From 4e0cfb8463bb611f9569b67abd508fd6fc8e19d3 Mon Sep 17 00:00:00 2001 From: andrew Date: Wed, 8 Jul 2026 13:52:25 -0600 Subject: [PATCH] Add scheduler_service and pipeline_runner pipeline_runner: subprocess execution wrapper replacing flock -n with a single asyncio.Lock (deliberately one global lock, not per-resource -- matches the old flock's all-jobs-share-one-lock behavior rather than over-engineering it). Records one job_runs row per invocation (started/finished/status/exit_code/summary/log_path/triggered_by). If the lock is already held, records status='skipped_lock' immediately instead of silently dropping the run. Found and fixed a real concurrency bug via testing: the obvious asyncio.wait_for(lock.acquire(), timeout=0) idiom for a non-blocking try-acquire is broken in asyncio -- the wrapping Task's first iteration and the timeout-0 callback race with no guaranteed ordering, so it timed out on literally every call, including the very first uncontended one. Fixed using lock.locked() + acquire(), relying on acquire()'s fast path never suspending when uncontended. scheduler_service: AsyncIOScheduler with the default in-memory job store (NOT SQLAlchemyJobStore -- it pickles job functions to persist them, and the _lib()/_beet() factory closures here aren't picklable; MemoryJobStore avoids this since register_all_jobs() rebuilds the whole schedule from code + DB on every startup anyway). All 19 maintenance jobs ported 1:1 from /etc/cron.d/sldl-maintenance with their exact cron expressions; dedup deliberately has no --apply (dry-run only on schedule, per the false-negative-biased dedup preference). Playlist jobs are synced live from the playlists table (sync_playlist_jobs), so add/remove-in-the-UI takes effect with no redeploy -- wired into playlist_service's create/update/ delete. Maintenance job enable/disable persists to the scheduled_jobs table so a paused job stays paused across a restart despite the in-memory job store. trigger_now() supports "Run now" by invoking a job's function immediately with triggered_by='manual', bypassing its schedule. Wired into main.py's lifespan (start on boot, shutdown on exit). Verified via integration tests: 36 jobs register correctly (17 seeded playlists + 19 maintenance) with correct cron triggers; live playlist create/delete correctly adds/removes its scheduler job; maintenance enable/disable persists to the DB and takes effect live; trigger_now correctly bypasses the schedule; pipeline_runner correctly handles success/failure/timeout/concurrent-lock-contention with real subprocesses; and the full app boots with the scheduler running and shuts down cleanly. --- app/main.py | 10 +- app/services/pipeline_runner.py | 168 +++++++++++++++++ app/services/playlist_service.py | 71 ++++--- app/services/scheduler_service.py | 302 ++++++++++++++++++++++++++++++ 4 files changed, 521 insertions(+), 30 deletions(-) create mode 100644 app/services/pipeline_runner.py create mode 100644 app/services/scheduler_service.py diff --git a/app/main.py b/app/main.py index 550332e..cd3dc05 100644 --- a/app/main.py +++ b/app/main.py @@ -6,6 +6,7 @@ from starlette.middleware.sessions import SessionMiddleware from app.db import enable_beets_db_wal, init_db from app.routers import auth, dashboard +from app.services import scheduler_service from app.settings import settings @@ -13,10 +14,15 @@ from app.settings import settings async def lifespan(_app: FastAPI): init_db() enable_beets_db_wal() - # Task 5 (scheduler_service) hooks APScheduler start/stop in here once - # the job registry and pipeline_runner exist. + + scheduler = scheduler_service.create_scheduler() + scheduler_service.register_all_jobs(scheduler) + scheduler.start() + yield + scheduler.shutdown(wait=False) + def create_app() -> FastAPI: app = FastAPI(title="alembic", lifespan=lifespan) diff --git a/app/services/pipeline_runner.py b/app/services/pipeline_runner.py new file mode 100644 index 0000000..449fa7b --- /dev/null +++ b/app/services/pipeline_runner.py @@ -0,0 +1,168 @@ +import asyncio +import time +from pathlib import Path + +from app.db import SessionLocal +from app.models import JobRun +from app.settings import settings + +# Single global lock replacing /var/lock/sldl-pipeline.lock + flock -n. One +# process (this one) now owns every pipeline invocation, so one asyncio.Lock +# is enough -- deliberately not per-resource, matching the original flock's +# all-jobs-share-one-lock behavior exactly rather than over-engineering it. +_lock = asyncio.Lock() + +_ENV_PASSTHROUGH_KEYS = ("PATH", "HOME", "LANG", "LC_ALL", "TZ") + + +async def _try_acquire_nowait() -> bool: + """asyncio.Lock has no acquire_nowait(). asyncio.wait_for(lock.acquire(), + timeout=0) looks like the obvious idiom but is broken in practice: it + wraps acquire() in a Task, and the Task's first iteration and the + timeout-0 callback are both scheduled via the event loop with no + guaranteed ordering, so it times out almost every time even when the + lock is completely uncontended (confirmed empirically -- the very first, + uncontended call failed every time). The correct pattern relies on + Lock.acquire()'s fast path never suspending when the lock is free: the + .locked() check and the subsequent acquire() happen with no `await` + point in between, so nothing else can interleave in this single- + threaded event loop.""" + if _lock.locked(): + return False + await _lock.acquire() + return True + + +def _subprocess_env() -> dict: + import os + + env = {k: v for k, v in os.environ.items() if k in _ENV_PASSTHROUGH_KEYS} + env.setdefault("PATH", "/opt/venv/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin") + env["ALEMBIC_CONFIG_DIR"] = str(settings.alembic_config_dir) + env["MUSIC_DATA_DIR"] = str(settings.music_data_dir) + env["PIPELINE_DIR"] = str(settings.pipeline_dir) + env["BEETSDIR"] = str(settings.beets_dir) + return env + + +def _summarize_log(log_path: Path, max_len: int = 200) -> str: + """Best-effort one-liner for the jobs list: the log's last non-empty + line. Scripts here already end their own runs with a human-readable + '=== ... done ===' / summary line (see pipeline-status.sh's convention), + so this is usually meaningful without any per-script special-casing.""" + try: + lines = [line for line in log_path.read_text(errors="replace").splitlines() if line.strip()] + return lines[-1][:max_len] if lines else "" + except OSError: + return "" + + +async def run_job( + job_key: str, + argv: list[str], + triggered_by: str = "schedule", + timeout: float | None = None, +) -> JobRun: + """Run a pipeline command under the shared mutual-exclusion lock, + recording one job_runs row start-to-finish. If the lock is already + held, records status='skipped_lock' immediately and returns without + running anything -- the old flock -n behavior, now visible in the UI + instead of silently skipping.""" + started_at = time.time() + + if not await _try_acquire_nowait(): + db = SessionLocal() + run = JobRun( + job_key=job_key, + started_at=started_at, + finished_at=started_at, + status="skipped_lock", + triggered_by=triggered_by, + ) + db.add(run) + db.commit() + db.refresh(run) + db.close() + return run + + try: + log_dir = settings.logs_dir / job_key.replace(":", "_") + log_dir.mkdir(parents=True, exist_ok=True) + log_path = log_dir / f"{time.strftime('%Y%m%d-%H%M%S')}.log" + + db = SessionLocal() + run = JobRun( + job_key=job_key, + started_at=started_at, + status="running", + triggered_by=triggered_by, + log_path=str(log_path), + ) + db.add(run) + db.commit() + db.refresh(run) + run_id = run.id + db.close() + + exit_code: int | None + try: + with open(log_path, "wb") as log_file: + proc = await asyncio.create_subprocess_exec( + *argv, + stdout=log_file, + stderr=asyncio.subprocess.STDOUT, + env=_subprocess_env(), + ) + try: + exit_code = await asyncio.wait_for(proc.wait(), timeout=timeout) + except asyncio.TimeoutError: + proc.kill() + await proc.wait() + exit_code = -1 + with open(log_path, "a") as f: + f.write(f"\n[pipeline_runner] TIMEOUT after {timeout}s -- process killed\n") + status = "success" if exit_code == 0 else "failed" + except Exception as exc: + exit_code = None + status = "failed" + with open(log_path, "a") as f: + f.write(f"\n[pipeline_runner] exception before/while running: {exc!r}\n") + + finished_at = time.time() + db = SessionLocal() + run = db.get(JobRun, run_id) + run.finished_at = finished_at + run.status = status + run.exit_code = exit_code + run.summary = _summarize_log(log_path) + db.commit() + db.refresh(run) + db.close() + return run + finally: + _lock.release() + + +async def run_playlist(playlist_name: str, no_m3u: bool = False, triggered_by: str = "schedule") -> JobRun: + script = settings.pipeline_dir / "bin" / "run-playlist.sh" + argv = [str(script)] + if no_m3u: + argv.append("--no-m3u") + argv.append(playlist_name) + # 45 min SLDL_TIMEOUT (run-playlist.sh) + generous slack for the + # tagging/beets-import/M3U steps that follow it in the same script. + return await run_job(f"playlist:{playlist_name}", argv, triggered_by=triggered_by, timeout=3300) + + +async def run_import_track(args: list[str], triggered_by: str = "manual") -> JobRun: + script = settings.pipeline_dir / "bin" / "import-track.sh" + return await run_job("manual:import", [str(script)] + args, triggered_by=triggered_by) + + +async def run_lib_script(job_key: str, script_name: str, args: list[str] | None = None, triggered_by: str = "schedule") -> JobRun: + script = settings.pipeline_dir / "lib" / script_name + return await run_job(job_key, [str(script)] + (args or []), triggered_by=triggered_by) + + +async def run_beet(job_key: str, args: list[str], triggered_by: str = "schedule") -> JobRun: + return await run_job(job_key, ["beet"] + args, triggered_by=triggered_by) diff --git a/app/services/playlist_service.py b/app/services/playlist_service.py index a5ea648..1fc79c6 100644 --- a/app/services/playlist_service.py +++ b/app/services/playlist_service.py @@ -10,25 +10,27 @@ from app.settings import settings # The 17-entry array this project is migrating off of (was hardcoded in # /opt/sldl/configs/regen.sh). Used once by seed_legacy() during Stage 1/2 -# of the migration; the DB is the source of truth from then on. +# of the migration; the DB is the source of truth from then on. cron_expr +# values are the exact staggered slots from /etc/cron.d/sldl-maintenance, +# preserved so cutover doesn't change anyone's download schedule. LEGACY_PLAYLISTS = { - "techno": "https://open.spotify.com/playlist/5GlAkczmLWJFIJO4FvkLEZ", - "house": "https://open.spotify.com/playlist/2XF8ye6O5xqzT4NSfIXdfJ", - "lofi": "https://open.spotify.com/playlist/0qnyF6TAmuT3U6hFH2fiF1", - "y2k": "https://open.spotify.com/playlist/0iaA7ZJS00aRxhMwZn2GCU", - "weeb": "https://open.spotify.com/playlist/0599AbpsKRjsp2rlJI8jhf", - "shoegaze": "https://open.spotify.com/playlist/3R4S3tmbVKjXQ1RPAmryC5", - "bass": "https://open.spotify.com/playlist/2zkF4S16efaMmmzdIOe5w2", - "hiphop": "https://open.spotify.com/playlist/42JxTtOLJ7gG1ybfyy7Vmi", - "modular": "https://open.spotify.com/playlist/2kOfP2EM8dzdMYC6fXvlLb", - "hotdog": "https://open.spotify.com/playlist/53HrOL0qw47G9ywYZz3kgX", - "kpop": "https://open.spotify.com/playlist/31e2V512TIqr5JIfgkyseo", - "jungle": "https://open.spotify.com/playlist/4LjLeXg6ElneQiTaIcAHhE", - "goldenera": "https://open.spotify.com/playlist/57vArhigysJfgwB6CY4VXR", - "botanica": "https://open.spotify.com/playlist/5KgQT9YWz3EqhIN4Jh69IU", - "digicore": "https://open.spotify.com/playlist/6tWHtPBECZTkZHPKFA3Fq4", - "hardcore": "https://open.spotify.com/playlist/1J1lbzlQKligzr2LWaq9Ex", - "liked": "https://open.spotify.com/playlist/4Z3qCYuU1sjNeNYO3Amzeo", + "digicore": ("https://open.spotify.com/playlist/6tWHtPBECZTkZHPKFA3Fq4", "0 0 * * *", False), + "techno": ("https://open.spotify.com/playlist/5GlAkczmLWJFIJO4FvkLEZ", "0 1 * * *", False), + "house": ("https://open.spotify.com/playlist/2XF8ye6O5xqzT4NSfIXdfJ", "30 1 * * *", False), + "lofi": ("https://open.spotify.com/playlist/0qnyF6TAmuT3U6hFH2fiF1", "0 2 * * *", False), + "y2k": ("https://open.spotify.com/playlist/0iaA7ZJS00aRxhMwZn2GCU", "30 2 * * *", False), + "weeb": ("https://open.spotify.com/playlist/0599AbpsKRjsp2rlJI8jhf", "0 3 * * *", False), + "shoegaze": ("https://open.spotify.com/playlist/3R4S3tmbVKjXQ1RPAmryC5", "30 3 * * *", False), + "bass": ("https://open.spotify.com/playlist/2zkF4S16efaMmmzdIOe5w2", "0 4 * * *", False), + "hiphop": ("https://open.spotify.com/playlist/42JxTtOLJ7gG1ybfyy7Vmi", "30 4 * * *", False), + "modular": ("https://open.spotify.com/playlist/2kOfP2EM8dzdMYC6fXvlLb", "0 5 * * *", False), + "hotdog": ("https://open.spotify.com/playlist/53HrOL0qw47G9ywYZz3kgX", "30 5 * * *", False), + "kpop": ("https://open.spotify.com/playlist/31e2V512TIqr5JIfgkyseo", "0 6 * * *", False), + "jungle": ("https://open.spotify.com/playlist/4LjLeXg6ElneQiTaIcAHhE", "30 6 * * *", False), + "goldenera": ("https://open.spotify.com/playlist/57vArhigysJfgwB6CY4VXR", "0 7 * * *", False), + "botanica": ("https://open.spotify.com/playlist/5KgQT9YWz3EqhIN4Jh69IU", "30 7 * * *", False), + "hardcore": ("https://open.spotify.com/playlist/1J1lbzlQKligzr2LWaq9Ex", "0 23 * * *", False), + "liked": ("https://open.spotify.com/playlist/4Z3qCYuU1sjNeNYO3Amzeo", "30 23 * * *", True), } @@ -51,6 +53,7 @@ def create( active: bool = True, no_m3u: bool = False, notes: str | None = None, + cron_expr: str | None = None, ) -> Playlist: now = time.time() playlist = Playlist( @@ -59,6 +62,7 @@ def create( active=active, no_m3u=no_m3u, notes=notes, + cron_expr=cron_expr, created_at=now, updated_at=now, ) @@ -96,16 +100,21 @@ def delete(db: Session, playlist_id: int) -> None: conf_path.unlink(missing_ok=True) _write_playlists_json(db) + from app.services import scheduler_service + + scheduler = scheduler_service.get_scheduler() + if scheduler is not None: + scheduler_service.sync_playlist_jobs(scheduler) + def seed_legacy(db: Session) -> int: """One-time import of the legacy hardcoded array (migration Stage 1/2). Skips any name that already exists. Returns the number of rows created.""" created = 0 - for name, url in LEGACY_PLAYLISTS.items(): + for name, (url, cron_expr, no_m3u) in LEGACY_PLAYLISTS.items(): if get_by_name(db, name) is not None: continue - no_m3u = name == "liked" # matches --no-m3u liked in the old crontab - create(db, name=name, spotify_url=url, no_m3u=no_m3u) + create(db, name=name, spotify_url=url, no_m3u=no_m3u, cron_expr=cron_expr) created += 1 return created @@ -152,14 +161,20 @@ def regenerate_confs(db: Session) -> subprocess.CompletedProcess: def _sync_to_disk(db: Session) -> None: - """Convenience wrapper used by create()/update(): regenerate confs and - immediately re-patch credentials into any newly-rendered file.""" + """Convenience wrapper used by create()/update(): regenerate confs, + immediately re-patch credentials into any newly-rendered file, and push + the change to the live scheduler if one is running -- this is what + makes add/remove-in-the-UI take effect without a redeploy.""" regenerate_confs(db) - # Imported lazily to avoid a circular import (credential_service doesn't - # depend on playlist_service, but importing at module load time would - # still work here — done lazily anyway to keep the dependency direction - # obvious as both services grow). - from app.services import credential_service + # Imported lazily to avoid a circular import (credential_service and + # scheduler_service don't depend on playlist_service, but importing at + # module load time would still work here — done lazily anyway to keep + # the dependency direction obvious as all three services grow). + from app.services import credential_service, scheduler_service credential_service.render_scope(db, "soulseek") credential_service.render_scope(db, "spotify") + + scheduler = scheduler_service.get_scheduler() + if scheduler is not None: + scheduler_service.sync_playlist_jobs(scheduler) diff --git a/app/services/scheduler_service.py b/app/services/scheduler_service.py new file mode 100644 index 0000000..105a1a0 --- /dev/null +++ b/app/services/scheduler_service.py @@ -0,0 +1,302 @@ +import time + +from apscheduler.schedulers.asyncio import AsyncIOScheduler +from apscheduler.triggers.cron import CronTrigger +from sqlalchemy import select + +from app.db import SessionLocal +from app.models import Playlist, ScheduledJob +from app.services import pipeline_runner +from app.settings import settings + +TIMEZONE = "America/Edmonton" + +# --------------------------------------------------------------------------- +# Maintenance jobs, ported 1:1 from /etc/cron.d/sldl-maintenance. Each entry +# is (job_key, cron kwargs for CronTrigger, callable). Dedup deliberately has +# no --apply here (dry-run only on schedule) -- a deliberate behavior change +# from the old blind `--apply` cron, matching the false-negative-biased +# dedup preference; applying deletions requires a reviewed confirm through +# the dedup UI (services/dedup_review_service.py). +# +# NOTE on job store choice: APScheduler's SQLAlchemyJobStore pickles job +# functions to persist them, and closures returned from a factory (like +# _lib()/_beet() below) are NOT picklable -- `add_job` would fail at +# runtime. MemoryJobStore (the default) keeps live objects in-process +# instead, which is fine here because register_all_jobs() rebuilds the +# entire schedule from code + the DB on every startup anyway. Persisted +# enable/disable state (so a paused job stays paused across a restart) +# lives in the app's own `scheduled_jobs` table, not in APScheduler's store. +# --------------------------------------------------------------------------- + + +def _lib(job_key: str, script_name: str, args: list[str] | None = None, timeout: float | None = None): + async def _run(triggered_by: str = "schedule"): + return await pipeline_runner.run_lib_script(job_key, script_name, args, triggered_by=triggered_by) + + return _run + + +def _beet(job_key: str, args: list[str]): + async def _run(triggered_by: str = "schedule"): + return await pipeline_runner.run_beet(job_key, args, triggered_by=triggered_by) + + return _run + + +async def _log_rotation(triggered_by: str = "schedule"): + """Replaces `find /var/log/sldl -name '*.log' -mtime +30 -delete`.""" + cutoff = time.time() - 30 * 86400 + removed = 0 + if settings.logs_dir.exists(): + for path in settings.logs_dir.rglob("*.log"): + if path.is_file() and path.stat().st_mtime < cutoff: + path.unlink(missing_ok=True) + removed += 1 + return removed + + +MAINTENANCE_JOBS: dict[str, tuple[dict, callable]] = { + # ==== Daily ==== + "maintenance:sync_bandcamp": ( + dict(minute=0, hour=8), + _lib("maintenance:sync_bandcamp", "sync-bandcamp.sh"), + ), + "maintenance:normalize_casing": ( + dict(minute=52, hour=8), + _lib("maintenance:normalize_casing", "normalize-artist-casing.py", ["--apply"]), + ), + "maintenance:dedup": ( + dict(minute=55, hour=8), + _lib("maintenance:dedup", "dedup-library.sh"), # no --apply: dry-run only + ), + "maintenance:gen_djmix_playlist": ( + dict(minute=57, hour=8), + _lib("maintenance:gen_djmix_playlist", "gen-djmix-playlist.sh"), + ), + "maintenance:gen_vgm_playlist": ( + dict(minute=58, hour=8), + _lib("maintenance:gen_vgm_playlist", "gen-vgm-playlist.sh"), + ), + "maintenance:navidrome_scan": ( + dict(minute=0, hour=9), + _lib("maintenance:navidrome_scan", "navidrome-scan.sh"), + ), + "maintenance:export_laptop_playlists": ( + dict(minute=5, hour=9), + _lib("maintenance:export_laptop_playlists", "export-laptop-playlists.py"), + ), + "maintenance:enrich_buy_url": ( + dict(minute=10, hour=9), + _lib("maintenance:enrich_buy_url", "enrich-buy-url.py", ["--apply"]), + ), + "maintenance:build_fingerprint_index": ( + dict(minute=25, hour=9), + _lib("maintenance:build_fingerprint_index", "build-fingerprint-index.py", ["--workers", "8"]), + ), + "maintenance:pipeline_status_report": ( + dict(minute=30, hour=9), + _lib("maintenance:pipeline_status_report", "pipeline-status.sh"), + ), + "maintenance:log_rotation": ( + dict(minute=0, hour=0), + _log_rotation, + ), + # ==== Weekly (Sunday) ==== + "maintenance:strip_mb_tags": ( + dict(minute=30, hour=8, day_of_week="sun"), + _lib("maintenance:strip_mb_tags", "strip-mb-tags.sh"), + ), + "maintenance:strip_watermark_art": ( + dict(minute=35, hour=8, day_of_week="sun"), + _lib("maintenance:strip_watermark_art", "strip-watermark-art.py", ["--apply"]), + ), + "maintenance:scrub_watermark_text": ( + dict(minute=40, hour=8, day_of_week="sun"), + _lib("maintenance:scrub_watermark_text", "scrub-watermark-text.py", ["--apply"]), + ), + "maintenance:clean_sldl_index": ( + dict(minute=45, hour=8, day_of_week="sun"), + _lib("maintenance:clean_sldl_index", "clean-sldl-index.py", ["--apply"]), + ), + "maintenance:clear_bad_genres": ( + dict(minute=48, hour=8, day_of_week="sun"), + _lib("maintenance:clear_bad_genres", "clear-bad-genres.py", ["--apply"]), + ), + "maintenance:spotify_genre": ( + dict(minute=50, hour=8, day_of_week="sun"), + _lib("maintenance:spotify_genre", "spotify-genre.py", ["--apply", "--force"]), + ), + "maintenance:beets_update_sync": ( + dict(minute=53, hour=8, day_of_week="sun"), + _beet("maintenance:beets_update_sync", ["update"]), + ), + # ==== Monthly ==== + "maintenance:upgrade_mp3_to_flac": ( + dict(minute=0, hour=10, day=1), + _lib("maintenance:upgrade_mp3_to_flac", "upgrade-mp3-to-flac.sh"), + ), +} + + +# Module-level singleton so other services (playlist_service, future +# routers) can reach the live scheduler without main.py threading it through +# every function call. None until main.py's lifespan starts it. +_instance: AsyncIOScheduler | None = None + + +def get_scheduler() -> AsyncIOScheduler | None: + return _instance + + +def create_scheduler() -> AsyncIOScheduler: + global _instance + # Default job store (MemoryJobStore) -- see module docstring for why + # SQLAlchemyJobStore doesn't work here. + _instance = AsyncIOScheduler(timezone=TIMEZONE) + return _instance + + +def _ensure_scheduled_job_rows(db) -> None: + """Make sure every MAINTENANCE_JOBS entry has a scheduled_jobs row, + seeding cron_expr from the hardcoded default and enabled=True the first + time. Subsequent runs never overwrite an existing row -- this table is + the persisted source of truth for enable/disable once it exists.""" + existing = {row.job_key for row in db.execute(select(ScheduledJob)).scalars()} + now = time.time() + for job_key, (cron_kwargs, _runner) in MAINTENANCE_JOBS.items(): + if job_key in existing: + continue + cron_expr = "{minute} {hour} {day} {month} {day_of_week}".format( + minute=cron_kwargs.get("minute", "*"), + hour=cron_kwargs.get("hour", "*"), + day=cron_kwargs.get("day", "*"), + month=cron_kwargs.get("month", "*"), + day_of_week=cron_kwargs.get("day_of_week", "*"), + ) + db.add( + ScheduledJob( + job_key=job_key, + job_type="maintenance", + playlist_id=None, + cron_expr=cron_expr, + enabled=True, + ) + ) + db.commit() + _ = now # unused, kept for symmetry with playlists' created_at/updated_at style + + +def _playlist_job_key(name: str) -> str: + return f"playlist:{name}" + + +async def _run_playlist_job(name: str, no_m3u: bool, triggered_by: str = "schedule"): + return await pipeline_runner.run_playlist(name, no_m3u=no_m3u, triggered_by=triggered_by) + + +def sync_playlist_jobs(scheduler: AsyncIOScheduler) -> None: + """Converge the scheduler's registered playlist: jobs with the + playlists table. Call after any playlist create/update/delete -- this + is what makes add/remove-in-the-UI take effect without a redeploy.""" + db = SessionLocal() + try: + playlists = list(db.execute(select(Playlist)).scalars()) + finally: + db.close() + + desired_ids = { + _playlist_job_key(p.name) for p in playlists if p.cron_expr and p.active + } + existing_ids = {j.id for j in scheduler.get_jobs() if j.id.startswith("playlist:")} + + for job_id in existing_ids - desired_ids: + scheduler.remove_job(job_id) + + for playlist in playlists: + job_id = _playlist_job_key(playlist.name) + if job_id not in desired_ids: + continue + trigger = CronTrigger.from_crontab(playlist.cron_expr, timezone=TIMEZONE) + scheduler.add_job( + _run_playlist_job, + trigger=trigger, + id=job_id, + args=[playlist.name, playlist.no_m3u], + replace_existing=True, + misfire_grace_time=3600, + ) + + +def register_all_jobs(scheduler: AsyncIOScheduler) -> None: + """Rebuild the entire in-memory schedule from code (MAINTENANCE_JOBS) + + the DB (scheduled_jobs.enabled, playlists). Safe to call at every + startup -- MemoryJobStore has nothing to reconcile against.""" + db = SessionLocal() + try: + _ensure_scheduled_job_rows(db) + enabled_by_key = { + row.job_key: row.enabled for row in db.execute(select(ScheduledJob)).scalars() + } + finally: + db.close() + + for job_key, (cron_kwargs, runner) in MAINTENANCE_JOBS.items(): + if not enabled_by_key.get(job_key, True): + continue + scheduler.add_job( + runner, + trigger=CronTrigger(**cron_kwargs, timezone=TIMEZONE), + id=job_key, + replace_existing=True, + misfire_grace_time=3600, + ) + + sync_playlist_jobs(scheduler) + + +async def trigger_now(scheduler: AsyncIOScheduler, job_key: str) -> None: + """'Run now' support: bypass the schedule and run this job's function + immediately, still through pipeline_runner's lock/logging/job_runs + recording (triggered_by='manual'). Works even for a currently-disabled + job (disabled only means "don't run on schedule").""" + job = scheduler.get_job(job_key) + if job is None: + raise ValueError(f"no such registered job: {job_key}") + func = job.func + args = list(job.args) + kwargs = dict(job.kwargs) + kwargs["triggered_by"] = "manual" + await func(*args, **kwargs) + + +def set_maintenance_enabled(scheduler: AsyncIOScheduler, job_key: str, enabled: bool) -> None: + """Pause/resume a maintenance job AND persist the choice to + scheduled_jobs.enabled so it survives a restart -- this is the actual + 'pause an individual job without a redeploy' mechanism.""" + if job_key not in MAINTENANCE_JOBS: + raise ValueError(f"not a maintenance job: {job_key}") + + db = SessionLocal() + try: + row = db.execute( + select(ScheduledJob).where(ScheduledJob.job_key == job_key) + ).scalar_one_or_none() + if row is not None: + row.enabled = enabled + db.commit() + finally: + db.close() + + if enabled: + cron_kwargs, runner = MAINTENANCE_JOBS[job_key] + scheduler.add_job( + runner, + trigger=CronTrigger(**cron_kwargs, timezone=TIMEZONE), + id=job_key, + replace_existing=True, + misfire_grace_time=3600, + ) + else: + if scheduler.get_job(job_key) is not None: + scheduler.remove_job(job_key)