From 2afbbfa0f2929837171a90bd27f2ab37fef5e14f Mon Sep 17 00:00:00 2001 From: Krispire Admin Date: Sun, 14 Jun 2026 00:18:11 +0530 Subject: [PATCH] fix: feed repitition --- internal/handler/social.go | 8 +++ internal/handler/soketi_webhook.go | 107 +++++++++++++++++++++++++++++ internal/repository/feed_repo.go | 76 ++++++++++++++++++++ internal/repository/post_repo.go | 27 ++++++-- internal/server/server.go | 18 ++++- internal/service/social_service.go | 38 +++++++++- 6 files changed, 263 insertions(+), 11 deletions(-) create mode 100644 internal/handler/soketi_webhook.go create mode 100644 internal/repository/feed_repo.go diff --git a/internal/handler/social.go b/internal/handler/social.go index 36138af..0c91906 100644 --- a/internal/handler/social.go +++ b/internal/handler/social.go @@ -259,6 +259,14 @@ func (h *PostHandler) Feed(w http.ResponseWriter, r *http.Request) { response.JSON(w, http.StatusOK, posts) } +// FeedReset — POST /feed/reset. Clears the caller's home-feed seen-set so the +// next page starts fresh. Called on logout and on page unload (the realtime +// socket dropping) to evict the set promptly rather than waiting for its TTL. +func (h *PostHandler) FeedReset(w http.ResponseWriter, r *http.Request) { + h.svc.ResetFeed(r.Context(), middleware.UserID(r.Context())) + response.NoContent(w) +} + // FeedNewUsers — GET /feed/new-users. Up to 5 accounts that joined in the last // 24h, for the feed's "new on the platform" discovery strip. Returns [] when // there are none (the client hides the section). diff --git a/internal/handler/soketi_webhook.go b/internal/handler/soketi_webhook.go new file mode 100644 index 0000000..8233775 --- /dev/null +++ b/internal/handler/soketi_webhook.go @@ -0,0 +1,107 @@ +package handler + +import ( + "crypto/hmac" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "io" + "net/http" + "strings" + + "github.com/prabhupadasays/spsays-bff/internal/repository" + "go.uber.org/zap" +) + +// notifChannelPrefix is the public per-user channel the frontend subscribes to +// for realtime notifications. Because it's keyed by user, a `channel_vacated` +// webhook on it means that user's *last* connection dropped — i.e. they went +// offline (across all their tabs) — so we can evict their feed seen-set. +const notifChannelPrefix = "notifications." + +// SoketiWebhookHandler receives Soketi (Pusher-compatible) webhooks. Soketi +// signs each delivery with the app secret (X-Pusher-Signature = hex +// HMAC-SHA256 of the raw body); we verify before acting. On channel_vacated +// for a per-user notifications channel we clear that user's home-feed seen-set +// — prompt cleanup of a true offline/disconnect, complementing the page-unload +// beacon and the set's own TTL. +type SoketiWebhookHandler struct { + appKey string + appSecret string + feed *repository.FeedRepo + logger *zap.Logger +} + +func NewSoketiWebhookHandler(appKey, appSecret string, feed *repository.FeedRepo, logger *zap.Logger) *SoketiWebhookHandler { + return &SoketiWebhookHandler{appKey: appKey, appSecret: appSecret, feed: feed, logger: logger} +} + +type soketiWebhookBody struct { + Events []struct { + Name string `json:"name"` + Channel string `json:"channel"` + } `json:"events"` +} + +// Handle — POST /webhooks/soketi. Public route: the HMAC signature is the +// authentication, not a session cookie. Returns 200 on a valid signature so +// Soketi doesn't retry; 401 on a bad/absent signature. +func (h *SoketiWebhookHandler) Handle(w http.ResponseWriter, r *http.Request) { + // Without a configured secret there's nothing to verify against — refuse + // rather than trust unsigned input. + if h.appSecret == "" { + w.WriteHeader(http.StatusServiceUnavailable) + return + } + body, err := io.ReadAll(io.LimitReader(r.Body, 1<<20)) // 1 MiB cap + if err != nil { + w.WriteHeader(http.StatusBadRequest) + return + } + if key := r.Header.Get("X-Pusher-Key"); h.appKey != "" && key != h.appKey { + h.logger.Warn("soketi webhook: app key mismatch", zap.String("got", key)) + w.WriteHeader(http.StatusUnauthorized) + return + } + if !validSoketiSignature(h.appSecret, r.Header.Get("X-Pusher-Signature"), body) { + // Almost always a secret mismatch between Soketi and the BFF. + h.logger.Warn("soketi webhook: bad signature", zap.Int("body_bytes", len(body))) + w.WriteHeader(http.StatusUnauthorized) + return + } + + var payload soketiWebhookBody + if err := json.Unmarshal(body, &payload); err != nil { + h.logger.Warn("soketi webhook: bad payload", zap.Error(err)) + w.WriteHeader(http.StatusBadRequest) + return + } + ctx := r.Context() + cleared := 0 + for _, e := range payload.Events { + // A per-user notifications channel emptying = that user is offline. + if e.Name != "channel_vacated" { + continue + } + if uid, ok := strings.CutPrefix(e.Channel, notifChannelPrefix); ok && uid != "" { + h.feed.Reset(ctx, uid) + cleared++ + } + } + h.logger.Debug("soketi webhook: verified", + zap.Int("events", len(payload.Events)), + zap.Int("feed_sets_cleared", cleared)) + w.WriteHeader(http.StatusOK) +} + +// validSoketiSignature is a constant-time check of the hex HMAC-SHA256 of the +// raw body keyed by the app secret. +func validSoketiSignature(secret, provided string, body []byte) bool { + if provided == "" { + return false + } + mac := hmac.New(sha256.New, []byte(secret)) + mac.Write(body) + expected := hex.EncodeToString(mac.Sum(nil)) + return hmac.Equal([]byte(expected), []byte(provided)) +} diff --git a/internal/repository/feed_repo.go b/internal/repository/feed_repo.go new file mode 100644 index 0000000..f2b72a2 --- /dev/null +++ b/internal/repository/feed_repo.go @@ -0,0 +1,76 @@ +package repository + +import ( + "context" + "time" + + "go.uber.org/zap" +) + +// feedSeenTTL bounds a home-feed "session": after this idle window the +// per-user seen-set expires and the feed starts fresh from the top. +const feedSeenTTL = 30 * time.Minute + +// FeedRepo tracks, per user, which post ids have already been served in the +// current home-feed session (a Redis SET) so paginated requests never repeat +// a post — and so the viewer-state-dependent feed ordering can't cause an +// OFFSET to drift over already-served rows. +// +// Nil-safe: with no Redis configured every method degrades to a no-op / +// empty, and the feed falls back to plain offset paging. +type FeedRepo struct { + redis *Redis + logger *zap.Logger +} + +func NewFeedRepo(redis *Redis, logger *zap.Logger) *FeedRepo { + return &FeedRepo{redis: redis, logger: logger} +} + +// Enabled reports whether the Redis-backed seen-set is available. +func (r *FeedRepo) Enabled() bool { return r != nil && r.redis != nil } + +func (r *FeedRepo) key(userID string) string { return "feed:seen:" + userID } + +// Seen returns the post ids already served to the user this session (empty +// when Redis is unavailable or the session is fresh). +func (r *FeedRepo) Seen(ctx context.Context, userID string) []string { + if !r.Enabled() { + return nil + } + ids, err := r.redis.Client.SMembers(ctx, r.key(userID)).Result() + if err != nil { + // Degrade rather than fail the feed — worst case is a possible repeat. + r.logger.Warn("feed seen: read failed", zap.String("user", userID), zap.Error(err)) + return nil + } + return ids +} + +// MarkSeen records the ids just served and refreshes the session TTL. +func (r *FeedRepo) MarkSeen(ctx context.Context, userID string, ids []string) { + if !r.Enabled() || len(ids) == 0 { + return + } + members := make([]any, len(ids)) + for i, id := range ids { + members[i] = id + } + pipe := r.redis.Client.Pipeline() + pipe.SAdd(ctx, r.key(userID), members...) + pipe.Expire(ctx, r.key(userID), feedSeenTTL) + if _, err := pipe.Exec(ctx); err != nil { + r.logger.Warn("feed seen: write failed", zap.String("user", userID), zap.Error(err)) + } +} + +// Reset clears the user's seen-set — called when a fresh feed session begins +// (the first page request) so reopening the feed serves from the top again. +func (r *FeedRepo) Reset(ctx context.Context, userID string) { + if !r.Enabled() { + return + } + if err := r.redis.Client.Del(ctx, r.key(userID)).Err(); err != nil { + r.logger.Warn("feed seen: reset failed", zap.String("user", userID), zap.Error(err)) + } +} diff --git a/internal/repository/post_repo.go b/internal/repository/post_repo.go index 7139914..37f9649 100644 --- a/internal/repository/post_repo.go +++ b/internal/repository/post_repo.go @@ -219,14 +219,27 @@ func (r *PostRepo) ListByUser(ctx context.Context, userID string, limit, offset // tier 0 — posts by people the viewer follows that they haven't liked yet // tier 1 — everything else (followed-and-liked, followers', strangers') // -// The viewer's own posts and anything involving a block (either direction) -// are excluded. The tier is computed in ORDER BY via correlated EXISTS, so no -// extra columns leak into scanPost. No total is computed (infinite scroll -// infers "more" from a full page); offset paging keeps it simple. At larger -// scale this would move to a precomputed fan-out / keyset cursor. -func (r *PostRepo) FeedForViewer(ctx context.Context, viewerID string, limit, offset int) ([]model.Post, error) { +// The viewer's own posts, anything involving a block (either direction), and +// any id in `exclude` (the already-served set) are filtered out. The tier is +// computed in ORDER BY via correlated EXISTS, so no extra columns leak into +// scanPost. No total is computed (infinite scroll infers "more" from a full +// page). +// +// With a non-empty `exclude` the caller drives pagination purely by exclusion +// (pass offset 0): since the tier depends on viewer state — liking a post +// moves it between tiers — an OFFSET would drift and repeat/skip rows, so the +// seen-set is the correct cursor. Without Redis, `exclude` is empty and OFFSET +// paginates (the frontend de-dups the rare drift). At larger scale this moves +// to a precomputed fan-out / keyset cursor. +func (r *PostRepo) FeedForViewer(ctx context.Context, viewerID string, limit, offset int, exclude []string) ([]model.Post, error) { + // nil would encode as SQL NULL, and `id <> ALL(NULL)` is NULL (excludes + // every row); a non-nil empty array means "exclude nothing". + if exclude == nil { + exclude = []string{} + } q := `SELECT ` + postSelectColumns + postSelectFrom + ` WHERE p.user_id <> $1 + AND p.id <> ALL($4) AND NOT EXISTS ( SELECT 1 FROM user_blocks b WHERE (b.blocker_id = $1 AND b.blocked_id = p.user_id) @@ -247,7 +260,7 @@ func (r *PostRepo) FeedForViewer(ctx context.Context, viewerID string, limit, of END) ASC, p.created_at DESC, p.id DESC LIMIT $2 OFFSET $3` - posts, _, err := r.collectPosts(ctx, 0, q, viewerID, limit, offset) + posts, _, err := r.collectPosts(ctx, 0, q, viewerID, limit, offset, exclude) return posts, err } diff --git a/internal/server/server.go b/internal/server/server.go index 84007af..beb70b8 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -56,6 +56,7 @@ type Server struct { collectionH *handler.CollectionHandler profileH *handler.ProfileHandler relationH *handler.RelationshipHandler + soketiWebhookH *handler.SoketiWebhookHandler notificationH *handler.NotificationHandler sadhnaH *handler.SadhnaHandler pushH *handler.PushHandler @@ -168,7 +169,13 @@ func (s *Server) initAuth(cfg *config.Config, logger *zap.Logger) { likeRepo := repository.NewLikeRepo(s.pg, s.redis, logger) mentionRepo := repository.NewMentionRepo(s.pg) commentRepo := repository.NewCommentRepo(s.pg, logger) - postSvc := service.NewPostService(postRepo, r2, likeRepo, mentionRepo, commentRepo) + // Per-user "already served" set so the home feed never repeats posts + // across pages (nil-safe — falls back to offset paging without Redis). + feedRepo := repository.NewFeedRepo(s.redis, logger) + postSvc := service.NewPostService(postRepo, r2, likeRepo, mentionRepo, commentRepo, feedRepo) + // Soketi signs webhooks with the app secret; on channel_vacated for a + // per-user notifications channel we evict that user's feed seen-set. + s.soketiWebhookH = handler.NewSoketiWebhookHandler(cfg.SoketiAppKey, cfg.SoketiAppSecret, feedRepo, logger) bookmarkSvc := service.NewBookmarkService(userRepo, bookmarkRepo) collectionSvc := service.NewCollectionService(userRepo, collectionRepo, postRepo, r2) @@ -765,6 +772,13 @@ func (s *Server) setupRouter() { // present, so the feed can carry `viewer_has_reshared` on each // post. Without a cookie the routes still respond — they just // return the anonymous shape (all-false). + // Soketi webhooks — server-to-server, authenticated by the HMAC + // signature (not a session). channel_vacated on a per-user + // notifications channel evicts that user's feed seen-set. + if s.soketiWebhookH != nil { + r.Post("/webhooks/soketi", s.soketiWebhookH.Handle) + } + // Home feed — the global, personalised feed of posts + its "new on // the platform" discovery strip. Auth-required (the feed is // per-viewer: ordering + relationship flags depend on the caller). @@ -777,6 +791,8 @@ func (s *Server) setupRouter() { r.Use(middleware.RequireAuth(s.authIssuer, s.userRepo)) r.Get("/", s.postH.Feed) r.Get("/new-users", s.postH.FeedNewUsers) + // Clear the per-user seen-set (logout / page unload). + r.Post("/reset", s.postH.FeedReset) }) }) diff --git a/internal/service/social_service.go b/internal/service/social_service.go index 0694174..5eca4d4 100644 --- a/internal/service/social_service.go +++ b/internal/service/social_service.go @@ -36,11 +36,12 @@ type PostService struct { likes *repository.LikeRepo mentions *repository.MentionRepo comments *repository.CommentRepo + feedSeen *repository.FeedRepo // per-user "already served" set for the home feed notif *NotificationService // optional; set via SetNotifier } -func NewPostService(posts *repository.PostRepo, r2 *repository.R2, likes *repository.LikeRepo, mentions *repository.MentionRepo, comments *repository.CommentRepo) *PostService { - return &PostService{posts: posts, r2: r2, likes: likes, mentions: mentions, comments: comments} +func NewPostService(posts *repository.PostRepo, r2 *repository.R2, likes *repository.LikeRepo, mentions *repository.MentionRepo, comments *repository.CommentRepo, feedSeen *repository.FeedRepo) *PostService { + return &PostService{posts: posts, r2: r2, likes: likes, mentions: mentions, comments: comments, feedSeen: feedSeen} } // SetNotifier wires the notification seam after construction (the @@ -224,7 +225,23 @@ func (s *PostService) Feed(ctx context.Context, viewerID string, limit, offset i if offset < 0 { offset = 0 } - posts, err := s.posts.FeedForViewer(ctx, viewerID, limit, offset) + + // Non-repetition: a per-user Redis "seen" set is the cursor. offset==0 is + // the start of a fresh session, so we clear it; every page then excludes + // already-served ids. Because the feed's tier ordering depends on viewer + // state (a like re-tiers a post), exclusion — not OFFSET — is what keeps + // pages from repeating or skipping rows. With Redis enabled we drive + // selection by the set alone (offset 0); without it we fall back to plain + // offset paging and the client de-dups the occasional drift. + if offset == 0 { + s.feedSeen.Reset(ctx, viewerID) + } + exclude := s.feedSeen.Seen(ctx, viewerID) + queryOffset := offset + if s.feedSeen.Enabled() { + queryOffset = 0 + } + posts, err := s.posts.FeedForViewer(ctx, viewerID, limit, queryOffset, exclude) if err != nil { return nil, err } @@ -243,9 +260,24 @@ func (s *PostService) Feed(ctx context.Context, viewerID string, limit, offset i if err := s.posts.HydrateViewerRelationships(ctx, viewerID, posts); err != nil { return nil, err } + // Record what we served so the next page skips it (no-op without Redis). + if len(posts) > 0 { + ids := make([]string, len(posts)) + for i := range posts { + ids[i] = posts[i].ID + } + s.feedSeen.MarkSeen(ctx, viewerID, ids) + } return posts, nil } +// ResetFeed clears the viewer's home-feed seen-set — called on logout and when +// the session ends (the page unloads / the realtime socket drops), so a stale +// set doesn't linger in Redis past its owner. No-op without Redis. +func (s *PostService) ResetFeed(ctx context.Context, viewerID string) { + s.feedSeen.Reset(ctx, viewerID) +} + // hydratePinnedSlice fills PinnedComment on a value slice. func (s *PostService) hydratePinnedSlice(ctx context.Context, viewerID string, posts []model.Post) error { if len(posts) == 0 {