commit e455f71d07954cb7721615a0a7ab5a953735b60d · bretton.dev/coves

+1763

tests/integration/comment_consumer_test.go

···

       1
       +
       package integration

     

       2
       +
       

     

       3
       +
       import (

     

       4
       +
       	"Coves/internal/atproto/jetstream"

     

       5
       +
       	"Coves/internal/core/comments"

     

       6
       +
       	"Coves/internal/db/postgres"

     

       7
       +
       	"context"

     

       8
       +
       	"fmt"

     

       9
       +
       	"testing"

     

       10
       +
       	"time"

     

       11
       +
       )

     

       12
       +
       

     

       13
       +
       func TestCommentConsumer_CreateComment(t *testing.T) {

     

       14
       +
       	db := setupTestDB(t)

     

       15
       +
       	defer func() {

     

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

     

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

     

       18
       +
       		}

     

       19
       +
       	}()

     

       20
       +
       

     

       21
       +
       	ctx := context.Background()

     

       22
       +
       	commentRepo := postgres.NewCommentRepository(db)

     

       23
       +
       	consumer := jetstream.NewCommentEventConsumer(commentRepo, db)

     

       24
       +
       

     

       25
       +
       	// Setup test data

     

       26
       +
       	testUser := createTestUser(t, db, "commenter.test", "did:plc:commenter123")

     

       27
       +
       	testCommunity, err := createFeedTestCommunity(db, ctx, "testcommunity", "owner.test")

     

       28
       +
       	if err != nil {

     

       29
       +
       		t.Fatalf("Failed to create test community: %v", err)

     

       30
       +
       	}

     

       31
       +
       	testPostURI := createTestPost(t, db, testCommunity, testUser.DID, "Test Post", 0, time.Now())

     

       32
       +
       

     

       33
       +
       	t.Run("Create comment on post", func(t *testing.T) {

     

       34
       +
       		rkey := generateTID()

     

       35
       +
       		uri := fmt.Sprintf("at://%s/social.coves.feed.comment/%s", testUser.DID, rkey)

     

       36
       +
       

     

       37
       +
       		// Simulate Jetstream comment create event

     

       38
       +
       		event := &jetstream.JetstreamEvent{

     

       39
       +
       			Did:  testUser.DID,

     

       40
       +
       			Kind: "commit",

     

       41
       +
       			Commit: &jetstream.CommitEvent{

     

       42
       +
       				Rev:        "test-rev",

     

       43
       +
       				Operation:  "create",

     

       44
       +
       				Collection: "social.coves.feed.comment",

     

       45
       +
       				RKey:       rkey,

     

       46
       +
       				CID:        "bafytest123",

     

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

     

       48
       +
       					"$type": "social.coves.feed.comment",

     

       49
       +
       					"content": "This is a test comment on a post!",

     

       50
       +
       					"reply": map[string]interface{}{

     

       51
       +
       						"root": map[string]interface{}{

     

       52
       +
       							"uri": testPostURI,

     

       53
       +
       							"cid": "bafypost",

     

       54
       +
       						},

     

       55
       +
       						"parent": map[string]interface{}{

     

       56
       +
       							"uri": testPostURI,

     

       57
       +
       							"cid": "bafypost",

     

       58
       +
       						},

     

       59
       +
       					},

     

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

     

       61
       +
       				},

     

       62
       +
       			},

     

       63
       +
       		}

     

       64
       +
       

     

       65
       +
       		// Handle the event

     

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

     

       67
       +
       		if err != nil {

     

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

     

       69
       +
       		}

     

       70
       +
       

     

       71
       +
       		// Verify comment was indexed

     

       72
       +
       		comment, err := commentRepo.GetByURI(ctx, uri)

     

       73
       +
       		if err != nil {

     

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

     

       75
       +
       		}

     

       76
       +
       

     

       77
       +
       		if comment.URI != uri {

     

       78
       +
       			t.Errorf("Expected URI %s, got %s", uri, comment.URI)

     

       79
       +
       		}

     

       80
       +
       

     

       81
       +
       		if comment.CommenterDID != testUser.DID {

     

       82
       +
       			t.Errorf("Expected commenter %s, got %s", testUser.DID, comment.CommenterDID)

     

       83
       +
       		}

     

       84
       +
       

     

       85
       +
       		if comment.Content != "This is a test comment on a post!" {

     

       86
       +
       			t.Errorf("Expected content 'This is a test comment on a post!', got %s", comment.Content)

     

       87
       +
       		}

     

       88
       +
       

     

       89
       +
       		if comment.RootURI != testPostURI {

     

       90
       +
       			t.Errorf("Expected root URI %s, got %s", testPostURI, comment.RootURI)

     

       91
       +
       		}

     

       92
       +
       

     

       93
       +
       		if comment.ParentURI != testPostURI {

     

       94
       +
       			t.Errorf("Expected parent URI %s, got %s", testPostURI, comment.ParentURI)

     

       95
       +
       		}

     

       96
       +
       

     

       97
       +
       		// Verify post comment count was incremented

     

       98
       +
       		var commentCount int

     

       99
       +
       		err = db.QueryRowContext(ctx, "SELECT comment_count FROM posts WHERE uri = $1", testPostURI).Scan(&commentCount)

     

       100
       +
       		if err != nil {

     

       101
       +
       			t.Fatalf("Failed to get post comment count: %v", err)

     

       102
       +
       		}

     

       103
       +
       

     

       104
       +
       		if commentCount != 1 {

     

       105
       +
       			t.Errorf("Expected post comment_count to be 1, got %d", commentCount)

     

       106
       +
       		}

     

       107
       +
       	})

     

       108
       +
       

     

       109
       +
       	t.Run("Idempotent create - duplicate event", func(t *testing.T) {

     

       110
       +
       		rkey := generateTID()

     

       111
       +
       

     

       112
       +
       		event := &jetstream.JetstreamEvent{

     

       113
       +
       			Did:  testUser.DID,

     

       114
       +
       			Kind: "commit",

     

       115
       +
       			Commit: &jetstream.CommitEvent{

     

       116
       +
       				Rev:        "test-rev",

     

       117
       +
       				Operation:  "create",

     

       118
       +
       				Collection: "social.coves.feed.comment",

     

       119
       +
       				RKey:       rkey,

     

       120
       +
       				CID:        "bafytest456",

     

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

     

       122
       +
       					"$type": "social.coves.feed.comment",

     

       123
       +
       					"content": "Idempotent test comment",

     

       124
       +
       					"reply": map[string]interface{}{

     

       125
       +
       						"root": map[string]interface{}{

     

       126
       +
       							"uri": testPostURI,

     

       127
       +
       							"cid": "bafypost",

     

       128
       +
       						},

     

       129
       +
       						"parent": map[string]interface{}{

     

       130
       +
       							"uri": testPostURI,

     

       131
       +
       							"cid": "bafypost",

     

       132
       +
       						},

     

       133
       +
       					},

     

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

     

       135
       +
       				},

     

       136
       +
       			},

     

       137
       +
       		}

     

       138
       +
       

     

       139
       +
       		// First creation

     

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

     

       141
       +
       		if err != nil {

     

       142
       +
       			t.Fatalf("First creation failed: %v", err)

     

       143
       +
       		}

     

       144
       +
       

     

       145
       +
       		// Get initial comment count

     

       146
       +
       		var initialCount int

     

       147
       +
       		err = db.QueryRowContext(ctx, "SELECT comment_count FROM posts WHERE uri = $1", testPostURI).Scan(&initialCount)

     

       148
       +
       		if err != nil {

     

       149
       +
       			t.Fatalf("Failed to get initial comment count: %v", err)

     

       150
       +
       		}

     

       151
       +
       

     

       152
       +
       		// Duplicate creation - should be idempotent

     

       153
       +
       		err = consumer.HandleEvent(ctx, event)

     

       154
       +
       		if err != nil {

     

       155
       +
       			t.Fatalf("Duplicate event should be handled gracefully: %v", err)

     

       156
       +
       		}

     

       157
       +
       

     

       158
       +
       		// Verify count wasn't incremented again

     

       159
       +
       		var finalCount int

     

       160
       +
       		err = db.QueryRowContext(ctx, "SELECT comment_count FROM posts WHERE uri = $1", testPostURI).Scan(&finalCount)

     

       161
       +
       		if err != nil {

     

       162
       +
       			t.Fatalf("Failed to get final comment count: %v", err)

     

       163
       +
       		}

     

       164
       +
       

     

       165
       +
       		if finalCount != initialCount {

     

       166
       +
       			t.Errorf("Comment count should not increase on duplicate event. Initial: %d, Final: %d", initialCount, finalCount)

     

       167
       +
       		}

     

       168
       +
       	})

     

       169
       +
       }

     

       170
       +
       

     

       171
       +
       func TestCommentConsumer_Threading(t *testing.T) {

     

       172
       +
       	db := setupTestDB(t)

     

       173
       +
       	defer func() {

     

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

     

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

     

       176
       +
       		}

     

       177
       +
       	}()

     

       178
       +
       

     

       179
       +
       	ctx := context.Background()

     

       180
       +
       	commentRepo := postgres.NewCommentRepository(db)

     

       181
       +
       	consumer := jetstream.NewCommentEventConsumer(commentRepo, db)

     

       182
       +
       

     

       183
       +
       	// Setup test data

     

       184
       +
       	testUser := createTestUser(t, db, "threader.test", "did:plc:threader123")

     

       185
       +
       	testCommunity, err := createFeedTestCommunity(db, ctx, "threadcommunity", "owner2.test")

     

       186
       +
       	if err != nil {

     

       187
       +
       		t.Fatalf("Failed to create test community: %v", err)

     

       188
       +
       	}

     

       189
       +
       	testPostURI := createTestPost(t, db, testCommunity, testUser.DID, "Threading Test", 0, time.Now())

     

       190
       +
       

     

       191
       +
       	t.Run("Create nested comment replies", func(t *testing.T) {

     

       192
       +
       		// Create first-level comment on post

     

       193
       +
       		comment1Rkey := generateTID()

     

       194
       +
       		comment1URI := fmt.Sprintf("at://%s/social.coves.feed.comment/%s", testUser.DID, comment1Rkey)

     

       195
       +
       

     

       196
       +
       		event1 := &jetstream.JetstreamEvent{

     

       197
       +
       			Did:  testUser.DID,

     

       198
       +
       			Kind: "commit",

     

       199
       +
       			Commit: &jetstream.CommitEvent{

     

       200
       +
       				Operation:  "create",

     

       201
       +
       				Collection: "social.coves.feed.comment",

     

       202
       +
       				RKey:       comment1Rkey,

     

       203
       +
       				CID:        "bafycomment1",

     

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

     

       205
       +
       					"content": "First level comment",

     

       206
       +
       					"reply": map[string]interface{}{

     

       207
       +
       						"root": map[string]interface{}{

     

       208
       +
       							"uri": testPostURI,

     

       209
       +
       							"cid": "bafypost",

     

       210
       +
       						},

     

       211
       +
       						"parent": map[string]interface{}{

     

       212
       +
       							"uri": testPostURI,

     

       213
       +
       							"cid": "bafypost",

     

       214
       +
       						},

     

       215
       +
       					},

     

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

     

       217
       +
       				},

     

       218
       +
       			},

     

       219
       +
       		}

     

       220
       +
       

     

       221
       +
       		err := consumer.HandleEvent(ctx, event1)

     

       222
       +
       		if err != nil {

     

       223
       +
       			t.Fatalf("Failed to create first-level comment: %v", err)

     

       224
       +
       		}

     

       225
       +
       

     

       226
       +
       		// Create second-level comment (reply to first comment)

     

       227
       +
       		comment2Rkey := generateTID()

     

       228
       +
       		comment2URI := fmt.Sprintf("at://%s/social.coves.feed.comment/%s", testUser.DID, comment2Rkey)

     

       229
       +
       

     

       230
       +
       		event2 := &jetstream.JetstreamEvent{

     

       231
       +
       			Did:  testUser.DID,

     

       232
       +
       			Kind: "commit",

     

       233
       +
       			Commit: &jetstream.CommitEvent{

     

       234
       +
       				Operation:  "create",

     

       235
       +
       				Collection: "social.coves.feed.comment",

     

       236
       +
       				RKey:       comment2Rkey,

     

       237
       +
       				CID:        "bafycomment2",

     

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

     

       239
       +
       					"content": "Second level comment (reply to first)",

     

       240
       +
       					"reply": map[string]interface{}{

     

       241
       +
       						"root": map[string]interface{}{

     

       242
       +
       							"uri": testPostURI,

     

       243
       +
       							"cid": "bafypost",

     

       244
       +
       						},

     

       245
       +
       						"parent": map[string]interface{}{

     

       246
       +
       							"uri": comment1URI,

     

       247
       +
       							"cid": "bafycomment1",

     

       248
       +
       						},

     

       249
       +
       					},

     

       250
       +
       					"createdAt": time.Now().Add(1 * time.Second).Format(time.RFC3339),

     

       251
       +
       				},

     

       252
       +
       			},

     

       253
       +
       		}

     

       254
       +
       

     

       255
       +
       		err = consumer.HandleEvent(ctx, event2)

     

       256
       +
       		if err != nil {

     

       257
       +
       			t.Fatalf("Failed to create second-level comment: %v", err)

     

       258
       +
       		}

     

       259
       +
       

     

       260
       +
       		// Verify threading structure

     

       261
       +
       		comment1, err := commentRepo.GetByURI(ctx, comment1URI)

     

       262
       +
       		if err != nil {

     

       263
       +
       			t.Fatalf("Failed to get first comment: %v", err)

     

       264
       +
       		}

     

       265
       +
       

     

       266
       +
       		comment2, err := commentRepo.GetByURI(ctx, comment2URI)

     

       267
       +
       		if err != nil {

     

       268
       +
       			t.Fatalf("Failed to get second comment: %v", err)

     

       269
       +
       		}

     

       270
       +
       

     

       271
       +
       		// Both should have same root (original post)

     

       272
       +
       		if comment1.RootURI != testPostURI {

     

       273
       +
       			t.Errorf("Comment1 root should be post URI, got %s", comment1.RootURI)

     

       274
       +
       		}

     

       275
       +
       

     

       276
       +
       		if comment2.RootURI != testPostURI {

     

       277
       +
       			t.Errorf("Comment2 root should be post URI, got %s", comment2.RootURI)

     

       278
       +
       		}

     

       279
       +
       

     

       280
       +
       		// Comment1 parent should be post

     

       281
       +
       		if comment1.ParentURI != testPostURI {

     

       282
       +
       			t.Errorf("Comment1 parent should be post URI, got %s", comment1.ParentURI)

     

       283
       +
       		}

     

       284
       +
       

     

       285
       +
       		// Comment2 parent should be comment1

     

       286
       +
       		if comment2.ParentURI != comment1URI {

     

       287
       +
       			t.Errorf("Comment2 parent should be comment1 URI, got %s", comment2.ParentURI)

     

       288
       +
       		}

     

       289
       +
       

     

       290
       +
       		// Verify reply count on comment1

     

       291
       +
       		if comment1.ReplyCount != 1 {

     

       292
       +
       			t.Errorf("Comment1 should have 1 reply, got %d", comment1.ReplyCount)

     

       293
       +
       		}

     

       294
       +
       

     

       295
       +
       		// Query all comments by root

     

       296
       +
       		allComments, err := commentRepo.ListByRoot(ctx, testPostURI, 100, 0)

     

       297
       +
       		if err != nil {

     

       298
       +
       			t.Fatalf("Failed to list comments by root: %v", err)

     

       299
       +
       		}

     

       300
       +
       

     

       301
       +
       		if len(allComments) != 2 {

     

       302
       +
       			t.Errorf("Expected 2 comments in thread, got %d", len(allComments))

     

       303
       +
       		}

     

       304
       +
       

     

       305
       +
       		// Query direct replies to post

     

       306
       +
       		directReplies, err := commentRepo.ListByParent(ctx, testPostURI, 100, 0)

     

       307
       +
       		if err != nil {

     

       308
       +
       			t.Fatalf("Failed to list direct replies to post: %v", err)

     

       309
       +
       		}

     

       310
       +
       

     

       311
       +
       		if len(directReplies) != 1 {

     

       312
       +
       			t.Errorf("Expected 1 direct reply to post, got %d", len(directReplies))

     

       313
       +
       		}

     

       314
       +
       

     

       315
       +
       		// Query replies to comment1

     

       316
       +
       		comment1Replies, err := commentRepo.ListByParent(ctx, comment1URI, 100, 0)

     

       317
       +
       		if err != nil {

     

       318
       +
       			t.Fatalf("Failed to list replies to comment1: %v", err)

     

       319
       +
       		}

     

       320
       +
       

     

       321
       +
       		if len(comment1Replies) != 1 {

     

       322
       +
       			t.Errorf("Expected 1 reply to comment1, got %d", len(comment1Replies))

     

       323
       +
       		}

     

       324
       +
       	})

     

       325
       +
       }

     

       326
       +
       

     

       327
       +
       func TestCommentConsumer_UpdateComment(t *testing.T) {

     

       328
       +
       	db := setupTestDB(t)

     

       329
       +
       	defer func() {

     

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

     

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

     

       332
       +
       		}

     

       333
       +
       	}()

     

       334
       +
       

     

       335
       +
       	ctx := context.Background()

     

       336
       +
       	commentRepo := postgres.NewCommentRepository(db)

     

       337
       +
       	consumer := jetstream.NewCommentEventConsumer(commentRepo, db)

     

       338
       +
       

     

       339
       +
       	// Setup test data

     

       340
       +
       	testUser := createTestUser(t, db, "editor.test", "did:plc:editor123")

     

       341
       +
       	testCommunity, err := createFeedTestCommunity(db, ctx, "editcommunity", "owner3.test")

     

       342
       +
       	if err != nil {

     

       343
       +
       		t.Fatalf("Failed to create test community: %v", err)

     

       344
       +
       	}

     

       345
       +
       	testPostURI := createTestPost(t, db, testCommunity, testUser.DID, "Edit Test", 0, time.Now())

     

       346
       +
       

     

       347
       +
       	t.Run("Update comment content preserves vote counts", func(t *testing.T) {

     

       348
       +
       		rkey := generateTID()

     

       349
       +
       		uri := fmt.Sprintf("at://%s/social.coves.feed.comment/%s", testUser.DID, rkey)

     

       350
       +
       

     

       351
       +
       		// Create initial comment

     

       352
       +
       		createEvent := &jetstream.JetstreamEvent{

     

       353
       +
       			Did:  testUser.DID,

     

       354
       +
       			Kind: "commit",

     

       355
       +
       			Commit: &jetstream.CommitEvent{

     

       356
       +
       				Operation:  "create",

     

       357
       +
       				Collection: "social.coves.feed.comment",

     

       358
       +
       				RKey:       rkey,

     

       359
       +
       				CID:        "bafyoriginal",

     

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

     

       361
       +
       					"content": "Original comment content",

     

       362
       +
       					"reply": map[string]interface{}{

     

       363
       +
       						"root": map[string]interface{}{

     

       364
       +
       							"uri": testPostURI,

     

       365
       +
       							"cid": "bafypost",

     

       366
       +
       						},

     

       367
       +
       						"parent": map[string]interface{}{

     

       368
       +
       							"uri": testPostURI,

     

       369
       +
       							"cid": "bafypost",

     

       370
       +
       						},

     

       371
       +
       					},

     

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

     

       373
       +
       				},

     

       374
       +
       			},

     

       375
       +
       		}

     

       376
       +
       

     

       377
       +
       		err := consumer.HandleEvent(ctx, createEvent)

     

       378
       +
       		if err != nil {

     

       379
       +
       			t.Fatalf("Failed to create comment: %v", err)

     

       380
       +
       		}

     

       381
       +
       

     

       382
       +
       		// Manually set vote counts to simulate votes

     

       383
       +
       		_, err = db.ExecContext(ctx, `

     

       384
       +
       			UPDATE comments

     

       385
       +
       			SET upvote_count = 5, downvote_count = 2, score = 3

     

       386
       +
       			WHERE uri = $1

     

       387
       +
       		`, uri)

     

       388
       +
       		if err != nil {

     

       389
       +
       			t.Fatalf("Failed to set vote counts: %v", err)

     

       390
       +
       		}

     

       391
       +
       

     

       392
       +
       		// Update the comment

     

       393
       +
       		updateEvent := &jetstream.JetstreamEvent{

     

       394
       +
       			Did:  testUser.DID,

     

       395
       +
       			Kind: "commit",

     

       396
       +
       			Commit: &jetstream.CommitEvent{

     

       397
       +
       				Operation:  "update",

     

       398
       +
       				Collection: "social.coves.feed.comment",

     

       399
       +
       				RKey:       rkey,

     

       400
       +
       				CID:        "bafyupdated",

     

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

     

       402
       +
       					"content": "EDITED: Updated comment content",

     

       403
       +
       					"reply": map[string]interface{}{

     

       404
       +
       						"root": map[string]interface{}{

     

       405
       +
       							"uri": testPostURI,

     

       406
       +
       							"cid": "bafypost",

     

       407
       +
       						},

     

       408
       +
       						"parent": map[string]interface{}{

     

       409
       +
       							"uri": testPostURI,

     

       410
       +
       							"cid": "bafypost",

     

       411
       +
       						},

     

       412
       +
       					},

     

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

     

       414
       +
       				},

     

       415
       +
       			},

     

       416
       +
       		}

     

       417
       +
       

     

       418
       +
       		err = consumer.HandleEvent(ctx, updateEvent)

     

       419
       +
       		if err != nil {

     

       420
       +
       			t.Fatalf("Failed to update comment: %v", err)

     

       421
       +
       		}

     

       422
       +
       

     

       423
       +
       		// Verify content updated

     

       424
       +
       		comment, err := commentRepo.GetByURI(ctx, uri)

     

       425
       +
       		if err != nil {

     

       426
       +
       			t.Fatalf("Failed to get updated comment: %v", err)

     

       427
       +
       		}

     

       428
       +
       

     

       429
       +
       		if comment.Content != "EDITED: Updated comment content" {

     

       430
       +
       			t.Errorf("Expected updated content, got %s", comment.Content)

     

       431
       +
       		}

     

       432
       +
       

     

       433
       +
       		// Verify CID updated

     

       434
       +
       		if comment.CID != "bafyupdated" {

     

       435
       +
       			t.Errorf("Expected CID to be updated to bafyupdated, got %s", comment.CID)

     

       436
       +
       		}

     

       437
       +
       

     

       438
       +
       		// Verify vote counts preserved

     

       439
       +
       		if comment.UpvoteCount != 5 {

     

       440
       +
       			t.Errorf("Expected upvote_count preserved at 5, got %d", comment.UpvoteCount)

     

       441
       +
       		}

     

       442
       +
       

     

       443
       +
       		if comment.DownvoteCount != 2 {

     

       444
       +
       			t.Errorf("Expected downvote_count preserved at 2, got %d", comment.DownvoteCount)

     

       445
       +
       		}

     

       446
       +
       

     

       447
       +
       		if comment.Score != 3 {

     

       448
       +
       			t.Errorf("Expected score preserved at 3, got %d", comment.Score)

     

       449
       +
       		}

     

       450
       +
       	})

     

       451
       +
       }

     

       452
       +
       

     

       453
       +
       func TestCommentConsumer_DeleteComment(t *testing.T) {

     

       454
       +
       	db := setupTestDB(t)

     

       455
       +
       	defer func() {

     

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

     

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

     

       458
       +
       		}

     

       459
       +
       	}()

     

       460
       +
       

     

       461
       +
       	ctx := context.Background()

     

       462
       +
       	commentRepo := postgres.NewCommentRepository(db)

     

       463
       +
       	consumer := jetstream.NewCommentEventConsumer(commentRepo, db)

     

       464
       +
       

     

       465
       +
       	// Setup test data

     

       466
       +
       	testUser := createTestUser(t, db, "deleter.test", "did:plc:deleter123")

     

       467
       +
       	testCommunity, err := createFeedTestCommunity(db, ctx, "deletecommunity", "owner4.test")

     

       468
       +
       	if err != nil {

     

       469
       +
       		t.Fatalf("Failed to create test community: %v", err)

     

       470
       +
       	}

     

       471
       +
       	testPostURI := createTestPost(t, db, testCommunity, testUser.DID, "Delete Test", 0, time.Now())

     

       472
       +
       

     

       473
       +
       	t.Run("Delete comment decrements parent count", func(t *testing.T) {

     

       474
       +
       		rkey := generateTID()

     

       475
       +
       		uri := fmt.Sprintf("at://%s/social.coves.feed.comment/%s", testUser.DID, rkey)

     

       476
       +
       

     

       477
       +
       		// Create comment

     

       478
       +
       		createEvent := &jetstream.JetstreamEvent{

     

       479
       +
       			Did:  testUser.DID,

     

       480
       +
       			Kind: "commit",

     

       481
       +
       			Commit: &jetstream.CommitEvent{

     

       482
       +
       				Operation:  "create",

     

       483
       +
       				Collection: "social.coves.feed.comment",

     

       484
       +
       				RKey:       rkey,

     

       485
       +
       				CID:        "bafydelete",

     

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

     

       487
       +
       					"content": "Comment to be deleted",

     

       488
       +
       					"reply": map[string]interface{}{

     

       489
       +
       						"root": map[string]interface{}{

     

       490
       +
       							"uri": testPostURI,

     

       491
       +
       							"cid": "bafypost",

     

       492
       +
       						},

     

       493
       +
       						"parent": map[string]interface{}{

     

       494
       +
       							"uri": testPostURI,

     

       495
       +
       							"cid": "bafypost",

     

       496
       +
       						},

     

       497
       +
       					},

     

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

     

       499
       +
       				},

     

       500
       +
       			},

     

       501
       +
       		}

     

       502
       +
       

     

       503
       +
       		err := consumer.HandleEvent(ctx, createEvent)

     

       504
       +
       		if err != nil {

     

       505
       +
       			t.Fatalf("Failed to create comment: %v", err)

     

       506
       +
       		}

     

       507
       +
       

     

       508
       +
       		// Get initial post comment count

     

       509
       +
       		var initialCount int

     

       510
       +
       		err = db.QueryRowContext(ctx, "SELECT comment_count FROM posts WHERE uri = $1", testPostURI).Scan(&initialCount)

     

       511
       +
       		if err != nil {

     

       512
       +
       			t.Fatalf("Failed to get initial comment count: %v", err)

     

       513
       +
       		}

     

       514
       +
       

     

       515
       +
       		// Delete comment

     

       516
       +
       		deleteEvent := &jetstream.JetstreamEvent{

     

       517
       +
       			Did:  testUser.DID,

     

       518
       +
       			Kind: "commit",

     

       519
       +
       			Commit: &jetstream.CommitEvent{

     

       520
       +
       				Operation:  "delete",

     

       521
       +
       				Collection: "social.coves.feed.comment",

     

       522
       +
       				RKey:       rkey,

     

       523
       +
       			},

     

       524
       +
       		}

     

       525
       +
       

     

       526
       +
       		err = consumer.HandleEvent(ctx, deleteEvent)

     

       527
       +
       		if err != nil {

     

       528
       +
       			t.Fatalf("Failed to delete comment: %v", err)

     

       529
       +
       		}

     

       530
       +
       

     

       531
       +
       		// Verify soft delete

     

       532
       +
       		comment, err := commentRepo.GetByURI(ctx, uri)

     

       533
       +
       		if err != nil {

     

       534
       +
       			t.Fatalf("Failed to get deleted comment: %v", err)

     

       535
       +
       		}

     

       536
       +
       

     

       537
       +
       		if comment.DeletedAt == nil {

     

       538
       +
       			t.Error("Expected deleted_at to be set, got nil")

     

       539
       +
       		}

     

       540
       +
       

     

       541
       +
       		// Verify post comment count decremented

     

       542
       +
       		var finalCount int

     

       543
       +
       		err = db.QueryRowContext(ctx, "SELECT comment_count FROM posts WHERE uri = $1", testPostURI).Scan(&finalCount)

     

       544
       +
       		if err != nil {

     

       545
       +
       			t.Fatalf("Failed to get final comment count: %v", err)

     

       546
       +
       		}

     

       547
       +
       

     

       548
       +
       		if finalCount != initialCount-1 {

     

       549
       +
       			t.Errorf("Expected comment count to decrease by 1. Initial: %d, Final: %d", initialCount, finalCount)

     

       550
       +
       		}

     

       551
       +
       	})

     

       552
       +
       

     

       553
       +
       	t.Run("Delete is idempotent", func(t *testing.T) {

     

       554
       +
       		rkey := generateTID()

     

       555
       +
       

     

       556
       +
       		// Create comment

     

       557
       +
       		createEvent := &jetstream.JetstreamEvent{

     

       558
       +
       			Did:  testUser.DID,

     

       559
       +
       			Kind: "commit",

     

       560
       +
       			Commit: &jetstream.CommitEvent{

     

       561
       +
       				Operation:  "create",

     

       562
       +
       				Collection: "social.coves.feed.comment",

     

       563
       +
       				RKey:       rkey,

     

       564
       +
       				CID:        "bafyidempdelete",

     

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

     

       566
       +
       					"content": "Idempotent delete test",

     

       567
       +
       					"reply": map[string]interface{}{

     

       568
       +
       						"root": map[string]interface{}{

     

       569
       +
       							"uri": testPostURI,

     

       570
       +
       							"cid": "bafypost",

     

       571
       +
       						},

     

       572
       +
       						"parent": map[string]interface{}{

     

       573
       +
       							"uri": testPostURI,

     

       574
       +
       							"cid": "bafypost",

     

       575
       +
       						},

     

       576
       +
       					},

     

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

     

       578
       +
       				},

     

       579
       +
       			},

     

       580
       +
       		}

     

       581
       +
       

     

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

     

       583
       +
       		if err != nil {

     

       584
       +
       			t.Fatalf("Failed to create comment: %v", err)

     

       585
       +
       		}

     

       586
       +
       

     

       587
       +
       		// First delete

     

       588
       +
       		deleteEvent := &jetstream.JetstreamEvent{

     

       589
       +
       			Did:  testUser.DID,

     

       590
       +
       			Kind: "commit",

     

       591
       +
       			Commit: &jetstream.CommitEvent{

     

       592
       +
       				Operation:  "delete",

     

       593
       +
       				Collection: "social.coves.feed.comment",

     

       594
       +
       				RKey:       rkey,

     

       595
       +
       			},

     

       596
       +
       		}

     

       597
       +
       

     

       598
       +
       		err = consumer.HandleEvent(ctx, deleteEvent)

     

       599
       +
       		if err != nil {

     

       600
       +
       			t.Fatalf("First delete failed: %v", err)

     

       601
       +
       		}

     

       602
       +
       

     

       603
       +
       		// Get count after first delete

     

       604
       +
       		var countAfterFirstDelete int

     

       605
       +
       		err = db.QueryRowContext(ctx, "SELECT comment_count FROM posts WHERE uri = $1", testPostURI).Scan(&countAfterFirstDelete)

     

       606
       +
       		if err != nil {

     

       607
       +
       			t.Fatalf("Failed to get count after first delete: %v", err)

     

       608
       +
       		}

     

       609
       +
       

     

       610
       +
       		// Second delete (idempotent)

     

       611
       +
       		err = consumer.HandleEvent(ctx, deleteEvent)

     

       612
       +
       		if err != nil {

     

       613
       +
       			t.Fatalf("Second delete should be idempotent: %v", err)

     

       614
       +
       		}

     

       615
       +
       

     

       616
       +
       		// Verify count didn't change

     

       617
       +
       		var countAfterSecondDelete int

     

       618
       +
       		err = db.QueryRowContext(ctx, "SELECT comment_count FROM posts WHERE uri = $1", testPostURI).Scan(&countAfterSecondDelete)

     

       619
       +
       		if err != nil {

     

       620
       +
       			t.Fatalf("Failed to get count after second delete: %v", err)

     

       621
       +
       		}

     

       622
       +
       

     

       623
       +
       		if countAfterSecondDelete != countAfterFirstDelete {

     

       624
       +
       			t.Errorf("Count should not change on duplicate delete. After first: %d, After second: %d", countAfterFirstDelete, countAfterSecondDelete)

     

       625
       +
       		}

     

       626
       +
       	})

     

       627
       +
       }

     

       628
       +
       

     

       629
       +
       func TestCommentConsumer_SecurityValidation(t *testing.T) {

     

       630
       +
       	db := setupTestDB(t)

     

       631
       +
       	defer func() {

     

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

     

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

     

       634
       +
       		}

     

       635
       +
       	}()

     

       636
       +
       

     

       637
       +
       	ctx := context.Background()

     

       638
       +
       	commentRepo := postgres.NewCommentRepository(db)

     

       639
       +
       	consumer := jetstream.NewCommentEventConsumer(commentRepo, db)

     

       640
       +
       

     

       641
       +
       	testUser := createTestUser(t, db, "security.test", "did:plc:security123")

     

       642
       +
       	testCommunity, err := createFeedTestCommunity(db, ctx, "seccommunity", "owner5.test")

     

       643
       +
       	if err != nil {

     

       644
       +
       		t.Fatalf("Failed to create test community: %v", err)

     

       645
       +
       	}

     

       646
       +
       	testPostURI := createTestPost(t, db, testCommunity, testUser.DID, "Security Test", 0, time.Now())

     

       647
       +
       

     

       648
       +
       	t.Run("Reject comment with empty content", func(t *testing.T) {

     

       649
       +
       		event := &jetstream.JetstreamEvent{

     

       650
       +
       			Did:  testUser.DID,

     

       651
       +
       			Kind: "commit",

     

       652
       +
       			Commit: &jetstream.CommitEvent{

     

       653
       +
       				Operation:  "create",

     

       654
       +
       				Collection: "social.coves.feed.comment",

     

       655
       +
       				RKey:       generateTID(),

     

       656
       +
       				CID:        "bafyinvalid",

     

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

     

       658
       +
       					"content": "", // Empty content

     

       659
       +
       					"reply": map[string]interface{}{

     

       660
       +
       						"root": map[string]interface{}{

     

       661
       +
       							"uri": testPostURI,

     

       662
       +
       							"cid": "bafypost",

     

       663
       +
       						},

     

       664
       +
       						"parent": map[string]interface{}{

     

       665
       +
       							"uri": testPostURI,

     

       666
       +
       							"cid": "bafypost",

     

       667
       +
       						},

     

       668
       +
       					},

     

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

     

       670
       +
       				},

     

       671
       +
       			},

     

       672
       +
       		}

     

       673
       +
       

     

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

     

       675
       +
       		if err == nil {

     

       676
       +
       			t.Error("Expected error for empty content, got nil")

     

       677
       +
       		}

     

       678
       +
       	})

     

       679
       +
       

     

       680
       +
       	t.Run("Reject comment with invalid root reference", func(t *testing.T) {

     

       681
       +
       		event := &jetstream.JetstreamEvent{

     

       682
       +
       			Did:  testUser.DID,

     

       683
       +
       			Kind: "commit",

     

       684
       +
       			Commit: &jetstream.CommitEvent{

     

       685
       +
       				Operation:  "create",

     

       686
       +
       				Collection: "social.coves.feed.comment",

     

       687
       +
       				RKey:       generateTID(),

     

       688
       +
       				CID:        "bafyinvalid2",

     

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

     

       690
       +
       					"content": "Valid content",

     

       691
       +
       					"reply": map[string]interface{}{

     

       692
       +
       						"root": map[string]interface{}{

     

       693
       +
       							"uri": "", // Missing URI

     

       694
       +
       							"cid": "bafypost",

     

       695
       +
       						},

     

       696
       +
       						"parent": map[string]interface{}{

     

       697
       +
       							"uri": testPostURI,

     

       698
       +
       							"cid": "bafypost",

     

       699
       +
       						},

     

       700
       +
       					},

     

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

     

       702
       +
       				},

     

       703
       +
       			},

     

       704
       +
       		}

     

       705
       +
       

     

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

     

       707
       +
       		if err == nil {

     

       708
       +
       			t.Error("Expected error for invalid root reference, got nil")

     

       709
       +
       		}

     

       710
       +
       	})

     

       711
       +
       

     

       712
       +
       	t.Run("Reject comment with invalid parent reference", func(t *testing.T) {

     

       713
       +
       		event := &jetstream.JetstreamEvent{

     

       714
       +
       			Did:  testUser.DID,

     

       715
       +
       			Kind: "commit",

     

       716
       +
       			Commit: &jetstream.CommitEvent{

     

       717
       +
       				Operation:  "create",

     

       718
       +
       				Collection: "social.coves.feed.comment",

     

       719
       +
       				RKey:       generateTID(),

     

       720
       +
       				CID:        "bafyinvalid3",

     

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

     

       722
       +
       					"content": "Valid content",

     

       723
       +
       					"reply": map[string]interface{}{

     

       724
       +
       						"root": map[string]interface{}{

     

       725
       +
       							"uri": testPostURI,

     

       726
       +
       							"cid": "bafypost",

     

       727
       +
       						},

     

       728
       +
       						"parent": map[string]interface{}{

     

       729
       +
       							"uri": testPostURI,

     

       730
       +
       							"cid": "", // Missing CID

     

       731
       +
       						},

     

       732
       +
       					},

     

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

     

       734
       +
       				},

     

       735
       +
       			},

     

       736
       +
       		}

     

       737
       +
       

     

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

     

       739
       +
       		if err == nil {

     

       740
       +
       			t.Error("Expected error for invalid parent reference, got nil")

     

       741
       +
       		}

     

       742
       +
       	})

     

       743
       +
       

     

       744
       +
       	t.Run("Reject comment with invalid DID format", func(t *testing.T) {

     

       745
       +
       		event := &jetstream.JetstreamEvent{

     

       746
       +
       			Did:  "invalid-did-format", // Bad DID

     

       747
       +
       			Kind: "commit",

     

       748
       +
       			Commit: &jetstream.CommitEvent{

     

       749
       +
       				Operation:  "create",

     

       750
       +
       				Collection: "social.coves.feed.comment",

     

       751
       +
       				RKey:       generateTID(),

     

       752
       +
       				CID:        "bafyinvalid4",

     

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

     

       754
       +
       					"content": "Valid content",

     

       755
       +
       					"reply": map[string]interface{}{

     

       756
       +
       						"root": map[string]interface{}{

     

       757
       +
       							"uri": testPostURI,

     

       758
       +
       							"cid": "bafypost",

     

       759
       +
       						},

     

       760
       +
       						"parent": map[string]interface{}{

     

       761
       +
       							"uri": testPostURI,

     

       762
       +
       							"cid": "bafypost",

     

       763
       +
       						},

     

       764
       +
       					},

     

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

     

       766
       +
       				},

     

       767
       +
       			},

     

       768
       +
       		}

     

       769
       +
       

     

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

     

       771
       +
       		if err == nil {

     

       772
       +
       			t.Error("Expected error for invalid DID format, got nil")

     

       773
       +
       		}

     

       774
       +
       	})

     

       775
       +
       

     

       776
       +
       	t.Run("Reject comment exceeding max content length", func(t *testing.T) {

     

       777
       +
       		event := &jetstream.JetstreamEvent{

     

       778
       +
       			Did:  testUser.DID,

     

       779
       +
       			Kind: "commit",

     

       780
       +
       			Commit: &jetstream.CommitEvent{

     

       781
       +
       				Operation:  "create",

     

       782
       +
       				Collection: "social.coves.feed.comment",

     

       783
       +
       				RKey:       generateTID(),

     

       784
       +
       				CID:        "bafytoobig",

     

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

     

       786
       +
       					"content": string(make([]byte, 30001)), // Exceeds 30000 byte limit

     

       787
       +
       					"reply": map[string]interface{}{

     

       788
       +
       						"root": map[string]interface{}{

     

       789
       +
       							"uri": testPostURI,

     

       790
       +
       							"cid": "bafypost",

     

       791
       +
       						},

     

       792
       +
       						"parent": map[string]interface{}{

     

       793
       +
       							"uri": testPostURI,

     

       794
       +
       							"cid": "bafypost",

     

       795
       +
       						},

     

       796
       +
       					},

     

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

     

       798
       +
       				},

     

       799
       +
       			},

     

       800
       +
       		}

     

       801
       +
       

     

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

     

       803
       +
       		if err == nil {

     

       804
       +
       			t.Error("Expected error for oversized content, got nil")

     

       805
       +
       		}

     

       806
       +
       		if err != nil && !contains(err.Error(), "exceeds maximum length") {

     

       807
       +
       			t.Errorf("Expected 'exceeds maximum length' error, got: %v", err)

     

       808
       +
       		}

     

       809
       +
       	})

     

       810
       +
       

     

       811
       +
       	t.Run("Reject comment with malformed parent URI", func(t *testing.T) {

     

       812
       +
       		event := &jetstream.JetstreamEvent{

     

       813
       +
       			Did:  testUser.DID,

     

       814
       +
       			Kind: "commit",

     

       815
       +
       			Commit: &jetstream.CommitEvent{

     

       816
       +
       				Operation:  "create",

     

       817
       +
       				Collection: "social.coves.feed.comment",

     

       818
       +
       				RKey:       generateTID(),

     

       819
       +
       				CID:        "bafymalformed",

     

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

     

       821
       +
       					"content": "Valid content",

     

       822
       +
       					"reply": map[string]interface{}{

     

       823
       +
       						"root": map[string]interface{}{

     

       824
       +
       							"uri": testPostURI,

     

       825
       +
       							"cid": "bafypost",

     

       826
       +
       						},

     

       827
       +
       						"parent": map[string]interface{}{

     

       828
       +
       							"uri": "at://malformed", // Invalid: missing collection/rkey

     

       829
       +
       							"cid": "bafyparent",

     

       830
       +
       						},

     

       831
       +
       					},

     

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

     

       833
       +
       				},

     

       834
       +
       			},

     

       835
       +
       		}

     

       836
       +
       

     

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

     

       838
       +
       		if err == nil {

     

       839
       +
       			t.Error("Expected error for malformed AT-URI, got nil")

     

       840
       +
       		}

     

       841
       +
       		if err != nil && !contains(err.Error(), "invalid parent URI") {

     

       842
       +
       			t.Errorf("Expected 'invalid parent URI' error, got: %v", err)

     

       843
       +
       		}

     

       844
       +
       	})

     

       845
       +
       

     

       846
       +
       	t.Run("Reject comment with malformed root URI", func(t *testing.T) {

     

       847
       +
       		event := &jetstream.JetstreamEvent{

     

       848
       +
       			Did:  testUser.DID,

     

       849
       +
       			Kind: "commit",

     

       850
       +
       			Commit: &jetstream.CommitEvent{

     

       851
       +
       				Operation:  "create",

     

       852
       +
       				Collection: "social.coves.feed.comment",

     

       853
       +
       				RKey:       generateTID(),

     

       854
       +
       				CID:        "bafymalformed2",

     

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

     

       856
       +
       					"content": "Valid content",

     

       857
       +
       					"reply": map[string]interface{}{

     

       858
       +
       						"root": map[string]interface{}{

     

       859
       +
       							"uri": "at://did:plc:test123", // Invalid: missing collection/rkey

     

       860
       +
       							"cid": "bafyroot",

     

       861
       +
       						},

     

       862
       +
       						"parent": map[string]interface{}{

     

       863
       +
       							"uri": testPostURI,

     

       864
       +
       							"cid": "bafyparent",

     

       865
       +
       						},

     

       866
       +
       					},

     

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

     

       868
       +
       				},

     

       869
       +
       			},

     

       870
       +
       		}

     

       871
       +
       

     

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

     

       873
       +
       		if err == nil {

     

       874
       +
       			t.Error("Expected error for malformed AT-URI, got nil")

     

       875
       +
       		}

     

       876
       +
       		if err != nil && !contains(err.Error(), "invalid root URI") {

     

       877
       +
       			t.Errorf("Expected 'invalid root URI' error, got: %v", err)

     

       878
       +
       		}

     

       879
       +
       	})

     

       880
       +
       }

     

       881
       +
       

     

       882
       +
       func TestCommentRepository_Queries(t *testing.T) {

     

       883
       +
       	db := setupTestDB(t)

     

       884
       +
       	defer func() {

     

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

     

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

     

       887
       +
       		}

     

       888
       +
       	}()

     

       889
       +
       

     

       890
       +
       	ctx := context.Background()

     

       891
       +
       	commentRepo := postgres.NewCommentRepository(db)

     

       892
       +
       

     

       893
       +
       	// Clean up any existing test data from previous runs

     

       894
       +
       	_, err := db.ExecContext(ctx, "DELETE FROM comments WHERE commenter_did LIKE 'did:plc:%'")

     

       895
       +
       	if err != nil {

     

       896
       +
       		t.Fatalf("Failed to clean up test comments: %v", err)

     

       897
       +
       	}

     

       898
       +
       

     

       899
       +
       	testUser := createTestUser(t, db, "query.test", "did:plc:query123")

     

       900
       +
       	testCommunity, err := createFeedTestCommunity(db, ctx, "querycommunity", "owner6.test")

     

       901
       +
       	if err != nil {

     

       902
       +
       		t.Fatalf("Failed to create test community: %v", err)

     

       903
       +
       	}

     

       904
       +
       	postURI := createTestPost(t, db, testCommunity, testUser.DID, "Query Test", 0, time.Now())

     

       905
       +
       

     

       906
       +
       	// Create a comment tree

     

       907
       +
       	// Post

     

       908
       +
       	//  |- Comment 1

     

       909
       +
       	//      |- Comment 2

     

       910
       +
       	//      |- Comment 3

     

       911
       +
       	//  |- Comment 4

     

       912
       +
       

     

       913
       +
       	comment1 := &comments.Comment{

     

       914
       +
       		URI:          fmt.Sprintf("at://%s/social.coves.feed.comment/1", testUser.DID),

     

       915
       +
       		CID:          "bafyc1",

     

       916
       +
       		RKey:         "1",

     

       917
       +
       		CommenterDID: testUser.DID,

     

       918
       +
       		RootURI:      postURI,

     

       919
       +
       		RootCID:      "bafypost",

     

       920
       +
       		ParentURI:    postURI,

     

       921
       +
       		ParentCID:    "bafypost",

     

       922
       +
       		Content:      "Comment 1",

     

       923
       +
       		Langs:        []string{},

     

       924
       +
       		CreatedAt:    time.Now(),

     

       925
       +
       	}

     

       926
       +
       

     

       927
       +
       	comment2 := &comments.Comment{

     

       928
       +
       		URI:          fmt.Sprintf("at://%s/social.coves.feed.comment/2", testUser.DID),

     

       929
       +
       		CID:          "bafyc2",

     

       930
       +
       		RKey:         "2",

     

       931
       +
       		CommenterDID: testUser.DID,

     

       932
       +
       		RootURI:      postURI,

     

       933
       +
       		RootCID:      "bafypost",

     

       934
       +
       		ParentURI:    comment1.URI,

     

       935
       +
       		ParentCID:    "bafyc1",

     

       936
       +
       		Content:      "Comment 2 (reply to 1)",

     

       937
       +
       		Langs:        []string{},

     

       938
       +
       		CreatedAt:    time.Now().Add(1 * time.Second),

     

       939
       +
       	}

     

       940
       +
       

     

       941
       +
       	comment3 := &comments.Comment{

     

       942
       +
       		URI:          fmt.Sprintf("at://%s/social.coves.feed.comment/3", testUser.DID),

     

       943
       +
       		CID:          "bafyc3",

     

       944
       +
       		RKey:         "3",

     

       945
       +
       		CommenterDID: testUser.DID,

     

       946
       +
       		RootURI:      postURI,

     

       947
       +
       		RootCID:      "bafypost",

     

       948
       +
       		ParentURI:    comment1.URI,

     

       949
       +
       		ParentCID:    "bafyc1",

     

       950
       +
       		Content:      "Comment 3 (reply to 1)",

     

       951
       +
       		Langs:        []string{},

     

       952
       +
       		CreatedAt:    time.Now().Add(2 * time.Second),

     

       953
       +
       	}

     

       954
       +
       

     

       955
       +
       	comment4 := &comments.Comment{

     

       956
       +
       		URI:          fmt.Sprintf("at://%s/social.coves.feed.comment/4", testUser.DID),

     

       957
       +
       		CID:          "bafyc4",

     

       958
       +
       		RKey:         "4",

     

       959
       +
       		CommenterDID: testUser.DID,

     

       960
       +
       		RootURI:      postURI,

     

       961
       +
       		RootCID:      "bafypost",

     

       962
       +
       		ParentURI:    postURI,

     

       963
       +
       		ParentCID:    "bafypost",

     

       964
       +
       		Content:      "Comment 4",

     

       965
       +
       		Langs:        []string{},

     

       966
       +
       		CreatedAt:    time.Now().Add(3 * time.Second),

     

       967
       +
       	}

     

       968
       +
       

     

       969
       +
       	// Create all comments

     

       970
       +
       	for i, c := range []*comments.Comment{comment1, comment2, comment3, comment4} {

     

       971
       +
       		if err := commentRepo.Create(ctx, c); err != nil {

     

       972
       +
       			t.Fatalf("Failed to create comment %d: %v", i+1, err)

     

       973
       +
       		}

     

       974
       +
       		t.Logf("Created comment %d: %s", i+1, c.URI)

     

       975
       +
       	}

     

       976
       +
       

     

       977
       +
       	// Verify comments were created

     

       978
       +
       	verifyCount, err := commentRepo.CountByParent(ctx, postURI)

     

       979
       +
       	if err != nil {

     

       980
       +
       		t.Fatalf("Failed to count comments: %v", err)

     

       981
       +
       	}

     

       982
       +
       	t.Logf("Direct replies to post after creation: %d", verifyCount)

     

       983
       +
       

     

       984
       +
       	t.Run("ListByRoot returns all comments in thread", func(t *testing.T) {

     

       985
       +
       		comments, err := commentRepo.ListByRoot(ctx, postURI, 100, 0)

     

       986
       +
       		if err != nil {

     

       987
       +
       			t.Fatalf("Failed to list by root: %v", err)

     

       988
       +
       		}

     

       989
       +
       

     

       990
       +
       		if len(comments) != 4 {

     

       991
       +
       			t.Errorf("Expected 4 comments, got %d", len(comments))

     

       992
       +
       		}

     

       993
       +
       	})

     

       994
       +
       

     

       995
       +
       	t.Run("ListByParent returns direct replies", func(t *testing.T) {

     

       996
       +
       		// Direct replies to post

     

       997
       +
       		postReplies, err := commentRepo.ListByParent(ctx, postURI, 100, 0)

     

       998
       +
       		if err != nil {

     

       999
       +
       			t.Fatalf("Failed to list post replies: %v", err)

     

       1000
       +
       		}

     

       1001
       +
       

     

       1002
       +
       		if len(postReplies) != 2 {

     

       1003
       +
       			t.Errorf("Expected 2 direct replies to post, got %d", len(postReplies))

     

       1004
       +
       		}

     

       1005
       +
       

     

       1006
       +
       		// Direct replies to comment1

     

       1007
       +
       		comment1Replies, err := commentRepo.ListByParent(ctx, comment1.URI, 100, 0)

     

       1008
       +
       		if err != nil {

     

       1009
       +
       			t.Fatalf("Failed to list comment1 replies: %v", err)

     

       1010
       +
       		}

     

       1011
       +
       

     

       1012
       +
       		if len(comment1Replies) != 2 {

     

       1013
       +
       			t.Errorf("Expected 2 direct replies to comment1, got %d", len(comment1Replies))

     

       1014
       +
       		}

     

       1015
       +
       	})

     

       1016
       +
       

     

       1017
       +
       	t.Run("CountByParent returns correct counts", func(t *testing.T) {

     

       1018
       +
       		postCount, err := commentRepo.CountByParent(ctx, postURI)

     

       1019
       +
       		if err != nil {

     

       1020
       +
       			t.Fatalf("Failed to count post replies: %v", err)

     

       1021
       +
       		}

     

       1022
       +
       

     

       1023
       +
       		if postCount != 2 {

     

       1024
       +
       			t.Errorf("Expected 2 direct replies to post, got %d", postCount)

     

       1025
       +
       		}

     

       1026
       +
       

     

       1027
       +
       		comment1Count, err := commentRepo.CountByParent(ctx, comment1.URI)

     

       1028
       +
       		if err != nil {

     

       1029
       +
       			t.Fatalf("Failed to count comment1 replies: %v", err)

     

       1030
       +
       		}

     

       1031
       +
       

     

       1032
       +
       		if comment1Count != 2 {

     

       1033
       +
       			t.Errorf("Expected 2 direct replies to comment1, got %d", comment1Count)

     

       1034
       +
       		}

     

       1035
       +
       	})

     

       1036
       +
       

     

       1037
       +
       	t.Run("ListByCommenter returns user's comments", func(t *testing.T) {

     

       1038
       +
       		userComments, err := commentRepo.ListByCommenter(ctx, testUser.DID, 100, 0)

     

       1039
       +
       		if err != nil {

     

       1040
       +
       			t.Fatalf("Failed to list by commenter: %v", err)

     

       1041
       +
       		}

     

       1042
       +
       

     

       1043
       +
       		if len(userComments) != 4 {

     

       1044
       +
       			t.Errorf("Expected 4 comments by user, got %d", len(userComments))

     

       1045
       +
       		}

     

       1046
       +
       	})

     

       1047
       +
       }

     

       1048
       +
       

     

       1049
       +
       // TestCommentConsumer_OutOfOrderReconciliation tests that parent counts are

     

       1050
       +
       // correctly reconciled when child comments arrive before their parent

     

       1051
       +
       func TestCommentConsumer_OutOfOrderReconciliation(t *testing.T) {

     

       1052
       +
       	db := setupTestDB(t)

     

       1053
       +
       	defer func() {

     

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

     

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

     

       1056
       +
       		}

     

       1057
       +
       	}()

     

       1058
       +
       

     

       1059
       +
       	ctx := context.Background()

     

       1060
       +
       	commentRepo := postgres.NewCommentRepository(db)

     

       1061
       +
       	consumer := jetstream.NewCommentEventConsumer(commentRepo, db)

     

       1062
       +
       

     

       1063
       +
       	testUser := createTestUser(t, db, "outoforder.test", "did:plc:outoforder123")

     

       1064
       +
       	testCommunity, err := createFeedTestCommunity(db, ctx, "ooo-community", "owner7.test")

     

       1065
       +
       	if err != nil {

     

       1066
       +
       		t.Fatalf("Failed to create test community: %v", err)

     

       1067
       +
       	}

     

       1068
       +
       	postURI := createTestPost(t, db, testCommunity, testUser.DID, "OOO Test Post", 0, time.Now())

     

       1069
       +
       

     

       1070
       +
       	t.Run("Child arrives before parent - count reconciled", func(t *testing.T) {

     

       1071
       +
       		// Scenario: User A creates comment C1 on post

     

       1072
       +
       		//           User B creates reply C2 to C1

     

       1073
       +
       		//           Jetstream delivers C2 before C1 (different repos)

     

       1074
       +
       		//           When C1 finally arrives, its reply_count should be 1, not 0

     

       1075
       +
       

     

       1076
       +
       		parentRkey := generateTID()

     

       1077
       +
       		parentURI := fmt.Sprintf("at://%s/social.coves.feed.comment/%s", testUser.DID, parentRkey)

     

       1078
       +
       

     

       1079
       +
       		childRkey := generateTID()

     

       1080
       +
       		childURI := fmt.Sprintf("at://%s/social.coves.feed.comment/%s", testUser.DID, childRkey)

     

       1081
       +
       

     

       1082
       +
       		// Step 1: Index child FIRST (before parent exists)

     

       1083
       +
       		childEvent := &jetstream.JetstreamEvent{

     

       1084
       +
       			Did:  testUser.DID,

     

       1085
       +
       			Kind: "commit",

     

       1086
       +
       			Commit: &jetstream.CommitEvent{

     

       1087
       +
       				Rev:        "child-rev",

     

       1088
       +
       				Operation:  "create",

     

       1089
       +
       				Collection: "social.coves.feed.comment",

     

       1090
       +
       				RKey:       childRkey,

     

       1091
       +
       				CID:        "bafychild",

     

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

     

       1093
       +
       					"$type":   "social.coves.feed.comment",

     

       1094
       +
       					"content": "This is a reply to a comment that doesn't exist yet!",

     

       1095
       +
       					"reply": map[string]interface{}{

     

       1096
       +
       						"root": map[string]interface{}{

     

       1097
       +
       							"uri": postURI,

     

       1098
       +
       							"cid": "bafypost",

     

       1099
       +
       						},

     

       1100
       +
       						"parent": map[string]interface{}{

     

       1101
       +
       							"uri": parentURI, // Points to parent that doesn't exist yet

     

       1102
       +
       							"cid": "bafyparent",

     

       1103
       +
       						},

     

       1104
       +
       					},

     

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

     

       1106
       +
       				},

     

       1107
       +
       			},

     

       1108
       +
       		}

     

       1109
       +
       

     

       1110
       +
       		err := consumer.HandleEvent(ctx, childEvent)

     

       1111
       +
       		if err != nil {

     

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

     

       1113
       +
       		}

     

       1114
       +
       

     

       1115
       +
       		// Verify child was indexed

     

       1116
       +
       		childComment, err := commentRepo.GetByURI(ctx, childURI)

     

       1117
       +
       		if err != nil {

     

       1118
       +
       			t.Fatalf("Child comment not indexed: %v", err)

     

       1119
       +
       		}

     

       1120
       +
       		if childComment.ParentURI != parentURI {

     

       1121
       +
       			t.Errorf("Expected child parent_uri %s, got %s", parentURI, childComment.ParentURI)

     

       1122
       +
       		}

     

       1123
       +
       

     

       1124
       +
       		// Step 2: Now index parent (arrives late due to Jetstream ordering)

     

       1125
       +
       		parentEvent := &jetstream.JetstreamEvent{

     

       1126
       +
       			Did:  testUser.DID,

     

       1127
       +
       			Kind: "commit",

     

       1128
       +
       			Commit: &jetstream.CommitEvent{

     

       1129
       +
       				Rev:        "parent-rev",

     

       1130
       +
       				Operation:  "create",

     

       1131
       +
       				Collection: "social.coves.feed.comment",

     

       1132
       +
       				RKey:       parentRkey,

     

       1133
       +
       				CID:        "bafyparent",

     

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

     

       1135
       +
       					"$type":   "social.coves.feed.comment",

     

       1136
       +
       					"content": "This is the parent comment arriving late",

     

       1137
       +
       					"reply": map[string]interface{}{

     

       1138
       +
       						"root": map[string]interface{}{

     

       1139
       +
       							"uri": postURI,

     

       1140
       +
       							"cid": "bafypost",

     

       1141
       +
       						},

     

       1142
       +
       						"parent": map[string]interface{}{

     

       1143
       +
       							"uri": postURI,

     

       1144
       +
       							"cid": "bafypost",

     

       1145
       +
       						},

     

       1146
       +
       					},

     

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

     

       1148
       +
       				},

     

       1149
       +
       			},

     

       1150
       +
       		}

     

       1151
       +
       

     

       1152
       +
       		err = consumer.HandleEvent(ctx, parentEvent)

     

       1153
       +
       		if err != nil {

     

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

     

       1155
       +
       		}

     

       1156
       +
       

     

       1157
       +
       		// Step 3: Verify parent was indexed with CORRECT reply_count

     

       1158
       +
       		parentComment, err := commentRepo.GetByURI(ctx, parentURI)

     

       1159
       +
       		if err != nil {

     

       1160
       +
       			t.Fatalf("Parent comment not indexed: %v", err)

     

       1161
       +
       		}

     

       1162
       +
       

     

       1163
       +
       		// THIS IS THE KEY TEST: Parent should have reply_count = 1 due to reconciliation

     

       1164
       +
       		if parentComment.ReplyCount != 1 {

     

       1165
       +
       			t.Errorf("Expected parent reply_count to be 1 (reconciled), got %d", parentComment.ReplyCount)

     

       1166
       +
       			t.Logf("This indicates out-of-order reconciliation failed!")

     

       1167
       +
       		}

     

       1168
       +
       

     

       1169
       +
       		// Verify via query as well

     

       1170
       +
       		count, err := commentRepo.CountByParent(ctx, parentURI)

     

       1171
       +
       		if err != nil {

     

       1172
       +
       			t.Fatalf("Failed to count parent replies: %v", err)

     

       1173
       +
       		}

     

       1174
       +
       		if count != 1 {

     

       1175
       +
       			t.Errorf("Expected 1 reply to parent, got %d", count)

     

       1176
       +
       		}

     

       1177
       +
       	})

     

       1178
       +
       

     

       1179
       +
       	t.Run("Multiple children arrive before parent", func(t *testing.T) {

     

       1180
       +
       		parentRkey := generateTID()

     

       1181
       +
       		parentURI := fmt.Sprintf("at://%s/social.coves.feed.comment/%s", testUser.DID, parentRkey)

     

       1182
       +
       

     

       1183
       +
       		// Index 3 children before parent

     

       1184
       +
       		for i := 1; i <= 3; i++ {

     

       1185
       +
       			childRkey := generateTID()

     

       1186
       +
       			childEvent := &jetstream.JetstreamEvent{

     

       1187
       +
       				Did:  testUser.DID,

     

       1188
       +
       				Kind: "commit",

     

       1189
       +
       				Commit: &jetstream.CommitEvent{

     

       1190
       +
       					Rev:        fmt.Sprintf("child-%d-rev", i),

     

       1191
       +
       					Operation:  "create",

     

       1192
       +
       					Collection: "social.coves.feed.comment",

     

       1193
       +
       					RKey:       childRkey,

     

       1194
       +
       					CID:        fmt.Sprintf("bafychild%d", i),

     

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

     

       1196
       +
       						"$type":   "social.coves.feed.comment",

     

       1197
       +
       						"content": fmt.Sprintf("Reply %d before parent", i),

     

       1198
       +
       						"reply": map[string]interface{}{

     

       1199
       +
       							"root": map[string]interface{}{

     

       1200
       +
       								"uri": postURI,

     

       1201
       +
       								"cid": "bafypost",

     

       1202
       +
       							},

     

       1203
       +
       							"parent": map[string]interface{}{

     

       1204
       +
       								"uri": parentURI,

     

       1205
       +
       								"cid": "bafyparent2",

     

       1206
       +
       							},

     

       1207
       +
       						},

     

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

     

       1209
       +
       					},

     

       1210
       +
       				},

     

       1211
       +
       			}

     

       1212
       +
       

     

       1213
       +
       			err := consumer.HandleEvent(ctx, childEvent)

     

       1214
       +
       			if err != nil {

     

       1215
       +
       				t.Fatalf("Failed to handle child %d event: %v", i, err)

     

       1216
       +
       			}

     

       1217
       +
       		}

     

       1218
       +
       

     

       1219
       +
       		// Now index parent

     

       1220
       +
       		parentEvent := &jetstream.JetstreamEvent{

     

       1221
       +
       			Did:  testUser.DID,

     

       1222
       +
       			Kind: "commit",

     

       1223
       +
       			Commit: &jetstream.CommitEvent{

     

       1224
       +
       				Rev:        "parent2-rev",

     

       1225
       +
       				Operation:  "create",

     

       1226
       +
       				Collection: "social.coves.feed.comment",

     

       1227
       +
       				RKey:       parentRkey,

     

       1228
       +
       				CID:        "bafyparent2",

     

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

     

       1230
       +
       					"$type":   "social.coves.feed.comment",

     

       1231
       +
       					"content": "Parent with 3 pre-existing children",

     

       1232
       +
       					"reply": map[string]interface{}{

     

       1233
       +
       						"root": map[string]interface{}{

     

       1234
       +
       							"uri": postURI,

     

       1235
       +
       							"cid": "bafypost",

     

       1236
       +
       						},

     

       1237
       +
       						"parent": map[string]interface{}{

     

       1238
       +
       							"uri": postURI,

     

       1239
       +
       							"cid": "bafypost",

     

       1240
       +
       						},

     

       1241
       +
       					},

     

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

     

       1243
       +
       				},

     

       1244
       +
       			},

     

       1245
       +
       		}

     

       1246
       +
       

     

       1247
       +
       		err := consumer.HandleEvent(ctx, parentEvent)

     

       1248
       +
       		if err != nil {

     

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

     

       1250
       +
       		}

     

       1251
       +
       

     

       1252
       +
       		// Verify parent has reply_count = 3

     

       1253
       +
       		parentComment, err := commentRepo.GetByURI(ctx, parentURI)

     

       1254
       +
       		if err != nil {

     

       1255
       +
       			t.Fatalf("Parent comment not indexed: %v", err)

     

       1256
       +
       		}

     

       1257
       +
       

     

       1258
       +
       		if parentComment.ReplyCount != 3 {

     

       1259
       +
       			t.Errorf("Expected parent reply_count to be 3 (reconciled), got %d", parentComment.ReplyCount)

     

       1260
       +
       		}

     

       1261
       +
       	})

     

       1262
       +
       }

     

       1263
       +
       

     

       1264
       +
       // TestCommentConsumer_Resurrection tests that soft-deleted comments can be recreated

     

       1265
       +
       // In atProto, deleted records' rkeys become available for reuse

     

       1266
       +
       func TestCommentConsumer_Resurrection(t *testing.T) {

     

       1267
       +
       	db := setupTestDB(t)

     

       1268
       +
       	defer func() {

     

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

     

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

     

       1271
       +
       		}

     

       1272
       +
       	}()

     

       1273
       +
       

     

       1274
       +
       	ctx := context.Background()

     

       1275
       +
       	commentRepo := postgres.NewCommentRepository(db)

     

       1276
       +
       	consumer := jetstream.NewCommentEventConsumer(commentRepo, db)

     

       1277
       +
       

     

       1278
       +
       	testUser := createTestUser(t, db, "resurrect.test", "did:plc:resurrect123")

     

       1279
       +
       	testCommunity, err := createFeedTestCommunity(db, ctx, "resurrect-community", "owner8.test")

     

       1280
       +
       	if err != nil {

     

       1281
       +
       		t.Fatalf("Failed to create test community: %v", err)

     

       1282
       +
       	}

     

       1283
       +
       	postURI := createTestPost(t, db, testCommunity, testUser.DID, "Resurrection Test", 0, time.Now())

     

       1284
       +
       

     

       1285
       +
       	rkey := generateTID()

     

       1286
       +
       	commentURI := fmt.Sprintf("at://%s/social.coves.feed.comment/%s", testUser.DID, rkey)

     

       1287
       +
       

     

       1288
       +
       	t.Run("Recreate deleted comment with same rkey", func(t *testing.T) {

     

       1289
       +
       		// Step 1: Create initial comment

     

       1290
       +
       		createEvent := &jetstream.JetstreamEvent{

     

       1291
       +
       			Did:  testUser.DID,

     

       1292
       +
       			Kind: "commit",

     

       1293
       +
       			Commit: &jetstream.CommitEvent{

     

       1294
       +
       				Rev:        "v1",

     

       1295
       +
       				Operation:  "create",

     

       1296
       +
       				Collection: "social.coves.feed.comment",

     

       1297
       +
       				RKey:       rkey,

     

       1298
       +
       				CID:        "bafyoriginal",

     

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

     

       1300
       +
       					"$type":   "social.coves.feed.comment",

     

       1301
       +
       					"content": "Original comment content",

     

       1302
       +
       					"reply": map[string]interface{}{

     

       1303
       +
       						"root": map[string]interface{}{

     

       1304
       +
       							"uri": postURI,

     

       1305
       +
       							"cid": "bafypost",

     

       1306
       +
       						},

     

       1307
       +
       						"parent": map[string]interface{}{

     

       1308
       +
       							"uri": postURI,

     

       1309
       +
       							"cid": "bafypost",

     

       1310
       +
       						},

     

       1311
       +
       					},

     

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

     

       1313
       +
       				},

     

       1314
       +
       			},

     

       1315
       +
       		}

     

       1316
       +
       

     

       1317
       +
       		err := consumer.HandleEvent(ctx, createEvent)

     

       1318
       +
       		if err != nil {

     

       1319
       +
       			t.Fatalf("Failed to create initial comment: %v", err)

     

       1320
       +
       		}

     

       1321
       +
       

     

       1322
       +
       		// Verify comment exists

     

       1323
       +
       		comment, err := commentRepo.GetByURI(ctx, commentURI)

     

       1324
       +
       		if err != nil {

     

       1325
       +
       			t.Fatalf("Comment not found after creation: %v", err)

     

       1326
       +
       		}

     

       1327
       +
       		if comment.Content != "Original comment content" {

     

       1328
       +
       			t.Errorf("Expected content 'Original comment content', got '%s'", comment.Content)

     

       1329
       +
       		}

     

       1330
       +
       		if comment.DeletedAt != nil {

     

       1331
       +
       			t.Errorf("Expected deleted_at to be nil, got %v", comment.DeletedAt)

     

       1332
       +
       		}

     

       1333
       +
       

     

       1334
       +
       		// Step 2: Delete the comment

     

       1335
       +
       		deleteEvent := &jetstream.JetstreamEvent{

     

       1336
       +
       			Did:  testUser.DID,

     

       1337
       +
       			Kind: "commit",

     

       1338
       +
       			Commit: &jetstream.CommitEvent{

     

       1339
       +
       				Rev:        "v2",

     

       1340
       +
       				Operation:  "delete",

     

       1341
       +
       				Collection: "social.coves.feed.comment",

     

       1342
       +
       				RKey:       rkey,

     

       1343
       +
       			},

     

       1344
       +
       		}

     

       1345
       +
       

     

       1346
       +
       		err = consumer.HandleEvent(ctx, deleteEvent)

     

       1347
       +
       		if err != nil {

     

       1348
       +
       			t.Fatalf("Failed to delete comment: %v", err)

     

       1349
       +
       		}

     

       1350
       +
       

     

       1351
       +
       		// Verify comment is soft-deleted

     

       1352
       +
       		comment, err = commentRepo.GetByURI(ctx, commentURI)

     

       1353
       +
       		if err != nil {

     

       1354
       +
       			t.Fatalf("Comment not found after deletion: %v", err)

     

       1355
       +
       		}

     

       1356
       +
       		if comment.DeletedAt == nil {

     

       1357
       +
       			t.Error("Expected deleted_at to be set, got nil")

     

       1358
       +
       		}

     

       1359
       +
       

     

       1360
       +
       		// Step 3: Recreate comment with same rkey (resurrection)

     

       1361
       +
       		// In atProto, this is a valid operation - user can reuse the rkey

     

       1362
       +
       		recreateEvent := &jetstream.JetstreamEvent{

     

       1363
       +
       			Did:  testUser.DID,

     

       1364
       +
       			Kind: "commit",

     

       1365
       +
       			Commit: &jetstream.CommitEvent{

     

       1366
       +
       				Rev:        "v3",

     

       1367
       +
       				Operation:  "create",

     

       1368
       +
       				Collection: "social.coves.feed.comment",

     

       1369
       +
       				RKey:       rkey, // Same rkey!

     

       1370
       +
       				CID:        "bafyresurrected",

     

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

     

       1372
       +
       					"$type":   "social.coves.feed.comment",

     

       1373
       +
       					"content": "Resurrected comment with new content",

     

       1374
       +
       					"reply": map[string]interface{}{

     

       1375
       +
       						"root": map[string]interface{}{

     

       1376
       +
       							"uri": postURI,

     

       1377
       +
       							"cid": "bafypost",

     

       1378
       +
       						},

     

       1379
       +
       						"parent": map[string]interface{}{

     

       1380
       +
       							"uri": postURI,

     

       1381
       +
       							"cid": "bafypost",

     

       1382
       +
       						},

     

       1383
       +
       					},

     

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

     

       1385
       +
       				},

     

       1386
       +
       			},

     

       1387
       +
       		}

     

       1388
       +
       

     

       1389
       +
       		err = consumer.HandleEvent(ctx, recreateEvent)

     

       1390
       +
       		if err != nil {

     

       1391
       +
       			t.Fatalf("Failed to resurrect comment: %v", err)

     

       1392
       +
       		}

     

       1393
       +
       

     

       1394
       +
       		// Step 4: Verify comment is resurrected with new content

     

       1395
       +
       		comment, err = commentRepo.GetByURI(ctx, commentURI)

     

       1396
       +
       		if err != nil {

     

       1397
       +
       			t.Fatalf("Comment not found after resurrection: %v", err)

     

       1398
       +
       		}

     

       1399
       +
       

     

       1400
       +
       		if comment.DeletedAt != nil {

     

       1401
       +
       			t.Errorf("Expected deleted_at to be NULL after resurrection, got %v", comment.DeletedAt)

     

       1402
       +
       		}

     

       1403
       +
       		if comment.Content != "Resurrected comment with new content" {

     

       1404
       +
       			t.Errorf("Expected resurrected content, got '%s'", comment.Content)

     

       1405
       +
       		}

     

       1406
       +
       		if comment.CID != "bafyresurrected" {

     

       1407
       +
       			t.Errorf("Expected CID 'bafyresurrected', got '%s'", comment.CID)

     

       1408
       +
       		}

     

       1409
       +
       

     

       1410
       +
       		// Verify parent count was restored (post should have comment_count = 1)

     

       1411
       +
       		var postCommentCount int

     

       1412
       +
       		err = db.QueryRowContext(ctx, "SELECT comment_count FROM posts WHERE uri = $1", postURI).Scan(&postCommentCount)

     

       1413
       +
       		if err != nil {

     

       1414
       +
       			t.Fatalf("Failed to check post comment count: %v", err)

     

       1415
       +
       		}

     

       1416
       +
       		if postCommentCount != 1 {

     

       1417
       +
       			t.Errorf("Expected post comment_count to be 1 after resurrection, got %d", postCommentCount)

     

       1418
       +
       		}

     

       1419
       +
       	})

     

       1420
       +
       

     

       1421
       +
       	t.Run("Recreate deleted comment with DIFFERENT parent", func(t *testing.T) {

     

       1422
       +
       		// Create two posts

     

       1423
       +
       		post1URI := createTestPost(t, db, testCommunity, testUser.DID, "Post 1", 0, time.Now())

     

       1424
       +
       		post2URI := createTestPost(t, db, testCommunity, testUser.DID, "Post 2", 0, time.Now())

     

       1425
       +
       

     

       1426
       +
       		rkey2 := generateTID()

     

       1427
       +
       		commentURI2 := fmt.Sprintf("at://%s/social.coves.feed.comment/%s", testUser.DID, rkey2)

     

       1428
       +
       

     

       1429
       +
       		// Step 1: Create comment on Post 1

     

       1430
       +
       		createEvent := &jetstream.JetstreamEvent{

     

       1431
       +
       			Did:  testUser.DID,

     

       1432
       +
       			Kind: "commit",

     

       1433
       +
       			Commit: &jetstream.CommitEvent{

     

       1434
       +
       				Rev:        "v1",

     

       1435
       +
       				Operation:  "create",

     

       1436
       +
       				Collection: "social.coves.feed.comment",

     

       1437
       +
       				RKey:       rkey2,

     

       1438
       +
       				CID:        "bafyv1",

     

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

     

       1440
       +
       					"$type":   "social.coves.feed.comment",

     

       1441
       +
       					"content": "Original on Post 1",

     

       1442
       +
       					"reply": map[string]interface{}{

     

       1443
       +
       						"root": map[string]interface{}{

     

       1444
       +
       							"uri": post1URI,

     

       1445
       +
       							"cid": "bafypost1",

     

       1446
       +
       						},

     

       1447
       +
       						"parent": map[string]interface{}{

     

       1448
       +
       							"uri": post1URI,

     

       1449
       +
       							"cid": "bafypost1",

     

       1450
       +
       						},

     

       1451
       +
       					},

     

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

     

       1453
       +
       				},

     

       1454
       +
       			},

     

       1455
       +
       		}

     

       1456
       +
       

     

       1457
       +
       		err := consumer.HandleEvent(ctx, createEvent)

     

       1458
       +
       		if err != nil {

     

       1459
       +
       			t.Fatalf("Failed to create comment on Post 1: %v", err)

     

       1460
       +
       		}

     

       1461
       +
       

     

       1462
       +
       		// Verify Post 1 has comment_count = 1

     

       1463
       +
       		var post1Count int

     

       1464
       +
       		err = db.QueryRowContext(ctx, "SELECT comment_count FROM posts WHERE uri = $1", post1URI).Scan(&post1Count)

     

       1465
       +
       		if err != nil {

     

       1466
       +
       			t.Fatalf("Failed to check post 1 count: %v", err)

     

       1467
       +
       		}

     

       1468
       +
       		if post1Count != 1 {

     

       1469
       +
       			t.Errorf("Expected Post 1 comment_count = 1, got %d", post1Count)

     

       1470
       +
       		}

     

       1471
       +
       

     

       1472
       +
       		// Step 2: Delete comment

     

       1473
       +
       		deleteEvent := &jetstream.JetstreamEvent{

     

       1474
       +
       			Did:  testUser.DID,

     

       1475
       +
       			Kind: "commit",

     

       1476
       +
       			Commit: &jetstream.CommitEvent{

     

       1477
       +
       				Rev:        "v2",

     

       1478
       +
       				Operation:  "delete",

     

       1479
       +
       				Collection: "social.coves.feed.comment",

     

       1480
       +
       				RKey:       rkey2,

     

       1481
       +
       			},

     

       1482
       +
       		}

     

       1483
       +
       

     

       1484
       +
       		err = consumer.HandleEvent(ctx, deleteEvent)

     

       1485
       +
       		if err != nil {

     

       1486
       +
       			t.Fatalf("Failed to delete comment: %v", err)

     

       1487
       +
       		}

     

       1488
       +
       

     

       1489
       +
       		// Verify Post 1 count decremented to 0

     

       1490
       +
       		err = db.QueryRowContext(ctx, "SELECT comment_count FROM posts WHERE uri = $1", post1URI).Scan(&post1Count)

     

       1491
       +
       		if err != nil {

     

       1492
       +
       			t.Fatalf("Failed to check post 1 count after delete: %v", err)

     

       1493
       +
       		}

     

       1494
       +
       		if post1Count != 0 {

     

       1495
       +
       			t.Errorf("Expected Post 1 comment_count = 0 after delete, got %d", post1Count)

     

       1496
       +
       		}

     

       1497
       +
       

     

       1498
       +
       		// Step 3: Recreate comment with same rkey but on Post 2 (different parent!)

     

       1499
       +
       		recreateEvent := &jetstream.JetstreamEvent{

     

       1500
       +
       			Did:  testUser.DID,

     

       1501
       +
       			Kind: "commit",

     

       1502
       +
       			Commit: &jetstream.CommitEvent{

     

       1503
       +
       				Rev:        "v3",

     

       1504
       +
       				Operation:  "create",

     

       1505
       +
       				Collection: "social.coves.feed.comment",

     

       1506
       +
       				RKey:       rkey2, // Same rkey!

     

       1507
       +
       				CID:        "bafyv3",

     

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

     

       1509
       +
       					"$type":   "social.coves.feed.comment",

     

       1510
       +
       					"content": "New comment on Post 2",

     

       1511
       +
       					"reply": map[string]interface{}{

     

       1512
       +
       						"root": map[string]interface{}{

     

       1513
       +
       							"uri": post2URI, // Different root!

     

       1514
       +
       							"cid": "bafypost2",

     

       1515
       +
       						},

     

       1516
       +
       						"parent": map[string]interface{}{

     

       1517
       +
       							"uri": post2URI, // Different parent!

     

       1518
       +
       							"cid": "bafypost2",

     

       1519
       +
       						},

     

       1520
       +
       					},

     

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

     

       1522
       +
       				},

     

       1523
       +
       			},

     

       1524
       +
       		}

     

       1525
       +
       

     

       1526
       +
       		err = consumer.HandleEvent(ctx, recreateEvent)

     

       1527
       +
       		if err != nil {

     

       1528
       +
       			t.Fatalf("Failed to resurrect comment on Post 2: %v", err)

     

       1529
       +
       		}

     

       1530
       +
       

     

       1531
       +
       		// Step 4: Verify threading references updated correctly

     

       1532
       +
       		comment, err := commentRepo.GetByURI(ctx, commentURI2)

     

       1533
       +
       		if err != nil {

     

       1534
       +
       			t.Fatalf("Failed to get resurrected comment: %v", err)

     

       1535
       +
       		}

     

       1536
       +
       

     

       1537
       +
       		// THIS IS THE CRITICAL TEST: Threading refs must point to Post 2, not Post 1

     

       1538
       +
       		if comment.ParentURI != post2URI {

     

       1539
       +
       			t.Errorf("Expected parent URI to be %s (Post 2), got %s (STALE!)", post2URI, comment.ParentURI)

     

       1540
       +
       		}

     

       1541
       +
       		if comment.RootURI != post2URI {

     

       1542
       +
       			t.Errorf("Expected root URI to be %s (Post 2), got %s (STALE!)", post2URI, comment.RootURI)

     

       1543
       +
       		}

     

       1544
       +
       		if comment.ParentCID != "bafypost2" {

     

       1545
       +
       			t.Errorf("Expected parent CID 'bafypost2', got '%s'", comment.ParentCID)

     

       1546
       +
       		}

     

       1547
       +
       

     

       1548
       +
       		// Verify counts are correct

     

       1549
       +
       		var post2Count int

     

       1550
       +
       		err = db.QueryRowContext(ctx, "SELECT comment_count FROM posts WHERE uri = $1", post2URI).Scan(&post2Count)

     

       1551
       +
       		if err != nil {

     

       1552
       +
       			t.Fatalf("Failed to check post 2 count: %v", err)

     

       1553
       +
       		}

     

       1554
       +
       		if post2Count != 1 {

     

       1555
       +
       			t.Errorf("Expected Post 2 comment_count = 1, got %d", post2Count)

     

       1556
       +
       		}

     

       1557
       +
       

     

       1558
       +
       		// Verify Post 1 count still 0 (not incremented by resurrection on Post 2)

     

       1559
       +
       		err = db.QueryRowContext(ctx, "SELECT comment_count FROM posts WHERE uri = $1", post1URI).Scan(&post1Count)

     

       1560
       +
       		if err != nil {

     

       1561
       +
       			t.Fatalf("Failed to check post 1 count after resurrection: %v", err)

     

       1562
       +
       		}

     

       1563
       +
       		if post1Count != 0 {

     

       1564
       +
       			t.Errorf("Expected Post 1 comment_count = 0 (unchanged), got %d", post1Count)

     

       1565
       +
       		}

     

       1566
       +
       	})

     

       1567
       +
       }

     

       1568
       +
       

     

       1569
       +
       // TestCommentConsumer_ThreadingImmutability tests that UPDATE events cannot change threading refs

     

       1570
       +
       func TestCommentConsumer_ThreadingImmutability(t *testing.T) {

     

       1571
       +
       	db := setupTestDB(t)

     

       1572
       +
       	defer func() {

     

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

     

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

     

       1575
       +
       		}

     

       1576
       +
       	}()

     

       1577
       +
       

     

       1578
       +
       	ctx := context.Background()

     

       1579
       +
       	commentRepo := postgres.NewCommentRepository(db)

     

       1580
       +
       	consumer := jetstream.NewCommentEventConsumer(commentRepo, db)

     

       1581
       +
       

     

       1582
       +
       	testUser := createTestUser(t, db, "immutable.test", "did:plc:immutable123")

     

       1583
       +
       	testCommunity, err := createFeedTestCommunity(db, ctx, "immutable-community", "owner9.test")

     

       1584
       +
       	if err != nil {

     

       1585
       +
       		t.Fatalf("Failed to create test community: %v", err)

     

       1586
       +
       	}

     

       1587
       +
       	postURI1 := createTestPost(t, db, testCommunity, testUser.DID, "Post 1", 0, time.Now())

     

       1588
       +
       	postURI2 := createTestPost(t, db, testCommunity, testUser.DID, "Post 2", 0, time.Now())

     

       1589
       +
       

     

       1590
       +
       	rkey := generateTID()

     

       1591
       +
       	commentURI := fmt.Sprintf("at://%s/social.coves.feed.comment/%s", testUser.DID, rkey)

     

       1592
       +
       

     

       1593
       +
       	t.Run("Reject UPDATE that changes parent URI", func(t *testing.T) {

     

       1594
       +
       		// Create comment on Post 1

     

       1595
       +
       		createEvent := &jetstream.JetstreamEvent{

     

       1596
       +
       			Did:  testUser.DID,

     

       1597
       +
       			Kind: "commit",

     

       1598
       +
       			Commit: &jetstream.CommitEvent{

     

       1599
       +
       				Rev:        "v1",

     

       1600
       +
       				Operation:  "create",

     

       1601
       +
       				Collection: "social.coves.feed.comment",

     

       1602
       +
       				RKey:       rkey,

     

       1603
       +
       				CID:        "bafycomment1",

     

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

     

       1605
       +
       					"$type":   "social.coves.feed.comment",

     

       1606
       +
       					"content": "Comment on Post 1",

     

       1607
       +
       					"reply": map[string]interface{}{

     

       1608
       +
       						"root": map[string]interface{}{

     

       1609
       +
       							"uri": postURI1,

     

       1610
       +
       							"cid": "bafypost1",

     

       1611
       +
       						},

     

       1612
       +
       						"parent": map[string]interface{}{

     

       1613
       +
       							"uri": postURI1,

     

       1614
       +
       							"cid": "bafypost1",

     

       1615
       +
       						},

     

       1616
       +
       					},

     

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

     

       1618
       +
       				},

     

       1619
       +
       			},

     

       1620
       +
       		}

     

       1621
       +
       

     

       1622
       +
       		err := consumer.HandleEvent(ctx, createEvent)

     

       1623
       +
       		if err != nil {

     

       1624
       +
       			t.Fatalf("Failed to create comment: %v", err)

     

       1625
       +
       		}

     

       1626
       +
       

     

       1627
       +
       		// Attempt to update comment to move it to Post 2 (should fail)

     

       1628
       +
       		updateEvent := &jetstream.JetstreamEvent{

     

       1629
       +
       			Did:  testUser.DID,

     

       1630
       +
       			Kind: "commit",

     

       1631
       +
       			Commit: &jetstream.CommitEvent{

     

       1632
       +
       				Rev:        "v2",

     

       1633
       +
       				Operation:  "update",

     

       1634
       +
       				Collection: "social.coves.feed.comment",

     

       1635
       +
       				RKey:       rkey,

     

       1636
       +
       				CID:        "bafycomment2",

     

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

     

       1638
       +
       					"$type":   "social.coves.feed.comment",

     

       1639
       +
       					"content": "Trying to hijack this comment to Post 2",

     

       1640
       +
       					"reply": map[string]interface{}{

     

       1641
       +
       						"root": map[string]interface{}{

     

       1642
       +
       							"uri": postURI2, // Changed!

     

       1643
       +
       							"cid": "bafypost2",

     

       1644
       +
       						},

     

       1645
       +
       						"parent": map[string]interface{}{

     

       1646
       +
       							"uri": postURI2, // Changed!

     

       1647
       +
       							"cid": "bafypost2",

     

       1648
       +
       						},

     

       1649
       +
       					},

     

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

     

       1651
       +
       				},

     

       1652
       +
       			},

     

       1653
       +
       		}

     

       1654
       +
       

     

       1655
       +
       		err = consumer.HandleEvent(ctx, updateEvent)

     

       1656
       +
       		if err == nil {

     

       1657
       +
       			t.Error("Expected error when changing threading references, got nil")

     

       1658
       +
       		}

     

       1659
       +
       		if err != nil && !contains(err.Error(), "threading references cannot be changed") {

     

       1660
       +
       			t.Errorf("Expected 'threading references cannot be changed' error, got: %v", err)

     

       1661
       +
       		}

     

       1662
       +
       

     

       1663
       +
       		// Verify comment still points to Post 1

     

       1664
       +
       		comment, err := commentRepo.GetByURI(ctx, commentURI)

     

       1665
       +
       		if err != nil {

     

       1666
       +
       			t.Fatalf("Failed to get comment: %v", err)

     

       1667
       +
       		}

     

       1668
       +
       		if comment.ParentURI != postURI1 {

     

       1669
       +
       			t.Errorf("Expected parent URI to remain %s, got %s", postURI1, comment.ParentURI)

     

       1670
       +
       		}

     

       1671
       +
       		if comment.RootURI != postURI1 {

     

       1672
       +
       			t.Errorf("Expected root URI to remain %s, got %s", postURI1, comment.RootURI)

     

       1673
       +
       		}

     

       1674
       +
       		// Content should NOT have been updated since the operation was rejected

     

       1675
       +
       		if comment.Content != "Comment on Post 1" {

     

       1676
       +
       			t.Errorf("Expected original content, got '%s'", comment.Content)

     

       1677
       +
       		}

     

       1678
       +
       	})

     

       1679
       +
       

     

       1680
       +
       	t.Run("Allow UPDATE that only changes content (threading unchanged)", func(t *testing.T) {

     

       1681
       +
       		rkey2 := generateTID()

     

       1682
       +
       		commentURI2 := fmt.Sprintf("at://%s/social.coves.feed.comment/%s", testUser.DID, rkey2)

     

       1683
       +
       

     

       1684
       +
       		// Create comment

     

       1685
       +
       		createEvent := &jetstream.JetstreamEvent{

     

       1686
       +
       			Did:  testUser.DID,

     

       1687
       +
       			Kind: "commit",

     

       1688
       +
       			Commit: &jetstream.CommitEvent{

     

       1689
       +
       				Rev:        "v1",

     

       1690
       +
       				Operation:  "create",

     

       1691
       +
       				Collection: "social.coves.feed.comment",

     

       1692
       +
       				RKey:       rkey2,

     

       1693
       +
       				CID:        "bafycomment3",

     

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

     

       1695
       +
       					"$type":   "social.coves.feed.comment",

     

       1696
       +
       					"content": "Original content",

     

       1697
       +
       					"reply": map[string]interface{}{

     

       1698
       +
       						"root": map[string]interface{}{

     

       1699
       +
       							"uri": postURI1,

     

       1700
       +
       							"cid": "bafypost1",

     

       1701
       +
       						},

     

       1702
       +
       						"parent": map[string]interface{}{

     

       1703
       +
       							"uri": postURI1,

     

       1704
       +
       							"cid": "bafypost1",

     

       1705
       +
       						},

     

       1706
       +
       					},

     

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

     

       1708
       +
       				},

     

       1709
       +
       			},

     

       1710
       +
       		}

     

       1711
       +
       

     

       1712
       +
       		err := consumer.HandleEvent(ctx, createEvent)

     

       1713
       +
       		if err != nil {

     

       1714
       +
       			t.Fatalf("Failed to create comment: %v", err)

     

       1715
       +
       		}

     

       1716
       +
       

     

       1717
       +
       		// Update content only (threading unchanged - should succeed)

     

       1718
       +
       		updateEvent := &jetstream.JetstreamEvent{

     

       1719
       +
       			Did:  testUser.DID,

     

       1720
       +
       			Kind: "commit",

     

       1721
       +
       			Commit: &jetstream.CommitEvent{

     

       1722
       +
       				Rev:        "v2",

     

       1723
       +
       				Operation:  "update",

     

       1724
       +
       				Collection: "social.coves.feed.comment",

     

       1725
       +
       				RKey:       rkey2,

     

       1726
       +
       				CID:        "bafycomment4",

     

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

     

       1728
       +
       					"$type":   "social.coves.feed.comment",

     

       1729
       +
       					"content": "Updated content",

     

       1730
       +
       					"reply": map[string]interface{}{

     

       1731
       +
       						"root": map[string]interface{}{

     

       1732
       +
       							"uri": postURI1, // Same

     

       1733
       +
       							"cid": "bafypost1",

     

       1734
       +
       						},

     

       1735
       +
       						"parent": map[string]interface{}{

     

       1736
       +
       							"uri": postURI1, // Same

     

       1737
       +
       							"cid": "bafypost1",

     

       1738
       +
       						},

     

       1739
       +
       					},

     

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

     

       1741
       +
       				},

     

       1742
       +
       			},

     

       1743
       +
       		}

     

       1744
       +
       

     

       1745
       +
       		err = consumer.HandleEvent(ctx, updateEvent)

     

       1746
       +
       		if err != nil {

     

       1747
       +
       			t.Fatalf("Expected update to succeed when threading unchanged, got error: %v", err)

     

       1748
       +
       		}

     

       1749
       +
       

     

       1750
       +
       		// Verify content was updated

     

       1751
       +
       		comment, err := commentRepo.GetByURI(ctx, commentURI2)

     

       1752
       +
       		if err != nil {

     

       1753
       +
       			t.Fatalf("Failed to get comment: %v", err)

     

       1754
       +
       		}

     

       1755
       +
       		if comment.Content != "Updated content" {

     

       1756
       +
       			t.Errorf("Expected updated content, got '%s'", comment.Content)

     

       1757
       +
       		}

     

       1758
       +
       		// Threading should remain unchanged

     

       1759
       +
       		if comment.ParentURI != postURI1 {

     

       1760
       +
       			t.Errorf("Expected parent URI %s, got %s", postURI1, comment.ParentURI)

     

       1761
       +
       		}

     

       1762
       +
       	})

     

       1763
       +
       }