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("subfeed.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 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 get_service_user(db: Session) -> User | None:
    return (
        db.execute(
            select(User)
            .join(OAuthToken)
            .where(OAuthToken.refresh_token_enc.is_not(None))
            .order_by(User.id)
        )
        .scalars()
        .first()
    )


def run_rss_poll(db: Session, channels: list[Channel] | None = None) -> int:
    if channels is None:
        channels = db.execute(select(Channel)).scalars().all()
    total = len(channels)
    new = 0
    done = 0
    # 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()
            except Exception:
                log.exception("RSS fetch failed for channel %s", channel.id)
                entries = []
            try:
                new += apply_rss_feed(db, channel, entries)
            except Exception:
                db.rollback()
                log.exception("RSS apply failed for channel %s", channel.id)
            done += 1
            progress.report(done, total, "rss_poll")
    return new


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 = (
        db.execute(
            select(User)
            .join(OAuthToken)
            .where(OAuthToken.refresh_token_enc.is_not(None))
        )
        .scalars()
        .all()
    )
    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,
    }