test(feed): keyset paging reproduces the unpaged order for every sort (E1 S2, TEST-2)

The feed's cursor paging (sort_keys/order_by/keyset_predicate/encode/decode) is
the NULLS-LAST + tiebreaker algebra where a page-boundary bug hides as a
duplicated or skipped row on the busiest endpoint, and had zero tests. Seed 7
videos on 2 channels with duplicate sort values and NULLs across
published_at/view_count/duration/title/subscriber_count, then assert — for every
sort — that walking the feed in pages of 2 via the cursor reproduces exactly the
order a single unpaged query returns, with no duplicates.

backend/tests/test_feed_keyset.py

@@ -0,0 +1,106 @@
+"""Feed keyset paging (TEST-2), DB-lane.
+
+The feed's cursor paging (feedquery.sort_keys / order_by / keyset_predicate / encode_cursor /
+decode_cursor) is the NULLS-LAST + tiebreaker algebra where a page boundary bug hides as a duplicated or
+skipped row — on the busiest endpoint, while the scheduler ingests mid-scroll. It had zero tests.
+
+The invariant here is strong and order-agnostic: for EVERY sort, walking the feed in pages of 2 via the
+cursor must reproduce EXACTLY the order a single unpaged query returns — same rows, same order, no dup, no
+skip. The seed is built to stress it: duplicate sort values (so the id tiebreaker is load-bearing) and
+NULLs in published_at / view_count / duration_seconds / title / subscriber_count (so nulls_last is
+exercised on both a first and a second key). (The malformed-cursor -> 400 path is covered by
+test_feed_cursor.py.)
+"""
+from datetime import datetime, timezone
+
+import pytest
+
+import app.state as state
+from app.models import Channel, Subscription, User, Video
+
+# Every sort feedquery.sort_keys defines except "relevance" (which needs a search query to rank).
+_SORTS = [
+    "newest", "oldest", "views", "views_asc", "duration_desc", "duration_asc",
+    "title", "title_desc", "subscribers", "subscribers_asc", "priority", "priority_asc", "shuffle",
+]
+
+
+def _dt(day: int) -> datetime:
+    return datetime(2024, 1, day, tzinfo=timezone.utc)
+
+
+def _seed(db):
+    # Two channels; one has a NULL subscriber_count (exercises the subscribers sort's nulls_last).
+    db.add_all([
+        Channel(id="UCa", title="Channel A", subscriber_count=1000),
+        Channel(id="UCb", title="Channel B", subscriber_count=None),
+    ])
+    user = User(email="keyset@example.com", email_verified=True)
+    db.add(user)
+    db.flush()
+    db.add_all([
+        Subscription(user_id=user.id, channel_id="UCa", priority=5, hidden=False),
+        Subscription(user_id=user.id, channel_id="UCb", priority=0, hidden=False),
+    ])
+    # 7 normal/organic videos. Deliberate duplicates (published_at day 1 x2 and day 3 x2, view_count
+    # 100 x2, duration 60 x2 / 30 x2) and NULLs spread across the sort columns.
+    rows = [
+        # id       ch     published    views  dur   title
+        ("vid00", "UCa", _dt(1),      100,   60,   "alpha"),
+        ("vid01", "UCa", _dt(1),      100,   60,   "beta"),
+        ("vid02", "UCa", _dt(2),      None,  None, "gamma"),
+        ("vid03", "UCb", None,        50,    120,  None),
+        ("vid04", "UCb", _dt(3),      200,   30,   "delta"),
+        ("vid05", "UCb", _dt(3),      None,  30,   "epsilon"),
+        ("vid06", "UCa", _dt(28),     None,  None, "zeta"),
+    ]
+    db.add_all([
+        Video(
+            id=vid, channel_id=ch, published_at=pub, view_count=views, duration_seconds=dur,
+            title=title, is_short=False, live_status="none", via_search=False, via_explore=False,
+        )
+        for vid, ch, pub, views, dur, title in rows
+    ])
+    db.commit()
+    return user, [r[0] for r in rows]
+
+
+def _page(client, sort, cursor):
+    url = f"/api/feed?sort={sort}&limit=2"
+    if cursor:
+        url += f"&cursor={cursor}"
+    resp = client.get(url)
+    assert resp.status_code == 200, resp.text
+    return resp.json()
+
+
+def test_feed_surfaces_the_whole_seed(db, as_user, monkeypatch):
+    monkeypatch.setattr(state, "_configured_cache", True)  # the request gate 503s an unconfigured app
+    user, ids = _seed(db)
+    body = as_user(user).get("/api/feed?limit=50").json()
+    assert {it["id"] for it in body["items"]} == set(ids)
+    assert body["has_more"] is False and body["next_cursor"] is None
+
+
+@pytest.mark.parametrize("sort", _SORTS)
+def test_cursor_walk_matches_the_unpaged_order(db, as_user, monkeypatch, sort):
+    monkeypatch.setattr(state, "_configured_cache", True)
+    user, _ = _seed(db)
+    client = as_user(user)
+
+    # Ground truth: the full order in one query (no cursor).
+    full = [it["id"] for it in client.get(f"/api/feed?sort={sort}&limit=50").json()["items"]]
+
+    # Walk it two at a time via the cursor and reassemble.
+    walked: list[str] = []
+    cursor = None
+    for _ in range(len(full) + 2):  # generous upper bound; the loop breaks on the last page
+        body = _page(client, sort, cursor)
+        walked.extend(it["id"] for it in body["items"])
+        cursor = body["next_cursor"]
+        if not body["has_more"]:
+            assert cursor is None
+            break
+
+    assert walked == full, f"{sort}: paged order diverged from the unpaged query"
+    assert len(walked) == len(set(walked)), f"{sort}: cursor paging duplicated a row"