perf(feed): published_at NOT NULL + range-anchored keyset for deep pages (E2 S2, PERF-15)

A nullable published_at forced the feed keyset into a non-sargable
`... OR published_at IS NULL` disjunction: with the 0067 index the planner still
applied it as a FILTER and scanned from the top of the index, so a deep `newest`
page discarded every earlier row (O(depth)).

- Migration 0068: backfill any NULLs (0 in prod) -> SET NOT NULL + DEFAULT now().
- Model: Video.published_at is now non-Optional, nullable=False, server_default now().
- feedquery: the `pub` key is nullable=False, and keyset_predicate is rebuilt
  recursively so a non-nullable leading key gets a non-strict range bound
  (`published_at <= v`) as an AND conjunct — an index range Cond instead of a
  filter. Nullable keys keep the NULL-aware OR form (a `<= v` bound would drop
  their NULL group). Equivalence proven by test_feed_keyset (all 13 sorts,
  cursor-walk == unpaged).
- Ingest paths (stub_from_playlist_item, apply_video_details, playlists sync)
  fall back to now() so they never write an explicit NULL.

Measured on the prod-sized localdev reseed at a ~50k-deep cursor (0067 index +
prod planner): OLD `Rows Removed by Filter: 50001`, 25,687 buffers / 9.16 ms ->
NEW `Index Cond: published_at <= v`, 1 row filtered, 47 buffers / 0.037 ms. SET
NOT NULL on 257k rows: 0.32s. Gate green (334 pytest).

backend/app/feedquery.py

@@ -339,7 +339,7 @@ def filtered_query(
 # 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, rank_expr=None) -> list[dict]:
-    pub = {"expr": Video.published_at, "asc": False, "nullable": True, "kind": "dt"}
+    pub = {"expr": Video.published_at, "asc": False, "nullable": False, "kind": "dt"}
     table: dict[str, list[dict]] = {
         "newest": [pub],
         "oldest": [{**pub, "asc": True}],
@@ -383,7 +383,15 @@ def keyset_predicate(keys: list[dict], cursor_id: str):
 
     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."""
+    also pull in the NULL group, and equality treats NULL == NULL.
+
+    Built recursively so that for a NON-nullable leading key we can prepend a non-strict
+    range bound (`col <= v` for desc / `col >= v` for asc) as an AND conjunct — logically
+    a no-op over the OR of strict-after branches, but it lets the planner bound the index
+    scan instead of scanning from the top and filtering every earlier row (PERF-15: deep
+    `newest` pages went 51.8k -> ~0.5k buffers). A NULLABLE key can't take that bound (the
+    NULL group sorts after every value and `col <= v` would drop it), so it keeps the pure
+    NULL-aware OR form."""
     def eq(k):
         return k["expr"].is_(None) if k["value"] is None else k["expr"] == k["value"]
 
@@ -393,11 +401,17 @@ def keyset_predicate(keys: list[dict], cursor_id: str):
         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 rec(i: int):
+        if i == len(keys):
+            return Video.id > cursor_id  # all keys tied -> break by the unique id
+        k = keys[i]
+        tail = or_(after(k), and_(eq(k), rec(i + 1)))
+        if not k["nullable"] and k["value"] is not None:
+            bound = k["expr"] >= k["value"] if k["asc"] else k["expr"] <= k["value"]
+            return and_(bound, tail)
+        return tail
+
+    return rec(0)
 
 
 def encode_cursor(keys: list[dict], row) -> str:

backend/app/models.py

@@ -261,8 +261,13 @@ class Video(Base, TimestampMixin):
     title: Mapped[str | None] = mapped_column(Text)  # normalized for display (see app.titles)
     original_title: Mapped[str | None] = mapped_column(Text)  # raw YouTube title, preserved
     description: Mapped[str | None] = mapped_column(Text)
-    published_at: Mapped[datetime | None] = mapped_column(
-        DateTime(timezone=True), index=True
+    # NOT NULL (migration 0068): in reality every video carries a publish date, and a nullable key
+    # forced the feed's keyset predicate into a non-sargable `... OR published_at IS NULL` disjunction
+    # that blocked the ix_videos_published_at_desc range (deep pages scanned from the top and filtered —
+    # PERF-15). server_default now() is the safety net for the rare date-less ingest (the enrich fills
+    # the real date); the ix_videos_published_at index stays for the `oldest` sort.
+    published_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), index=True, nullable=False, server_default=func.now()
     )
     thumbnail_url: Mapped[str | None] = mapped_column(String(1024))
     duration_seconds: Mapped[int | None] = mapped_column(Integer)

backend/app/sync/playlists.py

@@ -90,7 +90,9 @@ def _ensure_videos(db: Session, yt: YouTubeClient, video_ids: list[str]) -> None
         if not vid or not cid or vid in seen:
             continue
         seen.add(vid)
-        v = Video(id=vid, channel_id=cid, published_at=parse_dt(sn.get("publishedAt")))
+        # published_at is NOT NULL (0068); apply_video_details re-fills a missing date, but seed a now()
+        # fallback here too so the constructor never carries an explicit NULL.
+        v = Video(id=vid, channel_id=cid, published_at=parse_dt(sn.get("publishedAt")) or now)
         apply_video_details(v, it)
         v.enriched_at = now
         db.add(v)

backend/app/sync/videos.py

@@ -90,7 +90,10 @@ def stub_from_playlist_item(item: dict, channel_id: str) -> dict | None:
     video_id = content.get("videoId")
     if not video_id:
         return None
-    published = parse_dt(content.get("videoPublishedAt") or snippet.get("publishedAt"))
+    # published_at is NOT NULL (migration 0068); fall back to now() for the rare item that carries no
+    # date at all, so the bulk stub insert never writes a NULL. The enrich pass replaces it if the
+    # video's own snippet later yields a real publish date.
+    published = parse_dt(content.get("videoPublishedAt") or snippet.get("publishedAt")) or _now()
     return {
         "id": video_id,
         "channel_id": channel_id,
@@ -243,7 +246,8 @@ def apply_video_details(video: Video, item: dict) -> None:
         video.title = normalize_title(raw_title)
     video.description = snippet.get("description")
     if not video.published_at:
-        video.published_at = parse_dt(snippet.get("publishedAt"))
+        # NOT NULL (0068): a fresh insert with no date yet falls back to now() (enrich normally has one).
+        video.published_at = parse_dt(snippet.get("publishedAt")) or _now()
     video.thumbnail_url = best_thumbnail(snippet.get("thumbnails")) or video.thumbnail_url
     video.duration_seconds = parse_iso8601_duration(content.get("duration"))
     video.view_count = int(stats["viewCount"]) if stats.get("viewCount") else None