siftlode/backend/app/routes/scheduler.py

"""Admin Scheduler dashboard: a live view of what the background scheduler is doing —
per-job run activity (from the in-process scheduler), the work still queued (DB-derived),
and the shared quota picture."""
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy import and_, func, or_, select
from sqlalchemy.orm import Session

from app import quota, state
from app.config import settings
from app.db import get_db
from app.models import Channel, Subscription, User, Video
from app.routes.admin import admin_user
from app.scheduler import (
    JOB_INTERVALS,
    MAX_INTERVAL,
    MIN_INTERVAL,
    apply_interval,
    scheduler_snapshot,
    trigger_all,
    trigger_job,
)
from app.sync.runner import estimate_deep_backfill

router = APIRouter(prefix="/api/admin/scheduler", tags=["admin"])


@router.patch("/jobs/{job_id}")
def set_job_interval(
    job_id: str,
    payload: dict,
    _: User = Depends(admin_user),
    db: Session = Depends(get_db),
) -> dict:
    """Change how often a scheduler job runs (minutes). Persisted as an override and applied
    to the live scheduler immediately."""
    if job_id not in JOB_INTERVALS:
        raise HTTPException(status_code=404, detail="Unknown job")
    try:
        minutes = int(payload.get("interval_minutes"))
    except (TypeError, ValueError):
        raise HTTPException(status_code=400, detail="interval_minutes must be a number")
    if not (MIN_INTERVAL <= minutes <= MAX_INTERVAL):
        raise HTTPException(
            status_code=400,
            detail=f"interval must be between {MIN_INTERVAL} and {MAX_INTERVAL} minutes",
        )
    applied = apply_interval(db, job_id, minutes)
    return {"id": job_id, "interval_minutes": applied}


@router.post("/jobs/{job_id}/run")
def run_job_now(
    job_id: str,
    user: User = Depends(admin_user),
) -> dict:
    """Trigger a job to run immediately (in a background thread), independent of its
    interval. 409 if it's already running, 404 for an unknown job. On completion the
    triggering admin gets an inbox notification with the result."""
    try:
        result = trigger_job(job_id, actor_id=user.id)
    except KeyError:
        raise HTTPException(status_code=404, detail="Unknown job")
    if result == "already_running":
        raise HTTPException(status_code=409, detail="Job is already running")
    return {"id": job_id, "started": True}


@router.post("/run-all")
def run_all_now(user: User = Depends(admin_user)) -> dict:
    """Trigger every job that isn't already running. Returns the ids actually started."""
    return {"started": trigger_all(actor_id=user.id)}


@router.patch("/maintenance")
def set_maintenance_settings(
    payload: dict,
    _: User = Depends(admin_user),
    db: Session = Depends(get_db),
) -> dict:
    """Admin-tune the maintenance job's rolling re-validation batch (videos re-checked per
    run). Persisted to app_state; the job reads it each run."""
    try:
        value = int(payload.get("revalidate_batch"))
    except (TypeError, ValueError):
        raise HTTPException(status_code=400, detail="revalidate_batch must be a number")
    if not (state.MAINTENANCE_BATCH_MIN <= value <= state.MAINTENANCE_BATCH_MAX):
        raise HTTPException(
            status_code=400,
            detail=(
                f"revalidate_batch must be between {state.MAINTENANCE_BATCH_MIN} and "
                f"{state.MAINTENANCE_BATCH_MAX}"
            ),
        )
    return {"revalidate_batch": state.set_maintenance_batch(db, value)}


@router.get("")
def get_scheduler(
    _: User = Depends(admin_user), db: Session = Depends(get_db)
) -> dict:
    snap = scheduler_snapshot()

    # Work still queued (shared catalog, so these are instance-wide, not per-user).
    deep_requested = (
        select(Subscription.id)
        .where(
            Subscription.channel_id == Channel.id,
            Subscription.deep_requested.is_(True),
        )
        .exists()
    )
    deep_channels = (
        db.execute(
            select(Channel).where(Channel.backfill_done.is_(False), deep_requested)
        )
        .scalars()
        .all()
    )
    deep_eta = estimate_deep_backfill(db, deep_channels)

    queue = {
        "channels_recent_pending": db.scalar(
            select(func.count()).select_from(Channel).where(Channel.recent_synced_at.is_(None))
        )
        or 0,
        "channels_deep_pending": deep_eta["channels_pending"],
        "deep_eta_seconds": deep_eta["eta_seconds"],
        "videos_pending_enrich": db.scalar(
            select(func.count()).select_from(Video).where(Video.enriched_at.is_(None))
        )
        or 0,
        "videos_pending_shorts": db.scalar(
            select(func.count())
            .select_from(Video)
            .where(Video.shorts_probed.is_(False), Video.enriched_at.is_not(None))
        )
        or 0,
        "videos_live_refresh": db.scalar(
            select(func.count())
            .select_from(Video)
            .where(
                or_(
                    Video.live_status.in_(("live", "upcoming")),
                    and_(Video.live_status == "was_live", Video.duration_seconds.is_(None)),
                )
            )
        )
        or 0,
    }

    used = quota.units_used_today(db)
    quota_info = {
        "used_today": used,
        "remaining_today": quota.remaining_today(db),
        "daily_budget": settings.quota_daily_budget,
        "backfill_reserve": settings.backfill_quota_reserve,
    }

    return {
        **snap,
        "paused": state.is_sync_paused(db),
        "queue": queue,
        "quota": quota_info,
        "maintenance": {
            "revalidate_batch": state.get_maintenance_batch(db),
            "revalidate_batch_default": settings.maintenance_revalidate_batch,
            "min": state.MAINTENANCE_BATCH_MIN,
            "max": state.MAINTENANCE_BATCH_MAX,
        },
    }