feat(feed): keyset/cursor pagination replacing OFFSET/LIMIT

Replace deep-OFFSET paging (which scans-and-discards skipped rows on the
233k-row catalog and shifts/duplicates rows as the scheduler ingests mid-
scroll) with stable keyset pagination over an opaque cursor.

A single sort registry drives both the ORDER BY and a generic, NULL-aware
keyset WHERE so they can never drift, with Video.id as the unique final
tiebreaker. Covers every sort: newest/oldest, views, duration, title,
subscribers (joined), priority (coalesced multi-key) and the seeded
shuffle. The response now returns next_cursor instead of offset.

backend/app/routes/feed.py

@@ -1,3 +1,6 @@
+import base64
+import binascii
+import json
 from datetime import date, datetime, timedelta, timezone
 
 from fastapi import APIRouter, Depends, HTTPException, Query
@@ -278,21 +281,92 @@ def _feed_params(
     }
 
 
-SORTS = {
-    "newest": Video.published_at.desc().nulls_last(),
-    "oldest": Video.published_at.asc().nulls_last(),
-    "views": Video.view_count.desc().nulls_last(),
-    "views_asc": Video.view_count.asc().nulls_last(),
-    "duration_desc": Video.duration_seconds.desc().nulls_last(),
-    "duration_asc": Video.duration_seconds.asc().nulls_last(),
-    "title": func.lower(Video.title).asc().nulls_last(),
-    "title_desc": func.lower(Video.title).desc().nulls_last(),
-    "subscribers": Channel.subscriber_count.desc().nulls_last(),
-    "subscribers_asc": Channel.subscriber_count.asc().nulls_last(),
+# --- Keyset (cursor) pagination ---------------------------------------------
+#
+# Each sort is described as an ordered list of key columns (expression + direction
+# + whether it can be NULL + a JSON round-trip kind), always with Video.id as the
+# final, unique tiebreaker. This drives BOTH the ORDER BY and the keyset WHERE so
+# they can never drift apart — the prerequisite for correct cursor paging. Keyset
+# replaces OFFSET/LIMIT: deep scrolling no longer scans-and-discards skipped rows,
+# and the page boundary is stable even as the scheduler ingests new videos mid-scroll
+# (OFFSET would shift and duplicate/skip rows).
+def _sort_keys(sort: str, seed: int) -> list[dict]:
+    pub = {"expr": Video.published_at, "asc": False, "nullable": True, "kind": "dt"}
+    table: dict[str, list[dict]] = {
+        "newest": [pub],
+        "oldest": [{**pub, "asc": True}],
+        "views": [{"expr": Video.view_count, "asc": False, "nullable": True, "kind": "num"}],
+        "views_asc": [{"expr": Video.view_count, "asc": True, "nullable": True, "kind": "num"}],
+        "duration_desc": [{"expr": Video.duration_seconds, "asc": False, "nullable": True, "kind": "num"}],
+        "duration_asc": [{"expr": Video.duration_seconds, "asc": True, "nullable": True, "kind": "num"}],
+        "title": [{"expr": func.lower(Video.title), "asc": True, "nullable": True, "kind": "str"}],
+        "title_desc": [{"expr": func.lower(Video.title), "asc": False, "nullable": True, "kind": "str"}],
+        "subscribers": [{"expr": Channel.subscriber_count, "asc": False, "nullable": True, "kind": "num"}],
+        "subscribers_asc": [{"expr": Channel.subscriber_count, "asc": True, "nullable": True, "kind": "num"}],
         # Your per-channel priority (set in the channel manager), newest first within a tier.
         # coalesce keeps it null-safe in "all" scope where unsubscribed channels have no row.
-    "priority": func.coalesce(Subscription.priority, 0).desc(),
-}
+        "priority": [{"expr": func.coalesce(Subscription.priority, 0), "asc": False, "nullable": False, "kind": "num"}, pub],
+        "priority_asc": [{"expr": func.coalesce(Subscription.priority, 0), "asc": True, "nullable": False, "kind": "num"}, pub],
+        # Deterministic per-seed shuffle: the same seed yields the same total order, so
+        # the cursor stays valid across pages of one shuffled run.
+        "shuffle": [{"expr": func.md5(func.concat(Video.id, str(seed))), "asc": True, "nullable": False, "kind": "str"}],
+    }
+    return table.get(sort) or table["newest"]
+
+
+def _order_by(keys: list[dict]):
+    cols = []
+    for k in keys:
+        c = k["expr"].asc() if k["asc"] else k["expr"].desc()
+        cols.append(c.nulls_last() if k["nullable"] else c)
+    cols.append(Video.id.asc())  # unique final tiebreaker
+    return cols
+
+
+def _keyset_predicate(keys: list[dict], cursor_id: str):
+    """WHERE clause selecting rows strictly AFTER the cursor in the sort order.
+
+    Lexicographic over the key columns then id, NULL-aware for `nulls_last`: a NULL
+    sorts after every non-NULL value, so a cursor sitting on a non-NULL value must
+    also pull in the NULL group, and equality treats NULL == NULL."""
+    def eq(k):
+        return k["expr"].is_(None) if k["value"] is None else k["expr"] == k["value"]
+
+    def after(k):
+        if k["value"] is None:
+            return false()  # nulls are last; nothing sorts strictly after them
+        cmp = k["expr"] > k["value"] if k["asc"] else k["expr"] < k["value"]
+        return or_(cmp, k["expr"].is_(None)) if k["nullable"] else cmp
+
+    branches = []
+    for j in range(len(keys)):
+        branches.append(and_(*[eq(keys[i]) for i in range(j)], after(keys[j])))
+    branches.append(and_(*[eq(k) for k in keys], Video.id > cursor_id))
+    return or_(*branches)
+
+
+def _encode_cursor(keys: list[dict], row) -> str:
+    vals = []
+    for i, k in enumerate(keys):
+        v = getattr(row, f"_k{i}")
+        vals.append(v.isoformat() if (v is not None and k["kind"] == "dt") else v)
+    payload = json.dumps({"k": vals, "id": row.id})
+    return base64.urlsafe_b64encode(payload.encode()).decode()
+
+
+def _decode_cursor(keys: list[dict], cursor: str) -> tuple[list, str]:
+    try:
+        payload = json.loads(base64.urlsafe_b64decode(cursor.encode()))
+        raw = payload["k"]
+        if len(raw) != len(keys):
+            raise ValueError("key arity mismatch")
+        vals = [
+            datetime.fromisoformat(v) if (v is not None and k["kind"] == "dt") else v
+            for k, v in zip(keys, raw)
+        ]
+        return vals, payload["id"]
+    except (binascii.Error, json.JSONDecodeError, KeyError, ValueError, TypeError):
+        raise HTTPException(status_code=400, detail="Invalid feed cursor")
 
 
 @router.get("/feed")
@@ -301,29 +375,30 @@ def get_feed(
     sort: str = "newest",
     seed: int = 0,
     limit: int = Query(default=60, le=200),
-    offset: int = 0,
+    cursor: str | None = None,
     user: User = Depends(current_user),
     db: Session = Depends(get_db),
 ) -> dict:
     query, _status = _filtered_query(db, user, **params)
-    if sort in ("priority", "priority_asc"):
-        prio = func.coalesce(Subscription.priority, 0)
-        query = query.order_by(
-            prio.asc() if sort == "priority_asc" else prio.desc(),
-            Video.published_at.desc().nulls_last(),
-        )
-    else:
-        order = SORTS.get(sort)
-        if order is None and sort == "shuffle":
-            order = func.md5(func.concat(Video.id, str(seed)))
-        query = query.order_by(order if order is not None else SORTS["newest"])
+    keys = _sort_keys(sort, seed)
 
-    rows = db.execute(query.offset(offset).limit(limit + 1)).all()
+    # Expose each key column on the row so we can build the next cursor from the last item.
+    query = query.add_columns(*[k["expr"].label(f"_k{i}") for i, k in enumerate(keys)])
+    query = query.order_by(*_order_by(keys))
+
+    if cursor:
+        vals, cursor_id = _decode_cursor(keys, cursor)
+        for k, v in zip(keys, vals):
+            k["value"] = v
+        query = query.where(_keyset_predicate(keys, cursor_id))
+
+    rows = db.execute(query.limit(limit + 1)).all()
     has_more = len(rows) > limit
+    page = rows[:limit]
     return {
-        "items": [_serialize(r) for r in rows[:limit]],
+        "items": [_serialize(r) for r in page],
+        "next_cursor": _encode_cursor(keys, page[-1]) if (has_more and page) else None,
         "has_more": has_more,
-        "offset": offset,
         "limit": limit,
     }