perf(feed): a shuffle walks a stored random key instead of sorting the whole scope

Every video carries a random number drawn once (videos.shuffle_key, migration 0118, indexed with
id). A seed picks a point on that ring: the feed's Shuffle sort pages along it and wraps, and the
Home surprise reads a 500-video window of it and shuffles that window by the seed, so no video is
favoured for sitting after a wide gap. The md5(id || seed) order read and sorted the whole feed
scope on every call: ~350 ms on prod per surprise, 381-767 ms per Shuffle page on localdev, now
14 ms and 4 ms.

backend/app/discover/queries.py

@@ -10,29 +10,22 @@ explore-visible and in-progress for free, and cannot drift from them. The Plex h
 narrows EVERY rail by the Plex sidebar's `LibraryFilters`, which is right for a surface that HAS a
 sidebar and wrong for a landing page that does not.
 
-**One seed drives both halves.** The feed orders a shuffle by `md5(id || seed)` so a keyset cursor
-stays valid across pages; the Plex surprise rail uses `ORDER BY random()`, which is fine for a single
-un-paged draw but cannot be reproduced. A cross-source reroll needs both halves to move together —
-otherwise "Mást kérek" re-draws the Plex picks while the YouTube ones stand still — so the Plex side
-here takes the feed's seeded-md5 idiom instead of `random()`.
+**One seed drives both halves.** The YouTube half starts on the ring of stored random keys at the
+seed's point and shuffles that window by `md5(id || seed)`; the Plex surprise rail uses
+`ORDER BY random()`, which is fine for a single un-paged draw but cannot be reproduced. A
+cross-source reroll needs both halves to move together — otherwise "Mást kérek" re-draws the Plex
+picks while the YouTube ones stand still — so the Plex side here orders by the same seeded md5
+instead of `random()`.
 """
 
-import json
 from datetime import datetime, timezone
 
 from sqlalchemy import (
-    Float,
-    Select,
     and_,
-    cast,
     func,
-    literal,
-    literal_column,
     or_,
     select,
-    union_all,
 )
-from sqlalchemy.dialects.postgresql import JSONB
 from sqlalchemy.orm import Session, aliased
 
 from app import draw
@@ -43,6 +36,7 @@ from app.feedquery import (
     not_blocked_cond,
     order_by,
     serialize,
+    shuffle_walk,
     sort_keys,
 )
 from app.models import User
@@ -128,66 +122,50 @@ def yt_surprise(db: Session, user: User, budget: str, seed: int, limit: int) ->
     Shorts are opened up only for the shortest band: "I have ten minutes" is the one answer a Short is
     actually a good response to, and the feed's own default keeps them out otherwise."""
     lo, hi = band(budget)
-    params = dict(_FEED_BASE, min_duration=lo, max_duration=hi, include_shorts=budget == "short")
-    query, rank_expr = filtered_query(db, user, **params)
-    keys = sort_keys("shuffle", seed, rank_expr)
-
-    # Over-fetch, then drop anything already started. The feed's `unwatched` scope keeps watched and
-    # hidden out but deliberately keeps in-progress rows in (they belong in the feed). Here they must
-    # go: the resume tier has already had its chance at them, and it may have passed precisely because
-    # the REMAINING time did not fit the budget — offering the same item back as a surprise, whose
-    # runtime does fit, would label a half-finished thing "Meglepetés" and contradict the question.
-    take = limit * 4
-
-    def unstarted(rows) -> list[dict]:
-        return [r for r in map(serialize, rows) if not r["position_seconds"]]
+    params: dict = {
+        **_FEED_BASE,
+        "min_duration": lo,
+        "max_duration": hi,
+        "include_shorts": budget == "short",
+    }
+    query, _rank = filtered_query(db, user, **params)
 
+    # A window of the ring, not the whole scope: the seed's SURPRISE_WINDOW videos read along the
+    # stored random key (`shuffle_walk`, a millisecond), then shuffled by the seed. Dealing straight
+    # from the ring would favour for good every video that sits after a wide gap of it (the first
+    # video past a random point is the one whose gap the point fell into); shuffling the window makes
+    # each video's chance the window's share, even to within a few percent. A scope no larger than
+    # the window is the whole scope, in exactly the md5 order the draw always had.
+    #
+    # Anything already started is dropped. The feed's `unwatched` scope keeps watched and hidden out
+    # but deliberately keeps in-progress rows in (they belong in the feed). Here they must go: the
+    # resume tier has already had its chance at them, and it may have passed precisely because the
+    # REMAINING time did not fit the budget — offering the same item back as a surprise, whose runtime
+    # does fit, would label a half-finished thing "Meglepetés" and contradict the question.
+    window = [
+        r
+        for r in map(serialize, shuffle_walk(db, query, seed, SURPRISE_WINDOW))
+        if not r["position_seconds"]
+    ]
+    blind = draw.shuffled(window, seed, key=_video_id)
     channels = _channel_affinity(db, user)
     if not channels:
-        return unstarted(db.execute(query.order_by(*order_by(keys)).limit(take)).all())[:limit]
+        return blind[:limit]
 
-    # Both orders from ONE pass over the scope. The pool is the whole feed scope (~200k videos on a
-    # prod-sized catalog, ~390 ms a pass), so a second ordered query for the weighted draw doubled
-    # the cost; a materialized pool of (id, shuffle key, weighted key) is read twice instead.
-    weight = literal_column("1.0") + CHANNEL_WEIGHT * func.coalesce(
-        cast(cast(literal(json.dumps(channels)), JSONB).op("->>")(Video.channel_id), Float), 0.0
-    )
-    shuffle = func.md5(func.concat(Video.id, str(seed)))
-    pool = (
-        query.with_only_columns(
-            Video.id.label("vid"),
-            shuffle.label("plain"),
-            draw.weighted_key(draw.seeded_u(Video.id, seed), weight).label("leaning"),
-        )
-        .cte("surprise_pool")
-        .prefix_with("MATERIALIZED")
-    )
+    def weight(row: dict) -> float:
+        return 1.0 + CHANNEL_WEIGHT * channels.get(row["channel_id"], 0.0)
 
-    def top(side: int, *order) -> Select:
-        part = (
-            select(pool.c.vid, pool.c.plain, pool.c.leaning, literal(side).label("side"))
-            .order_by(*order, pool.c.vid)
-            .limit(take)
-            .subquery()
-        )
-        return select(part)
+    leaning = draw.leaning(window, seed, key=_video_id, weight=weight)
+    return draw.blend(leaning, blind, seed, limit, key=_video_id)
 
-    # One statement, so the pool is built once; each side's order is restored from its keys.
-    got = db.execute(union_all(top(0, pool.c.plain), top(1, pool.c.leaning, pool.c.plain))).all()
-    plain_ids = [
-        r.vid for r in sorted((r for r in got if r.side == 0), key=lambda r: (r.plain, r.vid))
-    ]
-    leaning_ids = [
-        r.vid
-        for r in sorted((r for r in got if r.side == 1), key=lambda r: (r.leaning, r.plain, r.vid))
-    ]
-    rows = {
-        r["id"]: r
-        for r in unstarted(db.execute(query.where(Video.id.in_({*plain_ids, *leaning_ids}))).all())
-    }
-    blind = [rows[i] for i in plain_ids if i in rows]
-    leaning = [rows[i] for i in leaning_ids if i in rows]
-    return draw.blend(leaning, blind, seed, limit, key=lambda r: r["id"])
+
+# How many videos of the ring a surprise draws from: wide enough that a channel the user watches has
+# a few in it, small enough to read in a millisecond or two.
+SURPRISE_WINDOW = 500
+
+
+def _video_id(row: dict) -> str:
+    return row["id"]
 
 
 # How much a channel the user watches most outweighs one they never watch (1 + this), and how far back