From a339a73bd6c772e86815d4f5b8d86ebb9d24da89 Mon Sep 17 00:00:00 2001 From: npeter83 Date: Tue, 16 Jun 2026 14:38:51 +0200 Subject: [PATCH] feat(scheduler): admin dashboard endpoint + per-job activity tracking Record each scheduler job's run activity in-process (running / last result / error / timestamps) and expose a snapshot. New admin-only GET /api/admin/scheduler returns the per-job activity (with APScheduler next-run times), the DB-derived work still queued (channels/videos pending, deep ETA, live-refresh backlog) and the shared quota picture. --- backend/app/main.py | 15 +++++- backend/app/routes/scheduler.py | 86 +++++++++++++++++++++++++++++++++ backend/app/scheduler.py | 78 +++++++++++++++++++++++++++++- 3 files changed, 177 insertions(+), 2 deletions(-) create mode 100644 backend/app/routes/scheduler.py diff --git a/backend/app/main.py b/backend/app/main.py index 34fef9e..ca0b01d 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -26,7 +26,19 @@ from starlette.middleware.sessions import SessionMiddleware from app import auth from app.config import settings -from app.routes import admin, channels, feed, health, me, playlists, quota, sync, tags, version +from app.routes import ( + admin, + channels, + feed, + health, + me, + playlists, + quota, + scheduler as scheduler_routes, + sync, + tags, + version, +) from app.scheduler import shutdown_scheduler, start_scheduler @@ -68,6 +80,7 @@ app.include_router(me.router) app.include_router(channels.router) app.include_router(playlists.router) app.include_router(admin.router) +app.include_router(scheduler_routes.router) app.include_router(quota.router) app.include_router(version.router) diff --git a/backend/app/routes/scheduler.py b/backend/app/routes/scheduler.py new file mode 100644 index 0000000..effa4c6 --- /dev/null +++ b/backend/app/routes/scheduler.py @@ -0,0 +1,86 @@ +"""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 +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 scheduler_snapshot +from app.sync.runner import estimate_deep_backfill + +router = APIRouter(prefix="/api/admin/scheduler", tags=["admin"]) + + +@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, + } diff --git a/backend/app/scheduler.py b/backend/app/scheduler.py index 8d6d0c3..467910f 100644 --- a/backend/app/scheduler.py +++ b/backend/app/scheduler.py @@ -1,6 +1,8 @@ """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 +import threading +from datetime import datetime, timezone from apscheduler.schedulers.background import BackgroundScheduler @@ -23,22 +25,96 @@ logger = logging.getLogger("subfeed.scheduler") _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() + +# Each interval job's configured period, surfaced so the dashboard can show "every N min". +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, +} + + +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 _job(name: str, fn) -> None: db = SessionLocal() + _record(name, running=True, last_started=_now_iso(), last_error=None) try: if is_sync_paused(db): logger.info("job %s skipped (sync paused)", name) + _record(name, running=False, status="skipped", last_finished=_now_iso(), + last_result="paused") return result = fn(db) logger.info("job %s -> %s", name, result) - except Exception: + _record(name, running=False, status="ok", last_finished=_now_iso(), + last_result=_summarize(result)) + except Exception as exc: db.rollback() logger.exception("job %s failed", name) + _record(name, running=False, status="error", last_finished=_now_iso(), + last_error=str(exc) or exc.__class__.__name__) finally: db.close() +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] = {} + 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 + with _activity_lock: + acts = {k: dict(v) for k, v in _activity.items()} + jobs = [] + for job_id, interval in JOB_INTERVALS.items(): + a = acts.get(job_id, {}) + jobs.append( + { + "id": job_id, + "interval_minutes": 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"), + "last_error": a.get("last_error"), + } + ) + return {"running_here": running_here, "enabled": settings.scheduler_enabled, "jobs": jobs} + + def _rss_job() -> None: _job("rss_poll", lambda db: run_rss_poll(db))