"""Redis access for the web service."""

from __future__ import annotations

import json
import logging
import queue
import threading
import time
import uuid
from datetime import datetime, timezone
from typing import Any, Callable

try:
    import redis
except ImportError:
    redis = None  # type: ignore[assignment]

logger = logging.getLogger("hot_tub.web")

KEY_CONTROL_LATEST = "control:latest"
CHAN_CONTROL_STATE = "control.state"
KEY_TEMPS_LATEST = "temps:latest"
CHAN_TEMPS_READING = "temps.reading"
KEY_TEMPS_SETUP = "temps:setup"
CHAN_TEMPS_SETUP = "temps.setup"
KEY_HEALTH_WEB = "health:web"
CHAN_HEALTH_WEB = "health.web"
KEY_HEALTH_CONTROL = "health:control"
KEY_HEALTH_TEMPS = "health:temps"
CHAN_COMMAND = "control.command"
CHAN_TEMPS_COMMAND = "temps.command"

SSE_CHANNELS = (CHAN_CONTROL_STATE, CHAN_TEMPS_READING, CHAN_TEMPS_SETUP)


def _utc_now() -> str:
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def _parse_ts(ts: str | None) -> datetime | None:
    if not ts:
        return None
    try:
        if ts.endswith("Z"):
            return datetime.fromisoformat(ts.replace("Z", "+00:00"))
        return datetime.fromisoformat(ts)
    except ValueError:
        return None


def payload_age_s(payload: dict[str, Any] | None, now: datetime | None = None) -> float | None:
    if payload is None:
        return None
    ts = _parse_ts(payload.get("ts"))
    if ts is None:
        return None
    if ts.tzinfo is None:
        ts = ts.replace(tzinfo=timezone.utc)
    now = now or datetime.now(timezone.utc)
    return (now - ts.astimezone(timezone.utc)).total_seconds()


class WebRedis:
    def __init__(self, redis_url: str) -> None:
        if redis is None:
            raise RuntimeError("redis package not installed (apt: python3-redis)")
        self._url = redis_url
        self._client: redis.Redis | None = None
        self._pubsub: redis.client.PubSub | None = None
        self._backoff_s = 1.0
        self._fanout_lock = threading.Lock()
        self._subscribers: list[queue.Queue[tuple[str, str]]] = []
        self._fanout_thread: threading.Thread | None = None
        self._fanout_stop = threading.Event()

    def _connect(self) -> redis.Redis:
        if self._client is None:
            self._client = redis.from_url(self._url, decode_responses=True)
        return self._client

    def _ensure(self) -> redis.Redis | None:
        try:
            client = self._connect()
            client.ping()
            self._backoff_s = 1.0
            return client
        except Exception as exc:
            logger.warning("Redis unavailable: %s (retry in %.0fs)", exc, self._backoff_s)
            self._client = None
            self._pubsub = None
            time.sleep(min(self._backoff_s, 30.0))
            self._backoff_s = min(self._backoff_s * 2, 30.0)
            return None

    def get_json(self, key: str) -> dict[str, Any] | None:
        client = self._ensure()
        if client is None:
            return None
        try:
            raw = client.get(key)
            if not raw:
                return None
            data = json.loads(raw)
            return data if isinstance(data, dict) else None
        except Exception as exc:
            logger.warning("GET %s failed: %s", key, exc)
            return None

    def publish_control_command(self, op: str, args: dict[str, Any]) -> str:
        cmd_id = str(uuid.uuid4())
        envelope = {
            "ts": _utc_now(),
            "source": "web",
            "id": cmd_id,
            "op": op,
            "args": args,
        }
        client = self._ensure()
        if client is None:
            raise RuntimeError("redis unavailable")
        body = json.dumps(envelope)
        client.publish(CHAN_COMMAND, body)
        return cmd_id

    def publish_temp_setup_command(self, op: str) -> str:
        cmd_id = str(uuid.uuid4())
        envelope = {"ts": _utc_now(), "source": "web", "id": cmd_id, "op": op}
        client = self._ensure()
        if client is None:
            raise RuntimeError("redis unavailable")
        body = json.dumps(envelope)
        client.publish(CHAN_TEMPS_COMMAND, body)
        return cmd_id

    def publish_health(self, payload: dict[str, Any]) -> bool:
        client = self._ensure()
        if client is None:
            return False
        body = json.dumps(payload)
        client.set(KEY_HEALTH_WEB, body)
        client.publish(CHAN_HEALTH_WEB, body)
        return True

    def _ensure_fanout(self) -> None:
        with self._fanout_lock:
            if self._fanout_thread is not None and self._fanout_thread.is_alive():
                return
            self._fanout_stop.clear()
            self._fanout_thread = threading.Thread(target=self._fanout_loop, daemon=True)
            self._fanout_thread.start()

    def _fanout_loop(self) -> None:
        while not self._fanout_stop.is_set():
            client = self._ensure()
            if client is None:
                time.sleep(1.0)
                continue
            try:
                if self._pubsub is None:
                    self._pubsub = client.pubsub(ignore_subscribe_messages=True)
                    self._pubsub.subscribe(*SSE_CHANNELS)
                msg = self._pubsub.get_message(timeout=1.0)
                if msg is None or msg.get("type") != "message":
                    continue
                channel = msg.get("channel")
                data = msg.get("data")
                if not isinstance(channel, str) or not isinstance(data, str):
                    continue
                with self._fanout_lock:
                    dead: list[queue.Queue[tuple[str, str]]] = []
                    for q in self._subscribers:
                        try:
                            q.put_nowait((channel, data))
                        except queue.Full:
                            dead.append(q)
                    for q in dead:
                        self._subscribers.remove(q)
            except Exception as exc:
                logger.warning("SSE fanout error: %s", exc)
                if self._pubsub is not None:
                    try:
                        self._pubsub.close()
                    except Exception:
                        pass
                    self._pubsub = None
                time.sleep(0.5)

    def subscribe_sse(self) -> queue.Queue[tuple[str, str]]:
        q: queue.Queue[tuple[str, str]] = queue.Queue(maxsize=64)
        with self._fanout_lock:
            self._subscribers.append(q)
        self._ensure_fanout()
        return q

    def unsubscribe_sse(self, q: queue.Queue[tuple[str, str]]) -> None:
        with self._fanout_lock:
            if q in self._subscribers:
                self._subscribers.remove(q)

    def close(self) -> None:
        self._fanout_stop.set()
        if self._fanout_thread is not None:
            self._fanout_thread.join(timeout=2.0)
        if self._pubsub is not None:
            try:
                self._pubsub.unsubscribe()
                self._pubsub.close()
            except Exception:
                pass
            self._pubsub = None
        if self._client is not None:
            try:
                self._client.close()
            except Exception:
                pass
            self._client = None