diff --git a/backend/app/discover/pins.py b/backend/app/discover/pins.py index 7a49ca47..98a6a0f9 100644 --- a/backend/app/discover/pins.py +++ b/backend/app/discover/pins.py @@ -8,14 +8,16 @@ second one. Every shelf card carries a resolved `next`: the thing that actually plays when you click the tile — a show's in-progress-or-next episode, a channel's newest unwatched video, a playlist's first -unwatched item. It costs a query per non-leaf pin, which is affordable precisely because this shelf is -short and hand-curated, and it buys the property the whole surface exists for: one click, no -intermediate page, no second decision. +unwatched item. That buys the property the whole surface exists for: one click, no intermediate page, +no second decision. Channels and playlists are resolved for the whole shelf at once (`_Batch`), so the +cost does not grow a query or two per pin. Domain module, so nothing here raises HTTP. `resolve_ref` returns None for "no such thing, or not yours" and the route turns that into its 404 — core never imports routes. """ +from dataclasses import dataclass, field + from sqlalchemy import func, select from sqlalchemy.orm import Session @@ -103,7 +105,34 @@ def next_position(db: Session, user: User) -> int: ) -def _shelf_card(db: Session, user: User, pin: ComfortPin) -> dict | None: +@dataclass +class _Batch: + """What the shelf resolves for every pinned channel and playlist in one go.""" + + channels: dict[str, Channel] = field(default_factory=dict) + channel_summary: dict[str, tuple[int, dict | None]] = field(default_factory=dict) + playlists: dict[int, Playlist] = field(default_factory=dict) + playlist_summary: dict[int, tuple[int, dict | None]] = field(default_factory=dict) + + @classmethod + def load(cls, db: Session, user: User, pins: list[ComfortPin]) -> "_Batch": + ch_ids = [p.channel_id for p in pins if p.channel_id is not None] + pl_ids = [p.playlist_id for p in pins if p.playlist_id is not None] + b = cls() + if ch_ids: + b.channels = { + c.id: c for c in db.scalars(select(Channel).where(Channel.id.in_(ch_ids))) + } + b.channel_summary = queries.channels_summary(db, user, ch_ids) + if pl_ids: + b.playlists = { + p.id: p for p in db.scalars(select(Playlist).where(Playlist.id.in_(pl_ids))) + } + b.playlist_summary = queries.playlists_summary(db, user, pl_ids) + return b + + +def _shelf_card(db: Session, user: User, pin: ComfortPin, batch: _Batch) -> dict | None: """One pin as a unified card with its resolved `next`. None when the pinned row has gone (the FK cascade removes the pin itself, so this is only the window between a delete and a refetch).""" card: dict | None = None @@ -117,11 +146,11 @@ def _shelf_card(db: Session, user: User, pin: ComfortPin) -> dict | None: nxt = dict(card) # a video plays itself elif pin.channel_id is not None: - ch = db.get(Channel, pin.channel_id) + ch = batch.channels.get(pin.channel_id) if ch is None: return None - card = cards.from_channel(ch, queries.channel_unwatched_count(db, user, ch.id)) - row = queries.channel_next_video(db, user, ch.id) + count, row = batch.channel_summary.get(ch.id, (0, None)) + card = cards.from_channel(ch, count) nxt = cards.from_video(row) if row else None elif pin.plex_item_id is not None: @@ -142,10 +171,10 @@ def _shelf_card(db: Session, user: User, pin: ComfortPin) -> dict | None: nxt = cards.from_plex(ep) if ep else None elif pin.playlist_id is not None: - pl = db.get(Playlist, pin.playlist_id) + pl = batch.playlists.get(pin.playlist_id) if pl is None: return None - count, row = queries.playlist_summary(db, user, pl.id) + count, row = batch.playlist_summary.get(pl.id, (0, None)) card = cards.from_playlist(pl, count, row.get("thumbnail_url") if row else None) nxt = cards.from_video(row) if row else None @@ -162,8 +191,10 @@ def _shelf_card(db: Session, user: User, pin: ComfortPin) -> dict | None: def shelf(db: Session, user: User) -> list[dict]: """The whole shelf, in the user's own order.""" out = [] - for pin in list_pins(db, user): - card = _shelf_card(db, user, pin) + pins = list_pins(db, user) + batch = _Batch.load(db, user, pins) + for pin in pins: + card = _shelf_card(db, user, pin, batch) if card is not None: out.append(card) return out diff --git a/backend/app/discover/queries.py b/backend/app/discover/queries.py index 403bd8c9..8b6e82f7 100644 --- a/backend/app/discover/queries.py +++ b/backend/app/discover/queries.py @@ -43,8 +43,9 @@ from app.plex import rollup from app.plex.cards import leaf_card, movie_card, show_card from app.plex.query import apply_show_state_filter, content_libs -# How many rows a single rail may carry. Matches `routes/plex/discover.py`'s RAIL_LIMIT on purpose — -# the two surfaces should feel like the same app, and a rail is a rail. +# How many rows a single rail may carry — Home's rails and the Plex module's Discover rails alike +# (`routes/plex/discover.py` imports it): the two surfaces should feel like the same app, and a rail +# is a rail. One definition, because a comment was all that held two copies in step. RAIL_LIMIT = 20 # The feed parameters a Home rail runs with. Spelled out rather than borrowed from the route's @@ -129,18 +130,6 @@ def yt_surprise(db: Session, user: User, budget: str, seed: int, limit: int) -> return [r for r in out if not r["position_seconds"]][:limit] -def channel_next_video(db: Session, user: User, channel_id: str) -> dict | None: - """The newest unwatched video of a pinned channel — what "put on that channel" resolves to.""" - # `library_source="all"`: a pinned channel means the whole channel, including videos that only - # ever arrived through a search. The feed's "organic" default is a feed-scoping choice, not a - # statement about what belongs to a channel. - params = dict(_FEED_BASE, channel_ids=[channel_id], scope="all", library_source="all") - query, rank_expr = filtered_query(db, user, **params) - keys = sort_keys("newest", 0, rank_expr) - row = db.execute(query.order_by(*order_by(keys)).limit(1)).first() - return serialize(row) if row is not None else None - - def yt_video_row(db: Session, user: User, video_id: str) -> dict | None: """One video as a feed row, with this user's watch state. @@ -158,31 +147,55 @@ def yt_video_row(db: Session, user: User, video_id: str) -> dict | None: return serialize(row) if row is not None else None -def channel_unwatched_count(db: Session, user: User, channel_id: str) -> int: - """How many unwatched videos a pinned channel is holding — the tile's only honest promise, since a - channel has no runtime of its own.""" - params = dict(_FEED_BASE, channel_ids=[channel_id], scope="all", library_source="all") - query, _ = filtered_query(db, user, **params) - return int(db.execute(select(func.count()).select_from(query.subquery())).scalar() or 0) +def channels_summary( + db: Session, user: User, channel_ids: list[str] +) -> dict[str, tuple[int, dict | None]]: + """For each pinned channel: (unwatched count, its newest unwatched video) — the tile's only honest + promise, since a channel has no runtime of its own, and what "put on that channel" resolves to. + + ONE query for the whole shelf, not two per channel: the count and the first row per channel are + window functions over the same filtered set. `library_source="all"`: a pinned channel means the + whole channel, including videos that only ever arrived through a search — the feed's "organic" + default is a feed-scoping choice, not a statement about what belongs to a channel.""" + if not channel_ids: + return {} + params = dict(_FEED_BASE, channel_ids=list(channel_ids), scope="all", library_source="all") + query, rank_expr = filtered_query(db, user, **params) + newest = order_by(sort_keys("newest", 0, rank_expr)) + per_channel = {"partition_by": Video.channel_id} + sub = query.add_columns( + func.row_number().over(**per_channel, order_by=newest).label("_rn"), + func.count().over(**per_channel).label("_total"), + ).subquery() + rows = db.execute(select(sub).where(sub.c._rn == 1)).all() + return {r.channel_id: (int(r._total), serialize(r)) for r in rows} -def playlist_summary(db: Session, user: User, playlist_id: int) -> tuple[int, dict | None]: - """(item count, the row to play next) for a pinned playlist — its first not-yet-watched item, or - its first item once the whole list is watched. The same row doubles as the tile's artwork, so a - shelf of playlists is scannable instead of a row of blank boxes.""" +def playlists_summary( + db: Session, user: User, playlist_ids: list[int] +) -> dict[int, tuple[int, dict | None]]: + """For each pinned playlist: (item count, the row to play next) — its first not-yet-watched item, + or its first item once the whole list is watched. The same row doubles as the tile's artwork, so + a shelf of playlists is scannable instead of a row of blank boxes. One query for the whole shelf.""" + if not playlist_ids: + return {} st = aliased(VideoState) rows = db.execute( - select(*feed_columns(user, st), PlaylistItem.position) + select(*feed_columns(user, st), PlaylistItem.playlist_id.label("_playlist")) .join(PlaylistItem, PlaylistItem.video_id == Video.id) .join(Channel, Channel.id == Video.channel_id) .outerjoin(st, and_(st.video_id == Video.id, st.user_id == user.id)) - .where(PlaylistItem.playlist_id == playlist_id) - .order_by(PlaylistItem.position.asc()) + .where(PlaylistItem.playlist_id.in_(playlist_ids)) + .order_by(PlaylistItem.playlist_id, PlaylistItem.position.asc()) ).all() - if not rows: - return 0, None - nxt = next((r for r in rows if (r.status or "new") not in ("watched", "hidden")), rows[0]) - return len(rows), serialize(nxt) + grouped: dict[int, list] = {} + for r in rows: + grouped.setdefault(r._playlist, []).append(r) + out: dict[int, tuple[int, dict | None]] = {} + for pid, items in grouped.items(): + nxt = next((r for r in items if (r.status or "new") not in ("watched", "hidden")), items[0]) + out[pid] = (len(items), serialize(nxt)) + return out # ── Plex ────────────────────────────────────────────────────────────────────────────────────────── diff --git a/backend/app/plex/rollup.py b/backend/app/plex/rollup.py index 53211250..7358338b 100644 --- a/backend/app/plex/rollup.py +++ b/backend/app/plex/rollup.py @@ -43,7 +43,9 @@ def rollup(cards: list[dict]) -> dict: ] resume = None if inprog: - resume = inprog[-1] # continue the latest-started episode + # The LAST in-progress episode in broadcast order — no timestamp is consulted. Home's + # `discover.queries.next_episode` picks the same one, so both Resume buttons agree. + resume = inprog[-1] else: resume = next((c for c in cards if c.get("status") != "watched"), None) return { diff --git a/backend/app/routes/plex/discover.py b/backend/app/routes/plex/discover.py index 3bf13054..6a6283a6 100644 --- a/backend/app/routes/plex/discover.py +++ b/backend/app/routes/plex/discover.py @@ -32,12 +32,12 @@ from app.plex.query import ( meta_filter_conds, top_art_keys, ) +from app.discover.queries import RAIL_LIMIT from app.plex.tmdb import resolve_recs from app.routes.plex._shared import _norm_lang router = APIRouter(prefix="/api/plex", tags=["plex"]) -RAIL_LIMIT = 20 GENRE_TILES = 12 SURPRISE_PER_KIND = 3 # Surprise-me: at most this many movies + this many shows (6 total) diff --git a/backend/tests/test_discover.py b/backend/tests/test_discover.py index 261001b7..a25beb5c 100644 --- a/backend/tests/test_discover.py +++ b/backend/tests/test_discover.py @@ -544,6 +544,73 @@ def test_shelf_resolves_a_playlist_to_its_first_unwatched_item(db): assert card["next"]["id"] == "vmed" +def _count_statements(db, fn) -> int: + """How many SQL statements `fn` sends through the session's connection.""" + from sqlalchemy import event + + seen: list[str] = [] + engine = db.get_bind() + listener = lambda *a, **k: seen.append(a[2]) # noqa: E731 — (conn, cursor, statement, …) + event.listen(engine, "before_cursor_execute", listener) + try: + fn() + finally: + event.remove(engine, "before_cursor_execute", listener) + return len(seen) + + +def test_the_shelf_resolves_each_channel_and_playlist_on_its_own_terms(db): + """Batched, the per-channel count and newest video come from one windowed query: each channel + must still get ITS count and ITS newest, not the shelf's overall first row.""" + user, ch = _seed_yt(db) + other = Channel(id="UCother", title="Other") + db.add(other) + db.flush() + db.add(Video(id="vother", channel_id=other.id, title="x", published_at=NOW - timedelta(days=9))) + pl = Playlist(user_id=user.id, name="Két elem") + empty = Playlist(user_id=user.id, name="Üres") + db.add_all([pl, empty]) + db.flush() + db.add(PlaylistItem(playlist_id=pl.id, video_id="vlong", position=0)) + for i, target in enumerate( + [ + {"channel_id": ch.id}, + {"channel_id": other.id}, + {"playlist_id": pl.id}, + {"playlist_id": empty.id}, + ] + ): + db.add(ComfortPin(user_id=user.id, position=i, **target)) + db.commit() + + got = [ + (c.get("unwatched_count", c.get("item_count")), (c["next"] or {}).get("id")) + for c in pins.shelf(db, user) + ] + assert got == [(3, "vshort"), (1, "vother"), (1, "vlong"), (0, None)] + + +def test_the_shelf_costs_the_same_queries_for_one_pin_or_many(db): + """A channel pin used to cost three queries and a playlist pin two, each; the shelf now + resolves every channel and every playlist together.""" + user, ch = _seed_yt(db) + extra = [Channel(id=f"UCx{i}", title=f"x{i}") for i in range(3)] + lists = [Playlist(user_id=user.id, name=f"p{i}") for i in range(4)] + db.add_all(extra + lists) + db.flush() + db.add(ComfortPin(user_id=user.id, channel_id=ch.id, position=0)) + db.add(ComfortPin(user_id=user.id, playlist_id=lists[0].id, position=1)) + db.commit() + few = _count_statements(db, lambda: pins.shelf(db, user)) + + for i, c in enumerate(extra): + db.add(ComfortPin(user_id=user.id, channel_id=c.id, position=10 + i)) + for i, p in enumerate(lists[1:]): + db.add(ComfortPin(user_id=user.id, playlist_id=p.id, position=20 + i)) + db.commit() + assert _count_statements(db, lambda: pins.shelf(db, user)) == few + + def test_a_pinned_video_survives_leaving_the_feed_scope(db): """The shelf is the fallback surface. An entry silently vanishing because the user unsubscribed from the channel is exactly the failure it must not have."""