siftlode/backend/app/sync/runner.py

"""Reusable sync routines shared by the scheduler and the manual /api/sync endpoints.

Channel/video data is shared across users, so background work acts through a "service
user" (any user with a stored refresh token) unless a YOUTUBE_API_KEY is configured for
public reads.
"""

import logging
import threading
from collections import Counter
from concurrent.futures import ThreadPoolExecutor, as_completed

from sqlalchemy import select
from sqlalchemy.orm import Session

from app import logs, progress, quota, sysconfig

log = logs.get_logger(__name__)
from app.db import SessionLocal
from app.models import Channel, OAuthToken, Subscription, User
from app.sync.explore import followed_channel
from app.sync.subscriptions import import_subscriptions
from app.sync.videos import (
    apply_rss_feed,
    backfill_channel_deep,
    backfill_channel_recent,
    enrich_pending,
    reconcile_full_history,
    refresh_live,
    run_shorts_classification,
)
from app.youtube.client import YouTubeClient
from app.youtube.rss import RssError, fetch_channel_feed

# RSS feeds are free and network-bound, so fetch many at once; DB writes stay single-threaded.
RSS_POLL_WORKERS = 16
# Failed channels named in a poll's summary warning; the rest are counted by reason.
RSS_FAILURE_EXAMPLES = 5


def token_users(db: Session) -> list[User]:
    """Every user with a stored YouTube refresh token (the read-scope service accounts), by id."""
    return list(
        db.execute(
            select(User)
            .join(OAuthToken)
            .where(OAuthToken.refresh_token_enc.is_not(None))
            .order_by(User.id)
        ).scalars()
    )


def get_service_user(db: Session) -> User | None:
    users = token_users(db)
    return users[0] if users else None


def run_rss_poll(db: Session, channels: list[Channel] | None = None) -> dict[str, int]:
    """Poll the free RSS feed of every followed channel (`followed_channel`). Returns `{"new": inserted, "failed": channels_that_did
    _not_complete}` — the failure count is part of the RESULT, not just a log line, so the scheduler
    card and the audit summary can tell "no new uploads" apart from "nothing worked" (an egress
    outage, or a database that can't accept writes, otherwise renders identically to a quiet hour).

    `failed` counts a channel whose feed could not be READ *or* whose rows could not be APPLIED —
    both leave the channel un-polled, and counting only the first would move the blind spot rather
    than remove it.

    Raises RuntimeError when NOTHING could be polled (`failed == total`, total > 0): a run that
    achieved literally nothing is a failed run, and only an exception turns the scheduler card red
    — a returned count leaves it green with the outage buried in the summary line. The message
    carries the FIRST underlying error, because on a small instance "all channels failed" is also
    what a single dead feed id looks like."""
    if channels is None:
        channels = db.execute(select(Channel).where(followed_channel())).scalars().all()
    total = len(channels)
    new = 0
    done = 0
    failed = 0
    # Kept for the total-outage message: "failed for all N channels" alone can't tell an egress
    # outage from one permanently dead feed id — which is the whole difference on an instance with
    # a single subscription, where BOTH render as 100% failed.
    first_error: str | None = None
    # What failed, as counts per reason (`http_404`, `transport`, `apply`, …) plus a few examples —
    # the ONE warning a poll writes. One line per failed channel was what turned YouTube's RSS outage
    # of 2026-09-16 into 11k warnings in four hours; each channel's own line is DEBUG now.
    reasons: Counter[str] = Counter()
    examples: list[str] = []

    def fail(reason: str, detail: str, *, unexpected: str | None = None) -> None:
        """Count a channel that did not get polled. `unexpected` is the message for a failure that
        is a bug on our side: its FIRST occurrence per reason gets a traceback at ERROR, the rest
        DEBUG — a read-only database fails every channel the same way, and 979 identical tracebacks
        tell nobody anything the first one didn't. Call it from inside the `except` block."""
        nonlocal failed, first_error
        if unexpected is not None:
            level = logging.ERROR if reasons[reason] == 0 else logging.DEBUG
            log.log(level, unexpected, exc_info=True)
        failed += 1
        reasons[reason] += 1
        first_error = first_error or detail
        if len(examples) < RSS_FAILURE_EXAMPLES:
            examples.append(detail)

    # Fetch feeds concurrently (network-bound, no DB), then apply inserts on this thread as
    # each fetch returns — a SQLAlchemy session isn't thread-safe. Pass the channel id (a
    # plain str) to the workers; the ORM instance is touched only here on the main thread.
    futures = {}
    with ThreadPoolExecutor(max_workers=RSS_POLL_WORKERS) as pool:
        for channel in channels:
            futures[pool.submit(fetch_channel_feed, channel.id)] = channel
        for fut in as_completed(futures):
            channel = futures[fut]
            try:
                entries = fut.result()
            # A failed poll is NOT an empty feed: skip the apply so `last_rss_at` stays where it
            # was (stamping it made a permanently-broken feed look freshly polled). The shared
            # counter/progress tail lives after the statement, so the two outcomes can stay two
            # clauses: an RssError is EXPECTED (404, timeout) and is only counted; anything else is a
            # bug on our side and earns a traceback (the first of its kind per poll).
            except RssError as exc:
                log.debug("RSS poll skipped for channel %s: %s", channel.id, exc)
                fail(exc.reason, f"{channel.id}: {exc}")
            # silent-ok: `fail(unexpected=…)` is the report — it logs at ERROR with the
            # traceback, once per reason per poll, which is the flood control this loop needs.
            except Exception as exc:  # noqa: BLE001 — one channel must not stop the poll
                fail(
                    "error", f"{channel.id}: {exc}", unexpected=f"RSS fetch failed for {channel.id}"
                )
            else:
                try:
                    new += apply_rss_feed(db, channel, entries)
                # silent-ok: `fail(unexpected=…)` below is the report — ERROR with the traceback,
                # once per reason per poll.
                except Exception as exc:  # noqa: BLE001 — one channel must not stop the poll
                    # A write failure (read-only DB, failover) leaves the channel just as un-polled
                    # as an unreachable feed — count it, or a wholly broken run reports "failed=0".
                    db.rollback()
                    fail(
                        "apply",
                        f"{channel.id} (apply): {exc}",
                        unexpected=f"RSS apply failed for {channel.id}",
                    )
            # One place that advances the counter, whatever happened above.
            done += 1
            progress.report(done, total, "rss_poll")
    if failed:
        log.warning(
            "RSS poll: %d/%d channels could not be polled",
            failed,
            total,
            extra=logs.event(
                "sync.rss.failed",
                failed=failed,
                total=total,
                reasons=dict(reasons),
                examples=examples,
            ),
        )
    if total and failed == total:
        # Nothing was polled at all — an egress outage (the tinyproxy-after-reboot incident), a
        # dead DNS, a read-only database. Returning normally would record the run as `ok` with the
        # failure hidden inside the result string, i.e. a green pill for a total outage.
        # The first error TRAVELS WITH the message on purpose: on an instance with one or two
        # subscriptions "all channels failed" is also what ONE dead feed id looks like, and a red
        # card the user can't act on is its own kind of dishonest. `channel: HTTP 404` says
        # "fix this subscription"; `Connection refused` says "the egress is down".
        raise RuntimeError(f"RSS poll failed for all {total} channels — first: {first_error}")
    return {"new": new, "failed": failed}


def run_enrich(db: Session, max_batches: int = 200, floor: int = 200) -> int:
    user = get_service_user(db)
    if user is None:
        return 0
    total = 0
    with YouTubeClient(db, user) as yt:
        for _ in range(max_batches):
            if quota.remaining_today(db) <= floor:
                break
            n = enrich_pending(db, yt)
            total += n
            progress.report(total, None, "enrich")  # total unknown (drains until empty)
            if n == 0:
                break
        # Revisit transient live/upcoming videos (enrich_pending is one-shot) so an ended
        # stream gains its final duration + was_live status instead of staying "live" forever.
        if quota.remaining_today(db) > floor:
            total += refresh_live(db, yt)
    return total


def run_shorts(db: Session) -> dict:
    return run_shorts_classification(db)


def run_subscription_resync(db: Session) -> dict:
    """Re-import every user's subscriptions so unsubscribes and new subscriptions on
    YouTube are reflected automatically."""
    users = token_users(db)
    for i, user in enumerate(users, 1):
        try:
            import_subscriptions(db, user)
        except Exception:
            db.rollback()
            log.exception("Subscription resync failed for user %s", user.id)
        progress.report(i, len(users), "subscriptions")
    return {"users": len(users)}


def run_recent_backfill(
    db: Session, channels: list[Channel] | None = None, max_channels: int | None = None
) -> dict:
    user = get_service_user(db)
    if user is None:
        return {"channels_processed": 0, "videos_added": 0, "reason": "no service user"}
    if channels is None:
        channels = (
            db.execute(
                select(Channel).where(Channel.recent_synced_at.is_(None), followed_channel())
            )
            .scalars()
            .all()
        )
    if max_channels:
        channels = channels[:max_channels]
    processed = 0
    videos_added = 0
    with YouTubeClient(db, user) as yt:
        for channel in channels:
            if quota.backfill_reserve_reached():
                break
            try:
                videos_added += backfill_channel_recent(db, yt, channel)
                processed += 1
                progress.report(processed, len(channels), "backfill_recent")
            except Exception:
                db.rollback()
                log.exception("Recent backfill failed for channel %s", channel.id)
    reconciled = reconcile_full_history(db)
    return {
        "channels_processed": processed,
        "videos_added": videos_added,
        "reconciled_full": reconciled,
    }


def run_deep_backfill(db: Session, max_channels: int = 10, max_pages: int = 5) -> dict:
    user = get_service_user(db)
    if user is None:
        return {"channels_processed": 0, "videos_added": 0, "reason": "no service user"}
    # Demand-driven: only deep-backfill channels at least one user has opted into
    # (Subscription.deep_requested). Recent backfill needs no opt-in: it covers every followed channel
    # (`followed_channel`).
    deep_requested = (
        select(Subscription.id)
        .where(
            Subscription.channel_id == Channel.id,
            Subscription.deep_requested.is_(True),
        )
        .exists()
    )
    channels = (
        db.execute(
            select(Channel).where(
                Channel.backfill_done.is_(False),
                Channel.recent_synced_at.is_not(None),
                deep_requested,
            )
        )
        .scalars()
        .all()
    )[:max_channels]
    processed = 0
    videos_added = 0
    with YouTubeClient(db, user) as yt:
        for channel in channels:
            if quota.backfill_reserve_reached():
                break
            # Claim the channel (or skip it if an interactive backfill already owns it) so the two
            # never advance the same cursor concurrently — symmetric with backfill_channel_now.
            with _backfill_lock:
                if channel.id in _backfill_in_progress:
                    continue
                _backfill_in_progress.add(channel.id)
            try:
                videos_added += backfill_channel_deep(db, yt, channel, max_pages)
                processed += 1
                progress.report(processed, len(channels), "backfill_deep")
            except Exception:
                db.rollback()
                log.exception("Deep backfill failed for channel %s", channel.id)
            finally:
                with _backfill_lock:
                    _backfill_in_progress.discard(channel.id)
    reconciled = reconcile_full_history(db)
    return {
        "channels_processed": processed,
        "videos_added": videos_added,
        "reconciled_full": reconciled,
    }


# Channels being deep-backfilled RIGHT NOW by EITHER path — the interactive backfill_channel_now (a
# FastAPI BackgroundTasks threadpool thread) OR the scheduler's run_deep_backfill (its own thread).
# BOTH register the channel they're paging (under the lock) so the other skips it: no two runs page
# the same channel's `backfill_cursor` concurrently, whichever started first. `_immediate_running`
# counts only the interactive path, capped so one user's burst can't drain the shared budget (the
# scheduler is already quota-paced). Single uvicorn process → one shared module state, lock suffices.
_backfill_in_progress: set[str] = set()
_immediate_running = 0
_backfill_lock = threading.Lock()
_MAX_CONCURRENT_IMMEDIATE = 2


def backfill_channel_now(channel_id: str, user_id: int, max_pages: int = 400) -> None:
    """User-initiated "get full history": deep-backfill ONE channel to completion right now, in a
    background task with its OWN session, so the full catalogue lands in seconds instead of over the
    paced scheduler's 10-minute ticks (max_pages=5/channel/tick). max_pages is high enough to finish
    any realistic channel in a single run (400 pages = 20 000 videos; e.g. BBC Earth's ~3400 = 69
    pages — the old 60-page cap stopped one page short of large channels, so their tail dribbled in
    over the scheduler's 10-minute ticks and looked stuck). backfill_channel_deep re-checks the shared
    budget before each page and yields at the reserve, so a high cap still can't drain the day's quota.
    Quota-guarded, capped to _MAX_CONCURRENT_IMMEDIATE interactive runs so one user can't drain the
    shared backfill budget by opting many channels in at once (the overflow, and any channel too large
    to finish under the reserve, is finished by the scheduler since deep_requested stays set), and
    mutually excluded per-channel with the scheduler via _backfill_in_progress. Best-effort — logs and
    swallows failures."""
    global _immediate_running
    with _backfill_lock:
        if channel_id in _backfill_in_progress or _immediate_running >= _MAX_CONCURRENT_IMMEDIATE:
            return  # already backfilling this channel, or too many interactive runs — leave it to the scheduler
        _backfill_in_progress.add(channel_id)
        _immediate_running += 1
    try:
        with SessionLocal() as db:
            user = get_service_user(db)
            if user is None:
                return
            channel = db.get(Channel, channel_id)
            if channel is None or channel.backfill_done or not channel.uploads_playlist_id:
                return
            if quota.backfill_reserve_reached():
                return
            try:
                # Attribute the spend to the requesting user under the full-backfill action, so it
                # shows up correctly in the usage breakdown (not the default OTHER/system bucket).
                with quota.attribute(user_id, quota.QuotaAction.VIDEOS_BACKFILL_FULL):
                    with YouTubeClient(db, user) as yt:
                        backfill_channel_deep(db, yt, channel, max_pages)
                    reconcile_full_history(db)
            except Exception:
                db.rollback()
                log.exception("Immediate backfill failed for channel %s", channel_id)
    finally:
        with _backfill_lock:
            _backfill_in_progress.discard(channel_id)
            _immediate_running -= 1


def estimate_deep_backfill(db: Session, channels: list[Channel]) -> dict:
    """Rough ETA to finish full-history backfill of `channels`. Quota-bound: total
    estimated API units / the daily budget left for backfill. Deliberately approximate —
    a channel's cost ~= playlistItems pages (1 unit / 50 videos) + enrichment (likewise)."""
    pending = [c for c in channels if not c.backfill_done]

    def units(c: Channel) -> int:
        videos = c.video_count or 0
        pages = max(1, (videos + 49) // 50)
        return 2 * pages

    total_units = sum(units(c) for c in pending)
    daily_budget = max(
        1,
        sysconfig.get_int(db, "quota_daily_budget")
        - sysconfig.get_int(db, "backfill_quota_reserve"),
    )
    eta_seconds = int(total_units / daily_budget * 86_400) if pending else 0
    return {
        "channels_pending": len(pending),
        "videos_pending_est": sum(c.video_count or 0 for c in pending),
        "units_est": total_units,
        "daily_budget": daily_budget,
        "eta_seconds": eta_seconds,
    }