feat: add comments

2026-06-13 01:29:55 +05:30
parent 1edee92865
commit 6ebaa36158
11 changed files with 968 additions and 10 deletions
@@ -0,0 +1,351 @@
+package repository
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/prabhupadasays/spsays-bff/internal/model"
+	"go.uber.org/zap"
+)
+
+var (
+	// ErrCommentNotFound — no comment with that id (or not in the expected
+	// scope, e.g. pinning a reply).
+	ErrCommentNotFound = errors.New("comment not found")
+	// ErrDuplicateComment — the same user re-posted identical text in the same
+	// scope (caught by the dedup unique index).
+	ErrDuplicateComment = errors.New("duplicate comment")
+)
+
+// CommentRepo handles post comments, replies, comment-likes, and pinning.
+// Comment likes are far lower-volume than post likes, so likes_count is kept
+// in sync transactionally rather than batched through Redis.
+type CommentRepo struct {
+	pg     *Postgres
+	logger *zap.Logger
+}
+
+func NewCommentRepo(pg *Postgres, logger *zap.Logger) *CommentRepo {
+	return &CommentRepo{pg: pg, logger: logger}
+}
+
+// nullUUIDArg returns nil for an empty id so a `$n::uuid` parameter binds to
+// SQL NULL (anonymous viewer) instead of failing uuid parsing on "".
+func nullUUIDArg(s string) *string {
+	if s == "" {
+		return nil
+	}
+	return &s
+}
+
+// commentSelect is the shared column list for reading a comment with its
+// author, the reply-to handle, the viewer's like state, and the reply count.
+// `vl` is a viewer-scoped LEFT JOIN on comment_likes; `ru` the reply-to user.
+const commentSelect = `
+	c.id, c.post_id, c.user_id, c.parent_id, c.body, c.likes_count,
+	c.pinned_at, c.created_at, c.updated_at,
+	u.name, u.handle, u.avatar_url,
+	ru.handle,
+	(vl.comment_id IS NOT NULL),
+	(SELECT count(*) FROM post_comments rc WHERE rc.parent_id = c.id)`
+
+// The viewer id always rides in parameter slot $2 across the read queries
+// below, so the shared viewer-scoped like join can hardcode it.
+const commentFrom = `
+	FROM post_comments c
+	JOIN users u ON u.id = c.user_id
+	LEFT JOIN users ru ON ru.id = c.reply_to_user_id
+	LEFT JOIN comment_likes vl ON vl.comment_id = c.id AND vl.user_id = $2::uuid`
+
+func scanComment(row pgx.Row, c *model.Comment) error {
+	return row.Scan(&c.ID, &c.PostID, &c.UserID, &c.ParentID, &c.Body, &c.LikesCount,
+		&c.PinnedAt, &c.CreatedAt, &c.UpdatedAt,
+		&c.AuthorName, &c.AuthorHandle, &c.AuthorAvatarURL,
+		&c.ReplyToHandle,
+		&c.ViewerHasLiked, &c.ReplyCount)
+}
+
+func scanComments(rows pgx.Rows) ([]model.Comment, error) {
+	defer rows.Close()
+	out := []model.Comment{}
+	for rows.Next() {
+		var c model.Comment
+		if err := scanComment(rows, &c); err != nil {
+			return nil, fmt.Errorf("scan comment: %w", err)
+		}
+		out = append(out, c)
+	}
+	return out, rows.Err()
+}
+
+// PostAuthor returns the id of the user who owns the post, or ErrPostNotFound.
+func (r *CommentRepo) PostAuthor(ctx context.Context, postID string) (string, error) {
+	var uid string
+	err := r.pg.Pool.QueryRow(ctx, `SELECT user_id FROM posts WHERE id = $1`, postID).Scan(&uid)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return "", ErrPostNotFound
+	}
+	if err != nil {
+		return "", fmt.Errorf("post author: %w", err)
+	}
+	return uid, nil
+}
+
+// Create inserts a comment. parentID is the thread root (nil for a top-level
+// comment); replyToUserID is who a reply addresses (nil for top-level).
+// Returns ErrDuplicateComment when the same user re-posts identical text in
+// the same scope. The returned Comment carries author fields.
+func (r *CommentRepo) Create(ctx context.Context, postID, userID string, parentID, replyToUserID *string, body string) (*model.Comment, error) {
+	var id string
+	err := r.pg.Pool.QueryRow(ctx, `
+		INSERT INTO post_comments (post_id, user_id, parent_id, reply_to_user_id, body)
+		VALUES ($1, $2, $3, $4, $5)
+		ON CONFLICT (post_id, user_id, COALESCE(parent_id, '00000000-0000-0000-0000-000000000000'::uuid), md5(body))
+		DO NOTHING
+		RETURNING id`,
+		postID, userID, parentID, replyToUserID, body,
+	).Scan(&id)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return nil, ErrDuplicateComment
+	}
+	if err != nil {
+		return nil, fmt.Errorf("insert comment: %w", err)
+	}
+	// The author hasn't liked their own brand-new comment, so viewer state is
+	// resolved against userID (yields false) — fine for the create response.
+	return r.Get(ctx, id, userID)
+}
+
+// Get returns one comment hydrated with author + viewer-like + reply count.
+func (r *CommentRepo) Get(ctx context.Context, id, viewerID string) (*model.Comment, error) {
+	q := `SELECT ` + commentSelect + commentFrom + ` WHERE c.id = $1`
+	c := &model.Comment{}
+	err := scanComment(r.pg.Pool.QueryRow(ctx, q, id, nullUUIDArg(viewerID)), c)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return nil, ErrCommentNotFound
+	}
+	if err != nil {
+		return nil, fmt.Errorf("get comment: %w", err)
+	}
+	return c, nil
+}
+
+// ListTopLevel returns a page of a post's top-level comments — pinned first,
+// then newest — plus the total top-level count.
+func (r *CommentRepo) ListTopLevel(ctx context.Context, postID, viewerID string, limit, offset int) ([]model.Comment, int, error) {
+	var total int
+	if err := r.pg.Pool.QueryRow(ctx,
+		`SELECT count(*) FROM post_comments WHERE post_id = $1 AND parent_id IS NULL`,
+		postID).Scan(&total); err != nil {
+		return nil, 0, fmt.Errorf("count comments: %w", err)
+	}
+	if total == 0 {
+		return []model.Comment{}, 0, nil
+	}
+	q := `SELECT ` + commentSelect + commentFrom + `
+		WHERE c.post_id = $1 AND c.parent_id IS NULL
+		ORDER BY (c.pinned_at IS NOT NULL) DESC, c.pinned_at DESC NULLS LAST, c.created_at DESC
+		LIMIT $3 OFFSET $4`
+	rows, err := r.pg.Pool.Query(ctx, q, postID, nullUUIDArg(viewerID), limit, offset)
+	if err != nil {
+		return nil, 0, fmt.Errorf("list comments: %w", err)
+	}
+	out, err := scanComments(rows)
+	return out, total, err
+}
+
+// ListReplies returns a page of a thread's replies (oldest first) + the total.
+func (r *CommentRepo) ListReplies(ctx context.Context, parentID, viewerID string, limit, offset int) ([]model.Comment, int, error) {
+	var total int
+	if err := r.pg.Pool.QueryRow(ctx,
+		`SELECT count(*) FROM post_comments WHERE parent_id = $1`, parentID).Scan(&total); err != nil {
+		return nil, 0, fmt.Errorf("count replies: %w", err)
+	}
+	if total == 0 {
+		return []model.Comment{}, 0, nil
+	}
+	q := `SELECT ` + commentSelect + commentFrom + `
+		WHERE c.parent_id = $1
+		ORDER BY c.created_at ASC
+		LIMIT $3 OFFSET $4`
+	rows, err := r.pg.Pool.Query(ctx, q, parentID, nullUUIDArg(viewerID), limit, offset)
+	if err != nil {
+		return nil, 0, fmt.Errorf("list replies: %w", err)
+	}
+	out, err := scanComments(rows)
+	return out, total, err
+}
+
+// HydratePinned attaches each post's pinned comment (if any) to
+// Post.PinnedComment in a single query for the page.
+func (r *CommentRepo) HydratePinned(ctx context.Context, viewerID string, posts []*model.Post) error {
+	ids := make([]string, 0, len(posts))
+	for _, p := range posts {
+		if p != nil {
+			ids = append(ids, p.ID)
+		}
+	}
+	if len(ids) == 0 {
+		return nil
+	}
+	q := `SELECT ` + commentSelect + commentFrom + `
+		WHERE c.post_id = ANY($1) AND c.pinned_at IS NOT NULL`
+	rows, err := r.pg.Pool.Query(ctx, q, ids, nullUUIDArg(viewerID))
+	if err != nil {
+		return fmt.Errorf("hydrate pinned: %w", err)
+	}
+	list, err := scanComments(rows)
+	if err != nil {
+		return err
+	}
+	byPost := make(map[string]*model.Comment, len(list))
+	for i := range list {
+		byPost[list[i].PostID] = &list[i]
+	}
+	for _, p := range posts {
+		if p == nil {
+			continue
+		}
+		if c, ok := byPost[p.ID]; ok {
+			p.PinnedComment = c
+		}
+	}
+	return nil
+}
+
+// Like / Unlike toggle the viewer's like on a comment and return the new
+// likes_count. Idempotent; the membership change and counter update happen in
+// one transaction so concurrent likes stay consistent.
+func (r *CommentRepo) Like(ctx context.Context, commentID, userID string) (int, error) {
+	return r.setLike(ctx, commentID, userID, true)
+}
+
+func (r *CommentRepo) Unlike(ctx context.Context, commentID, userID string) (int, error) {
+	return r.setLike(ctx, commentID, userID, false)
+}
+
+func (r *CommentRepo) setLike(ctx context.Context, commentID, userID string, like bool) (int, error) {
+	tx, err := r.pg.Pool.Begin(ctx)
+	if err != nil {
+		return 0, fmt.Errorf("begin like txn: %w", err)
+	}
+	defer tx.Rollback(ctx)
+
+	// Lock the comment row up front: this validates existence (clean 404
+	// instead of a leaked FK violation) and serializes concurrent likes so
+	// the counter stays exact.
+	var count int
+	if err := tx.QueryRow(ctx,
+		`SELECT likes_count FROM post_comments WHERE id = $1 FOR UPDATE`, commentID).Scan(&count); err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			return 0, ErrCommentNotFound
+		}
+		return 0, fmt.Errorf("read comment likes_count: %w", err)
+	}
+
+	if like {
+		tag, err := tx.Exec(ctx,
+			`INSERT INTO comment_likes (comment_id, user_id) VALUES ($1, $2) ON CONFLICT DO NOTHING`,
+			commentID, userID)
+		if err != nil {
+			return 0, fmt.Errorf("insert comment like: %w", err)
+		}
+		if tag.RowsAffected() > 0 {
+			count++
+			if _, err := tx.Exec(ctx,
+				`UPDATE post_comments SET likes_count = $2 WHERE id = $1`, commentID, count); err != nil {
+				return 0, fmt.Errorf("bump comment likes: %w", err)
+			}
+		}
+	} else {
+		tag, err := tx.Exec(ctx,
+			`DELETE FROM comment_likes WHERE comment_id = $1 AND user_id = $2`, commentID, userID)
+		if err != nil {
+			return 0, fmt.Errorf("delete comment like: %w", err)
+		}
+		if tag.RowsAffected() > 0 {
+			if count > 0 {
+				count--
+			}
+			if _, err := tx.Exec(ctx,
+				`UPDATE post_comments SET likes_count = $2 WHERE id = $1`, commentID, count); err != nil {
+				return 0, fmt.Errorf("drop comment likes: %w", err)
+			}
+		}
+	}
+
+	if err := tx.Commit(ctx); err != nil {
+		return 0, fmt.Errorf("commit like txn: %w", err)
+	}
+	return count, nil
+}
+
+// Pin makes commentID the post's single pinned comment (top-level only),
+// unpinning any previously pinned comment in the same transaction. Returns
+// ErrCommentNotFound when the comment isn't a top-level comment of postID.
+func (r *CommentRepo) Pin(ctx context.Context, postID, commentID string) error {
+	tx, err := r.pg.Pool.Begin(ctx)
+	if err != nil {
+		return fmt.Errorf("begin pin txn: %w", err)
+	}
+	defer tx.Rollback(ctx)
+
+	if _, err := tx.Exec(ctx,
+		`UPDATE post_comments SET pinned_at = NULL WHERE post_id = $1 AND pinned_at IS NOT NULL`,
+		postID); err != nil {
+		return fmt.Errorf("unpin existing: %w", err)
+	}
+	tag, err := tx.Exec(ctx,
+		`UPDATE post_comments SET pinned_at = now(), updated_at = now()
+		 WHERE id = $1 AND post_id = $2 AND parent_id IS NULL`,
+		commentID, postID)
+	if err != nil {
+		return fmt.Errorf("pin comment: %w", err)
+	}
+	if tag.RowsAffected() == 0 {
+		return ErrCommentNotFound
+	}
+	return tx.Commit(ctx)
+}
+
+// Unpin clears the pin on commentID for postID.
+func (r *CommentRepo) Unpin(ctx context.Context, postID, commentID string) error {
+	tag, err := r.pg.Pool.Exec(ctx,
+		`UPDATE post_comments SET pinned_at = NULL WHERE id = $1 AND post_id = $2`,
+		commentID, postID)
+	if err != nil {
+		return fmt.Errorf("unpin comment: %w", err)
+	}
+	if tag.RowsAffected() == 0 {
+		return ErrCommentNotFound
+	}
+	return nil
+}
+
+// Delete removes the comment. For a thread root this cascades to its replies
+// (and all comment_likes) via ON DELETE CASCADE. Authorization is the caller's
+// responsibility.
+func (r *CommentRepo) Delete(ctx context.Context, commentID string) error {
+	tag, err := r.pg.Pool.Exec(ctx, `DELETE FROM post_comments WHERE id = $1`, commentID)
+	if err != nil {
+		return fmt.Errorf("delete comment: %w", err)
+	}
+	if tag.RowsAffected() == 0 {
+		return ErrCommentNotFound
+	}
+	return nil
+}
+
+// CreateReport files a report against a comment.
+func (r *CommentRepo) CreateReport(ctx context.Context, reporterID, commentID, category, reason string) error {
+	if _, err := r.pg.Pool.Exec(ctx,
+		`INSERT INTO comment_reports (reporter_id, comment_id, category, reason)
+		 VALUES ($1, $2, $3, $4)`,
+		reporterID, commentID, category, reason,
+	); err != nil {
+		return fmt.Errorf("create comment report: %w", err)
+	}
+	return nil
+}
@@ -324,9 +324,12 @@ func (r *NotificationRepo) ListByUser(ctx context.Context, userID string, limit,
 	rows, err := r.pg.Pool.Query(ctx, `
 		SELECT n.id, n.user_id, n.actor_id, n.kind, n.post_id, n.actor_count,
 		       n.created_at, n.read_at,
-		       u.name, u.handle, u.avatar_url
+		       u.name, u.handle, u.avatar_url,
+		       pu.handle
 		FROM notifications n
 		LEFT JOIN users u ON u.id = n.actor_id
+		LEFT JOIN posts p ON p.id = n.post_id
+		LEFT JOIN users pu ON pu.id = p.user_id
 		WHERE n.user_id = $1
 		  AND NOT EXISTS (
 		      SELECT 1 FROM notification_prefs np
@@ -344,7 +347,8 @@ func (r *NotificationRepo) ListByUser(ctx context.Context, userID string, limit,
 		var n model.Notification
 		if err := rows.Scan(&n.ID, &n.UserID, &n.ActorID, &n.Kind, &n.PostID, &n.ActorCount,
 			&n.CreatedAt, &n.ReadAt,
-			&n.ActorName, &n.ActorHandle, &n.ActorAvatar); err != nil {
+			&n.ActorName, &n.ActorHandle, &n.ActorAvatar,
+			&n.PostAuthorHandle); err != nil {
 			return nil, fmt.Errorf("scan notification: %w", err)
 		}
 		out = append(out, n)
@@ -400,6 +404,36 @@ func (r *NotificationRepo) CreateMention(ctx context.Context, userID, actorID, p
 	return nil
 }

+// CreateComment persists a comment/reply notification linked to the post.
+// kind is 'post_comment' (to the post author) or 'comment_reply' (to the
+// replied-to comment's author). Like mentions, no aggregation.
+func (r *NotificationRepo) CreateComment(ctx context.Context, userID, actorID, postID, kind string) error {
+	if _, err := r.pg.Pool.Exec(ctx,
+		`INSERT INTO notifications (user_id, actor_id, kind, post_id)
+		 VALUES ($1, $2, $3, $4)`,
+		userID, actorID, kind, postID,
+	); err != nil {
+		return fmt.Errorf("create comment notification: %w", err)
+	}
+	return nil
+}
+
+// PostAuthorHandle returns the handle of the user who owns the post, for
+// building a deep-link. Empty string (no error) when the post is gone.
+func (r *NotificationRepo) PostAuthorHandle(ctx context.Context, postID string) (string, error) {
+	var h string
+	err := r.pg.Pool.QueryRow(ctx,
+		`SELECT u.handle FROM posts p JOIN users u ON u.id = p.user_id WHERE p.id = $1`,
+		postID).Scan(&h)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return "", nil
+	}
+	if err != nil {
+		return "", fmt.Errorf("post author handle: %w", err)
+	}
+	return h, nil
+}
+
 // CountUnread returns how many of a recipient's notifications are unread,
 // excluding kinds the user has turned off for the in-app channel (matching
 // ListByUser, so the badge never counts rows the feed won't show).