siftlode/backend/app/scheduler.py

"""Background scheduler: free RSS detection, enrichment of new videos, and quota-aware
recent-first / deep backfill. Runs inside the API process (single worker)."""
import logging

from apscheduler.schedulers.background import BackgroundScheduler

from app.config import settings
from app.db import SessionLocal
from app.sync.autotag import run_autotag_all
from app.sync.runner import (
    run_deep_backfill,
    run_enrich,
    run_recent_backfill,
    run_rss_poll,
    run_shorts,
    run_subscription_resync,
)

logger = logging.getLogger("subfeed.scheduler")

_scheduler: BackgroundScheduler | None = None


def _job(name: str, fn) -> None:
    db = SessionLocal()
    try:
        result = fn(db)
        logger.info("job %s -> %s", name, result)
    except Exception:
        db.rollback()
        logger.exception("job %s failed", name)
    finally:
        db.close()


def _rss_job() -> None:
    _job("rss_poll", lambda db: run_rss_poll(db))


def _enrich_job() -> None:
    _job("enrich", lambda db: run_enrich(db))


def _backfill_job() -> None:
    # Recent-first for not-yet-synced channels, then deep backfill for the rest.
    def work(db):
        recent = run_recent_backfill(db, max_channels=25)
        deep = run_deep_backfill(db, max_channels=10)
        return {"recent": recent, "deep": deep}

    _job("backfill", work)


def _autotag_job() -> None:
    _job("autotag", lambda db: run_autotag_all(db, only_missing=True))


def _shorts_job() -> None:
    _job("shorts", run_shorts)


def _subscriptions_job() -> None:
    _job("subscriptions", run_subscription_resync)


def start_scheduler() -> None:
    global _scheduler
    if not settings.scheduler_enabled or _scheduler is not None:
        return
    scheduler = BackgroundScheduler(timezone="UTC")
    scheduler.add_job(
        _rss_job, "interval", minutes=settings.rss_poll_minutes, id="rss_poll"
    )
    scheduler.add_job(
        _enrich_job, "interval", minutes=settings.enrich_interval_minutes, id="enrich"
    )
    scheduler.add_job(
        _backfill_job,
        "interval",
        minutes=settings.backfill_interval_minutes,
        id="backfill",
    )
    scheduler.add_job(
        _autotag_job,
        "interval",
        minutes=settings.autotag_interval_minutes,
        id="autotag",
    )
    scheduler.add_job(
        _shorts_job,
        "interval",
        minutes=settings.shorts_probe_interval_minutes,
        id="shorts",
    )
    scheduler.add_job(
        _subscriptions_job,
        "interval",
        minutes=settings.subscriptions_resync_minutes,
        id="subscriptions",
    )
    scheduler.start()
    _scheduler = scheduler
    logger.info("scheduler started")


def shutdown_scheduler() -> None:
    global _scheduler
    if _scheduler is not None:
        _scheduler.shutdown(wait=False)
        _scheduler = None