fix(e24): the review's three backend defects — blocklist, shelf order, next episode

The resume rail built its own SELECT instead of going through `filtered_query` and copied
only three of its predicates, so it dropped the per-user channel blocklist the module
docstring claims it inherits. Blocking a channel never touches `video_states`, so a
half-watched video from it kept its resume position and headed the rail — and, at the
default budget with the resume tier leading, became the landing page's hero: the app
answering "what should I watch" with the thing the user had just banned. The clause is now
`feedquery.not_blocked_cond`, shared with `filtered_query` rather than copied a second
time; the feed's SCOPE clauses stay deliberately unapplied, for the reason `yt_video_row`
already gives.

`next_position` returned the shelf's row COUNT while nothing ever renumbers positions —
`remove_pin` deletes, and every FK cascade removes a pin when its target disappears. Past
the first hole the count under-counts, so a new pin landed in the middle of the shelf and,
past the second, on a position another pin already held. `max(position) + 1`, the idiom
`routes/playlists` already uses.

`next_episode` took the FIRST in-progress episode in broadcast order while `plex.rollup`,
which drives the show page's Resume button, takes the last. One show therefore had two
different "continue watching" answers, with the Home tile ranking it by its most recent
touch and then offering the episode abandoned weeks earlier. The new test asserts the two
implementations against each other rather than against a literal.

All three mutation-checked: each new test goes red for the defect it names.

backend/app/discover/pins.py

@@ -16,7 +16,7 @@ Domain module, so nothing here raises HTTP. `resolve_ref` returns None for "no s
 yours" and the route turns that into its 404 — core never imports routes.
 """
 
-from sqlalchemy import select
+from sqlalchemy import func, select
 from sqlalchemy.orm import Session
 
 from app.discover import cards, queries
@@ -78,8 +78,21 @@ def list_pins(db: Session, user: User) -> list[ComfortPin]:
 
 
 def next_position(db: Session, user: User) -> int:
-    """New pins land at the end of the shelf."""
-    return len(list_pins(db, user))
+    """New pins land at the end of the shelf.
+
+    `max(position) + 1`, not the row count. Positions are never renumbered on removal — `remove_pin`
+    deletes the row, and every FK cascade silently removes a pin when the thing it points at goes
+    away — so past the first hole a count under-counts and would drop the new pin into the MIDDLE of
+    the shelf, or onto a position another pin already holds (nothing in the schema forbids that).
+    Same idiom as `routes/playlists._next_playlist_position`."""
+    return (
+        db.scalar(
+            select(func.coalesce(func.max(ComfortPin.position), -1)).where(
+                ComfortPin.user_id == user.id
+            )
+        )
+        + 1
+    )
 
 
 def _shelf_card(db: Session, user: User, pin: ComfortPin) -> dict | None:

backend/app/discover/queries.py

@@ -21,7 +21,14 @@ from sqlalchemy import and_, func, or_, select
 from sqlalchemy.orm import Session, aliased
 
 from app.discover.budget import band
-from app.feedquery import feed_columns, filtered_query, order_by, serialize, sort_keys
+from app.feedquery import (
+    feed_columns,
+    filtered_query,
+    not_blocked_cond,
+    order_by,
+    serialize,
+    sort_keys,
+)
 from app.models import (
     Channel,
     PlaylistItem,
@@ -78,7 +85,14 @@ def yt_in_progress(db: Session, user: User, limit: int = RAIL_LIMIT) -> list[tup
     watched/hidden) — but the query is built here rather than through `filtered_query` because this
     rail wants an ordering by `progress_updated_at`, and that column lives on the state alias the feed
     query keeps private. Driving from `video_states` is also the faster plan: it is a per-user table of
-    a few hundred rows, against a catalog of many thousands."""
+    a few hundred rows, against a catalog of many thousands.
+
+    Building it here means the feed's OTHER predicates have to be chosen deliberately rather than
+    inherited. `not_blocked_cond` is shared, because "never show me this channel" outranks "you
+    started it" — blocking leaves the state row untouched, so without this the video the user just
+    banned heads the resume rail. The feed's SCOPE clauses (subscribed / via_explore accessibility)
+    are deliberately NOT applied, for the reason `yt_video_row` gives: something you have already
+    started must stay resumable even after you unsubscribe or an exploration expires."""
     st = aliased(VideoState)
     q = (
         select(*feed_columns(user, st), st.progress_updated_at.label("progressed_at"))
@@ -88,6 +102,7 @@ def yt_in_progress(db: Session, user: User, limit: int = RAIL_LIMIT) -> list[tup
             st.position_seconds > 0,
             st.status.notin_(("watched", "hidden")),
             Video.unavailable_since.is_(None),
+            not_blocked_cond(user),
         )
         .order_by(st.progress_updated_at.desc().nullslast())
         .limit(limit)
@@ -281,7 +296,7 @@ _DONE = ("watched", "hidden")
 
 def next_episode(db: Session, uid: int, show_id: int) -> dict | None:
     """What "put that show on" means: the episode in progress, else the first unwatched one, else the
-    first — in broadcast order, the same `(season, episode)` ordering `detail.py::_episode_nav` uses.
+    first — over the same `(season, episode)` broadcast ordering `detail.py::_episode_nav` uses.
 
     Three ordered `LIMIT 1` probes rather than one scan picked apart in Python. The Python version
     loaded EVERY episode of the show to choose one of them — 236 rows for Friends, 227 for JAG — which
@@ -289,19 +304,28 @@ def next_episode(db: Session, uid: int, show_id: int) -> dict | None:
     could not, since it resolves a show per rail row on the landing page.
 
     Returned as a `leaf_card` so it carries the show title the episode row itself does not."""
+    asc = (
+        PlexItem.season_number.asc().nullsfirst(),
+        PlexItem.episode_number.asc().nullsfirst(),
+    )
     base = (
         select(PlexItem, PlexState)
         .outerjoin(PlexState, and_(PlexState.item_id == PlexItem.id, PlexState.user_id == uid))
         .where(PlexItem.show_id == show_id, PlexItem.kind == "episode")
-        .order_by(
-            PlexItem.season_number.asc().nullsfirst(),
-            PlexItem.episode_number.asc().nullsfirst(),
-        )
         .limit(1)
     )
-    # In progress. `position_seconds > 0` already requires the state row to exist, so no null guard.
+    # In progress — the LAST one in broadcast order, which is `plex.rollup.rollup`'s `inprog[-1]`,
+    # i.e. what the show page's own Resume button plays. Two episodes of one show are in progress at
+    # once more often than it sounds (`watch_sync._reconcile_inprogress_movies` documents that a
+    # scrub to the end leaves a stale `status='new', position>0` row until the daily reconcile), and
+    # taking opposite ends of that set is how one show came to have two different "continue" answers:
+    # the Home tile ranked it by its LATEST touch and then offered the EARLIEST started episode.
+    # `position_seconds > 0` already requires the state row to exist, so no null guard.
     pick = db.execute(
-        base.where(PlexState.position_seconds > 0, PlexState.status.notin_(_DONE))
+        base.where(PlexState.position_seconds > 0, PlexState.status.notin_(_DONE)).order_by(
+            PlexItem.season_number.desc().nullslast(),
+            PlexItem.episode_number.desc().nullslast(),
+        )
     ).first()
     # Unwatched — INCLUDING an episode with no state row at all, which is the common case and the one
     # a bare `status NOT IN (...)` silently drops (NULL is not false, it is unknown, so the row fails
@@ -309,11 +333,11 @@ def next_episode(db: Session, uid: int, show_id: int) -> dict | None:
     pick = (
         pick
         or db.execute(
-            base.where(or_(PlexState.id.is_(None), PlexState.status.notin_(_DONE)))
+            base.where(or_(PlexState.id.is_(None), PlexState.status.notin_(_DONE))).order_by(*asc)
         ).first()
     )
     # Fully watched show: start it again from the top.
-    pick = pick or db.execute(base).first()
+    pick = pick or db.execute(base.order_by(*asc)).first()
     if pick is None:
         return None