fix: feed repitition

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).

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))
+}

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))
+	}
+}

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)
+// The viewer's own posts, anything involving a block (either direction), and
-// are excluded. The tier is computed in ORDER BY via correlated EXISTS, so no
+// any id in `exclude` (the already-served set) are filtered out. The tier is
-// extra columns leak into scanPost. No total is computed (infinite scroll
+// computed in ORDER BY via correlated EXISTS, so no extra columns leak into
-// infers "more" from a full page); offset paging keeps it simple. At larger
+// scanPost. No total is computed (infinite scroll infers "more" from a full
-// scale this would move to a precomputed fan-out / keyset cursor.
+// page).
-func (r *PostRepo) FeedForViewer(ctx context.Context, viewerID string, limit, offset int) ([]model.Post, error) {
+//
+// 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
 }
 

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)
 			})
 		})
 

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 {
+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}
+	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 {