"""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), ) ) db.flush() settle_explored(db, channel) def settle_explored(db: Session, channel: Channel) -> None: """Hold the rule that a channel's videos are an exploration's only while nobody follows it, decided by who follows it now, not by the flag held in memory: a subscribe in another request may have cleared that while YouTube answered this one, and the videos that came in meanwhile would stay flagged. A channel someone follows, however the follow came (here or pulled in from YouTube), is the catalog's own, and so are its videos.""" if channel.from_explore and not queries.followed(db, channel.id): return 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 ) )