commit 5a3ad7c4ce77c389be63dac248b3d808674db0a6 · bretton.dev/coves

+1 -1

cmd/server/main.go

···

       322
        
       		postJetstreamURL = "ws://localhost:6008/subscribe?wantedCollections=social.coves.community.post"

     

       323
        
       	}

     

       324
        
       

     

       325
       -
       	postEventConsumer := jetstream.NewPostEventConsumer(postRepo, communityRepo, userService)

     

       326
        
       	postJetstreamConnector := jetstream.NewPostJetstreamConnector(postEventConsumer, postJetstreamURL)

     

       327
        
       

     

       328
        
       	go func() {

···

       322
        
       		postJetstreamURL = "ws://localhost:6008/subscribe?wantedCollections=social.coves.community.post"

     

       323
        
       	}

     

       324
        
       

     

       325
       +
       	postEventConsumer := jetstream.NewPostEventConsumer(postRepo, communityRepo, userService, db)

     

       326
        
       	postJetstreamConnector := jetstream.NewPostJetstreamConnector(postEventConsumer, postJetstreamURL)

     

       327
        
       

     

       328
        
       	go func() {

+99 -9

internal/atproto/jetstream/post_consumer.go

···

       5
        
       	"Coves/internal/core/posts"

     

       6
        
       	"Coves/internal/core/users"

     

       7
        
       	"context"

     

       0
        
       
     

       8
        
       	"encoding/json"

     

       9
        
       	"fmt"

     

       10
        
       	"log"

     
···

       19
        
       	postRepo      posts.Repository

     

       20
        
       	communityRepo communities.Repository

     

       21
        
       	userService   users.UserService

     

       0
        
       
     

       22
        
       }

     

       23
        
       

     

       24
        
       // NewPostEventConsumer creates a new Jetstream consumer for post events

     
···

       26
        
       	postRepo posts.Repository,

     

       27
        
       	communityRepo communities.Repository,

     

       28
        
       	userService users.UserService,

     

       0
        
       
     

       29
        
       ) *PostEventConsumer {

     

       30
        
       	return &PostEventConsumer{

     

       31
        
       		postRepo:      postRepo,

     

       32
        
       		communityRepo: communityRepo,

     

       33
        
       		userService:   userService,

     

       0
        
       
     

       34
        
       	}

     

       35
        
       }

     

       36
        
       

     
···

       127
        
       		}

     

       128
        
       	}

     

       129
        
       

     

       130
       -
       	// Index in AppView database (idempotent - safe for Jetstream replays)

     

       131
       -
       	err = c.postRepo.Create(ctx, post)

     

       132
       -
       	if err != nil {

     

       133
       -
       		// Check if it already exists (idempotency)

     

       134
       -
       		if posts.IsConflict(err) {

     

       135
       -
       			log.Printf("Post already indexed: %s", uri)

     

       136
       -
       			return nil

     

       137
       -
       		}

     

       138
       -
       		return fmt.Errorf("failed to index post: %w", err)

     

       139
        
       	}

     

       140
        
       

     

       141
        
       	log.Printf("✓ Indexed post: %s (author: %s, community: %s, rkey: %s)",

     

       142
        
       		uri, post.AuthorDID, post.CommunityDID, commit.RKey)

     

       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
        
       
     

       143
        
       	return nil

     

       144
        
       }

     

       145

···

       5
        
       	"Coves/internal/core/posts"

     

       6
        
       	"Coves/internal/core/users"

     

       7
        
       	"context"

     

       8
       +
       	"database/sql"

     

       9
        
       	"encoding/json"

     

       10
        
       	"fmt"

     

       11
        
       	"log"

     
···

       20
        
       	postRepo      posts.Repository

     

       21
        
       	communityRepo communities.Repository

     

       22
        
       	userService   users.UserService

     

       23
       +
       	db            *sql.DB // Direct DB access for atomic count reconciliation

     

       24
        
       }

     

       25
        
       

     

       26
        
       // NewPostEventConsumer creates a new Jetstream consumer for post events

     
···

       28
        
       	postRepo posts.Repository,

     

       29
        
       	communityRepo communities.Repository,

     

       30
        
       	userService users.UserService,

     

       31
       +
       	db *sql.DB,

     

       32
        
       ) *PostEventConsumer {

     

       33
        
       	return &PostEventConsumer{

     

       34
        
       		postRepo:      postRepo,

     

       35
        
       		communityRepo: communityRepo,

     

       36
        
       		userService:   userService,

     

       37
       +
       		db:            db,

     

       38
        
       	}

     

       39
        
       }

     

       40
        
       

     
···

       131
        
       		}

     

       132
        
       	}

     

       133
        
       

     

       134
       +
       	// Atomically: Index post + Reconcile comment count for out-of-order arrivals

     

       135
       +
       	if err := c.indexPostAndReconcileCounts(ctx, post); err != nil {

     

       136
       +
       		return fmt.Errorf("failed to index post and reconcile counts: %w", err)

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       137
        
       	}

     

       138
        
       

     

       139
        
       	log.Printf("✓ Indexed post: %s (author: %s, community: %s, rkey: %s)",

     

       140
        
       		uri, post.AuthorDID, post.CommunityDID, commit.RKey)

     

       141
       +
       	return nil

     

       142
       +
       }

     

       143
       +
       

     

       144
       +
       // indexPostAndReconcileCounts atomically indexes a post and reconciles comment counts

     

       145
       +
       // This fixes the race condition where comments arrive before their parent post

     

       146
       +
       func (c *PostEventConsumer) indexPostAndReconcileCounts(ctx context.Context, post *posts.Post) error {

     

       147
       +
       	tx, err := c.db.BeginTx(ctx, nil)

     

       148
       +
       	if err != nil {

     

       149
       +
       		return fmt.Errorf("failed to begin transaction: %w", err)

     

       150
       +
       	}

     

       151
       +
       	defer func() {

     

       152
       +
       		if rollbackErr := tx.Rollback(); rollbackErr != nil && rollbackErr != sql.ErrTxDone {

     

       153
       +
       			log.Printf("Failed to rollback transaction: %v", rollbackErr)

     

       154
       +
       		}

     

       155
       +
       	}()

     

       156
       +
       

     

       157
       +
       	// 1. Insert the post (idempotent with RETURNING clause)

     

       158
       +
       	var facetsJSON, embedJSON, labelsJSON sql.NullString

     

       159
       +
       

     

       160
       +
       	if post.ContentFacets != nil {

     

       161
       +
       		facetsJSON.String = *post.ContentFacets

     

       162
       +
       		facetsJSON.Valid = true

     

       163
       +
       	}

     

       164
       +
       

     

       165
       +
       	if post.Embed != nil {

     

       166
       +
       		embedJSON.String = *post.Embed

     

       167
       +
       		embedJSON.Valid = true

     

       168
       +
       	}

     

       169
       +
       

     

       170
       +
       	if post.ContentLabels != nil {

     

       171
       +
       		labelsJSON.String = *post.ContentLabels

     

       172
       +
       		labelsJSON.Valid = true

     

       173
       +
       	}

     

       174
       +
       

     

       175
       +
       	insertQuery := `

     

       176
       +
       		INSERT INTO posts (

     

       177
       +
       			uri, cid, rkey, author_did, community_did,

     

       178
       +
       			title, content, content_facets, embed, content_labels,

     

       179
       +
       			created_at, indexed_at

     

       180
       +
       		) VALUES (

     

       181
       +
       			$1, $2, $3, $4, $5,

     

       182
       +
       			$6, $7, $8, $9, $10,

     

       183
       +
       			$11, NOW()

     

       184
       +
       		)

     

       185
       +
       		ON CONFLICT (uri) DO NOTHING

     

       186
       +
       		RETURNING id

     

       187
       +
       	`

     

       188
       +
       

     

       189
       +
       	var postID int64

     

       190
       +
       	insertErr := tx.QueryRowContext(

     

       191
       +
       		ctx, insertQuery,

     

       192
       +
       		post.URI, post.CID, post.RKey, post.AuthorDID, post.CommunityDID,

     

       193
       +
       		post.Title, post.Content, facetsJSON, embedJSON, labelsJSON,

     

       194
       +
       		post.CreatedAt,

     

       195
       +
       	).Scan(&postID)

     

       196
       +
       

     

       197
       +
       	// If no rows returned, post already exists (idempotent - OK for Jetstream replays)

     

       198
       +
       	if insertErr == sql.ErrNoRows {

     

       199
       +
       		log.Printf("Post already indexed: %s (idempotent)", post.URI)

     

       200
       +
       		if commitErr := tx.Commit(); commitErr != nil {

     

       201
       +
       			return fmt.Errorf("failed to commit transaction: %w", commitErr)

     

       202
       +
       		}

     

       203
       +
       		return nil

     

       204
       +
       	}

     

       205
       +
       

     

       206
       +
       	if insertErr != nil {

     

       207
       +
       		return fmt.Errorf("failed to insert post: %w", insertErr)

     

       208
       +
       	}

     

       209
       +
       

     

       210
       +
       	// 2. Reconcile comment_count for this newly inserted post

     

       211
       +
       	// In case any comments arrived out-of-order before this post was indexed

     

       212
       +
       	// This is the CRITICAL FIX for the race condition identified in the PR review

     

       213
       +
       	reconcileQuery := `

     

       214
       +
       		UPDATE posts

     

       215
       +
       		SET comment_count = (

     

       216
       +
       			SELECT COUNT(*)

     

       217
       +
       			FROM comments c

     

       218
       +
       			WHERE c.parent_uri = $1 AND c.deleted_at IS NULL

     

       219
       +
       		)

     

       220
       +
       		WHERE id = $2

     

       221
       +
       	`

     

       222
       +
       	_, reconcileErr := tx.ExecContext(ctx, reconcileQuery, post.URI, postID)

     

       223
       +
       	if reconcileErr != nil {

     

       224
       +
       		log.Printf("Warning: Failed to reconcile comment_count for %s: %v", post.URI, reconcileErr)

     

       225
       +
       		// Continue anyway - this is a best-effort reconciliation

     

       226
       +
       	}

     

       227
       +
       

     

       228
       +
       	// Commit transaction

     

       229
       +
       	if err := tx.Commit(); err != nil {

     

       230
       +
       		return fmt.Errorf("failed to commit transaction: %w", err)

     

       231
       +
       	}

     

       232
       +
       

     

       233
        
       	return nil

     

       234
        
       }

     

       235

+1 -1

tests/integration/aggregator_e2e_test.go

···

       75
        
       

     

       76
        
       	// Setup consumers

     

       77
        
       	aggregatorConsumer := jetstream.NewAggregatorEventConsumer(aggregatorRepo)

     

       78
       -
       	postConsumer := jetstream.NewPostEventConsumer(postRepo, communityRepo, userService)

     

       79
        
       

     

       80
        
       	// Setup HTTP handlers

     

       81
        
       	getServicesHandler := aggregator.NewGetServicesHandler(aggregatorService)

···

       75
        
       

     

       76
        
       	// Setup consumers

     

       77
        
       	aggregatorConsumer := jetstream.NewAggregatorEventConsumer(aggregatorRepo)

     

       78
       +
       	postConsumer := jetstream.NewPostEventConsumer(postRepo, communityRepo, userService, db)

     

       79
        
       

     

       80
        
       	// Setup HTTP handlers

     

       81
        
       	getServicesHandler := aggregator.NewGetServicesHandler(aggregatorService)

+5 -5

tests/integration/post_e2e_test.go

···

       131
        
       		}

     

       132
        
       

     

       133
        
       		// STEP 3: Process event through Jetstream consumer

     

       134
       -
       		consumer := jetstream.NewPostEventConsumer(postRepo, communityRepo, userService)

     

       135
        
       		err := consumer.HandleEvent(ctx, &jetstreamEvent)

     

       136
        
       		if err != nil {

     

       137
        
       			t.Fatalf("Jetstream consumer failed to process event: %v", err)

     
···

       201
        
       			},

     

       202
        
       		}

     

       203
        
       

     

       204
       -
       		consumer := jetstream.NewPostEventConsumer(postRepo, communityRepo, userService)

     

       205
        
       		err := consumer.HandleEvent(ctx, &maliciousEvent)

     

       206
        
       

     

       207
        
       		// Should get security error

     
···

       241
        
       			},

     

       242
        
       		}

     

       243
        
       

     

       244
       -
       		consumer := jetstream.NewPostEventConsumer(postRepo, communityRepo, userService)

     

       245
        
       

     

       246
        
       		// First event - should succeed

     

       247
        
       		err := consumer.HandleEvent(ctx, &event)

     
···

       295
        
       			},

     

       296
        
       		}

     

       297
        
       

     

       298
       -
       		consumer := jetstream.NewPostEventConsumer(postRepo, communityRepo, userService)

     

       299
        
       

     

       300
        
       		// Should log warning but NOT fail (eventual consistency)

     

       301
        
       		// Note: This will fail due to foreign key constraint in current schema

     
···

       530
        
       			userService := users.NewUserService(userRepo, identityResolver, pdsURL)

     

       531
        
       

     

       532
        
       			// Create post consumer (same as main.go)

     

       533
       -
       			postConsumer := jetstream.NewPostEventConsumer(postRepo, communityRepo, userService)

     

       534
        
       

     

       535
        
       			// Channels to receive the event

     

       536
        
       			eventChan := make(chan *jetstream.JetstreamEvent, 10)

···

       131
        
       		}

     

       132
        
       

     

       133
        
       		// STEP 3: Process event through Jetstream consumer

     

       134
       +
       		consumer := jetstream.NewPostEventConsumer(postRepo, communityRepo, userService, db)

     

       135
        
       		err := consumer.HandleEvent(ctx, &jetstreamEvent)

     

       136
        
       		if err != nil {

     

       137
        
       			t.Fatalf("Jetstream consumer failed to process event: %v", err)

     
···

       201
        
       			},

     

       202
        
       		}

     

       203
        
       

     

       204
       +
       		consumer := jetstream.NewPostEventConsumer(postRepo, communityRepo, userService, db)

     

       205
        
       		err := consumer.HandleEvent(ctx, &maliciousEvent)

     

       206
        
       

     

       207
        
       		// Should get security error

     
···

       241
        
       			},

     

       242
        
       		}

     

       243
        
       

     

       244
       +
       		consumer := jetstream.NewPostEventConsumer(postRepo, communityRepo, userService, db)

     

       245
        
       

     

       246
        
       		// First event - should succeed

     

       247
        
       		err := consumer.HandleEvent(ctx, &event)

     
···

       295
        
       			},

     

       296
        
       		}

     

       297
        
       

     

       298
       +
       		consumer := jetstream.NewPostEventConsumer(postRepo, communityRepo, userService, db)

     

       299
        
       

     

       300
        
       		// Should log warning but NOT fail (eventual consistency)

     

       301
        
       		// Note: This will fail due to foreign key constraint in current schema

     
···

       530
        
       			userService := users.NewUserService(userRepo, identityResolver, pdsURL)

     

       531
        
       

     

       532
        
       			// Create post consumer (same as main.go)

     

       533
       +
       			postConsumer := jetstream.NewPostEventConsumer(postRepo, communityRepo, userService, db)

     

       534
        
       

     

       535
        
       			// Channels to receive the event

     

       536
        
       			eventChan := make(chan *jetstream.JetstreamEvent, 10)