From e9299051f5cdef2ace0c5d6958f6f50a3778f672 Mon Sep 17 00:00:00 2001 From: Krispire Admin Date: Tue, 9 Jun 2026 16:28:40 +0530 Subject: [PATCH] feat: add post like --- internal/handler/social.go | 61 +++- internal/model/social.go | 14 + internal/repository/like_repo.go | 305 ++++++++++++++++++ internal/repository/social_graph_repo.go | 29 +- internal/server/server.go | 20 +- internal/service/notification_service.go | 72 +++++ internal/service/social_service.go | 65 +++- internal/worker/likes_worker.go | 55 ++++ migrations/0030_create_post_likes.down.sql | 2 + migrations/0030_create_post_likes.up.sql | 26 ++ ...0031_notifications_like_aggregate.down.sql | 3 + .../0031_notifications_like_aggregate.up.sql | 14 + 12 files changed, 658 insertions(+), 8 deletions(-) create mode 100644 internal/repository/like_repo.go create mode 100644 internal/worker/likes_worker.go create mode 100644 migrations/0030_create_post_likes.down.sql create mode 100644 migrations/0030_create_post_likes.up.sql create mode 100644 migrations/0031_notifications_like_aggregate.down.sql create mode 100644 migrations/0031_notifications_like_aggregate.up.sql diff --git a/internal/handler/social.go b/internal/handler/social.go index c1c2e6b..090e0bc 100644 --- a/internal/handler/social.go +++ b/internal/handler/social.go @@ -125,6 +125,51 @@ func (h *PostHandler) Get(w http.ResponseWriter, r *http.Request) { response.JSON(w, http.StatusOK, p) } +// likeResp is returned by Like / Unlike — the new viewer state + effective +// total (likes_count + pending Redis delta) so the client can reconcile. +type likeResp struct { + Liked bool `json:"liked"` + LikesCount int `json:"likes_count"` +} + +// Like — POST /posts/{id}/like. Idempotent. RequireAuth. +func (h *PostHandler) Like(w http.ResponseWriter, r *http.Request) { + uid := middleware.UserID(r.Context()) + id := chi.URLParam(r, "id") + count, err := h.svc.Like(r.Context(), uid, id) + if err != nil { + mapSocialErr(w, err) + return + } + response.JSON(w, http.StatusOK, likeResp{Liked: true, LikesCount: count}) +} + +// Unlike — DELETE /posts/{id}/like. Idempotent. RequireAuth. +func (h *PostHandler) Unlike(w http.ResponseWriter, r *http.Request) { + uid := middleware.UserID(r.Context()) + id := chi.URLParam(r, "id") + count, err := h.svc.Unlike(r.Context(), uid, id) + if err != nil { + mapSocialErr(w, err) + return + } + response.JSON(w, http.StatusOK, likeResp{Liked: false, LikesCount: count}) +} + +// ListLikes — GET /posts/{id}/likes?limit=&offset=&q=. Paginated "liked by" +// users with optional name/handle search. Public (OptionalAuth). +func (h *PostHandler) ListLikes(w http.ResponseWriter, r *http.Request) { + id := chi.URLParam(r, "id") + limit, offset := parseLimitOffset(r) + q := strings.TrimSpace(r.URL.Query().Get("q")) + users, total, err := h.svc.Likers(r.Context(), id, q, limit, offset) + if err != nil { + mapSocialErr(w, err) + return + } + response.JSON(w, http.StatusOK, followListResp{Users: users, Total: total}) +} + func (h *PostHandler) Delete(w http.ResponseWriter, r *http.Request) { uid := middleware.UserID(r.Context()) if err := h.svc.Delete(r.Context(), chi.URLParam(r, "id"), uid); err != nil { @@ -443,10 +488,11 @@ type ProfileHandler struct { posts *repository.PostRepo collections *repository.CollectionRepo graph *service.SocialGraphService + likes *repository.LikeRepo } -func NewProfileHandler(users *repository.UserRepo, posts *repository.PostRepo, collections *repository.CollectionRepo, graph *service.SocialGraphService) *ProfileHandler { - return &ProfileHandler{users: users, posts: posts, collections: collections, graph: graph} +func NewProfileHandler(users *repository.UserRepo, posts *repository.PostRepo, collections *repository.CollectionRepo, graph *service.SocialGraphService, likes *repository.LikeRepo) *ProfileHandler { + return &ProfileHandler{users: users, posts: posts, collections: collections, graph: graph, likes: likes} } func (h *ProfileHandler) Get(w http.ResponseWriter, r *http.Request) { @@ -527,6 +573,17 @@ func (h *ProfileHandler) Feed(w http.ResponseWriter, r *http.Request) { } } } + // Like counts (effective) + viewer-liked state for every post in the page. + if len(posts) > 0 { + ptrs := make([]*model.Post, len(posts)) + for i := range posts { + ptrs[i] = &posts[i] + } + if err := h.likes.Hydrate(r.Context(), viewerID, ptrs); err != nil { + mapSocialErr(w, err) + return + } + } response.JSONWithMeta(w, http.StatusOK, posts, paginationMeta(total, limit, offset)) } diff --git a/internal/model/social.go b/internal/model/social.go index c490b82..1420fc8 100644 --- a/internal/model/social.go +++ b/internal/model/social.go @@ -39,6 +39,13 @@ type Post struct { // after the repo returns posts; always false for anonymous viewers. // Drives the post-card "Reshare ↔ Send" swap on the frontend. ViewerHasReshared bool `json:"viewer_has_reshared"` + + // Likes. LikesCount is the effective total = the denormalized + // posts.likes_count plus any still-pending Redis delta (hydrated by the + // like layer). ViewerHasLiked reflects the calling user's membership row; + // false for anonymous viewers. + LikesCount int `json:"likes_count"` + ViewerHasLiked bool `json:"viewer_has_liked"` } // PostImage represents one attached image on a post. @@ -167,6 +174,13 @@ type Notification struct { CreatedAt time.Time `json:"created_at"` ReadAt *time.Time `json:"read_at,omitempty"` + // Aggregation: for post_like, PostID is the liked post and ActorCount is + // how many likers have aggregated into this single row (the frontend + // renders "X and N-1 others liked your post"). Other kinds: PostID nil, + // ActorCount 1. + PostID *string `json:"post_id,omitempty"` + ActorCount int `json:"actor_count"` + ActorName *string `json:"actor_name,omitempty"` ActorHandle *string `json:"actor_handle,omitempty"` ActorAvatar *string `json:"actor_avatar_url,omitempty"` diff --git a/internal/repository/like_repo.go b/internal/repository/like_repo.go new file mode 100644 index 0000000..125a0bf --- /dev/null +++ b/internal/repository/like_repo.go @@ -0,0 +1,305 @@ +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 +} diff --git a/internal/repository/social_graph_repo.go b/internal/repository/social_graph_repo.go index 0c95e15..c66aa8a 100644 --- a/internal/repository/social_graph_repo.go +++ b/internal/repository/social_graph_repo.go @@ -319,7 +319,8 @@ func (r *NotificationRepo) Create(ctx context.Context, userID string, actorID *s // notifications or since-deleted actors). func (r *NotificationRepo) ListByUser(ctx context.Context, userID string, limit, offset int) ([]model.Notification, error) { rows, err := r.pg.Pool.Query(ctx, ` - SELECT n.id, n.user_id, n.actor_id, n.kind, n.created_at, n.read_at, + 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 FROM notifications n LEFT JOIN users u ON u.id = n.actor_id @@ -334,7 +335,8 @@ func (r *NotificationRepo) ListByUser(ctx context.Context, userID string, limit, out := []model.Notification{} for rows.Next() { var n model.Notification - if err := rows.Scan(&n.ID, &n.UserID, &n.ActorID, &n.Kind, &n.CreatedAt, &n.ReadAt, + 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 { return nil, fmt.Errorf("scan notification: %w", err) } @@ -343,6 +345,29 @@ func (r *NotificationRepo) ListByUser(ctx context.Context, userID string, limit, return out, rows.Err() } +// UpsertLike records (or aggregates) a like notification: one row per +// (recipient, post). On a repeat like it refreshes the most-recent actor, +// bumps actor_count, and re-marks the row unread + most-recent. Returns the +// new actor_count. +func (r *NotificationRepo) UpsertLike(ctx context.Context, userID, actorID, postID string) (int, error) { + var count int + err := r.pg.Pool.QueryRow(ctx, ` + INSERT INTO notifications (user_id, actor_id, kind, post_id, actor_count) + VALUES ($1, $2, 'post_like', $3, 1) + ON CONFLICT (user_id, post_id) WHERE kind = 'post_like' + DO UPDATE SET actor_id = EXCLUDED.actor_id, + actor_count = notifications.actor_count + 1, + read_at = NULL, + created_at = now() + RETURNING actor_count`, + userID, actorID, postID, + ).Scan(&count) + if err != nil { + return 0, fmt.Errorf("upsert like notification: %w", err) + } + return count, nil +} + // CountUnread returns how many of a recipient's notifications are unread. func (r *NotificationRepo) CountUnread(ctx context.Context, userID string) (int, error) { var n int diff --git a/internal/server/server.go b/internal/server/server.go index c6c96ef..f40ec5d 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -162,7 +162,10 @@ func (s *Server) initAuth(cfg *config.Config, logger *zap.Logger) { postRepo := repository.NewPostRepo(s.pg) bookmarkRepo := repository.NewBookmarkRepo(s.pg) collectionRepo := repository.NewCollectionRepo(s.pg) - postSvc := service.NewPostService(postRepo, r2) + // Likes: membership in Postgres, count delta batched through Redis + // (s.redis may be nil — LikeRepo then applies counts directly). + likeRepo := repository.NewLikeRepo(s.pg, s.redis, logger) + postSvc := service.NewPostService(postRepo, r2, likeRepo) bookmarkSvc := service.NewBookmarkService(userRepo, bookmarkRepo) collectionSvc := service.NewCollectionService(userRepo, collectionRepo, postRepo, r2) @@ -210,6 +213,9 @@ func (s *Server) initAuth(cfg *config.Config, logger *zap.Logger) { pushSender := service.NewPushSender(pushRepo, cfg.VAPIDPublicKey, cfg.VAPIDPrivateKey, cfg.VAPIDSubject, logger) s.pushH = handler.NewPushHandler(pushSender, pushRepo) notifSvc := service.NewNotificationService(notifRepo, userRepo, mail, rtClient, pushSender, logger, cfg.FrontendURL) + // Wire the notification seam into the post service so a new like notifies + // the author (push + realtime; not the liker themselves). + postSvc.SetNotifier(notifSvc) // Sādhana reminders: a background worker nudges users (email + push + // websocket) to chant / post a quote at their local time. Shares the @@ -223,11 +229,15 @@ func (s *Server) initAuth(cfg *config.Config, logger *zap.Logger) { } worker.NewReminderWorker(reminderSvc, logger).Run(wctx) + // Likes worker: folds the pending Redis like-deltas into posts.likes_count + // every 15 minutes (no-op when Redis is unavailable). + worker.NewLikesWorker(likeRepo, logger).Run(wctx) + graphSvc := service.NewSocialGraphService(followRepo, blockRepo, reportRepo, postRepo, userRepo, notifSvc, mail, cfg.FeedbackAdminEmails, logger) s.relationH = handler.NewRelationshipHandler(graphSvc) s.notificationH = handler.NewNotificationHandler(notifSvc) s.postH = handler.NewPostHandler(postSvc, graphSvc) - s.profileH = handler.NewProfileHandler(userRepo, postRepo, collectionRepo, graphSvc) + s.profileH = handler.NewProfileHandler(userRepo, postRepo, collectionRepo, graphSvc, likeRepo) // Sādhana streak: daily chanting log + the growing-plant state. Reads // the chanting half from sadhna_logs and the posting half from the @@ -631,6 +641,8 @@ func (s *Server) setupRouter() { r.Group(func(r chi.Router) { r.Use(middleware.OptionalAuth(s.authIssuer, s.userRepo)) r.Get("/{id}", s.postH.Get) + // Paginated "liked by" list — public, like the count itself. + r.Get("/{id}/likes", s.postH.ListLikes) }) // Protected — create / delete / image presign / "my posts" / // reshare actions. @@ -646,6 +658,10 @@ func (s *Server) setupRouter() { // double-clicks don't 4xx. r.Post("/{id}/reshare", s.postH.Reshare) r.Delete("/{id}/reshare", s.postH.Unreshare) + // Like / unlike — idempotent; membership is written now, the + // count is batched into likes_count via Redis. + r.Post("/{id}/like", s.postH.Like) + r.Delete("/{id}/like", s.postH.Unlike) // Report a post (category + optional reason) → admin email. r.Post("/{id}/report", s.postH.Report) }) diff --git a/internal/service/notification_service.go b/internal/service/notification_service.go index bb4148e..b64d70d 100644 --- a/internal/service/notification_service.go +++ b/internal/service/notification_service.go @@ -15,6 +15,7 @@ package service import ( "bytes" "context" + "fmt" "html/template" "strings" @@ -29,6 +30,7 @@ import ( const ( NotifFollowRequest = "follow_request" NotifFollowAccepted = "follow_accepted" + NotifPostLike = "post_like" // Reminder kinds are delivered live (email + push + a websocket toast) // but NOT persisted — they're ephemeral nudges, not history. @@ -96,6 +98,76 @@ func (s *NotificationService) NotifyFollowAccepted(ctx context.Context, recipien "%s accepted your follow request — Prabhupāda Says") } +// NotifyPostLike tells the post author that `likerID` liked their post. Sent +// only when the liker isn't the author (callers also guard). Unlike follows, +// likes are high-volume, so this delivers a persisted row + realtime nudge + +// Web Push but NO email. The push is tagged per-post so repeated likes on the +// same post collapse into one notification rather than stacking. +func (s *NotificationService) NotifyPostLike(ctx context.Context, postID, authorID, likerID string) { + if authorID == "" || likerID == "" || authorID == likerID { + return + } + // Aggregate into the single (author, post) like-notification row, getting + // back how many likers it now represents. + count, err := s.notifs.UpsertLike(ctx, authorID, likerID, postID) + if err != nil { + s.logger.Warn("aggregate like notification failed", + zap.String("recipient", authorID), zap.Error(err)) + count = 1 + } + + actorUser, _ := s.users.FindByID(ctx, likerID) + who := "Someone" + if actorUser != nil { + who = actorDisplay(actorUser) + } + body := likeMessage(who, count) + + // Realtime nudge — carry actor_count so the toast can read "and N others". + if s.rt != nil { + payload := map[string]any{"kind": NotifPostLike, "actor_count": count} + if actorUser != nil { + payload["actor_name"] = actorDisplay(actorUser) + payload["actor_handle"] = actorUser.Handle + payload["actor_avatar_url"] = actorUser.AvatarURL + } + if err := s.rt.Trigger(ctx, "notifications."+authorID, "notification", payload); err != nil { + s.logger.Warn("realtime like publish failed", zap.String("recipient", authorID), zap.Error(err)) + } + } + + // Web Push — tapping opens the author's own feed anchored to the post. + // Tagged per-post so repeated likes collapse into one notification. + if s.push == nil { + return + } + url := s.frontendURL + "/notifications" + if author, err := s.users.FindByID(ctx, authorID); err == nil && author != nil { + url = s.frontendURL + "/u/" + author.Handle + "/posts#post-" + postID + } + s.push.SendToUser(ctx, authorID, PushPayload{ + Title: "New Like", + Body: body, + URL: url, + Icon: s.iconURL(), + Tag: NotifPostLike + ":" + postID, + }) +} + +// likeMessage renders the aggregated like text: "X liked your post" or +// "X and N others liked your post". +func likeMessage(who string, count int) string { + if count <= 1 { + return who + " liked your post" + } + others := count - 1 + noun := "others" + if others == 1 { + noun = "other" + } + return fmt.Sprintf("%s and %d %s liked your post", who, others, noun) +} + // dispatch persists the notification and sends the matching email. All // failures are logged, never propagated. func (s *NotificationService) dispatch(ctx context.Context, recipientID, actorID, kind, subjectFmt string) { diff --git a/internal/service/social_service.go b/internal/service/social_service.go index 0b5d7b2..f5cd705 100644 --- a/internal/service/social_service.go +++ b/internal/service/social_service.go @@ -33,12 +33,18 @@ var ( type PostService struct { posts *repository.PostRepo r2 *repository.R2 + likes *repository.LikeRepo + notif *NotificationService // optional; set via SetNotifier } -func NewPostService(posts *repository.PostRepo, r2 *repository.R2) *PostService { - return &PostService{posts: posts, r2: r2} +func NewPostService(posts *repository.PostRepo, r2 *repository.R2, likes *repository.LikeRepo) *PostService { + return &PostService{posts: posts, r2: r2, likes: likes} } +// SetNotifier wires the notification seam after construction (the +// NotificationService is built later in server wiring than PostService). +func (s *PostService) SetNotifier(n *NotificationService) { s.notif = n } + // Create persists a new post. The caption is treated as opaque — we trust // the frontend to send the exact text the user selected. Source URL must be // a same-origin path (e.g. /books/bg/1/1?hl=…) — we strip the host if one @@ -72,6 +78,9 @@ func (s *PostService) Get(ctx context.Context, id, viewerID string) (*model.Post if err := s.hydrateReshareStates(ctx, viewerID, []*model.Post{p}); err != nil { return nil, err } + if err := s.likes.Hydrate(ctx, viewerID, []*model.Post{p}); err != nil { + return nil, err + } return p, nil } @@ -171,9 +180,58 @@ func (s *PostService) ListByHandle(ctx context.Context, handle, viewerID string, if err := s.hydrateReshareStatesSlice(ctx, viewerID, posts); err != nil { return nil, 0, err } + if err := s.hydrateLikesSlice(ctx, viewerID, posts); err != nil { + return nil, 0, err + } return posts, total, nil } +// hydrateLikesSlice fills LikesCount + ViewerHasLiked on a value slice. +func (s *PostService) hydrateLikesSlice(ctx context.Context, viewerID string, posts []model.Post) error { + if len(posts) == 0 { + return nil + } + ptrs := make([]*model.Post, len(posts)) + for i := range posts { + ptrs[i] = &posts[i] + } + return s.likes.Hydrate(ctx, viewerID, ptrs) +} + +// Like / Unlike toggle the caller's like on a post and return the effective +// total (likes_count + pending Redis delta). +func (s *PostService) Like(ctx context.Context, userID, postID string) (int, error) { + changed, err := s.likes.Like(ctx, postID, userID) + if err != nil { + return 0, err + } + count, err := s.likes.EffectiveCount(ctx, postID) + if err != nil { + return 0, err + } + // Notify the author on a genuinely new like by someone else. Fire-and- + // forget on a detached context so it isn't cancelled when the request + // returns; NotifyPostLike re-checks self-like internally too. + if changed && s.notif != nil { + if authorID, aerr := s.likes.PostAuthor(ctx, postID); aerr == nil && authorID != userID { + go s.notif.NotifyPostLike(context.Background(), postID, authorID, userID) + } + } + return count, nil +} + +func (s *PostService) Unlike(ctx context.Context, userID, postID string) (int, error) { + if _, err := s.likes.Unlike(ctx, postID, userID); err != nil { + return 0, err + } + return s.likes.EffectiveCount(ctx, postID) +} + +// Likers returns one page of users who liked the post + the total count. +func (s *PostService) Likers(ctx context.Context, postID, q string, limit, offset int) ([]model.UserSummary, int, error) { + return s.likes.ListLikers(ctx, postID, q, limit, offset) +} + // ListMine is the owner's view of their own profile feed. Every post // they own — authored or reshared — comes from posts.user_id, so the // viewer is by construction the owner. ViewerHasReshared is true for @@ -195,6 +253,9 @@ func (s *PostService) ListMine(ctx context.Context, userID string, limit, offset posts[i].ViewerHasReshared = true } } + if err := s.hydrateLikesSlice(ctx, userID, posts); err != nil { + return nil, 0, err + } return posts, total, nil } diff --git a/internal/worker/likes_worker.go b/internal/worker/likes_worker.go new file mode 100644 index 0000000..cd5a6cf --- /dev/null +++ b/internal/worker/likes_worker.go @@ -0,0 +1,55 @@ +package worker + +import ( + "context" + "time" + + "github.com/prabhupadasays/spsays-bff/internal/repository" + "go.uber.org/zap" +) + +// LikesWorker periodically folds the pending Redis like-deltas into +// posts.likes_count, so the hot counter is written in batches rather than on +// every like. It also does a best-effort final flush on shutdown so deltas +// aren't lost. +type LikesWorker struct { + repo *repository.LikeRepo + interval time.Duration + logger *zap.Logger +} + +func NewLikesWorker(repo *repository.LikeRepo, logger *zap.Logger) *LikesWorker { + return &LikesWorker{repo: repo, interval: 15 * time.Minute, logger: logger} +} + +// Run starts the flush loop in its own goroutine; it exits when ctx is done. +func (w *LikesWorker) Run(ctx context.Context) { + go func() { + w.logger.Info("likes worker started", zap.Duration("interval", w.interval)) + defer w.logger.Info("likes worker stopped") + + ticker := time.NewTicker(w.interval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + // Final flush so a graceful shutdown doesn't drop pending + // deltas. Fresh, short-lived context since ctx is cancelled. + flushCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + if _, err := w.repo.FlushDeltas(flushCtx); err != nil { + w.logger.Warn("likes worker: final flush failed", zap.Error(err)) + } + cancel() + return + case <-ticker.C: + n, err := w.repo.FlushDeltas(ctx) + if err != nil { + w.logger.Warn("likes worker: flush failed", zap.Error(err)) + } else if n > 0 { + w.logger.Info("likes worker: flushed", zap.Int("posts", n)) + } + } + } + }() +} diff --git a/migrations/0030_create_post_likes.down.sql b/migrations/0030_create_post_likes.down.sql new file mode 100644 index 0000000..0a8d26d --- /dev/null +++ b/migrations/0030_create_post_likes.down.sql @@ -0,0 +1,2 @@ +DROP TABLE IF EXISTS post_likes; +ALTER TABLE posts DROP COLUMN IF EXISTS likes_count; diff --git a/migrations/0030_create_post_likes.up.sql b/migrations/0030_create_post_likes.up.sql new file mode 100644 index 0000000..f0e4479 --- /dev/null +++ b/migrations/0030_create_post_likes.up.sql @@ -0,0 +1,26 @@ +-- Post likes. +-- +-- `post_likes` is the membership table (one row per user-per-post like), +-- written synchronously so a viewer's like state and the "liked by" list are +-- immediately consistent. The UNIQUE(post_id, user_id) makes liking idempotent. +-- +-- `posts.likes_count` is a denormalized counter that is NOT updated on every +-- like (that hot-row UPDATE doesn't scale for popular posts). Instead each +-- like/unlike nudges a per-post delta in Redis, and a background worker folds +-- the deltas into this column every 15 minutes. Reads return +-- likes_count + the still-pending Redis delta, so the displayed total is +-- always current. +ALTER TABLE posts ADD COLUMN IF NOT EXISTS likes_count INTEGER NOT NULL DEFAULT 0; + +CREATE TABLE IF NOT EXISTS post_likes ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + post_id UUID NOT NULL REFERENCES posts(id) ON DELETE CASCADE, + user_id UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + UNIQUE (post_id, user_id) +); + +-- Listing "who liked this post", newest first. +CREATE INDEX IF NOT EXISTS post_likes_post_idx ON post_likes (post_id, created_at DESC); +-- "Have I liked these posts?" — viewer-state hydration across a feed. +CREATE INDEX IF NOT EXISTS post_likes_user_idx ON post_likes (user_id); diff --git a/migrations/0031_notifications_like_aggregate.down.sql b/migrations/0031_notifications_like_aggregate.down.sql new file mode 100644 index 0000000..8dbbec3 --- /dev/null +++ b/migrations/0031_notifications_like_aggregate.down.sql @@ -0,0 +1,3 @@ +DROP INDEX IF EXISTS notifications_post_like_uq; +ALTER TABLE notifications DROP COLUMN IF EXISTS actor_count; +ALTER TABLE notifications DROP COLUMN IF EXISTS post_id; diff --git a/migrations/0031_notifications_like_aggregate.up.sql b/migrations/0031_notifications_like_aggregate.up.sql new file mode 100644 index 0000000..d10a844 --- /dev/null +++ b/migrations/0031_notifications_like_aggregate.up.sql @@ -0,0 +1,14 @@ +-- Aggregated like notifications. +-- +-- Likes are high-volume, so instead of one notification row per like we keep a +-- single row per (recipient, post) and aggregate: actor_id holds the most +-- recent liker, actor_count how many distinct likers have aggregated into it. +-- The partial unique index lets the like path UPSERT that one row (bumping the +-- count, refreshing the actor, re-marking unread) — so the activity list shows +-- "X and N others liked your post", not a flood. +ALTER TABLE notifications ADD COLUMN IF NOT EXISTS post_id UUID REFERENCES posts(id) ON DELETE CASCADE; +ALTER TABLE notifications ADD COLUMN IF NOT EXISTS actor_count INTEGER NOT NULL DEFAULT 1; + +CREATE UNIQUE INDEX IF NOT EXISTS notifications_post_like_uq + ON notifications (user_id, post_id) + WHERE kind = 'post_like';