feat(scheduler): admin run-now/run-all triggers + live progress + completion notices

Add per-job "Run now" buttons and a "Start all now" button to the admin Scheduler
dashboard (admin-gated endpoints). Triggers run the job in a background thread
independent of its interval, refusing a concurrent run (409). While running, the
long jobs (maintenance, enrich, backfill, shorts) report live progress through a
decoupled contextvar sink, shown as a progress bar on the job row via the existing
4s poll. A manually-triggered run posts a completion notification to the triggering
admin's inbox (scheduled runs stay silent to avoid spam); the inbox renders the
"scheduler" type trilingually from type+data. While here, give the maintenance job
its missing dashboard label/description in all three languages.

backend/app/scheduler.py

@@ -1,16 +1,19 @@
 """Background scheduler: free RSS detection, enrichment of new videos, and quota-aware
 recent-first / deep backfill. Runs inside the API process (single worker)."""
+import contextvars
 import logging
 import threading
 from datetime import datetime, timezone
+from typing import Callable
 
 from apscheduler.schedulers.background import BackgroundScheduler
 from sqlalchemy import select
 
-from app import quota
+from app import progress, quota
 from app.config import settings
 from app.db import SessionLocal
 from app.models import SchedulerSetting
+from app.notifications import create_notification
 from app.state import is_sync_paused
 from app.sync.autotag import run_autotag_all
 from app.sync.maintenance import run_maintenance
@@ -35,6 +38,13 @@ _scheduler: BackgroundScheduler | None = None
 _activity: dict[str, dict] = {}
 _activity_lock = threading.Lock()
 
+# Set (per-thread) by a manual "run now" trigger to the id of the admin who clicked, so that
+# run — and only that run, not the recurring scheduled ones — posts a completion notification
+# to their inbox. Unset (None) for scheduled runs.
+_manual_actor: contextvars.ContextVar[int | None] = contextvars.ContextVar(
+    "manual_actor", default=None
+)
+
 # 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] = {
@@ -90,24 +100,55 @@ def _record(name: str, **fields) -> None:
         _activity.setdefault(name, {}).update(fields)
 
 
+def _notify_done(db, actor_id: int, job_id: str, status: str, summary: str | None) -> None:
+    """Post a completion notice to the triggering admin's inbox (manual runs only). Best
+    effort — a notification failure must never mask the job's own outcome."""
+    try:
+        create_notification(
+            db,
+            user_id=actor_id,
+            type="scheduler",
+            title=f"{job_id}: {status}",  # English fallback; UI renders translated from data
+            body=summary,
+            data={"job_id": job_id, "status": status, "summary": summary},
+        )
+    except Exception:
+        db.rollback()
+        logger.exception("failed to write completion notification for job %s", job_id)
+
+
 def _job(name: str, fn) -> None:
     db = SessionLocal()
-    _record(name, running=True, last_started=_now_iso(), last_error=None)
+    actor_id = _manual_actor.get()  # set only for a manual "run now" trigger
+    _record(name, running=True, last_started=_now_iso(), last_error=None, progress=None)
+
+    def sink(current, total, phase):
+        _record(name, progress={"current": current, "total": total, "phase": phase})
+
     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")
+                     last_result="paused", progress=None)
+            if actor_id is not None:
+                _notify_done(db, actor_id, name, "skipped", "sync paused")
             return
-        result = fn(db)
+        with progress.bind(sink):
+            result = fn(db)
+        summary = _summarize(result)
         logger.info("job %s -> %s", name, result)
         _record(name, running=False, status="ok", last_finished=_now_iso(),
-                last_result=_summarize(result))
+                last_result=summary, progress=None)
+        if actor_id is not None:
+            _notify_done(db, actor_id, name, "ok", summary)
     except Exception as exc:
         db.rollback()
         logger.exception("job %s failed", name)
+        err = str(exc) or exc.__class__.__name__
         _record(name, running=False, status="error", last_finished=_now_iso(),
-                last_error=str(exc) or exc.__class__.__name__)
+                last_error=err, progress=None)
+        if actor_id is not None:
+            _notify_done(db, actor_id, name, "error", err)
     finally:
         db.close()
 
@@ -151,6 +192,7 @@ def scheduler_snapshot() -> dict:
                 "last_finished": a.get("last_finished"),
                 "last_result": a.get("last_result"),
                 "last_error": a.get("last_error"),
+                "progress": a.get("progress"),
             }
         )
     return {"running_here": running_here, "enabled": settings.scheduler_enabled, "jobs": jobs}
@@ -211,6 +253,58 @@ def _maintenance_job() -> None:
     _job("maintenance", work)
 
 
+# job_id -> wrapper. The single source of truth for which jobs exist and how to run one,
+# shared by start_scheduler (recurring registration) and trigger_job (manual "run now").
+JOB_FUNCS: dict[str, Callable[[], None]] = {
+    "rss_poll": _rss_job,
+    "enrich": _enrich_job,
+    "backfill": _backfill_job,
+    "autotag": _autotag_job,
+    "shorts": _shorts_job,
+    "subscriptions": _subscriptions_job,
+    "playlist_sync": _playlist_sync_job,
+    "maintenance": _maintenance_job,
+}
+
+
+def _is_running(job_id: str) -> bool:
+    with _activity_lock:
+        return bool(_activity.get(job_id, {}).get("running"))
+
+
+def trigger_job(job_id: str, actor_id: int | None = None) -> str:
+    """Run a job immediately in a background thread, independent of its interval schedule.
+    The wrapper handles its own DB session, activity tracking and pause-skip, so the live
+    dashboard reflects the run. Returns "started", "already_running", or raises KeyError for
+    an unknown job. Refusing a concurrent run keeps a manual trigger from overlapping a
+    scheduled run (APScheduler enforces max_instances=1 for the scheduled side).
+
+    `actor_id` (the admin who clicked "run now") is stashed in a contextvar inside the new
+    thread so the run posts a completion notification to that admin's inbox."""
+    fn = JOB_FUNCS.get(job_id)
+    if fn is None:
+        raise KeyError(job_id)
+    if _is_running(job_id):
+        return "already_running"
+
+    def run() -> None:
+        if actor_id is not None:
+            _manual_actor.set(actor_id)
+        fn()
+
+    threading.Thread(target=run, name=f"trigger-{job_id}", daemon=True).start()
+    return "started"
+
+
+def trigger_all(actor_id: int | None = None) -> list[str]:
+    """Trigger every job that isn't already running; returns the ids actually started."""
+    started = []
+    for job_id in JOB_FUNCS:
+        if trigger_job(job_id, actor_id) == "started":
+            started.append(job_id)
+    return started
+
+
 def start_scheduler() -> None:
     global _scheduler
     if not settings.scheduler_enabled or _scheduler is not None:
@@ -221,18 +315,8 @@ def start_scheduler() -> None:
         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,
-        "maintenance": _maintenance_job,
-    }
     scheduler = BackgroundScheduler(timezone="UTC")
-    for job_id, fn in fns.items():
+    for job_id, fn in JOB_FUNCS.items():
         scheduler.add_job(fn, "interval", minutes=iv[job_id], id=job_id)
     scheduler.start()
     _scheduler = scheduler