from __future__ import annotations

import os
from pathlib import Path
import threading
from typing import Any

from services.shared.core import utc_now_iso


class BridgeState:
    def __init__(self) -> None:
        self._lock = threading.Lock()
        self.ami_connected = False
        self.last_event_at: str | None = None
        self.last_error: str | None = None
        self.reconnect_requested = False

    def snapshot(self) -> dict[str, Any]:
        with self._lock:
            return {
                "ami_connected": self.ami_connected,
                "last_event_at": self.last_event_at,
                "last_error": self.last_error,
                "reconnect_requested": self.reconnect_requested,
            }

    def set_connected(self, value: bool) -> None:
        with self._lock:
            self.ami_connected = value
            if value:
                self.last_error = None

    def set_last_event(self) -> None:
        with self._lock:
            self.last_event_at = utc_now_iso()

    def set_error(self, message: str) -> None:
        with self._lock:
            self.last_error = message

    def request_reconnect(self) -> None:
        with self._lock:
            self.reconnect_requested = True

    def take_reconnect(self) -> bool:
        with self._lock:
            current = self.reconnect_requested
            self.reconnect_requested = False
            return current


_BACKGROUND_LOCK = threading.Lock()
_BACKGROUND_STOP_EVENT: threading.Event | None = None
_BACKGROUND_THREADS: list[threading.Thread] = []
_BRIDGE_SINGLETON_DB_CONNECTION = None
_BRIDGE_SINGLETON_FILE_HANDLE = None


def bridge_singleton_lock_path(*, database_url: str, cc_data_dir: str) -> Path:
    if database_url.startswith("sqlite:///"):
        db_path = Path(database_url[len("sqlite:///") :]).resolve()
        db_path.parent.mkdir(parents=True, exist_ok=True)
        return db_path.parent / "asterisk_bridge_service.singleton.lock"
    data_dir = Path(cc_data_dir).resolve()
    data_dir.mkdir(parents=True, exist_ok=True)
    return data_dir / "asterisk_bridge_service.singleton.lock"


def try_lock_file(handle) -> bool:
    try:
        handle.seek(0)
        handle.write(b"0")
        handle.flush()
        handle.seek(0)
    except OSError:
        return False
    if os.name == "nt":
        import msvcrt

        try:
            msvcrt.locking(handle.fileno(), msvcrt.LK_NBLCK, 1)
            return True
        except OSError:
            return False

    import fcntl

    try:
        fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
        return True
    except OSError:
        return False


def unlock_file(handle) -> None:
    handle.seek(0)
    if os.name == "nt":
        import msvcrt

        msvcrt.locking(handle.fileno(), msvcrt.LK_UNLCK, 1)
        return

    import fcntl

    fcntl.flock(handle.fileno(), fcntl.LOCK_UN)


def try_acquire_bridge_singleton_guard(
    *,
    database_url: str,
    engine: Any,
    lock_key: int,
    cc_data_dir: str,
) -> bool:
    global _BRIDGE_SINGLETON_DB_CONNECTION, _BRIDGE_SINGLETON_FILE_HANDLE

    if _BRIDGE_SINGLETON_DB_CONNECTION is not None or _BRIDGE_SINGLETON_FILE_HANDLE is not None:
        return True

    if database_url.startswith("postgresql"):
        connection = engine.raw_connection()
        try:
            cursor = connection.cursor()
            try:
                cursor.execute("SELECT pg_try_advisory_lock(%s)", (lock_key,))
                acquired_row = cursor.fetchone()
            finally:
                cursor.close()
            connection.commit()
        except Exception:
            connection.close()
            raise
        if acquired_row and bool(acquired_row[0]):
            _BRIDGE_SINGLETON_DB_CONNECTION = connection
            return True
        connection.close()
        return False

    lock_path = bridge_singleton_lock_path(database_url=database_url, cc_data_dir=cc_data_dir)
    lock_path.touch(exist_ok=True)
    handle = lock_path.open("r+b")
    if try_lock_file(handle):
        _BRIDGE_SINGLETON_FILE_HANDLE = handle
        return True
    try:
        handle.close()
    except OSError:
        pass
    return False


def release_bridge_singleton_guard(*, lock_key: int) -> None:
    global _BRIDGE_SINGLETON_DB_CONNECTION, _BRIDGE_SINGLETON_FILE_HANDLE

    if _BRIDGE_SINGLETON_DB_CONNECTION is not None:
        connection = _BRIDGE_SINGLETON_DB_CONNECTION
        _BRIDGE_SINGLETON_DB_CONNECTION = None
        try:
            cursor = connection.cursor()
            try:
                cursor.execute("SELECT pg_advisory_unlock(%s)", (lock_key,))
            finally:
                cursor.close()
            connection.commit()
        finally:
            connection.close()

    if _BRIDGE_SINGLETON_FILE_HANDLE is not None:
        handle = _BRIDGE_SINGLETON_FILE_HANDLE
        _BRIDGE_SINGLETON_FILE_HANDLE = None
        try:
            unlock_file(handle)
        finally:
            handle.close()


def bridge_singleton_db_connection():
    return _BRIDGE_SINGLETON_DB_CONNECTION


def bridge_singleton_file_handle():
    return _BRIDGE_SINGLETON_FILE_HANDLE


def background_threads_alive() -> list[threading.Thread]:
    return [thread for thread in _BACKGROUND_THREADS if thread.is_alive()]


def background_stop_event() -> threading.Event:
    global _BACKGROUND_STOP_EVENT
    if _BACKGROUND_STOP_EVENT is None or _BACKGROUND_STOP_EVENT.is_set():
        _BACKGROUND_STOP_EVENT = threading.Event()
    return _BACKGROUND_STOP_EVENT


def background_shutdown_timeout_seconds(*, poll_interval: float, failed_retry_interval_seconds: float) -> float:
    return max(10.0, poll_interval, failed_retry_interval_seconds) + 1.0


def start_background_threads(
    *,
    ami_loop,
    failed_retry_loop,
    ensure_guard,
    extra_loops: list[tuple[str, Any]] | None = None,
) -> None:
    global _BACKGROUND_THREADS

    with _BACKGROUND_LOCK:
        alive = background_threads_alive()
        if alive:
            _BACKGROUND_THREADS = alive
            ensure_guard()
            return

        stop_event = background_stop_event()
        ensure_guard()
        threads = [
            threading.Thread(target=ami_loop, args=(stop_event,), daemon=True, name="asterisk-ami-loop"),
            threading.Thread(
                target=failed_retry_loop,
                args=(stop_event,),
                daemon=True,
                name="asterisk-failed-retry-loop",
            ),
        ]
        for name, target in extra_loops or []:
            threads.append(threading.Thread(target=target, args=(stop_event,), daemon=True, name=name))
        _BACKGROUND_THREADS = threads
        try:
            for thread in threads:
                thread.start()
        except Exception:
            stop_event.set()
            _BACKGROUND_THREADS = []
            raise


def stop_background_threads(
    *,
    state: BridgeState,
    release_guard,
    shutdown_timeout_seconds: float,
) -> None:
    global _BACKGROUND_STOP_EVENT, _BACKGROUND_THREADS

    with _BACKGROUND_LOCK:
        stop_event = _BACKGROUND_STOP_EVENT
        threads = list(_BACKGROUND_THREADS)
        if stop_event is not None:
            stop_event.set()
        state.request_reconnect()

    for thread in threads:
        if thread.is_alive():
            thread.join(timeout=shutdown_timeout_seconds)

    with _BACKGROUND_LOCK:
        alive = background_threads_alive()
        if alive:
            _BACKGROUND_THREADS = alive
            return
        _BACKGROUND_THREADS = []
        _BACKGROUND_STOP_EVENT = None
        state.set_connected(False)
        release_guard()