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
from concurrent.futures import ThreadPoolExecutor, as_completed

from sqlalchemy import select
from sqlalchemy.orm import Session

from app import progress, quota, sysconfig

log = logging.getLogger("siftlode.sync")
from app.models import Channel, OAuthToken, Subscription, User
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


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 every channel's free RSS feed. 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)).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
    # 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 gets one warning line; anything
            # else is a bug on our side and earns a traceback.
            except RssError as exc:
                log.warning("RSS poll skipped for channel %s: %s", channel.id, exc)
                failed += 1
                first_error = first_error or f"{channel.id}: {exc}"
            except Exception as exc:
                log.exception("RSS fetch failed for channel %s", channel.id)
                failed += 1
                first_error = first_error or f"{channel.id}: {exc}"
            else:
                try:
                    new += apply_rss_feed(db, channel, entries)
                except Exception as exc:
                    # 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()
                    failed += 1
                    first_error = first_error or f"{channel.id} (apply): {exc}"
                    log.exception("RSS apply failed for channel %s", 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)
    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)))
            .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.remaining_today(db) <= sysconfig.get_int(db, "backfill_quota_reserve"):
                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 stays unconditional (cheap).
    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.remaining_today(db) <= sysconfig.get_int(db, "backfill_quota_reserve"):
                break
            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)
    reconciled = reconcile_full_history(db)
    return {
        "channels_processed": processed,
        "videos_added": videos_added,
        "reconciled_full": reconciled,
    }


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,
    }