refactor(discover): one shelf query per kind, one RAIL_LIMIT, a true rollup comment

The comfort shelf resolved each pinned channel with three queries (the
row, its unwatched count, its newest video) and each playlist with two.
channels_summary gets the count and the newest row per channel as window
functions over one filtered set, playlists_summary reads every pinned list
in one query, and the rows come in one IN() each — so the shelf's cost no
longer grows with its length. A test counts the statements.

RAIL_LIMIT had two definitions held in step by a comment; the Plex Discover
route now imports the domain's. And rollup's "continue the latest-started
episode" was false: it takes the last in-progress one in broadcast order,
which is also what Home's next_episode picks.

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."""