"""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 functools import lru_cache
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 import jellyseerr_stats  # noqa: F401 — registers the Jellyseerr stats provider
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,
)
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."""

    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 60)
            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 as _timeout_error:
            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 60)
            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 as _timeout_error:
            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 60)
            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 as _timeout_error:
            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 as _timeout_error:
            _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")


@lru_cache(maxsize=16)
def _qbittorrent_client(cache_key: tuple[str, str, str, str, int]) -> QbittorrentClient:
    """Build (or reuse) a qBittorrent client for a service.

    Cached per (service_id, base_url, username, password, timeout) so the
    authenticated session/SID cookie persists across widget fetches. A
    credentials or URL change produces a new cache key, so stale clients are
    not reused after reconfiguration. Mirrors the Jellyfin client cache in
    dependencies._jellyfin_client_for.
    """
    _service_id, base_url, username, password, timeout = cache_key
    return QbittorrentClient(base_url, username, password, timeout=timeout)


_QBITTORRENT_DOWNLOAD_STATES = frozenset(
    {"downloading", "forceddl", "stalleddl", "queueddl", "metadl", "forcedmetadl", "allocating", "checkingdl"}
)
_QBITTORRENT_UPLOAD_STATES = frozenset({"uploading", "forcedup", "stalledup", "queuedup", "checkingup"})
_QBITTORRENT_OTHER_ACTIVE_STATES = frozenset({"checkingresumedata", "moving"})


def _qbit_torrent_direction(torrent: dict[str, Any]) -> str | None:
    """Return the transfer direction for active qBittorrent states or speeds."""
    state = str(torrent.get("state") or "").lower()
    if state in _QBITTORRENT_DOWNLOAD_STATES:
        return "downloading"
    if state in _QBITTORRENT_UPLOAD_STATES:
        return "uploading"
    if _safe_int(torrent.get("dlspeed")) > 0:
        return "downloading"
    if _safe_int(torrent.get("upspeed")) > 0:
        return "uploading"
    return None


def _qbit_torrent_is_active(torrent: dict[str, Any]) -> bool:
    state = str(torrent.get("state") or "").lower()
    return bool(_qbit_torrent_direction(torrent)) or state in _QBITTORRENT_OTHER_ACTIVE_STATES


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

            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 "")
            timeout = int(service.config.get("timeout_seconds") or 60)
            if not base_url or not username or not password:
                return {"error": "qBittorrent service is missing base_url, username, or password"}

            # Reuse one authenticated client per service so the SID cookie
            # persists across fetches and we don't re-login on every widget
            # poll. qBittorrent verifies passwords with slow PBKDF2 hashing,
            # so logging in on every fetch (3 widgets x frequent polls)
            # saturates its web thread pool and the reverse proxy returns 504
            # 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)
            torrents = data.get("torrents", {})

            if widget_kind == "totals":
                by_state: dict[str, int] = {}
                by_direction = {"downloading": 0, "uploading": 0}
                for torrent in torrents.values():
                    state = str(torrent.get("state") or "unknown")
                    by_state[state] = by_state.get(state, 0) + 1
                    direction = _qbit_torrent_direction(torrent)
                    if direction:
                        by_direction[direction] += 1
                return {
                    "total": len(torrents),
                    "by_state": by_state,
                    "by_direction": by_direction,
                }

            if widget_kind == "active":
                active = []
                for torrent in torrents.values():
                    if not _qbit_torrent_is_active(torrent):
                        continue
                    direction = _qbit_torrent_direction(torrent)
                    active.append(
                        {
                            "name": torrent.get("name"),
                            "state": torrent.get("state"),
                            "direction": direction,
                            "size": torrent.get("size"),
                            "progress": torrent.get("progress"),
                            "dl_speed": torrent.get("dlspeed"),
                            "up_speed": torrent.get("upspeed"),
                        }
                    )
                return {"torrents": active}

            return {"error": f"Unknown qBittorrent widget kind: {widget_kind}"}
        except asyncio.TimeoutError as _timeout_error:
            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)


class StatsWidgetSource:
    """Generic source for ``stat`` and ``stats_overview`` widgets.

    Dispatches to the service type's registered :class:`StatsProvider`, so any
    stats-provider service gets these two widget kinds for free. The widgets
    router routes widget_kind in {"stat", "stats_overview"} here regardless of
    service type.
    """

    async def fetch(self, service: ServiceRecord | None, widget_kind: str, config: dict[str, Any]) -> dict[str, Any]:
        if service is None:
            return {"error": "Stats widget is missing its service"}
        provider = get_stats_provider(service.service_type)
        if provider is None:
            return {"error": f"No stats provider for service type '{service.service_type}'"}
        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 as _timeout_error:
            return {"error": "Stats fetch timed out"}
        except Exception as exc:
            logger.exception("stats provider failed service=%s", service.id)
            return {"error": f"Stats fetch failed: {exc}"}
        if result.detail and not result.stats:
            return {"error": result.detail}
        if widget_kind == "stats_overview":
            return {
                "stats": [{"key": s.key, "label": s.label, "value": s.value} for s in result.stats],
                "recent": result.recent,
            }
        stat_key = str(config.get("stat") or "")
        match = next((s for s in result.stats if s.key == stat_key), None)
        if match is None:
            return {"error": f"Unknown stat '{stat_key}'"}
        return {"key": match.key, "label": match.label, "value": match.value}


_STATS_ADAPTER = StatsWidgetSource()


def get_stats_adapter() -> WidgetSource | None:
    return _STATS_ADAPTER
