spsays-bff/internal/repository/like_repo.go

package repository

import (
	"context"
	"fmt"
	"strconv"
	"strings"

	"github.com/prabhupadasays/spsays-bff/internal/model"
	"go.uber.org/zap"
)

// likeDeltaPrefix namespaces the per-post pending-like counters in Redis. Each
// like/unlike INCRBY/DECRBYs likes:delta:{postID}; the LikesWorker folds these
// into posts.likes_count every 15 minutes (GETDEL + UPDATE). Reads add the
// still-pending delta on top of likes_count so the displayed total is current.
const likeDeltaPrefix = "likes:delta:"

// LikeRepo handles post-like membership (Postgres) and the write-behind count
// delta (Redis). Membership is written synchronously so viewer state and the
// "liked by" list stay consistent; only the hot counter is batched.
//
// `redis` is optional: when nil (Redis unavailable) likes still work, but the
// count is applied to posts.likes_count directly on each like instead of being
// batched.
type LikeRepo struct {
	pg     *Postgres
	redis  *Redis
	logger *zap.Logger
}

func NewLikeRepo(pg *Postgres, redis *Redis, logger *zap.Logger) *LikeRepo {
	return &LikeRepo{pg: pg, redis: redis, logger: logger}
}

func (r *LikeRepo) deltaKey(postID string) string { return likeDeltaPrefix + postID }

// Like records a membership row. Returns whether it was newly liked (false if
// the user had already liked it — idempotent). On a real change it nudges the
// pending count up.
func (r *LikeRepo) Like(ctx context.Context, postID, userID string) (bool, error) {
	tag, err := r.pg.Pool.Exec(ctx,
		`INSERT INTO post_likes (post_id, user_id) VALUES ($1, $2)
		 ON CONFLICT (post_id, user_id) DO NOTHING`,
		postID, userID,
	)
	if err != nil {
		return false, fmt.Errorf("insert like: %w", err)
	}
	if tag.RowsAffected() == 0 {
		return false, nil // already liked
	}
	r.bumpDelta(ctx, postID, 1)
	return true, nil
}

// Unlike removes the membership row. Returns whether it actually changed.
func (r *LikeRepo) Unlike(ctx context.Context, postID, userID string) (bool, error) {
	tag, err := r.pg.Pool.Exec(ctx,
		`DELETE FROM post_likes WHERE post_id = $1 AND user_id = $2`,
		postID, userID,
	)
	if err != nil {
		return false, fmt.Errorf("delete like: %w", err)
	}
	if tag.RowsAffected() == 0 {
		return false, nil // wasn't liked
	}
	r.bumpDelta(ctx, postID, -1)
	return true, nil
}

// bumpDelta adjusts the pending count. Prefers the Redis write-behind counter;
// falls back to a direct DB update when Redis is unavailable so the count is
// never lost.
func (r *LikeRepo) bumpDelta(ctx context.Context, postID string, d int64) {
	if r.redis != nil {
		if err := r.redis.Client.IncrBy(ctx, r.deltaKey(postID), d).Err(); err == nil {
			return
		}
		r.logger.Warn("likes: redis incr failed, applying to DB", zap.String("post", postID))
	}
	if _, err := r.pg.Pool.Exec(ctx,
		`UPDATE posts SET likes_count = GREATEST(0, likes_count + $2) WHERE id = $1`,
		postID, d,
	); err != nil {
		r.logger.Warn("likes: direct count update failed", zap.String("post", postID), zap.Error(err))
	}
}

// PostAuthor returns the user id that owns the post (for like notifications).
func (r *LikeRepo) PostAuthor(ctx context.Context, postID string) (string, error) {
	var uid string
	if err := r.pg.Pool.QueryRow(ctx,
		`SELECT user_id FROM posts WHERE id = $1`, postID,
	).Scan(&uid); err != nil {
		return "", fmt.Errorf("post author: %w", err)
	}
	return uid, nil
}

// EffectiveCount returns likes_count + the pending Redis delta for one post.
func (r *LikeRepo) EffectiveCount(ctx context.Context, postID string) (int, error) {
	var c int
	if err := r.pg.Pool.QueryRow(ctx,
		`SELECT likes_count FROM posts WHERE id = $1`, postID,
	).Scan(&c); err != nil {
		return 0, fmt.Errorf("read likes_count: %w", err)
	}
	if r.redis != nil {
		if d, err := r.redis.Client.Get(ctx, r.deltaKey(postID)).Int(); err == nil {
			c += d
		}
	}
	if c < 0 {
		c = 0
	}
	return c, nil
}

// Hydrate fills LikesCount (effective) and ViewerHasLiked on the given posts.
// Used by every post read path. viewerID "" leaves ViewerHasLiked false.
func (r *LikeRepo) Hydrate(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
	}

	// Denormalized counts from Postgres.
	counts := make(map[string]int, len(ids))
	rows, err := r.pg.Pool.Query(ctx,
		`SELECT id, likes_count FROM posts WHERE id = ANY($1)`, ids)
	if err != nil {
		return fmt.Errorf("hydrate like counts: %w", err)
	}
	for rows.Next() {
		var id string
		var c int
		if err := rows.Scan(&id, &c); err != nil {
			rows.Close()
			return fmt.Errorf("scan like count: %w", err)
		}
		counts[id] = c
	}
	rows.Close()
	if err := rows.Err(); err != nil {
		return err
	}

	// Pending deltas from Redis (best-effort).
	deltas := r.pendingDeltas(ctx, ids)

	// Viewer membership.
	liked := make(map[string]bool)
	if viewerID != "" {
		lr, err := r.pg.Pool.Query(ctx,
			`SELECT post_id FROM post_likes WHERE user_id = $1 AND post_id = ANY($2)`,
			viewerID, ids)
		if err != nil {
			return fmt.Errorf("hydrate viewer likes: %w", err)
		}
		for lr.Next() {
			var pid string
			if err := lr.Scan(&pid); err != nil {
				lr.Close()
				return fmt.Errorf("scan viewer like: %w", err)
			}
			liked[pid] = true
		}
		lr.Close()
		if err := lr.Err(); err != nil {
			return err
		}
	}

	for _, p := range posts {
		if p == nil {
			continue
		}
		c := counts[p.ID] + deltas[p.ID]
		if c < 0 {
			c = 0
		}
		p.LikesCount = c
		p.ViewerHasLiked = liked[p.ID]
	}
	return nil
}

// pendingDeltas returns the Redis like-deltas for the given posts (absent →
// zero). Best-effort: Redis errors yield an empty map.
func (r *LikeRepo) pendingDeltas(ctx context.Context, ids []string) map[string]int {
	out := make(map[string]int)
	if r.redis == nil || len(ids) == 0 {
		return out
	}
	keys := make([]string, len(ids))
	for i, id := range ids {
		keys[i] = r.deltaKey(id)
	}
	vals, err := r.redis.Client.MGet(ctx, keys...).Result()
	if err != nil {
		return out
	}
	for i, v := range vals {
		s, ok := v.(string)
		if !ok {
			continue
		}
		if n, err := strconv.Atoi(s); err == nil && n != 0 {
			out[ids[i]] = n
		}
	}
	return out
}

// ListLikers returns one page of users who liked the post (newest first),
// optionally filtered by a name/handle search, plus the total matching count.
func (r *LikeRepo) ListLikers(ctx context.Context, postID, q string, limit, offset int) ([]model.UserSummary, int, error) {
	like := "%" + q + "%"

	var total int
	if err := r.pg.Pool.QueryRow(ctx, `
		SELECT count(*)
		FROM post_likes l JOIN users u ON u.id = l.user_id
		WHERE l.post_id = $1
		  AND ($2 = '' OR u.name ILIKE $3 OR u.handle ILIKE $3)`,
		postID, q, like,
	).Scan(&total); err != nil {
		return nil, 0, fmt.Errorf("count likers: %w", err)
	}
	if total == 0 {
		return []model.UserSummary{}, 0, nil
	}

	rows, err := r.pg.Pool.Query(ctx, `
		SELECT u.handle, u.name, u.avatar_url
		FROM post_likes l JOIN users u ON u.id = l.user_id
		WHERE l.post_id = $1
		  AND ($2 = '' OR u.name ILIKE $3 OR u.handle ILIKE $3)
		ORDER BY l.created_at DESC, u.handle ASC
		LIMIT $4 OFFSET $5`,
		postID, q, like, limit, offset,
	)
	if err != nil {
		return nil, 0, fmt.Errorf("list likers: %w", err)
	}
	defer rows.Close()

	out := make([]model.UserSummary, 0, limit)
	for rows.Next() {
		var s model.UserSummary
		if err := rows.Scan(&s.Handle, &s.Name, &s.AvatarURL); err != nil {
			return nil, 0, fmt.Errorf("scan liker: %w", err)
		}
		out = append(out, s)
	}
	return out, total, rows.Err()
}

// FlushDeltas drains every pending Redis like-delta into posts.likes_count.
// Called periodically by the LikesWorker. Returns how many posts were updated.
// GETDEL makes each pop atomic; a failed UPDATE restores the delta so it's
// retried next cycle.
func (r *LikeRepo) FlushDeltas(ctx context.Context) (int, error) {
	if r.redis == nil {
		return 0, nil
	}
	var cursor uint64
	applied := 0
	for {
		keys, next, err := r.redis.Client.Scan(ctx, cursor, likeDeltaPrefix+"*", 200).Result()
		if err != nil {
			return applied, fmt.Errorf("scan like deltas: %w", err)
		}
		for _, key := range keys {
			postID := strings.TrimPrefix(key, likeDeltaPrefix)
			v, err := r.redis.Client.GetDel(ctx, key).Int()
			if err != nil || v == 0 {
				continue
			}
			if _, err := r.pg.Pool.Exec(ctx,
				`UPDATE posts SET likes_count = GREATEST(0, likes_count + $2) WHERE id = $1`,
				postID, v,
			); err != nil {
				// Restore so the delta isn't lost; retried next cycle.
				r.redis.Client.IncrBy(ctx, key, int64(v))
				r.logger.Warn("likes flush: update failed, delta restored",
					zap.String("post", postID), zap.Error(err))
				continue
			}
			applied++
		}
		cursor = next
		if cursor == 0 {
			break
		}
	}
	return applied, nil
}