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 sqlalchemy import select
from sqlalchemy.orm import Session

from app import quota
from app.config import settings

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 (
    backfill_channel_deep,
    backfill_channel_recent,
    enrich_pending,
    poll_rss_channel,
    run_shorts_classification,
)
from app.youtube.client import YouTubeClient


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()
    new = 0
    for channel in channels:
        try:
            new += poll_rss_channel(db, channel)
        except Exception:
            db.rollback()
            log.exception("RSS poll failed for channel %s", channel.id)
    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
            if n == 0:
                break
    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 user in users:
        try:
            import_subscriptions(db, user)
        except Exception:
            db.rollback()
            log.exception("Subscription resync failed for user %s", user.id)
    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) <= settings.backfill_quota_reserve:
                break
            try:
                videos_added += backfill_channel_recent(db, yt, channel)
                processed += 1
            except Exception:
                db.rollback()
                log.exception("Recent backfill failed for channel %s", channel.id)
    return {"channels_processed": processed, "videos_added": videos_added}


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) <= settings.backfill_quota_reserve:
                break
            try:
                videos_added += backfill_channel_deep(db, yt, channel, max_pages)
                processed += 1
            except Exception:
                db.rollback()
                log.exception("Deep backfill failed for channel %s", channel.id)
    return {"channels_processed": processed, "videos_added": videos_added}


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, settings.quota_daily_budget - settings.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,
    }