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 import quota
from app.config import settings
from app.db import SessionLocal
from app.state import is_sync_paused
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:
        if is_sync_paused(db):
            logger.info("job %s skipped (sync paused)", name)
            return
        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:
    def work(db):
        with quota.attribute(None, "enrich"):
            return run_enrich(db)

    _job("enrich", work)


def _backfill_job() -> None:
    # Recent-first for not-yet-synced channels, then deep backfill for the rest. All
    # background spend is attributed to the system (no actor), split by action.
    def work(db):
        with quota.attribute(None, "backfill_recent"):
            recent = run_recent_backfill(db, max_channels=25)
        with quota.attribute(None, "backfill_deep"):
            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:
    def work(db):
        with quota.attribute(None, "subscription_resync"):
            return run_subscription_resync(db)

    _job("subscriptions", work)


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