diff --git a/internal/atproto/jetstream/comment_consumer.go b/internal/atproto/jetstream/comment_consumer.go index 40f3f51..6ecb2a8 100644 --- a/internal/atproto/jetstream/comment_consumer.go +++ b/internal/atproto/jetstream/comment_consumer.go @@ -1,6 +1,7 @@ package jetstream import ( + "Coves/internal/atproto/utils" "Coves/internal/core/comments" "context" "database/sql" @@ -356,7 +357,7 @@ func (c *CommentEventConsumer) indexCommentAndUpdateCounts(ctx context.Context, // 2. Update parent counts atomically // Parent could be a post (increment comment_count) or a comment (increment reply_count) - // Try posts table first + // Parse collection from parent URI to determine target table // // FIXME(P1): Post comment_count reconciliation not implemented // When a comment arrives before its parent post (common with cross-repo Jetstream ordering), @@ -369,15 +370,39 @@ func (c *CommentEventConsumer) indexCommentAndUpdateCounts(ctx context.Context, // matches the post URI and set comment_count accordingly. // // Test demonstrating issue: TestCommentConsumer_PostCountReconciliation_Limitation - updatePostQuery := ` - UPDATE posts - SET comment_count = comment_count + 1 - WHERE uri = $1 AND deleted_at IS NULL - ` + collection := utils.ExtractCollectionFromURI(comment.ParentURI) + + var updateQuery string + switch collection { + case "social.coves.community.post": + // Comment on post - update posts.comment_count + updateQuery = ` + UPDATE posts + SET comment_count = comment_count + 1 + WHERE uri = $1 AND deleted_at IS NULL + ` + + case "social.coves.feed.comment": + // Reply to comment - update comments.reply_count + updateQuery = ` + UPDATE comments + SET reply_count = reply_count + 1 + WHERE uri = $1 AND deleted_at IS NULL + ` - result, err := tx.ExecContext(ctx, updatePostQuery, comment.ParentURI) + default: + // Unknown or unsupported parent collection + // Comment is still indexed, we just don't update parent counts + log.Printf("Comment parent has unsupported collection: %s (comment indexed, parent count not updated)", collection) + if commitErr := tx.Commit(); commitErr != nil { + return fmt.Errorf("failed to commit transaction: %w", commitErr) + } + return nil + } + + result, err := tx.ExecContext(ctx, updateQuery, comment.ParentURI) if err != nil { - return fmt.Errorf("failed to update post comment count: %w", err) + return fmt.Errorf("failed to update parent count: %w", err) } rowsAffected, err := result.RowsAffected() @@ -385,28 +410,9 @@ func (c *CommentEventConsumer) indexCommentAndUpdateCounts(ctx context.Context, return fmt.Errorf("failed to check update result: %w", err) } - // If no post was updated, parent is probably a comment + // If parent not found, that's OK (parent might not be indexed yet) if rowsAffected == 0 { - updateCommentQuery := ` - UPDATE comments - SET reply_count = reply_count + 1 - WHERE uri = $1 AND deleted_at IS NULL - ` - - result, err := tx.ExecContext(ctx, updateCommentQuery, comment.ParentURI) - if err != nil { - return fmt.Errorf("failed to update comment reply count: %w", err) - } - - rowsAffected, err := result.RowsAffected() - if err != nil { - return fmt.Errorf("failed to check update result: %w", err) - } - - // If neither post nor comment was found, that's OK (parent might not be indexed yet) - if rowsAffected == 0 { - 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) } // Commit transaction @@ -456,16 +462,40 @@ func (c *CommentEventConsumer) deleteCommentAndUpdateCounts(ctx context.Context, } // 2. Decrement parent counts atomically - // Parent could be a post or comment - try both (use GREATEST to prevent negative) - updatePostQuery := ` - UPDATE posts - SET comment_count = GREATEST(0, comment_count - 1) - WHERE uri = $1 AND deleted_at IS NULL - ` + // Parent could be a post or comment - parse collection to determine target table + collection := utils.ExtractCollectionFromURI(comment.ParentURI) + + var updateQuery string + switch collection { + case "social.coves.community.post": + // Comment on post - decrement posts.comment_count + updateQuery = ` + UPDATE posts + SET comment_count = GREATEST(0, comment_count - 1) + WHERE uri = $1 AND deleted_at IS NULL + ` + + case "social.coves.feed.comment": + // Reply to comment - decrement comments.reply_count + updateQuery = ` + UPDATE comments + SET reply_count = GREATEST(0, reply_count - 1) + WHERE uri = $1 AND deleted_at IS NULL + ` - result, err = tx.ExecContext(ctx, updatePostQuery, comment.ParentURI) + default: + // Unknown or unsupported parent collection + // Comment is still deleted, we just don't update parent counts + log.Printf("Comment parent has unsupported collection: %s (comment deleted, parent count not updated)", collection) + if commitErr := tx.Commit(); commitErr != nil { + return fmt.Errorf("failed to commit transaction: %w", commitErr) + } + return nil + } + + result, err = tx.ExecContext(ctx, updateQuery, comment.ParentURI) if err != nil { - return fmt.Errorf("failed to update post comment count: %w", err) + return fmt.Errorf("failed to update parent count: %w", err) } rowsAffected, err = result.RowsAffected() @@ -473,28 +503,9 @@ func (c *CommentEventConsumer) deleteCommentAndUpdateCounts(ctx context.Context, return fmt.Errorf("failed to check update result: %w", err) } - // If no post was updated, parent is probably a comment + // If parent not found, that's OK (parent might be deleted) if rowsAffected == 0 { - updateCommentQuery := ` - UPDATE comments - SET reply_count = GREATEST(0, reply_count - 1) - WHERE uri = $1 AND deleted_at IS NULL - ` - - result, err := tx.ExecContext(ctx, updateCommentQuery, comment.ParentURI) - if err != nil { - return fmt.Errorf("failed to update comment reply count: %w", err) - } - - rowsAffected, err := result.RowsAffected() - if err != nil { - return fmt.Errorf("failed to check update result: %w", err) - } - - // If neither was found, that's OK (parent might be deleted) - if rowsAffected == 0 { - 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) } // Commit transaction diff --git a/internal/atproto/jetstream/vote_consumer.go b/internal/atproto/jetstream/vote_consumer.go index 2996b27..3f97fbb 100644 --- a/internal/atproto/jetstream/vote_consumer.go +++ b/internal/atproto/jetstream/vote_consumer.go @@ -1,6 +1,7 @@ package jetstream import ( + "Coves/internal/atproto/utils" "Coves/internal/core/users" "Coves/internal/core/votes" "context" @@ -179,29 +180,61 @@ func (c *VoteEventConsumer) indexVoteAndUpdateCounts(ctx context.Context, vote * return fmt.Errorf("failed to insert vote: %w", err) } - // 2. Update post vote counts atomically - // Increment upvote_count or downvote_count based on direction - // Also update score (upvote_count - downvote_count) + // 2. Update vote counts on the subject (post or comment) + // Parse collection from subject URI to determine target table + collection := utils.ExtractCollectionFromURI(vote.SubjectURI) + var updateQuery string - if vote.Direction == "up" { - updateQuery = ` - UPDATE posts - SET upvote_count = upvote_count + 1, - score = upvote_count + 1 - downvote_count - WHERE uri = $1 AND deleted_at IS NULL - ` - } else { // "down" - updateQuery = ` - UPDATE posts - SET downvote_count = downvote_count + 1, - score = upvote_count - (downvote_count + 1) - WHERE uri = $1 AND deleted_at IS NULL - ` + switch collection { + case "social.coves.community.post": + // Vote on post - update posts table + if vote.Direction == "up" { + updateQuery = ` + UPDATE posts + SET upvote_count = upvote_count + 1, + score = upvote_count + 1 - downvote_count + WHERE uri = $1 AND deleted_at IS NULL + ` + } else { // "down" + updateQuery = ` + UPDATE posts + SET downvote_count = downvote_count + 1, + score = upvote_count - (downvote_count + 1) + WHERE uri = $1 AND deleted_at IS NULL + ` + } + + case "social.coves.feed.comment": + // Vote on comment - update comments table + if vote.Direction == "up" { + updateQuery = ` + UPDATE comments + SET upvote_count = upvote_count + 1, + score = upvote_count + 1 - downvote_count + WHERE uri = $1 AND deleted_at IS NULL + ` + } else { // "down" + updateQuery = ` + UPDATE comments + SET downvote_count = downvote_count + 1, + score = upvote_count - (downvote_count + 1) + WHERE uri = $1 AND deleted_at IS NULL + ` + } + + default: + // Unknown or unsupported collection + // Vote is still indexed in votes table, we just don't update denormalized counts + log.Printf("Vote subject has unsupported collection: %s (vote indexed, counts not updated)", collection) + if commitErr := tx.Commit(); commitErr != nil { + return fmt.Errorf("failed to commit transaction: %w", commitErr) + } + return nil } result, err := tx.ExecContext(ctx, updateQuery, vote.SubjectURI) if err != nil { - return fmt.Errorf("failed to update post counts: %w", err) + return fmt.Errorf("failed to update vote counts: %w", err) } rowsAffected, err := result.RowsAffected() @@ -209,10 +242,9 @@ func (c *VoteEventConsumer) indexVoteAndUpdateCounts(ctx context.Context, vote * return fmt.Errorf("failed to check update result: %w", err) } - // If post doesn't exist or is deleted, that's OK (vote still indexed) - // Future: We could validate post exists before indexing vote + // If subject doesn't exist or is deleted, that's OK (vote still indexed) if rowsAffected == 0 { - log.Printf("Warning: Post not found or deleted: %s (vote indexed anyway)", vote.SubjectURI) + log.Printf("Warning: Vote subject not found or deleted: %s (vote indexed anyway)", vote.SubjectURI) } // Commit transaction @@ -261,29 +293,61 @@ func (c *VoteEventConsumer) deleteVoteAndUpdateCounts(ctx context.Context, vote return nil } - // 2. Decrement post vote counts atomically - // Decrement upvote_count or downvote_count based on direction - // Also update score (use GREATEST to prevent negative counts) + // 2. Decrement vote counts on the subject (post or comment) + // Parse collection from subject URI to determine target table + collection := utils.ExtractCollectionFromURI(vote.SubjectURI) + var updateQuery string - if vote.Direction == "up" { - updateQuery = ` - UPDATE posts - SET upvote_count = GREATEST(0, upvote_count - 1), - score = GREATEST(0, upvote_count - 1) - downvote_count - WHERE uri = $1 AND deleted_at IS NULL - ` - } else { // "down" - updateQuery = ` - UPDATE posts - SET downvote_count = GREATEST(0, downvote_count - 1), - score = upvote_count - GREATEST(0, downvote_count - 1) - WHERE uri = $1 AND deleted_at IS NULL - ` + switch collection { + case "social.coves.community.post": + // Vote on post - update posts table + if vote.Direction == "up" { + updateQuery = ` + UPDATE posts + SET upvote_count = GREATEST(0, upvote_count - 1), + score = GREATEST(0, upvote_count - 1) - downvote_count + WHERE uri = $1 AND deleted_at IS NULL + ` + } else { // "down" + updateQuery = ` + UPDATE posts + SET downvote_count = GREATEST(0, downvote_count - 1), + score = upvote_count - GREATEST(0, downvote_count - 1) + WHERE uri = $1 AND deleted_at IS NULL + ` + } + + case "social.coves.feed.comment": + // Vote on comment - update comments table + if vote.Direction == "up" { + updateQuery = ` + UPDATE comments + SET upvote_count = GREATEST(0, upvote_count - 1), + score = GREATEST(0, upvote_count - 1) - downvote_count + WHERE uri = $1 AND deleted_at IS NULL + ` + } else { // "down" + updateQuery = ` + UPDATE comments + SET downvote_count = GREATEST(0, downvote_count - 1), + score = upvote_count - GREATEST(0, downvote_count - 1) + WHERE uri = $1 AND deleted_at IS NULL + ` + } + + default: + // Unknown or unsupported collection + // Vote is still deleted, we just don't update denormalized counts + log.Printf("Vote subject has unsupported collection: %s (vote deleted, counts not updated)", collection) + if commitErr := tx.Commit(); commitErr != nil { + return fmt.Errorf("failed to commit transaction: %w", commitErr) + } + return nil } result, err = tx.ExecContext(ctx, updateQuery, vote.SubjectURI) if err != nil { - return fmt.Errorf("failed to update post counts: %w", err) + return fmt.Errorf("failed to update vote counts: %w", err) } rowsAffected, err = result.RowsAffected() @@ -291,9 +355,9 @@ func (c *VoteEventConsumer) deleteVoteAndUpdateCounts(ctx context.Context, vote return fmt.Errorf("failed to check update result: %w", err) } - // If post doesn't exist or is deleted, that's OK (vote still deleted) + // If subject doesn't exist or is deleted, that's OK (vote still deleted) if rowsAffected == 0 { - log.Printf("Warning: Post not found or deleted: %s (vote deleted anyway)", vote.SubjectURI) + log.Printf("Warning: Vote subject not found or deleted: %s (vote deleted anyway)", vote.SubjectURI) } // Commit transaction