commit 78968f698d25edc11475accb0b4d7b52771acfe1 · bretton.dev/coves

+2 -1

cmd/server/main.go

···

       247
       247
        
       		log.Println("   Set SKIP_DID_WEB_VERIFICATION=false for production")

     

       248
       248
        
       	}

     

       249
       249
        
       

     

       250
       250
       -
       	communityEventConsumer := jetstream.NewCommunityEventConsumer(communityRepo, instanceDID, skipDIDWebVerification)

     

       250
       250
       +
       	// Pass identity resolver to consumer for PLC handle resolution (source of truth)

     

       251
       251
       +
       	communityEventConsumer := jetstream.NewCommunityEventConsumer(communityRepo, instanceDID, skipDIDWebVerification, identityResolver)

     

       251
       252
        
       	communityJetstreamConnector := jetstream.NewCommunityJetstreamConnector(communityEventConsumer, communityJetstreamURL)

     

       252
       253
        
       

     

       253
       254
        
       	go func() {

+65 -1

internal/atproto/jetstream/community_consumer.go

···

       1
       1
        
       package jetstream

     

       2
       2
        
       

     

       3
       3
        
       import (

     

       4
       4
       +
       	"Coves/internal/atproto/identity"

     

       4
       5
        
       	"Coves/internal/atproto/utils"

     

       5
       6
        
       	"Coves/internal/core/communities"

     

       6
       7
        
       	"context"

     
···

       19
       20
        
       // CommunityEventConsumer consumes community-related events from Jetstream

     

       20
       21
        
       type CommunityEventConsumer struct {

     

       21
       22
        
       	repo             communities.Repository           // Repository for community operations

     

       23
       23
       +
       	identityResolver interface{ Resolve(context.Context, string) (*identity.Identity, error) } // For resolving handles from DIDs

     

       22
       24
        
       	httpClient       *http.Client                     // Shared HTTP client with connection pooling

     

       23
       25
        
       	didCache         *lru.Cache[string, cachedDIDDoc] // Bounded LRU cache for .well-known verification results

     

       24
       26
        
       	wellKnownLimiter *rate.Limiter                    // Rate limiter for .well-known fetches

     
···

       35
       37
        
       // NewCommunityEventConsumer creates a new Jetstream consumer for community events

     

       36
       38
        
       // instanceDID: The DID of this Coves instance (for hostedBy verification)

     

       37
       39
        
       // skipVerification: Skip did:web verification (for dev mode)

     

       38
       38
       -
       func NewCommunityEventConsumer(repo communities.Repository, instanceDID string, skipVerification bool) *CommunityEventConsumer {

     

       40
       40
       +
       // identityResolver: Optional resolver for resolving handles from DIDs (can be nil for tests)

     

       41
       41
       +
       func NewCommunityEventConsumer(repo communities.Repository, instanceDID string, skipVerification bool, identityResolver interface{ Resolve(context.Context, string) (*identity.Identity, error) }) *CommunityEventConsumer {

     

       39
       42
        
       	// Create bounded LRU cache for DID document verification results

     

       40
       43
        
       	// Max 1000 entries to prevent unbounded memory growth (PR review feedback)

     

       41
       44
        
       	// Each entry ~100 bytes → max ~100KB memory overhead

     
···

       49
       52
        
       

     

       50
       53
        
       	return &CommunityEventConsumer{

     

       51
       54
        
       		repo:             repo,

     

       55
       55
       +
       		identityResolver: identityResolver, // Optional - can be nil for tests

     

       52
       56
        
       		instanceDID:      instanceDID,

     

       53
       57
        
       		skipVerification: skipVerification,

     

       54
       58
        
       		// Shared HTTP client with connection pooling for .well-known fetches

     
···

       129
       133
        
       		return fmt.Errorf("failed to parse community profile: %w", err)

     

       130
       134
        
       	}

     

       131
       135
        
       

     

       136
       136
       +
       	// atProto Best Practice: Handles are NOT stored in records (they're mutable, resolved from DIDs)

     

       137
       137
       +
       	// If handle is missing from record (new atProto-compliant records), resolve it from PLC/DID

     

       138
       138
       +
       	if profile.Handle == "" {

     

       139
       139
       +
       		if c.identityResolver != nil {

     

       140
       140
       +
       			// Production: Resolve handle from PLC (source of truth)

     

       141
       141
       +
       			// NO FALLBACK - if PLC is down, we fail and backfill later

     

       142
       142
       +
       			// This prevents creating communities with incorrect handles in federated scenarios

     

       143
       143
       +
       			identity, err := c.identityResolver.Resolve(ctx, did)

     

       144
       144
       +
       			if err != nil {

     

       145
       145
       +
       				return fmt.Errorf("failed to resolve handle from PLC for %s: %w (no fallback - will retry during backfill)", did, err)

     

       146
       146
       +
       			}

     

       147
       147
       +
       			profile.Handle = identity.Handle

     

       148
       148
       +
       			log.Printf("✓ Resolved handle from PLC: %s (did=%s, method=%s)",

     

       149
       149
       +
       				profile.Handle, did, identity.Method)

     

       150
       150
       +
       		} else {

     

       151
       151
       +
       			// Test mode only: construct deterministically when no resolver available

     

       152
       152
       +
       			profile.Handle = constructHandleFromProfile(profile)

     

       153
       153
       +
       			log.Printf("✓ Constructed handle (test mode): %s (name=%s, hostedBy=%s)",

     

       154
       154
       +
       				profile.Handle, profile.Name, profile.HostedBy)

     

       155
       155
       +
       		}

     

       156
       156
       +
       	}

     

       157
       157
       +
       

     

       132
       158
        
       	// SECURITY: Verify hostedBy claim matches handle domain

     

       133
       159
        
       	// This prevents malicious instances from claiming to host communities for domains they don't own

     

       134
       160
        
       	if err := c.verifyHostedByClaim(ctx, profile.Handle, profile.HostedBy); err != nil {

     
···

       223
       249
        
       	profile, err := parseCommunityProfile(commit.Record)

     

       224
       250
        
       	if err != nil {

     

       225
       251
        
       		return fmt.Errorf("failed to parse community profile: %w", err)

     

       252
       252
       +
       	}

     

       253
       253
       +
       

     

       254
       254
       +
       	// atProto Best Practice: Handles are NOT stored in records (they're mutable, resolved from DIDs)

     

       255
       255
       +
       	// If handle is missing from record (new atProto-compliant records), resolve it from PLC/DID

     

       256
       256
       +
       	if profile.Handle == "" {

     

       257
       257
       +
       		if c.identityResolver != nil {

     

       258
       258
       +
       			// Production: Resolve handle from PLC (source of truth)

     

       259
       259
       +
       			// NO FALLBACK - if PLC is down, we fail and backfill later

     

       260
       260
       +
       			// This prevents creating communities with incorrect handles in federated scenarios

     

       261
       261
       +
       			identity, err := c.identityResolver.Resolve(ctx, did)

     

       262
       262
       +
       			if err != nil {

     

       263
       263
       +
       				return fmt.Errorf("failed to resolve handle from PLC for %s: %w (no fallback - will retry during backfill)", did, err)

     

       264
       264
       +
       			}

     

       265
       265
       +
       			profile.Handle = identity.Handle

     

       266
       266
       +
       			log.Printf("✓ Resolved handle from PLC: %s (did=%s, method=%s)",

     

       267
       267
       +
       				profile.Handle, did, identity.Method)

     

       268
       268
       +
       		} else {

     

       269
       269
       +
       			// Test mode only: construct deterministically when no resolver available

     

       270
       270
       +
       			profile.Handle = constructHandleFromProfile(profile)

     

       271
       271
       +
       			log.Printf("✓ Constructed handle (test mode): %s (name=%s, hostedBy=%s)",

     

       272
       272
       +
       				profile.Handle, profile.Name, profile.HostedBy)

     

       273
       273
       +
       		}

     

       226
       274
        
       	}

     

       227
       275
        
       

     

       228
       276
        
       	// V2: Repository DID IS the community DID

     
···

       707
       755
        
       	}

     

       708
       756
        
       

     

       709
       757
        
       	return &profile, nil

     

       758
       758
       +
       }

     

       759
       759
       +
       

     

       760
       760
       +
       // constructHandleFromProfile constructs a deterministic handle from profile data

     

       761
       761
       +
       // Format: {name}.community.{instanceDomain}

     

       762
       762
       +
       // Example: gaming.community.coves.social

     

       763
       763
       +
       // This is ONLY used in test mode (when identity resolver is nil)

     

       764
       764
       +
       // Production MUST resolve handles from PLC (source of truth)

     

       765
       765
       +
       // Returns empty string if hostedBy is not did:web format (caller will fail validation)

     

       766
       766
       +
       func constructHandleFromProfile(profile *CommunityProfile) string {

     

       767
       767
       +
       	if !strings.HasPrefix(profile.HostedBy, "did:web:") {

     

       768
       768
       +
       		// hostedBy must be did:web format for handle construction

     

       769
       769
       +
       		// Return empty to trigger validation error in repository

     

       770
       770
       +
       		return ""

     

       771
       771
       +
       	}

     

       772
       772
       +
       	instanceDomain := strings.TrimPrefix(profile.HostedBy, "did:web:")

     

       773
       773
       +
       	return fmt.Sprintf("%s.community.%s", profile.Name, instanceDomain)

     

       710
       774
        
       }

     

       711
       775
        
       

     

       712
       776
        
       // extractContentVisibility extracts contentVisibility from subscription record with clamping

+6 -1

internal/db/postgres/community_repo.go

···

       22
       22
        
       

     

       23
       23
        
       // Create inserts a new community into the communities table

     

       24
       24
        
       func (r *postgresCommunityRepo) Create(ctx context.Context, community *communities.Community) (*communities.Community, error) {

     

       25
       25
       +
       	// Validate that handle is always provided (constructed by consumer)

     

       26
       26
       +
       	if community.Handle == "" {

     

       27
       27
       +
       		return nil, fmt.Errorf("handle is required (should be constructed by consumer before insert)")

     

       28
       28
       +
       	}

     

       29
       29
       +
       

     

       25
       30
        
       	query := `

     

       26
       31
        
       		INSERT INTO communities (

     

       27
       32
        
       			did, handle, name, display_name, description, description_facets,

     
···

       54
       59
        
       

     

       55
       60
        
       	err := r.db.QueryRowContext(ctx, query,

     

       56
       61
        
       		community.DID,

     

       57
       57
       -
       		community.Handle,

     

       62
       62
       +
       		community.Handle, // Always non-empty - constructed by AppView consumer

     

       58
       63
        
       		community.Name,

     

       59
       64
        
       		nullString(community.DisplayName),

     

       60
       65
        
       		nullString(community.Description),

+2 -1

tests/integration/community_blocking_test.go

···

       24
       24
        
       

     

       25
       25
        
       	repo := createBlockingTestCommunityRepo(t, db)

     

       26
       26
        
       	// Skip verification in tests

     

       27
       27
       -
       	consumer := jetstream.NewCommunityEventConsumer(repo, "did:web:coves.local", true)

     

       27
       27
       +
       	// Pass nil for identity resolver - not needed since consumer constructs handles from DIDs

     

       28
       28
       +
       	consumer := jetstream.NewCommunityEventConsumer(repo, "did:web:coves.local", true, nil)

     

       28
       29
        
       

     

       29
       30
        
       	// Create test community

     

       30
       31
        
       	testDID := fmt.Sprintf("did:plc:test-community-%d", time.Now().UnixNano())

+265 -14

tests/integration/community_consumer_test.go

···

       1
       1
        
       package integration

     

       2
       2
        
       

     

       3
       3
        
       import (

     

       4
       4
       +
       	"Coves/internal/atproto/identity"

     

       4
       5
        
       	"Coves/internal/atproto/jetstream"

     

       5
       6
        
       	"Coves/internal/core/communities"

     

       6
       7
        
       	"Coves/internal/db/postgres"

     

       7
       8
        
       	"context"

     

       9
       9
       +
       	"errors"

     

       8
       10
        
       	"fmt"

     

       9
       11
        
       	"testing"

     

       10
       12
        
       	"time"

     
···

       19
       21
        
       	}()

     

       20
       22
        
       

     

       21
       23
        
       	repo := postgres.NewCommunityRepository(db)

     

       22
       22
       -
       	// Skip verification in tests

     

       23
       23
       -
       	consumer := jetstream.NewCommunityEventConsumer(repo, "did:web:coves.local", true)

     

       24
       24
        
       	ctx := context.Background()

     

       25
       25
        
       

     

       26
       26
        
       	t.Run("creates community from firehose event", func(t *testing.T) {

     

       27
       27
        
       		uniqueSuffix := fmt.Sprintf("%d", time.Now().UnixNano())

     

       28
       28
        
       		communityDID := generateTestDID(uniqueSuffix)

     

       29
       29
       +
       		communityName := fmt.Sprintf("test-community-%s", uniqueSuffix)

     

       30
       30
       +
       		expectedHandle := fmt.Sprintf("%s.community.coves.local", communityName)

     

       31
       31
       +
       

     

       32
       32
       +
       		// Set up mock resolver for this test DID

     

       33
       33
       +
       		mockResolver := newMockIdentityResolver()

     

       34
       34
       +
       		mockResolver.resolutions[communityDID] = expectedHandle

     

       35
       35
       +
       		consumer := jetstream.NewCommunityEventConsumer(repo, "did:web:coves.local", true, mockResolver)

     

       29
       36
        
       

     

       30
       37
        
       		// Simulate a Jetstream commit event

     

       31
       38
        
       		event := &jetstream.JetstreamEvent{

     
···

       41
       48
        
       				Record: map[string]interface{}{

     

       42
       49
        
       					// Note: No 'did', 'handle', 'memberCount', or 'subscriberCount' in record

     

       43
       50
        
       					// These are resolved/computed by AppView, not stored in immutable records

     

       44
       44
       -
       					"name":        "test-community",

     

       51
       51
       +
       					"name":        communityName,

     

       45
       52
        
       					"displayName": "Test Community",

     

       46
       53
        
       					"description": "A test community",

     

       47
       54
        
       					"owner":       "did:web:coves.local",

     
···

       81
       88
        
       	t.Run("updates existing community", func(t *testing.T) {

     

       82
       89
        
       		uniqueSuffix := fmt.Sprintf("%d", time.Now().UnixNano())

     

       83
       90
        
       		communityDID := generateTestDID(uniqueSuffix)

     

       84
       84
       -
       		handle := fmt.Sprintf("!update-test-%s@coves.local", uniqueSuffix)

     

       91
       91
       +
       		communityName := "update-test"

     

       92
       92
       +
       		expectedHandle := fmt.Sprintf("%s.community.coves.local", communityName)

     

       93
       93
       +
       

     

       94
       94
       +
       		// Set up mock resolver for this test DID

     

       95
       95
       +
       		mockResolver := newMockIdentityResolver()

     

       96
       96
       +
       		mockResolver.resolutions[communityDID] = expectedHandle

     

       97
       97
       +
       		consumer := jetstream.NewCommunityEventConsumer(repo, "did:web:coves.local", true, mockResolver)

     

       85
       98
        
       

     

       86
       99
        
       		// Create initial community

     

       87
       100
        
       		initialCommunity := &communities.Community{

     

       88
       101
        
       			DID:                    communityDID,

     

       89
       89
       -
       			Handle:                 handle,

     

       90
       90
       -
       			Name:                   "update-test",

     

       102
       102
       +
       			Handle:                 expectedHandle,

     

       103
       103
       +
       			Name:                   communityName,

     

       91
       104
        
       			DisplayName:            "Original Name",

     

       92
       105
        
       			Description:            "Original description",

     

       93
       106
        
       			OwnerDID:               "did:web:coves.local",

     
···

       160
       173
        
       	t.Run("deletes community", func(t *testing.T) {

     

       161
       174
        
       		uniqueSuffix := fmt.Sprintf("%d", time.Now().UnixNano())

     

       162
       175
        
       		communityDID := generateTestDID(uniqueSuffix)

     

       176
       176
       +
       		communityName := "delete-test"

     

       177
       177
       +
       		expectedHandle := fmt.Sprintf("%s.community.coves.local", communityName)

     

       178
       178
       +
       

     

       179
       179
       +
       		// Set up mock resolver for this test DID

     

       180
       180
       +
       		mockResolver := newMockIdentityResolver()

     

       181
       181
       +
       		mockResolver.resolutions[communityDID] = expectedHandle

     

       182
       182
       +
       		consumer := jetstream.NewCommunityEventConsumer(repo, "did:web:coves.local", true, mockResolver)

     

       163
       183
        
       

     

       164
       184
        
       		// Create community to delete

     

       165
       185
        
       		community := &communities.Community{

     

       166
       186
        
       			DID:          communityDID,

     

       167
       167
       -
       			Handle:       fmt.Sprintf("!delete-test-%s@coves.local", uniqueSuffix),

     

       168
       168
       -
       			Name:         "delete-test",

     

       187
       187
       +
       			Handle:       expectedHandle,

     

       188
       188
       +
       			Name:         communityName,

     

       169
       189
        
       			OwnerDID:     "did:web:coves.local",

     

       170
       190
        
       			CreatedByDID: "did:plc:user123",

     

       171
       191
        
       			HostedByDID:  "did:web:coves.local",

     
···

       212
       232
        
       	}()

     

       213
       233
        
       

     

       214
       234
        
       	repo := postgres.NewCommunityRepository(db)

     

       215
       215
       -
       	// Skip verification in tests

     

       216
       216
       -
       	consumer := jetstream.NewCommunityEventConsumer(repo, "did:web:coves.local", true)

     

       217
       235
        
       	ctx := context.Background()

     

       218
       236
        
       

     

       219
       237
        
       	t.Run("creates subscription from event", func(t *testing.T) {

     

       220
       238
        
       		// Create a community first

     

       221
       239
        
       		uniqueSuffix := fmt.Sprintf("%d", time.Now().UnixNano())

     

       222
       240
        
       		communityDID := generateTestDID(uniqueSuffix)

     

       241
       241
       +
       		communityName := "sub-test"

     

       242
       242
       +
       		expectedHandle := fmt.Sprintf("%s.community.coves.local", communityName)

     

       243
       243
       +
       

     

       244
       244
       +
       		// Set up mock resolver for this test DID

     

       245
       245
       +
       		mockResolver := newMockIdentityResolver()

     

       246
       246
       +
       		mockResolver.resolutions[communityDID] = expectedHandle

     

       247
       247
       +
       		consumer := jetstream.NewCommunityEventConsumer(repo, "did:web:coves.local", true, mockResolver)

     

       223
       248
        
       

     

       224
       249
        
       		community := &communities.Community{

     

       225
       250
        
       			DID:          communityDID,

     

       226
       226
       -
       			Handle:       fmt.Sprintf("!sub-test-%s@coves.local", uniqueSuffix),

     

       227
       227
       -
       			Name:         "sub-test",

     

       251
       251
       +
       			Handle:       expectedHandle,

     

       252
       252
       +
       			Name:         communityName,

     

       228
       253
        
       			OwnerDID:     "did:web:coves.local",

     

       229
       254
        
       			CreatedByDID: "did:plc:user123",

     

       230
       255
        
       			HostedByDID:  "did:web:coves.local",

     
···

       297
       322
        
       	}()

     

       298
       323
        
       

     

       299
       324
        
       	repo := postgres.NewCommunityRepository(db)

     

       300
       300
       -
       	// Skip verification in tests

     

       301
       301
       -
       	consumer := jetstream.NewCommunityEventConsumer(repo, "did:web:coves.local", true)

     

       325
       325
       +
       	// Use mock resolver (though these tests don't create communities, so it won't be called)

     

       326
       326
       +
       	mockResolver := newMockIdentityResolver()

     

       327
       327
       +
       	consumer := jetstream.NewCommunityEventConsumer(repo, "did:web:coves.local", true, mockResolver)

     

       302
       328
        
       	ctx := context.Background()

     

       303
       329
        
       

     

       304
       330
        
       	t.Run("ignores identity events", func(t *testing.T) {

     
···

       340
       366
        
       		}

     

       341
       367
        
       	})

     

       342
       368
        
       }

     

       369
       369
       +
       

     

       370
       370
       +
       // mockIdentityResolver is a test double for identity resolution

     

       371
       371
       +
       type mockIdentityResolver struct {

     

       372
       372
       +
       	// Map of DID -> handle for successful resolutions

     

       373
       373
       +
       	resolutions map[string]string

     

       374
       374
       +
       	// If true, Resolve returns an error

     

       375
       375
       +
       	shouldFail bool

     

       376
       376
       +
       	// Track calls to verify invocation

     

       377
       377
       +
       	callCount int

     

       378
       378
       +
       	lastDID   string

     

       379
       379
       +
       }

     

       380
       380
       +
       

     

       381
       381
       +
       func newMockIdentityResolver() *mockIdentityResolver {

     

       382
       382
       +
       	return &mockIdentityResolver{

     

       383
       383
       +
       		resolutions: make(map[string]string),

     

       384
       384
       +
       	}

     

       385
       385
       +
       }

     

       386
       386
       +
       

     

       387
       387
       +
       func (m *mockIdentityResolver) Resolve(ctx context.Context, did string) (*identity.Identity, error) {

     

       388
       388
       +
       	m.callCount++

     

       389
       389
       +
       	m.lastDID = did

     

       390
       390
       +
       

     

       391
       391
       +
       	if m.shouldFail {

     

       392
       392
       +
       		return nil, errors.New("mock PLC resolution failure")

     

       393
       393
       +
       	}

     

       394
       394
       +
       

     

       395
       395
       +
       	handle, ok := m.resolutions[did]

     

       396
       396
       +
       	if !ok {

     

       397
       397
       +
       		return nil, fmt.Errorf("no resolution configured for DID: %s", did)

     

       398
       398
       +
       	}

     

       399
       399
       +
       

     

       400
       400
       +
       	return &identity.Identity{

     

       401
       401
       +
       		DID:        did,

     

       402
       402
       +
       		Handle:     handle,

     

       403
       403
       +
       		PDSURL:     "https://pds.example.com",

     

       404
       404
       +
       		ResolvedAt: time.Now(),

     

       405
       405
       +
       		Method:     identity.MethodHTTPS,

     

       406
       406
       +
       	}, nil

     

       407
       407
       +
       }

     

       408
       408
       +
       

     

       409
       409
       +
       func TestCommunityConsumer_PLCHandleResolution(t *testing.T) {

     

       410
       410
       +
       	db := setupTestDB(t)

     

       411
       411
       +
       	defer func() {

     

       412
       412
       +
       		if err := db.Close(); err != nil {

     

       413
       413
       +
       			t.Logf("Failed to close database: %v", err)

     

       414
       414
       +
       		}

     

       415
       415
       +
       	}()

     

       416
       416
       +
       

     

       417
       417
       +
       	repo := postgres.NewCommunityRepository(db)

     

       418
       418
       +
       	ctx := context.Background()

     

       419
       419
       +
       

     

       420
       420
       +
       	t.Run("resolves handle from PLC successfully", func(t *testing.T) {

     

       421
       421
       +
       		uniqueSuffix := fmt.Sprintf("%d", time.Now().UnixNano())

     

       422
       422
       +
       		communityDID := generateTestDID(uniqueSuffix)

     

       423
       423
       +
       		communityName := fmt.Sprintf("test-plc-%s", uniqueSuffix)

     

       424
       424
       +
       		expectedHandle := fmt.Sprintf("%s.community.coves.social", communityName)

     

       425
       425
       +
       

     

       426
       426
       +
       		// Create mock resolver

     

       427
       427
       +
       		mockResolver := newMockIdentityResolver()

     

       428
       428
       +
       		mockResolver.resolutions[communityDID] = expectedHandle

     

       429
       429
       +
       

     

       430
       430
       +
       		consumer := jetstream.NewCommunityEventConsumer(repo, "did:web:coves.local", true, mockResolver)

     

       431
       431
       +
       

     

       432
       432
       +
       		// Simulate Jetstream event without handle in record

     

       433
       433
       +
       		event := &jetstream.JetstreamEvent{

     

       434
       434
       +
       			Did:    communityDID,

     

       435
       435
       +
       			TimeUS: time.Now().UnixMicro(),

     

       436
       436
       +
       			Kind:   "commit",

     

       437
       437
       +
       			Commit: &jetstream.CommitEvent{

     

       438
       438
       +
       				Rev:        "rev123",

     

       439
       439
       +
       				Operation:  "create",

     

       440
       440
       +
       				Collection: "social.coves.community.profile",

     

       441
       441
       +
       				RKey:       "self",

     

       442
       442
       +
       				CID:        "bafy123abc",

     

       443
       443
       +
       				Record: map[string]interface{}{

     

       444
       444
       +
       					// No handle field - should trigger PLC resolution

     

       445
       445
       +
       					"name":        communityName,

     

       446
       446
       +
       					"displayName": "Test PLC Community",

     

       447
       447
       +
       					"description": "Testing PLC resolution",

     

       448
       448
       +
       					"owner":       "did:web:coves.local",

     

       449
       449
       +
       					"createdBy":   "did:plc:user123",

     

       450
       450
       +
       					"hostedBy":    "did:web:coves.local",

     

       451
       451
       +
       					"visibility":  "public",

     

       452
       452
       +
       					"federation": map[string]interface{}{

     

       453
       453
       +
       						"allowExternalDiscovery": true,

     

       454
       454
       +
       					},

     

       455
       455
       +
       					"createdAt": time.Now().Format(time.RFC3339),

     

       456
       456
       +
       				},

     

       457
       457
       +
       			},

     

       458
       458
       +
       		}

     

       459
       459
       +
       

     

       460
       460
       +
       		// Handle the event

     

       461
       461
       +
       		if err := consumer.HandleEvent(ctx, event); err != nil {

     

       462
       462
       +
       			t.Fatalf("Failed to handle event: %v", err)

     

       463
       463
       +
       		}

     

       464
       464
       +
       

     

       465
       465
       +
       		// Verify mock was called

     

       466
       466
       +
       		if mockResolver.callCount != 1 {

     

       467
       467
       +
       			t.Errorf("Expected 1 PLC resolution call, got %d", mockResolver.callCount)

     

       468
       468
       +
       		}

     

       469
       469
       +
       		if mockResolver.lastDID != communityDID {

     

       470
       470
       +
       			t.Errorf("Expected PLC resolution for DID %s, got %s", communityDID, mockResolver.lastDID)

     

       471
       471
       +
       		}

     

       472
       472
       +
       

     

       473
       473
       +
       		// Verify community was indexed with PLC-resolved handle

     

       474
       474
       +
       		community, err := repo.GetByDID(ctx, communityDID)

     

       475
       475
       +
       		if err != nil {

     

       476
       476
       +
       			t.Fatalf("Failed to get indexed community: %v", err)

     

       477
       477
       +
       		}

     

       478
       478
       +
       

     

       479
       479
       +
       		if community.Handle != expectedHandle {

     

       480
       480
       +
       			t.Errorf("Expected handle %s from PLC, got %s", expectedHandle, community.Handle)

     

       481
       481
       +
       		}

     

       482
       482
       +
       	})

     

       483
       483
       +
       

     

       484
       484
       +
       	t.Run("fails when PLC resolution fails (no fallback)", func(t *testing.T) {

     

       485
       485
       +
       		uniqueSuffix := fmt.Sprintf("%d", time.Now().UnixNano())

     

       486
       486
       +
       		communityDID := generateTestDID(uniqueSuffix)

     

       487
       487
       +
       		communityName := fmt.Sprintf("test-plc-fail-%s", uniqueSuffix)

     

       488
       488
       +
       

     

       489
       489
       +
       		// Create mock resolver that fails

     

       490
       490
       +
       		mockResolver := newMockIdentityResolver()

     

       491
       491
       +
       		mockResolver.shouldFail = true

     

       492
       492
       +
       

     

       493
       493
       +
       		consumer := jetstream.NewCommunityEventConsumer(repo, "did:web:coves.local", true, mockResolver)

     

       494
       494
       +
       

     

       495
       495
       +
       		// Simulate Jetstream event without handle in record

     

       496
       496
       +
       		event := &jetstream.JetstreamEvent{

     

       497
       497
       +
       			Did:    communityDID,

     

       498
       498
       +
       			TimeUS: time.Now().UnixMicro(),

     

       499
       499
       +
       			Kind:   "commit",

     

       500
       500
       +
       			Commit: &jetstream.CommitEvent{

     

       501
       501
       +
       				Rev:        "rev456",

     

       502
       502
       +
       				Operation:  "create",

     

       503
       503
       +
       				Collection: "social.coves.community.profile",

     

       504
       504
       +
       				RKey:       "self",

     

       505
       505
       +
       				CID:        "bafy456def",

     

       506
       506
       +
       				Record: map[string]interface{}{

     

       507
       507
       +
       					"name":        communityName,

     

       508
       508
       +
       					"displayName": "Test PLC Failure",

     

       509
       509
       +
       					"description": "Testing PLC failure",

     

       510
       510
       +
       					"owner":       "did:web:coves.local",

     

       511
       511
       +
       					"createdBy":   "did:plc:user123",

     

       512
       512
       +
       					"hostedBy":    "did:web:coves.local",

     

       513
       513
       +
       					"visibility":  "public",

     

       514
       514
       +
       					"federation": map[string]interface{}{

     

       515
       515
       +
       						"allowExternalDiscovery": true,

     

       516
       516
       +
       					},

     

       517
       517
       +
       					"createdAt": time.Now().Format(time.RFC3339),

     

       518
       518
       +
       				},

     

       519
       519
       +
       			},

     

       520
       520
       +
       		}

     

       521
       521
       +
       

     

       522
       522
       +
       		// Handle the event - should fail

     

       523
       523
       +
       		err := consumer.HandleEvent(ctx, event)

     

       524
       524
       +
       		if err == nil {

     

       525
       525
       +
       			t.Fatal("Expected error when PLC resolution fails, got nil")

     

       526
       526
       +
       		}

     

       527
       527
       +
       

     

       528
       528
       +
       		// Verify error message indicates PLC failure

     

       529
       529
       +
       		expectedErrSubstring := "failed to resolve handle from PLC"

     

       530
       530
       +
       		if !contains(err.Error(), expectedErrSubstring) {

     

       531
       531
       +
       			t.Errorf("Expected error containing '%s', got: %v", expectedErrSubstring, err)

     

       532
       532
       +
       		}

     

       533
       533
       +
       

     

       534
       534
       +
       		// Verify community was NOT indexed

     

       535
       535
       +
       		_, err = repo.GetByDID(ctx, communityDID)

     

       536
       536
       +
       		if !communities.IsNotFound(err) {

     

       537
       537
       +
       			t.Errorf("Expected community NOT to be indexed when PLC fails, but got: %v", err)

     

       538
       538
       +
       		}

     

       539
       539
       +
       

     

       540
       540
       +
       		// Verify mock was called (failure happened during resolution, not before)

     

       541
       541
       +
       		if mockResolver.callCount != 1 {

     

       542
       542
       +
       			t.Errorf("Expected 1 PLC resolution attempt, got %d", mockResolver.callCount)

     

       543
       543
       +
       		}

     

       544
       544
       +
       	})

     

       545
       545
       +
       

     

       546
       546
       +
       	t.Run("test mode rejects invalid hostedBy format", func(t *testing.T) {

     

       547
       547
       +
       		uniqueSuffix := fmt.Sprintf("%d", time.Now().UnixNano())

     

       548
       548
       +
       		communityDID := generateTestDID(uniqueSuffix)

     

       549
       549
       +
       		communityName := fmt.Sprintf("test-invalid-hosted-%s", uniqueSuffix)

     

       550
       550
       +
       

     

       551
       551
       +
       		// No identity resolver (test mode)

     

       552
       552
       +
       		consumer := jetstream.NewCommunityEventConsumer(repo, "did:web:coves.local", true, nil)

     

       553
       553
       +
       

     

       554
       554
       +
       		// Event with invalid hostedBy format (not did:web)

     

       555
       555
       +
       		event := &jetstream.JetstreamEvent{

     

       556
       556
       +
       			Did:    communityDID,

     

       557
       557
       +
       			TimeUS: time.Now().UnixMicro(),

     

       558
       558
       +
       			Kind:   "commit",

     

       559
       559
       +
       			Commit: &jetstream.CommitEvent{

     

       560
       560
       +
       				Rev:        "rev789",

     

       561
       561
       +
       				Operation:  "create",

     

       562
       562
       +
       				Collection: "social.coves.community.profile",

     

       563
       563
       +
       				RKey:       "self",

     

       564
       564
       +
       				CID:        "bafy789ghi",

     

       565
       565
       +
       				Record: map[string]interface{}{

     

       566
       566
       +
       					"name":        communityName,

     

       567
       567
       +
       					"displayName": "Test Invalid HostedBy",

     

       568
       568
       +
       					"description": "Testing validation",

     

       569
       569
       +
       					"owner":       "did:web:coves.local",

     

       570
       570
       +
       					"createdBy":   "did:plc:user123",

     

       571
       571
       +
       					"hostedBy":    "did:plc:invalid", // Invalid format - not did:web

     

       572
       572
       +
       					"visibility":  "public",

     

       573
       573
       +
       					"federation": map[string]interface{}{

     

       574
       574
       +
       						"allowExternalDiscovery": true,

     

       575
       575
       +
       					},

     

       576
       576
       +
       					"createdAt": time.Now().Format(time.RFC3339),

     

       577
       577
       +
       				},

     

       578
       578
       +
       			},

     

       579
       579
       +
       		}

     

       580
       580
       +
       

     

       581
       581
       +
       		// Handle the event - should fail due to empty handle

     

       582
       582
       +
       		err := consumer.HandleEvent(ctx, event)

     

       583
       583
       +
       		if err == nil {

     

       584
       584
       +
       			t.Fatal("Expected error for invalid hostedBy format in test mode, got nil")

     

       585
       585
       +
       		}

     

       586
       586
       +
       

     

       587
       587
       +
       		// Verify error is about handle being required

     

       588
       588
       +
       		expectedErrSubstring := "handle is required"

     

       589
       589
       +
       		if !contains(err.Error(), expectedErrSubstring) {

     

       590
       590
       +
       			t.Errorf("Expected error containing '%s', got: %v", expectedErrSubstring, err)

     

       591
       591
       +
       		}

     

       592
       592
       +
       	})

     

       593
       593
       +
       }

+5 -4

tests/integration/community_e2e_test.go

···

       142
       142
        
       		svc.SetPDSAccessToken(accessToken)

     

       143
       143
        
       	}

     

       144
       144
        
       

     

       145
       145
       -
       	// Skip verification in tests

     

       146
       146
       -
       	consumer := jetstream.NewCommunityEventConsumer(communityRepo, "did:web:coves.local", true)

     

       145
       145
       +
       	// Use real identity resolver with local PLC for production-like testing

     

       146
       146
       +
       	consumer := jetstream.NewCommunityEventConsumer(communityRepo, "did:web:coves.local", true, identityResolver)

     

       147
       147
        
       

     

       148
       148
        
       	// Setup HTTP server with XRPC routes

     

       149
       149
        
       	r := chi.NewRouter()

     
···

       434
       434
        
       					Collection: "social.coves.community.profile",

     

       435
       435
        
       					RKey:       rkey,

     

       436
       436
        
       					Record: map[string]interface{}{

     

       437
       437
       -
       						"did":         createResp.DID,    // Community's DID from response

     

       438
       438
       -
       						"handle":      createResp.Handle, // Community's handle from response

     

       437
       437
       +
       						// Note: No 'did' or 'handle' in record (atProto best practice)

     

       438
       438
       +
       						// These are mutable and resolved from DIDs, not stored in immutable records

     

       439
       439
        
       						"name":        createReq["name"],

     

       440
       440
        
       						"displayName": createReq["displayName"],

     

       441
       441
        
       						"description": createReq["description"],

     

       442
       442
        
       						"visibility":  createReq["visibility"],

     

       443
       443
        
       						// Server-side derives these from JWT auth (instanceDID is the authenticated user)

     

       444
       444
       +
       						"owner":     instanceDID,

     

       444
       445
        
       						"createdBy": instanceDID,

     

       445
       446
        
       						"hostedBy":  instanceDID,

     

       446
       447
        
       						"federation": map[string]interface{}{

+10 -5

tests/integration/community_hostedby_security_test.go

···

       23
       23
        
       

     

       24
       24
        
       	t.Run("rejects community with mismatched hostedBy domain", func(t *testing.T) {

     

       25
       25
        
       		// Create consumer with verification enabled

     

       26
       26
       -
       		consumer := jetstream.NewCommunityEventConsumer(repo, "did:web:coves.social", false)

     

       26
       26
       +
       		// Pass nil for identity resolver - not needed since consumer constructs handles from DIDs

     

       27
       27
       +
       		consumer := jetstream.NewCommunityEventConsumer(repo, "did:web:coves.social", false, nil)

     

       27
       28
        
       

     

       28
       29
        
       		uniqueSuffix := fmt.Sprintf("%d", time.Now().UnixNano())

     

       29
       30
        
       		communityDID := generateTestDID(uniqueSuffix)

     
···

       81
       82
        
       

     

       82
       83
        
       	t.Run("accepts community with matching hostedBy domain", func(t *testing.T) {

     

       83
       84
        
       		// Create consumer with verification enabled

     

       84
       84
       -
       		consumer := jetstream.NewCommunityEventConsumer(repo, "did:web:coves.social", false)

     

       85
       85
       +
       		// Pass nil for identity resolver - not needed since consumer constructs handles from DIDs

     

       86
       86
       +
       		consumer := jetstream.NewCommunityEventConsumer(repo, "did:web:coves.social", false, nil)

     

       85
       87
        
       

     

       86
       88
        
       		uniqueSuffix := fmt.Sprintf("%d", time.Now().UnixNano())

     

       87
       89
        
       		communityDID := generateTestDID(uniqueSuffix)

     
···

       134
       136
        
       

     

       135
       137
        
       	t.Run("rejects hostedBy with non-did:web format", func(t *testing.T) {

     

       136
       138
        
       		// Create consumer with verification enabled

     

       137
       137
       -
       		consumer := jetstream.NewCommunityEventConsumer(repo, "did:web:coves.social", false)

     

       139
       139
       +
       		// Pass nil for identity resolver - not needed since consumer constructs handles from DIDs

     

       140
       140
       +
       		consumer := jetstream.NewCommunityEventConsumer(repo, "did:web:coves.social", false, nil)

     

       138
       141
        
       

     

       139
       142
        
       		uniqueSuffix := fmt.Sprintf("%d", time.Now().UnixNano())

     

       140
       143
        
       		communityDID := generateTestDID(uniqueSuffix)

     
···

       179
       182
        
       

     

       180
       183
        
       	t.Run("skip verification flag bypasses all checks", func(t *testing.T) {

     

       181
       184
        
       		// Create consumer with verification DISABLED

     

       182
       182
       -
       		consumer := jetstream.NewCommunityEventConsumer(repo, "did:web:coves.social", true)

     

       185
       185
       +
       		// Pass nil for identity resolver - not needed since consumer constructs handles from DIDs

     

       186
       186
       +
       		consumer := jetstream.NewCommunityEventConsumer(repo, "did:web:coves.social", true, nil)

     

       183
       187
        
       

     

       184
       188
        
       		uniqueSuffix := fmt.Sprintf("%d", time.Now().UnixNano())

     

       185
       189
        
       		communityDID := generateTestDID(uniqueSuffix)

     
···

       306
       310
        
       

     

       307
       311
        
       	for _, tc := range testCases {

     

       308
       312
        
       		t.Run(tc.name, func(t *testing.T) {

     

       309
       309
       -
       			consumer := jetstream.NewCommunityEventConsumer(repo, "did:web:coves.social", false)

     

       313
       313
       +
       			// Pass nil for identity resolver - not needed since consumer constructs handles from DIDs

     

       314
       314
       +
       			consumer := jetstream.NewCommunityEventConsumer(repo, "did:web:coves.social", false, nil)

     

       310
       315
        
       

     

       311
       316
        
       			uniqueSuffix := fmt.Sprintf("%d", time.Now().UnixNano())

     

       312
       317
        
       			communityDID := generateTestDID(uniqueSuffix)

+4 -2

tests/integration/community_v2_validation_test.go

···

       21
       21
        
       

     

       22
       22
        
       	repo := postgres.NewCommunityRepository(db)

     

       23
       23
        
       	// Skip verification in tests

     

       24
       24
       -
       	consumer := jetstream.NewCommunityEventConsumer(repo, "did:web:coves.local", true)

     

       24
       24
       +
       	// Pass nil for identity resolver - not needed since consumer constructs handles from DIDs

     

       25
       25
       +
       	consumer := jetstream.NewCommunityEventConsumer(repo, "did:web:coves.local", true, nil)

     

       25
       26
        
       	ctx := context.Background()

     

       26
       27
        
       

     

       27
       28
        
       	t.Run("accepts V2 community with rkey=self", func(t *testing.T) {

     
···

       249
       250
        
       

     

       250
       251
        
       	repo := postgres.NewCommunityRepository(db)

     

       251
       252
        
       	// Skip verification in tests

     

       252
       252
       -
       	consumer := jetstream.NewCommunityEventConsumer(repo, "did:web:coves.local", true)

     

       253
       253
       +
       	// Pass nil for identity resolver - not needed since consumer constructs handles from DIDs

     

       254
       254
       +
       	consumer := jetstream.NewCommunityEventConsumer(repo, "did:web:coves.local", true, nil)

     

       253
       255
        
       	ctx := context.Background()

     

       254
       256
        
       

     

       255
       257
        
       	t.Run("indexes community with atProto handle", func(t *testing.T) {

+6 -3

tests/integration/subscription_indexing_test.go

···

       25
       25
        
       

     

       26
       26
        
       	repo := createTestCommunityRepo(t, db)

     

       27
       27
        
       	// Skip verification in tests

     

       28
       28
       -
       	consumer := jetstream.NewCommunityEventConsumer(repo, "did:web:coves.local", true)

     

       28
       28
       +
       	// Pass nil for identity resolver - not needed since consumer constructs handles from DIDs

     

       29
       29
       +
       	consumer := jetstream.NewCommunityEventConsumer(repo, "did:web:coves.local", true, nil)

     

       29
       30
        
       

     

       30
       31
        
       	// Create a test community first (with unique DID)

     

       31
       32
        
       	testDID := fmt.Sprintf("did:plc:test-community-%d", time.Now().UnixNano())

     
···

       249
       250
        
       

     

       250
       251
        
       	repo := createTestCommunityRepo(t, db)

     

       251
       252
        
       	// Skip verification in tests

     

       252
       252
       -
       	consumer := jetstream.NewCommunityEventConsumer(repo, "did:web:coves.local", true)

     

       253
       253
       +
       	// Pass nil for identity resolver - not needed since consumer constructs handles from DIDs

     

       254
       254
       +
       	consumer := jetstream.NewCommunityEventConsumer(repo, "did:web:coves.local", true, nil)

     

       253
       255
        
       

     

       254
       256
        
       	// Create test community (with unique DID)

     

       255
       257
        
       	testDID := fmt.Sprintf("did:plc:test-unsub-%d", time.Now().UnixNano())

     
···

       364
       366
        
       

     

       365
       367
        
       	repo := createTestCommunityRepo(t, db)

     

       366
       368
        
       	// Skip verification in tests

     

       367
       367
       -
       	consumer := jetstream.NewCommunityEventConsumer(repo, "did:web:coves.local", true)

     

       369
       369
       +
       	// Pass nil for identity resolver - not needed since consumer constructs handles from DIDs

     

       370
       370
       +
       	consumer := jetstream.NewCommunityEventConsumer(repo, "did:web:coves.local", true, nil)

     

       368
       371
        
       

     

       369
       372
        
       	// Create test community (with unique DID)

     

       370
       373
        
       	testDID := fmt.Sprintf("did:plc:test-subcount-%d", time.Now().UnixNano())

+12

tests/integration/user_test.go

···

       70
       70
        
       	return db

     

       71
       71
        
       }

     

       72
       72
        
       

     

       73
       73
       +
       // setupIdentityResolver creates an identity resolver configured for local PLC testing

     

       74
       74
       +
       func setupIdentityResolver(db *sql.DB) interface{ Resolve(context.Context, string) (*identity.Identity, error) } {

     

       75
       75
       +
       	plcURL := os.Getenv("PLC_DIRECTORY_URL")

     

       76
       76
       +
       	if plcURL == "" {

     

       77
       77
       +
       		plcURL = "http://localhost:3002" // Local PLC directory

     

       78
       78
       +
       	}

     

       79
       79
       +
       

     

       80
       80
       +
       	config := identity.DefaultConfig()

     

       81
       81
       +
       	config.PLCURL = plcURL

     

       82
       82
       +
       	return identity.NewResolver(db, config)

     

       83
       83
       +
       }

     

       84
       84
       +
       

     

       73
       85
        
       // generateTestDID generates a unique test DID for integration tests

     

       74
       86
        
       // V2.0: No longer uses DID generator - just creates valid did:plc strings

     

       75
       87
        
       func generateTestDID(suffix string) string {