perf(feed): relevance ranks the newest 2000 matches, the rest follow newest first

A broad word matched most of the library and ts_rank read every match:
"a" took 4.4 s a page on prod. The window's start is read once (0.1 s) and
rides the cursor, so a video ingested mid-scroll cannot repeat a row; the
page is read as its two halves, each on an index (80 ms + 5 ms).

backend/app/feedquery.py

@@ -12,7 +12,7 @@ import binascii
 import json
 from datetime import date, datetime, timedelta, timezone
 
-from sqlalchemy import BigInteger, Select, and_, cast, false, func, or_, select
+from sqlalchemy import BigInteger, Select, and_, case, cast, false, func, or_, select
 from sqlalchemy.orm import Session, aliased
 
 from app.collation import collated
@@ -265,7 +265,8 @@ def filtered_query(
     # the queries that surfaced the video > description; see Video.search_vector). YouTube-like
     # "fuzzy" matching: word-order-independent, multi-word AND, prefix on the word being typed,
     # accent-insensitive (unaccent_simple). The channel name still matches as a substring so
-    # channel searches work. `rank_expr` (weight-aware ts_rank) drives the "relevance" sort.
+    # channel searches work. `rank_expr` (weight-aware ts_rank) drives the "relevance" sort, over the
+    # newest RANK_WINDOW matches only (see there).
     rank_expr = None
     if q:
         # Resolve the channels whose NAME matches up front (there are only a few hundred channels, so
@@ -543,6 +544,54 @@ def sort_keys(sort: str, seed: int, rank_expr=None, lang: str | None = None) ->
     return table.get(sort) or table["newest"]
 
 
+# Relevance ranks only the newest RANK_WINDOW matches (E31): ts_rank reads every match's tsvector, and
+# a broad word matches most of the library ("a" ranked 193k rows in 6.8 s on prod, against 1.1 s to just
+# find them). The rows past the window keep rank 0, so they follow the ranked ones newest first; a word
+# that broad has no meaningful relevance order anyway, and a specific one fits the window whole.
+RANK_WINDOW = 2000
+
+
+def rank_window_start(db: Session, query: Select) -> datetime | None:
+    """The published_at the ranked window starts at: the RANK_WINDOW-th newest match, None when fewer
+    match (rank them all). The newest-first walk stops after RANK_WINDOW matches (40 ms for "a")."""
+    return db.scalar(
+        query.with_only_columns(Video.published_at)
+        .order_by(Video.published_at.desc())
+        .offset(RANK_WINDOW - 1)
+        .limit(1)
+    )
+
+
+def windowed_rank(rank_expr, since: datetime | None):
+    """`rank_expr` inside the window, 0 past it. CASE evaluates ts_rank only for the rows inside."""
+    if since is None:
+        return rank_expr
+    return case((Video.published_at >= since, rank_expr), else_=0)
+
+
+def windowed_relevance_rows(
+    db: Session, query: Select, keys: list[dict], window: datetime, cursor_id: str | None, n: int
+) -> list:
+    """Up to `n` rows of the windowed relevance order, read as its two halves: the window ranked, then
+    the rest newest first (every window row sorts before every row past it). As one ORDER BY the
+    planner computes the CASE for every match and sorts them all (2.7 s for "a" on prod); split, each
+    half rides an index (80 ms + 5 ms). With a cursor, `keys` carry its values (keyset_predicate)."""
+    pub_key = keys[1]  # relevance: [rank, published_at]
+    rows: list = []
+    if cursor_id is None or pub_key["value"] >= window:
+        inside = query.where(Video.published_at >= window).order_by(*order_by(keys))
+        if cursor_id is not None:
+            inside = inside.where(keyset_predicate(keys, cursor_id))
+        rows = list(db.execute(inside.limit(n)).all())
+    if len(rows) < n:
+        newest = [pub_key]  # past the window every rank is 0: the order is newest first
+        rest = query.where(Video.published_at < window).order_by(*order_by(newest))
+        if cursor_id is not None:
+            rest = rest.where(keyset_predicate(newest, cursor_id))
+        rows += db.execute(rest.limit(n - len(rows))).all()
+    return rows
+
+
 def order_by(keys: list[dict]):
     cols = []
     for k in keys:
@@ -589,16 +638,22 @@ def keyset_predicate(keys: list[dict], cursor_id: str):
     return rec(0)
 
 
-def encode_cursor(keys: list[dict], row) -> str:
+def encode_cursor(keys: list[dict], row, window: datetime | None = None) -> str:
+    """`window` is the relevance window's start: the cursor carries it, so every page ranks against the
+    window page one did. Recomputed per page, a video ingested mid-scroll would push the window's oldest
+    row to rank 0, behind the cursor, and the reader would see it twice."""
     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})
+    body: dict = {"k": vals, "id": row.id}
+    if window is not None:
+        body["w"] = window.isoformat()
+    payload = json.dumps(body)
     return base64.urlsafe_b64encode(payload.encode()).decode()
 
 
-def decode_cursor(keys: list[dict], cursor: str) -> tuple[list, str]:
+def decode_cursor(keys: list[dict], cursor: str) -> tuple[list, str, datetime | None]:
     try:
         payload = json.loads(base64.urlsafe_b64decode(cursor.encode()))
         raw = payload["k"]
@@ -608,7 +663,8 @@ def decode_cursor(keys: list[dict], cursor: str) -> tuple[list, str]:
             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"]
+        window = payload.get("w")
+        return vals, payload["id"], datetime.fromisoformat(window) if window else None
     except (binascii.Error, json.JSONDecodeError, KeyError, ValueError, TypeError) as exc:
         # Framework-free: the route handler translates this into an HTTP 400 so the
         # read-model stays web-layer-agnostic (see routes/feed.get_feed).