commit ff7cb2ef115baf086b2e7a97defd65a37a867288 · bretton.dev/coves

+99 -1

internal/atproto/jetstream/community_consumer.go

···

       1
        
       package jetstream

     

       2
        
       

     

       3
        
       import (

     

       0
        
       
     

       4
        
       	"Coves/internal/core/communities"

     

       5
        
       	"context"

     

       6
        
       	"encoding/json"

     
···

       35
        
       	// IMPORTANT: Collection names refer to RECORD TYPES in repositories, not XRPC procedures

     

       36
        
       	// - social.coves.community.profile: Community profile records (in community's own repo)

     

       37
        
       	// - social.coves.community.subscription: Subscription records (in user's repo)

     

       0
        
       
     

       38
        
       	//

     

       39
        
       	// XRPC procedures (social.coves.community.subscribe/unsubscribe) are just HTTP endpoints

     

       40
        
       	// that CREATE or DELETE records in these collections

     
···

       44
        
       	case "social.coves.community.subscription":

     

       45
        
       		// Handle both create (subscribe) and delete (unsubscribe) operations

     

       46
        
       		return c.handleSubscription(ctx, event.Did, commit)

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       47
        
       	default:

     

       48
        
       		// Not a community-related collection

     

       49
        
       		return nil

     
···

       267
        
       	uri := fmt.Sprintf("at://%s/social.coves.community.subscription/%s", userDID, commit.RKey)

     

       268
        
       

     

       269
        
       	// Create subscription entity

     

       0
        
       
     

       270
        
       	subscription := &communities.Subscription{

     

       271
        
       		UserDID:           userDID,

     

       272
        
       		CommunityDID:      communityDID,

     

       273
        
       		ContentVisibility: contentVisibility,

     

       274
       -
       		SubscribedAt:      time.Now(),

     

       275
        
       		RecordURI:         uri,

     

       276
        
       		RecordCID:         commit.CID,

     

       277
        
       	}

     
···

       325
        
       	}

     

       326
        
       

     

       327
        
       	log.Printf("✓ Removed subscription: %s -> %s", userDID, subscription.CommunityDID)

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       328
        
       	return nil

     

       329
        
       }

     

       330

···

       1
        
       package jetstream

     

       2
        
       

     

       3
        
       import (

     

       4
       +
       	"Coves/internal/atproto/utils"

     

       5
        
       	"Coves/internal/core/communities"

     

       6
        
       	"context"

     

       7
        
       	"encoding/json"

     
···

       36
        
       	// IMPORTANT: Collection names refer to RECORD TYPES in repositories, not XRPC procedures

     

       37
        
       	// - social.coves.community.profile: Community profile records (in community's own repo)

     

       38
        
       	// - social.coves.community.subscription: Subscription records (in user's repo)

     

       39
       +
       	// - social.coves.community.block: Block records (in user's repo)

     

       40
        
       	//

     

       41
        
       	// XRPC procedures (social.coves.community.subscribe/unsubscribe) are just HTTP endpoints

     

       42
        
       	// that CREATE or DELETE records in these collections

     
···

       46
        
       	case "social.coves.community.subscription":

     

       47
        
       		// Handle both create (subscribe) and delete (unsubscribe) operations

     

       48
        
       		return c.handleSubscription(ctx, event.Did, commit)

     

       49
       +
       	case "social.coves.community.block":

     

       50
       +
       		// Handle both create (block) and delete (unblock) operations

     

       51
       +
       		return c.handleBlock(ctx, event.Did, commit)

     

       52
        
       	default:

     

       53
        
       		// Not a community-related collection

     

       54
        
       		return nil

     
···

       272
        
       	uri := fmt.Sprintf("at://%s/social.coves.community.subscription/%s", userDID, commit.RKey)

     

       273
        
       

     

       274
        
       	// Create subscription entity

     

       275
       +
       	// Parse createdAt from record to preserve chronological ordering during replays

     

       276
        
       	subscription := &communities.Subscription{

     

       277
        
       		UserDID:           userDID,

     

       278
        
       		CommunityDID:      communityDID,

     

       279
        
       		ContentVisibility: contentVisibility,

     

       280
       +
       		SubscribedAt:      utils.ParseCreatedAt(commit.Record),

     

       281
        
       		RecordURI:         uri,

     

       282
        
       		RecordCID:         commit.CID,

     

       283
        
       	}

     
···

       331
        
       	}

     

       332
        
       

     

       333
        
       	log.Printf("✓ Removed subscription: %s -> %s", userDID, subscription.CommunityDID)

     

       334
       +
       	return nil

     

       335
       +
       }

     

       336
       +
       

     

       337
       +
       // handleBlock processes block create/delete events

     

       338
       +
       // CREATE operation = user blocked a community

     

       339
       +
       // DELETE operation = user unblocked a community

     

       340
       +
       func (c *CommunityEventConsumer) handleBlock(ctx context.Context, userDID string, commit *CommitEvent) error {

     

       341
       +
       	switch commit.Operation {

     

       342
       +
       	case "create":

     

       343
       +
       		return c.createBlock(ctx, userDID, commit)

     

       344
       +
       	case "delete":

     

       345
       +
       		return c.deleteBlock(ctx, userDID, commit)

     

       346
       +
       	default:

     

       347
       +
       		// Update operations shouldn't happen on blocks, but ignore gracefully

     

       348
       +
       		log.Printf("Ignoring unexpected operation on block: %s (userDID=%s, rkey=%s)",

     

       349
       +
       			commit.Operation, userDID, commit.RKey)

     

       350
       +
       		return nil

     

       351
       +
       	}

     

       352
       +
       }

     

       353
       +
       

     

       354
       +
       // createBlock indexes a new block

     

       355
       +
       func (c *CommunityEventConsumer) createBlock(ctx context.Context, userDID string, commit *CommitEvent) error {

     

       356
       +
       	if commit.Record == nil {

     

       357
       +
       		return fmt.Errorf("block create event missing record data")

     

       358
       +
       	}

     

       359
       +
       

     

       360
       +
       	// Extract community DID from record's subject field (following atProto conventions)

     

       361
       +
       	communityDID, ok := commit.Record["subject"].(string)

     

       362
       +
       	if !ok {

     

       363
       +
       		return fmt.Errorf("block record missing subject field")

     

       364
       +
       	}

     

       365
       +
       

     

       366
       +
       	// Build AT-URI for block record

     

       367
       +
       	// The record lives in the USER's repository

     

       368
       +
       	uri := fmt.Sprintf("at://%s/social.coves.community.block/%s", userDID, commit.RKey)

     

       369
       +
       

     

       370
       +
       	// Create block entity

     

       371
       +
       	// Parse createdAt from record to preserve chronological ordering during replays

     

       372
       +
       	block := &communities.CommunityBlock{

     

       373
       +
       		UserDID:      userDID,

     

       374
       +
       		CommunityDID: communityDID,

     

       375
       +
       		BlockedAt:    utils.ParseCreatedAt(commit.Record),

     

       376
       +
       		RecordURI:    uri,

     

       377
       +
       		RecordCID:    commit.CID,

     

       378
       +
       	}

     

       379
       +
       

     

       380
       +
       	// Index the block

     

       381
       +
       	// This is idempotent - safe for Jetstream replays

     

       382
       +
       	_, err := c.repo.BlockCommunity(ctx, block)

     

       383
       +
       	if err != nil {

     

       384
       +
       		// If already exists, that's fine (idempotency)

     

       385
       +
       		if communities.IsConflict(err) {

     

       386
       +
       			log.Printf("Block already indexed: %s -> %s", userDID, communityDID)

     

       387
       +
       			return nil

     

       388
       +
       		}

     

       389
       +
       		return fmt.Errorf("failed to index block: %w", err)

     

       390
       +
       	}

     

       391
       +
       

     

       392
       +
       	log.Printf("✓ Indexed block: %s -> %s", userDID, communityDID)

     

       393
       +
       	return nil

     

       394
       +
       }

     

       395
       +
       

     

       396
       +
       // deleteBlock removes a block from the index

     

       397
       +
       // DELETE operations don't include record data, so we need to look up the block

     

       398
       +
       // by its URI to find which community the user unblocked

     

       399
       +
       func (c *CommunityEventConsumer) deleteBlock(ctx context.Context, userDID string, commit *CommitEvent) error {

     

       400
       +
       	// Build AT-URI from the rkey

     

       401
       +
       	uri := fmt.Sprintf("at://%s/social.coves.community.block/%s", userDID, commit.RKey)

     

       402
       +
       

     

       403
       +
       	// Look up the block to get the community DID

     

       404
       +
       	// (DELETE operations don't include record data in Jetstream)

     

       405
       +
       	block, err := c.repo.GetBlockByURI(ctx, uri)

     

       406
       +
       	if err != nil {

     

       407
       +
       		if communities.IsNotFound(err) {

     

       408
       +
       			// Already deleted - this is fine (idempotency)

     

       409
       +
       			log.Printf("Block already deleted: %s", uri)

     

       410
       +
       			return nil

     

       411
       +
       		}

     

       412
       +
       		return fmt.Errorf("failed to find block for deletion: %w", err)

     

       413
       +
       	}

     

       414
       +
       

     

       415
       +
       	// Remove the block from the index

     

       416
       +
       	err = c.repo.UnblockCommunity(ctx, userDID, block.CommunityDID)

     

       417
       +
       	if err != nil {

     

       418
       +
       		if communities.IsNotFound(err) {

     

       419
       +
       			log.Printf("Block already removed: %s -> %s", userDID, block.CommunityDID)

     

       420
       +
       			return nil

     

       421
       +
       		}

     

       422
       +
       		return fmt.Errorf("failed to remove block: %w", err)

     

       423
       +
       	}

     

       424
       +
       

     

       425
       +
       	log.Printf("✓ Removed block: %s -> %s", userDID, block.CommunityDID)

     

       426
        
       	return nil

     

       427
        
       }

     

       428