refactor(plex): playlists, playback, Discover and admin through the unit; the plex package is fenced

The last Plex HTTP files do no database work of their own, so ALLOWED is empty: no route builds
SQL, touches the session or a unit's table. Discover's rails move whole into plex/discover.py; a
user's playlists, watch-state writes (the race-safe finish, an early checkpoint, a status, a
season or show at once) and the watch-sync link go through plex/queries.py and plex/service.py.

The plex package joins FENCED: it reads no other unit's tables, says whose a row is through
userscope (the rollup and the watch sync's 8 inline filters moved to owned_by), and commits only in
its named jobs — the library sync, TMDB enrichment, the watch sync and its background pushes.
A playlist's has-item probe keeps the plain key lookup, so a NUL in it is still the 422 a test pins.

Pins 178 -> 0; basedpyright baseline 118 -> 106.

backend/app/routes/plex/playback.py

@@ -1,6 +1,7 @@
 """Plex playback: image/subtitle proxies, HLS streaming, and watch-state writes."""
 
 import threading
+from typing import TYPE_CHECKING
 from datetime import datetime, timedelta, timezone
 
 import httpx
@@ -16,17 +17,13 @@ from fastapi import (
 )
 from fastapi.responses import FileResponse
 from pydantic import BaseModel
-from sqlalchemy import select, update
-from sqlalchemy.dialects.postgresql import insert as pg_insert
 from sqlalchemy.orm import Session
 
 from app.auth import current_user, resolved_user_id
 from app.db import SessionLocal, get_db
 from app.models import User
-from app.plex.models import PlexItem, PlexSeason, PlexShow, PlexState
-from app.plex import imgcache, subtitles
+from app.plex import imgcache, queries, service, subtitles
 from app.plex import paths as plex_paths
-from app.plex.lookups import by_rating_key
 from app.plex import people as plex_people
 from app.plex import stream as plex_stream
 from app.plex import watch_sync as plex_watch
@@ -35,6 +32,9 @@ from app.routes.plex._shared import item_or_404
 from app.refusal import refused
 from app.routes._common import ProgressIn, not_found
 
+if TYPE_CHECKING:
+    from app.plex.models import PlexItem
+
 router = APIRouter(prefix="/api/plex", tags=["plex"])
 
 
@@ -275,14 +275,7 @@ def _push_watch(
         return
     # Capture the row's freshest timestamp NOW so the background push can tell if a newer local edit
     # lands before it settles synced_to_plex (which would otherwise strand that newer value).
-    st = (
-        db.execute(
-            select(PlexState).where(PlexState.user_id == user_id, PlexState.item_id == item_id)
-        )
-        .scalars()
-        .first()
-    )
-    marker = plex_watch._state_marker(st)
+    marker = plex_watch._state_marker(queries.state(db, user_id, item_id))
     if action == "resume" and intent is not None:
         background.add_task(
             _push_resume_guarded, user_id, item_id, rating_key, time_ms, duration_ms, marker, intent
@@ -300,37 +293,6 @@ def _push_watch(
     )
 
 
-def _mark_finished(db: Session, user_id: int, item_id: int, now: datetime) -> bool:
-    """Mark the item watched (position cleared) and say whether THIS call made it so. Two near-end
-    checkpoints arrive together routinely (the 10 s tick and the debounced `seeked` flush), and a
-    read-then-write let both see "not watched" and both scrobble. The transition is decided by the
-    write itself: an insert that did not conflict, or an update conditional on the old status —
-    under READ COMMITTED the second writer re-reads the committed row and matches nothing."""
-    values = {
-        "status": "watched",
-        "watched_at": now,
-        "position_seconds": 0,
-        "progress_updated_at": now,
-        "synced_to_plex": False,
-    }
-    inserted = db.execute(
-        pg_insert(PlexState)
-        .values(user_id=user_id, item_id=item_id, **values)
-        .on_conflict_do_nothing(constraint="uq_plex_user_item")
-        .returning(PlexState.id)
-    ).first()
-    if inserted is not None:
-        return True
-    mine = (PlexState.user_id == user_id) & (PlexState.item_id == item_id)
-    if db.execute(
-        update(PlexState).where(mine, PlexState.status != "watched").values(values)
-    ).rowcount:
-        return True
-    # Already watched: a re-finish still restamps the row, as it always did — just no scrobble.
-    db.execute(update(PlexState).where(mine).values(values))
-    return False
-
-
 class ItemProgressIn(ProgressIn):
     """The shared checkpoint (its bounds included), plus whether the player is leaving the item."""
 
@@ -366,7 +328,7 @@ def item_progress(
     now = datetime.now(timezone.utc)
     near_end = duration > 0 and position > duration - _FINISH_MARGIN_S
     if near_end:
-        flipped = _mark_finished(db, user.id, it.id, now)
+        flipped = service.mark_finished(db, user.id, it.id, now)
         # The finish supersedes every resume push still in flight for this item: stamping it as the
         # latest push makes `_resume_push_superseded` drop them. Popping the entry instead (to keep
         # the map small) removed the very guard, and an older resume push could then land at Plex
@@ -380,33 +342,17 @@ def item_progress(
             _push_watch(background, db, user.id, it.id, rating_key, "watched")
         return {"ok": True, "status": "watched", "position_seconds": 0}
 
-    st = (
-        db.execute(
-            select(PlexState).where(PlexState.user_id == user.id, PlexState.item_id == it.id)
-        )
-        .scalars()
-        .first()
-    )
+    st = queries.state(db, user, it.id)
 
     if position < _PROGRESS_MIN_S:
-        if st is not None and st.status == "new":
-            db.delete(st)
-        elif st is not None:
-            st.position_seconds = 0
-            st.progress_updated_at = now
+        service.clear_progress(db, st, now)
         db.commit()
         return {"ok": True, "position_seconds": 0}
 
-    if st is None:
-        st = PlexState(user_id=user.id, item_id=it.id)
-        db.add(st)
-    st.position_seconds = position
-    st.progress_updated_at = now
     push_now = _should_push_resume(user.id, it.id, now, final)
-    if push_now:
-        # Mark dirty before the (background) push so the reconcile knows to re-push if it fails, and so
-        # `push_state_to_plex` can settle synced_to_plex once Plex confirms.
-        st.synced_to_plex = False
+    # Marked dirty before the (background) push so the reconcile knows to re-push if it fails, and so
+    # `push_state_to_plex` can settle synced_to_plex once Plex confirms.
+    service.checkpoint(db, user, it.id, st, position, now, dirty=push_now)
     db.commit()
     if push_now:
         _push_watch(
@@ -442,13 +388,7 @@ def item_state(
     status = body.status
     if status not in ("new", "watched", "hidden"):
         raise refused(400, "invalid_status")
-    st = (
-        db.execute(
-            select(PlexState).where(PlexState.user_id == user.id, PlexState.item_id == it.id)
-        )
-        .scalars()
-        .first()
-    )
+    st = queries.state(db, user, it.id)
     prev_status = st.status if st is not None else "new"
     if status == "new":
         # Does Plex hold anything for this item that the un-mark has to clear? Two things qualify: a
@@ -462,19 +402,12 @@ def item_state(
             st.status == "watched" or (st.position_seconds or 0) > 0
         )
         if st is not None:
-            db.delete(st)
+            service.forget_state(db, st)
             db.commit()
         if mirrored_on_plex:
             _push_watch(background, db, user.id, it.id, rating_key, "unwatched")
         return {"ok": True, "status": "new"}
-    if st is None:
-        st = PlexState(user_id=user.id, item_id=it.id)
-        db.add(st)
-    st.status = status
-    if status == "watched":
-        st.watched_at = datetime.now(timezone.utc)
-        st.position_seconds = 0
-    st.synced_to_plex = False
+    service.set_status(db, user, it.id, st, status)
     db.commit()
     if status == "watched" and prev_status != "watched":
         _push_watch(background, db, user.id, it.id, rating_key, "watched")
@@ -485,44 +418,15 @@ def _bulk_state(
     background: BackgroundTasks,
     db: Session,
     user_id: int,
-    eps: list[PlexItem],
+    eps: "list[PlexItem]",
     watched: bool,
 ) -> int:
     """Mark every episode in `eps` watched (or unwatched) for a user, mirroring each change to a
-    linked Plex account in the background (best-effort). Returns how many rows actually changed."""
-    now = datetime.now(timezone.utc)
-    existing = {
-        s.item_id: s
-        for s in db.execute(
-            select(PlexState).where(
-                PlexState.user_id == user_id,
-                PlexState.item_id.in_([e.id for e in eps] or [0]),
-            )
-        ).scalars()
-    }
+    linked Plex account in the background (best-effort). Returns how many rows actually changed.
+    The handler commits before the response — which is when the background push runs, so the push
+    always reads the committed marks."""
     pushable = plex_watch.link_for_push(db, user_id) is not None
-    to_push: list[tuple[int, str, str]] = []
-    for e in eps:
-        st = existing.get(e.id)
-        prev = st.status if st is not None else "new"
-        if watched:
-            if st is None:
-                st = PlexState(user_id=user_id, item_id=e.id)
-                db.add(st)
-            st.status = "watched"
-            st.watched_at = now
-            st.position_seconds = 0
-            st.synced_to_plex = False
-            if prev != "watched":
-                to_push.append((e.id, e.rating_key, "watched"))
-        else:  # → new: drop any state (but keep the user's private "hidden" marks)
-            if st is not None and prev != "hidden":
-                db.delete(st)
-                if prev != "new":
-                    to_push.append((e.id, e.rating_key, "unwatched"))
-    # Flush, not commit: the handler commits, and it does so before the response — which is when
-    # the background push below runs, so the push always reads the committed marks.
-    db.flush()
+    to_push = service.bulk_state(db, user_id, eps, watched)
     if pushable and to_push:
         # One coalesced background task for the whole batch (reuses a single session + Plex client),
         # not one task per episode — a 200-ep show mark is 1 task, not 200.
@@ -543,15 +447,11 @@ def show_state(
     db: Session = Depends(get_db),
 ) -> dict:
     """Mark a whole show watched/unwatched (all its episodes) for the current user + push to Plex."""
-    sh = by_rating_key(db, PlexShow, rating_key)
+    sh = queries.show(db, rating_key)
     if sh is None:
         raise not_found("show")
     watched = body.watched
-    eps = (
-        db.execute(select(PlexItem).where(PlexItem.show_id == sh.id, PlexItem.kind == "episode"))
-        .scalars()
-        .all()
-    )
+    eps = queries.episodes(db, sh)
     changed = _bulk_state(background, db, user.id, eps, watched)
     db.commit()
     return {"ok": True, "changed": changed, "watched": watched}
@@ -566,15 +466,11 @@ def season_state(
     db: Session = Depends(get_db),
 ) -> dict:
     """Mark a whole season watched/unwatched (all its episodes) for the current user + push to Plex."""
-    se = by_rating_key(db, PlexSeason, rating_key)
+    se = queries.season(db, rating_key)
     if se is None:
         raise not_found("season")
     watched = body.watched
-    eps = (
-        db.execute(select(PlexItem).where(PlexItem.season_id == se.id, PlexItem.kind == "episode"))
-        .scalars()
-        .all()
-    )
+    eps = queries.season_episodes(db, se)
     changed = _bulk_state(background, db, user.id, eps, watched)
     db.commit()
     return {"ok": True, "changed": changed, "watched": watched}