"""Durable state for the robot-local voice supervisor.

This module deliberately has no scheduler dependency. The voice process writes
locally first and the management plane may consume events later by cursor.
"""

from __future__ import annotations

import json
import sqlite3
import threading
import time
import uuid
from contextlib import contextmanager
from pathlib import Path


COMMAND_STATES = {
    "queued", "delivered", "acknowledged", "running", "succeeded",
    "failed", "expired", "superseded",
}


class SupervisorStore:
    def __init__(self, config_path: str, robot_id: str, max_events: int = 50_000) -> None:
        source = Path(config_path)
        self.path = source.with_suffix(source.suffix + ".supervisor.sqlite3")
        self.robot_id = robot_id
        self.max_events = max(1_000, int(max_events))
        self.lock = threading.RLock()
        self.path.parent.mkdir(parents=True, exist_ok=True)
        self._initialize()

    def _connect(self) -> sqlite3.Connection:
        connection = sqlite3.connect(self.path, timeout=10)
        connection.row_factory = sqlite3.Row
        connection.execute("PRAGMA journal_mode=WAL")
        connection.execute("PRAGMA synchronous=NORMAL")
        return connection

    @contextmanager
    def _db(self):
        connection = self._connect()
        try:
            with connection:
                yield connection
        finally:
            connection.close()

    def _initialize(self) -> None:
        with self.lock, self._db() as db:
            db.executescript(
                """
                CREATE TABLE IF NOT EXISTS events (
                    sequence INTEGER PRIMARY KEY AUTOINCREMENT,
                    event_id TEXT NOT NULL UNIQUE,
                    robot_id TEXT NOT NULL,
                    session_id TEXT,
                    type TEXT NOT NULL,
                    timestamp REAL NOT NULL,
                    payload TEXT NOT NULL,
                    acknowledged INTEGER NOT NULL DEFAULT 0
                );
                CREATE INDEX IF NOT EXISTS idx_voice_events_ack_sequence
                    ON events(acknowledged, sequence);
                CREATE TABLE IF NOT EXISTS state (
                    key TEXT PRIMARY KEY,
                    value TEXT NOT NULL
                );
                CREATE TABLE IF NOT EXISTS completed_commands (
                    command_id TEXT PRIMARY KEY,
                    status TEXT NOT NULL,
                    result TEXT NOT NULL,
                    completed_at REAL NOT NULL
                );
                """
            )

    def get_state(self, key: str, default=None):
        with self.lock, self._db() as db:
            row = db.execute("SELECT value FROM state WHERE key = ?", (key,)).fetchone()
        return json.loads(row["value"]) if row else default

    def set_state(self, key: str, value) -> None:
        encoded = json.dumps(value, separators=(",", ":"), ensure_ascii=True)
        with self.lock, self._db() as db:
            db.execute(
                "INSERT INTO state(key, value) VALUES(?, ?) "
                "ON CONFLICT(key) DO UPDATE SET value=excluded.value",
                (key, encoded),
            )

    def append_event(self, event_type: str, payload: dict, session_id: str | None = None) -> dict:
        event_id = str(uuid.uuid4())
        timestamp = time.time()
        with self.lock, self._db() as db:
            cursor = db.execute(
                "INSERT INTO events(event_id, robot_id, session_id, type, timestamp, payload) "
                "VALUES(?, ?, ?, ?, ?, ?)",
                (
                    event_id,
                    self.robot_id,
                    session_id,
                    event_type,
                    timestamp,
                    json.dumps(payload, separators=(",", ":"), ensure_ascii=True),
                ),
            )
            sequence = int(cursor.lastrowid)
            overflow = db.execute(
                "SELECT sequence FROM events ORDER BY sequence DESC LIMIT 1 OFFSET ?",
                (self.max_events,),
            ).fetchone()
            if overflow:
                db.execute("DELETE FROM events WHERE sequence <= ? AND acknowledged = 1", (overflow["sequence"],))
        return {
            "event_id": event_id,
            "sequence": sequence,
            "robot_id": self.robot_id,
            "session_id": session_id,
            "type": event_type,
            "timestamp": timestamp,
            "payload": payload,
        }

    def events_after(self, cursor: int, limit: int = 250) -> list[dict]:
        safe_limit = min(1_000, max(1, int(limit)))
        with self.lock, self._db() as db:
            rows = db.execute(
                "SELECT * FROM events WHERE sequence > ? ORDER BY sequence LIMIT ?",
                (max(0, int(cursor)), safe_limit),
            ).fetchall()
        return [
            {
                "event_id": row["event_id"],
                "sequence": row["sequence"],
                "robot_id": row["robot_id"],
                "session_id": row["session_id"],
                "type": row["type"],
                "timestamp": row["timestamp"],
                "payload": json.loads(row["payload"]),
            }
            for row in rows
        ]

    def acknowledge(self, sequence: int) -> int:
        highest = max(0, int(sequence))
        with self.lock, self._db() as db:
            db.execute("UPDATE events SET acknowledged = 1 WHERE sequence <= ?", (highest,))
            db.execute("DELETE FROM events WHERE acknowledged = 1 AND sequence <= ?", (highest,))
        return highest

    def stats(self) -> dict:
        with self.lock, self._db() as db:
            row = db.execute(
                "SELECT COUNT(*) total, COALESCE(MIN(sequence), 0) first_sequence, "
                "COALESCE(MAX(sequence), 0) last_sequence, "
                "SUM(CASE WHEN acknowledged = 0 THEN 1 ELSE 0 END) pending FROM events"
            ).fetchone()
        return {
            "total": int(row["total"] or 0),
            "pending": int(row["pending"] or 0),
            "first_sequence": int(row["first_sequence"] or 0),
            "last_sequence": int(row["last_sequence"] or 0),
        }

    def command_result(self, command_id: str) -> dict | None:
        with self.lock, self._db() as db:
            row = db.execute(
                "SELECT status, result, completed_at FROM completed_commands WHERE command_id = ?",
                (command_id,),
            ).fetchone()
        if not row:
            return None
        return {
            "status": row["status"],
            "result": json.loads(row["result"]),
            "completed_at": row["completed_at"],
        }

    def complete_command(self, command_id: str, status: str, result: dict) -> None:
        if status not in COMMAND_STATES:
            raise ValueError(f"invalid command state: {status}")
        with self.lock, self._db() as db:
            db.execute(
                "INSERT OR REPLACE INTO completed_commands(command_id, status, result, completed_at) "
                "VALUES(?, ?, ?, ?)",
                (command_id, status, json.dumps(result, ensure_ascii=True), time.time()),
            )