E5 S4: Plex Discover rail-surface (default landing)

A dedicated Plex "Discover" landing — a stack of horizontal Carousel rails
served from ONE consolidated read (serve-from-DB, no live Plex/TMDB call):

- GET /api/plex/discover → continue_watching (in-progress movies + shows, most
  recently progressed) · recently_added · genres (top-N tiles) ·
  because_you_watched (recs of the most-recent watched, TMDB-enriched movie,
  resolved to library rows by guid via serve_block's resolver).
- GET /api/plex/surprise → a fresh random, scope-bounded sample for the ↻ rail.

Frontend: PlexDiscover.tsx (rails reuse the S1 Carousel + S3 PlexPosterCard;
colored genre tiles → tap to filter the grid; ↻ reshuffles the surprise rail
alone). Discover is now the root sub-view; the flat grid becomes the bookmarkable
"Browse all" secondary view (?v=browse) reached by a header search/filter (a
browse-intent flip) or the new Discover⟷Browse toggle.

resolve_recs is promoted from _resolve_recs (now cross-module). Tests:
test_plex_discover.py (rail shape/order, continue-watching, because-you-watched
+ null, surprise bounds, scope) and PlexDiscover.dom.test.tsx; plexSubUrl
round-trip updated for discover-root/grid-browse. Backend lane 393, vitest 332.

backend/app/plex/tmdb.py

@@ -334,7 +334,7 @@ def run_enrich(db: Session) -> dict:
     return {"enriched": n}
 
 
-def _resolve_recs(db: Session, recs: list[dict] | None) -> list[dict]:
+def resolve_recs(db: Session, recs: list[dict] | None) -> list[dict]:
     """Tag each recommendation with whether it's in OUR library (matched by guid `tmdb:<id>`) and, if so,
     its rating_key — so the client can deep-link the ones we have and merely display the rest."""
     if not recs:
@@ -374,6 +374,6 @@ def serve_block(db: Session, row) -> dict | None:
         "crew": row.crew or [],
         "studios": row.studios or [],
         "production_countries": row.production_countries or [],
-        "recommendations": _resolve_recs(db, row.recommendations),
+        "recommendations": resolve_recs(db, row.recommendations),
         "backdrops": row.backdrops or [],
     }

backend/app/routes/plex/__init__.py

@@ -15,6 +15,7 @@ from app.routes.plex import (
     browse,
     collections,
     detail,
+    discover,
     people,
     playback,
     playlists,
@@ -25,6 +26,7 @@ router.include_router(admin.router)
 router.include_router(browse.router)
 router.include_router(collections.router)
 router.include_router(detail.router)
+router.include_router(discover.router)
 router.include_router(people.router)
 router.include_router(playback.router)
 router.include_router(playlists.router)

backend/app/routes/plex/discover.py

@@ -0,0 +1,331 @@
+"""Plex Discover: the rail-surface read-model (E5 S4).
+
+A consolidated payload for the dedicated Discover landing — a stack of horizontal rails:
+Continue watching · Recently added · Browse-by-genre tiles · Because you watched X — plus a
+separate, reshuffling Surprise-me sample. Serve-from-DB only: the recommendation rail reads the
+recs the S2 TMDB engine already persisted (`PlexItem.recommendations`), never a live TMDB call, and
+every card comes from the same `movie_card`/`show_card` serializers the grid uses (so rails inherit
+S1's `art` + S3's rating/summary for free). The rails carry EN metadata (like the other non-grid
+card paths); the HU overlay coalesces only in the main grid.
+"""
+
+from fastapi import APIRouter, Depends, Query
+from sqlalchemy import and_, case, func, or_, select
+from sqlalchemy.orm import Session, aliased
+
+from app.auth import current_user
+from app.db import get_db
+from app.models import PlexItem, PlexShow, PlexState, User
+from app.plex import rollup
+from app.plex.cards import movie_card, show_card
+from app.plex.query import genres_col
+from app.plex.tmdb import resolve_recs
+from app.routes.plex._shared import _content_libs, _norm_lang
+
+router = APIRouter(prefix="/api/plex", tags=["plex"])
+
+RAIL_LIMIT = 20
+GENRE_TILES = 12
+
+
+def _scope_libs(db: Session, scope: str) -> tuple[list[int], list[int]]:
+    """The movie/show library ids visible for this scope (empty list = out of scope)."""
+    movie_ids, show_ids = _content_libs(db)
+    return (
+        movie_ids if scope in ("movie", "both") else [],
+        show_ids if scope in ("show", "both") else [],
+    )
+
+
+def _movie_states(db: Session, uid: int, items: list[PlexItem]) -> dict[int, PlexState]:
+    """One batched fetch of the user's watch-state for a set of movies (avoids a per-card query)."""
+    ids = [it.id for it in items]
+    if not ids:
+        return {}
+    rows = db.execute(
+        select(PlexState).where(PlexState.user_id == uid, PlexState.item_id.in_(ids))
+    ).scalars().all()
+    return {r.item_id: r for r in rows}
+
+
+def _movie_cards(db: Session, uid: int, items: list[PlexItem]) -> list[dict]:
+    sts = _movie_states(db, uid, items)
+    return [movie_card(it, sts.get(it.id)) for it in items]
+
+
+def _show_cards(db: Session, uid: int, shows: list[PlexShow]) -> list[dict]:
+    statuses = rollup.state_map(db, uid, [s.id for s in shows])
+    return [show_card(s, statuses.get(s.id, "new")) for s in shows]
+
+
+def _recently_added(
+    db: Session, uid: int, movie_ids: list[int], show_ids: list[int]
+) -> list[dict]:
+    """Newest movies + shows, interleaved by `added_at` desc into one rail."""
+    movies = (
+        db.execute(
+            select(PlexItem)
+            .where(PlexItem.kind == "movie", PlexItem.library_id.in_(movie_ids))
+            .order_by(PlexItem.added_at.desc().nullslast())
+            .limit(RAIL_LIMIT)
+        ).scalars().all()
+        if movie_ids
+        else []
+    )
+    shows = (
+        db.execute(
+            select(PlexShow)
+            .where(PlexShow.library_id.in_(show_ids))
+            .order_by(PlexShow.added_at.desc().nullslast())
+            .limit(RAIL_LIMIT)
+        ).scalars().all()
+        if show_ids
+        else []
+    )
+    # Merge the two type-homogeneous lists by added_at, then serialize each side in one batch so the
+    # per-card state lookups stay batched (no N+1). The `(is not None, value)` key sorts a missing
+    # timestamp last under reverse.
+    tagged = [(m.added_at, "movie", m) for m in movies] + [
+        (s.added_at, "show", s) for s in shows
+    ]
+    tagged.sort(key=lambda t: (t[0] is not None, t[0]), reverse=True)
+    top = tagged[:RAIL_LIMIT]
+    mcards = {
+        id(m): c
+        for m, c in zip(
+            [t[2] for t in top if t[1] == "movie"],
+            _movie_cards(db, uid, [t[2] for t in top if t[1] == "movie"]),
+        )
+    }
+    scards = {
+        id(s): c
+        for s, c in zip(
+            [t[2] for t in top if t[1] == "show"],
+            _show_cards(db, uid, [t[2] for t in top if t[1] == "show"]),
+        )
+    }
+    return [mcards[id(o)] if kind == "movie" else scards[id(o)] for _, kind, o in top]
+
+
+def _continue_watching(
+    db: Session, uid: int, movie_ids: list[int], show_ids: list[int]
+) -> list[dict]:
+    """Resume rail: in-progress movies + partially-watched shows, most-recently-progressed first.
+    A movie's resume time is its own `progress_updated_at`; a show's is the max across its episodes."""
+    rows: list[tuple[object, object, dict]] = []  # (resumed_at, sentinel-ordinal, card)
+
+    if movie_ids:
+        mrows = db.execute(
+            select(PlexItem, PlexState)
+            .join(
+                PlexState,
+                and_(PlexState.item_id == PlexItem.id, PlexState.user_id == uid),
+            )
+            .where(
+                PlexItem.kind == "movie",
+                PlexItem.library_id.in_(movie_ids),
+                PlexState.position_seconds > 0,
+                # A NULL status is still "started" (mirrors apply_movie_state_filter's in_progress) —
+                # `notin_` alone would drop it, since `NULL NOT IN (...)` is NULL, not TRUE.
+                or_(
+                    PlexState.status.is_(None),
+                    PlexState.status.notin_(["watched", "hidden"]),
+                ),
+            )
+            .order_by(PlexState.progress_updated_at.desc().nullslast())
+            .limit(RAIL_LIMIT)
+        ).all()
+        for it, st in mrows:
+            rows.append((st.progress_updated_at, movie_card(it, st)))
+
+    if show_ids:
+        ps = aliased(PlexState)
+        prog = (
+            select(
+                PlexItem.show_id.label("show_id"),
+                func.max(ps.progress_updated_at).label("resumed_at"),
+                func.count(PlexItem.id).label("total"),
+                func.count(case((ps.status == "watched", 1))).label("watched"),
+                func.count(
+                    case(
+                        (
+                            and_(
+                                ps.position_seconds > 0,
+                                or_(ps.status.is_(None), ps.status != "watched"),
+                            ),
+                            1,
+                        )
+                    )
+                ).label("inprog"),
+            )
+            .outerjoin(ps, and_(ps.item_id == PlexItem.id, ps.user_id == uid))
+            .where(PlexItem.kind == "episode", PlexItem.show_id.isnot(None))
+            .group_by(PlexItem.show_id)
+            .subquery()
+        )
+        # in-progress == started but not fully watched (mirrors apply_show_state_filter's `in_progress`).
+        srows = db.execute(
+            select(PlexShow, prog.c.resumed_at)
+            .join(prog, prog.c.show_id == PlexShow.id)
+            .where(
+                PlexShow.library_id.in_(show_ids),
+                or_(prog.c.watched > 0, prog.c.inprog > 0),
+                prog.c.watched < prog.c.total,
+            )
+            .order_by(prog.c.resumed_at.desc().nullslast())
+            .limit(RAIL_LIMIT)
+        ).all()
+        for sh, resumed_at in srows:
+            rows.append((resumed_at, show_card(sh, "in_progress")))
+
+    rows.sort(key=lambda r: (r[0] is not None, r[0]), reverse=True)
+    return [card for _, card in rows[:RAIL_LIMIT]]
+
+
+def _genre_tiles(
+    db: Session, movie_ids: list[int], show_ids: list[int], lang: str
+) -> list[dict]:
+    """The most-common genres across the scope, as {value, count} tiles → tap to filter the grid.
+    The genre vocabulary is language-specific (coalesce(genres_hu, genres) in HU), matching the grid."""
+    counts: dict[str, int] = {}
+    for model, ids in ((PlexItem, movie_ids), (PlexShow, show_ids)):
+        if not ids:
+            continue
+        gcol = genres_col(model, lang)
+        base = select(func.jsonb_array_elements_text(gcol).label("g")).where(
+            model.library_id.in_(ids), func.jsonb_typeof(gcol) == "array"
+        )
+        if model is PlexItem:
+            base = base.where(model.kind == "movie")
+        expanded = base.subquery()
+        for g, c in db.execute(
+            select(expanded.c.g, func.count()).group_by(expanded.c.g)
+        ).all():
+            counts[g] = counts.get(g, 0) + c
+    top = sorted(counts.items(), key=lambda kv: (-kv[1], kv[0]))[:GENRE_TILES]
+    return [{"value": g, "count": c} for g, c in top]
+
+
+def _because_you_watched(
+    db: Session, uid: int, movie_ids: list[int], show_ids: list[int]
+) -> dict | None:
+    """Recommendations anchored on the user's most-recently-watched, TMDB-enriched movie. Only the
+    recs that resolve to OUR library (by guid) become cards — the rest aren't actionable here. Returns
+    None (rail hidden) when there's no enriched watch history or none of its recs are in the library."""
+    if not movie_ids:
+        return None
+    anchor = db.execute(
+        select(PlexItem)
+        .join(
+            PlexState,
+            and_(PlexState.item_id == PlexItem.id, PlexState.user_id == uid),
+        )
+        .where(
+            PlexItem.kind == "movie",
+            PlexItem.library_id.in_(movie_ids),
+            PlexState.status == "watched",
+            PlexItem.recommendations.isnot(None),
+        )
+        .order_by(PlexState.watched_at.desc().nullslast())
+        .limit(1)
+    ).scalars().first()
+    if anchor is None:
+        return None
+    resolved = resolve_recs(db, anchor.recommendations)
+    # Preserve TMDB's recommendation order, keep only in-library hits, and drop the anchor itself.
+    rks = [
+        r["rating_key"]
+        for r in resolved
+        if r.get("in_library") and r.get("rating_key") != anchor.rating_key
+    ]
+    if not rks:
+        return None
+    movies = {
+        m.rating_key: m
+        for m in db.execute(
+            select(PlexItem).where(
+                PlexItem.kind == "movie", PlexItem.rating_key.in_(rks)
+            )
+        ).scalars().all()
+    }
+    shows = {
+        s.rating_key: s
+        for s in db.execute(
+            select(PlexShow).where(PlexShow.rating_key.in_(rks))
+        ).scalars().all()
+    }
+    ordered_movies = [movies[rk] for rk in rks if rk in movies]
+    ordered_shows = [shows[rk] for rk in rks if rk in shows]
+    mcards = {id(m): c for m, c in zip(ordered_movies, _movie_cards(db, uid, ordered_movies))}
+    scards = {id(s): c for s, c in zip(ordered_shows, _show_cards(db, uid, ordered_shows))}
+    items: list[dict] = []
+    for rk in rks:
+        if rk in movies:
+            items.append(mcards[id(movies[rk])])
+        elif rk in shows:
+            items.append(scards[id(shows[rk])])
+    if not items:
+        return None
+    return {"anchor": {"id": anchor.rating_key, "title": anchor.title}, "items": items}
+
+
+@router.get("/discover")
+def discover(
+    scope: str = "both",
+    lang: str = "en",
+    user: User = Depends(current_user),
+    db: Session = Depends(get_db),
+) -> dict:
+    """The Discover landing payload — every rail in one pass (serve-from-DB). Empty rails come back as
+    empty lists (the client hides them); `because_you_watched` is null when there's no anchor."""
+    lang = _norm_lang(lang)
+    movie_ids, show_ids = _scope_libs(db, scope)
+    return {
+        "scope": scope,
+        "continue_watching": _continue_watching(db, user.id, movie_ids, show_ids),
+        "recently_added": _recently_added(db, user.id, movie_ids, show_ids),
+        "genres": _genre_tiles(db, movie_ids, show_ids, lang),
+        "because_you_watched": _because_you_watched(db, user.id, movie_ids, show_ids),
+    }
+
+
+@router.get("/surprise")
+def surprise(
+    scope: str = "both",
+    limit: int = Query(default=RAIL_LIMIT, ge=1, le=40),
+    user: User = Depends(current_user),
+    db: Session = Depends(get_db),
+) -> dict:
+    """A fresh random sample for the reshuffling Surprise-me rail (a random draw each call). Movies and
+    shows are drawn separately then interleaved so a lopsided library doesn't crowd out one type."""
+    movie_ids, show_ids = _scope_libs(db, scope)
+    movies = (
+        db.execute(
+            select(PlexItem)
+            .where(PlexItem.kind == "movie", PlexItem.library_id.in_(movie_ids))
+            .order_by(func.random())
+            .limit(limit)
+        ).scalars().all()
+        if movie_ids
+        else []
+    )
+    shows = (
+        db.execute(
+            select(PlexShow)
+            .where(PlexShow.library_id.in_(show_ids))
+            .order_by(func.random())
+            .limit(limit)
+        ).scalars().all()
+        if show_ids
+        else []
+    )
+    mcards = _movie_cards(db, user.id, movies)
+    scards = _show_cards(db, user.id, shows)
+    # Interleave movie/show cards, then trim to `limit`.
+    items: list[dict] = []
+    for i in range(max(len(mcards), len(scards))):
+        if i < len(mcards):
+            items.append(mcards[i])
+        if i < len(scards):
+            items.append(scards[i])
+    return {"scope": scope, "items": items[:limit]}