manage/backend/src/media_library_viewer_api/widgets/sources.py

"""Widget source adapters.

Adapters translate a widget instance into dashboard data. Service-bound widgets
are resolved against a :class:`ServiceRecord` (config + decrypted secrets); the
built-in widgets (backups, static) take ``service=None``.

Adapters never accept arbitrary commands and never store credentials — secrets
are decrypted in memory only for the duration of a fetch.
"""

from __future__ import annotations

import asyncio
import logging
import time
from dataclasses import dataclass, field
from typing import Any, Protocol

import requests

from media_library_viewer_api.clients.jellyfin import JellyfinClient
from media_library_viewer_api.clients.qbittorrent import QbittorrentClient
from media_library_viewer_api.domain.dashboard import (
    _map_sessions_to_activity_rows,
    build_backup_dashboard_summary,
)
from media_library_viewer_api.integrations.alertmanager import summarize_alerts
from media_library_viewer_api.services.qbittorrent_store import QbittorrentSampleStore
from media_library_viewer_api.services.settings_store import SettingsStore, get_settings_store
from media_library_viewer_api.services.task_runner import run_saved_task
from media_library_viewer_api.widgets.prometheus_range import (
    WINDOW_PRESETS,
    normalize_grafana_frames,
    normalize_prometheus_matrix,  # noqa: F401 — kept for future direct_url path (design decision 5)
    step_for_window,
)

logger = logging.getLogger(__name__)


@dataclass
class ServiceRecord:
    """Runtime view of a service instance with decrypted secrets."""

    id: str
    service_type: str
    name: str
    config: dict[str, Any] = field(default_factory=dict)
    secrets: dict[str, str] = field(default_factory=dict)
    enabled: bool = True


def build_service_record(store: SettingsStore, service_row: dict[str, Any]) -> ServiceRecord:
    """Build a :class:`ServiceRecord`, decrypting secrets in memory."""
    from media_library_viewer_api.services.secrets import decrypt_secrets

    return ServiceRecord(
        id=service_row["id"],
        service_type=service_row["service_type"],
        name=service_row["name"],
        config=service_row.get("config") or {},
        secrets=decrypt_secrets(service_row.get("secrets") or {}),
        enabled=bool(service_row.get("enabled", True)),
    )


class WidgetSource(Protocol):
    """Protocol for widget source adapters."""

    async def fetch(
        self,
        service: ServiceRecord | None,
        widget_kind: str,
        config: dict[str, Any],
    ) -> dict[str, Any]: ...


# ---------------------------------------------------------------------------
# Built-in (service-less) adapters
# ---------------------------------------------------------------------------


class BackupsWidgetSource:
    """Compute the backup dashboard summary from internal tables."""

    async def fetch(self, service: ServiceRecord | None, widget_kind: str, config: dict[str, Any]) -> dict[str, Any]:
        try:
            store = get_settings_store()
            summary = build_backup_dashboard_summary(store)
            return summary.model_dump()
        except Exception as exc:
            logger.exception("backups adapter failed")
            return {"error": f"Backup summary failed: {exc}"}


class StaticWidgetSource:
    """Return static text/markdown unchanged."""

    async def fetch(self, service: ServiceRecord | None, widget_kind: str, config: dict[str, Any]) -> dict[str, Any]:
        return {"text": config.get("text", "")}


# ---------------------------------------------------------------------------
# Service-bound adapters
# ---------------------------------------------------------------------------


class MetricSource:
    """Run PromQL queries through a Grafana gateway (``/api/ds/query``)."""

    async def fetch(self, service: ServiceRecord | None, widget_kind: str, config: dict[str, Any]) -> dict[str, Any]:
        try:
            if service is None:
                return {"error": "Prometheus widget is missing its service"}
            grafana_url = str(service.config.get("grafana_url") or "").rstrip("/")
            api_key = str(service.secrets.get("grafana_api_key") or "")
            datasource_uid = str(service.config.get("datasource_uid") or "prometheus")
            timeout = int(service.config.get("timeout_seconds") or 10)
            if widget_kind == "chart":
                return await self._fetch_chart(grafana_url, api_key, datasource_uid, timeout, config)
            if widget_kind == "gauge":
                return await self._fetch_gauge(grafana_url, api_key, datasource_uid, timeout, config)
            if widget_kind == "mean":
                return await self._fetch_mean(grafana_url, api_key, datasource_uid, timeout, config)
            # Default: instant-query metric path.
            return await self._fetch_metric(grafana_url, api_key, datasource_uid, timeout, config)
        except Exception as exc:
            logger.exception("prometheus adapter failed")
            return {"error": f"Prometheus query failed: {exc}"}

    async def _gateway_query(
        self,
        grafana_url: str,
        api_key: str,
        datasource_uid: str,
        timeout: int,
        promql: str,
        window_seconds: int | None = None,
        max_data_points: int = 200,
    ) -> dict[str, Any]:
        """POST ``{grafana_url}/api/ds/query``; return raw Grafana JSON or ``{error}``.

        - ``window_seconds=None`` → instant mapping (``from=now-1m, maxDataPoints=1``).
        - ``window_seconds=<N>`` → range query (``from=now-Ns``, step derived).
        """
        if not grafana_url:
            return {"error": "grafana_url is required"}
        if not api_key:
            return {"error": "grafana_api_key is required"}
        step = step_for_window(window_seconds) if window_seconds else 15
        interval_ms = step * 1000
        body = {
            "queries": [
                {
                    "datasource": {"uid": datasource_uid, "type": "prometheus"},
                    "expr": promql,
                    "format": "time_series",
                    "intervalMs": interval_ms,
                    "maxDataPoints": 1 if window_seconds is None else max_data_points,
                    "refId": "A",
                }
            ],
            "from": f"now-{window_seconds or 60}s" if window_seconds else "now-1m",
            "to": "now",
        }

        def _do_post() -> dict[str, Any]:
            resp = requests.post(
                f"{grafana_url}/api/ds/query",
                json=body,
                headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
                timeout=timeout,
            )
            resp.raise_for_status()
            return resp.json()

        try:
            return await asyncio.wait_for(asyncio.to_thread(_do_post), timeout=timeout)
        except asyncio.TimeoutError:
            return {"error": "Grafana query timed out"}
        except requests.RequestException as exc:
            logger.exception("grafana gateway query failed")
            return {"error": f"Grafana query failed: {exc}"}

    async def _fetch_chart(
        self, grafana_url: str, api_key: str, datasource_uid: str, timeout: int, config: dict[str, Any]
    ) -> dict[str, Any]:
        """Range query → ``{series}`` for the chart widget (GM-106)."""
        promql = config.get("promql")
        if not promql:
            return {"error": "promql is required"}
        window = WINDOW_PRESETS.get(config.get("window", "1h"), WINDOW_PRESETS["1h"])
        raw = await self._gateway_query(grafana_url, api_key, datasource_uid, timeout, promql, window_seconds=window)
        if "error" in raw:
            return raw
        return {"series": normalize_grafana_frames(raw)}

    async def _fetch_gauge(
        self, grafana_url: str, api_key: str, datasource_uid: str, timeout: int, config: dict[str, Any]
    ) -> dict[str, Any]:
        """Instant query → scalar for the gauge widget (GM-107).

        Scalar-only: a multi-series query returns an error. Threshold config is
        passed through for the frontend renderer.
        """
        promql = config.get("promql") or ""
        if not promql:
            return {"error": "promql is required"}
        raw = await self._gateway_query(grafana_url, api_key, datasource_uid, timeout, promql, window_seconds=None)
        if "error" in raw:
            return raw
        series = normalize_grafana_frames(raw)
        if len(series) != 1:
            return {"error": "Gauge requires a single-series query; refine your PromQL"}
        points = series[0]["points"]
        if not points:
            return {"error": "Gauge query returned no scalar value"}
        value = points[-1]["v"]
        if value is None:
            return {"error": "Gauge query returned no scalar value"}
        return {
            "value": value,
            "warn_at": config.get("warn_at"),
            "crit_at": config.get("crit_at"),
            "min": config.get("min"),
            "max": config.get("max"),
            "unit": config.get("unit"),
        }

    async def _fetch_mean(
        self, grafana_url: str, api_key: str, datasource_uid: str, timeout: int, config: dict[str, Any]
    ) -> dict[str, Any]:
        """Range query → client-side mean for the mean widget (GM-108).

        Runs a gateway range query over the configured window preset, averages
        all non-null numeric samples of the single series, and returns a scalar.
        Scalar-only: a multi-series query returns an error.
        """
        promql = config.get("promql")
        if not promql:
            return {"error": "promql is required"}
        window = WINDOW_PRESETS.get(config.get("window", "1h"), WINDOW_PRESETS["1h"])
        raw = await self._gateway_query(grafana_url, api_key, datasource_uid, timeout, promql, window_seconds=window)
        if "error" in raw:
            return raw
        series = normalize_grafana_frames(raw)
        if len(series) != 1:
            return {"error": "Mean requires a single-series query; refine your PromQL"}
        nums = [p["v"] for p in series[0]["points"] if p["v"] is not None]
        if not nums:
            return {"error": "Mean query returned no numeric samples in the window"}
        return {"value": sum(nums) / len(nums), "unit": config.get("unit")}

    async def _fetch_metric(
        self, grafana_url: str, api_key: str, datasource_uid: str, timeout: int, config: dict[str, Any]
    ) -> dict[str, Any]:
        """Instant query → ``{result}`` for the metric widget (GM-109).

        Returns ``{result: [{label, points}]}`` — the normalized series shape.
        The frontend ``PrometheusMetricWidget`` renders the last point of each
        series.
        """
        promql = config.get("promql") or ""
        if not promql:
            return {"error": "promql is required"}
        raw = await self._gateway_query(grafana_url, api_key, datasource_uid, timeout, promql, window_seconds=None)
        if "error" in raw:
            return raw
        return {"result": normalize_grafana_frames(raw)}


class AlertmanagerWidgetSource:
    """Fetch firing alerts from an Alertmanager service and summarize them."""

    async def fetch(self, service: ServiceRecord | None, widget_kind: str, config: dict[str, Any]) -> dict[str, Any]:
        try:
            if service is None:
                return {"error": "Alertmanager widget is missing its service"}
            base_url = str(service.config.get("base_url") or "").rstrip("/")
            timeout = int(service.config.get("timeout_seconds") or 5)
            severity_filter = config.get("severity_filter") or None
            headers: dict[str, str] = {}
            api_key = str(service.secrets.get("api_key") or "")
            if api_key:
                headers["Authorization"] = f"Bearer {api_key}"
            response = await asyncio.wait_for(
                asyncio.to_thread(
                    requests.get,
                    f"{base_url}/api/v1/alerts",
                    headers=headers,
                    timeout=timeout,
                ),
                timeout=timeout,
            )
            response.raise_for_status()
            payload = response.json()
            alerts = payload.get("data", []) if isinstance(payload, dict) else []
            return summarize_alerts(alerts, severity_filter=severity_filter)
        except asyncio.TimeoutError:
            return {"error": "Widget data fetch timed out"}
        except requests.RequestException as exc:
            logger.exception("alertmanager adapter failed")
            return {"error": f"Alertmanager query failed: {exc}"}
        except Exception as exc:
            logger.exception("alertmanager adapter failed")
            return {"error": f"Alertmanager query failed: {exc}"}


class JellyfinWidgetSource:
    """Fetch Jellyfin sessions and map them to activity rows."""

    async def fetch(self, service: ServiceRecord | None, widget_kind: str, config: dict[str, Any]) -> dict[str, Any]:
        timeout = 10
        try:
            if service is None:
                return {"error": "Jellyfin widget is missing its service"}
            base_url = str(service.config.get("base_url") or "")
            api_key = str(service.secrets.get("api_key") or "")
            timeout = int(service.config.get("timeout_seconds") or 10)
            client = await asyncio.wait_for(
                asyncio.to_thread(JellyfinClient, base_url, api_key, timeout),
                timeout=timeout,
            )
            sessions = await asyncio.wait_for(
                asyncio.to_thread(client.sessions),
                timeout=timeout,
            )
            if widget_kind == "now_playing":
                sessions = [
                    s for s in sessions if s.get("NowPlayingItem") and not s.get("PlayState", {}).get("IsPaused", True)
                ]
            rows = _map_sessions_to_activity_rows(sessions)
            return {"sessions": rows}
        except asyncio.TimeoutError:
            return {"error": "Widget data fetch timed out"}
        except Exception as exc:
            logger.exception("jellyfin adapter failed")
            return {"error": f"Jellyfin data fetch failed: {exc}"}


class SshTaskWidgetSource:
    """Run a saved task on an SSH task runner instance and log the run."""

    async def fetch(self, service: ServiceRecord | None, widget_kind: str, config: dict[str, Any]) -> dict[str, Any]:
        timeout = 30
        try:
            if service is None:
                return {"error": "SSH task widget is missing its service"}
            store = get_settings_store()
            task_id = config.get("task_id") or ""
            if not task_id:
                return {"error": "task_id is required"}
            task = store.get_task(task_id)
            if not task:
                return {"error": f"Task {task_id} not found"}
            if not task.get("enabled", True):
                return {"error": "Task is disabled"}

            timeout = int(service.config.get("timeout_seconds") or 30)
            result = await asyncio.wait_for(
                asyncio.to_thread(run_saved_task, store, task, service),
                timeout=timeout,
            )
            return {"exit_status": result.exit_status, "stdout": result.stdout, "stderr": result.stderr}
        except asyncio.TimeoutError:
            _record_timeout(service, config, timeout)
            return {"error": "Widget data fetch timed out"}
        except Exception as exc:
            logger.exception("ssh_task adapter failed")
            return {"error": f"SSH task failed: {exc}"}


def _record_timeout(service: ServiceRecord | None, config: dict[str, Any], timeout: int) -> None:
    try:
        store = get_settings_store()
        store.record_service_task_run(
            {
                "task_id": str(config.get("task_id") or ""),
                "service_id": service.id if service else "",
                "status": "timeout",
                "duration_ms": timeout * 1000,
                "error": f"Task timed out after {timeout}s",
            }
        )
    except Exception:  # pragma: no cover - logging best-effort
        logger.exception("failed to record ssh task timeout")


class QbittorrentWidgetSource:
    """Fetch qBittorrent data for totals, active, and speed widgets."""

    async def fetch(self, service: ServiceRecord | None, widget_kind: str, config: dict[str, Any]) -> dict[str, Any]:
        try:
            if service is None:
                return {"error": "qBittorrent widget is missing its service"}
            base_url = str(service.config.get("base_url") or "")
            username = str(service.secrets.get("username") or "")
            password = str(service.secrets.get("password") or "")
            timeout = int(service.config.get("timeout_seconds") or 10)
            if not base_url or not username or not password:
                return {"error": "qBittorrent service is missing base_url, username, or password"}

            client = QbittorrentClient(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":
                by_state: dict[str, int] = {}
                for t in torrents.values():
                    state = str(t.get("state", "unknown"))
                    by_state[state] = by_state.get(state, 0) + 1
                return {"total": len(torrents), "by_state": by_state}

            if widget_kind == "active":
                active = [
                    {
                        "name": t.get("name"),
                        "state": t.get("state"),
                        "size": t.get("size"),
                        "progress": t.get("progress"),
                        "dl_speed": t.get("dlspeed"),
                        "up_speed": t.get("upspeed"),
                    }
                    for t in torrents.values()
                    if str(t.get("state", "")) in {"downloading", "uploading"}
                ]
                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:
            return {"error": "qBittorrent data fetch timed out"}
        except Exception as exc:
            logger.exception("qbittorrent adapter failed")
            return {"error": f"qBittorrent fetch failed: {exc}"}


# ---------------------------------------------------------------------------
# Registries
# ---------------------------------------------------------------------------

SERVICE_ADAPTERS: dict[str, WidgetSource] = {
    "prometheus": MetricSource(),
    "qbittorrent": QbittorrentWidgetSource(),
    "alertmanager": AlertmanagerWidgetSource(),
    "jellyfin": JellyfinWidgetSource(),
    "ssh_tasks": SshTaskWidgetSource(),
}

BUILTIN_ADAPTERS: dict[str, WidgetSource] = {
    "backups": BackupsWidgetSource(),
    "static": StaticWidgetSource(),
}


def get_service_adapter(service_type: str) -> WidgetSource | None:
    return SERVICE_ADAPTERS.get(service_type)


def get_builtin_adapter(kind: str) -> WidgetSource | None:
    return BUILTIN_ADAPTERS.get(kind)