manage/backend/src/media_library_viewer_api/services/scheduler.py

"""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"]