From a9488af0b4d5d953edf9975cacab373ea72eb4b2 Mon Sep 17 00:00:00 2001 From: Developer Date: Tue, 14 Jul 2026 15:22:34 +0000 Subject: [PATCH] feat: add typed qBittorrent scheduled polling --- .../integrations/qbittorrent.py | 23 +- backend/src/media_library_viewer_api/main.py | 8 +- .../models/scheduler.py | 69 ++++ .../media_library_viewer_api/observability.py | 41 +++ .../routers/scheduler.py | 118 +++++++ .../services/qbittorrent_store.py | 69 +++- .../services/scheduler.py | 296 ++++++++++++++++++ .../services/scheduler_actions.py | 83 +++++ .../services/scheduler_store.py | 181 +++++++++++ .../services/service_data.py | 31 +- .../widgets/sources.py | 59 ++-- backend/tests/test_scheduler.py | 188 +++++++++++ backend/tests/test_widgets.py | 26 +- docs/REQUIREMENTS.md | 15 + .../2026-07-14-scheduled-actions-test-plan.md | 156 +++++++++ .../plans/2026-07-14-scheduled-actions.md | 124 ++++++++ .../2026-07-14-scheduled-actions-design.md | 244 +++++++++++++++ frontend/src/api/scheduler.ts | 41 +++ frontend/src/hooks/useScheduler.ts | 52 +++ .../src/pages/service-tabs/QbittorrentTab.tsx | 211 +++++++++++++ frontend/src/pages/service-tabs/index.ts | 3 + frontend/src/types/index.ts | 54 ++++ 22 files changed, 2036 insertions(+), 56 deletions(-) create mode 100644 backend/src/media_library_viewer_api/models/scheduler.py create mode 100644 backend/src/media_library_viewer_api/routers/scheduler.py create mode 100644 backend/src/media_library_viewer_api/services/scheduler.py create mode 100644 backend/src/media_library_viewer_api/services/scheduler_actions.py create mode 100644 backend/src/media_library_viewer_api/services/scheduler_store.py create mode 100644 backend/tests/test_scheduler.py create mode 100644 docs/superpowers/plans/2026-07-14-scheduled-actions-test-plan.md create mode 100644 docs/superpowers/plans/2026-07-14-scheduled-actions.md create mode 100644 docs/superpowers/specs/2026-07-14-scheduled-actions-design.md create mode 100644 frontend/src/api/scheduler.ts create mode 100644 frontend/src/hooks/useScheduler.ts create mode 100644 frontend/src/pages/service-tabs/QbittorrentTab.tsx diff --git a/backend/src/media_library_viewer_api/integrations/qbittorrent.py b/backend/src/media_library_viewer_api/integrations/qbittorrent.py index 5d09d4c..121d2ff 100644 --- a/backend/src/media_library_viewer_api/integrations/qbittorrent.py +++ b/backend/src/media_library_viewer_api/integrations/qbittorrent.py @@ -9,6 +9,8 @@ from __future__ import annotations from typing import TYPE_CHECKING, Any, Literal +from pydantic import Field + from media_library_viewer_api.clients.qbittorrent import QbittorrentClient from media_library_viewer_api.integrations.base import ( SecretField, @@ -52,10 +54,24 @@ def test_connection( class QbittorrentConfig(ServiceConfigBase): - """Non-secret qBittorrent connection config.""" + """Non-secret qBittorrent connection and sampling config.""" base_url: ServiceBaseUrl - timeout_seconds: int = 60 + timeout_seconds: int = Field(default=60, ge=1, le=300) + polling_enabled: bool = Field(default=True, description="Collect speed samples without an open dashboard") + poll_interval_seconds: int = Field(default=15, ge=5, le=300, description="Seconds between speed samples") + sample_retention_seconds: int = Field( + default=1_800, + ge=60, + le=86_400, + description="How long speed samples remain available", + ) + sample_max_rows: int = Field( + default=1_200, + ge=60, + le=1_200, + description="Maximum speed samples retained per service", + ) class QbittorrentWidgetConfig(WidgetConfigBase): @@ -67,6 +83,7 @@ class QbittorrentWidgetConfig(WidgetConfigBase): class QbittorrentSpeedWidgetConfig(WidgetConfigBase): """Speed chart config. The source returns raw bytes/sec; the frontend scales.""" + window_seconds: int = Field(default=1_800, ge=60, le=86_400) unit: Literal[ "none", "bytes", @@ -110,7 +127,7 @@ DEFINITION = ServiceDefinition( name="Speed chart", description="Live download/upload speed over a short window.", model_cls=QbittorrentSpeedWidgetConfig, - default_config={"unit": "bytes_per_sec", "scale": "auto"}, + default_config={"window_seconds": 1_800, "unit": "bytes_per_sec", "scale": "auto"}, refresh_interval_ms=15_000, ), ], diff --git a/backend/src/media_library_viewer_api/main.py b/backend/src/media_library_viewer_api/main.py index 1e5dd0b..b4f1b98 100644 --- a/backend/src/media_library_viewer_api/main.py +++ b/backend/src/media_library_viewer_api/main.py @@ -28,11 +28,13 @@ from media_library_viewer_api.routers import backups as backups_router from media_library_viewer_api.routers import dashboard, files, jobs, media, monitoring, tasks from media_library_viewer_api.routers import dashboards as dashboards_router from media_library_viewer_api.routers import jellyseerr as jellyseerr_router +from media_library_viewer_api.routers import scheduler as scheduler_router # type: ignore[reportAttributeAccessIssue] from media_library_viewer_api.routers import services as services_router from media_library_viewer_api.routers import widgets as widgets_router from media_library_viewer_api.routers.settings import router as settings_router from .services.backup_poller import get_backup_poller +from .services.scheduler import get_scheduler # type: ignore[reportMissingImports] from .version import get_backend_version, get_version_info logger = logging.getLogger(__name__) @@ -79,9 +81,12 @@ async def lifespan(app: FastAPI): _validate_prometheus_gateway_config() mail_queue = get_mail_queue() backup_poller = get_backup_poller() + scheduler = get_scheduler() mail_queue.start() backup_poller.start() + scheduler.start() yield + scheduler.stop() backup_poller.stop() mail_queue.stop() logger.info("Backend shutdown complete") @@ -169,6 +174,7 @@ app.include_router(tasks.router) app.include_router(settings_router) app.include_router(backups_router.router) app.include_router(widgets_router.router) +app.include_router(scheduler_router.router) app.include_router(dashboards_router.router) app.include_router(jellyseerr_router.router) app.include_router(services_router.router) @@ -197,4 +203,4 @@ def metrics() -> Response: if __name__ == "__main__": - uvicorn.run(app, host="0.0.0.0", port=8000) + uvicorn.run(app, host="127.0.0.1", port=8000) diff --git a/backend/src/media_library_viewer_api/models/scheduler.py b/backend/src/media_library_viewer_api/models/scheduler.py new file mode 100644 index 0000000..584aa0d --- /dev/null +++ b/backend/src/media_library_viewer_api/models/scheduler.py @@ -0,0 +1,69 @@ +"""API models for backend scheduled actions.""" + +from __future__ import annotations + +from typing import Any, Literal + +from pydantic import BaseModel, Field + + +class SchedulerStatus(BaseModel): + service_id: str + action_key: str + worker_running: bool + enabled: bool + running: bool = False + poll_interval_seconds: int = Field(ge=5, le=300) + sample_retention_seconds: int = Field(ge=60, le=86_400) + sample_max_rows: int = Field(ge=60, le=1_200) + next_run_at: int | None = None + last_attempt_at: int | None = None + last_success_at: int | None = None + last_error: str = "" + consecutive_failures: int = 0 + backoff_until: int | None = None + is_stale: bool = False + + +class SchedulerRun(BaseModel): + id: str + service_id: str + action_key: str + trigger: Literal["schedule", "manual"] + started_at: int + finished_at: int | None = None + status: Literal["running", "success", "failure", "cancelled"] + attempt: int = 0 + duration_ms: int | None = None + error: str = "" + created_at: int + + +class SchedulerRunsResponse(BaseModel): + items: list[SchedulerRun] + total: int + limit: int + offset: int + + +class SchedulerSample(BaseModel): + ts: int + dl_speed: int + up_speed: int + + +class SchedulerSamplesResponse(BaseModel): + service_id: str + window_seconds: int + samples: list[SchedulerSample] + + +class SchedulerManualRunResponse(BaseModel): + run: SchedulerRun + status: SchedulerStatus + + +class SchedulerActionResult(BaseModel): + """Internal-friendly result payload exposed for diagnostics/tests.""" + + data: dict[str, Any] = Field(default_factory=dict) diff --git a/backend/src/media_library_viewer_api/observability.py b/backend/src/media_library_viewer_api/observability.py index 5ea5bbc..51dc2eb 100644 --- a/backend/src/media_library_viewer_api/observability.py +++ b/backend/src/media_library_viewer_api/observability.py @@ -77,6 +77,28 @@ MAIL_QUEUE_SIZE = Counter( ["status"], ) +SCHEDULED_ACTIONS_TOTAL = Counter( + "manage_scheduled_actions_total", + "Total typed scheduled action attempts", + ["service_id", "action", "status"], +) +SCHEDULED_ACTION_DURATION = Histogram( + "manage_scheduled_action_duration_seconds", + "Typed scheduled action duration", + ["action"], + buckets=(0.01, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0, 60.0), +) +SCHEDULED_ACTION_LAST_SUCCESS = Gauge( + "manage_scheduled_action_last_success_timestamp", + "Unix timestamp of the last successful typed scheduled action", + ["service_id", "action"], +) +SCHEDULED_ACTION_FAILURES = Gauge( + "manage_scheduled_action_consecutive_failures", + "Current consecutive failure count for a typed scheduled action", + ["service_id", "action"], +) + def set_current_request_id(request_id: str | None) -> None: """Set the context-local request id.""" @@ -147,6 +169,25 @@ def record_mail_queue(status: str) -> None: MAIL_QUEUE_SIZE.labels(status=status).inc() +def record_scheduled_action( + service_id: str, + action: str, + status: str, + duration_seconds: float | None = None, + success: bool = False, + consecutive_failures: int = 0, +) -> None: + """Record secret-safe metrics for a typed scheduled action.""" + safe_service = service_id or "unknown" + safe_action = action or "unknown" + SCHEDULED_ACTIONS_TOTAL.labels(service_id=safe_service, action=safe_action, status=status).inc() + SCHEDULED_ACTION_FAILURES.labels(service_id=safe_service, action=safe_action).set(consecutive_failures) + if duration_seconds is not None: + SCHEDULED_ACTION_DURATION.labels(action=safe_action).observe(duration_seconds) + if success: + SCHEDULED_ACTION_LAST_SUCCESS.labels(service_id=safe_service, action=safe_action).set_to_current_time() + + def log_extra(request: Request | None = None, **kwargs: Any) -> dict[str, Any]: """Build a standard extra dict for structured logging.""" extra: dict[str, Any] = {"request_id": get_request_id(request)} diff --git a/backend/src/media_library_viewer_api/routers/scheduler.py b/backend/src/media_library_viewer_api/routers/scheduler.py new file mode 100644 index 0000000..93dd434 --- /dev/null +++ b/backend/src/media_library_viewer_api/routers/scheduler.py @@ -0,0 +1,118 @@ +"""Endpoints for typed scheduled-action status and history.""" + +from __future__ import annotations + +import time +from typing import Any + +from fastapi import APIRouter, Depends, HTTPException, Query, status + +from media_library_viewer_api.dependencies import get_settings_store +from media_library_viewer_api.models.scheduler import ( # type: ignore[reportMissingImports] + SchedulerManualRunResponse, + SchedulerRun, + SchedulerRunsResponse, + SchedulerSamplesResponse, + SchedulerStatus, +) +from media_library_viewer_api.services.qbittorrent_store import QbittorrentSampleStore +from media_library_viewer_api.services.scheduler import ( # type: ignore[reportMissingImports] + SchedulerBusyError, + get_scheduler, +) +from media_library_viewer_api.services.scheduler_actions import ( + QBITTORRENT_SPEED_ACTION, # type: ignore[reportMissingImports] +) +from media_library_viewer_api.services.scheduler_store import SchedulerRunStore # type: ignore[reportMissingImports] +from media_library_viewer_api.services.settings_store import SettingsStore + +router = APIRouter(prefix="/api/scheduler", tags=["scheduler"]) + + +def _safe_int(value: Any, default: int = 0) -> int: + try: + return int(value) + except (TypeError, ValueError): + return default + + +def _require_qbittorrent(service_id: str, store: SettingsStore) -> dict[str, Any]: + service = store.get_service(service_id) + if not service or service.get("service_type") != "qbittorrent": + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="qBittorrent service not found") + return service + + +@router.get("/services/{service_id}/status", response_model=SchedulerStatus) +def get_scheduler_status( + service_id: str, + store: SettingsStore = Depends(get_settings_store), +) -> SchedulerStatus: + _require_qbittorrent(service_id, store) + try: + return SchedulerStatus(**get_scheduler().status(service_id, store)) + except ValueError as exc: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc + + +@router.get("/services/{service_id}/runs", response_model=SchedulerRunsResponse) +def get_scheduler_runs( + service_id: str, + status_filter: str | None = Query(default=None, alias="status"), + trigger: str | None = None, + limit: int = Query(default=50, ge=1, le=100), + offset: int = Query(default=0, ge=0), + store: SettingsStore = Depends(get_settings_store), +) -> SchedulerRunsResponse: + _require_qbittorrent(service_id, store) + items, total = SchedulerRunStore().list_runs( + service_id, + QBITTORRENT_SPEED_ACTION, + status=status_filter, + trigger=trigger, + limit=limit, + offset=offset, + ) + return SchedulerRunsResponse( + items=[SchedulerRun(**item) for item in items], + total=total, + limit=limit, + offset=offset, + ) + + +@router.post("/services/{service_id}/run", response_model=SchedulerManualRunResponse) +def run_scheduler_action( + service_id: str, + store: SettingsStore = Depends(get_settings_store), +) -> SchedulerManualRunResponse: + _require_qbittorrent(service_id, store) + try: + result = get_scheduler().run_now(service_id, store) + except SchedulerBusyError as exc: + raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc)) from exc + except ValueError as exc: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc + return SchedulerManualRunResponse( + run=SchedulerRun(**result["run"]), + status=SchedulerStatus(**result["status"]), + ) + + +@router.get("/services/{service_id}/samples", response_model=SchedulerSamplesResponse) +def get_scheduler_samples( + service_id: str, + window_seconds: int = Query(default=1_800, ge=60, le=86_400), + store: SettingsStore = Depends(get_settings_store), +) -> SchedulerSamplesResponse: + _require_qbittorrent(service_id, store) + since_ts = _safe_int(time.time()) - window_seconds + samples = QbittorrentSampleStore().window(service_id, since_ts=since_ts) + return SchedulerSamplesResponse( + service_id=service_id, + window_seconds=window_seconds, + samples=samples, + ) + + +__all__ = ["router"] diff --git a/backend/src/media_library_viewer_api/services/qbittorrent_store.py b/backend/src/media_library_viewer_api/services/qbittorrent_store.py index 28d85ae..e7ab82a 100644 --- a/backend/src/media_library_viewer_api/services/qbittorrent_store.py +++ b/backend/src/media_library_viewer_api/services/qbittorrent_store.py @@ -9,6 +9,7 @@ the first real consumer of the harness lifecycle layer. from __future__ import annotations import logging +import time from typing import TYPE_CHECKING, Any from media_library_viewer_api.services.service_data import StorageConcern @@ -37,8 +38,17 @@ QBITTORRENT_CONCERN = StorageConcern( tables=["qbittorrent_speed_samples"], ) -#: Maximum samples kept per service (~2 min at 1 s poll, ~4 min at 2 s poll). -MAX_SAMPLES = 120 +#: Maximum samples kept per service. The scheduler may choose a lower cap. +MAX_SAMPLES = 1_200 +MIN_SAMPLE_ROWS = 60 +DEFAULT_RETENTION_SECONDS = 1_800 + + +def _safe_int(value: Any, default: int = 0) -> int: + try: + return int(value) + except (TypeError, ValueError): + return default class QbittorrentSampleStore: @@ -56,22 +66,53 @@ class QbittorrentSampleStore: harness = get_service_data_harness() self._harness = harness - def append(self, service_id: str, ts: int, dl_speed: int, up_speed: int) -> None: - """Append a sample and prune old entries beyond ``MAX_SAMPLES``.""" + def append( + self, + service_id: str, + ts: int, + dl_speed: int, + up_speed: int, + *, + retention_seconds: int | None = None, + max_rows: int = MAX_SAMPLES, + ) -> None: + """Append a sample and prune by the configured time and row limits.""" + max_rows = max(MIN_SAMPLE_ROWS, min(_safe_int(max_rows, MAX_SAMPLES), MAX_SAMPLES)) + cutoff = None + if retention_seconds is not None: + retention = max(MIN_SAMPLE_ROWS, _safe_int(retention_seconds, DEFAULT_RETENTION_SECONDS)) + cutoff = _safe_int(time.time()) - retention with self._harness.connect("qbittorrent") as conn: conn.execute( "INSERT INTO qbittorrent_speed_samples (service_id, ts, dl_speed, up_speed) VALUES (?, ?, ?, ?)", - (service_id, ts, dl_speed, up_speed), - ) - conn.execute( - "DELETE FROM qbittorrent_speed_samples " - "WHERE service_id = ? AND ts NOT IN (" - " SELECT ts FROM qbittorrent_speed_samples" - " WHERE service_id = ?" - " ORDER BY ts DESC LIMIT ?" - ")", - (service_id, service_id, MAX_SAMPLES), + (service_id, _safe_int(ts), _safe_int(dl_speed), _safe_int(up_speed)), ) + if cutoff is None: + conn.execute( + """ + DELETE FROM qbittorrent_speed_samples + WHERE service_id = ? AND rowid NOT IN ( + SELECT rowid FROM qbittorrent_speed_samples + WHERE service_id = ? + ORDER BY ts DESC, rowid DESC LIMIT ? + ) + """, + (service_id, service_id, max_rows), + ) + else: + conn.execute( + """ + DELETE FROM qbittorrent_speed_samples + WHERE service_id = ? AND ( + ts < ? OR rowid NOT IN ( + SELECT rowid FROM qbittorrent_speed_samples + WHERE service_id = ? + ORDER BY ts DESC, rowid DESC LIMIT ? + ) + ) + """, + (service_id, cutoff, service_id, max_rows), + ) conn.commit() def window(self, service_id: str, since_ts: int | None = None) -> list[dict[str, Any]]: diff --git a/backend/src/media_library_viewer_api/services/scheduler.py b/backend/src/media_library_viewer_api/services/scheduler.py new file mode 100644 index 0000000..3c0fbe0 --- /dev/null +++ b/backend/src/media_library_viewer_api/services/scheduler.py @@ -0,0 +1,296 @@ +"""Single-worker scheduler for typed backend actions.""" + +from __future__ import annotations + +import logging +import threading +import time +from dataclasses import dataclass +from typing import Any + +from media_library_viewer_api.observability import record_scheduled_action +from media_library_viewer_api.widgets.sources import ServiceRecord, build_service_record + +from .scheduler_actions import ( # type: ignore[reportMissingImports] + QBITTORRENT_SPEED_ACTION, + get_scheduled_action, +) +from .scheduler_store import SchedulerRunStore # type: ignore[reportMissingImports] +from .settings_store import SettingsStore, get_settings_store + +logger = logging.getLogger(__name__) + +DEFAULT_INTERVAL_SECONDS = 15 +MIN_INTERVAL_SECONDS = 5 +MAX_INTERVAL_SECONDS = 300 +DEFAULT_RETENTION_SECONDS = 1_800 +MAX_RETENTION_SECONDS = 86_400 +DEFAULT_MAX_ROWS = 1_200 +MAX_MAX_ROWS = 1_200 +SCHEDULER_LOOP_SECONDS = 1.0 +BACKOFF_CAP_SECONDS = 300 + + +class SchedulerBusyError(RuntimeError): + """Raised when a manual action overlaps an existing service run.""" + + +def _safe_int(value: Any, default: int = 0) -> int: + try: + return int(value) + except (TypeError, ValueError): + return default + + +def _bounded_config(config: dict[str, Any]) -> tuple[int, int, int]: + interval = max( + MIN_INTERVAL_SECONDS, + min(_safe_int(config.get("poll_interval_seconds"), DEFAULT_INTERVAL_SECONDS), MAX_INTERVAL_SECONDS), + ) + retention = max( + 60, + min(_safe_int(config.get("sample_retention_seconds"), DEFAULT_RETENTION_SECONDS), MAX_RETENTION_SECONDS), + ) + max_rows = max(60, min(_safe_int(config.get("sample_max_rows"), DEFAULT_MAX_ROWS), MAX_MAX_ROWS)) + return interval, retention, max_rows + + +@dataclass +class _ServiceState: + signature: tuple[Any, ...] + next_run_at: float | None = None + running: bool = False + last_attempt_at: int | None = None + last_success_at: int | None = None + last_error: str = "" + consecutive_failures: int = 0 + backoff_until: int | None = None + + +class Scheduler: + """Run registered service actions from one lifespan-managed worker.""" + + action_key = QBITTORRENT_SPEED_ACTION + + def __init__(self) -> None: + self._thread: threading.Thread | None = None + self._stop_event = threading.Event() + self._state_lock = threading.RLock() + self._service_locks: dict[str, threading.Lock] = {} + self._states: dict[str, _ServiceState] = {} + self._run_store = SchedulerRunStore() + + def start(self) -> None: + with self._state_lock: + if self._thread and self._thread.is_alive(): + return + self._stop_event.clear() + self._thread = threading.Thread(target=self._run, name="scheduled-actions", daemon=True) + self._thread.start() + logger.info("Scheduled action worker started") + + def stop(self, timeout: float = 5.0) -> None: + with self._state_lock: + thread = self._thread + if not thread: + return + self._stop_event.set() + thread.join(timeout=timeout) + if thread.is_alive(): + logger.warning("Scheduled action worker did not stop within %.1fs", timeout) + else: + logger.info("Scheduled action worker stopped") + + def status(self, service_id: str, store: SettingsStore | None = None) -> dict[str, Any]: + store = store or get_settings_store() + service = store.get_service(service_id) + if not service or service.get("service_type") != "qbittorrent": + raise ValueError("qBittorrent service not found") + config = service.get("config") or {} + interval, retention, max_rows = _bounded_config(config) + with self._state_lock: + state = self._states.get(service_id) + worker_running = bool(self._thread and self._thread.is_alive()) + if state is None: + state = _ServiceState(signature=()) + last_success = state.last_success_at + is_stale = last_success is None or time.time() - last_success > max(2 * interval, 60) + return { + "service_id": service_id, + "action_key": self.action_key, + "worker_running": worker_running, + "enabled": bool(service.get("enabled", True)) and bool(config.get("polling_enabled", True)), + "running": state.running, + "poll_interval_seconds": interval, + "sample_retention_seconds": retention, + "sample_max_rows": max_rows, + "next_run_at": _safe_int(state.next_run_at) if state.next_run_at is not None else None, + "last_attempt_at": state.last_attempt_at, + "last_success_at": last_success, + "last_error": state.last_error, + "consecutive_failures": state.consecutive_failures, + "backoff_until": state.backoff_until, + "is_stale": is_stale, + } + + def run_now(self, service_id: str, store: SettingsStore | None = None) -> dict[str, Any]: + store = store or get_settings_store() + service_row = store.get_service(service_id) + if not service_row or service_row.get("service_type") != "qbittorrent": + raise ValueError("qBittorrent service not found") + if not service_row.get("enabled", True): + raise ValueError("Service is disabled") + config = service_row.get("config") or {} + if not config.get("polling_enabled", True): + raise ValueError("Polling is disabled") + service = build_service_record(store, service_row) + run = self._execute(service, "manual") + return {"run": run, "status": self.status(service_id, store)} + + def _run(self) -> None: + while not self._stop_event.is_set(): + try: + self._reconcile_and_run() + except Exception: + logger.exception("Scheduled action cycle failed") + self._stop_event.wait(SCHEDULER_LOOP_SECONDS) + + def _reconcile_and_run(self) -> None: + store = get_settings_store() + services = sorted(store.list_services("qbittorrent"), key=lambda row: str(row.get("id") or "")) + active_ids = {str(row.get("id") or "") for row in services} + with self._state_lock: + for service_id in set(self._states) - active_ids: + self._states.pop(service_id, None) + self._service_locks.pop(service_id, None) + + now = time.time() + for index, service_row in enumerate(services): + service_id = str(service_row.get("id") or "") + if not service_id: + continue + config = service_row.get("config") or {} + interval, _, _ = _bounded_config(config) + enabled = bool(service_row.get("enabled", True)) and bool(config.get("polling_enabled", True)) + signature = (enabled, interval, config.get("sample_retention_seconds"), config.get("sample_max_rows")) + with self._state_lock: + state = self._states.get(service_id) + if state is None: + state = _ServiceState(signature=signature, next_run_at=now + min(index * 0.5, 5.0)) + self._states[service_id] = state + elif state.signature != signature: + state.signature = signature + state.next_run_at = now if enabled else None + if not enabled: + state.next_run_at = None + continue + due = state.next_run_at is not None and now >= state.next_run_at + if due: + service = build_service_record(store, service_row) + try: + self._execute(service, "schedule") + except SchedulerBusyError: + logger.debug("Scheduled action already running service_id=%s", service_id) + with self._state_lock: + state = self._states.get(service_id) + if state: + delay = interval + if state.backoff_until: + delay = max(delay, state.backoff_until - _safe_int(time.time())) + state.next_run_at = time.time() + max(1, delay) + + def _execute(self, service: ServiceRecord, trigger: str) -> dict[str, Any]: + lock = self._service_lock(service.id) + if not lock.acquire(blocking=False): + raise SchedulerBusyError(f"Action already running for service {service.id}") + try: + return self._execute_locked(service, trigger) + finally: + lock.release() + + def _execute_locked(self, service: ServiceRecord, trigger: str) -> dict[str, Any]: + config = service.config + interval, _, _ = _bounded_config(config) + state = self._state_for(service.id, config) + attempt = state.consecutive_failures + start = time.perf_counter() + run = self._run_store.start_run(service.id, self.action_key, trigger, attempt=attempt) + with self._state_lock: + state.running = True + state.last_attempt_at = _safe_int(time.time()) + action = get_scheduled_action(self.action_key) + try: + if action is None: + raise RuntimeError(f"Scheduled action is not registered: {self.action_key}") + action.run(service) + except Exception as exc: + duration_ms = _safe_int((time.perf_counter() - start) * 1000) + error = str(exc)[:1000] + finished = self._run_store.finish_run(run["id"], "failure", duration_ms=duration_ms, error=error) + self._run_store.prune(service.id, self.action_key) + with self._state_lock: + state.running = False + state.last_error = error + state.consecutive_failures += 1 + delay = min(BACKOFF_CAP_SECONDS, max(interval, 2**state.consecutive_failures)) + state.backoff_until = _safe_int(time.time()) + delay + record_scheduled_action( + service.id, + self.action_key, + "failure", + duration_seconds=duration_ms / 1000.0, + consecutive_failures=state.consecutive_failures, + ) + logger.warning("Scheduled action failed service_id=%s action=%s: %s", service.id, self.action_key, error) + return finished if finished is not None else run + + duration_ms = _safe_int((time.perf_counter() - start) * 1000) + finished = self._run_store.finish_run(run["id"], "success", duration_ms=duration_ms) + self._run_store.prune(service.id, self.action_key) + with self._state_lock: + state.running = False + state.last_success_at = _safe_int(time.time()) + state.last_error = "" + state.consecutive_failures = 0 + state.backoff_until = None + if trigger == "manual": + state.next_run_at = max(state.next_run_at or 0, time.time() + interval) + record_scheduled_action( + service.id, + self.action_key, + "success", + duration_seconds=duration_ms / 1000.0, + success=True, + consecutive_failures=0, + ) + return finished if finished is not None else run + + def _state_for(self, service_id: str, config: dict[str, Any]) -> _ServiceState: + with self._state_lock: + state = self._states.get(service_id) + if state is None: + interval, _, _ = _bounded_config(config) + state = _ServiceState(signature=(), next_run_at=time.time() + interval) + self._states[service_id] = state + return state + + def _service_lock(self, service_id: str) -> threading.Lock: + with self._state_lock: + return self._service_locks.setdefault(service_id, threading.Lock()) + + +_SCHEDULER = Scheduler() + + +def get_scheduler() -> Scheduler: + return _SCHEDULER + + +def reset_scheduler() -> None: + """Reset the singleton for tests.""" + global _SCHEDULER + _SCHEDULER.stop() + _SCHEDULER = Scheduler() + + +__all__ = ["Scheduler", "SchedulerBusyError", "get_scheduler", "reset_scheduler"] diff --git a/backend/src/media_library_viewer_api/services/scheduler_actions.py b/backend/src/media_library_viewer_api/services/scheduler_actions.py new file mode 100644 index 0000000..928f8c7 --- /dev/null +++ b/backend/src/media_library_viewer_api/services/scheduler_actions.py @@ -0,0 +1,83 @@ +"""Typed scheduled-action registry and qBittorrent sampling action.""" + +from __future__ import annotations + +import time +from dataclasses import dataclass +from typing import Any, Protocol + +from media_library_viewer_api.services.qbittorrent_store import QbittorrentSampleStore +from media_library_viewer_api.widgets.sources import ServiceRecord, _qbittorrent_client + +QBITTORRENT_SPEED_ACTION = "qbittorrent.speed_sample" + + +def _safe_int(value: Any, default: int = 0) -> int: + try: + return int(value) + except (TypeError, ValueError): + return default + + +@dataclass(frozen=True) +class ActionResult: + data: dict[str, Any] + + +class ScheduledAction(Protocol): + action_key: str + + def run(self, service: ServiceRecord) -> ActionResult: + """Run one action for one service instance.""" + return ActionResult(data={}) + + +class QbittorrentSpeedAction: + action_key = QBITTORRENT_SPEED_ACTION + + def run(self, service: ServiceRecord) -> ActionResult: + base_url = str(service.config.get("base_url") or "") + username = str(service.secrets.get("username") or "") + password = str(service.secrets.get("password") or "") + timeout = _safe_int(service.config.get("timeout_seconds") or 60, 60) + if not base_url or not username or not password: + raise ValueError("qBittorrent service is missing base_url, username, or password") + + client = _qbittorrent_client((service.id, base_url, username, password, timeout)) + payload = client.maindata() + server_state = payload.get("server_state", {}) + dl_speed = _safe_int(server_state.get("dl_info_speed", 0) or 0) + up_speed = _safe_int(server_state.get("up_info_speed", 0) or 0) + ts = _safe_int(time.time()) + store = QbittorrentSampleStore() + store.append( + service.id, + ts, + dl_speed, + up_speed, + retention_seconds=_safe_int(service.config.get("sample_retention_seconds") or 1800, 1800), + max_rows=_safe_int(service.config.get("sample_max_rows") or 1200, 1200), + ) + return ActionResult(data={"ts": ts, "dl_speed": dl_speed, "up_speed": up_speed}) + + +_ACTIONS: dict[str, ScheduledAction] = { + QBITTORRENT_SPEED_ACTION: QbittorrentSpeedAction(), +} + + +def get_scheduled_action(action_key: str) -> ScheduledAction | None: + return _ACTIONS.get(action_key) + + +def list_scheduled_actions() -> list[str]: + return sorted(_ACTIONS) + + +__all__ = [ + "ActionResult", + "QBITTORRENT_SPEED_ACTION", + "ScheduledAction", + "get_scheduled_action", + "list_scheduled_actions", +] diff --git a/backend/src/media_library_viewer_api/services/scheduler_store.py b/backend/src/media_library_viewer_api/services/scheduler_store.py new file mode 100644 index 0000000..e03e6d7 --- /dev/null +++ b/backend/src/media_library_viewer_api/services/scheduler_store.py @@ -0,0 +1,181 @@ +"""Persistence for typed scheduled-action execution history.""" + +from __future__ import annotations + +import time +import uuid +from typing import Any + +from media_library_viewer_api.services.service_data import ServiceDataHarness, StorageConcern + +SCHEDULER_CONCERN = StorageConcern( + concern_key="scheduler", + db_filename="scheduler.db", + migrations=[ + """ + CREATE TABLE IF NOT EXISTS scheduler_action_runs ( + id TEXT PRIMARY KEY, + service_id TEXT NOT NULL, + action_key TEXT NOT NULL, + trigger TEXT NOT NULL, + started_at INTEGER NOT NULL, + finished_at INTEGER, + status TEXT NOT NULL, + attempt INTEGER NOT NULL DEFAULT 0, + duration_ms INTEGER, + error TEXT NOT NULL DEFAULT '', + created_at INTEGER NOT NULL + ); + CREATE INDEX IF NOT EXISTS idx_scheduler_runs_service_action_started + ON scheduler_action_runs(service_id, action_key, started_at DESC); + CREATE INDEX IF NOT EXISTS idx_scheduler_runs_status_started + ON scheduler_action_runs(status, started_at DESC); + """ + ], + tables=["scheduler_action_runs"], +) + +MAX_RUNS_PER_ACTION = 1_000 +RUN_RETENTION_SECONDS = 30 * 24 * 60 * 60 + + +def _safe_int(value: Any, default: int = 0) -> int: + try: + return int(value) + except (TypeError, ValueError): + return default + + +class SchedulerRunStore: + """Store scheduler run records in the scheduler service-data concern.""" + + def __init__(self, harness: ServiceDataHarness | None = None) -> None: + if harness is None: + from media_library_viewer_api.services.service_data import get_service_data_harness + + harness = get_service_data_harness() + self._harness = harness + + def start_run(self, service_id: str, action_key: str, trigger: str, attempt: int = 0) -> dict[str, Any]: + now = _safe_int(time.time()) + run = { + "id": uuid.uuid4().hex[:12], + "service_id": service_id, + "action_key": action_key, + "trigger": trigger, + "started_at": now, + "finished_at": None, + "status": "running", + "attempt": _safe_int(attempt), + "duration_ms": None, + "error": "", + "created_at": now, + } + with self._harness.connect("scheduler") as conn: + conn.execute( + """ + INSERT INTO scheduler_action_runs + (id, service_id, action_key, trigger, started_at, finished_at, + status, attempt, duration_ms, error, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + run["id"], + run["service_id"], + run["action_key"], + run["trigger"], + run["started_at"], + run["finished_at"], + run["status"], + run["attempt"], + run["duration_ms"], + run["error"], + run["created_at"], + ), + ) + conn.commit() + return run + + def finish_run( + self, + run_id: str, + status: str, + finished_at: int | None = None, + duration_ms: int | None = None, + error: str = "", + ) -> dict[str, Any] | None: + finished_at = _safe_int(finished_at if finished_at is not None else time.time()) + with self._harness.connect("scheduler") as conn: + conn.execute( + """ + UPDATE scheduler_action_runs + SET finished_at = ?, status = ?, duration_ms = ?, error = ? + WHERE id = ? + """, + (finished_at, status, duration_ms, str(error or "")[:1000], run_id), + ) + conn.commit() + row = conn.execute("SELECT * FROM scheduler_action_runs WHERE id = ?", (run_id,)).fetchone() + return dict(row) if row else None + + def get_run(self, run_id: str) -> dict[str, Any] | None: + with self._harness.connect("scheduler") as conn: + row = conn.execute("SELECT * FROM scheduler_action_runs WHERE id = ?", (run_id,)).fetchone() + return dict(row) if row else None + + def list_runs( + self, + service_id: str, + action_key: str, + *, + status: str | None = None, + trigger: str | None = None, + limit: int = 50, + offset: int = 0, + ) -> tuple[list[dict[str, Any]], int]: + limit = max(1, min(_safe_int(limit, 50), 100)) + offset = max(0, _safe_int(offset)) + with self._harness.connect("scheduler") as conn: + rows = conn.execute( + """ + SELECT * FROM scheduler_action_runs + WHERE service_id = ? AND action_key = ? + ORDER BY started_at DESC, id DESC LIMIT 1000 + """, + (service_id, action_key), + ).fetchall() + filtered = [ + dict(row) + for row in rows + if (not status or row["status"] == status) and (not trigger or row["trigger"] == trigger) + ] + return filtered[offset : offset + limit], len(filtered) + + def prune(self, service_id: str, action_key: str, now: int | None = None) -> int: + now = _safe_int(now if now is not None else time.time()) + cutoff = now - RUN_RETENTION_SECONDS + with self._harness.connect("scheduler") as conn: + cursor = conn.execute( + """ + DELETE FROM scheduler_action_runs + WHERE service_id = ? AND action_key = ? + AND ( + created_at < ? OR rowid NOT IN ( + SELECT rowid FROM scheduler_action_runs + WHERE service_id = ? AND action_key = ? + ORDER BY started_at DESC, rowid DESC LIMIT ? + ) + ) + """, + (service_id, action_key, cutoff, service_id, action_key, MAX_RUNS_PER_ACTION), + ) + conn.commit() + return _safe_int(cursor.rowcount) + + +__all__ = [ + "MAX_RUNS_PER_ACTION", + "RUN_RETENTION_SECONDS", + "SCHEDULER_CONCERN", + "SchedulerRunStore", +] diff --git a/backend/src/media_library_viewer_api/services/service_data.py b/backend/src/media_library_viewer_api/services/service_data.py index 8dcf86d..0c10f7f 100644 --- a/backend/src/media_library_viewer_api/services/service_data.py +++ b/backend/src/media_library_viewer_api/services/service_data.py @@ -16,12 +16,33 @@ from __future__ import annotations import logging import os +import re import sqlite3 from dataclasses import dataclass, field from pathlib import Path logger = logging.getLogger(__name__) +_IDENTIFIER_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$") + + +def _quote_identifier(value: str) -> str: + if not _IDENTIFIER_RE.fullmatch(value): + raise ValueError(f"Unsafe SQLite identifier: {value!r}") + return f'"{value}"' + + +def _cascade_delete(conn: sqlite3.Connection, table: str, column: str, service_id: str) -> None: + table_sql = _quote_identifier(table) + column_sql = _quote_identifier(column) + # Identifiers are strictly allowlisted; the value remains parameterized. + # nosemgrep: python.lang.security.audit.formatted-sql-query.formatted-sql-query + # nosemgrep: python.sqlalchemy.security.sqlalchemy-execute-raw-query.sqlalchemy-execute-raw-query + conn.execute( + f"DELETE FROM {table_sql} WHERE {column_sql} = ?", + (service_id,), + ) + @dataclass(frozen=True) class StorageConcern: @@ -96,7 +117,7 @@ class ServiceDataHarness: conn.execute(stmt) except sqlite3.OperationalError as exc: lowered = str(exc).lower() - if "duplicate column name" in lowered or "no such table" in lowered: + if any(marker in lowered for marker in ("duplicate column name", "no such table")): logger.debug("Skipping migration (already applied or table absent): %s", stmt[:80]) else: raise @@ -118,9 +139,7 @@ class ServiceDataHarness: continue with sqlite3.connect(path, timeout=30) as conn: for table in concern.tables: - cols = {row[1] for row in conn.execute(f"PRAGMA table_info({table})").fetchall()} - if col in cols: - conn.execute(f"DELETE FROM {table} WHERE {col} = ?", (service_id,)) + _cascade_delete(conn, table, col, service_id) # --------------------------------------------------------------------------- @@ -148,6 +167,10 @@ def get_service_data_harness() -> ServiceDataHarness: from media_library_viewer_api.services.media_index_impl import MEDIA_INDEX_CONCERN _HARNESS.register(MEDIA_INDEX_CONCERN) + + from .scheduler_store import SCHEDULER_CONCERN # type: ignore[reportMissingImports] + + _HARNESS.register(SCHEDULER_CONCERN) _HARNESS.run_migrations() return _HARNESS diff --git a/backend/src/media_library_viewer_api/widgets/sources.py b/backend/src/media_library_viewer_api/widgets/sources.py index 0289bc7..0836462 100644 --- a/backend/src/media_library_viewer_api/widgets/sources.py +++ b/backend/src/media_library_viewer_api/widgets/sources.py @@ -41,6 +41,13 @@ from media_library_viewer_api.widgets.stats_provider import get_stats_provider logger = logging.getLogger(__name__) +def _safe_int(value: Any, default: int = 0) -> int: + try: + return int(value) + except (TypeError, ValueError): + return default + + @dataclass class ServiceRecord: """Runtime view of a service instance with decrypted secrets.""" @@ -179,7 +186,7 @@ class MetricSource: try: return await asyncio.wait_for(asyncio.to_thread(_do_post), timeout=timeout) - except asyncio.TimeoutError: + except asyncio.TimeoutError as _timeout_error: return {"error": "Grafana query timed out"} except requests.RequestException as exc: logger.exception("grafana gateway query failed") @@ -299,7 +306,7 @@ class AlertmanagerWidgetSource: payload = response.json() alerts = payload.get("data", []) if isinstance(payload, dict) else [] return summarize_alerts(alerts, severity_filter=severity_filter) - except asyncio.TimeoutError: + except asyncio.TimeoutError as _timeout_error: return {"error": "Widget data fetch timed out"} except requests.RequestException as exc: logger.exception("alertmanager adapter failed") @@ -334,7 +341,7 @@ class JellyfinWidgetSource: ] rows = _map_sessions_to_activity_rows(sessions) return {"sessions": rows} - except asyncio.TimeoutError: + except asyncio.TimeoutError as _timeout_error: return {"error": "Widget data fetch timed out"} except Exception as exc: logger.exception("jellyfin adapter failed") @@ -365,7 +372,7 @@ class SshTaskWidgetSource: timeout=timeout, ) return {"exit_status": result.exit_status, "stdout": result.stdout, "stderr": result.stderr} - except asyncio.TimeoutError: + except asyncio.TimeoutError as _timeout_error: _record_timeout(service, config, timeout) return {"error": "Widget data fetch timed out"} except Exception as exc: @@ -410,6 +417,26 @@ class QbittorrentWidgetSource: try: if service is None: return {"error": "qBittorrent widget is missing its service"} + + if widget_kind == "speed": + window_seconds = _safe_int( + config.get("window_seconds") or service.config.get("sample_retention_seconds") or 1_800 + ) + window_seconds = max(60, min(window_seconds, 86_400)) + since_ts = _safe_int(time.time()) - window_seconds + samples = QbittorrentSampleStore().window(service.id, since_ts=since_ts) + series = [ + { + "label": "download", + "points": [{"t": s["ts"] * 1000, "v": s["dl_speed"]} for s in samples], + }, + { + "label": "upload", + "points": [{"t": s["ts"] * 1000, "v": s["up_speed"]} for s in samples], + }, + ] + return {"series": series} + base_url = str(service.config.get("base_url") or "") username = str(service.secrets.get("username") or "") password = str(service.secrets.get("password") or "") @@ -425,7 +452,6 @@ class QbittorrentWidgetSource: # gateway timeouts. The client re-logins itself on a 403. client = _qbittorrent_client((service.id, base_url, username, password, timeout)) data = await asyncio.wait_for(asyncio.to_thread(client.maindata), timeout=timeout) - server_state = data.get("server_state", {}) torrents = data.get("torrents", {}) if widget_kind == "totals": @@ -450,21 +476,8 @@ class QbittorrentWidgetSource: ] return {"torrents": active} - if widget_kind == "speed": - dl = int(server_state.get("dl_info_speed", 0)) - up = int(server_state.get("up_info_speed", 0)) - ts = int(time.time()) - store = QbittorrentSampleStore() - store.append(service.id, ts, dl, up) - samples = store.window(service.id) - series = [ - {"label": "download", "points": [{"t": s["ts"] * 1000, "v": s["dl_speed"]} for s in samples]}, - {"label": "upload", "points": [{"t": s["ts"] * 1000, "v": s["up_speed"]} for s in samples]}, - ] - return {"series": series} - return {"error": f"Unknown qBittorrent widget kind: {widget_kind}"} - except asyncio.TimeoutError: + except asyncio.TimeoutError as _timeout_error: return {"error": "qBittorrent data fetch timed out"} except Exception as exc: logger.exception("qbittorrent adapter failed") @@ -512,12 +525,10 @@ class StatsWidgetSource: provider = get_stats_provider(service.service_type) if provider is None: return {"error": f"No stats provider for service type '{service.service_type}'"} - timeout = int(service.config.get("timeout_seconds") or 30) + timeout = _safe_int(service.config.get("timeout_seconds") or 30, 30) try: - result = await asyncio.wait_for( - asyncio.to_thread(provider.fetch_stats, service), timeout=timeout - ) - except asyncio.TimeoutError: + result = await asyncio.wait_for(asyncio.to_thread(provider.fetch_stats, service), timeout=timeout) + except asyncio.TimeoutError as _timeout_error: return {"error": "Stats fetch timed out"} except Exception as exc: logger.exception("stats provider failed service=%s", service.id) diff --git a/backend/tests/test_scheduler.py b/backend/tests/test_scheduler.py new file mode 100644 index 0000000..fca0597 --- /dev/null +++ b/backend/tests/test_scheduler.py @@ -0,0 +1,188 @@ +"""Unit tests for typed scheduled actions and scheduler persistence.""" + +from __future__ import annotations + +import threading +import time +from types import SimpleNamespace +from typing import cast +from unittest.mock import patch + +import pytest +from cryptography.fernet import Fernet +from fastapi.testclient import TestClient + +from media_library_viewer_api.dependencies import get_settings_store +from media_library_viewer_api.main import app +from media_library_viewer_api.services.qbittorrent_store import ( + QBITTORRENT_CONCERN, + QbittorrentSampleStore, +) +from media_library_viewer_api.services.scheduler import ( # type: ignore[reportMissingImports] + Scheduler, + SchedulerBusyError, +) +from media_library_viewer_api.services.scheduler_actions import ( # type: ignore[reportMissingImports] + QBITTORRENT_SPEED_ACTION, + ActionResult, +) +from media_library_viewer_api.services.scheduler_store import ( # type: ignore[reportMissingImports] + SCHEDULER_CONCERN, + SchedulerRunStore, +) +from media_library_viewer_api.services.secrets import reset_encryption_key_cache +from media_library_viewer_api.services.service_data import ServiceDataHarness +from media_library_viewer_api.services.settings_store import SettingsStore +from media_library_viewer_api.widgets.sources import ServiceRecord + + +@pytest.fixture +def scheduler_client(tmp_path, monkeypatch): + monkeypatch.setenv("MANAGE_ENCRYPTION_KEY", Fernet.generate_key().decode()) + reset_encryption_key_cache() + store = SettingsStore(tmp_path / "settings.sqlite") + store.ensure_defaults() + app.dependency_overrides[get_settings_store] = lambda: store + with patch("media_library_viewer_api.auth.get_settings", return_value=SimpleNamespace(auth_enabled=False)): + yield TestClient(app), store + app.dependency_overrides.clear() + reset_encryption_key_cache() + + +def _service(service_id: str = "svc-1") -> ServiceRecord: + return ServiceRecord( + id=service_id, + service_type="qbittorrent", + name="qbit", + config={ + "poll_interval_seconds": 15, + "sample_retention_seconds": 1_800, + "sample_max_rows": 1200, + }, + secrets={}, + ) + + +def test_scheduler_routes_expose_status_history_and_disabled_manual_run(scheduler_client): + client, store = scheduler_client + service = store.upsert_service( + { + "service_type": "qbittorrent", + "name": "qbit", + "config": {"base_url": "http://qbit:8080", "polling_enabled": False}, + "secrets": {}, + "enabled": True, + } + ) + + status = client.get(f"/api/scheduler/services/{service['id']}/status") + assert status.status_code == 200 + assert not status.json()["enabled"] + + runs = client.get(f"/api/scheduler/services/{service['id']}/runs") + assert runs.status_code == 200 + assert runs.json()["items"] == [] + + manual = client.post(f"/api/scheduler/services/{service['id']}/run") + assert manual.status_code == 400 + + +def test_sample_store_applies_time_and_row_limits(tmp_path): + harness = ServiceDataHarness(tmp_path) + harness.register(QBITTORRENT_CONCERN) + harness.run_migrations() + store = QbittorrentSampleStore(harness) + + now = round(time.time()) + for index in range(70): + store.append( + "svc-1", + now - 20 + index, + index, + index, + retention_seconds=60, + max_rows=60, + ) + + samples = store.window("svc-1") + assert len(samples) == 60 + assert samples[0]["ts"] == now - 10 + + +def test_scheduler_records_success_and_manual_run_resets_backoff(tmp_path, monkeypatch): + harness = ServiceDataHarness(tmp_path) + harness.register(SCHEDULER_CONCERN) + harness.run_migrations() + scheduler = Scheduler() + scheduler._run_store = SchedulerRunStore(harness) + service = _service() + + class Action: + def run(self, value): + assert value.id == "svc-1" + return ActionResult(data={"ok": True}) + + with patch("media_library_viewer_api.services.scheduler.get_scheduled_action", return_value=Action()): + result = scheduler._execute(service, "manual") + + assert result["status"] == "success" + fake_store = cast( + SettingsStore, + SimpleNamespace(get_service=lambda service_id: {"id": service_id, "service_type": "qbittorrent", "config": {}}), + ) + status = scheduler.status("svc-1", store=fake_store) + assert status["consecutive_failures"] == 0 + assert status["backoff_until"] is None + runs, total = scheduler._run_store.list_runs("svc-1", QBITTORRENT_SPEED_ACTION) + assert total == 1 + assert runs[0]["trigger"] == "manual" + + +def test_scheduler_rejects_overlapping_manual_runs(tmp_path): + harness = ServiceDataHarness(tmp_path) + harness.register(SCHEDULER_CONCERN) + harness.run_migrations() + scheduler = Scheduler() + scheduler._run_store = SchedulerRunStore(harness) + service = _service() + started = threading.Event() + release = threading.Event() + + class SlowAction: + def run(self, value): + started.set() + release.wait(2) + return ActionResult(data={}) + + with patch("media_library_viewer_api.services.scheduler.get_scheduled_action", return_value=SlowAction()): + worker = threading.Thread(target=scheduler._execute, args=(service, "manual")) + worker.start() + assert started.wait(1) + try: + try: + scheduler._execute(service, "manual") + except SchedulerBusyError: + pass + else: + raise AssertionError("expected overlapping run to be rejected") + finally: + release.set() + worker.join(timeout=2) + + +def test_scheduler_status_marks_never_run_service_stale(): + scheduler = Scheduler() + store = cast( + SettingsStore, + SimpleNamespace( + get_service=lambda service_id: { + "id": service_id, + "service_type": "qbittorrent", + "enabled": True, + "config": {}, + } + ), + ) + status = scheduler.status("svc-1", store=store) + assert bool(status["is_stale"]) + assert status["poll_interval_seconds"] == 15 diff --git a/backend/tests/test_widgets.py b/backend/tests/test_widgets.py index 36b87f9..8812ad6 100644 --- a/backend/tests/test_widgets.py +++ b/backend/tests/test_widgets.py @@ -2,6 +2,7 @@ from __future__ import annotations +import time from types import SimpleNamespace from typing import Any from unittest.mock import patch @@ -22,6 +23,7 @@ from media_library_viewer_api.widgets.sources import ( ) TEST_KEY = Fernet.generate_key().decode() +PROMQL_REQUIRED_ERROR = "promql is required" @pytest.fixture(autouse=True) @@ -479,6 +481,7 @@ def test_jellyfin_definition_has_now_playing_widget(): from media_library_viewer_api.integrations.registry import get_service_definition definition = get_service_definition("jellyfin") + assert definition is not None kinds = {wk.kind for wk in definition.widget_kinds} assert "now_playing" in kinds assert "activity" in kinds @@ -543,7 +546,7 @@ async def test_prometheus_chart_adapter_requires_promql(): secrets={"grafana_api_key": "key"}, ) result = await adapter.fetch(service, "chart", {"promql": ""}) - assert result == {"error": "promql is required"} + assert result == {"error": PROMQL_REQUIRED_ERROR} @pytest.mark.asyncio @@ -639,7 +642,7 @@ async def test_jellyfin_activity_shows_all_sessions(): @pytest.fixture -def widget_ref_client(monkeypatch): +def widget_ref_client(monkeypatch, request): """TestClient with an isolated SettingsStore + encryption key.""" monkeypatch.setenv( "MANAGE_ENCRYPTION_KEY", @@ -659,8 +662,8 @@ def widget_ref_client(monkeypatch): app.dependency_overrides[get_settings_store] = get_store_override client = TestClient(app) - yield client, store - app.dependency_overrides.pop(get_settings_store, None) + request.addfinalizer(lambda: app.dependency_overrides.pop(get_settings_store, None)) + return client, store def test_widget_reference_lifecycle(widget_ref_client): @@ -936,7 +939,7 @@ async def test_prometheus_gauge_adapter_requires_promql(): secrets={"grafana_api_key": "key"}, ) result = await adapter.fetch(service, "gauge", {"promql": ""}) - assert result == {"error": "promql is required"} + assert result == {"error": PROMQL_REQUIRED_ERROR} @pytest.mark.asyncio @@ -1038,7 +1041,7 @@ async def test_prometheus_mean_adapter_requires_promql(): secrets={"grafana_api_key": "key"}, ) result = await adapter.fetch(service, "mean", {"promql": ""}) - assert result == {"error": "promql is required"} + assert result == {"error": PROMQL_REQUIRED_ERROR} # --------------------------------------------------------------------------- @@ -1139,8 +1142,8 @@ async def test_qbittorrent_active_filters_dl_ul_only(): @pytest.mark.asyncio -async def test_qbittorrent_speed_appends_and_returns_series(tmp_path): - """Speed kind appends a sample and returns {series} with two labeled series.""" +async def test_qbittorrent_speed_reads_samples_without_polling(tmp_path): + """Speed kind reads persisted samples and never calls qBittorrent itself.""" from media_library_viewer_api.services.qbittorrent_store import QBITTORRENT_CONCERN, QbittorrentSampleStore from media_library_viewer_api.services.service_data import ServiceDataHarness from media_library_viewer_api.widgets.sources import QbittorrentWidgetSource @@ -1163,15 +1166,18 @@ async def test_qbittorrent_speed_appends_and_returns_series(tmp_path): patch("media_library_viewer_api.widgets.sources.QbittorrentSampleStore") as mock_store_cls, ): mock_client.return_value.maindata.return_value = _fake_qbit_maindata() - # Wire the mock store to a real isolated store + # Wire the mock store to a real isolated store and seed headless data. real_store = QbittorrentSampleStore(harness) + real_store.append("svc-speed", round(time.time()), 500000, 1000) mock_store_cls.return_value = real_store result = await adapter.fetch(service, "speed", {}) + mock_client.assert_not_called() + assert "series" in result labels = [s["label"] for s in result["series"]] assert labels == ["download", "upload"] - # The sample just appended should be present + # The scheduler-supplied sample should be present. dl_points = result["series"][0]["points"] assert len(dl_points) >= 1 # timestamps multiplied by 1000 for JS epoch diff --git a/docs/REQUIREMENTS.md b/docs/REQUIREMENTS.md index d609bfe..a2c927b 100644 --- a/docs/REQUIREMENTS.md +++ b/docs/REQUIREMENTS.md @@ -296,6 +296,21 @@ Multiple instances per service type are supported. Services are managed from the These do not reference a service. +### Backend Scheduled Actions and qBittorrent Polling + +- The backend should collect qBittorrent speed samples independently of browser or dashboard presence. +- Scheduled work should use a typed, explicitly registered action system; arbitrary widgets, SSH commands, and user-provided code must not be executable through the scheduler. +- The first scheduled action is qBittorrent speed sampling. The initial deployment assumes one scheduler-capable backend worker; multiple replicas must not silently duplicate polls. +- qBittorrent polling should be opt-out by default for enabled service instances and configurable per service with a 15-second default interval bounded to 5–300 seconds. +- Sample retention should be configurable by duration and maximum rows, defaulting to 30 minutes and 1,200 rows, with duration bounded to 1–24 hours and the row cap enforced server-side. +- The scheduler should run immediately after startup with per-service staggering, use fixed-delay execution, prevent overlap/backlog, and reconcile configuration changes without a backend restart. +- Poll failures should remain enabled, be persisted, and retry with bounded exponential backoff. A successful scheduled or manual run should clear backoff. +- The qBittorrent widget-data endpoint must become read-only; only the scheduler may contact qBittorrent and append samples. +- The service UI should expose polling settings, current status, stale-data state, a manual `Run now` action, selectable chart windows, and paginated scheduled-action history. +- Scheduled-action runs should use dedicated generic records, retain at most 30 days or 1,000 runs per service/action, and never store secrets or raw credentials. +- Disabling a qBittorrent service pauses polling while retaining history; deleting the service purges its samples and scheduler history through the existing cascade-delete behavior. +- Persistent polling failures should be visible in the service UI and application metrics; a new notification channel is not required for the first release. + ### Security - Service secrets (API keys, tokens, passphrases) are **encrypted at rest** with diff --git a/docs/superpowers/plans/2026-07-14-scheduled-actions-test-plan.md b/docs/superpowers/plans/2026-07-14-scheduled-actions-test-plan.md new file mode 100644 index 0000000..68584e8 --- /dev/null +++ b/docs/superpowers/plans/2026-07-14-scheduled-actions-test-plan.md @@ -0,0 +1,156 @@ +# Scheduled Actions Test Plan / QA Checklist + +## Test objectives + +Verify that qBittorrent speed collection is backend-owned, configurable per service, resilient to transient failures, observable, and safe when the frontend is not open. + +## Backend unit tests + +### Configuration + +- [ ] Existing qBittorrent configs receive polling defaults without changing secrets. +- [ ] `polling_enabled` accepts booleans and defaults to enabled. +- [ ] Poll intervals below 5 seconds or above 300 seconds are rejected. +- [ ] Retention below 60 seconds or above 86,400 seconds is rejected. +- [ ] Sample caps below the minimum or above 1,200 are rejected. +- [ ] Credential-looking fields remain rejected from service config. + +### Sample storage + +- [ ] Samples remain isolated by `service_id`. +- [ ] Timestamp retention removes rows older than the configured window. +- [ ] Row-cap retention removes the oldest rows when the cap is exceeded. +- [ ] Both retention rules are applied together. +- [ ] Samples are ordered oldest to newest for chart responses. +- [ ] Service cascade deletion removes samples. +- [ ] Empty and missing databases initialize safely. + +### Scheduler core + +- [ ] Only registered action keys can execute. +- [ ] Disabled services do not run. +- [ ] Enabled services run immediately after startup with deterministic staggering. +- [ ] A normal run waits the configured interval after completion. +- [ ] A slow run cannot overlap itself. +- [ ] A slow run does not create queued backlog entries. +- [ ] Stop interrupts the wait and joins the worker within the configured timeout. +- [ ] A configuration change is applied on the next reconciliation cycle. +- [ ] Disabling a service allows an in-flight run to finish, then prevents new runs. +- [ ] Deleting a service removes its schedule state and history. + +### Backoff and manual runs + +- [ ] A failed run is persisted with safe error text and increments failure state. +- [ ] Retry delay increases exponentially and respects the configured cap. +- [ ] Backoff does not create duplicate queued runs. +- [ ] A successful scheduled run clears failures and backoff. +- [ ] A successful manual run clears failures and backoff. +- [ ] A failed manual run follows normal failure persistence and retry behavior. +- [ ] Manual runs do not alter the configured interval. + +### qBittorrent action and widget separation + +- [ ] The scheduled action calls qBittorrent and appends exactly one sample per successful poll. +- [ ] The qBittorrent client cache is reused correctly per service. +- [ ] Timeouts become failed runs without blocking the scheduler indefinitely. +- [ ] The `speed` widget adapter reads samples but does not call qBittorrent. +- [ ] Multiple widgets or browser tabs do not multiply samples. +- [ ] A failed latest poll still returns last-known samples with stale status. + +## Backend API tests + +- [ ] Scheduler status requires normal API authentication. +- [ ] Status returns effective configuration, last attempt, last success, failure count, backoff, and stale state. +- [ ] Missing service returns 404 or the project’s established service error shape. +- [ ] Disabled service status is explicit and does not run an action. +- [ ] Run history is paginated and supports status/trigger filters. +- [ ] Run history is bounded by 30 days and 1,000 records per service/action. +- [ ] Manual-run endpoint returns a typed result and records the attempt. +- [ ] Samples endpoint accepts a display window and is read-only. +- [ ] Widget-data requests never cause a qBittorrent external call. +- [ ] Responses never expose usernames, passwords, API keys, headers, or raw payloads. + +## Observability tests + +- [ ] Run counters increment for success and failure. +- [ ] Duration metrics record completed attempts. +- [ ] Last-success gauges update only after successful sampling. +- [ ] Failure/stale gauges reset after recovery. +- [ ] Metric labels are bounded and contain no secrets or raw URLs. +- [ ] Structured logs include action/service/status context and sanitize errors. + +## Frontend unit/component tests + +- [ ] qBittorrent schedule fields render only for qBittorrent services. +- [ ] Invalid interval, retention, and cap values show validation feedback. +- [ ] Save preserves existing encrypted-secret behavior. +- [ ] Disabled polling clearly shows paused state. +- [ ] Status card renders healthy, running, backoff, stale, disabled, and never-run states. +- [ ] `Run now` shows pending state and disables duplicate clicks. +- [ ] Successful manual run refreshes status/history and clears backoff display. +- [ ] Failed manual run renders a safe error. +- [ ] Chart window selector requests samples without changing sampler settings. +- [ ] Stale warning appears while last-known speed data remains visible. +- [ ] Run history renders pagination, trigger, status, duration, timestamp, and error details. +- [ ] Empty history and no-data states are readable on mobile. + +## Integration / lifespan tests + +- [ ] Starting the FastAPI lifespan starts the scheduler exactly once. +- [ ] Repeated `start()` calls do not create duplicate workers. +- [ ] Lifespan shutdown stops the scheduler and does not leak a thread. +- [ ] A test app can override the scheduler/action registry cleanly. +- [ ] Existing mail queue and backup poller lifecycle behavior remains unchanged. + +## Manual QA scenarios + +### Headless collection + +1. Configure an enabled qBittorrent service. +2. Start the backend without opening the frontend. +3. Wait for at least two intervals. +4. Query scheduler status and samples directly. +5. Confirm samples and successful run records exist. + +### Duplicate prevention + +1. Open the speed widget in multiple browser tabs. +2. Compare sample count growth to scheduler run count. +3. Confirm browser refreshes do not add samples or external qBittorrent calls. + +### Outage and recovery + +1. Make qBittorrent unreachable. +2. Confirm failures and increasing backoff appear in status/history. +3. Confirm last-known samples remain visible with a stale warning. +4. Restore qBittorrent. +5. Confirm the next successful scheduled or manual run clears backoff and stale state. + +### Configuration reload + +1. Change the interval and retention in the qBittorrent service editor. +2. Confirm the existing worker remains alive. +3. Confirm the new effective values appear after reconciliation. +4. Confirm pruning follows the new retention/cap. + +### Service lifecycle + +1. Disable a service and confirm no new runs are created while history remains. +2. Re-enable it and confirm polling resumes. +3. Delete it and confirm service-owned samples and run records are removed. + +### Deployment constraint + +1. Run the documented single-worker deployment. +2. Confirm one scheduler worker is active. +3. Verify the deployment documentation warns against multiple scheduler-capable replicas. + +## Release gate + +- [ ] Backend test suite passes. +- [ ] Frontend tests pass. +- [ ] Frontend lint passes with no new violations. +- [ ] Frontend build passes. +- [ ] No secret appears in logs, API responses, metrics, or scheduler run records. +- [ ] Project maps are patched for new files and validated. +- [ ] Requirements and runbook documentation match the shipped behavior. diff --git a/docs/superpowers/plans/2026-07-14-scheduled-actions.md b/docs/superpowers/plans/2026-07-14-scheduled-actions.md new file mode 100644 index 0000000..33b8bf9 --- /dev/null +++ b/docs/superpowers/plans/2026-07-14-scheduled-actions.md @@ -0,0 +1,124 @@ +# Typed Scheduler and qBittorrent Polling Implementation Plan + +> This plan implements the approved design in `docs/superpowers/specs/2026-07-14-scheduled-actions-design.md`. + +**Goal:** Move qBittorrent speed collection into a backend-owned typed scheduler while adding per-service controls, run history, stale-data status, and a manual run action. + +**Scope:** qBittorrent speed polling only. The scheduler registry is extensible, but arbitrary widgets, SSH tasks, and other integrations remain out of scope. + +## Delivery slices + +### Slice 1 — Contracts and persistence + +- [ ] Add validated qBittorrent config fields to `backend/src/media_library_viewer_api/integrations/qbittorrent.py`: + - `polling_enabled` default `true`; + - `poll_interval_seconds` default `15`, range `5..300`; + - `sample_retention_seconds` default `1800`, range `60..86400`; + - `sample_max_rows` default `1200`, range `60..1200`. +- [ ] Add migration/default normalization tests for existing qBittorrent records. +- [ ] Extend `QbittorrentSampleStore` to prune by retention timestamp and capped row count. +- [ ] Add a generic scheduler storage concern with run records, indexes, pruning, and cascade deletion by service ID. +- [ ] Add typed backend models for scheduler status, run records, samples, and manual-run responses. + +**Acceptance:** Existing service records validate without edits; qBittorrent samples remain isolated by service; scheduler records cannot contain secrets; deletion removes service-owned samples and runs. + +### Slice 2 — Typed scheduler core + +- [ ] Create a scheduler action protocol and registry. +- [ ] Implement a single-worker, lifespan-managed scheduler coordinator with responsive stop behavior. +- [ ] Add qBittorrent speed sampling as the first registered action. +- [ ] Extract external polling from `QbittorrentWidgetSource` into a reusable sampler/action helper. +- [ ] Implement immediate startup execution with deterministic staggering. +- [ ] Implement fixed-delay, no-overlap execution and bounded exponential backoff. +- [ ] Reconcile enabled services/config changes on each cycle. +- [ ] Add safe structured logs and Prometheus metrics. +- [ ] Start/stop the scheduler in `main.py` alongside the existing mail queue and backup poller. + +**Acceptance:** With no frontend open, enabled qBittorrent services append samples; one slow service cannot create overlapping runs or a backlog; shutdown joins the worker; a successful manual or scheduled run resets backoff. + +### Slice 3 — Read-only APIs + +- [ ] Add `backend/src/media_library_viewer_api/routers/scheduler.py`. +- [ ] Add status, paginated runs, manual-run, and read-only samples endpoints. +- [ ] Keep service configuration writes on the existing service-instance API. +- [ ] Change the qBittorrent speed widget adapter to read samples only. +- [ ] Add stale-data calculation and safe error truncation. +- [ ] Add API tests for disabled/missing services, stale data, pagination, manual runs, backoff, and authentication. + +**Acceptance:** Opening or refreshing a speed widget never contacts qBittorrent and never appends a sample; API responses expose timestamps and status but no credentials. + +### Slice 4 — Frontend controls and history + +- [ ] Add scheduler TypeScript types, API functions, and React Query hooks. +- [ ] Add qBittorrent schedule controls to the existing schema-driven service editor. +- [ ] Add status/backoff/stale-data presentation and a `Run now` action. +- [ ] Add user-selectable chart windows. +- [ ] Add a paginated run-history table with safe error details. +- [ ] Keep UI refreshes separate from sampler cadence. +- [ ] Add frontend tests for validation, disabled state, stale warning, manual-run reset, chart-window selection, and run-history rendering. + +**Acceptance:** Operators can configure, inspect, and manually trigger qBittorrent polling from the service surface without opening the dashboard; the chart remains useful during outages and identifies stale data. + +### Slice 5 — Documentation and operational verification + +- [ ] Update `docs/REQUIREMENTS.md` with scheduler requirements and the one-worker constraint. +- [ ] Update deployment/runbook documentation with scheduler startup, shutdown, and replica guidance. +- [ ] Add migration/recovery notes for sample and run-history retention. +- [ ] Run backend tests, frontend tests, lint, and build. +- [ ] Verify a headless collection scenario against a mocked qBittorrent service. +- [ ] Verify project-map artifacts after files are added. + +## Suggested file map + +### Backend + +| File | Change | +| --- | --- | +| `backend/src/media_library_viewer_api/integrations/qbittorrent.py` | Schedule config schema and defaults | +| `backend/src/media_library_viewer_api/services/qbittorrent_store.py` | Duration/cap pruning and sample queries | +| `backend/src/media_library_viewer_api/services/service_data.py` | Register scheduler run concern | +| `backend/src/media_library_viewer_api/services/scheduler.py` | Worker lifecycle, reconciliation, timing, backoff | +| `backend/src/media_library_viewer_api/services/scheduler_actions.py` | Typed registry and qBittorrent action | +| `backend/src/media_library_viewer_api/services/scheduler_store.py` | Run-record persistence and pruning | +| `backend/src/media_library_viewer_api/models/scheduler.py` | Response/request models | +| `backend/src/media_library_viewer_api/routers/scheduler.py` | Status, history, samples, manual-run API | +| `backend/src/media_library_viewer_api/widgets/sources.py` | Make qBittorrent speed reads side-effect free | +| `backend/src/media_library_viewer_api/main.py` | Start/stop scheduler | +| `backend/tests/test_scheduler.py` | Scheduler lifecycle/timing/backoff tests | +| `backend/tests/test_scheduler_api.py` | Endpoint and auth tests | +| `backend/tests/test_service_data.py` | Migration/cascade coverage | +| `backend/tests/test_widgets.py` | Read-only qBittorrent widget coverage | + +### Frontend + +| File | Change | +| --- | --- | +| `frontend/src/types/scheduler.ts` | Scheduler status/run/sample types | +| `frontend/src/api/scheduler.ts` | Typed endpoint wrappers | +| `frontend/src/hooks/useScheduler.ts` | Queries and manual-run mutation | +| `frontend/src/pages/ServicesPage.tsx` | qBittorrent schedule controls/status surface | +| `frontend/src/pages/ServicePage.tsx` or qBittorrent service tab | Status, chart window, history surface | +| `frontend/src/widgets/QbittorrentSpeedWidget.tsx` | Read-only sample window and stale warning | +| `frontend/src/integrations/registry.ts` | Schedule metadata/config exposure if needed | +| `frontend/src/types/index.ts` | Shared exports | + +## Risks and mitigations + +- **Duplicate polling:** widget adapter becomes read-only; only scheduler action calls qBittorrent. +- **Multiple backend workers:** document and log the one-worker constraint; do not silently duplicate work. +- **Unbounded storage:** prune by both duration and row cap; test pruning under rapid polling. +- **Credential leakage:** reuse existing secret resolution and sanitize run errors/log fields. +- **Scheduler shutdown races:** use a stop event, per-action lock, and bounded joins; test lifespan shutdown. +- **Config changes during a run:** let the current run finish, then reconcile on the next cycle. +- **Stale but useful data:** return samples plus explicit stale status rather than blanking the chart. + +## Verification commands + +```bash +cd backend && PYTHONPATH=src pytest +cd frontend && npm test +cd frontend && npm run lint +cd frontend && npm run build +``` + +Do not begin implementation until the final module names, retry cap/jitter, and chart-window response shape are confirmed during the implementation pass. diff --git a/docs/superpowers/specs/2026-07-14-scheduled-actions-design.md b/docs/superpowers/specs/2026-07-14-scheduled-actions-design.md new file mode 100644 index 0000000..c66b2c5 --- /dev/null +++ b/docs/superpowers/specs/2026-07-14-scheduled-actions-design.md @@ -0,0 +1,244 @@ +# Typed Scheduled Actions and qBittorrent Polling Design + +**Date:** 2026-07-14 +**Status:** Proposed + +## Overview + +Manage currently collects qBittorrent speed samples as a side effect of a browser polling the widget-data endpoint. This design moves collection into a backend-owned typed scheduler so samples continue when no page is open, while keeping widget reads read-only. + +The first scheduled action is qBittorrent speed polling. The scheduler is intentionally extensible but does not execute arbitrary widgets, SSH commands, or user-provided code. + +## Goals + +- Collect qBittorrent download/upload speed independently of browser presence. +- Configure polling per qBittorrent service instance. +- Preserve per-service SQLite isolation and existing cascade-delete behavior. +- Provide current status, stale-data state, run history, and a manual `Run now` action. +- Reuse the existing lifespan worker pattern and remain safe under the single-backend-worker deployment model. +- Expose metrics and structured logs without adding a new notification channel. + +## Non-goals + +- Distributed scheduling across replicas. +- External worker infrastructure or a task queue. +- Scheduling arbitrary saved SSH tasks. +- Moving every widget-backed integration to the scheduler in this release. +- A global scheduler administration page. + +## Decisions + +| Area | Decision | +| --- | --- | +| Architecture | Typed action registry with qBittorrent as the first action | +| Worker model | One lifespan-managed backend worker; deployment must run one scheduler-capable backend process | +| Activation | Enabled qBittorrent services poll by default; polling is opt-out | +| Defaults | 15-second interval; valid range 5–300 seconds | +| Samples | 30-minute default history; valid range 1–24 hours; configurable lower sample cap with a hard maximum of 1,200 rows | +| Timing | Immediate first run with per-service startup staggering; fixed delay after completion | +| Concurrency | No overlapping runs and no queued missed ticks per service | +| Failure | Keep enabled, record failure, retry with bounded exponential backoff | +| Manual run | Supported; successful manual run clears backoff | +| Widget data | Read-only; scheduler is the only qBittorrent sampler | +| Run history | Dedicated generic scheduled-action run records; retain 30 days or 1,000 runs per service/action | +| Service lifecycle | Disable pauses and retains history; delete purges service-owned data through cascade deletion | +| UI | Controls and status/history live with each qBittorrent service | +| Alerts | UI and metrics only in this release | + +## Current and target flow + +### Current + +```text +React Query interval + -> GET /api/widgets/instances/{id}/data + -> QbittorrentWidgetSource.fetch() + -> qBittorrent API + -> append speed sample + -> return chart data +``` + +### Target + +```text +Backend lifespan + -> TypedScheduler + -> registered QbittorrentSpeedAction + -> qBittorrent API + -> QbittorrentSampleStore + -> SchedulerRunStore + +React Query / service UI + -> scheduler status/history/sample endpoints + -> read-only SQLite queries +``` + +## Configuration model + +The existing qBittorrent service config gains validated non-secret fields: + +```json +{ + "base_url": "https://qbit.example", + "timeout_seconds": 60, + "polling_enabled": true, + "poll_interval_seconds": 15, + "sample_retention_seconds": 1800, + "sample_max_rows": 1200 +} +``` + +Suggested validation: + +- `polling_enabled`: boolean, default `true` for backward compatibility. +- `poll_interval_seconds`: integer from 5 through 300, default 15. +- `sample_retention_seconds`: integer from 60 through 86,400, default 1,800. +- `sample_max_rows`: integer from 60 through 1,200, default 1,200. The upper bound is a server safety limit, not merely a UI hint. +- Secrets remain exclusively in the existing encrypted secret fields. + +The service type metadata must expose these fields so the existing schema-driven service editor renders them. Existing qBittorrent records receive defaults through normalization rather than a destructive migration. + +## Scheduler architecture + +### Registry and contracts + +Add a small scheduler service with explicit action registration: + +```python +class ScheduledAction(Protocol): + action_key: str + async def run(self, service: ServiceRecord, context: ActionContext) -> ActionResult: ... +``` + +The registry maps an action key to its implementation and metadata. The first key is `qbittorrent.speed_sample`. The scheduler never evaluates arbitrary config as executable code. + +A scheduler cycle should: + +1. Read enabled service records. +2. Select services whose typed action is enabled. +3. Reconcile changed interval/enabled settings. +4. Run due actions serially per service. +5. Persist a run record and update in-memory status. +6. Wait using a stop event so shutdown is responsive. + +A thread-based coordinator is appropriate for the first release because `BackupAlertPoller` already establishes the project’s lifespan-managed worker pattern and qBittorrent’s client is blocking. The action may use the existing authenticated client cache, but the sampler should be extracted from the widget adapter so collection and presentation are not coupled. + +### Timing and backoff + +- First eligible service run starts immediately after startup, with a small deterministic stagger based on service ordering. +- Normal scheduling uses fixed delay: the next due time is calculated after the previous attempt completes. +- A per-service action lock prevents overlap. +- A failed run uses bounded exponential backoff, capped below the configured interval’s operational maximum. Backoff must not enqueue missed runs. +- A successful scheduled or manual run resets consecutive failures and clears `backoff_until`. +- Config changes are observed during the next reconciliation cycle; an interval change affects the next due calculation. +- Disabling a service prevents new work and allows the current run to finish before the action becomes idle. + +### Single-worker constraint + +The initial design assumes one backend process owns scheduler execution. Running multiple Uvicorn workers or replicas would duplicate polls and run records. Startup logs and operational documentation must make this constraint explicit. A future distributed lease can be added without changing the action contract. + +## Storage + +### Speed samples + +Extend `QbittorrentSampleStore` to prune by both: + +- `service_id` and `ts >= now - sample_retention_seconds`; +- most recent `sample_max_rows`, bounded by 1,200. + +The existing `qbittorrent_speed_samples` table remains the source for chart data. Its API should accept a requested display window and return ordered samples. Deleting a service must continue to cascade into this concern. + +### Scheduled-action runs + +Add a generic scheduler storage concern, separate from `service_task_runs`, with fields equivalent to: + +| Field | Description | +| --- | --- | +| `id` | Run identifier | +| `service_id` | Owning service instance | +| `action_key` | Registered action key, e.g. `qbittorrent.speed_sample` | +| `trigger` | `schedule` or `manual` | +| `started_at` / `finished_at` | Attempt timing | +| `status` | `running`, `success`, `failure`, `backoff`, or `cancelled` | +| `duration_ms` | Elapsed time | +| `attempt` | Retry/backoff attempt number | +| `error` | Secret-safe error text, truncated | +| `created_at` | Record creation time | + +Indexes should cover `(service_id, action_key, started_at DESC)` and `(status, started_at DESC)`. Prune records older than 30 days and enforce a maximum of 1,000 records per service/action. + +No credentials, request headers, or raw qBittorrent payloads may be stored in run history. + +## Backend API + +Add a dedicated scheduler router. Exact response models should be typed and should not expose secrets. + +| Endpoint | Purpose | +| --- | --- | +| `GET /api/scheduler/services/{service_id}/status` | Current action state, last attempt/success, stale state, failures, backoff, and effective config | +| `GET /api/scheduler/services/{service_id}/runs` | Paginated run history with status/trigger filters | +| `POST /api/scheduler/services/{service_id}/run` | Run the registered qBittorrent action immediately; return a run/status response | +| `GET /api/scheduler/services/{service_id}/samples` | Read-only speed samples for a selected display window | + +The existing `PUT /api/services/instances/{id}` remains the write path for schedule configuration. The widget-data endpoint must stop calling qBittorrent for the `speed` kind; it should read samples through the same store/query helper used by the scheduler API. + +Manual runs must use the same action registry and persistence path as scheduled runs. A successful manual run clears backoff; a failed manual run records the failure and applies the same bounded retry state. + +## Stale-data semantics + +The status response should include `last_success_at`, `last_error`, `consecutive_failures`, `backoff_until`, and `is_stale`. Suggested initial stale rule: + +```text +is_stale = no successful run + OR now - last_success_at > max(2 * effective_interval, 60 seconds) +``` + +The speed widget should retain and render the last known samples with a warning containing the last-success time and current error. It should not replace useful history with an empty state solely because the latest poll failed. + +## Frontend design + +The existing schema-driven qBittorrent service editor should gain a scheduling section containing: + +- polling enabled switch; +- interval field with bounds/error text; +- sample retention duration; +- maximum sample rows; +- effective next-run and last-success summary; +- `Run now` button; +- current failure/backoff message. + +A qBittorrent service detail/editor surface should also contain: + +- stale-data banner; +- user-selectable chart windows (for example 5m, 30m, 1h, all retained); +- speed chart sourced from read-only sample data; +- paginated run-history table with trigger, status, duration, timestamp, and safe error detail; +- loading, empty, disabled, and failed states. + +Add typed API functions, React Query hooks, and types under the existing `frontend/src/api`, `frontend/src/hooks`, and `frontend/src/types` patterns. Poll status/history at a slower UI cadence than the sampler; the UI must not drive collection. + +## Observability + +Add secret-safe metrics using bounded labels: + +- scheduled action runs total by action and status; +- scheduled action duration by action; +- last successful run timestamp by action/service; +- current consecutive failures or stale state by action/service. + +Avoid labels containing URLs, usernames, API keys, raw errors, or unbounded widget IDs. Structured logs should include service ID, action key, trigger, status, duration, and request ID where available. + +## Security and operational constraints + +- Only registered action keys can execute. +- Service credentials are loaded through existing decryption helpers and are never returned or persisted in run records. +- Manual-run endpoints use existing JWT/API authentication. +- The scheduler must stop cleanly during lifespan shutdown and should not leave a new thread running after tests finish. +- The deployment documentation must state the one-worker scheduler constraint. + +## Open implementation details + +- Choose final module names and whether scheduler run storage belongs in a new service-data concern or a dedicated settings-store table. +- Define exact retry cap and jitter values. +- Decide whether the scheduler status endpoint returns one action or a list of registered actions. +- Finalize chart window/downsampling behavior for the 24-hour/1,200-row maximum. diff --git a/frontend/src/api/scheduler.ts b/frontend/src/api/scheduler.ts new file mode 100644 index 0000000..5d70cee --- /dev/null +++ b/frontend/src/api/scheduler.ts @@ -0,0 +1,41 @@ +import { get, post } from "./shared"; +import type { + SchedulerManualRunResponse, + SchedulerRunsResponse, + SchedulerSamplesResponse, + SchedulerStatus, +} from "../types"; + +export function fetchSchedulerStatus( + serviceId: string, +): Promise { + return get(`/api/scheduler/services/${serviceId}/status`); +} + +export function fetchSchedulerRuns( + serviceId: string, + limit = 20, +): Promise { + return get( + `/api/scheduler/services/${serviceId}/runs`, + { limit: String(limit) }, + ); +} + +export function fetchSchedulerSamples( + serviceId: string, + windowSeconds: number, +): Promise { + return get( + `/api/scheduler/services/${serviceId}/samples`, + { window_seconds: String(windowSeconds) }, + ); +} + +export function runSchedulerAction( + serviceId: string, +): Promise { + return post( + `/api/scheduler/services/${serviceId}/run`, + ); +} diff --git a/frontend/src/hooks/useScheduler.ts b/frontend/src/hooks/useScheduler.ts new file mode 100644 index 0000000..f977a71 --- /dev/null +++ b/frontend/src/hooks/useScheduler.ts @@ -0,0 +1,52 @@ +import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query"; +import { + fetchSchedulerRuns, + fetchSchedulerSamples, + fetchSchedulerStatus, + runSchedulerAction, +} from "../api/scheduler"; + +export function useSchedulerStatus(serviceId: string) { + return useQuery({ + queryKey: ["scheduler", "status", serviceId], + queryFn: () => fetchSchedulerStatus(serviceId), + enabled: Boolean(serviceId), + refetchInterval: 15_000, + }); +} + +export function useSchedulerRuns(serviceId: string) { + return useQuery({ + queryKey: ["scheduler", "runs", serviceId], + queryFn: () => fetchSchedulerRuns(serviceId), + enabled: Boolean(serviceId), + refetchInterval: 15_000, + }); +} + +export function useSchedulerSamples(serviceId: string, windowSeconds: number) { + return useQuery({ + queryKey: ["scheduler", "samples", serviceId, windowSeconds], + queryFn: () => fetchSchedulerSamples(serviceId, windowSeconds), + enabled: Boolean(serviceId), + refetchInterval: 15_000, + }); +} + +export function useRunSchedulerAction() { + const queryClient = useQueryClient(); + return useMutation({ + mutationFn: runSchedulerAction, + onSuccess: (_result, serviceId) => { + queryClient.invalidateQueries({ + queryKey: ["scheduler", "status", serviceId], + }); + queryClient.invalidateQueries({ + queryKey: ["scheduler", "runs", serviceId], + }); + queryClient.invalidateQueries({ + queryKey: ["scheduler", "samples", serviceId], + }); + }, + }); +} diff --git a/frontend/src/pages/service-tabs/QbittorrentTab.tsx b/frontend/src/pages/service-tabs/QbittorrentTab.tsx new file mode 100644 index 0000000..8f3383d --- /dev/null +++ b/frontend/src/pages/service-tabs/QbittorrentTab.tsx @@ -0,0 +1,211 @@ +import { Activity, Clock, Play, RefreshCw, TriangleAlert } from "lucide-react"; +import { useState } from "react"; +import { LineSeriesChart } from "../../components/LineSeriesChart"; +import { + useRunSchedulerAction, + useSchedulerRuns, + useSchedulerSamples, + useSchedulerStatus, +} from "../../hooks/useScheduler"; +import type { ServiceInstance } from "../../types"; +import { Alert, AlertDescription, AlertTitle } from "@/components/ui/alert"; +import { Badge } from "@/components/ui/badge"; +import { Button } from "@/components/ui/button"; +import { Card, CardContent, CardHeader, CardTitle } from "@/components/ui/card"; +import { + Select, + SelectContent, + SelectItem, + SelectTrigger, + SelectValue, +} from "@/components/ui/select"; +import { Skeleton } from "@/components/ui/skeleton"; + +const WINDOWS = [ + { value: 900, label: "15 minutes" }, + { value: 1800, label: "30 minutes" }, + { value: 3600, label: "1 hour" }, + { value: 21600, label: "6 hours" }, + { value: 86400, label: "24 hours" }, +]; + +function formatTimestamp(value: number | null): string { + return value ? new Date(value * 1000).toLocaleString() : "Never"; +} + +function statusVariant( + status: string, +): "default" | "secondary" | "destructive" | "outline" { + if (status === "success") return "default"; + if (status === "failure") return "destructive"; + if (status === "running") return "secondary"; + return "outline"; +} + +export function QbittorrentTab({ instance }: { instance: ServiceInstance }) { + const [windowSeconds, setWindowSeconds] = useState(1800); + const status = useSchedulerStatus(instance.id); + const samples = useSchedulerSamples(instance.id, windowSeconds); + const runs = useSchedulerRuns(instance.id); + const runNow = useRunSchedulerAction(); + const stale = Boolean(status.data?.enabled && status.data.is_stale); + const hasError = Boolean(status.data?.last_error || runNow.error); + + const chartSeries = samples.data + ? [ + { + label: "download", + points: samples.data.samples.map((sample) => ({ + t: sample.ts * 1000, + v: sample.dl_speed, + })), + }, + { + label: "upload", + points: samples.data.samples.map((sample) => ({ + t: sample.ts * 1000, + v: sample.up_speed, + })), + }, + ] + : []; + + return ( +
+
+
+ + Headless speed polling + {status.data && ( + + {status.data.enabled ? "Enabled" : "Paused"} + + )} +
+ +
+ + {(stale || hasError) && ( + + + + {hasError ? "Polling error" : "Data may be stale"} + + + {status.data?.last_error || + runNow.error?.message || + "No successful sample has been recorded recently."} + + + )} + + + + + + Speed history + + + + + {samples.isLoading ? ( + + ) : ( + + )} + + + +
+ + +
Last success
+
+ {formatTimestamp(status.data?.last_success_at ?? null)} +
+ + + + +
Next poll
+
+ {formatTimestamp(status.data?.next_run_at ?? null)} +
+ + + + +
+ Consecutive failures +
+
+ {status.data?.consecutive_failures ?? 0} +
+ + +
+ + + + + + Recent runs + + + + {runs.isLoading ? ( + + ) : runs.data?.items.length ? ( + runs.data.items.map((run) => ( +
+
+ + {run.status} + + + {run.trigger} · {formatTimestamp(run.started_at)} + +
+ {run.error && ( + {run.error} + )} +
+ )) + ) : ( +
+ No runs recorded yet. +
+ )} + + +
+ ); +} diff --git a/frontend/src/pages/service-tabs/index.ts b/frontend/src/pages/service-tabs/index.ts index c03b8b6..91b212d 100644 --- a/frontend/src/pages/service-tabs/index.ts +++ b/frontend/src/pages/service-tabs/index.ts @@ -16,6 +16,7 @@ import { ActionsTab } from "./ActionsTab"; import { JobsTab } from "./JobsTab"; import { UsersTab } from "./UsersTab"; import { MessagingTab } from "./MessagingTab"; +import { QbittorrentTab } from "./QbittorrentTab"; export type ServiceTabComponent = ComponentType<{ instance: ServiceInstance }>; @@ -57,6 +58,8 @@ export function serviceContentTabs(serviceType: string): ContentTab[] { return [{ label: "Alerts", Component: AlertsTab }]; case "prometheus": return [{ label: "Metrics", Component: MetricsTab }]; + case "qbittorrent": + return [{ label: "Speed", Component: QbittorrentTab }]; default: return []; } diff --git a/frontend/src/types/index.ts b/frontend/src/types/index.ts index e5ac9c1..dd6a8af 100644 --- a/frontend/src/types/index.ts +++ b/frontend/src/types/index.ts @@ -464,6 +464,60 @@ export interface ServiceInstance { updated_at: number; } +export interface SchedulerStatus { + service_id: string; + action_key: string; + worker_running: boolean; + enabled: boolean; + running: boolean; + poll_interval_seconds: number; + sample_retention_seconds: number; + sample_max_rows: number; + next_run_at: number | null; + last_attempt_at: number | null; + last_success_at: number | null; + last_error: string; + consecutive_failures: number; + backoff_until: number | null; + is_stale: boolean; +} + +export interface SchedulerRun { + id: string; + service_id: string; + action_key: string; + trigger: "schedule" | "manual"; + started_at: number; + finished_at: number | null; + status: "running" | "success" | "failure" | "cancelled"; + attempt: number; + duration_ms: number | null; + error: string; + created_at: number; +} + +export interface SchedulerRunsResponse { + items: SchedulerRun[]; + total: number; + limit: number; + offset: number; +} + +export interface SchedulerSamplesResponse { + service_id: string; + window_seconds: number; + samples: Array<{ + ts: number; + dl_speed: number; + up_speed: number; + }>; +} + +export interface SchedulerManualRunResponse { + run: SchedulerRun; + status: SchedulerStatus; +} + export interface ServiceInstanceInput { id?: string | null; service_type: string;