fix(jetstream): handle duplicate comment events gracefully

A community based topic aggregation platform built on atproto

Use ON CONFLICT DO NOTHING for comment indexing to handle race
conditions from duplicate Jetstream events (at-least-once delivery).
This eliminates duplicate key constraint errors when the same event
is delivered multiple times.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>

bretton.dev 3 months ago 3add6604 c6f51cc4

+12

1 changed file

expand all

unified split

internal

atproto

jetstream

comment_consumer.go

+12

internal/atproto/jetstream/comment_consumer.go

··· 310 310 311 311 } else if checkErr == sql.ErrNoRows { 312 312 // Comment doesn't exist - insert new comment 313 + // Use ON CONFLICT DO NOTHING to handle race conditions gracefully 314 + // (e.g., duplicate Jetstream events from reconnections/retries) 313 315 insertQuery := ` 314 316 INSERT INTO comments ( 315 317 uri, cid, rkey, commenter_did, ··· 322 324 $9, $10, $11, $12, $13, 323 325 $14, $15 324 326 ) 327 + ON CONFLICT (uri) DO NOTHING 325 328 RETURNING id 326 329 ` 327 330 ··· 332 335 comment.Content, comment.ContentFacets, comment.Embed, comment.ContentLabels, pq.Array(comment.Langs), 333 336 comment.CreatedAt, time.Now(), 334 337 ).Scan(&commentID) 338 + if err == sql.ErrNoRows { 339 + // ON CONFLICT triggered - comment was inserted by concurrent process 340 + // This is an idempotent replay, skip gracefully 341 + log.Printf("Comment already indexed (concurrent insert): %s", comment.URI) 342 + if commitErr := tx.Commit(); commitErr != nil { 343 + return fmt.Errorf("failed to commit transaction: %w", commitErr) 344 + } 345 + return nil 346 + } 335 347 if err != nil { 336 348 return fmt.Errorf("failed to insert comment: %w", err) 337 349 }