from __future__ import annotations

import atexit
import json
import logging
import os
import queue
import threading
import time
from collections import defaultdict
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Mapping

try:
    import fcntl
except ImportError:  # pragma: no cover - indisponível no Windows
    fcntl = None


def _sanitize_for_filename(value: str) -> str:
    raw = (value or "").strip()
    if not raw:
        return "unknown"

    allowed = []
    for ch in raw:
        if ch.isalnum() or ch in {"-", "_", "."}:
            allowed.append(ch)
        else:
            allowed.append("_")

    sanitized = "".join(allowed).strip("._")
    return sanitized or "unknown"


def _json_safe(value: Any) -> Any:
    if value is None:
        return None
    if isinstance(value, (str, int, float, bool)):
        return value
    if isinstance(value, Path):
        return str(value)
    if isinstance(value, Mapping):
        return {str(key): _json_safe(item) for key, item in value.items()}
    if isinstance(value, (list, tuple, set)):
        return [_json_safe(item) for item in value]
    return str(value)


def _bounded_env_int(name: str, default: int, *, minimum: int, maximum: int) -> int:
    try:
        value = int(os.getenv(name, str(default)) or str(default))
    except (TypeError, ValueError):
        value = default
    return max(minimum, min(maximum, value))


class _WriterCommand:
    __slots__ = ("kind", "event")

    def __init__(self, kind: str) -> None:
        self.kind = kind
        self.event = threading.Event()


class _TimelineWriter:
    """Serializa timelines numa thread dedicada, fora do event loop."""

    def __init__(
        self,
        max_queue: int,
        *,
        warning_interval_s: float = 60.0,
    ) -> None:
        self._queue: queue.Queue[tuple[Path, str] | _WriterCommand] = queue.Queue(
            maxsize=max(1, max_queue)
        )
        self._thread: threading.Thread | None = None
        self._lifecycle_lock = threading.Lock()
        self._stats_lock = threading.Lock()
        self._closed = False
        self._dropped = 0
        self._write_errors = 0
        self._last_drop_warning_at = 0.0
        self._last_error_warning_at = 0.0
        self._warning_interval_s = max(1.0, float(warning_interval_s))
        self._logger = logging.getLogger(f"{__name__}.writer")

    @property
    def dropped(self) -> int:
        with self._stats_lock:
            return self._dropped

    @property
    def write_errors(self) -> int:
        with self._stats_lock:
            return self._write_errors

    def start(self) -> bool:
        with self._lifecycle_lock:
            if self._closed:
                return False
            if self._thread is not None and self._thread.is_alive():
                return True
            self._thread = threading.Thread(
                target=self._run,
                name="call-timeline-writer",
                daemon=True,
            )
            self._thread.start()
            return True

    def submit(self, path: Path, line: str) -> bool:
        if not self.start():
            self._record_drop(reason="writer_closed")
            return False
        reason = ""
        with self._lifecycle_lock:
            if self._closed:
                reason = "writer_closed"
            else:
                try:
                    self._queue.put_nowait((path, line))
                    return True
                except queue.Full:
                    reason = "queue_full"
        self._record_drop(reason=reason)
        return False

    def flush(self, timeout: float | None = 2.0) -> bool:
        with self._lifecycle_lock:
            if self._closed:
                return self._queue.empty()
        if not self.start():
            return False
        command = _WriterCommand("flush")
        deadline = None if timeout is None else time.monotonic() + max(0.0, timeout)
        with self._lifecycle_lock:
            if self._closed:
                return self._queue.empty()
            if not self._enqueue_command(command, deadline):
                return False
        remaining = None if deadline is None else max(0.0, deadline - time.monotonic())
        return command.event.wait(remaining)

    def shutdown(self, timeout: float | None = 2.0) -> bool:
        with self._lifecycle_lock:
            if self._closed:
                thread = self._thread
                return thread is None or not thread.is_alive()
            self._closed = True
            thread = self._thread

        if thread is None:
            return True

        command = _WriterCommand("shutdown")
        deadline = None if timeout is None else time.monotonic() + max(0.0, timeout)
        if not self._enqueue_command(command, deadline):
            return False
        remaining = None if deadline is None else max(0.0, deadline - time.monotonic())
        if not command.event.wait(remaining):
            return False
        remaining = None if deadline is None else max(0.0, deadline - time.monotonic())
        thread.join(remaining)
        return not thread.is_alive()

    def _enqueue_command(
        self,
        command: _WriterCommand,
        deadline: float | None,
    ) -> bool:
        try:
            if deadline is None:
                self._queue.put(command)
            else:
                self._queue.put(command, timeout=max(0.0, deadline - time.monotonic()))
            return True
        except queue.Full:
            return False

    def _run(self) -> None:
        while True:
            first = self._queue.get()
            batch: list[tuple[Path, str]] = []
            commands: list[_WriterCommand] = []
            self._collect(first, batch, commands)
            while True:
                try:
                    item = self._queue.get_nowait()
                except queue.Empty:
                    break
                self._collect(item, batch, commands)

            if batch:
                self._write_batch(batch)

            should_stop = False
            for command in commands:
                command.event.set()
                should_stop = should_stop or command.kind == "shutdown"
            if should_stop:
                return

    @staticmethod
    def _collect(
        item: tuple[Path, str] | _WriterCommand,
        batch: list[tuple[Path, str]],
        commands: list[_WriterCommand],
    ) -> None:
        if isinstance(item, _WriterCommand):
            commands.append(item)
        else:
            batch.append(item)

    def _write_batch(self, batch: list[tuple[Path, str]]) -> None:
        by_path: dict[Path, list[str]] = defaultdict(list)
        for path, line in batch:
            by_path[path].append(line)
        for path, lines in by_path.items():
            try:
                path.parent.mkdir(parents=True, exist_ok=True)
                with path.open("a", encoding="utf-8") as handle:
                    if fcntl is not None:
                        fcntl.flock(handle.fileno(), fcntl.LOCK_EX)
                    try:
                        handle.write("".join(f"{line}\n" for line in lines))
                    finally:
                        if fcntl is not None:
                            fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
            except Exception as exc:
                self._record_write_error(path=path, exc=exc)

    def _record_drop(self, *, reason: str) -> None:
        now = time.monotonic()
        with self._stats_lock:
            self._dropped += 1
            dropped = self._dropped
            should_warn = now - self._last_drop_warning_at >= self._warning_interval_s
            if should_warn:
                self._last_drop_warning_at = now
        if should_warn:
            self._logger.warning(
                "CALL_TIMELINE_ASYNC_DROP | reason=%s | dropped_total=%s | queue_size=%s",
                reason,
                dropped,
                self._queue.qsize(),
            )

    def _record_write_error(self, *, path: Path, exc: Exception) -> None:
        now = time.monotonic()
        with self._stats_lock:
            self._write_errors += 1
            errors = self._write_errors
            should_warn = now - self._last_error_warning_at >= self._warning_interval_s
            if should_warn:
                self._last_error_warning_at = now
        if should_warn:
            self._logger.warning(
                "CALL_TIMELINE_ASYNC_WRITE_FAIL | path=%s | error=%s: %s | errors_total=%s",
                path,
                type(exc).__name__,
                exc,
                errors,
            )


_WRITER = _TimelineWriter(
    _bounded_env_int(
        "CALL_TIMELINE_QUEUE_MAX",
        10_000,
        minimum=1,
        maximum=1_000_000,
    ),
    warning_interval_s=_bounded_env_int(
        "ASYNC_IO_WARNING_INTERVAL_S",
        60,
        minimum=1,
        maximum=3_600,
    ),
)
atexit.register(_WRITER.shutdown)


class CallTimeline:
    def __init__(
        self,
        *,
        logger: logging.Logger,
        component: str,
        timeline_id: str,
        protocol: str = "",
        room: str = "",
        session_id: str = "",
        phone_number: str = "",
        origin_unix_ms: int | None = None,
    ) -> None:
        self._logger = logger
        self._component = (component or "unknown").strip().lower()
        self._timeline_id = _sanitize_for_filename(timeline_id or session_id or protocol or room)
        self._protocol = (protocol or "").strip()
        self._room = (room or "").strip()
        self._session_id = (session_id or "").strip()
        self._phone_number = (phone_number or "").strip()
        self._origin_unix_ms = int(origin_unix_ms or round(time.time() * 1000))
        self._enabled = os.getenv("CALL_TIMELINE_ENABLED", "1") == "1"
        self._console_enabled = os.getenv("CALL_TIMELINE_CONSOLE", "0") == "1"
        self._dir = Path(os.getenv("CALL_TIMELINE_DIR", "./timeline"))
        self._path = self._dir / f"{self._timeline_id}.jsonl"
        if self._enabled:
            _WRITER.start()

    @property
    def origin_unix_ms(self) -> int:
        return self._origin_unix_ms

    @property
    def path(self) -> Path:
        return self._path

    def emit(self, event: str, **fields: Any) -> None:
        if not self._enabled:
            return

        now_ms = round(time.time() * 1000)
        record = {
            "ts": datetime.now(timezone.utc).isoformat(),
            "t_rel_ms": max(0, int(now_ms - self._origin_unix_ms)),
            "component": self._component,
            "event": (event or "unknown").strip(),
            "timeline_id": self._timeline_id,
            "protocol": self._protocol,
            "room": self._room,
            "session_id": self._session_id,
            "phone_number": self._phone_number,
        }
        for key, value in fields.items():
            record[str(key)] = _json_safe(value)

        line = json.dumps(record, ensure_ascii=False)
        self._append_line(line)

        if self._console_enabled:
            self._logger.info("TIMELINE | %s", line)

    def _append_line(self, line: str) -> None:
        _WRITER.submit(self._path, line)

    def flush(self, timeout: float | None = 2.0) -> bool:
        """Aguarda eventos já enfileirados; não deve ser chamado no loop de áudio."""
        return _WRITER.flush(timeout=timeout)

    @property
    def dropped_events(self) -> int:
        return _WRITER.dropped
