diff --git a/backend/alembic/versions/0015_scheduler_settings.py b/backend/alembic/versions/0015_scheduler_settings.py new file mode 100644 index 0000000..5a7f590 --- /dev/null +++ b/backend/alembic/versions/0015_scheduler_settings.py @@ -0,0 +1,30 @@ +"""scheduler job interval overrides + +Revision ID: 0015_scheduler_settings +Revises: 0014_demo_account +Create Date: 2026-06-16 + +Adds scheduler_settings: a per-job run-interval override (minutes), editable from the admin +Scheduler dashboard. Absent row = use the env/config default, so this is purely additive. +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "0015_scheduler_settings" +down_revision: Union[str, None] = "0014_demo_account" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "scheduler_settings", + sa.Column("job_id", sa.String(length=40), primary_key=True), + sa.Column("interval_minutes", sa.Integer(), nullable=False), + ) + + +def downgrade() -> None: + op.drop_table("scheduler_settings") diff --git a/backend/app/models.py b/backend/app/models.py index 55f694c..f99e8ec 100644 --- a/backend/app/models.py +++ b/backend/app/models.py @@ -325,6 +325,17 @@ class AppState(Base): ) +class SchedulerSetting(Base): + """Admin override for a scheduler job's run interval (minutes). One row per job id; + absent = use the env/config default. DB-backed so it's editable from the admin UI at + runtime without a redeploy (env stays a fallback).""" + + __tablename__ = "scheduler_settings" + + job_id: Mapped[str] = mapped_column(String(40), primary_key=True) + interval_minutes: Mapped[int] = mapped_column(Integer) + + class Playlist(Base): """A per-user named, ordered collection of videos from the shared catalog. diff --git a/backend/app/routes/scheduler.py b/backend/app/routes/scheduler.py index effa4c6..11eeeea 100644 --- a/backend/app/routes/scheduler.py +++ b/backend/app/routes/scheduler.py @@ -1,7 +1,7 @@ """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 fastapi import APIRouter, Depends, HTTPException from sqlalchemy import and_, func, or_, select from sqlalchemy.orm import Session @@ -10,12 +10,36 @@ 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.scheduler import JOB_INTERVALS, MAX_INTERVAL, MIN_INTERVAL, apply_interval, scheduler_snapshot 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.get("") def get_scheduler( _: User = Depends(admin_user), db: Session = Depends(get_db) diff --git a/backend/app/scheduler.py b/backend/app/scheduler.py index 467910f..656db73 100644 --- a/backend/app/scheduler.py +++ b/backend/app/scheduler.py @@ -5,10 +5,12 @@ import threading from datetime import datetime, timezone from apscheduler.schedulers.background import BackgroundScheduler +from sqlalchemy import select from app import quota from app.config import settings from app.db import SessionLocal +from app.models import SchedulerSetting from app.state import is_sync_paused from app.sync.autotag import run_autotag_all from app.sync.playlists import sync_all_playlists @@ -32,7 +34,8 @@ _scheduler: BackgroundScheduler | None = None _activity: dict[str, dict] = {} _activity_lock = threading.Lock() -# Each interval job's configured period, surfaced so the dashboard can show "every N min". +# 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, @@ -43,6 +46,38 @@ JOB_INTERVALS: dict[str, int] = { "playlist_sync": settings.playlist_sync_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) + logger.info("rescheduled job %s to every %s min", job_id, minutes) + return minutes + def _now_iso() -> str: return datetime.now(timezone.utc).isoformat() @@ -90,19 +125,23 @@ def scheduler_snapshot() -> dict: 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, interval in JOB_INTERVALS.items(): + for job_id, default_interval in JOB_INTERVALS.items(): a = acts.get(job_id, {}) jobs.append( { "id": job_id, - "interval_minutes": interval, + "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"), @@ -166,43 +205,24 @@ 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() + fns = { + "rss_poll": _rss_job, + "enrich": _enrich_job, + "backfill": _backfill_job, + "autotag": _autotag_job, + "shorts": _shorts_job, + "subscriptions": _subscriptions_job, + "playlist_sync": _playlist_sync_job, + } 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.add_job( - _playlist_sync_job, - "interval", - minutes=settings.playlist_sync_minutes, - id="playlist_sync", - ) + for job_id, fn in fns.items(): + scheduler.add_job(fn, "interval", minutes=iv[job_id], id=job_id) scheduler.start() _scheduler = scheduler logger.info("scheduler started")