diff --git a/CLAUDE.md b/CLAUDE.md index 66d0071e..40cd024a 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -63,7 +63,7 @@ how it gets built. Concretely: ## Architecture shape (don't fight it) `routes/*` (HTTP only: validation, auth-dependency, serialization) → domain packages `sync/`, `plex/`, -`downloads/`, `reader/`, `discover/`, `youtube/`, `issues/` (shared by routes + scheduler + worker) → flat core (`auth`, `apitokens`, `quota`, `jobruns`, +`downloads/`, `reader/`, `discover/`, `catalog/`, `youtube/`, `issues/` (shared by routes + scheduler + worker) → flat core (`auth`, `apitokens`, `quota`, `jobruns`, `sysconfig`, `state`, `audit`, `notifications`, `userscope`, `ratelimit`, `bodylimit`, `progress`, `collation`, `security`, `logs`, `logsink`, `diagnostics`, `debugsession`) → `models/db/config`. **Dependency direction is one-way: core never imports routes.** Frontend: `main.tsx` provider tree → `App.tsx` shell (a shared `/watch` link boots from its own diff --git a/backend/.basedpyright/baseline.json b/backend/.basedpyright/baseline.json index 5468b8db..3fc631dc 100644 --- a/backend/.basedpyright/baseline.json +++ b/backend/.basedpyright/baseline.json @@ -640,48 +640,6 @@ } } ], - "./app/routes/channels.py": [ - { - "code": "reportReturnType", - "range": { - "startColumn": 11, - "endColumn": 11, - "lineCount": 6 - } - }, - { - "code": "reportArgumentType", - "range": { - "startColumn": 8, - "endColumn": 15, - "lineCount": 1 - } - }, - { - "code": "reportArgumentType", - "range": { - "startColumn": 49, - "endColumn": 56, - "lineCount": 1 - } - }, - { - "code": "reportAttributeAccessIssue", - "range": { - "startColumn": 38, - "endColumn": 44, - "lineCount": 1 - } - }, - { - "code": "reportAttributeAccessIssue", - "range": { - "startColumn": 33, - "endColumn": 39, - "lineCount": 1 - } - } - ], "./app/routes/discover.py": [ { "code": "reportInvalidTypeForm", @@ -967,14 +925,6 @@ } ], "./app/routes/scheduler.py": [ - { - "code": "reportArgumentType", - "range": { - "startColumn": 42, - "endColumn": 55, - "lineCount": 1 - } - }, { "code": "reportOptionalMemberAccess", "range": { @@ -992,42 +942,6 @@ } } ], - "./app/routes/search.py": [ - { - "code": "reportArgumentType", - "range": { - "startColumn": 25, - "endColumn": 31, - "lineCount": 1 - } - }, - { - "code": "reportAttributeAccessIssue", - "range": { - "startColumn": 6, - "endColumn": 14, - "lineCount": 1 - } - } - ], - "./app/routes/sync.py": [ - { - "code": "reportArgumentType", - "range": { - "startColumn": 42, - "endColumn": 65, - "lineCount": 1 - } - }, - { - "code": "reportAttributeAccessIssue", - "range": { - "startColumn": 42, - "endColumn": 50, - "lineCount": 1 - } - } - ], "./app/sync/explore.py": [ { "code": "reportAttributeAccessIssue", diff --git a/backend/app/catalog/queries.py b/backend/app/catalog/queries.py new file mode 100644 index 00000000..6d98ba61 --- /dev/null +++ b/backend/app/catalog/queries.py @@ -0,0 +1,296 @@ +"""Catalog's reads: channels and videos, and each user's footprint in them (subscriptions, blocks, +explorations), for the HTTP layer and for the core code that asks about them. + +A user's row is read through `userscope`, so someone else's is the same answer as a missing one. +Catalog does not read another unit's tables: what it needs from Reader (a playlist's channels, a +user's tag links) it asks `reader.queries`. +""" + +from datetime import datetime + +from sqlalchemy import Select, and_, case, func, or_, select +from sqlalchemy.orm import Session + +from app.catalog.models import ( + LIVE_OR_UPCOMING, + BlockedChannel, + Channel, + ExploredChannel, + Subscription, + Video, +) +from app.collation import collated +from app.models import User +from app.sync.explore import followed_channel +from app.userscope import owned_by + +# --- channels -------------------------------------------------------------------------------- + + +def channel(db: Session, channel_id: str) -> Channel | None: + return db.get(Channel, channel_id) + + +def channel_title(db: Session, channel_id: str) -> str | None: + return db.scalar(select(Channel.title).where(Channel.id == channel_id)) + + +def channel_titles(db: Session, ids: set[str]) -> dict[str, str]: + """Each channel's title by id; a channel with no title yet is left out.""" + if not ids: + return {} + rows = db.execute(select(Channel.id, Channel.title).where(Channel.id.in_(ids))) + return {cid: title for cid, title in rows.tuples() if title is not None} + + +def channel_aggregates(channel_ids: list[str]) -> Select: + """Per-channel stored count, last upload and total duration, in one grouped pass. Every column it + reads is in `ix_videos_channel_pub_cover` (migration 0095), so Postgres answers it from the index + alone; a test holds that (test_channel_list_index_only).""" + return ( + select( + Video.channel_id, + func.count(Video.id), + func.max(Video.published_at), + func.coalesce(func.sum(Video.duration_seconds), 0), + ) + .where(Video.channel_id.in_(channel_ids)) + .group_by(Video.channel_id) + ) + + +def aggregates(db: Session, channel_ids: list[str]) -> list[tuple[str, int, datetime | None, int]]: + """`channel_aggregates`, run: (channel id, stored videos, last upload, total seconds).""" + if not channel_ids: + return [] + return list(db.execute(channel_aggregates(channel_ids)).tuples()) + + +def video_counts(db: Session, channel_id: str) -> tuple[int, int, int]: + """A channel's stored videos, its Shorts and its live/upcoming ones, in one grouped pass.""" + n_all, n_short, n_live = db.execute( + select( + func.count(Video.id), + func.count(Video.id).filter(Video.is_short.is_(True)), + func.count(Video.id).filter(Video.live_status.in_(LIVE_OR_UPCOMING)), + ).where(Video.channel_id == channel_id) + ).one() + return int(n_all or 0), int(n_short or 0), int(n_live or 0) + + +def deep_queued(db: Session, channel_ids: list[str]) -> set[str]: + """Which of these channels at least one user (any user) has asked full history for.""" + if not channel_ids: + return set() + return set( + db.execute( + select(Subscription.channel_id) + .where( + Subscription.channel_id.in_(channel_ids), + Subscription.deep_requested.is_(True), + ) + .distinct() + ).scalars() + ) + + +def followed(db: Session, channel_id: str) -> bool: + """Whether anyone subscribes to the channel.""" + return ( + db.execute( + select(Subscription.id).where(Subscription.channel_id == channel_id).limit(1) + ).first() + is not None + ) + + +# --- a user's subscriptions, blocks and explorations ---------------------------------------- + + +def subscriptions(db: Session, user: User, lang: str) -> list[tuple[Subscription, Channel]]: + """The user's subscriptions with their channels, highest priority first, then by title in the + reader's alphabet.""" + return list( + db.execute( + select(Subscription, Channel) + .join(Channel, Channel.id == Subscription.channel_id) + .where(owned_by(Subscription.user_id, user)) + .order_by(Subscription.priority.desc(), collated(Channel.title, lang)) + ).tuples() + ) + + +def subscription(db: Session, user: User, channel_id: str) -> Subscription | None: + return db.execute( + select(Subscription).where( + owned_by(Subscription.user_id, user), Subscription.channel_id == channel_id + ) + ).scalar_one_or_none() + + +def subscribed_channels(user: User): + """The channel ids the user subscribes to, as a subquery.""" + return select(Subscription.channel_id).where(owned_by(Subscription.user_id, user)) + + +def exploration(db: Session, user: User, channel_id: str) -> ExploredChannel | None: + return db.execute( + select(ExploredChannel).where( + owned_by(ExploredChannel.user_id, user), ExploredChannel.channel_id == channel_id + ) + ).scalar_one_or_none() + + +def is_blocked(db: Session, user: User, channel_id: str) -> bool: + return ( + db.execute( + select(BlockedChannel.id).where( + owned_by(BlockedChannel.user_id, user), BlockedChannel.channel_id == channel_id + ) + ).first() + is not None + ) + + +def blocked_ids(db: Session, user: User) -> set[str]: + return set( + db.scalars(select(BlockedChannel.channel_id).where(owned_by(BlockedChannel.user_id, user))) + ) + + +def blocked_channels(db: Session, user: User, lang: str) -> list[Channel]: + return list( + db.execute( + select(Channel) + .join(BlockedChannel, BlockedChannel.channel_id == Channel.id) + .where(owned_by(BlockedChannel.user_id, user)) + .order_by(collated(Channel.title, lang)) + ).scalars() + ) + + +# --- sync progress --------------------------------------------------------------------------- + + +def catalog_totals(db: Session, user: User) -> dict: + """The instance's catalog size and this user's subscription count (the sync status card).""" + return { + "subscriptions": db.scalar( + select(func.count()) + .select_from(Subscription) + .where(owned_by(Subscription.user_id, user)) + ), + "channels_total": db.scalar(select(func.count()).select_from(Channel)), + "channels_backfilling": db.scalar( + select(func.count()).select_from(Channel).where(Channel.backfill_done.is_(False)) + ), + "videos_total": db.scalar(select(func.count()).select_from(Video)), + "pending_enrich": db.scalar( + select(func.count()).select_from(Video).where(Video.enriched_at.is_(None)) + ), + } + + +def my_progress(db: Session, user: User) -> tuple[int, int, int, int]: + """Over the channels this user subscribes to: how many, how many have details, how many their + recent uploads, how many their whole history.""" + total, details_synced, recent_synced, all_deep_done = db.execute( + select( + func.count(Channel.id), + func.count(Channel.details_synced_at), + func.count(Channel.recent_synced_at), + func.sum(case((Channel.backfill_done.is_(True), 1), else_=0)), + ) + .select_from(Channel) + .join( + Subscription, + and_(Subscription.channel_id == Channel.id, owned_by(Subscription.user_id, user)), + ) + ).one() + return total or 0, details_synced or 0, recent_synced or 0, int(all_deep_done or 0) + + +def my_deep_requested(db: Session, user: User) -> list[Channel]: + """The channels this user asked full history for.""" + return list( + db.execute( + select(Channel).join( + Subscription, + and_( + Subscription.channel_id == Channel.id, + owned_by(Subscription.user_id, user), + Subscription.deep_requested.is_(True), + ), + ) + ).scalars() + ) + + +def my_video_count(db: Session, user: User) -> int: + """The videos of the channels this user subscribes to and has not hidden.""" + return ( + db.scalar( + select(func.count(Video.id)).join( + Subscription, + and_( + Subscription.channel_id == Video.channel_id, + owned_by(Subscription.user_id, user), + Subscription.hidden.is_(False), + ), + ) + ) + or 0 + ) + + +def video_total(db: Session) -> int: + return db.scalar(select(func.count(Video.id))) or 0 + + +def deep_backlog(db: Session) -> list[Channel]: + """Every channel someone asked full history for that does not have it yet.""" + deep_requested = ( + select(Subscription.id) + .where(Subscription.channel_id == Channel.id, Subscription.deep_requested.is_(True)) + .exists() + ) + return list( + db.execute( + select(Channel).where(Channel.backfill_done.is_(False), deep_requested) + ).scalars() + ) + + +def work_queue(db: Session) -> dict: + """What the catalog's jobs still have queued, instance-wide (the scheduler card).""" + return { + # The same selection run_recent_backfill works through, or the card counts channels the + # job will never touch and never reaches zero. + "channels_recent_pending": db.scalar( + select(func.count()) + .select_from(Channel) + .where(Channel.recent_synced_at.is_(None), followed_channel()) + ) + or 0, + "videos_pending_enrich": db.scalar( + select(func.count()).select_from(Video).where(Video.enriched_at.is_(None)) + ) + or 0, + "videos_pending_shorts": db.scalar( + select(func.count()) + .select_from(Video) + .where(Video.shorts_probed.is_(False), Video.enriched_at.is_not(None)) + ) + or 0, + "videos_live_refresh": db.scalar( + select(func.count()) + .select_from(Video) + .where( + or_( + Video.live_status.in_(("live", "upcoming")), + and_(Video.live_status == "was_live", Video.duration_seconds.is_(None)), + ) + ) + ) + or 0, + } diff --git a/backend/app/catalog/search.py b/backend/app/catalog/search.py new file mode 100644 index 00000000..95ca4bc7 --- /dev/null +++ b/backend/app/catalog/search.py @@ -0,0 +1,232 @@ +"""Live YouTube search, the catalog's side: what a page of results leaves in the shared catalog. + +A result becomes a catalog row so it renders with the normal feed cards and the in-app player and +gains per-user state. The ingest runs in steps with a YouTube call between each, and each step's +rows are committed before the next call: an enrich that fails rolls back to the stubs, which stay +(the FK targets for what follows, backfilled later). Results are flagged (`videos.via_search`, +`channels.from_search`) so the Library can hide this search-discovered noise; Shorts and +live/upcoming videos are never kept. The HTTP layer (`routes/search.py`) runs the paging, the caps +and the refusals. +""" + +from collections.abc import Sequence +from concurrent.futures import ThreadPoolExecutor +from datetime import datetime, timezone + +from sqlalchemy import delete, func, select, update +from sqlalchemy.dialects.postgresql import insert as pg_insert +from sqlalchemy.engine import CursorResult +from sqlalchemy.orm import Session + +from app import logs, quota, sysconfig +from app.catalog import queries +from app.catalog.models import LIVE_OR_UPCOMING, Channel, SearchFind, Subscription, Video +from app.models import User +from app.reader import queries as reader_queries +from app.sync.subscriptions import apply_channel_details +from app.sync.videos import apply_video_details, insert_stubs, parse_dt +from app.userscope import owned_by +from app.utils import like_escape +from app.youtube.client import YouTubeClient, YouTubeError +from app.youtube.shorts import make_client, probe_is_short + +log = logs.get_logger(__name__) + + +def classify_shorts(db: Session, videos: Sequence[Video]) -> None: + """Confirm/deny Shorts for the freshly-ingested videos so none slip into the feed. + + Mirrors the scheduler's run_shorts_classification, but scoped to just these rows and run + synchronously (the search is an explicit, awaited action). Non-candidates (too long, no + duration, or live) are finalised without a probe; the rest hit youtube.com/shorts (no + quota). Undetermined probes are left for the scheduler to retry.""" + probe_max = sysconfig.get_int(db, "shorts_probe_max_seconds") + to_probe: list[Video] = [] + for v in videos: + if v.shorts_probed: + continue + if v.enriched_at is None: + # Enrichment failed (proxy blip) or YouTube omitted the row — duration is unknown, + # so DON'T finalise shorts_probed off a null duration; leave it for the scheduler + # (which only classifies enriched rows) to retry. Otherwise a real Short would leak. + continue + if v.live_status != "none" or v.duration_seconds is None or v.duration_seconds > probe_max: + v.is_short = False + v.shorts_probed = True + else: + to_probe.append(v) + if to_probe: + with make_client() as client: + + def work(v: Video): + return v, probe_is_short(client, v.id) + + with ThreadPoolExecutor(max_workers=16) as pool: + for v, result in pool.map(work, to_probe): + if result is None: + continue + v.is_short = result + v.shorts_probed = True + db.commit() + + +def ingest_candidates( + db: Session, yt: YouTubeClient, user: User, candidates: list[dict] +) -> list[str]: + """Materialise one page of search candidates into the catalog (channel stubs + enrich, + video stubs flagged via_search, enrich, Shorts classification) and return the KEEPABLE + video ids — non-Short, non-live — in the input relevance order.""" + ids = [it["id"] for it in candidates] + if not ids: + return [] + + # 1) Channels are the FK target — create stubs for unknown ones (flagged from_search), + # then enrich the new ones (channels.list, 1 unit). Stubs survive an enrich failure. + channel_ids = list({it["channel_id"] for it in candidates}) + existing = set(db.scalars(select(Channel.id).where(Channel.id.in_(channel_ids))).all()) + new_channel_ids = [c for c in channel_ids if c not in existing] + if new_channel_ids: + titles = {it["channel_id"]: it["channel_title"] for it in candidates} + for cid in new_channel_ids: + db.add(Channel(id=cid, title=titles.get(cid), from_search=True)) + db.commit() + try: + with quota.attribute(user.id, quota.QuotaAction.CHANNELS_DISCOVER): + details = yt.get_channels(new_channel_ids) + apply_channel_details(db, details, new_channel_ids) + db.commit() + except YouTubeError: + # The stubs remain and the details can be backfilled later — but until they are, those + # channels sit in the results with no thumbnail and no subscriber count, looking like + # channels that have neither. + log.warning( + "the channels found by a search could not be enriched", + extra=logs.event("search.channel_enrich_failed", channels=len(new_channel_ids)), + ) + db.rollback() # details can be backfilled later; the stubs remain + + # 2) Video stubs, flagged via_search. Existing rows keep their flags (ON CONFLICT DO + # NOTHING), so an organically-synced video stays organic. + # published_at is NOT NULL (migration 0068); the scrape source exposes no publish date + # (only a relative "3 years ago"), so fall back to now() to satisfy the constraint — the + # enrich in step 3 overwrites it with the video's real snippet date. + now = datetime.now(timezone.utc) + insert_stubs( + db, + [ + { + "id": it["id"], + "channel_id": it["channel_id"], + "title": it["title"], + "published_at": parse_dt(it["published_at"]) or now, + "thumbnail_url": it["thumbnail_url"], + "via_search": True, + } + for it in candidates + ], + ) + + # 3) Enrich (videos.list, 1 unit/50) so cards have duration/stats and we can judge live. + videos = db.scalars(select(Video).where(Video.id.in_(ids))).all() + try: + with quota.attribute(user.id, quota.QuotaAction.VIDEOS_ENRICH): + detail_by_id = {it["id"]: it for it in yt.get_videos(ids)} + except YouTubeError: + # Every candidate then stays unenriched, and an unenriched video cannot be judged live or + # Short — so it is dropped. A working search returns nothing and reads as "no results". + log.warning( + "the videos found by a search could not be enriched", + extra=logs.event("search.video_enrich_failed", candidates=len(ids)), + ) + detail_by_id = {} + enriched_at = datetime.now( + timezone.utc + ) # when THIS enrich ran (distinct from the stub placeholder) + for v in videos: + item = detail_by_id.get(v.id) + if item is None: + continue + apply_video_details(v, item) + v.enriched_at = enriched_at + db.commit() + + # 4) Confirm/deny Shorts (no quota) so none enter via search. + classify_shorts(db, videos) + + keep = { + v.id + for v in videos + if v.enriched_at is not None and not v.is_short and v.live_status not in LIVE_OR_UPCOMING + } + return [vid for vid in ids if vid in keep] + + +def _fold_term(db: Session, video_ids: list[str], term: str) -> None: + """Append `term` to each video's space-joined `search_terms`, unless it is already one of them. + + "Already one of them" is a whole-term match, as a literal: a substring test skipped `cat` on a + video already found for `category`, and an unescaped one read the `_` in `my_channel` as a + wildcard — either way the term was never folded in, and local search lost that relevance.""" + padded = func.concat(" ", func.coalesce(Video.search_terms, ""), " ") + db.execute( + update(Video) + .where(Video.id.in_(video_ids)) + .where(padded.notilike(f"% {like_escape(term)} %")) + .values( + search_terms=func.trim( + func.coalesce(Video.search_terms, "").op("||")(" ").op("||")(term) + ) + ) + ) + + +def new_first(db: Session, user: User, ids: list[str]) -> list[str]: + """Reorder results so genuinely-new discoveries come first and things the user already has + (a subscribed channel, or a video they've watched / in-progress / saved / playlisted) sink + to the back — keeping the relevance order within each group.""" + if not ids: + return ids + have = set( + db.scalars( + select(Video.id).where( + Video.id.in_(ids), Video.channel_id.in_(queries.subscribed_channels(user)) + ) + ) + ) | reader_queries.held_by(db, user, ids) + return [i for i in ids if i not in have] + [i for i in ids if i in have] + + +def record_finds(db: Session, user: User, video_ids: list[str], term: str) -> None: + """These results as THIS user's search finds (the Mine feed, the Source filter), and the term + folded into each one's `search_terms` so local full-text search inherits YouTube's relevance. + Idempotent.""" + db.execute( + pg_insert(SearchFind) + .values([{"user_id": user.id, "video_id": vid} for vid in video_ids]) + .on_conflict_do_nothing(index_elements=["user_id", "video_id"]) + ) + _fold_term(db, video_ids, term) + + +def clear(db: Session, user: User, video_ids: list[str]) -> int: + """Discard these results "as if never added": drop this user's finds of them, then delete the + ones nobody kept — a `via_search` video no reader holds (`reader.queries.kept_by_a_reader`), + whose channel nobody subscribes to, and that no other user still has as a find. Returns how + many videos went.""" + db.execute( + delete(SearchFind).where( + owned_by(SearchFind.user_id, user), SearchFind.video_id.in_(video_ids) + ) + ) + subscribed = select(Subscription.id).where(Subscription.channel_id == Video.channel_id).exists() + other_find = select(SearchFind.id).where(SearchFind.video_id == Video.id).exists() + result = db.execute( + delete(Video).where( + Video.id.in_(video_ids), + Video.via_search.is_(True), + ~reader_queries.kept_by_a_reader(Video.id), + ~subscribed, + ~other_find, + ) + ) + return result.rowcount if isinstance(result, CursorResult) else 0 diff --git a/backend/app/catalog/service.py b/backend/app/catalog/service.py new file mode 100644 index 00000000..1004a96e --- /dev/null +++ b/backend/app/catalog/service.py @@ -0,0 +1,133 @@ +"""Catalog's writes a request makes: a user's subscription settings, what they explore or block, a +subscription recorded or dropped, a channel's sync markers reset. + +Every function here flushes and leaves the commit to its caller (the handler). The YouTube side of a +subscribe or an unsubscribe is the handler's (it needs the user's client and its refusals); what +lands here is the local record of it. The ingest jobs (`sync/`, `catalog/search`) write the +catalog's shared rows themselves. +""" + +from datetime import datetime, timezone + +from sqlalchemy import delete, update +from sqlalchemy.engine import CursorResult +from sqlalchemy.orm import Session + +from app.catalog import queries +from app.catalog.models import BlockedChannel, Channel, ExploredChannel, Subscription, Video +from app.models import User +from app.userscope import owned_by + + +def remember_own_channel(user: User, channel_id: str) -> None: + """The user's own YouTube channel, so discovery never offers it to them.""" + user.yt_channel_id = channel_id + + +def update_subscription( + db: Session, + sub: Subscription, + *, + priority: int | None = None, + hidden: bool | None = None, + deep_requested: bool | None = None, +) -> bool: + """Change the settings given (None leaves one as it is). Returns whether full history was just + turned ON — the caller then pulls the channel now rather than at the scheduler's next tick.""" + turned_on = False + if priority is not None: + sub.priority = priority + if hidden is not None: + sub.hidden = hidden + if deep_requested is not None: + # Turning it off won't undo already-fetched videos; it stops further deep paging if nobody + # else still wants it. + turned_on = deep_requested and not sub.deep_requested + sub.deep_requested = deep_requested + db.flush() + return turned_on + + +def set_deep_all(db: Session, user: User, on: bool) -> int: + """Every one of the user's channels in (or out of) full-history backfill; how many rows moved.""" + result = db.execute( + update(Subscription).where(owned_by(Subscription.user_id, user)).values(deep_requested=on) + ) + return result.rowcount if isinstance(result, CursorResult) else 0 + + +def explore(db: Session, user: User, channel: Channel) -> None: + """Record (or refresh the grace clock of) the user's exploration of a channel, and flag the + channel ephemeral unless someone follows it — the cleanup job's keep rule reads both.""" + existing = queries.exploration(db, user, channel.id) + if existing is None: + db.add(ExploredChannel(user_id=user.id, channel_id=channel.id)) + else: + existing.created_at = datetime.now(timezone.utc) + if not channel.from_explore and not queries.followed(db, channel.id): + channel.from_explore = True + db.flush() + + +def reset_backfill(db: Session, channel: Channel) -> None: + """Fetch the channel again from scratch: its backfill markers cleared, and full history asked + for on EVERY subscription to it — the deep scheduler gates on an EXISTS over those, so the admin + who resets it need not subscribe themselves.""" + channel.backfill_done = False + channel.backfill_cursor = None + channel.recent_synced_at = None + db.execute( + update(Subscription) + .where(Subscription.channel_id == channel.id) + .values(deep_requested=True) + ) + db.flush() + + +def record_subscription(db: Session, user: User, channel: Channel, yt_subscription_id: str) -> None: + """The subscription YouTube now holds, recorded here. Subscribing is the "keep" signal: an + explored channel becomes an organic one, so the cleanup job leaves it alone and its videos join + the shared Library like any subscribed channel's uploads.""" + db.add( + Subscription( + user_id=user.id, + channel_id=channel.id, + yt_subscription_id=yt_subscription_id or None, + subscribed_at=datetime.now(timezone.utc), + ) + ) + if channel.from_explore: + channel.from_explore = False + db.execute( + update(Video) + .where(Video.channel_id == channel.id, Video.via_explore.is_(True)) + .values(via_explore=False) + ) + db.flush() + + +def drop_subscription(db: Session, sub: Subscription) -> None: + db.delete(sub) + db.flush() + + +def block(db: Session, user: User, channel_id: str) -> None: + """Block a channel for this user (idempotent), and end any exploration of it so the cleanup job + can reclaim it.""" + if queries.is_blocked(db, user, channel_id): + return + db.add(BlockedChannel(user_id=user.id, channel_id=channel_id)) + db.execute( + delete(ExploredChannel).where( + owned_by(ExploredChannel.user_id, user), ExploredChannel.channel_id == channel_id + ) + ) + db.flush() + + +def unblock(db: Session, user: User, channel_id: str) -> None: + db.execute( + delete(BlockedChannel).where( + owned_by(BlockedChannel.user_id, user), BlockedChannel.channel_id == channel_id + ) + ) diff --git a/backend/app/reader/queries.py b/backend/app/reader/queries.py index 80beaac5..453b7e88 100644 --- a/backend/app/reader/queries.py +++ b/backend/app/reader/queries.py @@ -10,9 +10,10 @@ 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.collation import collated +from app.feedquery import feed_columns, serialize from app.models import User -from app.catalog.models import Channel, Video +from app.catalog.models import Channel, Subscription, Video from app.reader.models import ChannelTag, Playlist, PlaylistItem, SavedView, Tag, VideoState from app.userscope import owned_by, owned_row @@ -269,3 +270,92 @@ def tags_in_feed_views(db: Session) -> set[int]: for (filters,) in db.execute(select(SavedView.filters).where(SavedView.source == "feed")) for t in view_tag_ids(filters) } + + +def held_by(db: Session, user: User, video_ids: list[str]) -> set[str]: + """Which of these videos the user already holds: watched, in progress or hidden, or in one of + their playlists (a live search sinks them below the new finds).""" + if not video_ids: + return set() + has_state = ( + select(VideoState.id) + .where(VideoState.video_id == Video.id, owned_by(VideoState.user_id, user)) + .exists() + ) + in_playlist = ( + select(PlaylistItem.id) + .join(Playlist, Playlist.id == PlaylistItem.playlist_id) + .where(PlaylistItem.video_id == Video.id, owned_by(Playlist.user_id, user)) + .exists() + ) + return set( + db.scalars(select(Video.id).where(Video.id.in_(video_ids), or_(has_state, in_playlist))) + ) + + +def feed_rows_by_ids(db: Session, user: User, video_ids: list[str]) -> list[dict]: + """These videos serialized with the feed's projection (their channel, this user's watch state, + saved), in the order given.""" + if not video_ids: + return [] + state = aliased(VideoState) + rows = db.execute( + select(*feed_columns(user, state)) + .join(Channel, Channel.id == Video.channel_id) + .outerjoin(state, and_(state.video_id == Video.id, owned_by(state.user_id, user))) + .where(Video.id.in_(video_ids)) + ).all() + by_id = {r.id: serialize(r) for r in rows} + return [by_id[i] for i in video_ids if i in by_id] + + +def playlist_channels(db: Session, user: User, exclude: str | None, lang: str) -> list: + """The channels of the videos in the user's playlists that they don't subscribe to (`exclude`: + their own channel), each with how many of those videos it has and across how many playlists; + most present first, then by title in the reader's alphabet.""" + subscribed = select(Subscription.channel_id).where(owned_by(Subscription.user_id, user)) + q = ( + select( + Channel, + func.count(func.distinct(PlaylistItem.video_id)), + func.count(func.distinct(Playlist.id)), + ) + .select_from(PlaylistItem) + .join(Playlist, Playlist.id == PlaylistItem.playlist_id) + .join(Video, Video.id == PlaylistItem.video_id) + .join(Channel, Channel.id == Video.channel_id) + .where(owned_by(Playlist.user_id, user), Channel.id.not_in(subscribed)) + ) + if exclude: + q = q.where(Channel.id != exclude) + return list( + db.execute( + q.group_by(Channel.id).order_by( + func.count(func.distinct(PlaylistItem.video_id)).desc(), + collated(Channel.title, lang), + ) + ).all() + ) + + +def user_tag_links(db: Session, user: User, channel_id: str | None = None) -> dict[str, list[int]]: + """The user's own tag links by channel (one channel's, when given).""" + stmt = select(ChannelTag.channel_id, ChannelTag.tag_id).where( + owned_by(ChannelTag.user_id, user) + ) + if channel_id is not None: + stmt = stmt.where(ChannelTag.channel_id == channel_id) + out: dict[str, list[int]] = {} + for cid, tag_id in db.execute(stmt).tuples(): + out.setdefault(cid, []).append(tag_id) + return out + + +def user_tag_link(db: Session, user: User, channel_id: str, tag_id: int) -> ChannelTag | None: + return db.execute( + select(ChannelTag).where( + ChannelTag.channel_id == channel_id, + ChannelTag.tag_id == tag_id, + owned_by(ChannelTag.user_id, user), + ) + ).scalar_one_or_none() diff --git a/backend/app/reader/service.py b/backend/app/reader/service.py index f07bb7e7..676e1862 100644 --- a/backend/app/reader/service.py +++ b/backend/app/reader/service.py @@ -464,3 +464,17 @@ def reorder_items(db: Session, pl: Playlist, order: list[str]) -> int: touch(pl) db.flush() return pos + + +def attach_user_tag(db: Session, user: User, channel_id: str, tag_id: int) -> None: + """The user's own label on a channel (idempotent). The caller has checked the tag is theirs.""" + if queries.user_tag_link(db, user, channel_id, tag_id) is None: + db.add(ChannelTag(channel_id=channel_id, tag_id=tag_id, user_id=user.id, source="user")) + db.flush() + + +def detach_user_tag(db: Session, user: User, channel_id: str, tag_id: int) -> None: + row = queries.user_tag_link(db, user, channel_id, tag_id) + if row is not None: + db.delete(row) + db.flush() diff --git a/backend/app/routes/channels.py b/backend/app/routes/channels.py index f681ab87..fc2a995d 100644 --- a/backend/app/routes/channels.py +++ b/backend/app/routes/channels.py @@ -4,28 +4,22 @@ hidden, tags) and per-channel sync state. Local reads/writes plus YouTube-side w from datetime import datetime, timezone from functools import partial +from typing import TYPE_CHECKING from fastapi import APIRouter, Depends from pydantic import BaseModel -from sqlalchemy import Select, func, select, update from sqlalchemy.orm import Session -from app.collation import collated, sorted_words +from app.collation import sorted_words from app.mailtext import lang_for_user from app import logs, quota, scheduler from app.auth import admin_user, current_user, require_write_scope from app.db import get_db from app.models import User -from app.catalog.models import ( - LIVE_OR_UPCOMING, - BlockedChannel, - Channel, - ExploredChannel, - Subscription, - Video, -) -from app.reader.models import ChannelTag, Playlist, PlaylistItem, Tag -from app.routes._common import get_or_404, not_found, owned_or_404, youtube_refusal +from app.catalog import queries, service +from app.reader import queries as reader_queries +from app.reader import service as reader_service +from app.routes._common import not_found, youtube_refusal from app.sync.explore import explore_ingest_page from app.sync.runner import backfill_channel_now, run_recent_backfill from app.sync.subscriptions import apply_channel_details, details_stale @@ -33,35 +27,17 @@ from app.utils import iso from app.youtube.client import YouTubeClient, YouTubeError from app.refusal import refused +if TYPE_CHECKING: + from app.catalog.models import Channel, Subscription + router = APIRouter(prefix="/api/channels", tags=["channels"]) log = logs.get_logger(__name__) -def channel_aggregates(channel_ids: list[str]) -> Select: - """Per-channel stored count, last upload and total duration, in one grouped pass. Every column it - reads is in `ix_videos_channel_pub_cover` (migration 0095), so Postgres answers it from the index - alone; a test holds that (test_channel_list_index_only).""" - return ( - select( - Video.channel_id, - func.count(Video.id), - func.max(Video.published_at), - func.coalesce(func.sum(Video.duration_seconds), 0), - ) - .where(Video.channel_id.in_(channel_ids)) - .group_by(Video.channel_id) - ) - - @router.get("") def list_channels(user: User = Depends(current_user), db: Session = Depends(get_db)) -> list[dict]: - rows = db.execute( - select(Subscription, Channel) - .join(Channel, Channel.id == Subscription.channel_id) - .where(Subscription.user_id == user.id) - .order_by(Subscription.priority.desc(), collated(Channel.title, lang_for_user(user))) - ).all() + rows = queries.subscriptions(db, user, lang_for_user(user)) # Per-channel aggregates over the shared catalog (channel_aggregates), computed on read rather than # denormalised: an Index Only Scan over a ~20 MB index, where it used to read every video's heap page @@ -70,36 +46,21 @@ def list_channels(user: User = Depends(current_user), db: Session = Depends(get_ # computed here — the manager table dropped its N/S/L column.) channel_ids = [c.id for _, c in rows] agg: dict[str, dict] = {} - if channel_ids: - for cid, n, last_up, total_dur in db.execute(channel_aggregates(channel_ids)).all(): - agg[cid] = { - "stored": n, - "last_video_at": iso(last_up), - "total_duration_seconds": int(total_dur or 0), - } + for cid, n, last_up, total_dur in queries.aggregates(db, channel_ids): + agg[cid] = { + "stored": n, + "last_video_at": iso(last_up), + "total_duration_seconds": int(total_dur or 0), + } # Channels already in the *global* deep-backfill queue — i.e. at least one user (any # user) has requested full history and it isn't done yet. Their whole back-catalogue is # coming for everyone (video data is shared), so the UI shouldn't offer "get full # history" to a user who simply hasn't personally opted in. - deep_queued: set[str] = set() - if channel_ids: - for (cid,) in db.execute( - select(Subscription.channel_id) - .where( - Subscription.channel_id.in_(channel_ids), - Subscription.deep_requested.is_(True), - ) - .distinct() - ).all(): - deep_queued.add(cid) + deep_queued = queries.deep_queued(db, channel_ids) # The user's personal tag links, grouped by channel. - tags_by_channel: dict[str, list[int]] = {} - for cid, tag_id in db.execute( - select(ChannelTag.channel_id, ChannelTag.tag_id).where(ChannelTag.user_id == user.id) - ).all(): - tags_by_channel.setdefault(cid, []).append(tag_id) + tags_by_channel = reader_queries.user_tag_links(db, user) return [ { @@ -128,27 +89,7 @@ DISCOVERY_ENRICH_CAP = 200 def _discovery_rows(db: Session, user: User) -> list: """The user's playlist channels they don't subscribe to (and that aren't their own), each with how many playlist videos come from it and across how many playlists.""" - subscribed = select(Subscription.channel_id).where(Subscription.user_id == user.id) - q = ( - select( - Channel, - func.count(func.distinct(PlaylistItem.video_id)), - func.count(func.distinct(Playlist.id)), - ) - .select_from(PlaylistItem) - .join(Playlist, Playlist.id == PlaylistItem.playlist_id) - .join(Video, Video.id == PlaylistItem.video_id) - .join(Channel, Channel.id == Video.channel_id) - .where(Playlist.user_id == user.id, Channel.id.not_in(subscribed)) - ) - if user.yt_channel_id: - q = q.where(Channel.id != user.yt_channel_id) - return db.execute( - q.group_by(Channel.id).order_by( - func.count(func.distinct(PlaylistItem.video_id)).desc(), - collated(Channel.title, lang_for_user(user)), - ) - ).all() + return reader_queries.playlist_channels(db, user, user.yt_channel_id, lang_for_user(user)) @router.get("/discovery") @@ -170,7 +111,7 @@ def discover_channels( ): own = yt.get_my_channel_id() if own: - user.yt_channel_id = own + service.remember_own_channel(user, own) db.commit() except YouTubeError as exc: # No traceback anywhere in this file's YouTube handlers: for an upstream answer @@ -226,7 +167,7 @@ def discover_channels( ] -def _channel_summary(ch: Channel) -> dict: +def _channel_summary(ch: "Channel") -> dict: """The channel fields common to the list and discovery projections.""" return { "id": ch.id, @@ -240,7 +181,7 @@ def _channel_summary(ch: Channel) -> dict: def _channel_detail_dict( - channel: Channel, + channel: "Channel", *, subscribed: bool, explored: bool, @@ -309,7 +250,7 @@ def channel_detail( needed (a stub) and refreshes it once it is stale (`DETAILS_MAX_AGE`), which is how a channel nobody subscribes to (an explored one, outside the subscription import) gets a banner that still resolves. The enrich uses the API key (1 unit, cached after).""" - channel = get_or_404(db, Channel, channel_id, what="channel") + channel = _channel_or_404(db, channel_id) # No `published_at is None` sentinel: a terminated channel never gets one, so it re-fetched on # every view; rows enriched before that field existed are long past DETAILS_MAX_AGE anyway. stale = details_stale(channel, datetime.now(timezone.utc)) @@ -321,7 +262,7 @@ def channel_detail( ): apply_channel_details(db, yt.get_channels([channel_id]), [channel_id]) db.commit() - channel = db.get(Channel, channel_id) + channel = _channel_or_404(db, channel_id) except YouTubeError as exc: log.warning( "channel detail enrich failed: %s", @@ -331,41 +272,15 @@ def channel_detail( ), ) db.rollback() - sub = db.execute( - select(Subscription).where( - Subscription.user_id == user.id, Subscription.channel_id == channel_id - ) - ).scalar_one_or_none() + sub = queries.subscription(db, user, channel_id) subscribed = sub is not None - explored = ( - db.execute( - select(ExploredChannel.id).where( - ExploredChannel.user_id == user.id, ExploredChannel.channel_id == channel_id - ) - ).first() - is not None - ) - blocked = _is_blocked(db, user, channel_id) + explored = queries.exploration(db, user, channel_id) is not None + blocked = queries.is_blocked(db, user, channel_id) # Stored total + the Normal/Shorts/Live split in one grouped pass (same filters as the manager # list), so the channel page can show the per-type breakdown that moved off the table. - n_all, n_short, n_live = db.execute( - select( - func.count(Video.id), - func.count(Video.id).filter(Video.is_short.is_(True)), - func.count(Video.id).filter(Video.live_status.in_(LIVE_OR_UPCOMING)), - ).where(Video.channel_id == channel_id) - ).one() - stored = int(n_all or 0) - count_short = int(n_short or 0) - count_live = int(n_live or 0) + stored, count_short, count_live = queries.video_counts(db, channel_id) count_normal = max(0, stored - count_short - count_live) - tag_ids = list( - db.scalars( - select(ChannelTag.tag_id).where( - ChannelTag.user_id == user.id, ChannelTag.channel_id == channel_id - ) - ).all() - ) + tag_ids = reader_queries.user_tag_links(db, user, channel_id).get(channel_id, []) return _channel_detail_dict( channel, subscribed=subscribed, @@ -395,28 +310,15 @@ def explore_channel( it, then ingests one page of its uploads (newest-first) marked via_explore + enriched. Returns the next page token so the page can "load more". Guarded against the shared backfill quota reserve.""" - channel = get_or_404(db, Channel, channel_id, what="channel") - if _is_blocked(db, user, channel_id): + channel = _channel_or_404(db, channel_id) + if queries.is_blocked(db, user, channel_id): raise refused(403, "channel_blocked") if quota.backfill_reserve_reached(): raise refused(429, "quota_reserve") # Record/refresh the exploration (DB only) before ingesting, so the explorer can see the # videos as soon as they land. Flag the channel ephemeral unless it's organically followed. - existing = db.execute( - select(ExploredChannel).where( - ExploredChannel.user_id == user.id, ExploredChannel.channel_id == channel_id - ) - ).scalar_one_or_none() - if existing is None: - db.add(ExploredChannel(user_id=user.id, channel_id=channel_id)) - else: - existing.created_at = datetime.now(timezone.utc) # refresh the grace clock - followed = db.execute( - select(Subscription.id).where(Subscription.channel_id == channel_id).limit(1) - ).first() - if followed is None and not channel.from_explore: - channel.from_explore = True + service.explore(db, user, channel) db.commit() try: @@ -430,7 +332,7 @@ def explore_channel( if details_stale(channel, datetime.now(timezone.utc)): apply_channel_details(db, yt.get_channels([channel_id]), [channel_id]) db.commit() - channel = db.get(Channel, channel_id) + channel = _channel_or_404(db, channel_id) result = explore_ingest_page(db, yt, channel, page_token) except YouTubeError as exc: raise youtube_refusal(exc) from exc @@ -442,30 +344,20 @@ def explore_channel( } -def _user_subscription(db: Session, user: User, channel_id: str) -> Subscription: - sub = db.execute( - select(Subscription).where( - Subscription.user_id == user.id, Subscription.channel_id == channel_id - ) - ).scalar_one_or_none() +def _channel_or_404(db: Session, channel_id: str) -> "Channel": + channel = queries.channel(db, channel_id) + if channel is None: + raise not_found("channel") + return channel + + +def _user_subscription(db: Session, user: User, channel_id: str) -> "Subscription": + sub = queries.subscription(db, user, channel_id) if sub is None: raise not_found("subscription") return sub -def _is_blocked(db: Session, user: User, channel_id: str) -> bool: - """Whether this user has blocked this channel.""" - return ( - db.execute( - select(BlockedChannel.id).where( - BlockedChannel.user_id == user.id, - BlockedChannel.channel_id == channel_id, - ) - ).first() - is not None - ) - - class ChannelUpdateIn(BaseModel): # PATCH semantics: a field left unset (None) is untouched. Every field is optional so the # UI can send just the one control it changed. int/bool coercion + rejection of garbage @@ -483,19 +375,14 @@ def update_channel( db: Session = Depends(get_db), ) -> dict: sub = _user_subscription(db, user, channel_id) - channel = db.get(Channel, channel_id) - deep_turned_on = False - if body.priority is not None: - sub.priority = body.priority - if body.hidden is not None: - sub.hidden = body.hidden - if body.deep_requested is not None: - # Opt this channel into (or out of) full-history backfill. The deep scheduler - # picks the flag up on its next run; turning it off won't undo already-fetched - # videos, it just stops further deep paging if nobody else still wants it. - want = body.deep_requested - deep_turned_on = want and not sub.deep_requested - sub.deep_requested = want + channel = queries.channel(db, channel_id) + deep_turned_on = service.update_subscription( + db, + sub, + priority=body.priority, + hidden=body.hidden, + deep_requested=body.deep_requested, + ) db.commit() # Opting a channel into full history should give an immediate feed: if we haven't even @@ -539,19 +426,8 @@ def reset_backfill( state (a "reset" trigger). Clears the channel's backfill markers, opts it back into deep backfill, and re-runs its recent pull immediately; the deep scheduler re-pages the full back-catalog on its next run. Idempotent — videos upsert by id, so nothing duplicates.""" - channel = get_or_404(db, Channel, channel_id, what="channel") - channel.backfill_done = False - channel.backfill_cursor = None - channel.recent_synced_at = None - # Re-arm the demand-driven deep scheduler for this channel. run_deep_backfill gates on an - # EXISTS over subscriptions with deep_requested=True, so flip it on EVERY subscriber rather - # than requiring the admin to be personally subscribed (they reset from the manager, not - # their own feed). No-op row count if the channel has no subscribers. - db.execute( - update(Subscription) - .where(Subscription.channel_id == channel_id) - .values(deep_requested=True) - ) + channel = _channel_or_404(db, channel_id) + service.reset_backfill(db, channel) db.commit() with quota.attribute(user.id, quota.QuotaAction.VIDEOS_BACKFILL_RECENT): run_recent_backfill(db, [channel], max_channels=1) @@ -569,13 +445,8 @@ def subscribe( local subscription with YouTube's returned resource id and enriches the channel's metadata if it's still a stub, so the new row renders properly in the manager. It does NOT pull the channel's videos — the scheduler picks those up on its next run, like any subscribed channel.""" - channel = get_or_404(db, Channel, channel_id, what="channel") - existing = db.execute( - select(Subscription).where( - Subscription.user_id == user.id, Subscription.channel_id == channel_id - ) - ).scalar_one_or_none() - if existing is not None: + channel = _channel_or_404(db, channel_id) + if queries.subscription(db, user, channel_id) is not None: raise refused(409, "already_subscribed") with ( quota.attribute(user.id, quota.QuotaAction.CHANNELS_SUBSCRIBE), @@ -615,24 +486,7 @@ def subscribe( reason=exc.reason, ), ) - db.add( - Subscription( - user_id=user.id, - channel_id=channel_id, - yt_subscription_id=yt_sub_id or None, - subscribed_at=datetime.now(timezone.utc), - ) - ) - # Subscribing is the "keep" signal: a previously-explored channel becomes a normal organic - # one. Drop its ephemeral flags so the cleanup job leaves it alone and its videos stop being - # per-user private (they join the shared Library like any subscribed channel's uploads). - if channel.from_explore: - channel.from_explore = False - db.execute( - update(Video) - .where(Video.channel_id == channel_id, Video.via_explore.is_(True)) - .values(via_explore=False) - ) + service.record_subscription(db, user, channel, yt_sub_id) db.commit() return {"ok": True, "subscribed": channel_id} @@ -659,7 +513,7 @@ def unsubscribe( # 422 (a real, explainable error → speaking modal), not 502 which the client treats as # a transient "connection lost" gateway blip. raise youtube_refusal(exc) from exc - db.delete(sub) + service.drop_subscription(db, sub) db.commit() return {"ok": True, "unsubscribed": channel_id} @@ -667,16 +521,7 @@ def unsubscribe( @router.get("/blocked/list") def list_blocked(user: User = Depends(current_user), db: Session = Depends(get_db)) -> list[dict]: """The user's blocked channels (videos from these are excluded from their search/feed/explore).""" - rows = ( - db.execute( - select(Channel) - .join(BlockedChannel, BlockedChannel.channel_id == Channel.id) - .where(BlockedChannel.user_id == user.id) - .order_by(collated(Channel.title, lang_for_user(user))) - ) - .scalars() - .all() - ) + rows = queries.blocked_channels(db, user, lang_for_user(user)) return [ {"id": c.id, "title": c.title, "handle": c.handle, "thumbnail_url": c.thumbnail_url} for c in rows @@ -692,16 +537,9 @@ def block_channel( """Block a channel for this user: its videos won't appear in their live search (and won't be ingested for them) and are hidden from their feed / explore. Idempotent. The now-hidden un-kept search/explore videos are reclaimed by the discovery-cleanup job in due course.""" - get_or_404(db, Channel, channel_id, what="channel") - if not _is_blocked(db, user, channel_id): - db.add(BlockedChannel(user_id=user.id, channel_id=channel_id)) - # Stop any active exploration of it so it can be cleaned up. - db.execute( - ExploredChannel.__table__.delete().where( - (ExploredChannel.user_id == user.id) & (ExploredChannel.channel_id == channel_id) - ) - ) - db.commit() + _channel_or_404(db, channel_id) + service.block(db, user, channel_id) + db.commit() return {"ok": True, "blocked": channel_id} @@ -711,11 +549,7 @@ def unblock_channel( user: User = Depends(current_user), db: Session = Depends(get_db), ) -> dict: - db.execute( - BlockedChannel.__table__.delete().where( - (BlockedChannel.user_id == user.id) & (BlockedChannel.channel_id == channel_id) - ) - ) + service.unblock(db, user, channel_id) db.commit() return {"ok": True, "unblocked": channel_id} @@ -732,26 +566,12 @@ def attach_tag( db: Session = Depends(get_db), ) -> dict: _user_subscription(db, user, channel_id) - # Only the user's own tags may be attached as personal links (owned_or_404 → 404 if the tag - # is missing OR belongs to someone else, without leaking which). - owned_or_404(db, Tag, body.tag_id, user, what="tag") - exists = db.execute( - select(ChannelTag).where( - ChannelTag.channel_id == channel_id, - ChannelTag.tag_id == body.tag_id, - ChannelTag.user_id == user.id, - ) - ).scalar_one_or_none() - if exists is None: - db.add( - ChannelTag( - channel_id=channel_id, - tag_id=body.tag_id, - user_id=user.id, - source="user", - ) - ) - db.commit() + # Only the user's own tags may be attached as personal links (a 404 if the tag is missing OR + # belongs to someone else, without leaking which). + if reader_queries.own_tag(db, user, body.tag_id) is None: + raise not_found("tag") + reader_service.attach_user_tag(db, user, channel_id, body.tag_id) + db.commit() return {"ok": True, "channel_id": channel_id, "tag_id": body.tag_id} @@ -762,14 +582,6 @@ def detach_tag( user: User = Depends(current_user), db: Session = Depends(get_db), ) -> dict: - row = db.execute( - select(ChannelTag).where( - ChannelTag.channel_id == channel_id, - ChannelTag.tag_id == tag_id, - ChannelTag.user_id == user.id, - ) - ).scalar_one_or_none() - if row is not None: - db.delete(row) - db.commit() + reader_service.detach_user_tag(db, user, channel_id, tag_id) + db.commit() return {"ok": True, "channel_id": channel_id, "tag_id": tag_id} diff --git a/backend/app/routes/runs.py b/backend/app/routes/runs.py index 466a38be..2672d802 100644 --- a/backend/app/routes/runs.py +++ b/backend/app/routes/runs.py @@ -19,7 +19,7 @@ from app import scheduler from app.auth import admin_user from app.db import get_db from app.models import JobRun, LogEvent, User -from app.catalog.models import Channel +from app.catalog import queries as catalog_queries from app.routes import _account as account from app.routes._keyset import PAGE_MAX, accounts, older_than, page, utc from app.runstatus import RunStatus @@ -117,10 +117,7 @@ def list_runs( lambda r: r.started_at, ) channels = {r.target for r in runs if r.job in scheduler.AD_HOC_JOBS and r.target} - titles: dict[str, str] = {} - if channels: - rows = db.execute(select(Channel.id, Channel.title).where(Channel.id.in_(channels))) - titles = {cid: title for cid, title in rows.tuples() if title is not None} + titles = catalog_queries.channel_titles(db, channels) lines: dict[int, int] = {} if runs: counted = db.execute( diff --git a/backend/app/routes/scheduler.py b/backend/app/routes/scheduler.py index 76172806..f09b4b45 100644 --- a/backend/app/routes/scheduler.py +++ b/backend/app/routes/scheduler.py @@ -4,7 +4,6 @@ and the shared quota picture.""" from fastapi import APIRouter, Depends from pydantic import BaseModel -from sqlalchemy import and_, func, or_, select from sqlalchemy.orm import Session from app import audit, quota, state, sysconfig @@ -13,7 +12,7 @@ from app.auth import admin_user from app.config import settings from app.db import get_db from app.models import User -from app.catalog.models import Channel, Subscription, Video +from app.catalog import queries as catalog_queries from app.scheduler import ( JOB_INTERVALS, MAX_INTERVAL, @@ -24,7 +23,6 @@ from app.scheduler import ( trigger_all, trigger_job, ) -from app.sync.explore import followed_channel from app.sync.runner import estimate_deep_backfill from app.routes._common import not_found from app.refusal import refused @@ -123,58 +121,14 @@ def get_scheduler(_: User = Depends(admin_user), db: Session = Depends(get_db)) snap = scheduler_snapshot(db) # Work still queued (shared catalog, so these are instance-wide, not per-user). - deep_requested = ( - select(Subscription.id) - .where( - Subscription.channel_id == Channel.id, - Subscription.deep_requested.is_(True), - ) - .exists() - ) - deep_channels = ( - db.execute(select(Channel).where(Channel.backfill_done.is_(False), deep_requested)) - .scalars() - .all() - ) - deep_eta = estimate_deep_backfill(db, deep_channels) - - queue = { - "channels_recent_pending": db.scalar( - # The same selection run_recent_backfill works through, or the card counts channels - # the job will never touch and never reaches zero. - select(func.count()) - .select_from(Channel) - .where(Channel.recent_synced_at.is_(None), followed_channel()) - ) - or 0, - # The whole opted-in BACKLOG, deliberately not `run_deep_backfill`'s next-pass selection - # (which also wants `recent_synced_at`): the same figure /api/sync/my-status shows the user, - # and a channel missing its recent pull is one the same job tick fetches first, not one the - # job will never touch. Narrowing it here would under-report work that is genuinely queued. - "channels_deep_pending": deep_eta["channels_pending"], - "deep_eta_seconds": deep_eta["eta_seconds"], - "videos_pending_enrich": db.scalar( - select(func.count()).select_from(Video).where(Video.enriched_at.is_(None)) - ) - or 0, - "videos_pending_shorts": db.scalar( - select(func.count()) - .select_from(Video) - .where(Video.shorts_probed.is_(False), Video.enriched_at.is_not(None)) - ) - or 0, - "videos_live_refresh": db.scalar( - select(func.count()) - .select_from(Video) - .where( - or_( - Video.live_status.in_(("live", "upcoming")), - and_(Video.live_status == "was_live", Video.duration_seconds.is_(None)), - ) - ) - ) - or 0, - } + deep_eta = estimate_deep_backfill(db, catalog_queries.deep_backlog(db)) + queue = catalog_queries.work_queue(db) + # The whole opted-in BACKLOG, deliberately not `run_deep_backfill`'s next-pass selection + # (which also wants `recent_synced_at`): the same figure /api/sync/my-status shows the user, + # and a channel missing its recent pull is one the same job tick fetches first, not one the + # job will never touch. Narrowing it here would under-report work that is genuinely queued. + queue["channels_deep_pending"] = deep_eta["channels_pending"] + queue["deep_eta_seconds"] = deep_eta["eta_seconds"] used = quota.units_used_today(db) quota_info = { diff --git a/backend/app/routes/search.py b/backend/app/routes/search.py index 33d3188b..9271ea65 100644 --- a/backend/app/routes/search.py +++ b/backend/app/routes/search.py @@ -11,37 +11,24 @@ regardless of source. Ingested results are flagged (videos.via_search / channels.from_search) so the Library can hide this search-discovered "noise" by default. Shorts and currently-live/upcoming videos are never ingested this way (Shorts are confirmed by the youtube.com/shorts probe; live/upcoming are -dropped from the search response up front).""" +dropped from the search response up front). -from concurrent.futures import ThreadPoolExecutor -from datetime import datetime, timezone +This is the HTTP side — the paging, the caps and the refusals; what a page of results leaves in the +catalog is `catalog/search`.""" from fastapi import APIRouter, Depends from pydantic import BaseModel -from sqlalchemy import and_, delete, func, or_, select, update -from sqlalchemy.dialects.postgresql import insert as pg_insert -from sqlalchemy.orm import Session, aliased +from sqlalchemy.orm import Session from app import logs, quota, sysconfig from app.auth import current_user +from app.catalog import queries +from app.catalog import search as catalog_search from app.db import get_db from app.models import User -from app.catalog.models import ( - LIVE_OR_UPCOMING, - BlockedChannel, - Channel, - SearchFind, - Subscription, - Video, -) -from app.reader.models import Playlist, PlaylistItem, VideoState -from app.feedquery import feed_columns, serialize -from app.utils import like_escape -from app.sync.subscriptions import apply_channel_details -from app.sync.videos import insert_stubs, apply_video_details, parse_dt +from app.reader import queries as reader_queries from app.youtube import search_scrape from app.youtube.client import YouTubeClient, YouTubeError -from app.youtube.shorts import make_client, probe_is_short from app.refusal import refused from app.routes._common import youtube_refusal @@ -52,198 +39,6 @@ router = APIRouter(prefix="/api/search", tags=["search"]) SEARCH_COST = 100 -def _classify_shorts(db: Session, videos: list[Video]) -> None: - """Confirm/deny Shorts for the freshly-ingested videos so none slip into the feed. - - Mirrors the scheduler's run_shorts_classification, but scoped to just these rows and run - synchronously (the search is an explicit, awaited action). Non-candidates (too long, no - duration, or live) are finalised without a probe; the rest hit youtube.com/shorts (no - quota). Undetermined probes are left for the scheduler to retry.""" - probe_max = sysconfig.get_int(db, "shorts_probe_max_seconds") - to_probe: list[Video] = [] - for v in videos: - if v.shorts_probed: - continue - if v.enriched_at is None: - # Enrichment failed (proxy blip) or YouTube omitted the row — duration is unknown, - # so DON'T finalise shorts_probed off a null duration; leave it for the scheduler - # (which only classifies enriched rows) to retry. Otherwise a real Short would leak. - continue - if v.live_status != "none" or v.duration_seconds is None or v.duration_seconds > probe_max: - v.is_short = False - v.shorts_probed = True - else: - to_probe.append(v) - if to_probe: - with make_client() as client: - - def work(v: Video): - return v, probe_is_short(client, v.id) - - with ThreadPoolExecutor(max_workers=16) as pool: - for v, result in pool.map(work, to_probe): - if result is None: - continue - v.is_short = result - v.shorts_probed = True - db.commit() - - -def _serialize_ids(db: Session, user: User, ids: list[str]) -> list[dict]: - """Serialize the given video ids with the shared feed projection (joined channel + this - user's watch state + saved), preserving the input (search relevance) order.""" - if not ids: - return [] - state = aliased(VideoState) - rows = db.execute( - select(*feed_columns(user, state)) - .join(Channel, Channel.id == Video.channel_id) - .outerjoin(state, and_(state.video_id == Video.id, state.user_id == user.id)) - .where(Video.id.in_(ids)) - ).all() - by_id = {r.id: serialize(r) for r in rows} - return [by_id[i] for i in ids if i in by_id] - - -def _ingest_candidates( - db: Session, yt: YouTubeClient, user: User, candidates: list[dict] -) -> list[str]: - """Materialise one page of search candidates into the catalog (channel stubs + enrich, - video stubs flagged via_search, enrich, Shorts classification) and return the KEEPABLE - video ids — non-Short, non-live — in the input relevance order.""" - ids = [it["id"] for it in candidates] - if not ids: - return [] - - # 1) Channels are the FK target — create stubs for unknown ones (flagged from_search), - # then enrich the new ones (channels.list, 1 unit). Stubs survive an enrich failure. - channel_ids = list({it["channel_id"] for it in candidates}) - existing = set(db.scalars(select(Channel.id).where(Channel.id.in_(channel_ids))).all()) - new_channel_ids = [c for c in channel_ids if c not in existing] - if new_channel_ids: - titles = {it["channel_id"]: it["channel_title"] for it in candidates} - for cid in new_channel_ids: - db.add(Channel(id=cid, title=titles.get(cid), from_search=True)) - db.commit() - try: - with quota.attribute(user.id, quota.QuotaAction.CHANNELS_DISCOVER): - details = yt.get_channels(new_channel_ids) - apply_channel_details(db, details, new_channel_ids) - db.commit() - except YouTubeError: - # The stubs remain and the details can be backfilled later — but until they are, those - # channels sit in the results with no thumbnail and no subscriber count, looking like - # channels that have neither. - log.warning( - "the channels found by a search could not be enriched", - extra=logs.event("search.channel_enrich_failed", channels=len(new_channel_ids)), - ) - db.rollback() # details can be backfilled later; the stubs remain - - # 2) Video stubs, flagged via_search. Existing rows keep their flags (ON CONFLICT DO - # NOTHING), so an organically-synced video stays organic. - # published_at is NOT NULL (migration 0068); the scrape source exposes no publish date - # (only a relative "3 years ago"), so fall back to now() to satisfy the constraint — the - # enrich in step 3 overwrites it with the video's real snippet date. - now = datetime.now(timezone.utc) - insert_stubs( - db, - [ - { - "id": it["id"], - "channel_id": it["channel_id"], - "title": it["title"], - "published_at": parse_dt(it["published_at"]) or now, - "thumbnail_url": it["thumbnail_url"], - "via_search": True, - } - for it in candidates - ], - ) - - # 3) Enrich (videos.list, 1 unit/50) so cards have duration/stats and we can judge live. - videos = db.scalars(select(Video).where(Video.id.in_(ids))).all() - try: - with quota.attribute(user.id, quota.QuotaAction.VIDEOS_ENRICH): - detail_by_id = {it["id"]: it for it in yt.get_videos(ids)} - except YouTubeError: - # Every candidate then stays unenriched, and an unenriched video cannot be judged live or - # Short — so it is dropped. A working search returns nothing and reads as "no results". - log.warning( - "the videos found by a search could not be enriched", - extra=logs.event("search.video_enrich_failed", candidates=len(ids)), - ) - detail_by_id = {} - enriched_at = datetime.now( - timezone.utc - ) # when THIS enrich ran (distinct from the stub placeholder) - for v in videos: - item = detail_by_id.get(v.id) - if item is None: - continue - apply_video_details(v, item) - v.enriched_at = enriched_at - db.commit() - - # 4) Confirm/deny Shorts (no quota) so none enter via search. - _classify_shorts(db, videos) - - keep = { - v.id - for v in videos - if v.enriched_at is not None and not v.is_short and v.live_status not in LIVE_OR_UPCOMING - } - return [vid for vid in ids if vid in keep] - - -def _new_first(db: Session, user: User, ids: list[str]) -> list[str]: - """Reorder results so genuinely-new discoveries come first and things the user already has - (a subscribed channel, or a video they've watched / in-progress / saved / playlisted) sink - to the back — keeping the relevance order within each group.""" - if not ids: - return ids - sub_ch = select(Subscription.channel_id).where(Subscription.user_id == user.id) - has_state = ( - select(VideoState.id) - .where(VideoState.video_id == Video.id, VideoState.user_id == user.id) - .exists() - ) - in_playlist = ( - select(PlaylistItem.id) - .join(Playlist, Playlist.id == PlaylistItem.playlist_id) - .where(PlaylistItem.video_id == Video.id, Playlist.user_id == user.id) - .exists() - ) - have = set( - db.scalars( - select(Video.id).where( - Video.id.in_(ids), - or_(Video.channel_id.in_(sub_ch), has_state, in_playlist), - ) - ).all() - ) - return [i for i in ids if i not in have] + [i for i in ids if i in have] - - -def _fold_term(db: Session, video_ids: list[str], term: str) -> None: - """Append `term` to each video's space-joined `search_terms`, unless it is already one of them. - - "Already one of them" is a whole-term match, as a literal: a substring test skipped `cat` on a - video already found for `category`, and an unescaped one read the `_` in `my_channel` as a - wildcard — either way the term was never folded in, and local search lost that relevance.""" - padded = func.concat(" ", func.coalesce(Video.search_terms, ""), " ") - db.execute( - update(Video) - .where(Video.id.in_(video_ids)) - .where(padded.notilike(f"% {like_escape(term)} %")) - .values( - search_terms=func.trim( - func.coalesce(Video.search_terms, "").op("||")(" ").op("||")(term) - ) - ) - ) - - @router.get("/youtube") def search_youtube( q: str | None = None, @@ -285,9 +80,7 @@ def search_youtube( raise refused(429, "quota_spent") # Channels this user has blocked are dropped before ingestion — never shown, never stored. - blocked = set( - db.scalars(select(BlockedChannel.channel_id).where(BlockedChannel.user_id == user.id)).all() - ) + blocked = queries.blocked_ids(db, user) collected: list[str] = [] next_cursor = cursor @@ -328,7 +121,7 @@ def search_youtube( # write/shape failure degrades to a 422 with a message on this JSON endpoint instead of a # bare 500 hidden behind the FE's generic "search failed" (same rationale as B-core-1). try: - collected.extend(_ingest_candidates(db, yt, user, candidates)) + collected.extend(catalog_search.ingest_candidates(db, yt, user, candidates)) except Exception as exc: # any ingest failure, not just YouTubeError db.rollback() # The term is NOT in this line. A search string is exactly what the privacy @@ -376,17 +169,12 @@ def search_youtube( # Record these as THIS user's search finds (Mine feed / Source filter), and fold the # query into each result's search_terms so local full-text search inherits YouTube's # relevance. Idempotent; the generated search_vector updates automatically. - db.execute( - pg_insert(SearchFind) - .values([{"user_id": user.id, "video_id": vid} for vid in ordered]) - .on_conflict_do_nothing(index_elements=["user_id", "video_id"]) - ) - _fold_term(db, ordered, term) + catalog_search.record_finds(db, user, ordered, term) db.commit() - ordered = _new_first(db, user, ordered) + ordered = catalog_search.new_first(db, user, ordered) return { - "items": _serialize_ids(db, user, ordered), + "items": reader_queries.feed_rows_by_ids(db, user, ordered), "next_cursor": next_cursor, "has_more": bool(next_cursor), "source": source, @@ -410,23 +198,6 @@ def clear_search( ids = body.video_ids if not ids: return {"ok": True, "removed": 0} - db.execute( - delete(SearchFind).where(SearchFind.user_id == user.id, SearchFind.video_id.in_(ids)) - ) - db.commit() - has_state = select(VideoState.id).where(VideoState.video_id == Video.id).exists() - in_playlist = select(PlaylistItem.id).where(PlaylistItem.video_id == Video.id).exists() - subscribed = select(Subscription.id).where(Subscription.channel_id == Video.channel_id).exists() - other_find = select(SearchFind.id).where(SearchFind.video_id == Video.id).exists() - removed = db.execute( - delete(Video).where( - Video.id.in_(ids), - Video.via_search.is_(True), - ~has_state, - ~in_playlist, - ~subscribed, - ~other_find, - ) - ).rowcount + removed = catalog_search.clear(db, user, ids) db.commit() return {"ok": True, "removed": removed} diff --git a/backend/app/routes/sync.py b/backend/app/routes/sync.py index 64bc0762..b8cca834 100644 --- a/backend/app/routes/sync.py +++ b/backend/app/routes/sync.py @@ -1,5 +1,4 @@ from fastapi import APIRouter, Depends, HTTPException -from sqlalchemy import and_, case, func, select, update from sqlalchemy.orm import Session from app import audit, quota, state @@ -7,7 +6,7 @@ from app.audit import AuditAction from app.auth import admin_user, current_user, has_read_scope from app.db import get_db from app.models import User -from app.catalog.models import Channel, Subscription, Video +from app.catalog import queries, service from app.routes._common import youtube_refusal from app.ratelimit import RateLimiter from app.sync.runner import estimate_deep_backfill @@ -48,19 +47,8 @@ def sync_subscriptions(user: User = Depends(current_user), db: Session = Depends @router.get("/status") def sync_status(user: User = Depends(current_user), db: Session = Depends(get_db)) -> dict: - sub_count = db.scalar( - select(func.count()).select_from(Subscription).where(Subscription.user_id == user.id) - ) return { - "subscriptions": sub_count, - "channels_total": db.scalar(select(func.count()).select_from(Channel)), - "channels_backfilling": db.scalar( - select(func.count()).select_from(Channel).where(Channel.backfill_done.is_(False)) - ), - "videos_total": db.scalar(select(func.count()).select_from(Video)), - "pending_enrich": db.scalar( - select(func.count()).select_from(Video).where(Video.enriched_at.is_(None)) - ), + **queries.catalog_totals(db, user), "quota_used_today": quota.units_used_today(db), "quota_remaining_today": quota.remaining_today(db), "paused": state.is_sync_paused(db), @@ -71,51 +59,21 @@ def sync_status(user: User = Depends(current_user), db: Session = Depends(get_db @router.get("/my-status") def my_status(user: User = Depends(current_user), db: Session = Depends(get_db)) -> dict: """Per-user sync progress: how far the channels *this* user subscribes to have synced.""" - sub_join = and_(Subscription.channel_id == Channel.id, Subscription.user_id == user.id) - total, details_synced, recent_synced, all_deep_done = db.execute( - select( - func.count(Channel.id), - func.count(Channel.details_synced_at), - func.count(Channel.recent_synced_at), - func.sum(case((Channel.backfill_done.is_(True), 1), else_=0)), - ) - .select_from(Channel) - .join(Subscription, sub_join) - ).one() - total = total or 0 - all_deep_done = int(all_deep_done or 0) + total, details_synced, recent_synced, all_deep_done = queries.my_progress(db, user) # Per-user full-history (deep) backfill: channels this user has opted in, and an ETA # for the still-pending ones (quota-bound estimate). - deep_requested_channels = ( - db.execute( - select(Channel).join( - Subscription, - and_(sub_join, Subscription.deep_requested.is_(True)), - ) - ) - .scalars() - .all() - ) + deep_requested_channels = queries.my_deep_requested(db, user) deep_eta = estimate_deep_backfill(db, deep_requested_channels) # "done / requested" is shown as one ratio, so both halves count the SAME channels: the ones this # user asked full history for. Counting every subscribed channel whose back-catalog happens to be # complete (another user asked, or it was short) gave "342/329". deep_done = sum(1 for c in deep_requested_channels if c.backfill_done) - my_videos = db.scalar( - select(func.count(Video.id)).join( - Subscription, - and_( - Subscription.channel_id == Video.channel_id, - Subscription.user_id == user.id, - Subscription.hidden.is_(False), - ), - ) - ) + my_videos = queries.my_video_count(db, user) # Whole-catalog total (every video any user's subscriptions have pulled in), so the # header can show "yours" against the shared library size. - total_videos = db.scalar(select(func.count(Video.id))) + total_videos = queries.video_total(db) return { "channels_total": total, "channels_details_synced": details_synced, @@ -126,8 +84,8 @@ def my_status(user: User = Depends(current_user), db: Session = Depends(get_db)) "channels_deep_requested": len(deep_requested_channels), "deep_pending_count": deep_eta["channels_pending"], "deep_eta_seconds": deep_eta["eta_seconds"], - "my_videos": my_videos or 0, - "total_videos": total_videos or 0, + "my_videos": my_videos, + "total_videos": total_videos, "quota_used_today": quota.units_used_today(db), "quota_remaining_today": quota.remaining_today(db), "paused": state.is_sync_paused(db), @@ -144,11 +102,9 @@ def deep_all( on: bool = True, ) -> dict: """Opt every one of the user's channels into (or out of) full-history backfill.""" - result = db.execute( - update(Subscription).where(Subscription.user_id == user.id).values(deep_requested=on) - ) + updated = service.set_deep_all(db, user, on) db.commit() - return {"ok": True, "updated": result.rowcount, "deep_requested": on} + return {"ok": True, "updated": updated, "deep_requested": on} @router.post("/pause") diff --git a/backend/app/scheduler.py b/backend/app/scheduler.py index 57e1147b..c2a3ae09 100644 --- a/backend/app/scheduler.py +++ b/backend/app/scheduler.py @@ -9,14 +9,13 @@ from typing import Callable from apscheduler.executors.pool import ThreadPoolExecutor from apscheduler.jobstores.base import ConflictingIdError from apscheduler.schedulers.background import BackgroundScheduler -from sqlalchemy import select from app import audit, debugsession, jobruns, logs, logsink, progress, quota, sysconfig from app.config import settings from app.db import SessionLocal from app.downloads.gc import run_download_gc from app.models import JobRun -from app.catalog.models import Channel +from app.catalog import queries as catalog_queries from app.plex.sync import sync as run_plex_sync from app.plex.tmdb import run_enrich as run_tmdb_enrich from app.plex.tmdb import run_people_enrich @@ -306,7 +305,7 @@ def scheduler_snapshot(db) -> dict: "running": going is not None, "target": shown.target if shown else None, "target_title": ( - db.scalar(select(Channel.title).where(Channel.id == shown.target)) + catalog_queries.channel_title(db, shown.target) if shown is not None and shown.target else None ), diff --git a/backend/app/sync/subscriptions.py b/backend/app/sync/subscriptions.py index 79a66364..301cd019 100644 --- a/backend/app/sync/subscriptions.py +++ b/backend/app/sync/subscriptions.py @@ -11,6 +11,7 @@ from app.models import User from app.catalog.models import Channel, Subscription from app.utils import normalize_description from app.youtube.client import YouTubeClient, YouTubeError, best_thumbnail +from app.userscope import owned_by log = logs.get_logger(__name__) @@ -173,7 +174,7 @@ def import_subscriptions(db: Session, user: User) -> dict: new_channels += 1 sub_row = db.execute( select(Subscription).where( - Subscription.user_id == user.id, Subscription.channel_id == cid + owned_by(Subscription.user_id, user.id), Subscription.channel_id == cid ) ).scalar_one_or_none() if sub_row is None: @@ -186,7 +187,9 @@ def import_subscriptions(db: Session, user: User) -> dict: # (see `should_prune`). removed = 0 existing = ( - db.execute(select(Subscription).where(Subscription.user_id == user.id)).scalars().all() + db.execute(select(Subscription).where(owned_by(Subscription.user_id, user.id))) + .scalars() + .all() ) prune_skipped = not should_prune(len(fetched), len(existing)) if prune_skipped: diff --git a/backend/tests/test_catalog_service.py b/backend/tests/test_catalog_service.py new file mode 100644 index 00000000..c01d0e9b --- /dev/null +++ b/backend/tests/test_catalog_service.py @@ -0,0 +1,295 @@ +"""Catalog's service and queries, one use case each, below the HTTP layer. + +The routes are thin callers now, and their own tests stay the contract a client sees. These hold +what each write does and leaves on its own: that it flushes and leaves the commit to its caller, +that a row of someone else's reads as a missing one, and the rule each write carries (an explored +channel flagged only while nobody follows it, a subscription that keeps it, a cleared search that +spares what anyone holds). DB-backed (real Postgres). +""" + +from datetime import datetime, timedelta, timezone + +from sqlalchemy import select + +from app.catalog import queries, search, service +from app.catalog.models import ( + BlockedChannel, + Channel, + ExploredChannel, + SearchFind, + Subscription, + Video, +) +from app.reader import queries as reader_queries +from app.reader import service as reader_service + + +def _channel(db, cid="UCc", **kw): + ch = Channel(id=cid, title=kw.pop("title", cid), **kw) + db.add(ch) + db.commit() + return ch + + +def _video(db, vid, cid="UCc", **kw): + v = Video(id=vid, channel_id=cid, published_at=datetime.now(timezone.utc), **kw) + db.add(v) + db.commit() + return v + + +def _sub(db, user, cid="UCc", **kw): + s = Subscription(user_id=user.id, channel_id=cid, **kw) + db.add(s) + db.commit() + return s + + +def _uncommitted(db): + return bool(db.new or db.dirty or db.deleted) or db.in_transaction() + + +# --- whose a row is ------------------------------------------------------------------------ + + +def test_a_users_subscriptions_blocks_and_explorations_are_theirs_alone(db, make_user): + owner, stranger = make_user(), make_user() + ch = _channel(db) + _sub(db, owner) + service.explore(db, owner, ch) + service.block(db, owner, "UCc") + db.commit() + assert queries.subscription(db, owner, "UCc") is not None + assert queries.subscription(db, stranger, "UCc") is None + assert queries.exploration(db, stranger, "UCc") is None + assert queries.is_blocked(db, stranger, "UCc") is False + assert queries.blocked_ids(db, stranger) == set() + assert [s.channel_id for s, _ in queries.subscriptions(db, stranger, "en")] == [] + + +# --- subscription settings ------------------------------------------------------------------- + + +def test_full_history_reports_only_the_switch_on(db, make_user): + user = make_user() + _channel(db) + sub = _sub(db, user) + assert service.update_subscription(db, sub, priority=3) is False + assert service.update_subscription(db, sub, deep_requested=True) is True + assert service.update_subscription(db, sub, deep_requested=True) is False # already on + assert _uncommitted(db) + db.commit() + assert (sub.priority, sub.hidden, sub.deep_requested) == (3, False, True) + + +def test_full_history_for_all_moves_only_the_callers_rows(db, make_user): + user, other = make_user(), make_user() + _channel(db) + _sub(db, user) + theirs = _sub(db, other) + assert service.set_deep_all(db, user, True) == 1 + db.commit() + db.expire_all() + assert theirs.deep_requested is False + + +def test_a_reset_asks_every_subscriber_for_full_history(db, make_user): + a, b = make_user(), make_user() + ch = _channel(db, backfill_done=True, recent_synced_at=datetime.now(timezone.utc)) + _sub(db, a) + _sub(db, b) + service.reset_backfill(db, ch) + db.commit() + db.expire_all() + assert (ch.backfill_done, ch.recent_synced_at) == (False, None) + assert set(db.scalars(select(Subscription.deep_requested))) == {True} + + +# --- explore, subscribe, block --------------------------------------------------------------- + + +def test_an_exploration_is_one_row_and_flags_the_channel_only_while_nobody_follows_it( + db, make_user +): + explorer, follower = make_user(), make_user() + lone = _channel(db, "UClone") + kept = _channel(db, "UCkept") + _sub(db, follower, "UCkept") + service.explore(db, explorer, lone) + service.explore(db, explorer, kept) + db.commit() + first = queries.exploration(db, explorer, "UClone").created_at + service.explore(db, explorer, lone) # refreshes the grace clock + db.commit() + assert db.scalar(select(ExploredChannel.id).where(ExploredChannel.channel_id == "UClone")) + assert len(db.scalars(select(ExploredChannel.id)).all()) == 2 + assert queries.exploration(db, explorer, "UClone").created_at >= first + assert (lone.from_explore, kept.from_explore) == (True, False) + + +def test_subscribing_keeps_an_explored_channel_and_its_videos(db, make_user): + user = make_user() + ch = _channel(db, from_explore=True) + v = _video(db, "v1", via_explore=True) + service.record_subscription(db, user, ch, "") + db.commit() + db.expire_all() + sub = queries.subscription(db, user, "UCc") + assert (sub.yt_subscription_id, ch.from_explore, v.via_explore) == (None, False, False) + service.drop_subscription(db, sub) + db.commit() + assert queries.subscription(db, user, "UCc") is None + + +def test_a_block_is_once_and_ends_the_exploration(db, make_user): + user = make_user() + ch = _channel(db) + service.explore(db, user, ch) + db.commit() + service.block(db, user, "UCc") + service.block(db, user, "UCc") + db.commit() + assert len(db.scalars(select(BlockedChannel.id)).all()) == 1 + assert queries.exploration(db, user, "UCc") is None + assert [c.id for c in queries.blocked_channels(db, user, "en")] == ["UCc"] + service.unblock(db, user, "UCc") + db.commit() + assert queries.is_blocked(db, user, "UCc") is False + + +# --- what the status cards count ------------------------------------------------------------- + + +def test_my_progress_counts_only_my_channels_and_my_unhidden_videos(db, make_user): + user, other = make_user(), make_user() + now = datetime.now(timezone.utc) + _channel(db, "UCa", details_synced_at=now, recent_synced_at=now, backfill_done=True) + _channel(db, "UCb") + _channel(db, "UCtheirs") + _video(db, "a1", "UCa") + _video(db, "b1", "UCb") + _video(db, "t1", "UCtheirs") + _sub(db, user, "UCa", deep_requested=True) + _sub(db, user, "UCb", hidden=True) + _sub(db, other, "UCtheirs") + assert queries.my_progress(db, user) == (2, 1, 1, 1) + assert [c.id for c in queries.my_deep_requested(db, user)] == ["UCa"] + assert queries.my_video_count(db, user) == 1 + assert queries.video_total(db) == 3 + assert queries.catalog_totals(db, user)["subscriptions"] == 2 + + +def test_a_channel_with_no_title_is_left_out_of_the_titles(db): + _channel(db, "UCt", title="Titled") + db.add(Channel(id="UCn", title=None)) + db.commit() + assert queries.channel_titles(db, {"UCt", "UCn", "UCgone"}) == {"UCt": "Titled"} + assert queries.channel_title(db, "UCt") == "Titled" + + +def test_the_deep_backlog_is_the_unfinished_channels_someone_asked_for(db, make_user): + user = make_user() + _channel(db, "UCwant") + _channel(db, "UCdone", backfill_done=True) + _channel(db, "UCnobody") + _sub(db, user, "UCwant", deep_requested=True) + _sub(db, user, "UCdone", deep_requested=True) + _sub(db, user, "UCnobody") + assert [c.id for c in queries.deep_backlog(db)] == ["UCwant"] + assert queries.deep_queued(db, ["UCwant", "UCdone", "UCnobody"]) == {"UCwant", "UCdone"} + + +# --- live search ----------------------------------------------------------------------------- + + +def test_a_cleared_search_spares_what_anyone_holds(db, make_user): + user, other, follower = make_user(), make_user(), make_user() + _channel(db, "UCs") + _channel(db, "UCfollowed") + _sub(db, follower, "UCfollowed") + for vid in ("loose", "watched", "found_by_other"): + _video(db, vid, "UCs", via_search=True) + _video(db, "followed", "UCfollowed", via_search=True) + _video(db, "organic", "UCs") + ids = ["loose", "watched", "found_by_other", "followed", "organic"] + search.record_finds(db, user, ids, "cat") + search.record_finds(db, user, ids, "cat") # idempotent + search.record_finds(db, other, ["found_by_other"], "dog") + reader_service.mark(db, other, "watched", "watched") + db.commit() + assert search.clear(db, user, ids) == 1 + db.commit() + left = set(db.scalars(select(Video.id))) + assert left == {"watched", "found_by_other", "followed", "organic"} + assert db.scalars(select(SearchFind.user_id)).all() == [other.id] + + +def test_a_search_puts_what_the_user_already_has_last(db, make_user): + user = make_user() + _channel(db, "UCs") + _channel(db, "UCmine") + _sub(db, user, "UCmine") + for vid, cid in (("new1", "UCs"), ("mine", "UCmine"), ("seen", "UCs"), ("new2", "UCs")): + _video(db, vid, cid) + reader_service.mark(db, user, "seen", "hidden") + db.commit() + assert search.new_first(db, user, ["mine", "new1", "seen", "new2"]) == [ + "new1", + "new2", + "mine", + "seen", + ] + + +# --- what Catalog asks Reader ---------------------------------------------------------------- + + +def test_a_users_tag_links_are_made_once_and_only_theirs_are_read(db, make_user): + user, other = make_user(), make_user() + _channel(db) + mine = reader_service.create_tag(db, user, "Mine", "other", None) + theirs = reader_service.create_tag(db, other, "Theirs", "other", None) + reader_service.attach_user_tag(db, user, "UCc", mine.id) + reader_service.attach_user_tag(db, user, "UCc", mine.id) + reader_service.attach_user_tag(db, other, "UCc", theirs.id) + db.commit() + assert reader_queries.user_tag_links(db, user) == {"UCc": [mine.id]} + assert reader_queries.user_tag_links(db, user, "UCelse") == {} + reader_service.detach_user_tag(db, user, "UCc", mine.id) + db.commit() + assert reader_queries.user_tag_links(db, user) == {} + assert reader_queries.user_tag_links(db, other) == {"UCc": [theirs.id]} + + +def test_discovery_offers_the_playlist_channels_not_followed_and_not_ones_own(db, make_user): + user = make_user() + for cid in ("UCnew", "UCfollowed", "UCown"): + _channel(db, cid) + for vid, cid in (("n1", "UCnew"), ("n2", "UCnew"), ("f1", "UCfollowed"), ("o1", "UCown")): + _video(db, vid, cid) + _sub(db, user, "UCfollowed") + pl = reader_service.create_playlist(db, user, "L") + for vid in ("n1", "n2", "f1", "o1"): + reader_service.add_item(db, pl, vid, mark_dirty=False) + db.commit() + rows = reader_queries.playlist_channels(db, user, "UCown", "en") + assert [(ch.id, n, lists) for ch, n, lists in rows] == [("UCnew", 2, 1)] + + +def test_an_exploration_older_than_its_grace_is_still_one_row(db, make_user): + """The grace clock is refreshed in place: re-exploring after the cutoff does not add a row.""" + user = make_user() + ch = _channel(db) + db.add( + ExploredChannel( + user_id=user.id, + channel_id="UCc", + created_at=datetime.now(timezone.utc) - timedelta(days=30), + ) + ) + db.commit() + service.explore(db, user, ch) + db.commit() + rows = db.scalars(select(ExploredChannel)).all() + assert len(rows) == 1 + assert rows[0].created_at > datetime.now(timezone.utc) - timedelta(minutes=1) diff --git a/backend/tests/test_channel_list_index_only.py b/backend/tests/test_channel_list_index_only.py index 397a8555..0fcc0d32 100644 --- a/backend/tests/test_channel_list_index_only.py +++ b/backend/tests/test_channel_list_index_only.py @@ -15,7 +15,7 @@ from sqlalchemy.dialects import postgresql from app.feedquery import filtered_query from app.catalog.models import Video -from app.routes.channels import channel_aggregates +from app.catalog.queries import channel_aggregates INDEX = "ix_videos_channel_pub_cover" diff --git a/backend/tests/test_like_literals.py b/backend/tests/test_like_literals.py index fc763e1d..265c24a5 100644 --- a/backend/tests/test_like_literals.py +++ b/backend/tests/test_like_literals.py @@ -10,7 +10,7 @@ from datetime import datetime, timezone from sqlalchemy import select from app.catalog.models import Channel, Subscription, Video -from app.routes.search import _fold_term +from app.catalog.search import _fold_term def _video(vid: str, ch: str, title: str, search_terms: str | None = None) -> Video: diff --git a/backend/tests/test_route_conventions.py b/backend/tests/test_route_conventions.py index 4ea0e820..59b4bb7d 100644 --- a/backend/tests/test_route_conventions.py +++ b/backend/tests/test_route_conventions.py @@ -20,9 +20,6 @@ COMMITTING_HELPERS = { "not hand the token back", "auth.py:_complete_link": "the tail of the Google callback route, which answers with its result", "auth.py:_register_account": "a background task with its own session, i.e. a job wrapper", - "search.py:_ingest_candidates": "stages around YouTube calls: the stubs must survive an enrich " - "that fails", - "search.py:_classify_shorts": "the last stage of that ingest, after its probes", "auth.py:upsert_pending_invite": "commits only when it writes — the contract its two " "callers rely on (`_register_account` persists its own flushed user because a no-op here " "commits nothing)", diff --git a/backend/tests/test_search_scrape_ingest.py b/backend/tests/test_search_scrape_ingest.py index 90da002c..8d817880 100644 --- a/backend/tests/test_search_scrape_ingest.py +++ b/backend/tests/test_search_scrape_ingest.py @@ -16,7 +16,7 @@ from datetime import datetime, timezone from app import sysconfig from app.models import User from app.catalog.models import Video -from app.routes.search import _ingest_candidates +from app.catalog.search import ingest_candidates class _FakeYT: @@ -71,7 +71,7 @@ def test_scrape_ingest_dateless_stub_ingests_and_takes_real_enrich_date(db): "vid_bbbbbbb": _enrich_item("vid_bbbbbbb", "2023-10-30T17:00:08Z"), } - kept = _ingest_candidates(db, _FakeYT(enrich), user, candidates) + kept = ingest_candidates(db, _FakeYT(enrich), user, candidates) # No IntegrityError, and both non-Short/non-live videos are kept in input order. assert kept == ["vid_aaaaaaa", "vid_bbbbbbb"] @@ -89,7 +89,7 @@ def test_scrape_ingest_enrich_miss_keeps_nonnull_placeholder(db): db.add(user) db.commit() - kept = _ingest_candidates(db, _FakeYT({}), user, [_candidate("vid_ccccccc", "UC_chan_c")]) + kept = ingest_candidates(db, _FakeYT({}), user, [_candidate("vid_ccccccc", "UC_chan_c")]) assert kept == [] c = db.get(Video, "vid_ccccccc") @@ -101,7 +101,7 @@ def test_search_youtube_route_returns_422_when_ingest_fails(db, monkeypatch, ove # Route-level guard (B-core-1): an ingest failure with nothing yet collected must surface as a 422 # + message on this JSON endpoint (the FE hides 5xx behind a generic "connection lost" + auto-retry, # burning a second quota-costed search), and the session is rolled back. Here the page fetch succeeds - # (one candidate) and _ingest_candidates raises → the endpoint returns 422. + # (one candidate) and ingest_candidates raises → the endpoint returns 422. from fastapi.testclient import TestClient import app.state as state @@ -144,7 +144,7 @@ def test_search_youtube_route_returns_422_when_ingest_fails(db, monkeypatch, ove def _boom(db_, yt, user_, candidates): raise RuntimeError("ingest exploded") - monkeypatch.setattr(search_mod, "_ingest_candidates", _boom) + monkeypatch.setattr(search_mod.catalog_search, "ingest_candidates", _boom) override_dep(get_db, db.info["request"]) override_dep(current_user, lambda: user) diff --git a/backend/tests/test_unit_fences.py b/backend/tests/test_unit_fences.py index c51b968e..125b3b1b 100644 --- a/backend/tests/test_unit_fences.py +++ b/backend/tests/test_unit_fences.py @@ -110,7 +110,6 @@ UNIT_ROUTES = { # What each HTTP file still does, by file: the debt a unit's sprint pays off. Exact counts. A core # route's count is only the unit tables it imports; it is paid by the sprint of the unit it reaches. ALLOWED = { - "channels.py": 83, "plex/_shared.py": 1, "plex/admin.py": 12, "plex/browse.py": 93, @@ -120,10 +119,6 @@ ALLOWED = { "plex/people.py": 20, "plex/playback.py": 42, "plex/playlists.py": 80, - "runs.py": 1, - "scheduler.py": 3, - "search.py": 46, - "sync.py": 38, } # The session methods that read or write data. `commit` and `rollback` stay with the handler. @@ -341,11 +336,22 @@ UNIT_PACKAGES = { "sync/playlist_merge.py", "sync/tagsuggest.py", ), + # The catalog's ingest jobs: the subscription import, the video sync and backfills, the explore + # and search ingests, the maintenance re-checks and the auto-tagger. + "catalog": ( + "catalog/", + "sync/autotag.py", + "sync/explore.py", + "sync/maintenance.py", + "sync/runner.py", + "sync/subscriptions.py", + "sync/videos.py", + ), } # The units fenced so far; the rest join with their sprint. -FENCED = {"capture", "reader"} +FENCED = {"capture", "reader", "catalog"} # Whose tables a unit's own code may READ. Core code reads a fenced unit's tables through its queries. -READS = {"capture": {"catalog", "plex"}, "reader": {"catalog", "plex"}} +READS = {"capture": {"catalog", "plex"}, "reader": {"catalog", "plex"}, "catalog": set()} # The columns that say whose a row is. OWNER_COLUMNS = {"user_id", "to_user_id", "from_user_id", "created_by"} # Where a fenced unit's code may commit: the jobs that own their session (the worker and what it @@ -372,6 +378,32 @@ UNIT_COMMITS = { "sync/playlists.py:merge_playlist", "sync/playlists.py:repull_playlist", }, + # The catalog's ingest jobs commit per page or batch, a YouTube call between each: a page + # fetched and stored is not lost when the next call fails, and an enrich that fails rolls back + # to the stubs the step before committed (a search's results, a playlist's unseen videos). + "catalog": { + "catalog/search.py:classify_shorts", + "catalog/search.py:ingest_candidates", + "sync/autotag.py:apply_channel_autotags", + "sync/autotag.py:run_autotag_all", + "sync/explore.py:_enrich_new", + "sync/explore.py:purge_unkept_explored", + "sync/explore.py:purge_unkept_search", + "sync/maintenance.py:_notify_removed", + "sync/maintenance.py:_recheck_flagged", + "sync/maintenance.py:_revalidate_rolling", + "sync/runner.py:run_feed_fallback", + "sync/runner.py:_stamp_fallback_attempt", + "sync/subscriptions.py:import_subscriptions", + "sync/videos.py:insert_stubs", + "sync/videos.py:apply_rss_feed", + "sync/videos.py:backfill_channel_recent", + "sync/videos.py:backfill_channel_deep", + "sync/videos.py:reconcile_full_history", + "sync/videos.py:_apply_video_batch", + "sync/videos.py:run_shorts_classification", + "sync/videos.py:ensure_videos", + }, } SQL_WRITES = {"insert", "update", "delete", "pg_insert"} @@ -506,7 +538,9 @@ PACKAGE_CASES = { ("downloads/x.py", "from app.models import Video\nupdate(Video)"): 1, ("downloads/x.py", "from app.downloads.models import DownloadJob\nDownloadJob()"): 0, ("search.py", "from app.downloads.models import DownloadJob"): 1, # core reads capture - ("search.py", "from app.models import Video\nVideo()"): 0, # neither side fenced yet + ("search.py", "from app.models import PlexItem\nPlexItem()"): 0, # neither side fenced yet + ("search.py", "from app.models import Video\nVideo()"): 2, # core reads and writes catalog + ("sync/videos.py", "from app.models import Playlist"): 1, # no catalog→reader edge ("downloads/x.py", "q = DownloadJob.user_id == 1"): 1, ("downloads/x.py", "q = owned_by(DownloadJob.user_id, 1)"): 0, ("downloads/x.py", "def f(db):\n db.commit()"): 1, diff --git a/docs/gates.md b/docs/gates.md index 987d7a1c..88f0e772 100644 --- a/docs/gates.md +++ b/docs/gates.md @@ -113,7 +113,7 @@ project checks needs a row. | CVE gate | HIGH/CRITICAL with a fix available only | `--ignore-unfixed` + `--severity HIGH,CRITICAL`: a finding with no fix cannot be acted on by an upgrade; it shows in the scan output. A fixable one fails the publish, and the only way past it is `.trivyignore.yaml`. | 0 open (E32 S1) | | npm audit | `--omit=dev --audit-level=high` | What ships to the browser, at high or above. The dev toolchain's advisories were burned down by hand in E32 S1 (18 → 0). | 0 | | knip | `entry` — one file added | Not a narrowing: knip's Vite plugin reads entries from index.html only, so the shared link's page (`watch.html` → `src/watchMain.tsx`) is named here, or the lane reports that whole entry unused. It adds files to the lane, never takes any out. | 1 entry | -| unit fence | the database work the HTTP layer still does, pinned | `backend/tests/test_unit_fences.py`: a unit's route file may not build SQL, call the session's data methods, set a model's attributes or import a unit's tables; a core route may not import a unit's tables. Switched on unit by unit: `ALLOWED` pins each file's count exactly, so a rise fails and a fall fails until the pin is lowered — the list may only shrink. End state: empty, and the pin mechanism goes. A unit in `FENCED` is held below its HTTP layer too: its code reads another unit's tables only along `READS`, writes them never, filters owners through `userscope`, and commits only where `UNIT_COMMITS` names a job's own session. | 826 in 23 files (E42 S0) → 667 in 19 (S1: Capture 0, fenced) → 525 in 14 (S2: Reader 0, fenced) | +| unit fence | the database work the HTTP layer still does, pinned | `backend/tests/test_unit_fences.py`: a unit's route file may not build SQL, call the session's data methods, set a model's attributes or import a unit's tables; a core route may not import a unit's tables. Switched on unit by unit: `ALLOWED` pins each file's count exactly, so a rise fails and a fall fails until the pin is lowered — the list may only shrink. End state: empty, and the pin mechanism goes. A unit in `FENCED` is held below its HTTP layer too: its code reads another unit's tables only along `READS`, writes them never, filters owners through `userscope`, and commits only where `UNIT_COMMITS` names a job's own session. | 826 in 23 files (E42 S0) → 667 in 19 (S1: Capture 0, fenced) → 525 in 14 (S2: Reader 0, fenced) → 354 in 9 (S3: Catalog 0, fenced; core routes 0) | | React Compiler | the functions it skips, pinned | `components/compilerCoverage.contract.test.ts` — the list may only shrink. End state: only VirtualGrid (TanStack Virtual, an IncompatibleLibrary) remains. | 93 (E32 S1) → 29 (E32 S3) → 1, VirtualGrid (E32 S4) | ## Accepted findings