"""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, sysconfig 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": sysconfig.get_int(db, "quota_daily_budget"), "backfill_reserve": sysconfig.get_int(db, "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, }, }