refactor(reader): the HTTP layer reads through queries and writes through the service

Feed, playlists, saved views, tags and the Home pins do no database work in their routes any more:
reads go through reader/queries (and feedquery for a feed page, its counts and facets), writes
through reader/service and discover/pins, which flush; the handlers commit. Reader is fenced below
its HTTP layer too: its owner filters go through userscope, and the catalog's jobs ask Reader
instead of joining its tables (the auto-tagger's system tags, the cleanup jobs' keep rule, the
maintenance notice). A playlist sync's catalog ingest moves to sync/videos.

backend/app/discover/pins.py

@@ -27,6 +27,7 @@ from app.reader.models import ComfortPin, Playlist
 from app.plex import rollup
 from app.plex.cards import SHOW_NEW, leaf_card, show_card
 from app.plex.lookups import by_rating_key
+from app.userscope import owned_by, owned_row
 
 # kind -> the ComfortPin column that holds it. The client's vocabulary and the schema's, in one place.
 PIN_COLUMNS: dict[str, str] = {
@@ -93,7 +94,7 @@ def reorder(db: Session, user: User, ids: list[int]) -> None:
     # old slots — and one global lock order means two reorders cannot deadlock.
     locked = db.execute(
         select(ComfortPin)
-        .where(ComfortPin.user_id == user.id)
+        .where(owned_by(ComfortPin.user_id, user.id))
         .order_by(ComfortPin.id)
         .with_for_update()
         .execution_options(populate_existing=True)
@@ -118,7 +119,7 @@ def list_pins(db: Session, user: User) -> list[ComfortPin]:
     return list(
         db.execute(
             select(ComfortPin)
-            .where(ComfortPin.user_id == user.id)
+            .where(owned_by(ComfortPin.user_id, user.id))
             .order_by(ComfortPin.position.asc(), ComfortPin.id.asc())
         )
         .scalars()
@@ -133,17 +134,44 @@ def next_position(db: Session, user: User) -> int:
     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`."""
+    Same idiom as a new playlist's place in the rail (`reader.service`)."""
     return (
         db.scalar(
             select(func.coalesce(func.max(ComfortPin.position), -1)).where(
-                ComfortPin.user_id == user.id
+                owned_by(ComfortPin.user_id, user.id)
             )
         )
         + 1
     )
 
 
+def own_pin(db: Session, user: User, pin_id: int) -> ComfortPin | None:
+    return owned_row(db, ComfortPin, pin_id, user)
+
+
+def pin_of(db: Session, user: User, column: str, value: object) -> ComfortPin | None:
+    """The user's pin of this thing (`column`/`value` from `resolve_ref`), if it is on the shelf."""
+    return db.execute(
+        select(ComfortPin).where(
+            owned_by(ComfortPin.user_id, user), getattr(ComfortPin, column) == value
+        )
+    ).scalar_one_or_none()
+
+
+def add(db: Session, user: User, column: str, value: object) -> ComfortPin:
+    """A new pin at the end of the shelf. A second pin of the same thing (another tab, at the same
+    moment) raises `IntegrityError` at the flush: the unique constraint is the guard."""
+    pin = ComfortPin(user_id=user.id, position=next_position(db, user), **{column: value})
+    db.add(pin)
+    db.flush()
+    return pin
+
+
+def remove(db: Session, pin: ComfortPin) -> None:
+    db.delete(pin)
+    db.flush()
+
+
 @dataclass
 class _Batch:
     """What the shelf resolves for every pinned channel and playlist in one go."""

backend/app/discover/queries.py

@@ -34,6 +34,7 @@ from app.reader.models import PlaylistItem, VideoState
 from app.plex import rollup
 from app.plex.cards import display_title, leaf_card, movie_card, show_card
 from app.plex.query import apply_show_state_filter, content_libs
+from app.userscope import owned_by
 
 # 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
@@ -201,7 +202,7 @@ def _unseen_movie_conds(uid: int):
         .select_from(PlexState)
         .where(
             PlexState.item_id == PlexItem.id,
-            PlexState.user_id == uid,
+            owned_by(PlexState.user_id, uid),
             or_(
                 PlexState.status.in_(("watched", "hidden")),
                 # Started counts as "not a surprise" too — see the note in `yt_surprise`.
@@ -229,7 +230,9 @@ def plex_in_progress(
     if movie_libs:
         mq = (
             select(PlexItem, PlexState)
-            .join(PlexState, and_(PlexState.item_id == PlexItem.id, PlexState.user_id == uid))
+            .join(
+                PlexState, and_(PlexState.item_id == PlexItem.id, owned_by(PlexState.user_id, uid))
+            )
             .where(
                 PlexItem.kind == "movie",
                 PlexItem.library_id.in_(movie_libs),
@@ -323,7 +326,9 @@ def next_episode(db: Session, uid: int, show_id: int, *, lang: str) -> dict | No
     )
     base = (
         select(PlexItem, PlexState)
-        .outerjoin(PlexState, and_(PlexState.item_id == PlexItem.id, PlexState.user_id == uid))
+        .outerjoin(
+            PlexState, and_(PlexState.item_id == PlexItem.id, owned_by(PlexState.user_id, uid))
+        )
         .where(PlexItem.show_id == show_id, PlexItem.kind == "episode")
         .limit(1)
     )

backend/app/feedquery.py

@@ -1,10 +1,11 @@
-"""Feed read-model: the pure query-building and row-serialization layer behind the feed.
+"""Feed read-model: Reader's query-building, paging and row serialization behind the feed.
 
-Extracted from `routes/feed.py` so the read-model has one home instead of living inside the
-router module — the feed route handlers and the playlist-detail query share these helpers (the
-projection, the filter builder, and keyset paging). Nothing here defines a route or reaches for
-request state; it's all `(db, user, params) -> query`, and a malformed cursor raises a plain
-`ValueError` for the route handler to translate (the tsquery sanitizer lives in `app.utils`).
+One home for the read-model rather than the router module: the feed routes, the playlist-detail
+query and Home share these helpers (the projection, the filter builder, and keyset paging), and the
+feed's routes only validate and answer — a page, its counts and its facets are built here. Nothing
+here defines a route or reaches for request state; it's all `(db, user, params) -> query`, and a
+malformed cursor raises `BadCursor` for the route handler to translate (the tsquery sanitizer lives
+in `app.utils`).
 """
 
 import base64
@@ -28,6 +29,7 @@ from app.models import (
 )
 from app.reader.models import ChannelTag, Playlist, PlaylistItem, Tag, VideoState
 from app.utils import iso, like_escape, to_tsquery_str
+from app.userscope import owned_by
 
 
 def _channel_url(channel_id: str, handle: str | None) -> str:
@@ -44,7 +46,7 @@ def _saved_expr(user: User):
         .select_from(PlaylistItem)
         .join(Playlist, Playlist.id == PlaylistItem.playlist_id)
         .where(
-            Playlist.user_id == user.id,
+            owned_by(Playlist.user_id, user.id),
             Playlist.kind == "watch_later",
             PlaylistItem.video_id == Video.id,
         )
@@ -99,7 +101,7 @@ def serialize(row) -> dict:
 
 def _tag_visible(user: User):
     """A ChannelTag row this user may see: a shared system tag (user_id NULL) or their own."""
-    return or_(ChannelTag.user_id.is_(None), ChannelTag.user_id == user.id)
+    return or_(ChannelTag.user_id.is_(None), owned_by(ChannelTag.user_id, user.id))
 
 
 # The pseudo-category of a selected tag id no Tag row answers (a saved view or the feed state outlives
@@ -118,7 +120,7 @@ def tags_by_category(db: Session, user: User, tag_ids: list[int]) -> dict[str, l
     if not tag_ids:
         return {}  # skip the pointless `WHERE id IN ()` round-trip on the no-tag-filter hot path
     out: dict[str, list[int]] = {}
-    visible = or_(Tag.user_id.is_(None), Tag.user_id == user.id)
+    visible = or_(Tag.user_id.is_(None), owned_by(Tag.user_id, user.id))
     for tid, cat in db.execute(
         select(Tag.id, Tag.category).where(Tag.id.in_(tag_ids), visible)
     ).all():
@@ -138,7 +140,7 @@ def not_blocked_cond(user: User):
     position — and the one query that copied only PART of this predicate set is exactly how that
     video came back as the landing page's suggestion after the user had said "never again"."""
     return Video.channel_id.notin_(
-        select(BlockedChannel.channel_id).where(BlockedChannel.user_id == user.id)
+        select(BlockedChannel.channel_id).where(owned_by(BlockedChannel.user_id, user.id))
     )
 
 
@@ -183,7 +185,7 @@ def filtered_query(
             Subscription,
             and_(
                 Subscription.channel_id == Video.channel_id,
-                Subscription.user_id == user.id,
+                owned_by(Subscription.user_id, user.id),
             ),
         )
     else:
@@ -195,12 +197,12 @@ def filtered_query(
             Subscription,
             and_(
                 Subscription.channel_id == Video.channel_id,
-                Subscription.user_id == user.id,
+                owned_by(Subscription.user_id, user.id),
                 Subscription.hidden.is_(False),
             ),
         ).outerjoin(
             SearchFind,
-            and_(SearchFind.video_id == Video.id, SearchFind.user_id == user.id),
+            and_(SearchFind.video_id == Video.id, owned_by(SearchFind.user_id, user.id)),
         )
 
     # The state-gated views (watched/hidden/in_progress) only ever return videos that HAVE a
@@ -223,8 +225,8 @@ def filtered_query(
     # catalog. Applied in BOTH scopes; the channel page reaches its ephemeral videos by
     # asking for scope=all + channel_ids (the channel is in this accessible set).
     accessible_explore = (
-        select(Subscription.channel_id).where(Subscription.user_id == user.id)
-    ).union(select(ExploredChannel.channel_id).where(ExploredChannel.user_id == user.id))
+        select(Subscription.channel_id).where(owned_by(Subscription.user_id, user.id))
+    ).union(select(ExploredChannel.channel_id).where(owned_by(ExploredChannel.user_id, user.id)))
     query = query.where(
         or_(
             Video.via_explore.is_(False),
@@ -342,7 +344,7 @@ def filtered_query(
     if untagged:
         query = query.where(
             Video.channel_id.notin_(
-                select(ChannelTag.channel_id).where(ChannelTag.user_id == user.id)
+                select(ChannelTag.channel_id).where(owned_by(ChannelTag.user_id, user.id))
             )
         )
 
@@ -692,3 +694,98 @@ def decode_cursor(keys: list[dict], cursor: str) -> tuple[list, str, datetime |
         # Framework-free: the route handler translates this into an HTTP 400 so the
         # read-model stays web-layer-agnostic (see routes/feed.get_feed).
         raise ValueError("Invalid feed cursor") from exc
+
+
+class BadCursor(ValueError):
+    """A `cursor` this module did not write (or one from another sort): the caller's bad input."""
+
+
+def feed_page(
+    db: Session,
+    user: User,
+    params: dict,
+    *,
+    sort: str,
+    seed: int,
+    limit: int,
+    cursor: str | None,
+    lang: str,
+) -> dict:
+    """One page of the feed, and the cursor for the next. A malformed cursor raises `BadCursor`."""
+    query, rank_expr = filtered_query(db, user, **params)
+    keys = sort_keys(sort, seed, rank_expr, lang)
+
+    vals, cursor_id, window = None, None, None
+    if cursor:
+        try:
+            vals, cursor_id, window = decode_cursor(keys, cursor)
+        except ValueError as e:
+            raise BadCursor(str(e)) from e
+    ranked = sort == "relevance" and rank_expr is not None
+    if ranked and not cursor:
+        window = rank_window_start(db, query)
+    if ranked and window is not None:
+        keys = sort_keys(sort, seed, windowed_rank(rank_expr, window), lang)
+
+    # Expose each key column on the row so we can build the next cursor from the last item.
+    query = query.add_columns(*[k["expr"].label(f"_k{i}") for i, k in enumerate(keys)])
+    if vals is not None:
+        for k, v in zip(keys, vals, strict=True):
+            k["value"] = v
+    if ranked and window is not None:
+        rows = windowed_relevance_rows(db, query, keys, window, cursor_id, limit + 1)
+    else:
+        query = query.order_by(*order_by(keys))
+        if cursor_id is not None:
+            query = query.where(keyset_predicate(keys, cursor_id))
+        rows = db.execute(query.limit(limit + 1)).all()
+    has_more = len(rows) > limit
+    page = rows[:limit]
+    return {
+        "items": [serialize(r) for r in page],
+        "next_cursor": (
+            encode_cursor(keys, page[-1], window if ranked else None)
+            if (has_more and page)
+            else None
+        ),
+        "has_more": has_more,
+        "limit": limit,
+    }
+
+
+def feed_count(db: Session, user: User, **params) -> int:
+    """How many videos the filters match — the reference oracle `feed_summary`'s count is tested
+    against."""
+    query, _rank = filtered_query(db, user, **params)
+    return db.scalar(select(func.count()).select_from(query.subquery())) or 0
+
+
+def facet_counts(db: Session, user: User, **params) -> dict[int, int]:
+    """Per-tag channel counts for the current filter context, one pass per category — the reference
+    oracle `feed_summary`'s facets are tested against.
+
+    Each category is counted with every other filter applied but its own selections ignored
+    (drill-down faceting: selecting one topic doesn't zero out the rest). A conjunctive (AND)
+    category keeps them applied, so each remaining chip narrows to channels that ALSO have every
+    selected tag, and tags that can't co-occur drop to zero."""
+    visible = _tag_visible(user)
+    counts: dict[int, int] = {}
+    for category in ("language", "topic", "other"):
+        conjunctive = params.get("tag_modes", {}).get(category) == "and"
+        base, _rank = filtered_query(
+            db,
+            user,
+            **{**params, "exclude_tag_category": None if conjunctive else category},
+        )
+        channels = base.with_only_columns(Video.channel_id).distinct().subquery()
+        rows = db.execute(
+            select(ChannelTag.tag_id, func.count(func.distinct(ChannelTag.channel_id)))
+            .select_from(channels)
+            .join(ChannelTag, ChannelTag.channel_id == channels.c.channel_id)
+            .join(Tag, and_(Tag.id == ChannelTag.tag_id, Tag.category == category))
+            .where(visible)
+            .group_by(ChannelTag.tag_id)
+        ).all()
+        for tag_id, count in rows:
+            counts[tag_id] = count
+    return counts

backend/app/reader/queries.py

@@ -0,0 +1,270 @@
+"""Reader's reads: a user's labels, saved views, watch state and playlists, for the HTTP layer and
+for the other units that ask about them.
+
+Every read of a user's row goes through `userscope`, so "someone else's row" is the same answer as
+a missing one. Reader reads Catalog's videos along its declared edge (a playlist's cover and
+length, whether a video exists); another unit asks the questions at the end of this module rather
+than joining Reader's tables itself.
+"""
+
+from sqlalchemy import ColumnElement, and_, func, or_, select
+from sqlalchemy.orm import InstrumentedAttribute, Session, aliased
+
+from app.feedquery import feed_columns
+from app.models import Channel, User, Video
+from app.reader.models import ChannelTag, Playlist, PlaylistItem, SavedView, Tag, VideoState
+from app.userscope import owned_by, owned_row
+
+# --- catalog, read along the edge ------------------------------------------------------------
+
+
+def video(db: Session, video_id: str) -> Video | None:
+    return db.get(Video, video_id)
+
+
+# --- tags ------------------------------------------------------------------------------------
+
+
+def tags_with_counts(db: Session, user: User) -> list[tuple[Tag, int]]:
+    """The tags this user sees (the shared system ones and their own), each with how many channel
+    links it has, by category then name."""
+    return [
+        (tag, count)
+        for tag, count in db.execute(
+            select(Tag, func.count(ChannelTag.id))
+            .outerjoin(ChannelTag, ChannelTag.tag_id == Tag.id)
+            .where(or_(Tag.user_id.is_(None), owned_by(Tag.user_id, user)))
+            .group_by(Tag.id)
+            .order_by(Tag.category, Tag.name)
+        ).tuples()
+    ]
+
+
+def own_tag(db: Session, user: User, tag_id: int) -> Tag | None:
+    """One of the user's own tags. A system tag (no owner) is nobody's to edit, so it reads as
+    missing too."""
+    return owned_row(db, Tag, tag_id, user)
+
+
+# --- saved views -----------------------------------------------------------------------------
+
+
+def views(db: Session, user: User, source: str) -> list[SavedView]:
+    return list(
+        db.execute(
+            select(SavedView)
+            .where(owned_by(SavedView.user_id, user), SavedView.source == source)
+            .order_by(SavedView.position, SavedView.id)
+        ).scalars()
+    )
+
+
+def all_views(db: Session, user: User) -> list[SavedView]:
+    return list(db.execute(select(SavedView).where(owned_by(SavedView.user_id, user))).scalars())
+
+
+def own_view(db: Session, user: User, view_id: int) -> SavedView | None:
+    return owned_row(db, SavedView, view_id, user)
+
+
+# --- watch state -----------------------------------------------------------------------------
+
+
+def video_state(db: Session, user: User, video_id: str) -> VideoState | None:
+    return db.execute(
+        select(VideoState).where(
+            owned_by(VideoState.user_id, user), VideoState.video_id == video_id
+        )
+    ).scalar_one_or_none()
+
+
+# --- playlists -------------------------------------------------------------------------------
+
+
+def own_playlist(
+    db: Session, user: User, playlist_id: int, *, lock: bool = False
+) -> Playlist | None:
+    return owned_row(db, Playlist, playlist_id, user, lock=lock)
+
+
+def playlists_for(db: Session, user: User) -> list[Playlist]:
+    """The user's playlists in their rail order."""
+    return list(
+        db.execute(
+            select(Playlist)
+            .where(owned_by(Playlist.user_id, user))
+            .order_by(Playlist.position, Playlist.created_at)
+        ).scalars()
+    )
+
+
+def watch_later(db: Session, user: User) -> Playlist | None:
+    return db.scalar(
+        select(Playlist).where(owned_by(Playlist.user_id, user), Playlist.kind == "watch_later")
+    )
+
+
+def item_counts(db: Session, ids: list[int]) -> dict[int, int]:
+    return dict(
+        db.execute(
+            select(PlaylistItem.playlist_id, func.count())
+            .where(PlaylistItem.playlist_id.in_(ids))
+            .group_by(PlaylistItem.playlist_id)
+        )
+        .tuples()
+        .all()
+    )
+
+
+def durations(db: Session, ids: list[int]) -> dict[int, int]:
+    """Each playlist's running time: the sum of its videos' lengths."""
+    return {
+        pid: int(total or 0)
+        for pid, total in db.execute(
+            select(PlaylistItem.playlist_id, func.coalesce(func.sum(Video.duration_seconds), 0))
+            .join(Video, Video.id == PlaylistItem.video_id)
+            .where(PlaylistItem.playlist_id.in_(ids))
+            .group_by(PlaylistItem.playlist_id)
+        ).tuples()
+    }
+
+
+def containing(db: Session, ids: list[int], video_id: str) -> set[int]:
+    """Which of these playlists already hold the video (the add-to-playlist popover's ticks)."""
+    return set(
+        db.execute(
+            select(PlaylistItem.playlist_id).where(
+                PlaylistItem.playlist_id.in_(ids), PlaylistItem.video_id == video_id
+            )
+        ).scalars()
+    )
+
+
+def covers(db: Session, ids: list[int]) -> dict[int, str | None]:
+    """Each playlist's cover: its first item's thumbnail, for every list in ONE window query rather
+    than a lookup per playlist."""
+    rn = (
+        func.row_number()
+        .over(partition_by=PlaylistItem.playlist_id, order_by=PlaylistItem.position)
+        .label("rn")
+    )
+    first_items = (
+        select(PlaylistItem.playlist_id.label("pid"), Video.thumbnail_url.label("thumb"), rn)
+        .join(Video, Video.id == PlaylistItem.video_id)
+        .where(PlaylistItem.playlist_id.in_(ids))
+        .subquery()
+    )
+    return dict(
+        db.execute(select(first_items.c.pid, first_items.c.thumb).where(first_items.c.rn == 1))
+        .tuples()
+        .all()
+    )
+
+
+def playlist_totals(db: Session, playlist_id: int) -> tuple[int, int, str | None]:
+    """One playlist's item count, running time and cover. An outer join, so an empty list still
+    answers (0, 0, None)."""
+    count, total = db.execute(
+        select(func.count(PlaylistItem.id), func.coalesce(func.sum(Video.duration_seconds), 0))
+        .select_from(PlaylistItem)
+        .outerjoin(Video, Video.id == PlaylistItem.video_id)
+        .where(PlaylistItem.playlist_id == playlist_id)
+    ).one()
+    cover = db.scalar(
+        select(Video.thumbnail_url)
+        .join(PlaylistItem, PlaylistItem.video_id == Video.id)
+        .where(PlaylistItem.playlist_id == playlist_id)
+        .order_by(PlaylistItem.position)
+        .limit(1)
+    )
+    return count or 0, total or 0, cover
+
+
+def playlist_rows(db: Session, user: User, playlist_id: int) -> list:
+    """A playlist's videos as feed rows (the same columns the feed serializes), in list order."""
+    state = aliased(VideoState)
+    return list(
+        db.execute(
+            select(*feed_columns(user, state))
+            .join(PlaylistItem, PlaylistItem.video_id == Video.id)
+            .join(Channel, Channel.id == Video.channel_id)
+            .outerjoin(state, and_(state.video_id == Video.id, owned_by(state.user_id, user)))
+            .where(PlaylistItem.playlist_id == playlist_id)
+            # `id` breaks a position tie the same way on every read (the renumbering add does too).
+            .order_by(PlaylistItem.position, PlaylistItem.id)
+        ).all()
+    )
+
+
+def items(db: Session, playlist_id: int) -> list[PlaylistItem]:
+    """A playlist's item rows, in list order."""
+    return list(
+        db.execute(
+            select(PlaylistItem)
+            .where(PlaylistItem.playlist_id == playlist_id)
+            .order_by(PlaylistItem.position, PlaylistItem.id)
+        ).scalars()
+    )
+
+
+def item(db: Session, playlist_id: int, video_id: str) -> PlaylistItem | None:
+    return db.scalar(
+        select(PlaylistItem).where(
+            PlaylistItem.playlist_id == playlist_id, PlaylistItem.video_id == video_id
+        )
+    )
+
+
+# --- for the other units ---------------------------------------------------------------------
+
+
+def kept_by_a_reader(video_id: InstrumentedAttribute[str]) -> ColumnElement[bool]:
+    """A video someone holds: watched, in progress or hidden (a `VideoState`), or in one of their
+    playlists. The catalog's cleanup jobs keep such a video; as a clause, so their DELETE stays
+    one statement."""
+    return or_(
+        select(VideoState.id).where(VideoState.video_id == video_id).exists(),
+        select(PlaylistItem.id).where(PlaylistItem.video_id == video_id).exists(),
+    )
+
+
+def playlisted_by(db: Session, video_ids: list[str]) -> list[tuple[int, str]]:
+    """(user id, video id) for every user who has one of these videos in a playlist."""
+    return list(
+        db.execute(
+            select(Playlist.user_id, PlaylistItem.video_id)
+            .join(PlaylistItem, PlaylistItem.playlist_id == Playlist.id)
+            .where(PlaylistItem.video_id.in_(video_ids))
+        ).tuples()
+    )
+
+
+def has_auto_tags(channel_id: InstrumentedAttribute[str]) -> ColumnElement[bool]:
+    """A channel the auto-tagger has already labelled, as a clause for its "only the missing" scan."""
+    return (
+        select(ChannelTag.id)
+        .where(
+            ChannelTag.channel_id == channel_id,
+            ChannelTag.source == "auto",
+            ChannelTag.user_id.is_(None),
+        )
+        .exists()
+    )
+
+
+def view_tag_ids(filters: object) -> list[int]:
+    """The tag ids a stored view's filters name; anything else in that slot names none."""
+    tags = filters.get("tags") if isinstance(filters, dict) else None
+    if not isinstance(tags, list):
+        return []
+    return [t for t in tags if type(t) is int]  # `type`, not isinstance: True is not a tag id
+
+
+def tags_in_feed_views(db: Session) -> set[int]:
+    """Every tag id any user's saved feed view filters on. Read defensively (`view_tag_ids`): a view
+    stored before its route checked the shape must not raise for everyone's job."""
+    return {
+        t
+        for (filters,) in db.execute(select(SavedView.filters).where(SavedView.source == "feed"))
+        for t in view_tag_ids(filters)
+    }

backend/app/reader/service.py

@@ -0,0 +1,466 @@
+"""Reader's writes: a user's labels and saved views, where they are in a video, their playlists'
+items and order — and the system tags the catalog's auto-tagger hands over.
+
+Every function here flushes and leaves the commit to its caller (the handler, or the job that owns
+its session). A write that must not double up keeps its guard where it is written: an upsert on the
+unique key, a row lock taken before a presence check. The comfort shelf's writes are Home's
+(`discover.pins`); a playlist's YouTube sync is `sync.playlists`.
+"""
+
+from datetime import datetime, timezone
+
+from sqlalchemy import exists, func, select, update
+from sqlalchemy.dialects.postgresql import insert as pg_insert
+from sqlalchemy.orm import Session
+
+from app.models import User
+from app.reader import queries
+from app.reader.models import ChannelTag, Playlist, PlaylistItem, SavedView, Tag, VideoState
+from app.sync.playlists import lock_playlists, recompute_dirty
+from app.userscope import owned_by
+
+TAG_CATEGORIES = ("language", "topic", "other")
+VIEW_NAME_MAX = 80
+
+# --- tags ------------------------------------------------------------------------------------
+
+
+def create_tag(db: Session, user: User, name: str, category: str, color: str | None) -> Tag:
+    """A tag of the user's own. An unknown category falls back to "other" rather than refusing. A
+    name the user already has raises `IntegrityError` at the flush."""
+    tag = Tag(
+        user_id=user.id,
+        name=name,
+        color=color,
+        category=category if category in TAG_CATEGORIES else "other",
+    )
+    db.add(tag)
+    db.flush()
+    return tag
+
+
+def update_tag(tag: Tag, *, sent: set[str], name: str | None, color: str | None) -> None:
+    """Change the fields the caller sent: `color` sent as None clears it."""
+    if "name" in sent and name is not None:
+        tag.name = name
+    if "color" in sent:
+        tag.color = color
+
+
+def delete_tag(db: Session, tag: Tag) -> None:
+    db.delete(tag)  # its channel links cascade
+    db.flush()
+
+
+def ensure_system_tag(db: Session, name: str, category: str, color: str | None) -> Tag:
+    """The shared system tag of this name, made the first time a channel earns it."""
+    tag = db.execute(
+        select(Tag).where(Tag.user_id.is_(None), Tag.name == name)
+    ).scalar_one_or_none()
+    if tag is None:
+        tag = Tag(user_id=None, name=name, category=category, color=color)
+        db.add(tag)
+        db.flush()
+    return tag
+
+
+def set_auto_tags(db: Session, channel_id: str, desired: dict[int, float | None]) -> None:
+    """Make a channel's automatic tag links exactly `desired` (tag id → confidence). Only the
+    auto-tagger's own links move; a user's links on the channel are never touched."""
+    existing = {
+        ct.tag_id: ct
+        for ct in db.execute(
+            select(ChannelTag).where(
+                ChannelTag.channel_id == channel_id,
+                ChannelTag.user_id.is_(None),
+                ChannelTag.source == "auto",
+            )
+        ).scalars()
+    }
+    for tag_id, ct in existing.items():
+        if tag_id not in desired:
+            db.delete(ct)
+    for tag_id, conf in desired.items():
+        ct = existing.get(tag_id)
+        if ct is None:
+            db.add(
+                ChannelTag(
+                    channel_id=channel_id,
+                    tag_id=tag_id,
+                    user_id=None,
+                    source="auto",
+                    confidence=conf,
+                )
+            )
+        else:
+            ct.confidence = conf
+    db.flush()
+
+
+def prune_orphan_system_tags(db: Session) -> int:
+    """Delete the system tags no channel links to any more (after a mapping change) — but not one a
+    saved feed view filters on. Views hold tag IDS, and a deleted tag comes back under a new id when
+    a channel earns it again, so the view's filter would point at nothing; the feed drops an unknown
+    id, and the view showed the whole feed under its old name. The kept ids are filtered here rather
+    than in SQL: a `NOT IN` over every view's ids could pass the driver's bind-parameter limit."""
+    in_views = queries.tags_in_feed_views(db)
+    orphans = [
+        tag
+        for tag in db.execute(
+            select(Tag).where(Tag.user_id.is_(None), ~exists().where(ChannelTag.tag_id == Tag.id))
+        ).scalars()
+        if tag.id not in in_views
+    ]
+    for tag in orphans:
+        db.delete(tag)
+    db.flush()
+    return len(orphans)
+
+
+# --- saved views -----------------------------------------------------------------------------
+
+
+def create_view(db: Session, user: User, source: str, name: str, filters: dict) -> SavedView:
+    """A view appended after the user's others of the same module (each module numbers its own
+    0..N). A name the user already has there raises `IntegrityError` at the flush."""
+    maxpos = db.execute(
+        select(func.coalesce(func.max(SavedView.position), -1)).where(
+            owned_by(SavedView.user_id, user), SavedView.source == source
+        )
+    ).scalar_one()
+    view = SavedView(
+        user_id=user.id,
+        source=source,
+        name=name[:VIEW_NAME_MAX],
+        filters=filters,
+        position=maxpos + 1,
+    )
+    db.add(view)
+    db.flush()
+    return view
+
+
+def update_view(
+    db: Session,
+    view: SavedView,
+    *,
+    sent: set[str],
+    name: str | None = None,
+    filters: dict | None = None,
+    is_default: bool | None = None,
+) -> None:
+    """Change the fields the caller sent. Making a view the default clears the flag on the user's
+    other views of the same module first: at most one default per (user, source)."""
+    if "name" in sent and name is not None:
+        view.name = name[:VIEW_NAME_MAX]
+    if "filters" in sent and filters is not None:
+        view.filters = filters
+    if "is_default" in sent:
+        make_default = bool(is_default)
+        if make_default:
+            db.execute(
+                update(SavedView)
+                .where(
+                    owned_by(SavedView.user_id, view.user_id),
+                    SavedView.source == view.source,
+                    SavedView.id != view.id,
+                )
+                .values(is_default=False)
+            )
+        view.is_default = make_default
+    db.flush()
+
+
+def delete_view(db: Session, view: SavedView) -> None:
+    db.delete(view)
+    db.flush()
+
+
+def reorder_views(db: Session, user: User, ids: list[int]) -> None:
+    """Number the listed views 0..n in the listed order; ids that are not the user's are skipped."""
+    by_id = {v.id: v for v in queries.all_views(db, user)}
+    pos = 0
+    for vid in ids:
+        v = by_id.get(vid)
+        if v is not None:
+            v.position = pos
+            pos += 1
+    db.flush()
+
+
+# --- watch state -----------------------------------------------------------------------------
+
+
+def unmark(db: Session, user: User, video_id: str) -> None:
+    """Back to "new". A row that still holds a resume point keeps it (un-marking "watched" restores
+    the in-progress state, not where the user left off); otherwise the row goes. The player fires
+    this beside a checkpoint that may delete the same row: the caller's commit then raises
+    `StaleDataError`, which is that lost race and nothing worse."""
+    row = queries.video_state(db, user, video_id)
+    if row is None:
+        return
+    if row.position_seconds:
+        row.status = "new"
+        row.watched_at = None
+    else:
+        db.delete(row)
+    db.flush()
+
+
+def mark(db: Session, user: User, video_id: str, status: str) -> None:
+    """ "watched" or "hidden", as one atomic upsert on (user, video). The player marks "watched" at
+    the end of playback at the instant a checkpoint may delete the in-progress row; a SELECT then
+    UPDATE raised `StaleDataError` on that race, and an INSERT … ON CONFLICT is immune to it.
+    "hidden" keeps any earlier `watched_at`."""
+    now = datetime.now(timezone.utc)
+    set_: dict = {"status": status, "updated_at": now}
+    if status == "watched":
+        set_["watched_at"] = now
+    db.execute(
+        pg_insert(VideoState)
+        .values(
+            user_id=user.id,
+            video_id=video_id,
+            status=status,
+            watched_at=now if status == "watched" else None,
+        )
+        .on_conflict_do_update(constraint="uq_user_video", set_=set_)
+    )
+
+
+def clear_state(db: Session, user: User, video_id: str) -> None:
+    """The video back to pristine for this user: status, resume point and `watched_at` all go.
+    Playlist membership lives elsewhere and stays."""
+    row = queries.video_state(db, user, video_id)
+    if row is not None:
+        db.delete(row)
+        db.flush()
+
+
+def set_progress(db: Session, user: User, video_id: str, position: int) -> None:
+    """The player's resume point. 0 clears it: a row holding nothing else goes, one with a status
+    (watched, hidden) keeps it with the position zeroed. The status itself is never changed here."""
+    row = queries.video_state(db, user, video_id)
+    now = datetime.now(timezone.utc)
+    if not position:
+        if row is None:
+            return
+        if row.status == "new":
+            db.delete(row)
+        else:
+            row.position_seconds = 0
+            row.progress_updated_at = now
+        db.flush()
+        return
+    if row is None:
+        row = VideoState(user_id=user.id, video_id=video_id)
+        db.add(row)
+    row.position_seconds = position
+    row.progress_updated_at = now
+    db.flush()
+
+
+# --- playlists -------------------------------------------------------------------------------
+
+
+class MoveRefused(Exception):
+    """A move between playlists that cannot happen: `reason` is the code the UI translates."""
+
+    def __init__(self, reason: str):
+        super().__init__(reason)
+        self.reason = reason
+
+
+def touch(pl: Playlist) -> None:
+    """Bump the playlist's `updated_at` for an edit that only writes `playlist_items`.
+
+    `UpdatedAtMixin`'s `onupdate` fires on an UPDATE of the PLAYLIST row, and add / remove /
+    reorder / move write the item rows instead — so without this the rail's "recently changed"
+    sort would only ever see renames. The assignment is what makes the column mean "last time
+    this list changed", which is what the rail promises the reader.
+
+    The DATABASE's clock, like the mixin's own `onupdate`: one sort key written from two clocks
+    (this process's and Postgres's) could order a rename and an item edit the wrong way round."""
+    pl.updated_at = func.now()
+
+
+def _next_playlist_position(db: Session, user_id: int) -> int:
+    """Append position for a new playlist in this user's rail."""
+    return (
+        db.execute(
+            select(func.coalesce(func.max(Playlist.position), -1)).where(
+                owned_by(Playlist.user_id, user_id)
+            )
+        ).scalar_one()
+        + 1
+    )
+
+
+def next_item_position(db: Session, playlist_id: int) -> int:
+    """Append position for a new item at the end of a playlist. Locks the playlist row first: two
+    adds (two tabs, the rail and the page) otherwise both read the same max and land on one
+    position, and the list's order between them is then arbitrary."""
+    lock_playlists(db, playlist_id)
+    return (
+        db.execute(
+            select(func.coalesce(func.max(PlaylistItem.position), -1)).where(
+                PlaylistItem.playlist_id == playlist_id
+            )
+        ).scalar_one()
+        + 1
+    )
+
+
+def create_playlist(db: Session, user: User, name: str) -> Playlist:
+    pl = Playlist(user_id=user.id, name=name, position=_next_playlist_position(db, user.id))
+    db.add(pl)
+    db.flush()
+    return pl
+
+
+def watch_later(db: Session, user: User) -> Playlist:
+    """The user's built-in Watch later playlist, created on first use."""
+    pl = queries.watch_later(db, user)
+    if pl is None:
+        pl = Playlist(
+            user_id=user.id,
+            name="Watch later",
+            kind="watch_later",
+            position=_next_playlist_position(db, user.id),
+        )
+        db.add(pl)
+        db.flush()  # the id, for the item the caller adds next
+    return pl
+
+
+def add_item(db: Session, pl: Playlist, video_id: str, *, mark_dirty: bool) -> bool:
+    """Append a video to a playlist if not already present (idempotent). Returns whether it
+    was added. `mark_dirty` recomputes the YouTube-sync dirty flag (skip for Watch later).
+    Locked BEFORE the presence check: two adds of one video both saw "absent" and the second one's
+    insert hit uq_playlist_video as a 500."""
+    lock_playlists(db, pl.id)
+    if queries.item(db, pl.id, video_id) is not None:
+        return False
+    db.add(
+        PlaylistItem(playlist_id=pl.id, video_id=video_id, position=next_item_position(db, pl.id))
+    )
+    if mark_dirty:
+        recompute_dirty(db, pl)
+    touch(pl)
+    db.flush()
+    return True
+
+
+def remove_item(db: Session, pl: Playlist, video_id: str, *, mark_dirty: bool) -> bool:
+    """Remove a video from a playlist if present (idempotent). Returns whether it was removed.
+    `mark_dirty` recomputes the YouTube-sync dirty flag (skip for Watch later)."""
+    lock_playlists(db, pl.id)
+    found = queries.item(db, pl.id, video_id)
+    if found is None:
+        return False
+    db.delete(found)
+    if mark_dirty:
+        recompute_dirty(db, pl)
+    touch(pl)
+    db.flush()
+    return True
+
+
+def place_item(db: Session, pl: Playlist, video_id: str, position: int) -> None:
+    """Put a video already in the list at `position` (clamped to the end), renumbering the whole
+    list: positions can have gaps after removals, so "index N" is only meaningful over the
+    ordered rows."""
+    rows = queries.items(db, pl.id)
+    new = next(r for r in rows if r.video_id == video_id)
+    rows.remove(new)
+    rows.insert(min(position, len(rows)), new)
+    for pos, row in enumerate(rows):
+        row.position = pos
+    db.flush()
+
+
+def add_to_playlist(db: Session, pl: Playlist, video_id: str, position: int | None) -> None:
+    """Add a video to one of the user's lists, at `position` when given (the list page's "Undo" of
+    a removal puts the row back where it was in ONE request), else at the end. A video already in
+    the list stays where it is."""
+    added = add_item(db, pl, video_id, mark_dirty=False)
+    if added and position is not None:
+        place_item(db, pl, video_id, position)
+    recompute_dirty(db, pl)
+    db.flush()
+
+
+def rename_playlist(db: Session, pl: Playlist, name: str) -> None:
+    """A new name moves `dirty`; the caller holds the playlist's lock (the scheduled mirror refresh
+    reads `dirty` under it)."""
+    if name != pl.name:
+        pl.name = name
+        recompute_dirty(db, pl)
+    db.flush()
+
+
+def delete_playlist(db: Session, pl: Playlist) -> None:
+    """The playlist first, as every item write: the cascade deletes the items after the list row
+    is locked."""
+    lock_playlists(db, pl.id)
+    db.delete(pl)  # items cascade
+    db.flush()
+
+
+def move_item(db: Session, src: Playlist, dst: Playlist, video_id: str) -> None:
+    """Move one video from `src` into `dst`, in the caller's one transaction.
+
+    The client-side shape — add to the target, then remove from the source — leaves the video in
+    BOTH lists whenever the second call fails. Idempotent: a video already in the target is simply
+    dropped from the source (moving into a list that has it is a removal, not a duplicate). Raises
+    `MoveRefused` for the same list twice, or a video a concurrent edit already took out."""
+    if src.id == dst.id:
+        raise MoveRefused("same_playlist")
+    # Both lists, in id order: a move A→B beside a move B→A would otherwise lock crosswise.
+    lock_playlists(db, src.id, dst.id)
+    row = queries.item(db, src.id, video_id)
+    if row is None:
+        raise MoveRefused("not_in_playlist")
+    dst_changed = False
+    if queries.item(db, dst.id, video_id) is None:
+        # Re-parent the existing row rather than delete+insert: `added_at` ("when I saved this")
+        # survives the move, and there is no window where the video belongs to neither list. The
+        # check above holds: both lists are locked, and an insert into the target (another tab's
+        # add, a YouTube merge) waits for the lock — its foreign key needs the target row.
+        row.playlist_id = dst.id
+        row.position = next_item_position(db, dst.id)
+        dst_changed = True
+    if not dst_changed:
+        db.delete(row)
+    # The source always changed; the target only when the row landed in it. Touching a target
+    # that already had the video promoted an untouched list in the rail's "recently changed" sort.
+    recompute_dirty(db, src)
+    touch(src)
+    if dst_changed:
+        recompute_dirty(db, dst)
+        touch(dst)
+    db.flush()
+
+
+def reorder_items(db: Session, pl: Playlist, order: list[str]) -> int:
+    """Number the listed videos 0..n in the listed order. Items the list left out (added in another
+    tab meanwhile) keep their relative order after them, so no two items share a position. The
+    caller holds the playlist's lock. Returns how many items the list has."""
+    by_video = {it.video_id: it for it in queries.items(db, pl.id)}
+    pos = 0
+    seen: set[str] = set()
+    for vid in order:
+        it = by_video.get(vid)
+        if it is not None:
+            it.position = pos
+            pos += 1
+            seen.add(vid)
+    for it in sorted(
+        (it for vid, it in by_video.items() if vid not in seen), key=lambda i: i.position
+    ):
+        it.position = pos
+        pos += 1
+    recompute_dirty(db, pl)
+    touch(pl)
+    db.flush()
+    return pos

backend/app/routes/discover.py

@@ -13,7 +13,6 @@ from typing import Literal
 
 from fastapi import APIRouter, Depends
 from pydantic import BaseModel
-from sqlalchemy import select
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
@@ -23,9 +22,8 @@ from app.discover import hero as hero_mod
 from app.discover import pins, rails
 from app.discover.budget import Budget, DEFAULT_BUDGET
 from app.models import User
-from app.reader.models import ComfortPin
 from app.plex.cards import norm_lang
-from app.routes._common import owned_or_404, not_found
+from app.routes._common import not_found
 
 router = APIRouter(prefix="/api/discover", tags=["discover"])
 
@@ -141,30 +139,19 @@ def add_pin(
         raise not_found("item")
     column, value = resolved
 
-    existing = db.execute(
-        select(ComfortPin).where(
-            ComfortPin.user_id == user.id,
-            getattr(ComfortPin, column) == value,
-        )
-    ).scalar_one_or_none()
+    existing = pins.pin_of(db, user, column, value)
     if existing is not None:
         return {"ok": True, "id": existing.id, "created": False}
 
-    pin = ComfortPin(user_id=user.id, position=pins.next_position(db, user), **{column: value})
-    db.add(pin)
     try:
+        pin = pins.add(db, user, column, value)
         db.commit()
     except IntegrityError:
         # A concurrent pin of the same thing from a second tab. The unique constraint did its job;
         # the user's intent ("be on my shelf") is satisfied either way, so report the winner rather
         # than a 500 the user can do nothing with.
         db.rollback()
-        winner = db.execute(
-            select(ComfortPin).where(
-                ComfortPin.user_id == user.id,
-                getattr(ComfortPin, column) == value,
-            )
-        ).scalar_one_or_none()
+        winner = pins.pin_of(db, user, column, value)
         if winner is None:
             # Not the unique constraint, then: the only other way in is the target row being pruned
             # between `resolve_ref` confirming it and this INSERT (a Plex sync dropping a removed
@@ -181,8 +168,10 @@ def remove_pin(
     user: User = Depends(current_user),
     db: Session = Depends(get_db),
 ) -> dict:
-    pin = owned_or_404(db, ComfortPin, pin_id, user, what="pin")
-    db.delete(pin)
+    pin = pins.own_pin(db, user, pin_id)
+    if pin is None:
+        raise not_found("pin")
+    pins.remove(db, pin)
     db.commit()
     return {"deleted": pin_id}
 

backend/app/routes/feed.py

@@ -1,9 +1,7 @@
-from datetime import date, datetime, timezone
+from datetime import date
 
 from fastapi import APIRouter, Depends, Query
 from pydantic import BaseModel
-from sqlalchemy import and_, func, or_, select
-from sqlalchemy.dialects.postgresql import insert as pg_insert
 from sqlalchemy.orm import Session
 from sqlalchemy.orm.exc import StaleDataError
 
@@ -11,29 +9,10 @@ from app import quota
 from app.auth import current_user
 from app.db import get_db
 from app.mailtext import lang_for_user
-from app.feedquery import (
-    decode_cursor,
-    encode_cursor,
-    feed_summary,
-    feed_years,
-    filtered_query,
-    keyset_predicate,
-    order_by,
-    rank_window_start,
-    serialize,
-    sort_keys,
-    windowed_rank,
-    windowed_relevance_rows,
-)
-from app.models import User, Video
-from app.reader.models import ChannelTag, Tag, VideoState
-from app.routes._common import (
-    ProgressIn,
-    get_or_404,
-    kept_position,
-    not_found,
-    youtube_refusal,
-)
+from app.feedquery import BadCursor, facet_counts, feed_count, feed_page, feed_summary, feed_years
+from app.models import User
+from app.reader import queries, service
+from app.routes._common import ProgressIn, kept_position, not_found, youtube_refusal
 from app.sync.videos import parse_iso8601_duration
 from app.utils import iso
 from app.youtube.client import YouTubeClient, YouTubeError
@@ -103,49 +82,20 @@ def get_feed(
     user: User = Depends(current_user),
     db: Session = Depends(get_db),
 ) -> dict:
-    query, rank_expr = filtered_query(db, user, **params)
-    lang = lang_for_user(user)
-    keys = sort_keys(sort, seed, rank_expr, lang)
-
-    vals, cursor_id, window = None, None, None
-    if cursor:
-        # The read-model raises a plain ValueError on a malformed cursor; translate it to the
-        # HTTP 400 here so feedquery stays web-framework-free.
-        try:
-            vals, cursor_id, window = decode_cursor(keys, cursor)
-        except ValueError:
-            # `from None`: the ValueError is an expected bad-input signal, not an error to chain.
-            raise refused(400, "invalid_cursor") from None
-    ranked = sort == "relevance" and rank_expr is not None
-    if ranked and not cursor:
-        window = rank_window_start(db, query)
-    if ranked and window is not None:
-        keys = sort_keys(sort, seed, windowed_rank(rank_expr, window), lang)
-
-    # Expose each key column on the row so we can build the next cursor from the last item.
-    query = query.add_columns(*[k["expr"].label(f"_k{i}") for i, k in enumerate(keys)])
-    if vals is not None:
-        for k, v in zip(keys, vals, strict=True):
-            k["value"] = v
-    if ranked and window is not None:
-        rows = windowed_relevance_rows(db, query, keys, window, cursor_id, limit + 1)
-    else:
-        query = query.order_by(*order_by(keys))
-        if cursor_id is not None:
-            query = query.where(keyset_predicate(keys, cursor_id))
-        rows = db.execute(query.limit(limit + 1)).all()
-    has_more = len(rows) > limit
-    page = rows[:limit]
-    return {
-        "items": [serialize(r) for r in page],
-        "next_cursor": (
-            encode_cursor(keys, page[-1], window if ranked else None)
-            if (has_more and page)
-            else None
-        ),
-        "has_more": has_more,
-        "limit": limit,
-    }
+    try:
+        return feed_page(
+            db,
+            user,
+            params,
+            sort=sort,
+            seed=seed,
+            limit=limit,
+            cursor=cursor,
+            lang=lang_for_user(user),
+        )
+    except BadCursor:
+        # `from None`: a malformed cursor is an expected bad-input signal, not an error to chain.
+        raise refused(400, "invalid_cursor") from None
 
 
 @router.get("/feed/count")
@@ -157,9 +107,7 @@ def get_feed_count(
     # SUPERSEDED by /feed/summary (PERF-13) — the app no longer calls this. Kept as the reference
     # oracle that test_feed_summary_matches_count_plus_facets asserts /feed/summary's `count` against;
     # remove both together with that test if this endpoint is ever retired.
-    query, _rank = filtered_query(db, user, **params)
-    total = db.scalar(select(func.count()).select_from(query.subquery()))
-    return {"count": total or 0}
+    return {"count": feed_count(db, user, **params)}
 
 
 @router.get("/facets")
@@ -181,31 +129,7 @@ def get_facets(
     SUPERSEDED by /feed/summary (PERF-13) — the app no longer calls this. Kept as the reference
     oracle for test_feed_summary_matches_count_plus_facets (it asserts /feed/summary's `facets`
     against these `counts`); retire the two together."""
-    visible = or_(ChannelTag.user_id.is_(None), ChannelTag.user_id == user.id)
-    counts: dict[int, int] = {}
-    for category in ("language", "topic", "other"):
-        # Disjunctive (OR) facets drop the category's own selections so its chips keep
-        # independent counts (you can OR more of them in). Conjunctive (AND, topics only)
-        # keeps them applied, so each remaining chip narrows to channels that ALSO have all
-        # already-selected topics — and tags that can't co-occur drop to zero (hidden).
-        conjunctive = params.get("tag_modes", {}).get(category) == "and"
-        base, _rank = filtered_query(
-            db,
-            user,
-            **{**params, "exclude_tag_category": None if conjunctive else category},
-        )
-        channels = base.with_only_columns(Video.channel_id).distinct().subquery()
-        rows = db.execute(
-            select(ChannelTag.tag_id, func.count(func.distinct(ChannelTag.channel_id)))
-            .select_from(channels)
-            .join(ChannelTag, ChannelTag.channel_id == channels.c.channel_id)
-            .join(Tag, and_(Tag.id == ChannelTag.tag_id, Tag.category == category))
-            .where(visible)
-            .group_by(ChannelTag.tag_id)
-        ).all()
-        for tag_id, count in rows:
-            counts[tag_id] = count
-    return {"counts": counts}
+    return {"counts": facet_counts(db, user, **params)}
 
 
 @router.get("/feed/summary")
@@ -234,6 +158,11 @@ def get_feed_years(
     return {"years": feed_years(db, user, **params)}
 
 
+def _video_or_404(db: Session, video_id: str) -> None:
+    if queries.video(db, video_id) is None:
+        raise not_found("video")
+
+
 class VideoStateIn(BaseModel):
     status: str
 
@@ -248,50 +177,17 @@ def set_video_state(
     status = body.status
     if status not in VALID_STATES:
         raise refused(400, "invalid_status")
-    get_or_404(db, Video, video_id, what="video")
-
-    now = datetime.now(timezone.utc)
-
+    _video_or_404(db, video_id)
     if status == "new":
-        # Un-marking: keep the row if it still holds a resume position (un-marking
-        # "watched" should restore the in-progress state, not wipe where the user left
-        # off); otherwise drop it. The in-app player fires this alongside a progress
-        # checkpoint that may DELETE the same row, so tolerate it vanishing under us.
-        row = db.execute(
-            select(VideoState).where(VideoState.user_id == user.id, VideoState.video_id == video_id)
-        ).scalar_one_or_none()
-        if row is not None:
-            if row.position_seconds:
-                row.status = "new"
-                row.watched_at = None
-            else:
-                db.delete(row)
-            try:
-                db.commit()
-            # silent-ok: the documented lost race — a concurrent player checkpoint already removed
-            # or moved this row. The state the caller asked for is the state that holds either way.
-            except StaleDataError:
-                db.rollback()
+        try:
+            service.unmark(db, user, video_id)
+            db.commit()
+        # silent-ok: the documented lost race — a concurrent player checkpoint already removed or
+        # moved this row (`service.unmark`). The state the caller asked for holds either way.
+        except StaleDataError:
+            db.rollback()
         return {"ok": True, "video_id": video_id, "status": "new"}
-
-    # watched / hidden: atomic upsert. The player marks "watched" at end-of-playback at
-    # the same instant a progress checkpoint may DELETE the in-progress row; a plain
-    # SELECT-then-UPDATE then raises StaleDataError ("0 rows matched") on that race. An
-    # INSERT … ON CONFLICT DO UPDATE keyed on uq_user_video is immune to it.
-    set_: dict = {"status": status, "updated_at": now}
-    if status == "watched":
-        set_["watched_at"] = now  # hidden keeps any existing watched_at
-    stmt = (
-        pg_insert(VideoState)
-        .values(
-            user_id=user.id,
-            video_id=video_id,
-            status=status,
-            watched_at=now if status == "watched" else None,
-        )
-        .on_conflict_do_update(constraint="uq_user_video", set_=set_)
-    )
-    db.execute(stmt)
+    service.mark(db, user, video_id, status)
     db.commit()
     return {"ok": True, "video_id": video_id, "status": status}
 
@@ -305,16 +201,12 @@ def clear_video_state(
     """Reset a video to pristine for this user — drop the whole VideoState row (status,
     watch position and watched_at), as if it had never been opened. Saved/playlist membership
     lives elsewhere and is untouched."""
-    row = db.execute(
-        select(VideoState).where(VideoState.user_id == user.id, VideoState.video_id == video_id)
-    ).scalar_one_or_none()
-    if row is not None:
-        db.delete(row)
-        try:
-            db.commit()
-        # silent-ok: the row went under the delete, which is the outcome the delete wanted.
-        except StaleDataError:
-            db.rollback()
+    try:
+        service.clear_state(db, user, video_id)
+        db.commit()
+    # silent-ok: the row went under the delete, which is the outcome the delete wanted.
+    except StaleDataError:
+        db.rollback()
     return {"ok": True, "video_id": video_id, "status": "new", "position_seconds": 0}
 
 
@@ -330,36 +222,17 @@ def set_video_progress(
     partially-watched video can render a progress bar and match the "in progress" filter.
     Trivially-early and near-finished positions clear the position rather than store it."""
     position = kept_position(body)
-
-    get_or_404(db, Video, video_id, what="video")
-
-    row = db.execute(
-        select(VideoState).where(VideoState.user_id == user.id, VideoState.video_id == video_id)
-    ).scalar_one_or_none()
-
+    _video_or_404(db, video_id)
     if not position:
-        # Nothing meaningful to store: drop a status-less row entirely, otherwise just
-        # zero the position (a saved/hidden video keeps its status).
-        if row is not None:
-            if row.status == "new":
-                db.delete(row)
-            else:
-                row.position_seconds = 0
-                row.progress_updated_at = datetime.now(timezone.utc)
-            # A concurrent state change (e.g. the end-of-playback "watched" mark) may
-            # have removed/updated this row already — tolerate the lost race.
-            try:
-                db.commit()
-            # silent-ok: exactly the race the comment above describes; the response stays true.
-            except StaleDataError:
-                db.rollback()
+        try:
+            service.set_progress(db, user, video_id, 0)
+            db.commit()
+        # silent-ok: a concurrent state change (the end-of-playback "watched" mark) removed or
+        # updated this row already — the lost race; the response stays true.
+        except StaleDataError:
+            db.rollback()
         return {"ok": True, "video_id": video_id, "position_seconds": 0}
-
-    if row is None:
-        row = VideoState(user_id=user.id, video_id=video_id)
-        db.add(row)
-    row.position_seconds = position
-    row.progress_updated_at = datetime.now(timezone.utc)
+    service.set_progress(db, user, video_id, position)
     db.commit()
     return {"ok": True, "video_id": video_id, "position_seconds": position}
 
@@ -377,7 +250,7 @@ def get_video_detail(
     our DB (e.g. a YouTube link inside another video's description that the in-app
     player navigated to) is resolved via the YouTube API (videos.list, 1 unit,
     attributed to the requesting user)."""
-    v = db.get(Video, video_id)
+    v = queries.video(db, video_id)
     if v is not None:
         return {
             "id": v.id,

backend/app/routes/playlists.py

@@ -3,32 +3,34 @@ CRUD (create / rename / delete, add / remove / reorder items) plus YouTube two-w
 mirror the user's own YouTube playlists in (`/sync-youtube`) and push/revert local edits back
 out (`/{id}/push-plan`, `/{id}/revert-youtube`), behind the optional write scope."""
 
+from typing import TYPE_CHECKING
+
 from fastapi import APIRouter, Depends, HTTPException
 from pydantic import BaseModel, Field
-from sqlalchemy import func, select, and_
-from sqlalchemy.orm import Session, aliased
+from sqlalchemy.orm import Session
 
 from app import quota
 from app.auth import current_user, has_read_scope, has_write_scope, require_write_scope
 from app.db import get_db
-from app.feedquery import feed_columns, serialize
-from app.models import Channel, User, Video
-from app.reader.models import Playlist, PlaylistItem, VideoState
+from app.feedquery import serialize
+from app.models import User
+from app.reader import queries, service
 from app.sync.playlists import (
     PlaylistGone,
-    lock_playlists,
     merge_playlist,
     plan_merge,
-    recompute_dirty,
     pull_user_playlists,
     repull_playlist,
     sync_claim,
     user_sync_claim,
 )
 from app.ratelimit import RateLimiter
-from app.routes._common import get_or_404, not_found, owned_or_404, youtube_refusal
+from app.routes._common import not_found, youtube_refusal
 from app.youtube.client import YouTubeClient, YouTubeError
 
+if TYPE_CHECKING:
+    from app.reader.models import Playlist
+
 router = APIRouter(prefix="/api/playlists", tags=["playlists"])
 
 # SEC-14, same reasoning as the sync module's limiter: a playlist pull lists every playlist the
@@ -57,90 +59,20 @@ def _refused(status: int, reason: str) -> HTTPException:
     return HTTPException(status_code=status, detail={"code": "playlist", "reason": reason})
 
 
-def _touch(pl: Playlist) -> None:
-    """Bump the playlist's `updated_at` for an edit that only writes `playlist_items`.
-
-    `UpdatedAtMixin`'s `onupdate` fires on an UPDATE of the PLAYLIST row, and add / remove /
-    reorder / move write the item rows instead — so without this the rail's "recently changed"
-    sort would only ever see renames. The assignment is what makes the column mean "last time
-    this list changed", which is what the rail promises the reader.
-
-    The DATABASE's clock, like the mixin's own `onupdate`: one sort key written from two clocks
-    (this process's and Postgres's) could order a rename and an item edit the wrong way round."""
-    pl.updated_at = func.now()
+def _own_playlist(db: Session, user: User, playlist_id: int, *, lock: bool = False) -> "Playlist":
+    pl = queries.own_playlist(db, user, playlist_id, lock=lock)
+    if pl is None:
+        raise not_found("playlist")
+    return pl
 
 
-def _next_playlist_position(db: Session, user_id: int) -> int:
-    """Append position for a new playlist in this user's rail."""
-    return (
-        db.scalar(
-            select(func.coalesce(func.max(Playlist.position), -1)).where(
-                Playlist.user_id == user_id
-            )
-        )
-        + 1
-    )
-
-
-def _next_item_position(db: Session, playlist_id: int) -> int:
-    """Append position for a new item at the end of a playlist. Locks the playlist row first: two
-    adds (two tabs, the rail and the page) otherwise both read the same max and land on one
-    position, and the list's order between them is then arbitrary."""
-    lock_playlists(db, playlist_id)
-    return (
-        db.scalar(
-            select(func.coalesce(func.max(PlaylistItem.position), -1)).where(
-                PlaylistItem.playlist_id == playlist_id
-            )
-        )
-        + 1
-    )
-
-
-def _add_item(db: Session, pl: Playlist, video_id: str, *, mark_dirty: bool) -> bool:
-    """Append a video to a playlist if not already present (idempotent). Returns whether it
-    was added. `mark_dirty` recomputes the YouTube-sync dirty flag (skip for Watch later).
-    Locked BEFORE the presence check: two adds of one video both saw "absent" and the second one's
-    insert hit uq_playlist_video as a 500."""
-    lock_playlists(db, pl.id)
-    exists = db.scalar(
-        select(PlaylistItem).where(
-            PlaylistItem.playlist_id == pl.id, PlaylistItem.video_id == video_id
-        )
-    )
-    if exists is not None:
-        return False
-    db.add(
-        PlaylistItem(playlist_id=pl.id, video_id=video_id, position=_next_item_position(db, pl.id))
-    )
-    if mark_dirty:
-        recompute_dirty(db, pl)
-    _touch(pl)
-    db.flush()  # the handler commits (CLAUDE.md: helpers flush)
-    return True
-
-
-def _remove_item(db: Session, pl: Playlist, video_id: str, *, mark_dirty: bool) -> bool:
-    """Remove a video from a playlist if present (idempotent). Returns whether it was removed.
-    `mark_dirty` recomputes the YouTube-sync dirty flag (skip for Watch later)."""
-    lock_playlists(db, pl.id)
-    item = db.scalar(
-        select(PlaylistItem).where(
-            PlaylistItem.playlist_id == pl.id, PlaylistItem.video_id == video_id
-        )
-    )
-    if item is None:
-        return False
-    db.delete(item)
-    if mark_dirty:
-        recompute_dirty(db, pl)
-    _touch(pl)
-    db.flush()  # the handler commits (CLAUDE.md: helpers flush)
-    return True
+def _video_or_404(db: Session, video_id: str) -> None:
+    if queries.video(db, video_id) is None:
+        raise not_found("video")
 
 
 def _summary(
-    pl: Playlist,
+    pl: "Playlist",
     count: int,
     total_duration: int = 0,
     has_video: bool | None = None,
@@ -157,7 +89,7 @@ def _summary(
         "total_duration_seconds": total_duration,
         "cover_thumbnail": cover,
         # When this list last CHANGED (rename, add/remove/move/reorder) — the rail's
-        # "recently changed first" sort reads it. See _touch for why it isn't automatic.
+        # "recently changed first" sort reads it. See `service.touch` for why it isn't automatic.
         "updated_at": pl.updated_at,
     }
     if has_video is not None:
@@ -173,73 +105,14 @@ def list_playlists(
 ) -> list[dict]:
     """The user's playlists (newest custom order). With `contains=<video_id>`, each entry
     also reports whether that video is already in it — used by the add-to-playlist popover."""
-    pls = (
-        db.execute(
-            select(Playlist)
-            .where(Playlist.user_id == user.id)
-            .order_by(Playlist.position, Playlist.created_at)
-        )
-        .scalars()
-        .all()
-    )
+    pls = queries.playlists_for(db, user)
     if not pls:
         return []
     ids = [p.id for p in pls]
-    counts = dict(
-        db.execute(
-            select(PlaylistItem.playlist_id, func.count())
-            .where(PlaylistItem.playlist_id.in_(ids))
-            .group_by(PlaylistItem.playlist_id)
-        ).all()
-    )
-    durations = dict(
-        db.execute(
-            select(
-                PlaylistItem.playlist_id,
-                func.coalesce(func.sum(Video.duration_seconds), 0),
-            )
-            .join(Video, Video.id == PlaylistItem.video_id)
-            .where(PlaylistItem.playlist_id.in_(ids))
-            .group_by(PlaylistItem.playlist_id)
-        ).all()
-    )
-    member: set[int] = set()
-    if contains:
-        member = set(
-            db.execute(
-                select(PlaylistItem.playlist_id).where(
-                    PlaylistItem.playlist_id.in_(ids),
-                    PlaylistItem.video_id == contains,
-                )
-            )
-            .scalars()
-            .all()
-        )
-    # Batch the cover thumbnails (first item per playlist by position) in ONE window query
-    # instead of a per-playlist SELECT inside _summary (the old N+1).
-    rn = (
-        func.row_number()
-        .over(
-            partition_by=PlaylistItem.playlist_id,
-            order_by=PlaylistItem.position,
-        )
-        .label("rn")
-    )
-    first_items = (
-        select(
-            PlaylistItem.playlist_id.label("pid"),
-            Video.thumbnail_url.label("thumb"),
-            rn,
-        )
-        .join(Video, Video.id == PlaylistItem.video_id)
-        .where(PlaylistItem.playlist_id.in_(ids))
-        .subquery()
-    )
-    covers = dict(
-        db.execute(
-            select(first_items.c.pid, first_items.c.thumb).where(first_items.c.rn == 1)
-        ).all()
-    )
+    counts = queries.item_counts(db, ids)
+    durations = queries.durations(db, ids)
+    member = queries.containing(db, ids, contains) if contains else set()
+    covers = queries.covers(db, ids)
     return [
         _summary(
             p,
@@ -270,29 +143,11 @@ def create_playlist(
     name = body.name.strip()
     if not name:
         raise _refused(400, "name_required")
-    pl = Playlist(user_id=user.id, name=name, position=_next_playlist_position(db, user.id))
-    db.add(pl)
+    pl = service.create_playlist(db, user, name)
     db.commit()
     return _summary(pl, 0)
 
 
-def _watch_later(db: Session, user: User) -> Playlist:
-    """The user's built-in Watch later playlist, created on first use."""
-    pl = db.scalar(
-        select(Playlist).where(Playlist.user_id == user.id, Playlist.kind == "watch_later")
-    )
-    if pl is None:
-        pl = Playlist(
-            user_id=user.id,
-            name="Watch later",
-            kind="watch_later",
-            position=_next_playlist_position(db, user.id),
-        )
-        db.add(pl)
-        db.flush()  # the id, for the item the caller adds next; the handler commits
-    return pl
-
-
 @router.post("/watch-later")
 def add_watch_later(
     body: VideoRefIn,
@@ -301,9 +156,9 @@ def add_watch_later(
 ) -> dict:
     """Bookmark shortcut: add a video to the built-in Watch later playlist."""
     video_id = body.video_id
-    get_or_404(db, Video, video_id, what="video")
-    pl = _watch_later(db, user)
-    _add_item(db, pl, video_id, mark_dirty=False)
+    _video_or_404(db, video_id)
+    pl = service.watch_later(db, user)
+    service.add_item(db, pl, video_id, mark_dirty=False)
     db.commit()
     return {"ok": True, "saved": True, "playlist_id": pl.id}
 
@@ -314,11 +169,9 @@ def remove_watch_later(
     user: User = Depends(current_user),
     db: Session = Depends(get_db),
 ) -> dict:
-    pl = db.scalar(
-        select(Playlist).where(Playlist.user_id == user.id, Playlist.kind == "watch_later")
-    )
+    pl = queries.watch_later(db, user)
     if pl is not None:
-        _remove_item(db, pl, video_id, mark_dirty=False)
+        service.remove_item(db, pl, video_id, mark_dirty=False)
         db.commit()
     return {"ok": True, "saved": False}
 
@@ -355,7 +208,7 @@ def revert_youtube(
     """Reset one YouTube-linked playlist to its current YouTube state, discarding local
     edits (and clearing the dirty flag). The user-facing 'undo all my changes' for a list
     once the in-session undo history is gone."""
-    pl = owned_or_404(db, Playlist, playlist_id, user, what="playlist")
+    pl = _own_playlist(db, user, playlist_id)
     if not pl.yt_playlist_id:
         raise _refused(400, "not_on_youtube")
     if not has_read_scope(user):
@@ -376,10 +229,10 @@ def revert_youtube(
             raise youtube_refusal(exc, playlist=True) from exc
 
 
-def _pushable(db: Session, user: User, playlist_id: int) -> Playlist:
+def _pushable(db: Session, user: User, playlist_id: int) -> "Playlist":
     """A local playlist the user may push to YouTube (create or update). The built-in
     Watch later and read-only YouTube mirrors are never pushable."""
-    pl = owned_or_404(db, Playlist, playlist_id, user, what="playlist")
+    pl = _own_playlist(db, user, playlist_id)
     if pl.kind == "watch_later":
         raise _refused(400, "watch_later_not_pushable")
     return pl
@@ -449,17 +302,8 @@ def get_playlist(
     user: User = Depends(current_user),
     db: Session = Depends(get_db),
 ) -> dict:
-    pl = owned_or_404(db, Playlist, playlist_id, user, what="playlist")
-    state = aliased(VideoState)
-    rows = db.execute(
-        select(*feed_columns(user, state))
-        .join(PlaylistItem, PlaylistItem.video_id == Video.id)
-        .join(Channel, Channel.id == Video.channel_id)
-        .outerjoin(state, and_(state.video_id == Video.id, state.user_id == user.id))
-        .where(PlaylistItem.playlist_id == pl.id)
-        # `id` breaks a position tie the same way on every read (the renumbering add does too).
-        .order_by(PlaylistItem.position, PlaylistItem.id)
-    ).all()
+    pl = _own_playlist(db, user, playlist_id)
+    rows = queries.playlist_rows(db, user, pl.id)
     return {
         "id": pl.id,
         "name": pl.name,
@@ -483,34 +327,15 @@ def rename_playlist(
     db: Session = Depends(get_db),
 ) -> dict:
     # Locked: the new name moves `dirty`, which the scheduled mirror refresh reads under this lock.
-    pl = owned_or_404(db, Playlist, playlist_id, user, what="playlist", lock=True)
+    pl = _own_playlist(db, user, playlist_id, lock=True)
     if "name" in body.model_fields_set:
         name = (body.name or "").strip()
         if not name:
             raise _refused(400, "name_required")
-        if name != pl.name:
-            pl.name = name
-            recompute_dirty(db, pl)
+        service.rename_playlist(db, pl, name)
     db.commit()
-    # One aggregate for count + duration (outer join so an empty playlist still returns 0),
-    # plus a single cover lookup — was 3 separate round-trips.
-    count, total = db.execute(
-        select(
-            func.count(PlaylistItem.id),
-            func.coalesce(func.sum(Video.duration_seconds), 0),
-        )
-        .select_from(PlaylistItem)
-        .outerjoin(Video, Video.id == PlaylistItem.video_id)
-        .where(PlaylistItem.playlist_id == pl.id)
-    ).one()
-    cover = db.scalar(
-        select(Video.thumbnail_url)
-        .join(PlaylistItem, PlaylistItem.video_id == Video.id)
-        .where(PlaylistItem.playlist_id == pl.id)
-        .order_by(PlaylistItem.position)
-        .limit(1)
-    )
-    return _summary(pl, count or 0, total or 0, cover=cover)
+    count, total, cover = queries.playlist_totals(db, pl.id)
+    return _summary(pl, count, total, cover=cover)
 
 
 @router.delete("/{playlist_id}")
@@ -520,7 +345,7 @@ def delete_playlist(
     user: User = Depends(current_user),
     db: Session = Depends(get_db),
 ) -> dict:
-    pl = owned_or_404(db, Playlist, playlist_id, user, what="playlist")
+    pl = _own_playlist(db, user, playlist_id)
     if pl.kind == "watch_later":
         raise _refused(400, "watch_later_undeletable")
     # Not in the middle of a sync of it (`sync_claim`): a push that read the list before the delete
@@ -541,9 +366,7 @@ def delete_playlist(
                         yt.delete_playlist(pl.yt_playlist_id)
             except YouTubeError as exc:
                 raise youtube_refusal(exc, playlist=True) from exc
-        # The playlist first, as every item write: the cascade deletes the items before the list row.
-        lock_playlists(db, pl.id)
-        db.delete(pl)  # items cascade
+        service.delete_playlist(db, pl)
         db.commit()
     return {"deleted": playlist_id}
 
@@ -562,26 +385,10 @@ def add_item(
     user: User = Depends(current_user),
     db: Session = Depends(get_db),
 ) -> dict:
-    pl = owned_or_404(db, Playlist, playlist_id, user, what="playlist")
+    pl = _own_playlist(db, user, playlist_id)
     video_id = body.video_id
-    get_or_404(db, Video, video_id, what="video")
-    added = _add_item(db, pl, video_id, mark_dirty=False)
-    if added and body.position is not None:
-        # Renumber the whole list with the new row at `position` (clamped to the end): positions
-        # can have gaps after removals, so "index N" is only meaningful over the ordered rows.
-        rows = list(
-            db.execute(
-                select(PlaylistItem)
-                .where(PlaylistItem.playlist_id == pl.id)
-                .order_by(PlaylistItem.position, PlaylistItem.id)
-            ).scalars()
-        )
-        new = next(r for r in rows if r.video_id == video_id)
-        rows.remove(new)
-        rows.insert(min(body.position, len(rows)), new)
-        for pos, row in enumerate(rows):
-            row.position = pos
-    recompute_dirty(db, pl)
+    _video_or_404(db, video_id)
+    service.add_to_playlist(db, pl, video_id, body.position)
     db.commit()
     return {"ok": True, "playlist_id": pl.id, "video_id": video_id}
 
@@ -593,8 +400,8 @@ def remove_item(
     user: User = Depends(current_user),
     db: Session = Depends(get_db),
 ) -> dict:
-    pl = owned_or_404(db, Playlist, playlist_id, user, what="playlist")
-    _remove_item(db, pl, video_id, mark_dirty=True)
+    pl = _own_playlist(db, user, playlist_id)
+    service.remove_item(db, pl, video_id, mark_dirty=True)
     db.commit()
     return {"ok": True, "playlist_id": pl.id, "video_id": video_id}
 
@@ -618,49 +425,21 @@ def move_item(
     exactly the kind of half-applied state this codebase treats as a bug rather than an edge
     case. Idempotent: if the video is already in the target, the source row is simply dropped
     (moving into a list that already has it is a removal, not a duplicate)."""
-    src = owned_or_404(db, Playlist, playlist_id, user, what="playlist")
+    src = _own_playlist(db, user, playlist_id)
     # Its own reason: "this playlist doesn't exist any more" about a TARGET that is gone was read as
     # the list on screen — which is still there.
-    dst = owned_or_404(db, Playlist, body.target_playlist_id, user, what="move_target")
-    if src.id == dst.id:
+    dst = queries.own_playlist(db, user, body.target_playlist_id)
+    if dst is None:
+        raise not_found("move_target")
+    try:
+        service.move_item(db, src, dst, video_id)
+    except service.MoveRefused as e:
+        # A same-list move is a bad request; a video a concurrent edit (another tab, a YouTube
+        # re-pull) already took out is gone.
+        status = 400 if e.reason == "same_playlist" else 404
         raise HTTPException(
-            status_code=400, detail={"code": "playlist_move", "reason": "same_playlist"}
-        )
-    # Both lists, in id order: a move A→B beside a move B→A would otherwise lock crosswise.
-    lock_playlists(db, src.id, dst.id)
-    item = db.scalar(
-        select(PlaylistItem).where(
-            PlaylistItem.playlist_id == src.id, PlaylistItem.video_id == video_id
-        )
-    )
-    if item is None:
-        # A concurrent edit (another tab, or a YouTube re-pull) already took it out.
-        raise HTTPException(
-            status_code=404, detail={"code": "playlist_move", "reason": "not_in_playlist"}
-        )
-    already = db.scalar(
-        select(PlaylistItem).where(
-            PlaylistItem.playlist_id == dst.id, PlaylistItem.video_id == video_id
-        )
-    )
-    dst_changed = False
-    if already is None:
-        # Re-parent the existing row rather than delete+insert: `added_at` ("when I saved this")
-        # survives the move, and there is no window where the video belongs to neither list. The
-        # check above holds: both lists are locked, and an insert into the target (another tab's
-        # add, a YouTube merge) waits for the lock — its foreign key needs the target row.
-        item.playlist_id = dst.id
-        item.position = _next_item_position(db, dst.id)
-        dst_changed = True
-    if not dst_changed:
-        db.delete(item)
-    # The source always changed; the target only when the row landed in it. `_touch` on a target
-    # that already had the video promoted an untouched list in the rail's "recently changed" sort.
-    recompute_dirty(db, src)
-    _touch(src)
-    if dst_changed:
-        recompute_dirty(db, dst)
-        _touch(dst)
+            status_code=status, detail={"code": "playlist_move", "reason": e.reason}
+        ) from None
     db.commit()
     return {"ok": True, "from_playlist_id": src.id, "to_playlist_id": dst.id, "video_id": video_id}
 
@@ -677,30 +456,7 @@ def reorder_items(
     db: Session = Depends(get_db),
 ) -> dict:
     # Locked: a concurrent add between reading the items and writing the order would collide.
-    pl = owned_or_404(db, Playlist, playlist_id, user, what="playlist", lock=True)
-    order = body.video_ids
-    items = {
-        it.video_id: it
-        for it in db.execute(
-            select(PlaylistItem).where(PlaylistItem.playlist_id == pl.id)
-        ).scalars()
-    }
-    pos = 0
-    seen: set[str] = set()
-    for vid in order:
-        it = items.get(vid)
-        if it is not None:
-            it.position = pos
-            pos += 1
-            seen.add(vid)
-    # Any items the payload omitted (e.g. added concurrently in another tab) keep their relative
-    # order but get fresh trailing positions, so no two items collide on the same position.
-    for it in sorted(
-        (it for vid, it in items.items() if vid not in seen), key=lambda i: i.position
-    ):
-        it.position = pos
-        pos += 1
-    recompute_dirty(db, pl)
-    _touch(pl)
+    pl = _own_playlist(db, user, playlist_id, lock=True)
+    pos = service.reorder_items(db, pl, body.video_ids)
     db.commit()
     return {"ok": True, "playlist_id": pl.id, "count": pos}

backend/app/routes/saved_views.py

@@ -1,16 +1,20 @@
+from typing import TYPE_CHECKING
+
 from fastapi import APIRouter, Depends
 from pydantic import BaseModel, field_validator
-from sqlalchemy import func, select, update
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
 from app.auth import current_user
 from app.db import get_db
 from app.models import User
-from app.reader.models import SavedView
-from app.routes._common import owned_or_404
+from app.reader import queries, service
+from app.routes._common import not_found
 from app.refusal import refused
 
+if TYPE_CHECKING:
+    from app.reader.models import SavedView
+
 # Saved "smart views": per-user named snapshots of a module's filters (feed or plex, keyed by
 # `source`).
 router = APIRouter(prefix="/api/saved-views", tags=["saved-views"])
@@ -24,7 +28,7 @@ def _norm_source(source: str) -> str:
     return source if source in _SOURCES else "feed"
 
 
-def _serialize(v: SavedView) -> dict:
+def _serialize(v: "SavedView") -> dict:
     return {
         "id": v.id,
         "name": v.name,
@@ -34,23 +38,20 @@ def _serialize(v: SavedView) -> dict:
     }
 
 
+def _own_view(db: Session, user: User, view_id: int) -> "SavedView":
+    view = queries.own_view(db, user, view_id)
+    if view is None:
+        raise not_found("view")
+    return view
+
+
 @router.get("")
 def list_views(
     source: str = "feed",
     user: User = Depends(current_user),
     db: Session = Depends(get_db),
 ) -> list[dict]:
-    source = _norm_source(source)
-    rows = (
-        db.execute(
-            select(SavedView)
-            .where(SavedView.user_id == user.id, SavedView.source == source)
-            .order_by(SavedView.position, SavedView.id)
-        )
-        .scalars()
-        .all()
-    )
-    return [_serialize(v) for v in rows]
+    return [_serialize(v) for v in queries.views(db, user, _norm_source(source))]
 
 
 # A feed view's tag pick is read by more than the feed: the autotag job keeps every tag a view names
@@ -87,22 +88,8 @@ def create_view(
     name = body.name.strip()
     if not name:
         raise refused(400, "name_required")
-    source = _norm_source(body.source)
-    # Position is per (user, source), so each module's views number 0..N independently.
-    maxpos = db.execute(
-        select(func.coalesce(func.max(SavedView.position), -1)).where(
-            SavedView.user_id == user.id, SavedView.source == source
-        )
-    ).scalar_one()
-    view = SavedView(
-        user_id=user.id,
-        source=source,
-        name=name[:80],
-        filters=body.filters,
-        position=maxpos + 1,
-    )
-    db.add(view)
     try:
+        view = service.create_view(db, user, _norm_source(body.source), name, body.filters)
         db.commit()
     except IntegrityError:
         db.rollback()
@@ -127,32 +114,17 @@ def update_view(
     user: User = Depends(current_user),
     db: Session = Depends(get_db),
 ) -> dict:
-    view = owned_or_404(db, SavedView, view_id, user, what="view")
+    view = _own_view(db, user, view_id)
     sent = body.model_fields_set
-    if "name" in sent:
-        name = (body.name or "").strip()
-        if not name:
-            raise refused(400, "name_required")
-        view.name = name[:80]
-    if "filters" in sent:
-        if body.filters is None:
-            raise refused(400, "filters_not_object")
-        view.filters = body.filters
-    if "is_default" in sent:
-        make_default = bool(body.is_default)
-        if make_default:
-            # At most one default per (user, source): clear the flag on this module's others first.
-            db.execute(
-                update(SavedView)
-                .where(
-                    SavedView.user_id == user.id,
-                    SavedView.source == view.source,
-                    SavedView.id != view.id,
-                )
-                .values(is_default=False)
-            )
-        view.is_default = make_default
+    name = (body.name or "").strip()
+    if "name" in sent and not name:
+        raise refused(400, "name_required")
+    if "filters" in sent and body.filters is None:
+        raise refused(400, "filters_not_object")
     try:
+        service.update_view(
+            db, view, sent=sent, name=name, filters=body.filters, is_default=body.is_default
+        )
         db.commit()
     except IntegrityError:
         db.rollback()
@@ -166,8 +138,7 @@ def delete_view(
     user: User = Depends(current_user),
     db: Session = Depends(get_db),
 ) -> dict:
-    view = owned_or_404(db, SavedView, view_id, user, what="view")
-    db.delete(view)
+    service.delete_view(db, _own_view(db, user, view_id))
     db.commit()
     return {"deleted": view_id}
 
@@ -182,14 +153,6 @@ def reorder_views(
     user: User = Depends(current_user),
     db: Session = Depends(get_db),
 ) -> dict:
-    ids = body.ids
-    rows = db.execute(select(SavedView).where(SavedView.user_id == user.id)).scalars().all()
-    by_id = {v.id: v for v in rows}
-    pos = 0
-    for vid in ids:
-        v = by_id.get(vid)
-        if v is not None:
-            v.position = pos
-            pos += 1
+    service.reorder_views(db, user, body.ids)
     db.commit()
     return {"ok": True}

backend/app/routes/tags.py

@@ -1,20 +1,32 @@
+from typing import TYPE_CHECKING
+
 from fastapi import APIRouter, Depends
 from pydantic import BaseModel
-from sqlalchemy import func, or_, select
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
 from app.auth import current_user
 from app.db import get_db
 from app.models import User
-from app.reader.models import ChannelTag, Tag
-from app.routes._common import owned_or_404
+from app.reader import queries, service
+from app.routes._common import not_found
 from app.sync.tagsuggest import compute_tag_suggestions
 from app.refusal import refused
 
+if TYPE_CHECKING:
+    from app.reader.models import Tag
+
 router = APIRouter(prefix="/api/tags", tags=["tags"])
 
 
+def _own_tag(db: Session, user: User, tag_id: int) -> "Tag":
+    """One of the user's own tags, or a 404 — a system tag (no owner) is not theirs to edit."""
+    tag = queries.own_tag(db, user, tag_id)
+    if tag is None:
+        raise not_found("tag")
+    return tag
+
+
 @router.get("/suggestions")
 def tag_suggestions(
     user: User = Depends(current_user), db: Session = Depends(get_db)
@@ -28,13 +40,6 @@ def tag_suggestions(
 
 @router.get("")
 def list_tags(user: User = Depends(current_user), db: Session = Depends(get_db)) -> list[dict]:
-    rows = db.execute(
-        select(Tag, func.count(ChannelTag.id))
-        .outerjoin(ChannelTag, ChannelTag.tag_id == Tag.id)
-        .where(or_(Tag.user_id.is_(None), Tag.user_id == user.id))
-        .group_by(Tag.id)
-        .order_by(Tag.category, Tag.name)
-    ).all()
     return [
         {
             "id": tag.id,
@@ -44,7 +49,7 @@ def list_tags(user: User = Depends(current_user), db: Session = Depends(get_db))
             "system": tag.user_id is None,
             "channel_count": count,
         }
-        for tag, count in rows
+        for tag, count in queries.tags_with_counts(db, user)
     ]
 
 
@@ -63,16 +68,8 @@ def create_tag(
     name = body.name.strip()
     if not name:
         raise refused(400, "name_required")
-    # An unknown category silently falls back to "other" (unchanged behaviour) rather than 422.
-    category = body.category if body.category in ("language", "topic", "other") else "other"
-    tag = Tag(
-        user_id=user.id,
-        name=name,
-        color=body.color,
-        category=category,
-    )
-    db.add(tag)
     try:
+        tag = service.create_tag(db, user, name, body.category, body.color)
         db.commit()
     except IntegrityError:
         db.rollback()
@@ -100,17 +97,13 @@ def update_tag(
     user: User = Depends(current_user),
     db: Session = Depends(get_db),
 ) -> dict:
-    # System tags (user_id NULL) are not user-editable: owned_or_404 404s them too.
-    tag = owned_or_404(db, Tag, tag_id, user, what="tag")
+    tag = _own_tag(db, user, tag_id)
     sent = body.model_fields_set
-    if "name" in sent:
-        name = (body.name or "").strip()
-        if not name:
-            raise refused(400, "name_required")
-        tag.name = name
-    if "color" in sent:
-        tag.color = body.color
+    name = (body.name or "").strip()
+    if "name" in sent and not name:
+        raise refused(400, "name_required")
     try:
+        service.update_tag(tag, sent=sent, name=name, color=body.color)
         db.commit()
     except IntegrityError:
         db.rollback()
@@ -130,8 +123,6 @@ def delete_tag(
     user: User = Depends(current_user),
     db: Session = Depends(get_db),
 ) -> dict:
-    # System tags (user_id NULL) are not user-editable: owned_or_404 404s them too.
-    tag = owned_or_404(db, Tag, tag_id, user, what="tag")
-    db.delete(tag)  # ChannelTag rows cascade on tag delete.
+    service.delete_tag(db, _own_tag(db, user, tag_id))
     db.commit()
     return {"deleted": tag_id}

backend/app/sync/autotag.py

@@ -8,12 +8,13 @@ regenerated freely; user tags are never touched here.
 
 import re
 
-from sqlalchemy import and_, exists, func, select
+from sqlalchemy import func, select
 from sqlalchemy.orm import Session
 
 from app import logs, progress, sysconfig
 from app.models import Channel, Video
-from app.reader.models import ChannelTag, SavedView, Tag
+from app.reader import queries as reader_queries
+from app.reader import service as reader_service
 
 log = logs.get_logger(__name__)
 
@@ -210,71 +211,30 @@ def compute_channel_topics(db: Session, channel: Channel) -> set[str]:
     return topics
 
 
-def ensure_system_tag(db: Session, name: str, category: str, color: str | None) -> Tag:
-    tag = db.execute(
-        select(Tag).where(Tag.user_id.is_(None), Tag.name == name)
-    ).scalar_one_or_none()
-    if tag is None:
-        tag = Tag(user_id=None, name=name, category=category, color=color)
-        db.add(tag)
-        db.flush()
-    return tag
-
-
 def apply_channel_autotags(db: Session, channel: Channel) -> None:
     desired: dict[int, float | None] = {}
 
     lang, confidence = detect_channel_language(db, channel)
     if lang:
-        tag = ensure_system_tag(db, LANG_NAMES.get(lang, lang.upper()), "language", LANG_COLOR)
+        tag = reader_service.ensure_system_tag(
+            db, LANG_NAMES.get(lang, lang.upper()), "language", LANG_COLOR
+        )
         desired[tag.id] = confidence
 
     for topic in compute_channel_topics(db, channel):
-        tag = ensure_system_tag(db, topic, "topic", TOPIC_COLORS.get(topic, DEFAULT_TOPIC_COLOR))
+        tag = reader_service.ensure_system_tag(
+            db, topic, "topic", TOPIC_COLORS.get(topic, DEFAULT_TOPIC_COLOR)
+        )
         desired.setdefault(tag.id, None)
 
-    existing = {
-        ct.tag_id: ct
-        for ct in db.execute(
-            select(ChannelTag).where(
-                ChannelTag.channel_id == channel.id,
-                ChannelTag.user_id.is_(None),
-                ChannelTag.source == "auto",
-            )
-        ).scalars()
-    }
-    for tag_id, ct in existing.items():
-        if tag_id not in desired:
-            db.delete(ct)
-    for tag_id, conf in desired.items():
-        ct = existing.get(tag_id)
-        if ct is None:
-            db.add(
-                ChannelTag(
-                    channel_id=channel.id,
-                    tag_id=tag_id,
-                    user_id=None,
-                    source="auto",
-                    confidence=conf,
-                )
-            )
-        else:
-            ct.confidence = conf
+    reader_service.set_auto_tags(db, channel.id, desired)
     db.commit()
 
 
 def run_autotag_all(db: Session, only_missing: bool = False) -> dict:
     query = select(Channel)
     if only_missing:
-        query = query.where(
-            ~exists().where(
-                and_(
-                    ChannelTag.channel_id == Channel.id,
-                    ChannelTag.source == "auto",
-                    ChannelTag.user_id.is_(None),
-                )
-            )
-        )
+        query = query.where(~reader_queries.has_auto_tags(Channel.id))
     channels = db.execute(query).scalars().all()
     tagged = 0
     for i, channel in enumerate(channels, 1):
@@ -285,40 +245,6 @@ def run_autotag_all(db: Session, only_missing: bool = False) -> dict:
             db.rollback()
             log.exception("Auto-tagging failed for channel %s", channel.id)
         progress.report(i, len(channels), "autotag")
-    removed = _cleanup_orphan_system_tags(db)
-    return {"channels_tagged": tagged, "orphan_tags_removed": removed}
-
-
-def _view_tag_ids(filters: object) -> list[int]:
-    """The tag ids a stored view's filters name; anything else in that slot names none."""
-    tags = filters.get("tags") if isinstance(filters, dict) else None
-    if not isinstance(tags, list):
-        return []
-    return [t for t in tags if type(t) is int]  # `type`, not isinstance: True is not a tag id
-
-
-def _cleanup_orphan_system_tags(db: Session) -> int:
-    """Delete system tags that no longer link to any channel (e.g. after a mapping change) — but not
-    one a saved feed view filters on. Views hold tag IDS, and a deleted tag comes back under a new id
-    when a channel earns it again, so the view's filter would point at nothing; the feed drops an
-    unknown id, and the view showed the whole feed under its old name.
-
-    The views are read defensively, and the kept ids are filtered here rather than in SQL: a view
-    stored before its route checked the shape (or written around it) must not raise for everyone's
-    job, nor push a `NOT IN` past the driver's bind-parameter limit."""
-    in_views = {
-        t
-        for (filters,) in db.execute(select(SavedView.filters).where(SavedView.source == "feed"))
-        for t in _view_tag_ids(filters)
-    }
-    orphans = [
-        tag
-        for tag in db.execute(
-            select(Tag).where(Tag.user_id.is_(None), ~exists().where(ChannelTag.tag_id == Tag.id))
-        ).scalars()
-        if tag.id not in in_views
-    ]
-    for tag in orphans:
-        db.delete(tag)
+    removed = reader_service.prune_orphan_system_tags(db)
     db.commit()
-    return len(orphans)
+    return {"channels_tagged": tagged, "orphan_tags_removed": removed}

backend/app/sync/explore.py

@@ -15,7 +15,7 @@ from sqlalchemy.orm import Session
 
 from app import logs, sysconfig
 from app.models import Channel, ExploredChannel, SearchFind, Subscription, Video
-from app.reader.models import PlaylistItem, VideoState
+from app.reader import queries as reader_queries
 from app.sync.videos import insert_stubs, stub_from_playlist_item, apply_video_details
 from app.youtube.client import YouTubeClient, YouTubeError, is_playlist_gone
 
@@ -122,15 +122,12 @@ def purge_unkept_explored(db: Session) -> dict:
     if not orphan_ids:
         return {"expired": expired, "videos_deleted": 0, "channels_deleted": 0}
 
-    state_exists = select(VideoState.id).where(VideoState.video_id == Video.id).exists()
-    pl_exists = select(PlaylistItem.id).where(PlaylistItem.video_id == Video.id).exists()
     find_exists = select(SearchFind.id).where(SearchFind.video_id == Video.id).exists()
     videos_deleted = db.execute(
         delete(Video).where(
             Video.channel_id.in_(orphan_ids),
             Video.via_explore.is_(True),
-            ~state_exists,
-            ~pl_exists,
+            ~reader_queries.kept_by_a_reader(Video.id),
             ~find_exists,
         )
     ).rowcount
@@ -159,15 +156,12 @@ def purge_unkept_search(db: Session, grace_days: int | None = None) -> dict:
     0 quota."""
     grace = sysconfig.get_int(db, "search_grace_days") if grace_days is None else grace_days
     cutoff = datetime.now(timezone.utc) - timedelta(days=grace)
-    state_exists = select(VideoState.id).where(VideoState.video_id == Video.id).exists()
-    pl_exists = select(PlaylistItem.id).where(PlaylistItem.video_id == Video.id).exists()
     sub_exists = select(Subscription.id).where(Subscription.channel_id == Video.channel_id).exists()
     deleted = db.execute(
         delete(Video).where(
             Video.via_search.is_(True),
             Video.created_at < cutoff,
-            ~state_exists,
-            ~pl_exists,
+            ~reader_queries.kept_by_a_reader(Video.id),
             ~sub_exists,
         )
     ).rowcount

backend/app/sync/maintenance.py

@@ -28,7 +28,7 @@ from sqlalchemy.orm import Session
 from app import logs, progress, quota, sysconfig
 from app.config import settings
 from app.models import Video
-from app.reader.models import Playlist, PlaylistItem
+from app.reader import queries as reader_queries
 from app.notifications import create_notification
 from app.sync.runner import get_service_user
 from app.sync.videos import apply_video_details, parse_dt
@@ -89,11 +89,7 @@ def _collect_user_impact(db: Session, videos: list[Video]) -> dict[int, list[dic
     # Anyone who had it in a playlist — which now also covers "saved": Watch later became a
     # built-in playlist, so "saved" is no longer a VideoState status (the old status=="saved"
     # scan matched nothing and under-counted impacted users).
-    for user_id, video_id in db.execute(
-        select(Playlist.user_id, PlaylistItem.video_id)
-        .join(PlaylistItem, PlaylistItem.playlist_id == Playlist.id)
-        .where(PlaylistItem.video_id.in_(ids))
-    ):
+    for user_id, video_id in reader_queries.playlisted_by(db, ids):
         impact.setdefault(user_id, set()).add(video_id)
     return {
         user_id: [

backend/app/sync/playlists.py

@@ -12,7 +12,6 @@ import json
 from collections.abc import Callable, Iterator
 from contextlib import AbstractContextManager, contextmanager
 
-from datetime import datetime, timezone
 from typing import TypeVar
 
 from sqlalchemy import Engine, create_engine, delete, func, select
@@ -21,13 +20,14 @@ from sqlalchemy.pool import NullPool
 
 from app import logs, progress, quota
 from app.auth import has_read_scope
-from app.models import Channel, User, Video
+from app.models import User, Video
 from app.reader.models import Playlist, PlaylistItem
 from app.sync.playlist_merge import merge_plan
 from app.sync.runner import token_users
 from app.sync.subscriptions import blank_listing, refuse_blank_listing, vanished
-from app.sync.videos import apply_video_details, parse_dt
+from app.sync.videos import ensure_videos
 from app.youtube.client import YouTubeClient, YouTubeError, is_playlist_gone
+from app.userscope import owned_by
 
 log = logs.get_logger(__name__)
 # Cost of a single YouTube write op (playlists/playlistItems insert/update/delete).
@@ -228,55 +228,11 @@ def recompute_dirty(db: Session, pl: Playlist) -> None:
     pl.dirty = pl.synced_fingerprint != current_fingerprint(db, pl)
 
 
-def _ensure_videos(db: Session, yt: YouTubeClient, video_ids: list[str]) -> None:
-    """Make sure every given video id exists in the catalog, fetching + ingesting the
-    missing ones (and a stub channel for any unknown channel). Deleted/private videos that
-    the API won't return are simply left out."""
-    if not video_ids:
-        return
-    have = set(db.execute(select(Video.id).where(Video.id.in_(video_ids))).scalars().all())
-    missing = [v for v in dict.fromkeys(video_ids) if v not in have]
-    if not missing:
-        return
-    items = yt.get_videos(missing)
-    chan_ids = {
-        it.get("snippet", {}).get("channelId")
-        for it in items
-        if it.get("snippet", {}).get("channelId")
-    }
-    existing_ch = set(
-        db.execute(select(Channel.id).where(Channel.id.in_(chan_ids))).scalars().all()
-    )
-    now = datetime.now(timezone.utc)
-    for it in items:
-        sn = it.get("snippet", {})
-        cid = sn.get("channelId")
-        if cid and cid not in existing_ch:
-            db.add(Channel(id=cid, title=sn.get("channelTitle")))
-            existing_ch.add(cid)
-    db.flush()
-    seen: set[str] = set()
-    for it in items:
-        vid = it.get("id")
-        sn = it.get("snippet", {})
-        cid = sn.get("channelId")
-        if not vid or not cid or vid in seen:
-            continue
-        seen.add(vid)
-        # published_at is NOT NULL (0068); apply_video_details re-fills a missing date, but seed a now()
-        # fallback here too so the constructor never carries an explicit NULL.
-        v = Video(id=vid, channel_id=cid, published_at=parse_dt(sn.get("publishedAt")) or now)
-        apply_video_details(v, it)
-        v.enriched_at = now
-        db.add(v)
-    db.commit()
-
-
 def _replace_items_from_youtube(db: Session, pl: Playlist, ids: list[str]) -> list[str]:
     """Mirror a YouTube playlist's ordered video ids into `pl`'s local items (authoritative
     replace): keep only those present in the catalog in their YT order (de-duplicated), swap the
     items out, and reset the synced fingerprint + dirty flag. The caller has ingested `ids`
-    (`_ensure_videos`, which commits) and THEN taken `lock_playlists`, set `pl.name`, and commits
+    (`sync.videos.ensure_videos`, which commits) and THEN taken `lock_playlists`, set `pl.name`, and commits
     after: an ingest in here re-fetched every id YouTube will not return (a private or deleted
     video) and its commit let go of the lock before the replace. Returns the final ordered ids."""
     present = (
@@ -347,7 +303,9 @@ def pull_user_playlists(db: Session, user: User) -> dict:
         existing = {
             pl.yt_playlist_id: pl
             for pl in db.execute(
-                select(Playlist).where(Playlist.user_id == user.id, Playlist.source == "youtube")
+                select(Playlist).where(
+                    owned_by(Playlist.user_id, user.id), Playlist.source == "youtube"
+                )
             ).scalars()
         }
         # YouTube playlists that we already own as local (exported) playlists must not be
@@ -355,7 +313,7 @@ def pull_user_playlists(db: Session, user: User) -> dict:
         owned_yt_ids = set(
             db.execute(
                 select(Playlist.yt_playlist_id).where(
-                    Playlist.user_id == user.id,
+                    owned_by(Playlist.user_id, user.id),
                     Playlist.source == "local",
                     Playlist.yt_playlist_id.is_not(None),
                 )
@@ -402,7 +360,7 @@ def pull_user_playlists(db: Session, user: User) -> dict:
                     continue
                 # Ingest first — it commits — so the row lock below is held from the dirty check
                 # to the commit that replaces the items.
-                _ensure_videos(db, yt, ids)
+                ensure_videos(db, yt, ids)
                 # `dirty` read above is a snapshot from before the YouTube calls (sessions keep
                 # loaded rows past a commit): a remove or reorder made meanwhile must not be
                 # replaced by YouTube's order. Item writes lock the playlist row, so this waits for
@@ -613,7 +571,7 @@ def merge_playlist(db: Session, user: User, pl: Playlist, *, live=None) -> dict:
         prev_unavailable = parse_unavailable(pl)
         still_listed = [v for v in yt_ids if v in prev_unavailable]
         if still_listed:
-            _ensure_videos(db, yt, still_listed)
+            ensure_videos(db, yt, still_listed)
             healed = set(
                 db.execute(select(Video.id).where(Video.id.in_(still_listed))).scalars().all()
             )
@@ -625,7 +583,7 @@ def merge_playlist(db: Session, user: User, pl: Playlist, *, live=None) -> dict:
         # --- Siftlode side (free, direct): pull YouTube-adds in, drop YouTube-deletes, land the
         # merged order authoritatively. Do this first so local reflects the merge even if a later
         # YouTube write trips the breaker.
-        _ensure_videos(db, yt, mp.add_to_local)
+        ensure_videos(db, yt, mp.add_to_local)
         # `local` was read before the YouTube calls above: an add or remove committed meanwhile
         # (another tab) would be undone by the replace below. Lock the list — the ingest's commit is
         # behind us — and plan again from what is there now. Nothing new needs ingesting: a video
@@ -750,7 +708,7 @@ def repull_playlist(db: Session, user: User, pl: Playlist) -> dict:
             )
         ids = list(yt.iter_my_playlist_video_ids(pl.yt_playlist_id))
         refuse_blank_listing(ids, snippet.get("item_count"))
-        _ensure_videos(db, yt, ids)
+        ensure_videos(db, yt, ids)
         _lock_or_gone(db, pl)
         pl.name = snippet.get("title") or pl.name
         ordered = _replace_items_from_youtube(db, pl, ids)

backend/app/sync/tagsuggest.py

@@ -25,6 +25,7 @@ from sqlalchemy.orm import Session
 
 from app.models import Subscription, User
 from app.reader.models import ChannelTag, Tag
+from app.userscope import owned_by
 
 # A user tag needs at least this many manually-tagged example channels before we trust its
 # profile — fewer is noise (one gaming-cooking channel shouldn't make "cooking" mean "gaming").
@@ -58,7 +59,9 @@ def compute_tag_suggestions(db: Session, user: User) -> dict[str, list[dict]]:
     # Channel Manager shows ghost chips only on non-hidden rows, so suggesting for hidden channels would
     # inflate the header count past what's shown and let bulk-accept tag channels the user can't see).
     subs = db.execute(
-        select(Subscription.channel_id, Subscription.hidden).where(Subscription.user_id == user.id)
+        select(Subscription.channel_id, Subscription.hidden).where(
+            owned_by(Subscription.user_id, user.id)
+        )
     ).all()
     subscribed: set[str] = {cid for cid, _ in subs}
     hidden: set[str] = {cid for cid, h in subs if h}
@@ -75,7 +78,7 @@ def compute_tag_suggestions(db: Session, user: User) -> dict[str, list[dict]]:
     user_tagged: set[str] = set()
     for tag_id, cid in db.execute(
         select(ChannelTag.tag_id, ChannelTag.channel_id).where(
-            ChannelTag.user_id == user.id, ChannelTag.source == "user"
+            owned_by(ChannelTag.user_id, user.id), ChannelTag.source == "user"
         )
     ).all():
         if cid not in subscribed:

backend/app/sync/videos.py

@@ -418,3 +418,48 @@ def run_shorts_classification(db: Session, limit: int | None = None) -> dict:
                 progress.report(probed, len(candidates), "probe")
     db.commit()
     return {"probed": probed, "shorts": shorts}
+
+
+def ensure_videos(db: Session, yt: YouTubeClient, video_ids: list[str]) -> None:
+    """Make sure every given video id exists in the catalog, fetching + ingesting the
+    missing ones (and a stub channel for any unknown channel). Deleted/private videos that
+    the API won't return are simply left out. The catalog's write for a playlist sync that names
+    videos it has never seen; commits."""
+    if not video_ids:
+        return
+    have = set(db.execute(select(Video.id).where(Video.id.in_(video_ids))).scalars().all())
+    missing = [v for v in dict.fromkeys(video_ids) if v not in have]
+    if not missing:
+        return
+    items = yt.get_videos(missing)
+    chan_ids = {
+        it.get("snippet", {}).get("channelId")
+        for it in items
+        if it.get("snippet", {}).get("channelId")
+    }
+    existing_ch = set(
+        db.execute(select(Channel.id).where(Channel.id.in_(chan_ids))).scalars().all()
+    )
+    now = datetime.now(timezone.utc)
+    for it in items:
+        sn = it.get("snippet", {})
+        cid = sn.get("channelId")
+        if cid and cid not in existing_ch:
+            db.add(Channel(id=cid, title=sn.get("channelTitle")))
+            existing_ch.add(cid)
+    db.flush()
+    seen: set[str] = set()
+    for it in items:
+        vid = it.get("id")
+        sn = it.get("snippet", {})
+        cid = sn.get("channelId")
+        if not vid or not cid or vid in seen:
+            continue
+        seen.add(vid)
+        # published_at is NOT NULL (0068); apply_video_details re-fills a missing date, but seed a now()
+        # fallback here too so the constructor never carries an explicit NULL.
+        v = Video(id=vid, channel_id=cid, published_at=parse_dt(sn.get("publishedAt")) or now)
+        apply_video_details(v, it)
+        v.enriched_at = now
+        db.add(v)
+    db.commit()