"""Persistent application settings stored in a small SQLite database.

The store manages machine definitions, machine services, and per-machine
application configuration so the frontend can present local and remote targets
in the same UI.
"""

from __future__ import annotations

import json
import logging
import sqlite3
import time
import uuid
from io import StringIO
from pathlib import Path
from typing import Any

import paramiko

from media_library_viewer_api.models.widgets import _validate_config_keys

logger = logging.getLogger(__name__)

DEFAULT_SETTINGS_PATH = Path(".cache/media_library_viewer/settings.sqlite")
LOCAL_MACHINE_ID = "local"
DEFAULT_SERVICES = ["monitoring", "files"]


def _default_local_machine() -> dict[str, Any]:
    return {
        "id": LOCAL_MACHINE_ID,
        "name": "This machine",
        "mode": "local",
        "enabled": True,
        "services": list(DEFAULT_SERVICES),
        "host": "",
        "port": 22,
        "username": "",
        "key_directory": "",
        "key_name": "",
        "ssh_key_id": "",
        "ssh_private_key": "",
        "ssh_private_key_passphrase": "",
        "password": "",
        "node_exporter_enabled": False,
        "node_exporter_port": 9100,
        "node_exporter_scrape_host": "",
        "notes": "",
    }


class SettingsStore:
    """SQLite-backed settings store for machine definitions and history."""

    def __init__(self, db_path: Path | str = DEFAULT_SETTINGS_PATH):
        self.db_path = Path(db_path)
        self.db_path.parent.mkdir(parents=True, exist_ok=True)

    def connect(self) -> sqlite3.Connection:
        conn = sqlite3.connect(self.db_path, timeout=30)
        conn.row_factory = sqlite3.Row
        conn.execute("PRAGMA journal_mode=WAL")
        conn.execute("PRAGMA busy_timeout=30000")
        return conn

    def init_schema(self) -> None:
        with self.connect() as conn:
            conn.execute(
                """
                CREATE TABLE IF NOT EXISTS monitoring_machines (
                    id TEXT PRIMARY KEY,
                    name TEXT NOT NULL,
                    mode TEXT NOT NULL,
                    enabled INTEGER NOT NULL,
                    config_json TEXT NOT NULL,
                    created_at INTEGER NOT NULL,
                    updated_at INTEGER NOT NULL
                )
                """
            )
            conn.execute("CREATE INDEX IF NOT EXISTS idx_monitoring_machines_mode ON monitoring_machines(mode)")
            # The legacy SSH-scraping monitor (MonitoringPoller) was decommissioned;
            # metrics now live in Prometheus/node_exporter. Drop the orphan
            # table on startup so existing databases get a clean slate.
            conn.execute("DROP TABLE IF EXISTS monitoring_machine_actions")
            conn.execute(
                """
                CREATE TABLE IF NOT EXISTS ssh_keys (
                    id TEXT PRIMARY KEY,
                    name TEXT NOT NULL,
                    private_key TEXT NOT NULL,
                    passphrase TEXT NOT NULL,
                    public_key TEXT NOT NULL DEFAULT '',
                    fingerprint TEXT NOT NULL DEFAULT '',
                    notes TEXT NOT NULL,
                    created_at INTEGER NOT NULL,
                    updated_at INTEGER NOT NULL
                )
                """
            )
            existing_key_columns = {row[1] for row in conn.execute("PRAGMA table_info(ssh_keys)").fetchall()}
            if "public_key" not in existing_key_columns:
                conn.execute("ALTER TABLE ssh_keys ADD COLUMN public_key TEXT NOT NULL DEFAULT ''")
            if "fingerprint" not in existing_key_columns:
                conn.execute("ALTER TABLE ssh_keys ADD COLUMN fingerprint TEXT NOT NULL DEFAULT ''")
            conn.execute(
                """
                CREATE TABLE IF NOT EXISTS saved_tasks (
                    id TEXT PRIMARY KEY,
                    name TEXT NOT NULL,
                    task_type TEXT NOT NULL,
                    content TEXT NOT NULL,
                    enabled INTEGER NOT NULL,
                    default_service_id TEXT NOT NULL,
                    notes TEXT NOT NULL,
                    created_at INTEGER NOT NULL,
                    updated_at INTEGER NOT NULL
                )
                """
            )
            conn.execute("CREATE INDEX IF NOT EXISTS idx_saved_tasks_name ON saved_tasks(name)")
            # saved_tasks.default_machine_id → default_service_id (saved tasks now
            # target ssh_tasks service instances). Migrate existing columns.
            saved_tasks_cols = {row[1] for row in conn.execute("PRAGMA table_info(saved_tasks)").fetchall()}
            if "default_service_id" not in saved_tasks_cols and "default_machine_id" in saved_tasks_cols:
                conn.execute("ALTER TABLE saved_tasks RENAME COLUMN default_machine_id TO default_service_id")
            # Run history for saved tasks now lives in service_task_runs; the
            # legacy machine-based table is dropped.
            conn.execute("DROP TABLE IF EXISTS saved_task_runs")
            conn.execute(
                """
                CREATE TABLE IF NOT EXISTS dashboard_shortcuts (
                    id TEXT PRIMARY KEY,
                    label TEXT NOT NULL,
                    shortcut_type TEXT NOT NULL,
                    target_json TEXT NOT NULL,
                    enabled INTEGER NOT NULL,
                    created_at INTEGER NOT NULL,
                    updated_at INTEGER NOT NULL
                )
                """
            )
            conn.execute(
                "CREATE INDEX IF NOT EXISTS idx_dashboard_shortcuts_type ON dashboard_shortcuts(shortcut_type)"
            )
            conn.execute(
                """
                CREATE TABLE IF NOT EXISTS dashboard_widgets (
                    id TEXT PRIMARY KEY,
                    addon_id TEXT NOT NULL,
                    widget_type TEXT NOT NULL,
                    title TEXT NOT NULL,
                    config_json TEXT NOT NULL DEFAULT '{}',
                    enabled INTEGER NOT NULL DEFAULT 1,
                    sort_order INTEGER NOT NULL DEFAULT 0,
                    created_at INTEGER NOT NULL,
                    updated_at INTEGER NOT NULL
                )
                """
            )
            conn.execute("CREATE INDEX IF NOT EXISTS idx_dashboard_widgets_sort ON dashboard_widgets(sort_order)")
            widget_cols = {row[1] for row in conn.execute("PRAGMA table_info(dashboard_widgets)").fetchall()}
            if "service_id" not in widget_cols:
                conn.execute("ALTER TABLE dashboard_widgets ADD COLUMN service_id TEXT")
            if "widget_kind" not in widget_cols:
                conn.execute("ALTER TABLE dashboard_widgets ADD COLUMN widget_kind TEXT")
            conn.execute(
                """
                CREATE TABLE IF NOT EXISTS widget_references (
                    id TEXT PRIMARY KEY,
                    dashboard_scope TEXT NOT NULL,
                    widget_id TEXT NOT NULL,
                    sort_order INTEGER DEFAULT 0,
                    created_at INTEGER NOT NULL,
                    FOREIGN KEY (widget_id) REFERENCES dashboard_widgets(id) ON DELETE CASCADE
                )
                """
            )
            conn.execute("CREATE INDEX IF NOT EXISTS idx_widget_references_scope ON widget_references(dashboard_scope)")
            conn.execute("""
                CREATE TABLE IF NOT EXISTS backup_jobs (
                    id TEXT PRIMARY KEY,
                    name TEXT NOT NULL UNIQUE,
                    source TEXT,
                    target TEXT,
                    schedule_interval_seconds INTEGER,
                    created_at INTEGER NOT NULL
                )
            """)
            backup_job_cols = {col[1] for col in conn.execute("PRAGMA table_info(backup_jobs)").fetchall()}
            if "service_id" not in backup_job_cols:
                conn.execute("ALTER TABLE backup_jobs ADD COLUMN service_id TEXT")
            conn.execute("""
                CREATE TABLE IF NOT EXISTS backup_runs (
                    id TEXT PRIMARY KEY,
                    job_id TEXT NOT NULL,
                    started_at INTEGER NOT NULL,
                    ended_at INTEGER,
                    status TEXT NOT NULL,
                    bytes_transferred INTEGER,
                    duration_ms INTEGER,
                    error_message TEXT,
                    details_json TEXT,
                    created_at INTEGER NOT NULL,
                    FOREIGN KEY (job_id) REFERENCES backup_jobs(id)
                )
            """)
            conn.execute("CREATE INDEX IF NOT EXISTS idx_backup_runs_job_id ON backup_runs(job_id)")
            conn.execute("CREATE INDEX IF NOT EXISTS idx_backup_runs_status ON backup_runs(status)")
            conn.execute("""
                CREATE TABLE IF NOT EXISTS backup_alerts (
                    id TEXT PRIMARY KEY,
                    job_id TEXT NOT NULL,
                    run_id TEXT,
                    alert_type TEXT NOT NULL,
                    severity TEXT NOT NULL,
                    message TEXT NOT NULL,
                    acknowledged INTEGER NOT NULL DEFAULT 0,
                    resolved_at INTEGER,
                    created_at INTEGER NOT NULL,
                    FOREIGN KEY (job_id) REFERENCES backup_jobs(id),
                    FOREIGN KEY (run_id) REFERENCES backup_runs(id)
                )
            """)
            conn.execute("CREATE INDEX IF NOT EXISTS idx_backup_alerts_job_id ON backup_alerts(job_id)")
            conn.execute("CREATE INDEX IF NOT EXISTS idx_backup_alerts_acknowledged ON backup_alerts(acknowledged)")
            conn.execute(
                """
                CREATE TABLE IF NOT EXISTS services (
                    id TEXT PRIMARY KEY,
                    service_type TEXT NOT NULL,
                    name TEXT NOT NULL,
                    config_json TEXT NOT NULL DEFAULT '{}',
                    secrets_json TEXT NOT NULL DEFAULT '{}',
                    enabled INTEGER NOT NULL DEFAULT 1,
                    created_at INTEGER NOT NULL,
                    updated_at INTEGER NOT NULL
                )
                """
            )
            conn.execute("CREATE INDEX IF NOT EXISTS idx_services_type ON services(service_type)")
            conn.execute(
                """
                CREATE TABLE IF NOT EXISTS service_task_runs (
                    id TEXT PRIMARY KEY,
                    task_id TEXT NOT NULL,
                    service_id TEXT NOT NULL,
                    status TEXT NOT NULL,
                    exit_status INTEGER,
                    duration_ms INTEGER,
                    stdout_tail TEXT NOT NULL DEFAULT '',
                    stderr_tail TEXT NOT NULL DEFAULT '',
                    error TEXT NOT NULL DEFAULT '',
                    created_at INTEGER NOT NULL
                )
                """
            )
            conn.execute(
                "CREATE INDEX IF NOT EXISTS idx_service_task_runs_service "
                "ON service_task_runs(service_id, created_at DESC)"
            )
            conn.execute(
                "CREATE INDEX IF NOT EXISTS idx_service_task_runs_task ON service_task_runs(task_id, created_at DESC)"
            )
            conn.execute(
                """
                CREATE TABLE IF NOT EXISTS named_dashboards (
                    id TEXT PRIMARY KEY,
                    label TEXT NOT NULL,
                    slug TEXT NOT NULL UNIQUE,
                    sort_order INTEGER NOT NULL DEFAULT 0,
                    payload_json TEXT NOT NULL DEFAULT '{}',
                    created_at INTEGER NOT NULL,
                    updated_at INTEGER NOT NULL
                )
                """
            )

    @staticmethod
    def _normalize_services(value: Any, fallback: list[str] | None = None) -> list[str]:
        if isinstance(value, str):
            items = [part.strip() for part in value.split(",")]
        elif isinstance(value, list):
            items = [str(part).strip() for part in value]
        else:
            items = list(fallback or DEFAULT_SERVICES)
        services = [item for item in items if item]
        if not services:
            services = list(fallback or DEFAULT_SERVICES)
        deduped: list[str] = []
        for service in services:
            if service not in deduped:
                deduped.append(service)
        return deduped

    def _row_to_machine(self, row: sqlite3.Row) -> dict[str, Any]:
        data = json.loads(row["config_json"])
        default_services = DEFAULT_SERVICES if row["id"] == LOCAL_MACHINE_ID else []
        services = self._normalize_services(data.get("services"), default_services)
        return {
            "id": row["id"],
            "name": row["name"],
            "mode": row["mode"],
            "enabled": bool(row["enabled"]),
            "services": services,
            "host": data.get("host", ""),
            "port": int(data.get("port", 22) or 22),
            "username": data.get("username", ""),
            "key_directory": data.get("key_directory", ""),
            "key_name": data.get("key_name", ""),
            "ssh_key_id": data.get("ssh_key_id", ""),
            "ssh_private_key_set": bool(data.get("ssh_private_key")),
            "ssh_private_key_passphrase_set": bool(data.get("ssh_private_key_passphrase")),
            "password_set": bool(data.get("password")),
            "node_exporter_enabled": bool(data.get("node_exporter_enabled", False)),
            "node_exporter_port": int(data.get("node_exporter_port", 9100) or 9100),
            "node_exporter_scrape_host": data.get("node_exporter_scrape_host", ""),
            "notes": data.get("notes", ""),
            "created_at": row["created_at"],
            "updated_at": row["updated_at"],
        }

    def _normalize_machine_payload(
        self,
        payload: dict[str, Any],
        machine_id: str | None = None,
    ) -> dict[str, Any]:
        current = self.get_machine(machine_id) if machine_id else None
        machine_id = str(payload.get("id") or machine_id or uuid.uuid4().hex[:12]).strip() or uuid.uuid4().hex[:12]
        mode = str(payload.get("mode") or (current or {}).get("mode") or "local").strip().lower()
        if mode not in {"local", "ssh"}:
            mode = "local"
        enabled = bool(payload.get("enabled", (current or {}).get("enabled", True)))
        name = str(payload.get("name") or (current or {}).get("name") or "").strip() or (
            "This machine" if mode == "local" else machine_id
        )
        services = self._normalize_services(payload.get("services"), (current or {}).get("services", []))

        def _current_str(field: str, default: str = "") -> str:
            return str(
                payload.get(field) if payload.get(field) is not None else (current or {}).get(field, default) or default
            ).strip()

        host = _current_str("host")
        port = int(payload.get("port") or (current or {}).get("port", 22) or 22)
        username = _current_str("username")
        key_directory = _current_str("key_directory")
        key_name = _current_str("key_name")
        ssh_key_id = _current_str("ssh_key_id")
        ssh_private_key = payload.get("ssh_private_key")
        if ssh_private_key in (None, ""):
            ssh_private_key = (current or {}).get("ssh_private_key", "")
        ssh_private_key = str(ssh_private_key or "")
        ssh_private_key_passphrase = payload.get("ssh_private_key_passphrase")
        if ssh_private_key_passphrase in (None, ""):
            ssh_private_key_passphrase = (current or {}).get("ssh_private_key_passphrase", "")
        ssh_private_key_passphrase = str(ssh_private_key_passphrase or "")
        password = payload.get("password")
        if password in (None, ""):
            password = (current or {}).get("password", "")
        password = str(password or "")
        node_exporter_enabled = bool(
            payload.get("node_exporter_enabled")
            if payload.get("node_exporter_enabled") is not None
            else (current or {}).get("node_exporter_enabled", False)
        )
        node_exporter_port_raw = payload.get("node_exporter_port")
        if node_exporter_port_raw is None:
            node_exporter_port_raw = (current or {}).get("node_exporter_port", 9100)
        node_exporter_port = int(node_exporter_port_raw or 9100)
        node_exporter_scrape_host = _current_str("node_exporter_scrape_host")
        notes = _current_str("notes")
        if mode == "local":
            host = host or "localhost"
            username = username or ""
        return {
            "id": machine_id,
            "name": name,
            "mode": mode,
            "enabled": enabled,
            "services": services,
            "host": host,
            "port": port,
            "username": username,
            "key_directory": key_directory,
            "key_name": key_name,
            "ssh_key_id": ssh_key_id,
            "ssh_private_key": ssh_private_key,
            "ssh_private_key_passphrase": ssh_private_key_passphrase,
            "password": password,
            "node_exporter_enabled": node_exporter_enabled,
            "node_exporter_port": node_exporter_port,
            "node_exporter_scrape_host": node_exporter_scrape_host,
            "notes": notes,
        }

    def _seed_local_machine(self) -> None:
        """Seed the default local machine if none exists."""
        machine = _default_local_machine()
        now = int(time.time())
        config = {
            "services": machine["services"],
            "host": machine["host"],
            "port": machine["port"],
            "username": machine["username"],
            "key_directory": machine["key_directory"],
            "key_name": machine["key_name"],
            "ssh_key_id": machine.get("ssh_key_id", ""),
            "ssh_private_key": "",
            "ssh_private_key_passphrase": "",
            "password": "",
            "node_exporter_enabled": machine["node_exporter_enabled"],
            "node_exporter_port": machine["node_exporter_port"],
            "node_exporter_scrape_host": machine["node_exporter_scrape_host"],
            "notes": machine["notes"],
        }
        with self.connect() as conn:
            conn.execute(
                """
                INSERT INTO monitoring_machines (id, name, mode, enabled, config_json, created_at, updated_at)
                VALUES (?, ?, ?, ?, ?, ?, ?)
                """,
                (
                    machine["id"],
                    machine["name"],
                    machine["mode"],
                    1,
                    json.dumps(config),
                    now,
                    now,
                ),
            )

    def _seed_dashboard_widgets(self) -> None:
        """Default widget seeding was removed.

        Widgets are now service-bound (or built-in). A fresh install starts with
        no widgets; the user configures services and adds widgets from the UI.
        Kept as a no-op so :meth:`ensure_defaults` callers are unchanged.
        """
        return None

    def ensure_defaults(self) -> None:
        self.init_schema()
        with self.connect() as conn:
            row = conn.execute("SELECT COUNT(*) FROM monitoring_machines").fetchone()
            if not row or int(row[0]) == 0:
                self._seed_local_machine()
        self._migrate_jellyseerr_into_jellyfin()

    def _migrate_jellyseerr_into_jellyfin(self) -> None:
        """Absorb standalone ``jellyseerr`` services into their paired Jellyfin.

        Idempotent: once no ``jellyseerr`` rows remain the method is a no-op.
        Pairing policy: exactly-one Jellyfin merges; multiple picks the first
        Jellyfin whose ``jellyseerr_url`` is still empty; no Jellyfin or all
        paired -> drop with a logged warning.
        """
        from media_library_viewer_api.services.secrets import decrypt_value

        self.init_schema()
        jellyseerr_rows: list[sqlite3.Row] = []
        with self.connect() as conn:
            jellyseerr_rows = conn.execute(
                "SELECT * FROM services WHERE service_type = 'jellyseerr' ORDER BY name ASC"
            ).fetchall()
        if not jellyseerr_rows:
            return

        jellyfin_rows = self.list_services("jellyfin")
        for js_row in jellyseerr_rows:
            js_config = json.loads(js_row["config_json"] or "{}")
            js_secrets = json.loads(js_row["secrets_json"] or "{}")
            js_url = str(js_config.get("base_url", "")).strip()
            js_api_key = str(js_secrets.get("api_key", "")).strip()
            # Decrypt the api_key (secrets are stored encrypted; config is plaintext).
            if js_api_key:
                try:
                    js_api_key = decrypt_value(js_api_key)
                except Exception:
                    logger.warning("could not decrypt jellyseerr api_key for %r", js_row["name"])
                    js_api_key = ""
            js_name = js_row["name"]

            target = None
            if len(jellyfin_rows) == 1:
                target = jellyfin_rows[0]
            elif len(jellyfin_rows) > 1:
                for jf in jellyfin_rows:
                    if not str(jf["config"].get("jellyseerr_url", "")).strip():
                        target = jf
                        break

            if target:
                merged_config = dict(target["config"])
                merged_config["jellyseerr_url"] = js_url
                merged_config["jellyseerr_api_key"] = js_api_key
                self.upsert_service(
                    {
                        "id": target["id"],
                        "service_type": "jellyfin",
                        "name": target["name"],
                        "config": merged_config,
                        "enabled": target["enabled"],
                    },
                    secret_values={"api_key": str(target["secrets"].get("api_key", ""))},
                )
                logger.info("migrated jellyseerr service %r into jellyfin service %r", js_name, target["name"])
            else:
                logger.warning(
                    "dropped unpaired jellyseerr service %r; reconfigure manually on the Jellyfin instance",
                    js_name,
                )

            with self.connect() as conn:
                conn.execute("DELETE FROM services WHERE id = ?", (js_row["id"],))
                conn.commit()

    def list_machines(self) -> list[dict[str, Any]]:
        self.init_schema()
        with self.connect() as conn:
            rows = conn.execute(
                "SELECT * FROM monitoring_machines ORDER BY CASE WHEN id = ? THEN 0 ELSE 1 END, name COLLATE NOCASE",
                (LOCAL_MACHINE_ID,),
            ).fetchall()
        return [self._row_to_machine(row) for row in rows]

    def get_machine(self, machine_id: str | None) -> dict[str, Any] | None:
        if not machine_id:
            return None
        self.init_schema()
        with self.connect() as conn:
            row = conn.execute("SELECT * FROM monitoring_machines WHERE id = ?", (machine_id,)).fetchone()
        return self._row_to_machine(row) if row else None

    def get_machine_config(self, machine_id: str | None) -> dict[str, Any] | None:
        """Return the full machine config including secrets."""
        if not machine_id:
            return None
        self.init_schema()
        with self.connect() as conn:
            row = conn.execute("SELECT * FROM monitoring_machines WHERE id = ?", (machine_id,)).fetchone()
        if not row:
            return None
        data = json.loads(row["config_json"])
        return {
            "id": row["id"],
            "name": row["name"],
            "mode": row["mode"],
            "enabled": bool(row["enabled"]),
            "services": self._normalize_services(
                data.get("services"),
                DEFAULT_SERVICES if row["id"] == LOCAL_MACHINE_ID else [],
            ),
            "host": data.get("host", ""),
            "port": int(data.get("port", 22) or 22),
            "username": data.get("username", ""),
            "key_directory": data.get("key_directory", ""),
            "key_name": data.get("key_name", ""),
            "ssh_key_id": data.get("ssh_key_id", ""),
            "ssh_private_key": data.get("ssh_private_key", ""),
            "ssh_private_key_passphrase": data.get("ssh_private_key_passphrase", ""),
            "password": data.get("password", ""),
            "node_exporter_enabled": bool(data.get("node_exporter_enabled", False)),
            "node_exporter_port": int(data.get("node_exporter_port", 9100) or 9100),
            "node_exporter_scrape_host": data.get("node_exporter_scrape_host", ""),
            "notes": data.get("notes", ""),
        }

    def list_machines_for_service(self, service: str) -> list[dict[str, Any]]:
        return [
            machine
            for machine in self.list_machines()
            if service in machine.get("services", []) and machine.get("enabled")
        ]

    def get_machine_for_service(self, service: str, machine_id: str | None = None) -> dict[str, Any] | None:
        if machine_id:
            machine = self.get_machine(machine_id)
            if machine and service in machine.get("services", []) and machine.get("enabled"):
                return machine
            return machine if machine else None
        machines = self.list_machines_for_service(service)
        return machines[0] if machines else None

    def upsert_machine(self, payload: dict[str, Any], machine_id: str | None = None) -> dict[str, Any]:
        self.init_schema()
        machine = self._normalize_machine_payload(payload, machine_id)
        now = int(time.time())
        config = {
            "services": machine["services"],
            "host": machine["host"],
            "port": machine["port"],
            "username": machine["username"],
            "key_directory": machine["key_directory"],
            "key_name": machine["key_name"],
            "ssh_key_id": machine.get("ssh_key_id", ""),
            "ssh_private_key": machine["ssh_private_key"],
            "ssh_private_key_passphrase": machine["ssh_private_key_passphrase"],
            "password": machine["password"],
            "node_exporter_enabled": machine["node_exporter_enabled"],
            "node_exporter_port": machine["node_exporter_port"],
            "node_exporter_scrape_host": machine["node_exporter_scrape_host"],
            "notes": machine["notes"],
        }
        with self.connect() as conn:
            existing = conn.execute(
                "SELECT created_at FROM monitoring_machines WHERE id = ?",
                (machine["id"],),
            ).fetchone()
            created_at = int(existing[0]) if existing else now
            conn.execute(
                """
                INSERT INTO monitoring_machines (id, name, mode, enabled, config_json, created_at, updated_at)
                VALUES (?, ?, ?, ?, ?, ?, ?)
                ON CONFLICT(id) DO UPDATE SET
                    name = excluded.name,
                    mode = excluded.mode,
                    enabled = excluded.enabled,
                    config_json = excluded.config_json,
                    updated_at = excluded.updated_at
                """,
                (
                    machine["id"],
                    machine["name"],
                    machine["mode"],
                    1 if machine["enabled"] else 0,
                    json.dumps(config),
                    created_at,
                    now,
                ),
            )
        return self.get_machine(machine["id"]) or machine

    def delete_machine(self, machine_id: str) -> None:
        self.init_schema()
        with self.connect() as conn:
            conn.execute("DELETE FROM monitoring_machines WHERE id = ?", (machine_id,))

    @staticmethod
    def _private_key_summary(private_key: str) -> dict[str, str]:
        if not private_key:
            return {"public_key": "", "fingerprint": ""}
        key_classes = [paramiko.Ed25519Key, paramiko.RSAKey, paramiko.ECDSAKey]
        for key_class in key_classes:
            try:
                key = key_class.from_private_key(StringIO(private_key))
                fingerprint = ":".join(f"{b:02x}" for b in key.get_fingerprint())
                return {"public_key": f"{key.get_name()} {key.get_base64()}", "fingerprint": fingerprint}
            except Exception:
                continue
        return {"public_key": "", "fingerprint": ""}

    def _row_to_ssh_key(self, row: sqlite3.Row, usage_count: int = 0) -> dict[str, Any]:
        summary = self._private_key_summary(str(row["private_key"] or ""))
        public_key = str(row["public_key"] or summary["public_key"] or "")
        fingerprint = str(row["fingerprint"] or summary["fingerprint"] or "")
        return {
            "id": row["id"],
            "name": row["name"],
            "private_key_set": bool(row["private_key"]),
            "passphrase_set": bool(row["passphrase"]),
            "public_key": public_key,
            "fingerprint": fingerprint,
            "usage_count": usage_count,
            "notes": row["notes"],
            "created_at": row["created_at"],
            "updated_at": row["updated_at"],
        }

    def _normalize_ssh_key_payload(self, payload: dict[str, Any], key_id: str | None = None) -> dict[str, Any]:
        current = self.get_ssh_key(key_id) if key_id else None
        key_id = str(payload.get("id") or key_id or uuid.uuid4().hex[:12]).strip() or uuid.uuid4().hex[:12]
        name = str(payload.get("name") or (current or {}).get("name") or key_id).strip() or key_id
        private_key = payload.get("private_key")
        if private_key in (None, ""):
            private_key = (current or {}).get("private_key", "")
        private_key = str(private_key or "")
        passphrase = payload.get("passphrase")
        if passphrase in (None, ""):
            passphrase = (current or {}).get("passphrase", "")
        passphrase = str(passphrase or "")
        notes = str(
            payload.get("notes") if payload.get("notes") is not None else (current or {}).get("notes", "") or ""
        ).strip()
        summary = self._private_key_summary(private_key)
        public_key = str(
            payload.get("public_key")
            if payload.get("public_key") is not None
            else (current or {}).get("public_key", "") or summary["public_key"] or ""
        ).strip()
        fingerprint = str(
            payload.get("fingerprint")
            if payload.get("fingerprint") is not None
            else (current or {}).get("fingerprint", "") or summary["fingerprint"] or ""
        ).strip()
        return {
            "id": key_id,
            "name": name,
            "private_key": private_key,
            "passphrase": passphrase,
            "public_key": public_key,
            "fingerprint": fingerprint,
            "notes": notes,
        }

    def list_ssh_keys(self) -> list[dict[str, Any]]:
        self.init_schema()
        machines = self.list_machines()
        usage_counts: dict[str, int] = {}
        for machine in machines:
            ssh_key_id = str(machine.get("ssh_key_id") or "").strip()
            if ssh_key_id:
                usage_counts[ssh_key_id] = usage_counts.get(ssh_key_id, 0) + 1
        with self.connect() as conn:
            rows = conn.execute("SELECT * FROM ssh_keys ORDER BY name COLLATE NOCASE").fetchall()
        return [self._row_to_ssh_key(row, usage_counts.get(row["id"], 0)) for row in rows]

    def get_ssh_key(self, key_id: str | None) -> dict[str, Any] | None:
        if not key_id:
            return None
        self.init_schema()
        with self.connect() as conn:
            row = conn.execute("SELECT * FROM ssh_keys WHERE id = ?", (key_id,)).fetchone()
        if not row:
            return None
        summary = self._private_key_summary(str(row["private_key"] or ""))
        return {
            "id": row["id"],
            "name": row["name"],
            "private_key": row["private_key"],
            "passphrase": row["passphrase"],
            "notes": row["notes"],
            "public_key": str(row["public_key"] or summary["public_key"] or ""),
            "fingerprint": str(row["fingerprint"] or summary["fingerprint"] or ""),
        }

    def upsert_ssh_key(self, payload: dict[str, Any], key_id: str | None = None) -> dict[str, Any]:
        self.init_schema()
        key = self._normalize_ssh_key_payload(payload, key_id)
        now = int(time.time())
        with self.connect() as conn:
            existing = conn.execute("SELECT created_at FROM ssh_keys WHERE id = ?", (key["id"],)).fetchone()
            created_at = int(existing[0]) if existing else now
            conn.execute(
                """
                INSERT INTO ssh_keys (
                    id, name, private_key, passphrase, public_key, fingerprint,
                    notes, created_at, updated_at
                )
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
                ON CONFLICT(id) DO UPDATE SET
                    name = excluded.name,
                    private_key = excluded.private_key,
                    passphrase = excluded.passphrase,
                    public_key = excluded.public_key,
                    fingerprint = excluded.fingerprint,
                    notes = excluded.notes,
                    updated_at = excluded.updated_at
                """,
                (
                    key["id"],
                    key["name"],
                    key["private_key"],
                    key["passphrase"],
                    key["public_key"],
                    key["fingerprint"],
                    key["notes"],
                    created_at,
                    now,
                ),
            )
        return self.get_ssh_key(key["id"]) or key

    def delete_ssh_key(self, key_id: str) -> None:
        self.init_schema()
        with self.connect() as conn:
            conn.execute("DELETE FROM ssh_keys WHERE id = ?", (key_id,))

    def _row_to_task(self, row: sqlite3.Row) -> dict[str, Any]:
        return {
            "id": row["id"],
            "name": row["name"],
            "task_type": row["task_type"],
            "content": row["content"],
            "enabled": bool(row["enabled"]),
            "default_service_id": row["default_service_id"],
            "notes": row["notes"],
            "created_at": row["created_at"],
            "updated_at": row["updated_at"],
        }

    def _normalize_task_payload(self, payload: dict[str, Any], task_id: str | None = None) -> dict[str, Any]:
        current = self.get_task(task_id) if task_id else None
        task_id = str(payload.get("id") or task_id or uuid.uuid4().hex[:12]).strip() or uuid.uuid4().hex[:12]
        name = str(payload.get("name") or (current or {}).get("name") or task_id).strip() or task_id
        task_type = str(payload.get("task_type") or (current or {}).get("task_type") or "shell").strip().lower()
        if task_type not in {"shell", "python"}:
            task_type = "shell"
        content = str(
            payload.get("content") if payload.get("content") is not None else (current or {}).get("content", "") or ""
        )
        enabled = bool(payload.get("enabled", (current or {}).get("enabled", True)))
        default_service_id = str(
            payload.get("default_service_id")
            if payload.get("default_service_id") is not None
            else (current or {}).get("default_service_id", "") or ""
        ).strip()
        notes = str(
            payload.get("notes") if payload.get("notes") is not None else (current or {}).get("notes", "") or ""
        ).strip()
        return {
            "id": task_id,
            "name": name,
            "task_type": task_type,
            "content": content,
            "enabled": enabled,
            "default_service_id": default_service_id,
            "notes": notes,
        }

    def list_tasks(self) -> list[dict[str, Any]]:
        self.init_schema()
        with self.connect() as conn:
            rows = conn.execute("SELECT * FROM saved_tasks ORDER BY name COLLATE NOCASE").fetchall()
        return [self._row_to_task(row) for row in rows]

    def get_task(self, task_id: str | None) -> dict[str, Any] | None:
        if not task_id:
            return None
        self.init_schema()
        with self.connect() as conn:
            row = conn.execute("SELECT * FROM saved_tasks WHERE id = ?", (task_id,)).fetchone()
        return self._row_to_task(row) if row else None

    def upsert_task(self, payload: dict[str, Any], task_id: str | None = None) -> dict[str, Any]:
        self.init_schema()
        task = self._normalize_task_payload(payload, task_id)
        now = int(time.time())
        with self.connect() as conn:
            existing = conn.execute("SELECT created_at FROM saved_tasks WHERE id = ?", (task["id"],)).fetchone()
            created_at = int(existing[0]) if existing else now
            conn.execute(
                """
                INSERT INTO saved_tasks (
                    id, name, task_type, content, enabled, default_service_id,
                    notes, created_at, updated_at
                )
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
                ON CONFLICT(id) DO UPDATE SET
                    name = excluded.name,
                    task_type = excluded.task_type,
                    content = excluded.content,
                    enabled = excluded.enabled,
                    default_service_id = excluded.default_service_id,
                    notes = excluded.notes,
                    updated_at = excluded.updated_at
                """,
                (
                    task["id"],
                    task["name"],
                    task["task_type"],
                    task["content"],
                    1 if task["enabled"] else 0,
                    task["default_service_id"],
                    task["notes"],
                    created_at,
                    now,
                ),
            )
        return self.get_task(task["id"]) or task

    def delete_task(self, task_id: str) -> None:
        self.init_schema()
        with self.connect() as conn:
            conn.execute("DELETE FROM saved_tasks WHERE id = ?", (task_id,))

    def _row_to_shortcut(self, row: sqlite3.Row) -> dict[str, Any]:
        target = json.loads(row["target_json"] or "{}")
        return {
            "id": row["id"],
            "label": row["label"],
            "shortcut_type": row["shortcut_type"],
            "enabled": bool(row["enabled"]),
            "icon": target.get("icon", ""),
            "url": target.get("url", ""),
            "task_id": target.get("task_id", ""),
            "machine_id": target.get("machine_id", ""),
            "user_id": target.get("user_id", ""),
            "notes": target.get("notes", ""),
            "created_at": row["created_at"],
            "updated_at": row["updated_at"],
        }

    def _normalize_shortcut_payload(self, payload: dict[str, Any], shortcut_id: str | None = None) -> dict[str, Any]:
        current = self.get_shortcut(shortcut_id) if shortcut_id else None
        shortcut_id = str(payload.get("id") or shortcut_id or uuid.uuid4().hex[:12]).strip() or uuid.uuid4().hex[:12]
        shortcut_type = (
            str(payload.get("shortcut_type") or (current or {}).get("shortcut_type") or "website").strip().lower()
        )
        if shortcut_type not in {"website", "action", "user"}:
            shortcut_type = "website"
        label = str(payload.get("label") or (current or {}).get("label") or "").strip() or shortcut_id
        enabled = bool(payload.get("enabled", (current or {}).get("enabled", True)))

        def _field(field: str, default: str = "") -> str:
            return str(
                payload.get(field) if payload.get(field) is not None else (current or {}).get(field, default) or default
            ).strip()

        icon = _field("icon")
        url = _field("url")
        task_id = _field("task_id")
        machine_id = _field("machine_id")
        user_id = _field("user_id")
        notes = _field("notes")
        return {
            "id": shortcut_id,
            "label": label,
            "shortcut_type": shortcut_type,
            "enabled": enabled,
            "target": {
                "icon": icon,
                "url": url,
                "task_id": task_id,
                "machine_id": machine_id,
                "user_id": user_id,
                "notes": notes,
            },
        }

    def list_shortcuts(self) -> list[dict[str, Any]]:
        self.init_schema()
        with self.connect() as conn:
            rows = conn.execute("SELECT * FROM dashboard_shortcuts ORDER BY label COLLATE NOCASE").fetchall()
        return [self._row_to_shortcut(row) for row in rows]

    def get_shortcut(self, shortcut_id: str | None) -> dict[str, Any] | None:
        if not shortcut_id:
            return None
        self.init_schema()
        with self.connect() as conn:
            row = conn.execute("SELECT * FROM dashboard_shortcuts WHERE id = ?", (shortcut_id,)).fetchone()
        return self._row_to_shortcut(row) if row else None

    def upsert_shortcut(self, payload: dict[str, Any], shortcut_id: str | None = None) -> dict[str, Any]:
        self.init_schema()
        shortcut = self._normalize_shortcut_payload(payload, shortcut_id)
        now = int(time.time())
        with self.connect() as conn:
            existing = conn.execute(
                "SELECT created_at FROM dashboard_shortcuts WHERE id = ?",
                (shortcut["id"],),
            ).fetchone()
            created_at = int(existing[0]) if existing else now
            conn.execute(
                """
                INSERT INTO dashboard_shortcuts (id, label, shortcut_type, target_json, enabled, created_at, updated_at)
                VALUES (?, ?, ?, ?, ?, ?, ?)
                ON CONFLICT(id) DO UPDATE SET
                    label = excluded.label,
                    shortcut_type = excluded.shortcut_type,
                    target_json = excluded.target_json,
                    enabled = excluded.enabled,
                    updated_at = excluded.updated_at
                """,
                (
                    shortcut["id"],
                    shortcut["label"],
                    shortcut["shortcut_type"],
                    json.dumps(shortcut["target"]),
                    1 if shortcut["enabled"] else 0,
                    created_at,
                    now,
                ),
            )
        return self.get_shortcut(shortcut["id"]) or shortcut

    def delete_shortcut(self, shortcut_id: str) -> None:
        self.init_schema()
        with self.connect() as conn:
            conn.execute("DELETE FROM dashboard_shortcuts WHERE id = ?", (shortcut_id,))

    # ------------------------------------------------------------------
    # Backup jobs
    # ------------------------------------------------------------------

    def _row_to_job(self, row: sqlite3.Row) -> dict[str, Any]:
        return {
            "id": row["id"],
            "name": row["name"],
            "source": row["source"],
            "target": row["target"],
            "schedule_interval_seconds": row["schedule_interval_seconds"],
            "service_id": row["service_id"],
            "created_at": row["created_at"],
        }

    def _normalize_backup_job_payload(self, payload: dict[str, Any], job_id: str | None = None) -> dict[str, Any]:
        current = self.get_backup_job(job_id) if job_id else None
        job_id = str(payload.get("id") or job_id or uuid.uuid4().hex[:12]).strip() or uuid.uuid4().hex[:12]
        name = str(payload.get("name") or (current or {}).get("name") or job_id).strip() or job_id
        source = str(
            payload.get("source") if payload.get("source") is not None else (current or {}).get("source", "") or ""
        ).strip()
        target = str(
            payload.get("target") if payload.get("target") is not None else (current or {}).get("target", "") or ""
        ).strip()
        schedule_interval_seconds = payload.get("schedule_interval_seconds")
        if schedule_interval_seconds is None:
            schedule_interval_seconds = (current or {}).get("schedule_interval_seconds")
        if schedule_interval_seconds is not None:
            schedule_interval_seconds = int(schedule_interval_seconds)
        service_id = str(
            payload.get("service_id")
            if payload.get("service_id") is not None
            else (current or {}).get("service_id", "") or ""
        ).strip()
        return {
            "id": job_id,
            "name": name,
            "source": source,
            "target": target,
            "schedule_interval_seconds": schedule_interval_seconds,
            "service_id": service_id,
        }

    def get_backup_job_by_name(self, name: str) -> dict[str, Any] | None:
        self.init_schema()
        with self.connect() as conn:
            row = conn.execute("SELECT * FROM backup_jobs WHERE name = ?", (name,)).fetchone()
        return self._row_to_job(row) if row else None

    def upsert_backup_job(self, payload: dict[str, Any]) -> dict[str, Any]:
        self.init_schema()
        job = self._normalize_backup_job_payload(payload)
        now = int(time.time())
        with self.connect() as conn:
            existing = conn.execute("SELECT created_at FROM backup_jobs WHERE id = ?", (job["id"],)).fetchone()
            created_at = int(existing[0]) if existing else now
            conn.execute(
                """
                INSERT INTO backup_jobs (id, name, source, target, schedule_interval_seconds, service_id, created_at)
                VALUES (?, ?, ?, ?, ?, ?, ?)
                ON CONFLICT(id) DO UPDATE SET
                    name = excluded.name,
                    source = excluded.source,
                    target = excluded.target,
                    schedule_interval_seconds = excluded.schedule_interval_seconds,
                    service_id = excluded.service_id
                ON CONFLICT(name) DO UPDATE SET
                    source = excluded.source,
                    target = excluded.target,
                    schedule_interval_seconds = excluded.schedule_interval_seconds,
                    service_id = excluded.service_id
                """,
                (
                    job["id"],
                    job["name"],
                    job["source"],
                    job["target"],
                    job["schedule_interval_seconds"],
                    job["service_id"],
                    created_at,
                ),
            )
        return self.get_backup_job(job["id"]) or job

    def get_backup_job(self, job_id: str | None) -> dict[str, Any] | None:
        if not job_id:
            return None
        self.init_schema()
        with self.connect() as conn:
            row = conn.execute("SELECT * FROM backup_jobs WHERE id = ?", (job_id,)).fetchone()
        return self._row_to_job(row) if row else None

    def list_backup_jobs(self) -> list[dict[str, Any]]:
        self.init_schema()
        with self.connect() as conn:
            rows = conn.execute("SELECT * FROM backup_jobs ORDER BY created_at DESC").fetchall()
        return [self._row_to_job(row) for row in rows]

    # ------------------------------------------------------------------
    # Backup runs
    # ------------------------------------------------------------------

    def _row_to_run(self, row: sqlite3.Row) -> dict[str, Any]:
        details = json.loads(row["details_json"]) if row["details_json"] else {}
        return {
            "id": row["id"],
            "job_id": row["job_id"],
            "started_at": row["started_at"],
            "ended_at": row["ended_at"],
            "status": row["status"],
            "bytes_transferred": row["bytes_transferred"],
            "duration_ms": row["duration_ms"],
            "error_message": row["error_message"],
            "details": details,
            "created_at": row["created_at"],
        }

    def create_backup_run(self, payload: dict[str, Any]) -> dict[str, Any]:
        self.init_schema()
        run_id = str(payload.get("id") or uuid.uuid4().hex[:12]).strip() or uuid.uuid4().hex[:12]
        now = int(time.time())
        job_id = str(payload["job_id"])
        started_at = int(payload["started_at"])
        ended_at = payload.get("ended_at")
        if ended_at is not None:
            ended_at = int(ended_at)
        status = str(payload.get("status", "pending"))
        bytes_transferred = payload.get("bytes_transferred")
        if bytes_transferred is not None:
            bytes_transferred = int(bytes_transferred)
        duration_ms = payload.get("duration_ms")
        if duration_ms is not None:
            duration_ms = int(duration_ms)
        error_message = str(payload.get("error_message") or "")
        details = payload.get("details", {})
        details_json = json.dumps(details) if details else None
        with self.connect() as conn:
            conn.execute(
                """
                INSERT INTO backup_runs (
                    id, job_id, started_at, ended_at, status, bytes_transferred,
                    duration_ms, error_message, details_json, created_at
                )
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                """,
                (
                    run_id,
                    job_id,
                    started_at,
                    ended_at,
                    status,
                    bytes_transferred,
                    duration_ms,
                    error_message,
                    details_json,
                    now,
                ),
            )
        return self.get_backup_run(run_id) or {
            "id": run_id,
            "job_id": job_id,
            "started_at": started_at,
            "ended_at": ended_at,
            "status": status,
            "bytes_transferred": bytes_transferred,
            "duration_ms": duration_ms,
            "error_message": error_message,
            "details": details,
            "created_at": now,
        }

    def get_backup_run(self, run_id: str | None) -> dict[str, Any] | None:
        if not run_id:
            return None
        self.init_schema()
        with self.connect() as conn:
            row = conn.execute("SELECT * FROM backup_runs WHERE id = ?", (run_id,)).fetchone()
        return self._row_to_run(row) if row else None

    def list_backup_runs(
        self,
        job_id: str | None = None,
        status: str | None = None,
        limit: int = 50,
    ) -> list[dict[str, Any]]:
        self.init_schema()
        clauses: list[str] = []
        params: list[Any] = []
        if job_id:
            clauses.append("job_id = ?")
            params.append(job_id)
        if status:
            clauses.append("status = ?")
            params.append(status)
        where = f"WHERE {' AND '.join(clauses)}" if clauses else ""
        sql = f"SELECT * FROM backup_runs {where} ORDER BY created_at DESC LIMIT ?"
        params.append(max(1, min(int(limit), 200)))
        with self.connect() as conn:
            rows = conn.execute(sql, params).fetchall()
        return [self._row_to_run(row) for row in rows]

    def get_latest_backup_run(self, job_id: str) -> dict[str, Any] | None:
        self.init_schema()
        with self.connect() as conn:
            row = conn.execute(
                "SELECT * FROM backup_runs WHERE job_id = ? ORDER BY created_at DESC LIMIT 1",
                (job_id,),
            ).fetchone()
        return self._row_to_run(row) if row else None

    # ------------------------------------------------------------------
    # Backup alerts
    # ------------------------------------------------------------------

    def _row_to_alert(self, row: sqlite3.Row) -> dict[str, Any]:
        return {
            "id": row["id"],
            "job_id": row["job_id"],
            "run_id": row["run_id"],
            "alert_type": row["alert_type"],
            "severity": row["severity"],
            "message": row["message"],
            "acknowledged": bool(row["acknowledged"]),
            "resolved_at": row["resolved_at"],
            "created_at": row["created_at"],
        }

    def create_backup_alert(self, payload: dict[str, Any]) -> dict[str, Any]:
        self.init_schema()
        alert_id = str(payload.get("id") or uuid.uuid4().hex[:12]).strip() or uuid.uuid4().hex[:12]
        now = int(time.time())
        job_id = str(payload["job_id"])
        run_id = payload.get("run_id")
        if run_id is not None:
            run_id = str(run_id)
        alert_type = str(payload.get("alert_type", "generic"))
        severity = str(payload.get("severity", "warning"))
        message = str(payload.get("message", ""))
        with self.connect() as conn:
            conn.execute(
                """
                INSERT INTO backup_alerts (
                    id, job_id, run_id, alert_type, severity, message,
                    acknowledged, resolved_at, created_at
                )
                VALUES (?, ?, ?, ?, ?, ?, 0, NULL, ?)
                """,
                (alert_id, job_id, run_id, alert_type, severity, message, now),
            )
        return self.get_backup_alert(alert_id) or {
            "id": alert_id,
            "job_id": job_id,
            "run_id": run_id,
            "alert_type": alert_type,
            "severity": severity,
            "message": message,
            "acknowledged": False,
            "resolved_at": None,
            "created_at": now,
        }

    def get_backup_alert(self, alert_id: str | None) -> dict[str, Any] | None:
        if not alert_id:
            return None
        self.init_schema()
        with self.connect() as conn:
            row = conn.execute("SELECT * FROM backup_alerts WHERE id = ?", (alert_id,)).fetchone()
        return self._row_to_alert(row) if row else None

    def list_backup_alerts(
        self,
        job_id: str | None = None,
        acknowledged: bool | None = None,
        severity: str | None = None,
    ) -> list[dict[str, Any]]:
        self.init_schema()
        clauses: list[str] = []
        params: list[Any] = []
        if job_id:
            clauses.append("job_id = ?")
            params.append(job_id)
        if acknowledged is not None:
            clauses.append("acknowledged = ?")
            params.append(1 if acknowledged else 0)
        if severity:
            clauses.append("severity = ?")
            params.append(severity)
        where = f"WHERE {' AND '.join(clauses)}" if clauses else ""
        sql = f"SELECT * FROM backup_alerts {where} ORDER BY created_at DESC"
        with self.connect() as conn:
            rows = conn.execute(sql, params).fetchall()
        return [self._row_to_alert(row) for row in rows]

    def acknowledge_backup_alert(self, alert_id: str) -> dict[str, Any] | None:
        self.init_schema()
        with self.connect() as conn:
            conn.execute(
                "UPDATE backup_alerts SET acknowledged = 1 WHERE id = ?",
                (alert_id,),
            )
        return self.get_backup_alert(alert_id)

    def resolve_backup_alerts_for_job(self, job_id: str, alert_type: str) -> int:
        self.init_schema()
        now = int(time.time())
        with self.connect() as conn:
            cur = conn.execute(
                "UPDATE backup_alerts SET resolved_at = ? WHERE job_id = ? AND alert_type = ? AND resolved_at IS NULL",
                (now, job_id, alert_type),
            )
            return int(cur.rowcount or 0)

    def prune_backup_alerts(self, cutoff_ts: int) -> int:
        self.init_schema()
        with self.connect() as conn:
            cur = conn.execute(
                "DELETE FROM backup_alerts WHERE created_at < ?",
                (int(cutoff_ts),),
            )
            return int(cur.rowcount or 0)

    # ------------------------------------------------------------------
    # Generic key-value settings
    # ------------------------------------------------------------------

    def get_settings(self) -> dict[str, Any]:
        """Return all generic settings as a dict."""
        self.init_schema()
        with self.connect() as conn:
            conn.execute(
                """
                CREATE TABLE IF NOT EXISTS app_settings (
                    key TEXT PRIMARY KEY,
                    value TEXT NOT NULL,
                    updated_at INTEGER NOT NULL
                )
                """
            )
            rows = conn.execute("SELECT key, value FROM app_settings").fetchall()
        return {row["key"]: row["value"] for row in rows}

    def get_setting(self, key: str, default: Any = None) -> Any:
        """Return a single setting value or default."""
        self.init_schema()
        with self.connect() as conn:
            conn.execute(
                """
                CREATE TABLE IF NOT EXISTS app_settings (
                    key TEXT PRIMARY KEY,
                    value TEXT NOT NULL,
                    updated_at INTEGER NOT NULL
                )
                """
            )
            row = conn.execute("SELECT value FROM app_settings WHERE key = ?", (key,)).fetchone()
        return row["value"] if row else default

    def update_setting(self, key: str, value: str) -> None:
        """Set a generic key-value setting."""
        self.init_schema()
        now = int(time.time())
        with self.connect() as conn:
            conn.execute(
                """
                CREATE TABLE IF NOT EXISTS app_settings (
                    key TEXT PRIMARY KEY,
                    value TEXT NOT NULL,
                    updated_at INTEGER NOT NULL
                )
                """
            )
            conn.execute(
                """
                INSERT INTO app_settings (key, value, updated_at)
                VALUES (?, ?, ?)
                ON CONFLICT(key) DO UPDATE SET
                    value = excluded.value,
                    updated_at = excluded.updated_at
                """,
                (key, value, now),
            )

    # ------------------------------------------------------------------
    # Dashboard widgets
    # ------------------------------------------------------------------

    def _row_to_widget(self, row: sqlite3.Row) -> dict[str, Any]:
        keys = row.keys()
        return {
            "id": row["id"],
            "addon_id": row["addon_id"],
            "widget_type": row["widget_type"],
            "service_id": row["service_id"] if "service_id" in keys else None,
            "widget_kind": row["widget_kind"] if "widget_kind" in keys else None,
            "title": row["title"],
            "config": json.loads(row["config_json"] or "{}"),
            "enabled": bool(row["enabled"]),
            "sort_order": int(row["sort_order"]),
            "created_at": row["created_at"],
            "updated_at": row["updated_at"],
        }

    def _normalize_widget_payload(
        self,
        payload: dict[str, Any],
        widget_id: str | None = None,
    ) -> dict[str, Any]:
        current = self.get_widget(widget_id) if widget_id else None
        widget_id = str(payload.get("id") or widget_id or uuid.uuid4().hex[:12]).strip() or uuid.uuid4().hex[:12]
        service_id = str(payload.get("service_id") or (current or {}).get("service_id") or "").strip() or None
        widget_kind = str(payload.get("widget_kind") or (current or {}).get("widget_kind", "")).strip()
        title = str(payload.get("title") or (current or {}).get("title", "") or "").strip()
        config = payload.get("config", (current or {}).get("config", {}))
        if not isinstance(config, dict):
            config = {}
        # Defense-in-depth: reject credential keys at the store layer too.
        _validate_config_keys(config)
        enabled = bool(payload.get("enabled", (current or {}).get("enabled", True)))
        sort_order = int(payload.get("sort_order", (current or {}).get("sort_order", 0)) or 0)
        # Legacy label kept for diagnostics; new code uses service_id + widget_kind.
        widget_type = f"{service_id}:{widget_kind}" if widget_kind else ""
        return {
            "id": widget_id,
            "addon_id": "",
            "widget_type": widget_type,
            "service_id": service_id,
            "widget_kind": widget_kind,
            "title": title,
            "config": config,
            "enabled": enabled,
            "sort_order": sort_order,
        }

    def list_widgets(
        self,
        service_id: str | None = None,
        *,
        scope: str | None = None,
        all_widgets: bool = True,
    ) -> list[dict[str, Any]]:
        """List widget instances, optionally filtered.

        - ``service_id=X``: only widgets for service X.
        - ``scope="dashboard"``: only widgets with NULL service_id.
        - ``scope="service"``: only widgets with a non-null service_id.
        - ``all_widgets=True, service_id=None, scope=None``: all widgets.
        """
        self.init_schema()
        clauses: list[str] = []
        params: list[Any] = []
        if service_id is not None:
            clauses.append("service_id = ?")
            params.append(service_id)
        if scope == "dashboard":
            clauses.append("service_id IS NULL")
        elif scope == "service":
            clauses.append("service_id IS NOT NULL")
        where = f" WHERE {' AND '.join(clauses)}" if clauses else ""
        with self.connect() as conn:
            rows = conn.execute(
                f"SELECT * FROM dashboard_widgets{where} ORDER BY sort_order ASC, created_at ASC",
                params,
            ).fetchall()
        return [self._row_to_widget(row) for row in rows]

    def get_widget(self, widget_id: str | None) -> dict[str, Any] | None:
        if not widget_id:
            return None
        self.init_schema()
        with self.connect() as conn:
            row = conn.execute("SELECT * FROM dashboard_widgets WHERE id = ?", (widget_id,)).fetchone()
        return self._row_to_widget(row) if row else None

    def upsert_widget(self, payload: dict[str, Any], widget_id: str | None = None) -> dict[str, Any]:
        self.init_schema()
        widget = self._normalize_widget_payload(payload, widget_id)
        now = int(time.time())
        with self.connect() as conn:
            existing = conn.execute(
                "SELECT created_at FROM dashboard_widgets WHERE id = ?",
                (widget["id"],),
            ).fetchone()
            created_at = int(existing[0]) if existing else now
            conn.execute(
                """
                INSERT INTO dashboard_widgets (
                    id, addon_id, widget_type, service_id, widget_kind, title,
                    config_json, enabled, sort_order, created_at, updated_at
                )
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                ON CONFLICT(id) DO UPDATE SET
                    addon_id = excluded.addon_id,
                    widget_type = excluded.widget_type,
                    service_id = excluded.service_id,
                    widget_kind = excluded.widget_kind,
                    title = excluded.title,
                    config_json = excluded.config_json,
                    enabled = excluded.enabled,
                    sort_order = excluded.sort_order,
                    updated_at = excluded.updated_at
                """,
                (
                    widget["id"],
                    widget["addon_id"],
                    widget["widget_type"],
                    widget["service_id"],
                    widget["widget_kind"],
                    widget["title"],
                    json.dumps(widget["config"]),
                    1 if widget["enabled"] else 0,
                    widget["sort_order"],
                    created_at,
                    now,
                ),
            )
        return self.get_widget(widget["id"]) or widget

    def delete_widget(self, widget_id: str) -> None:
        self.init_schema()
        with self.connect() as conn:
            conn.execute("DELETE FROM dashboard_widgets WHERE id = ?", (widget_id,))
            conn.execute("DELETE FROM widget_references WHERE widget_id = ?", (widget_id,))

    # ------------------------------------------------------------------
    # Widget references (live-link widgets across dashboards)
    # ------------------------------------------------------------------

    def list_widget_references(self, dashboard_scope: str) -> list[dict[str, Any]]:
        """List widget references for a dashboard scope, joined with widget data."""
        self.init_schema()
        with self.connect() as conn:
            rows = conn.execute(
                """
                SELECT wr.id AS ref_id, wr.dashboard_scope, wr.widget_id, wr.sort_order,
                       wr.created_at AS ref_created_at
                FROM widget_references wr
                WHERE wr.dashboard_scope = ?
                ORDER BY wr.sort_order ASC, wr.created_at ASC
                """,
                (dashboard_scope,),
            ).fetchall()
        result: list[dict[str, Any]] = []
        for row in rows:
            widget = self.get_widget(row["widget_id"])
            if not widget:
                continue
            result.append(
                {
                    "id": row["ref_id"],
                    "dashboard_scope": row["dashboard_scope"],
                    "widget_id": row["widget_id"],
                    "sort_order": int(row["sort_order"]),
                    "created_at": row["ref_created_at"],
                    "widget": widget,
                }
            )
        return result

    def create_widget_reference(self, dashboard_scope: str, widget_id: str, sort_order: int = 0) -> dict[str, Any]:
        self.init_schema()
        widget = self.get_widget(widget_id)
        if not widget:
            raise ValueError(f"Widget {widget_id} not found")
        ref_id = uuid.uuid4().hex[:12]
        now = int(time.time())
        with self.connect() as conn:
            conn.execute(
                """
                INSERT INTO widget_references (id, dashboard_scope, widget_id, sort_order, created_at)
                VALUES (?, ?, ?, ?, ?)
                """,
                (ref_id, dashboard_scope, widget_id, sort_order, now),
            )
        return {
            "id": ref_id,
            "dashboard_scope": dashboard_scope,
            "widget_id": widget_id,
            "sort_order": sort_order,
            "created_at": now,
            "widget": widget,
        }

    def delete_widget_reference(self, reference_id: str) -> None:
        self.init_schema()
        with self.connect() as conn:
            conn.execute("DELETE FROM widget_references WHERE id = ?", (reference_id,))

    def update_widget_reference(self, reference_id: str, sort_order: int) -> dict[str, Any]:
        """Update only the sort_order on a widget reference (per-dashboard reordering)."""
        self.init_schema()
        with self.connect() as conn:
            row = conn.execute(
                "SELECT * FROM widget_references WHERE id = ?",
                (reference_id,),
            ).fetchone()
            if not row:
                raise ValueError(f"Reference {reference_id} not found")
            conn.execute(
                "UPDATE widget_references SET sort_order = ? WHERE id = ?",
                (sort_order, reference_id),
            )
        widget = self.get_widget(row["widget_id"])
        return {
            "id": row["id"],
            "dashboard_scope": row["dashboard_scope"],
            "widget_id": row["widget_id"],
            "sort_order": sort_order,
            "created_at": int(row["created_at"]),
            "widget": widget,
        }

    def detach_widget_reference(self, reference_id: str, dashboard_scope: str) -> dict[str, Any]:
        """Clone the referenced widget into a new standalone instance owned by the scope."""
        self.init_schema()
        with self.connect() as conn:
            row = conn.execute(
                "SELECT widget_id FROM widget_references WHERE id = ?",
                (reference_id,),
            ).fetchone()
        if not row:
            raise ValueError(f"Reference {reference_id} not found")
        source = self.get_widget(row["widget_id"])
        if not source:
            raise ValueError(f"Source widget {row['widget_id']} not found")
        # Clone: copy the widget verbatim including service_id (so service-bound
        # widgets keep working), only the id/created_at change.
        cloned = self.upsert_widget(
            {
                "service_id": source.get("service_id"),
                "widget_kind": source["widget_kind"],
                "title": source["title"],
                "config": source["config"],
                "enabled": source["enabled"],
                "sort_order": source["sort_order"],
            }
        )
        self.delete_widget_reference(reference_id)
        return cloned

    # ------------------------------------------------------------------
    # Service registry
    # ------------------------------------------------------------------

    def _row_to_service(self, row: sqlite3.Row) -> dict[str, Any]:
        secrets_blob = json.loads(row["secrets_json"] or "{}")
        return {
            "id": row["id"],
            "service_type": row["service_type"],
            "name": row["name"],
            "config": json.loads(row["config_json"] or "{}"),
            "secrets": secrets_blob,
            "enabled": bool(row["enabled"]),
            "created_at": row["created_at"],
            "updated_at": row["updated_at"],
        }

    def list_services(self, service_type: str | None = None) -> list[dict[str, Any]]:
        self.init_schema()
        with self.connect() as conn:
            if service_type:
                rows = conn.execute(
                    "SELECT * FROM services WHERE service_type = ? ORDER BY name ASC",
                    (service_type,),
                ).fetchall()
            else:
                rows = conn.execute("SELECT * FROM services ORDER BY name ASC").fetchall()
        return [self._row_to_service(row) for row in rows]

    def get_service(self, service_id: str) -> dict[str, Any] | None:
        self.init_schema()
        with self.connect() as conn:
            row = conn.execute("SELECT * FROM services WHERE id = ?", (service_id,)).fetchone()
        return self._row_to_service(row) if row else None

    def _normalize_service_payload(
        self,
        payload: dict[str, Any],
        service_id: str | None = None,
    ) -> dict[str, Any]:
        current = self.get_service(service_id) if service_id else None
        service_id = str(payload.get("id") or service_id or uuid.uuid4().hex[:12]).strip() or uuid.uuid4().hex[:12]
        service_type = str(payload.get("service_type") or (current or {}).get("service_type", "")).strip()
        name = str(payload.get("name") or (current or {}).get("name", "") or "").strip()
        config = payload.get("config", (current or {}).get("config", {}))
        if not isinstance(config, dict):
            config = {}
        enabled = bool(payload.get("enabled", (current or {}).get("enabled", True)))
        return {
            "id": service_id,
            "service_type": service_type,
            "name": name,
            "config": config,
            "enabled": enabled,
        }

    def upsert_service(
        self,
        payload: dict[str, Any],
        secret_values: dict[str, str] | None = None,
        service_id: str | None = None,
    ) -> dict[str, Any]:
        """Insert or update a service instance.

        ``secret_values`` carries plaintext secrets to encrypt and store. A key
        absent from ``secret_values`` preserves the existing ciphertext; a key
        mapped to an empty string clears it.
        """
        self.init_schema()
        service = self._normalize_service_payload(payload, service_id)
        now = int(time.time())

        existing = self.get_service(service["id"])
        secrets_blob: dict[str, str]
        if existing is not None:
            secrets_blob = dict(existing["secrets"])
        else:
            secrets_blob = {}
        if secret_values:
            from media_library_viewer_api.services.secrets import encrypt_value

            for key, value in secret_values.items():
                if value == "":
                    secrets_blob.pop(key, None)
                else:
                    secrets_blob[key] = encrypt_value(value)

        with self.connect() as conn:
            created_at = int(existing["created_at"]) if existing else now
            conn.execute(
                """
                INSERT INTO services (
                    id, service_type, name, config_json, secrets_json,
                    enabled, created_at, updated_at
                )
                VALUES (?, ?, ?, ?, ?, ?, ?, ?)
                ON CONFLICT(id) DO UPDATE SET
                    service_type = excluded.service_type,
                    name = excluded.name,
                    config_json = excluded.config_json,
                    secrets_json = excluded.secrets_json,
                    enabled = excluded.enabled,
                    updated_at = excluded.updated_at
                """,
                (
                    service["id"],
                    service["service_type"],
                    service["name"],
                    json.dumps(service["config"]),
                    json.dumps(secrets_blob),
                    1 if service["enabled"] else 0,
                    created_at,
                    now,
                ),
            )
        return self.get_service(service["id"]) or service

    def delete_service(self, service_id: str) -> None:
        """Delete a service and cascade-delete widgets + harness data."""
        self.init_schema()
        with self.connect() as conn:
            # The service_id column on dashboard_widgets is added in a later
            # slice; only cascade when it is present.
            widget_cols = {row[1] for row in conn.execute("PRAGMA table_info(dashboard_widgets)").fetchall()}
            if "service_id" in widget_cols:
                conn.execute(
                    "DELETE FROM dashboard_widgets WHERE service_id = ?",
                    (service_id,),
                )
            conn.execute("DELETE FROM services WHERE id = ?", (service_id,))

        # Cascade-delete harness-managed data (best-effort: the service row is
        # already removed; data cleanup must not block service deletion).
        try:
            from media_library_viewer_api.services.service_data import get_service_data_harness

            get_service_data_harness().cascade_delete(service_id)
        except Exception:
            logger.exception("Failed to cascade-delete harness data for service %s", service_id)

    def record_service_task_run(self, payload: dict[str, Any]) -> dict[str, Any]:
        """Append a service task run history row."""
        self.init_schema()
        run_id = str(payload.get("id") or uuid.uuid4().hex[:12])
        now = int(time.time())
        with self.connect() as conn:
            conn.execute(
                """
                INSERT INTO service_task_runs (
                    id, task_id, service_id, status, exit_status, duration_ms,
                    stdout_tail, stderr_tail, error, created_at
                )
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                """,
                (
                    run_id,
                    str(payload.get("task_id") or ""),
                    str(payload.get("service_id") or ""),
                    str(payload.get("status") or "error"),
                    payload.get("exit_status"),
                    payload.get("duration_ms"),
                    str(payload.get("stdout_tail") or "")[:8000],
                    str(payload.get("stderr_tail") or "")[:8000],
                    str(payload.get("error") or "")[:1000],
                    int(payload.get("created_at") or now),
                ),
            )
        return {"id": run_id}

    def list_service_task_runs(
        self,
        service_id: str | None = None,
        task_id: str | None = None,
        limit: int = 50,
    ) -> list[dict[str, Any]]:
        self.init_schema()
        clauses: list[str] = []
        params: list[Any] = []
        if service_id:
            clauses.append("service_id = ?")
            params.append(service_id)
        if task_id:
            clauses.append("task_id = ?")
            params.append(task_id)
        where = ("WHERE " + " AND ".join(clauses)) if clauses else ""
        params.append(int(limit))
        with self.connect() as conn:
            rows = conn.execute(
                f"SELECT * FROM service_task_runs {where} ORDER BY created_at DESC LIMIT ?",
                params,
            ).fetchall()
        return [
            {
                "id": row["id"],
                "task_id": row["task_id"],
                "service_id": row["service_id"],
                "status": row["status"],
                "exit_status": row["exit_status"],
                "duration_ms": row["duration_ms"],
                "stdout_tail": row["stdout_tail"],
                "stderr_tail": row["stderr_tail"],
                "error": row["error"],
                "created_at": row["created_at"],
            }
            for row in rows
        ]

    # ------------------------------------------------------------------
    # Named dashboards
    # ------------------------------------------------------------------

    @staticmethod
    def _slugify(label: str) -> str:
        import re

        slug = re.sub(r"[^a-z0-9]+", "-", label.lower()).strip("-")
        return slug or "dashboard"

    def _unique_slug(self, slug: str, exclude_id: str | None = None) -> str:
        self.init_schema()
        base = slug
        suffix = 1
        with self.connect() as conn:
            while True:
                row = conn.execute(
                    "SELECT id FROM named_dashboards WHERE slug = ? AND id != ?",
                    (slug, exclude_id or ""),
                ).fetchone()
                if not row:
                    return slug
                suffix += 1
                slug = f"{base}-{suffix}"

    def _row_to_dashboard(self, row: sqlite3.Row) -> dict[str, Any]:
        return {
            "id": row["id"],
            "label": row["label"],
            "slug": row["slug"],
            "sort_order": row["sort_order"],
            "payload": json.loads(row["payload_json"] or "{}"),
            "created_at": row["created_at"],
            "updated_at": row["updated_at"],
        }

    def list_dashboards(self) -> list[dict[str, Any]]:
        self.init_schema()
        with self.connect() as conn:
            rows = conn.execute(
                "SELECT * FROM named_dashboards ORDER BY sort_order ASC, label COLLATE NOCASE"
            ).fetchall()
        return [self._row_to_dashboard(row) for row in rows]

    def get_dashboard(self, dashboard_id: str | None) -> dict[str, Any] | None:
        if not dashboard_id:
            return None
        self.init_schema()
        with self.connect() as conn:
            row = conn.execute("SELECT * FROM named_dashboards WHERE id = ?", (dashboard_id,)).fetchone()
        return self._row_to_dashboard(row) if row else None

    def get_dashboard_by_slug(self, slug: str | None) -> dict[str, Any] | None:
        if not slug:
            return None
        self.init_schema()
        with self.connect() as conn:
            row = conn.execute("SELECT * FROM named_dashboards WHERE slug = ?", (slug,)).fetchone()
        return self._row_to_dashboard(row) if row else None

    def upsert_dashboard(self, payload: dict[str, Any], dashboard_id: str | None = None) -> dict[str, Any]:
        self.init_schema()
        current = self.get_dashboard(dashboard_id) if dashboard_id else None
        dash_id = str(payload.get("id") or dashboard_id or uuid.uuid4().hex[:12]).strip()
        label = str(payload.get("label") or (current or {}).get("label") or "Dashboard").strip()
        slug = str(payload.get("slug") or "").strip() or self._slugify(label)
        slug = self._unique_slug(slug, exclude_id=dash_id)
        sort_order = payload.get("sort_order")
        if sort_order is None:
            sort_order = (current or {}).get("sort_order", 0)
        sort_order = int(sort_order)
        payload_data = payload.get("payload")
        if payload_data is None:
            payload_data = (current or {}).get("payload", {})
        now = int(time.time())
        with self.connect() as conn:
            existing = conn.execute("SELECT created_at FROM named_dashboards WHERE id = ?", (dash_id,)).fetchone()
            created_at = int(existing[0]) if existing else now
            conn.execute(
                """
                INSERT INTO named_dashboards (id, label, slug, sort_order, payload_json, created_at, updated_at)
                VALUES (?, ?, ?, ?, ?, ?, ?)
                ON CONFLICT(id) DO UPDATE SET
                    label = excluded.label,
                    slug = excluded.slug,
                    sort_order = excluded.sort_order,
                    payload_json = excluded.payload_json,
                    updated_at = excluded.updated_at
                """,
                (
                    dash_id,
                    label,
                    slug,
                    sort_order,
                    json.dumps(payload_data),
                    created_at,
                    now,
                ),
            )
        return self.get_dashboard(dash_id) or {"id": dash_id, "label": label, "slug": slug}

    def delete_dashboard(self, dashboard_id: str) -> None:
        self.init_schema()
        with self.connect() as conn:
            conn.execute("DELETE FROM named_dashboards WHERE id = ?", (dashboard_id,))


_store: SettingsStore | None = None


def get_settings_store() -> SettingsStore:
    global _store
    if _store is None:
        _store = SettingsStore()
    return _store
