bretton.dev
+1
-1
cmd/server/main.go
+1
-1
cmd/server/main.go
···postJetstreamURL = "ws://localhost:6008/subscribe?wantedCollections=social.coves.community.post"postJetstreamConnector := jetstream.NewPostJetstreamConnector(postEventConsumer, postJetstreamURL)
+276
-24
docs/COMMENT_SYSTEM_IMPLEMENTATION.md
+276
-24
docs/COMMENT_SYSTEM_IMPLEMENTATION.md
···This document details the complete implementation of the comment system for Coves, a forum-like atProto social media platform. The comment system follows the established vote system pattern, with comments living in user repositories and being indexed by the AppView via Jetstream firehose.-**Last Updated:** November 6, 2025 (Final PR review fixes complete - lexicon compliance, data integrity, SQL correctness)···+2. `internal/atproto/jetstream/vote_consumer.go` - Refactored for comment support with URI parsing+3. `internal/atproto/jetstream/comment_consumer.go` - Applied URI parsing pattern for consistency+- Viewer state population (authenticated with vote, authenticated without vote, unauthenticated)+After Phase 2B implementation, a thorough PR review identified several critical issues and improvements that were addressed before production deployment:+- **Problem:** When a comment arrives before its parent post (common with Jetstream's cross-repository event ordering), the post update returns 0 rows affected. Later when the post is indexed, there was NO reconciliation logic to count pre-existing comments, causing posts to have permanently stale `comment_count` values.+- **Impact:** Posts would show incorrect comment counts indefinitely, breaking UX and violating data integrity+- **Solution:** Implemented reconciliation in post consumer (similar to existing pattern in comment consumer)+- Updates `comment_count` atomically: `SET comment_count = (SELECT COUNT(*) FROM comments WHERE parent_uri = $1)`+- **Initial Request:** Change `log.Printf("...%v", err)` to `log.Printf("...%w", err)` in vote consumer+- **Outcome:** No changes needed; error is properly returned on next line to preserve error chain+- **Issue:** Rich text facets, embeds, and labels are stored in database but not deserialized in API responses+- **Decision:** Per original Phase 2C plan, defer JSON field deserialization (already marked with TODO comments)+- **Rationale:** Phase 2C explicitly covers "complete record" population - no scope creep needed+- **Problem:** Taking address of type-asserted variables directly from type assertion could be risky during refactoring+- **Problem:** Function returned empty string for malformed URIs with no clear indication in documentation+- **Problem:** Query selected unused columns (`cid`, `created_at`) that weren't accessed by service+**Total Implementation Effort:** Phase 2B initial (5-7 hours) + PR hardening (6-8 hours) = **~11-15 hours**···-The comment system has successfully completed **Phase 1 (Indexing)** and **Phase 2A (Query API)**, providing a production-ready threaded discussion system for Coves:+The comment system has successfully completed **Phase 1 (Indexing)**, **Phase 2A (Query API)**, and **Phase 2B (Vote Integration)** with comprehensive production hardening, providing a production-ready threaded discussion system for Coves:The implementation provides a solid foundation for building rich threaded discussions in Coves while maintaining compatibility with the broader atProto ecosystem and following established patterns from platforms like Lemmy and Reddit.···TEST_DATABASE_URL="postgres://test_user:test_password@localhost:5434/coves_test?sslmode=disable" \+TEST_DATABASE_URL="postgres://test_user:test_password@localhost:5434/coves_test?sslmode=disable" \···+**Documentation:** Comprehensive implementation guide covering all phases, PR reviews, and production considerations
+69
-58
internal/atproto/jetstream/comment_consumer.go
+69
-58
internal/atproto/jetstream/comment_consumer.go
······// When a comment arrives before its parent post (common with cross-repo Jetstream ordering),···+log.Printf("Comment parent has unsupported collection: %s (comment indexed, parent count not updated)", collection)···-log.Printf("Warning: Parent not found or deleted: %s (comment indexed anyway)", comment.ParentURI)+log.Printf("Warning: Parent not found or deleted: %s (comment indexed anyway)", comment.ParentURI)···+log.Printf("Comment parent has unsupported collection: %s (comment deleted, parent count not updated)", collection)···-log.Printf("Warning: Parent not found or deleted: %s (comment deleted anyway)", comment.ParentURI)+log.Printf("Warning: Parent not found or deleted: %s (comment deleted anyway)", comment.ParentURI)
+99
-9
internal/atproto/jetstream/post_consumer.go
+99
-9
internal/atproto/jetstream/post_consumer.go
············+func (c *PostEventConsumer) indexPostAndReconcileCounts(ctx context.Context, post *posts.Post) error {
+105
-41
internal/atproto/jetstream/vote_consumer.go
+105
-41
internal/atproto/jetstream/vote_consumer.go
······+log.Printf("Vote subject has unsupported collection: %s (vote indexed, counts not updated)", collection)···+log.Printf("Warning: Vote subject not found or deleted: %s (vote indexed anyway)", vote.SubjectURI)···+log.Printf("Vote subject has unsupported collection: %s (vote deleted, counts not updated)", collection)···+log.Printf("Warning: Vote subject not found or deleted: %s (vote deleted anyway)", vote.SubjectURI)
+19
internal/atproto/utils/record_utils.go
+19
internal/atproto/utils/record_utils.go
···
+60
-14
internal/core/comments/comment_service.go
+60
-14
internal/core/comments/comment_service.go
··················-// voteState, err := s.commentRepo.GetVoteStateForComments(ctx, *viewerDID, []string{comment.URI})
+1139
internal/core/comments/comment_service_test.go
+1139
internal/core/comments/comment_service_test.go
···+listByParentWithHotRankFunc func(ctx context.Context, parentURI string, sort string, timeframe string, limit int, cursor *string) ([]*Comment, *string, error)+listByParentsBatchFunc func(ctx context.Context, parentURIs []string, sort string, limitPerParent int) (map[string][]*Comment, error)+getVoteStateForCommentsFunc func(ctx context.Context, viewerDID string, commentURIs []string) (map[string]interface{}, error)+func (m *mockCommentRepo) ListByRoot(ctx context.Context, rootURI string, limit, offset int) ([]*Comment, error) {+func (m *mockCommentRepo) ListByParent(ctx context.Context, parentURI string, limit, offset int) ([]*Comment, error) {+func (m *mockCommentRepo) ListByCommenter(ctx context.Context, commenterDID string, limit, offset int) ([]*Comment, error) {+func (m *mockCommentRepo) GetByURIsBatch(ctx context.Context, uris []string) (map[string]*Comment, error) {+func (m *mockCommentRepo) GetVoteStateForComments(ctx context.Context, viewerDID string, commentURIs []string) (map[string]interface{}, error) {+func (m *mockUserRepo) UpdateHandle(ctx context.Context, did, newHandle string) (*users.User, error) {+func (m *mockCommunityRepo) Create(ctx context.Context, community *communities.Community) (*communities.Community, error) {+func (m *mockCommunityRepo) GetByDID(ctx context.Context, did string) (*communities.Community, error) {+func (m *mockCommunityRepo) GetByHandle(ctx context.Context, handle string) (*communities.Community, error) {+func (m *mockCommunityRepo) Update(ctx context.Context, community *communities.Community) (*communities.Community, error) {+func (m *mockCommunityRepo) UpdateCredentials(ctx context.Context, did, accessToken, refreshToken string) error {+func (m *mockCommunityRepo) List(ctx context.Context, req communities.ListCommunitiesRequest) ([]*communities.Community, int, error) {+func (m *mockCommunityRepo) Search(ctx context.Context, req communities.SearchCommunitiesRequest) ([]*communities.Community, int, error) {+func (m *mockCommunityRepo) Subscribe(ctx context.Context, subscription *communities.Subscription) (*communities.Subscription, error) {+func (m *mockCommunityRepo) SubscribeWithCount(ctx context.Context, subscription *communities.Subscription) (*communities.Subscription, error) {+func (m *mockCommunityRepo) Unsubscribe(ctx context.Context, userDID, communityDID string) error {+func (m *mockCommunityRepo) UnsubscribeWithCount(ctx context.Context, userDID, communityDID string) error {+func (m *mockCommunityRepo) GetSubscription(ctx context.Context, userDID, communityDID string) (*communities.Subscription, error) {+func (m *mockCommunityRepo) GetSubscriptionByURI(ctx context.Context, recordURI string) (*communities.Subscription, error) {+func (m *mockCommunityRepo) ListSubscriptions(ctx context.Context, userDID string, limit, offset int) ([]*communities.Subscription, error) {+func (m *mockCommunityRepo) ListSubscribers(ctx context.Context, communityDID string, limit, offset int) ([]*communities.Subscription, error) {+func (m *mockCommunityRepo) BlockCommunity(ctx context.Context, block *communities.CommunityBlock) (*communities.CommunityBlock, error) {+func (m *mockCommunityRepo) UnblockCommunity(ctx context.Context, userDID, communityDID string) error {+func (m *mockCommunityRepo) GetBlock(ctx context.Context, userDID, communityDID string) (*communities.CommunityBlock, error) {+func (m *mockCommunityRepo) GetBlockByURI(ctx context.Context, recordURI string) (*communities.CommunityBlock, error) {+func (m *mockCommunityRepo) ListBlockedCommunities(ctx context.Context, userDID string, limit, offset int) ([]*communities.CommunityBlock, error) {+func (m *mockCommunityRepo) IsBlocked(ctx context.Context, userDID, communityDID string) (bool, error) {+func (m *mockCommunityRepo) CreateMembership(ctx context.Context, membership *communities.Membership) (*communities.Membership, error) {+func (m *mockCommunityRepo) GetMembership(ctx context.Context, userDID, communityDID string) (*communities.Membership, error) {+func (m *mockCommunityRepo) UpdateMembership(ctx context.Context, membership *communities.Membership) (*communities.Membership, error) {+func (m *mockCommunityRepo) ListMembers(ctx context.Context, communityDID string, limit, offset int) ([]*communities.Membership, error) {+func (m *mockCommunityRepo) CreateModerationAction(ctx context.Context, action *communities.ModerationAction) (*communities.ModerationAction, error) {+func (m *mockCommunityRepo) ListModerationActions(ctx context.Context, communityDID string, limit, offset int) ([]*communities.ModerationAction, error) {+func (m *mockCommunityRepo) IncrementMemberCount(ctx context.Context, communityDID string) error {+func (m *mockCommunityRepo) DecrementMemberCount(ctx context.Context, communityDID string) error {+func (m *mockCommunityRepo) IncrementSubscriberCount(ctx context.Context, communityDID string) error {+func (m *mockCommunityRepo) DecrementSubscriberCount(ctx context.Context, communityDID string) error {+func (m *mockCommunityRepo) IncrementPostCount(ctx context.Context, communityDID string) error {+func createTestComment(uri string, commenterDID string, commenterHandle string, rootURI string, parentURI string, replyCount int) *Comment {+comment1 := createTestComment("at://did:plc:commenter123/comment/1", commenterDID, "commenter.test", postURI, postURI, 0)+comment2 := createTestComment("at://did:plc:commenter123/comment/2", commenterDID, "commenter.test", postURI, postURI, 0)+commentRepo.listByParentWithHotRankFunc = func(ctx context.Context, parentURI string, sort string, timeframe string, limit int, cursor *string) ([]*Comment, *string, error) {+commentRepo.listByParentWithHotRankFunc = func(ctx context.Context, parentURI string, sort string, timeframe string, limit int, cursor *string) ([]*Comment, *string, error) {+comment1 := createTestComment(comment1URI, commenterDID, "commenter.test", postURI, postURI, 0)+commentRepo.listByParentWithHotRankFunc = func(ctx context.Context, parentURI string, sort string, timeframe string, limit int, cursor *string) ([]*Comment, *string, error) {+commentRepo.getVoteStateForCommentsFunc = func(ctx context.Context, viewerDID string, commentURIs []string) (map[string]interface{}, error) {+comment1 := createTestComment("at://did:plc:commenter123/comment/1", commenterDID, "commenter.test", postURI, postURI, 0)+commentRepo.listByParentWithHotRankFunc = func(ctx context.Context, parentURI string, sort string, timeframe string, limit int, cursor *string) ([]*Comment, *string, error) {+comment1 := createTestComment("at://did:plc:commenter123/comment/1", commenterDID, "commenter.test", postURI, postURI, 0)+commentRepo.listByParentWithHotRankFunc = func(ctx context.Context, parentURI string, sort string, timeframe string, limit int, cursor *string) ([]*Comment, *string, error) {+commentRepo.listByParentWithHotRankFunc = func(ctx context.Context, parentURI string, sort string, timeframe string, limit int, cursor *string) ([]*Comment, *string, error) {+service := NewCommentService(commentRepo, userRepo, postRepo, communityRepo).(*commentService)+deletedComment := createTestComment("at://did:plc:commenter123/comment/1", "did:plc:commenter123", "commenter.test", postURI, postURI, 0)+normalComment := createTestComment("at://did:plc:commenter123/comment/2", "did:plc:commenter123", "commenter.test", postURI, postURI, 0)+service := NewCommentService(commentRepo, userRepo, postRepo, communityRepo).(*commentService)+result := service.buildThreadViews(context.Background(), []*Comment{deletedComment, normalComment}, 10, "hot", nil)+parentComment := createTestComment(parentURI, "did:plc:commenter123", "commenter.test", postURI, postURI, 1)+childComment := createTestComment(childURI, "did:plc:commenter123", "commenter.test", postURI, parentURI, 0)+commentRepo.listByParentsBatchFunc = func(ctx context.Context, parentURIs []string, sort string, limitPerParent int) (map[string][]*Comment, error) {+service := NewCommentService(commentRepo, userRepo, postRepo, communityRepo).(*commentService)+result := service.buildThreadViews(context.Background(), []*Comment{parentComment}, 1, "hot", nil)+parentComment := createTestComment("at://did:plc:commenter123/comment/1", "did:plc:commenter123", "commenter.test", postURI, postURI, 5)+service := NewCommentService(commentRepo, userRepo, postRepo, communityRepo).(*commentService)+result := service.buildThreadViews(context.Background(), []*Comment{parentComment}, 0, "hot", nil)+comment := createTestComment(commentURI, "did:plc:commenter123", "commenter.test", postURI, postURI, 0)+service := NewCommentService(commentRepo, userRepo, postRepo, communityRepo).(*commentService)+comment := createTestComment(commentURI, "did:plc:commenter123", "commenter.test", postURI, postURI, 0)+service := NewCommentService(commentRepo, userRepo, postRepo, communityRepo).(*commentService)+comment := createTestComment(childCommentURI, "did:plc:commenter123", "commenter.test", postURI, parentCommentURI, 0)+service := NewCommentService(commentRepo, userRepo, postRepo, communityRepo).(*commentService)+comment := createTestComment(commentURI, "did:plc:commenter123", "commenter.test", postURI, postURI, 0)+service := NewCommentService(commentRepo, userRepo, postRepo, communityRepo).(*commentService)+comment := createTestComment(commentURI, "did:plc:commenter123", "commenter.test", postURI, postURI, 0)+service := NewCommentService(commentRepo, userRepo, postRepo, communityRepo).(*commentService)
+3
-6
internal/db/postgres/comment_repo.go
+3
-6
internal/db/postgres/comment_repo.go
·········
+1
-1
tests/integration/aggregator_e2e_test.go
+1
-1
tests/integration/aggregator_e2e_test.go
···
+565
tests/integration/comment_vote_test.go
+565
tests/integration/comment_vote_test.go
···
+5
-5
tests/integration/post_e2e_test.go
+5
-5
tests/integration/post_e2e_test.go
···············