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

from app import quota
from app.config import settings
from app.models import Channel, OAuthToken, User
from app.sync.videos import (
    backfill_channel_deep,
    backfill_channel_recent,
    enrich_pending,
    poll_rss_channel,
)
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()
    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_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()
    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"}
    channels = (
        db.execute(
            select(Channel).where(
                Channel.backfill_done.is_(False),
                Channel.recent_synced_at.is_not(None),
            )
        )
        .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()
    return {"channels_processed": processed, "videos_added": videos_added}