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 contextvars
import threading
from datetime import datetime, timezone
from typing import Callable

from apscheduler.schedulers.background import BackgroundScheduler
from sqlalchemy import select

from app import audit, debugsession, logs, logsink, progress, quota
from app.audit import AuditAction
from app.config import settings
from app.db import SessionLocal
from app.demo import reset_demo_state
from app.downloads.gc import run_download_gc
from app.models import SchedulerSetting, User
from app.plex.sync import sync as run_plex_sync
from app.plex.tmdb import run_enrich as run_tmdb_enrich
from app.plex.tmdb import run_people_enrich
from app.plex.watch_sync import run_plex_watch_reconcile, run_plex_watch_sync
from app.notifications import create_notification
from app.state import is_sync_paused
from app.sync.autotag import run_autotag_all
from app.sync.explore import purge_ephemeral
from app.sync.maintenance import run_maintenance
from app.sync.playlists import sync_all_playlists
from app.sync.runner import (
    run_deep_backfill,
    run_enrich,
    run_recent_backfill,
    run_rss_poll,
    run_shorts,
    run_subscription_resync,
)

log = logs.get_logger(__name__)
_scheduler: BackgroundScheduler | None = None

# Per-job run activity, kept in-memory for the admin Scheduler dashboard to read (the
# scheduler lives in the same single-worker process as the API, so a request can read it
# directly). Ephemeral by design — it reflects "what the scheduler is doing right now and
# how the last runs went" since this process started; durable history isn't the goal here.
_activity: dict[str, dict] = {}
_activity_lock = threading.Lock()

# Set (per-thread) by a manual "run now" trigger to the id of the admin who clicked, so that
# run — and only that run, not the recurring scheduled ones — posts a completion notification
# to their inbox. Unset (None) for scheduled runs.
_manual_actor: contextvars.ContextVar[int | None] = contextvars.ContextVar(
    "manual_actor", default=None
)

# Each interval job's env/config default period (minutes). The admin can override any of
# these at runtime (stored in scheduler_settings, applied live); see load_intervals.
JOB_INTERVALS: dict[str, int] = {
    "rss_poll": settings.rss_poll_minutes,
    "enrich": settings.enrich_interval_minutes,
    "backfill": settings.backfill_interval_minutes,
    "autotag": settings.autotag_interval_minutes,
    "shorts": settings.shorts_probe_interval_minutes,
    "subscriptions": settings.subscriptions_resync_minutes,
    "playlist_sync": settings.playlist_sync_minutes,
    "maintenance": settings.maintenance_interval_minutes,
    "demo_reset": settings.demo_reset_minutes,
    "explore_cleanup": settings.explore_cleanup_minutes,
    "download_gc": settings.download_gc_minutes,
    "plex_sync": settings.plex_sync_interval_min,
    "plex_watch_sync": settings.plex_watch_sync_interval_min,
    "plex_watch_reconcile": settings.plex_watch_reconcile_interval_min,
    "tmdb_enrich": settings.tmdb_enrich_interval_min,
    "people_enrich": settings.people_enrich_interval_min,
    "audit_gc": settings.audit_gc_minutes,
    "log_gc": settings.log_gc_minutes,
}

# Sane bounds for an admin-set interval (minutes).
MIN_INTERVAL = 1
MAX_INTERVAL = 1440  # one day


def load_intervals(db) -> dict[str, int]:
    """Effective per-job intervals: the env/config defaults overlaid with any admin overrides
    saved in scheduler_settings."""
    iv = dict(JOB_INTERVALS)
    for row in db.execute(select(SchedulerSetting)).scalars():
        if row.job_id in iv and row.interval_minutes:
            iv[row.job_id] = row.interval_minutes
    return iv


def apply_interval(db, job_id: str, minutes: int) -> int:
    """Persist a job's interval override and reschedule the live job immediately (if the
    scheduler runs in this process). Returns the applied value."""
    if job_id not in JOB_INTERVALS:
        raise KeyError(job_id)
    minutes = max(MIN_INTERVAL, min(MAX_INTERVAL, int(minutes)))
    row = db.get(SchedulerSetting, job_id)
    if row is None:
        db.add(SchedulerSetting(job_id=job_id, interval_minutes=minutes))
    else:
        row.interval_minutes = minutes
    db.commit()
    if _scheduler is not None and _scheduler.get_job(job_id) is not None:
        _scheduler.reschedule_job(job_id, trigger="interval", minutes=minutes)
        log.info("rescheduled job %s to every %s min", job_id, minutes)
    return minutes


def _now_iso() -> str:
    return datetime.now(timezone.utc).isoformat()


def _record(name: str, **fields) -> None:
    with _activity_lock:
        _activity.setdefault(name, {}).update(fields)


def _notify_done(
    db, actor_id: int, job_id: str, status: str, summary: str | None, result: dict | None = None
) -> None:
    """Post a completion notice to the triggering admin's inbox (manual runs only). Best
    effort — a notification failure must never mask the job's own outcome."""
    try:
        create_notification(
            db,
            user_id=actor_id,
            type="scheduler",
            title=f"{job_id}: {status}",  # English fallback; UI renders translated from data
            body=summary,
            data={"job_id": job_id, "status": status, "summary": summary, "result": result},
        )
    except Exception:
        db.rollback()
        log.exception("failed to write completion notification for job %s", job_id)


def _run_changed(result) -> bool:
    """Did an automatic job run actually do something worth an audit row? (No-op polls don't.)
    A "change" = a truthy NUMERIC count; status-string dicts like {"skipped": "disabled"} (Plex
    off — the default) or {"skipped": "no demo account"} are no-ops, not changes, so they don't
    flood the trail every interval."""
    if result is None:
        return False
    if isinstance(result, dict):
        # A phased job (backfill's recent/deep, people_enrich's bios/portraits) nests a result per
        # phase: a count inside a phase is a change too.
        return any(
            _run_changed(v) if isinstance(v, dict) else isinstance(v, (int, float)) and bool(v)
            for v in result.values()
        )
    if isinstance(result, (int, float)):
        return result != 0
    return bool(result)


def _audit_run(db, actor_id, job_id, status, summary, result) -> None:
    """Write ONE run-summary audit row per job run (never per-item). Manual runs (an admin
    clicked "run now" → actor_id set) and errors are always logged; automatic runs only when
    they changed something, so routine no-op polls don't spam the trail. Best-effort."""
    if actor_id is None and status == "ok" and not _run_changed(result):
        return
    # The result as data too (E30): the audit log names it in the reader's language, as the inbox
    # does; `summary` stays the English line of the row's own summary column.
    data = _result_data(result)
    try:
        audit.record(
            db,
            actor_id,
            AuditAction.JOB_RUN,
            target_type="job",
            target_id=job_id,
            summary=f"{job_id}: {status}" + (f" — {summary}" if summary else ""),
            after={"status": status, "summary": summary, **({"result": data} if data else {})},
        )
        db.commit()
    except Exception:
        db.rollback()
        log.exception("failed to write audit row for job %s", job_id)


def _job(name: str, fn) -> None:
    # The one place both run paths converge (manual trigger AND scheduled fire), so claim the run
    # HERE, atomically: whichever reaches this first wins, the other skips. This is the real
    # single-run guard — trigger_job's pre-check is only a fast best-effort answer for the UI.
    if not _claim_running(name):
        log.info("job %s already running — skipping this trigger", name)
        return
    db = SessionLocal()
    actor_id = _manual_actor.get()  # set only for a manual "run now" trigger
    _record(
        name, last_started=_now_iso(), last_error=None, progress=None
    )  # running already claimed

    def sink(current, total, phase):
        _record(name, progress={"current": current, "total": total, "phase": phase})

    try:
        if is_sync_paused(db):
            log.info("job %s skipped (sync paused)", name)
            _record(
                name,
                status="skipped",
                last_finished=_now_iso(),
                last_result="paused",
                result=_PAUSED,
                progress=None,
            )
            if actor_id is not None:
                _notify_done(db, actor_id, name, "skipped", "sync paused", _PAUSED)
                _audit_run(db, actor_id, name, "skipped", "sync paused", _PAUSED)
            return
        with progress.bind(sink):
            result = fn(db)
        summary = _summarize(result)
        data = _result_data(result)
        log.info("job %s -> %s", name, result)
        _record(
            name,
            status="ok",
            last_finished=_now_iso(),
            last_result=summary,
            result=data,
            progress=None,
        )
        if actor_id is not None:
            _notify_done(db, actor_id, name, "ok", summary, data)
        _audit_run(db, actor_id, name, "ok", summary, result)
    except Exception as exc:
        db.rollback()
        log.exception("job %s failed", name, extra=logs.event("scheduler.job.failed", job=name))
        err = str(exc) or exc.__class__.__name__
        _record(
            name,
            status="error",
            last_finished=_now_iso(),
            last_error=err,
            result=None,
            progress=None,
        )
        if actor_id is not None:
            _notify_done(db, actor_id, name, "error", err)
        _audit_run(db, actor_id, name, "error", err, None)
    finally:
        # ALWAYS release the claim, even if the body raised before its own status update — so a
        # failure can never wedge the job as permanently "running".
        _record(name, running=False)
        db.close()


# A skipped run's result, in the same shape as a job's own `{"skipped": …}`.
_PAUSED = {"skipped": "sync paused"}


def _result_data(result) -> dict | None:
    """A job's result as the client names it in the reader's language (E30): the dict itself (counts,
    a `skipped` / `reason` word, a nested dict per phase), `enrich`'s bare count as one. The one-line
    `_summarize` is the English fallback."""
    if isinstance(result, dict):
        return result or None
    if isinstance(result, int) and not isinstance(result, bool):
        return {"enriched": result}
    return None


def _summarize(result) -> str | None:
    """Compact one-line rendering of a job's return value for the dashboard."""
    if result is None:
        return None
    if isinstance(result, dict):
        return ", ".join(f"{k}={v}" for k, v in result.items()) or None
    return str(result)


def scheduler_snapshot() -> dict:
    """What the scheduler is doing, for the admin dashboard. `running_here` is False in
    environments where the scheduler is disabled (e.g. local dev shares the prod-ish DB but
    runs no scheduler), so the UI can say so while still showing DB-derived queue/quota."""
    running_here = _scheduler is not None
    next_runs: dict[str, str | None] = {}
    live_intervals: dict[str, int] = {}
    if _scheduler is not None:
        for job in _scheduler.get_jobs():
            nr = getattr(job, "next_run_time", None)
            next_runs[job.id] = nr.astimezone(timezone.utc).isoformat() if nr else None
            iv = getattr(job.trigger, "interval", None)
            if iv is not None:
                live_intervals[job.id] = round(iv.total_seconds() / 60)
    with _activity_lock:
        acts = {k: dict(v) for k, v in _activity.items()}
    jobs = []
    for job_id, default_interval in JOB_INTERVALS.items():
        a = acts.get(job_id, {})
        jobs.append(
            {
                "id": job_id,
                "interval_minutes": live_intervals.get(job_id, default_interval),
                "next_run": next_runs.get(job_id),
                "running": a.get("running", False),
                "status": a.get("status"),
                "last_started": a.get("last_started"),
                "last_finished": a.get("last_finished"),
                "last_result": a.get("last_result"),
                "result": a.get("result"),
                "last_error": a.get("last_error"),
                "progress": a.get("progress"),
            }
        )
    return {"running_here": running_here, "enabled": settings.scheduler_enabled, "jobs": jobs}


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


def _enrich_job() -> None:
    def work(db):
        with quota.attribute(None, quota.QuotaAction.VIDEOS_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, quota.QuotaAction.VIDEOS_BACKFILL_RECENT):
            recent = run_recent_backfill(db, max_channels=25)
        with quota.attribute(None, quota.QuotaAction.VIDEOS_BACKFILL_FULL):
            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, quota.QuotaAction.SUBSCRIPTIONS_RESYNC):
            return run_subscription_resync(db)

    _job("subscriptions", work)


def _playlist_sync_job() -> None:
    # Mirror each read-scope user's YouTube playlists; per-user quota attribution is
    # handled inside sync_all_playlists.
    _job("playlist_sync", sync_all_playlists)


def _maintenance_job() -> None:
    def work(db):
        with quota.attribute(None, quota.QuotaAction.MAINTENANCE_REVALIDATE):
            return run_maintenance(db)

    _job("maintenance", work)


def _demo_reset_job() -> None:
    # Auto-clean the shared demo sandbox. No-op (and never creates one) if there's no demo account.
    def work(db):
        demo = db.query(User).filter(User.is_demo.is_(True)).one_or_none()
        if demo is None:
            return {"skipped": "no demo account"}
        return {"playlists_seeded": reset_demo_state(db, demo)}

    _job("demo_reset", work)


def _explore_cleanup_job() -> None:
    # Reclaim un-kept ephemeral discovery content (explored channels + live-search results) after
    # their grace periods. 0 quota (pure DB), so no quota attribution.
    _job("explore_cleanup", purge_ephemeral)


def _download_gc_job() -> None:
    # Retention GC for the download center: pre-expiry warnings, TTL deletion, LRU eviction.
    # Pure disk/DB work (no YouTube quota).
    _job("download_gc", run_download_gc)


def _plex_sync_job() -> None:
    # Mirror the enabled Plex library sections into the local catalog. Pure metadata (no YouTube
    # quota); a no-op when the Plex module is disabled.
    _job("plex_sync", run_plex_sync)


def _plex_watch_sync_job() -> None:
    # Incremental Plex→Siftlode watch-state pull (history + on-deck) + re-push of unsynced local
    # states. Cheap; a no-op when Plex is disabled or no owner link has sync enabled.
    _job("plex_watch_sync", run_plex_watch_sync)


def _plex_watch_reconcile_job() -> None:
    # Full watch-state reconcile (whole-section rescan) — catches un-watches the incremental feed
    # can't. Heavier, so it runs far less often (daily default).
    _job("plex_watch_reconcile", run_plex_watch_reconcile)


def _tmdb_enrich_job() -> None:
    # Enrich a batch of mirrored Plex titles from TMDB (keywords/crew/budget/revenue/TMDb score/
    # recommendations/backdrops), matched by guid. External metadata (no YouTube quota); a no-op when
    # the TMDB key is unset or the Plex catalog is empty.
    _job("tmdb_enrich", run_tmdb_enrich)


def _people_enrich_job() -> None:
    # Make the people local (E30 S6): a batch of person bios from TMDB and a batch of portraits resized
    # through Plex, so a person page and a cast strip never wait on an outside service.
    _job("people_enrich", run_people_enrich)


def _audit_gc_job() -> None:
    # Prune audit-log rows older than the admin-set retention (audit_retention_days; 0 = keep
    # forever). Pure DB, no quota.
    def work(db):
        from app import sysconfig

        return {"reaped": audit.gc(db, sysconfig.get_int(db, "audit_retention_days"))}

    _job("audit_gc", work)


def _log_gc_job() -> None:
    # Prune log_events rows older than log_retention_days — the limit the privacy policy states, so
    # this runs hourly rather than daily. Pure DB, no quota.
    def work(db):
        from app import sysconfig

        days = sysconfig.get_int(db, "log_retention_days")
        reaped = logsink.gc(db, days)
        # Debug sessions (E16) go with the rows they bundle; an open one past its cap is closed.
        closed = debugsession.sweep(db)
        sessions = debugsession.gc(db, days)
        db.commit()
        return {"reaped": reaped, "sessions_closed": closed, "sessions_reaped": sessions}

    _job("log_gc", work)


# job_id -> wrapper. The single source of truth for which jobs exist and how to run one,
# shared by start_scheduler (recurring registration) and trigger_job (manual "run now").
JOB_FUNCS: dict[str, Callable[[], None]] = {
    "rss_poll": _rss_job,
    "enrich": _enrich_job,
    "backfill": _backfill_job,
    "autotag": _autotag_job,
    "shorts": _shorts_job,
    "subscriptions": _subscriptions_job,
    "playlist_sync": _playlist_sync_job,
    "maintenance": _maintenance_job,
    "demo_reset": _demo_reset_job,
    "explore_cleanup": _explore_cleanup_job,
    "download_gc": _download_gc_job,
    "plex_sync": _plex_sync_job,
    "plex_watch_sync": _plex_watch_sync_job,
    "plex_watch_reconcile": _plex_watch_reconcile_job,
    "tmdb_enrich": _tmdb_enrich_job,
    "people_enrich": _people_enrich_job,
    "audit_gc": _audit_gc_job,
    "log_gc": _log_gc_job,
}


def _is_running(job_id: str) -> bool:
    with _activity_lock:
        return bool(_activity.get(job_id, {}).get("running"))


def _claim_running(job_id: str) -> bool:
    """Atomically claim a job as running: True if it was idle (and marks it running now), False if
    it was already running. Holding the lock across BOTH the check and the set is what makes the
    single-run guard race-free — called once at the top of _job so a manual trigger and a scheduled
    fire (or two rapid triggers) converge here and exactly one wins. Released in _job's finally."""
    with _activity_lock:
        if _activity.get(job_id, {}).get("running"):
            return False
        _activity.setdefault(job_id, {})["running"] = True
        return True


def running_job_ids() -> set[str]:
    """Ids of jobs executing right now (any trigger). Lets non-admin surfaces — e.g. the
    header's sync indicator — reflect real activity instead of just pending-work counts."""
    with _activity_lock:
        return {jid for jid, a in _activity.items() if a.get("running")}


def trigger_job(job_id: str, actor_id: int | None = None) -> str:
    """Run a job immediately in a background thread, independent of its interval schedule.
    The wrapper handles its own DB session, activity tracking and pause-skip, so the live
    dashboard reflects the run. Returns "started", "already_running", or raises KeyError for
    an unknown job. Refusing a concurrent run keeps a manual trigger from overlapping a
    scheduled run (APScheduler enforces max_instances=1 for the scheduled side).

    `actor_id` (the admin who clicked "run now") is stashed in a contextvar inside the new
    thread so the run posts a completion notification to that admin's inbox."""
    fn = JOB_FUNCS.get(job_id)
    if fn is None:
        raise KeyError(job_id)
    # Best-effort fast answer for the UI; the authoritative single-run guard is _job's atomic claim
    # (so a manual trigger and a scheduled fire can't both run). A race here just means the loser's
    # _job claims, sees it's taken, and skips — no double run.
    if _is_running(job_id):
        return "already_running"

    def run() -> None:
        if actor_id is not None:
            _manual_actor.set(actor_id)
        fn()

    threading.Thread(target=run, name=f"trigger-{job_id}", daemon=True).start()
    return "started"


def trigger_all(actor_id: int | None = None) -> list[str]:
    """Trigger every job that isn't already running; returns the ids actually started."""
    started = []
    for job_id in JOB_FUNCS:
        if trigger_job(job_id, actor_id) == "started":
            started.append(job_id)
    return started


def start_scheduler() -> None:
    global _scheduler
    if not settings.scheduler_enabled or _scheduler is not None:
        return
    # Effective intervals = env defaults overlaid with any admin overrides from the DB.
    db = SessionLocal()
    try:
        iv = load_intervals(db)
    finally:
        db.close()
    scheduler = BackgroundScheduler(timezone="UTC")
    for job_id, fn in JOB_FUNCS.items():
        scheduler.add_job(fn, "interval", minutes=iv[job_id], id=job_id)
    scheduler.start()
    _scheduler = scheduler
    log.info("scheduler started")


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