feat(discover): Surprise me leans on what you watch, and stays a surprise

The Home hero's surprise tier and the Plex Discover rail were uniform draws over the unseen. They
are now weighted draws (`app.draw`, Efraimidis-Spirakis over the same md5 shuffle): a title's
weight is 1 + its affinity. Plex: the genres of the latest watched titles (half-life 180 days, a
show counts once) and the TMDB recommendations stored on them (`plex.affinity`). YouTube: the
channels the user watches (half-life 90 days). One slot of three stays a blind draw, and with no
history the order is exactly the old one. No visible change.

The YouTube draw builds its pool once (a materialized CTE read twice): the scope is ~200k videos on
a prod-sized catalog, and a second ordered pass doubled the cost. Measured on localdev: a reroll
210 -> 292 ms.

backend/app/discover/queries.py

@@ -17,9 +17,25 @@ otherwise "Mást kérek" re-draws the Plex picks while the YouTube ones stand st
 here takes the feed's seeded-md5 idiom instead of `random()`.
 """
 
-from sqlalchemy import and_, func, or_, select
+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
 from app.discover.budget import band
 from app.feedquery import (
     feed_columns,
@@ -33,7 +49,7 @@ from app.models import User
 from app.plex.models import PlexItem, PlexShow, PlexState
 from app.catalog.models import Channel, Video
 from app.reader.models import PlaylistItem, VideoState
-from app.plex import rollup
+from app.plex import affinity, rollup
 from app.plex.cards import display_title, leaf_card, movie_card, show_card
 from app.plex.query import apply_show_state_filter, content_libs
 from app.userscope import owned_by
@@ -115,14 +131,93 @@ def yt_surprise(db: Session, user: User, budget: str, seed: int, limit: int) ->
     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.
-    rows = db.execute(query.order_by(*order_by(keys)).limit(limit * 4)).all()
-    out = [serialize(r) for r in rows]
-    return [r for r in out if not r["position_seconds"]][:limit]
+    take = limit * 4
+
+    def unstarted(rows) -> list[dict]:
+        return [r for r in map(serialize, rows) if not r["position_seconds"]]
+
+    channels = _channel_affinity(db, user)
+    if not channels:
+        return unstarted(db.execute(query.order_by(*order_by(keys)).limit(take)).all())[: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 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)
+
+    # 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 much a channel the user watches most outweighs one they never watch (1 + this), and how far back
+# the watched history counts: the latest WATCHED_RECENT watches, the recent ones weighing more.
+CHANNEL_WEIGHT = 3.0
+WATCHED_RECENT = 300
+WATCHED_HALF_LIFE_DAYS = 90
+
+
+def _channel_affinity(db: Session, user: User) -> dict[str, float]:
+    """Each channel the user watched lately, scored 0..1 against the most watched: the sum of its
+    watches, each weighing half as much every WATCHED_HALF_LIFE_DAYS. Empty with no history."""
+    rows = db.execute(
+        select(Video.channel_id, VideoState.watched_at)
+        .join(Video, Video.id == VideoState.video_id)
+        .where(owned_by(VideoState.user_id, user), VideoState.status == "watched")
+        .order_by(VideoState.watched_at.desc().nullslast())
+        .limit(WATCHED_RECENT)
+    ).all()
+    now = datetime.now(timezone.utc)
+    scores: dict[str, float] = {}
+    for channel_id, watched_at in rows:
+        age = (
+            (now - watched_at).total_seconds() / 86400 if watched_at else 2 * WATCHED_HALF_LIFE_DAYS
+        )
+        scores[channel_id] = scores.get(channel_id, 0.0) + 0.5 ** (
+            max(0.0, age) / WATCHED_HALF_LIFE_DAYS
+        )
+    top = max(scores.values(), default=0.0)
+    return {k: v / top for k, v in scores.items()} if top else {}
 
 
 def yt_video_row(db: Session, user: User, video_id: str) -> dict | None:
@@ -289,21 +384,24 @@ def plex_surprise(
     if not movie_libs:
         return []
     lo, hi = band(budget)
-    q = (
-        select(PlexItem)
-        .where(
-            PlexItem.kind == "movie",
-            PlexItem.library_id.in_(movie_libs),
-            _unseen_movie_conds(uid),
-        )
-        .order_by(_seeded(PlexItem.rating_key, seed))
-        .limit(limit)
+    q = select(PlexItem).where(
+        PlexItem.kind == "movie",
+        PlexItem.library_id.in_(movie_libs),
+        _unseen_movie_conds(uid),
     )
     if lo is not None:
         q = q.where(PlexItem.duration_s >= lo)
     if hi is not None:
         q = q.where(PlexItem.duration_s <= hi)
-    items = db.execute(q).scalars().all()
+    shuffled = _seeded(PlexItem.rating_key, seed)
+    blind = db.execute(q.order_by(shuffled, PlexItem.id).limit(limit)).scalars().all()
+    leaning = affinity.for_user(db, uid)
+    if leaning:
+        key = draw.weighted_key(draw.seeded_u(PlexItem.rating_key, seed), leaning.weight(PlexItem))
+        weighted = db.execute(q.order_by(key, shuffled, PlexItem.id).limit(limit)).scalars().all()
+        items = draw.blend(list(weighted), list(blind), seed, limit, key=lambda it: it.id)
+    else:
+        items = list(blind)
     states = rollup.item_state_map(db, uid, [it.id for it in items])
     return [movie_card(it, states.get(it.id), lang=lang) for it in items]