#!/usr/bin/env python3
"""Watch fireworks requests via Linux inotify or Win11 HTTP polling (no pip deps)."""

from __future__ import annotations

import ctypes
import errno
import json
import os
import select
import struct
import sys
import threading
import urllib.error
import urllib.request
from typing import Callable

from fireworks_ipc import (
    fireworks_request_is_fresh,
    read_fireworks_request,
    resolve_fireworks_request_path,
)

IN_MODIFY = 0x00000002
IN_CLOSE_WRITE = 0x00000008
IN_MOVED_FROM = 0x00000040
IN_MOVED_TO = 0x00000080
IN_CREATE = 0x00000100
IN_DELETE = 0x00000200

_DIR_MASK = IN_CREATE | IN_DELETE | IN_MODIFY | IN_CLOSE_WRITE | IN_MOVED_FROM | IN_MOVED_TO


def _libc() -> ctypes.CDLL:
    if sys.platform.startswith("linux"):
        return ctypes.CDLL("libc.so.6", use_errno=True)
    raise OSError("inotify requires Linux")


def _inotify_init() -> int:
    fd = _libc().inotify_init()
    if fd < 0:
        err = ctypes.get_errno()
        raise OSError(err, os.strerror(err), "inotify_init")
    return fd


def _inotify_add_watch(fd: int, path: str, mask: int) -> int:
    watch = _libc().inotify_add_watch(fd, path.encode("utf-8"), mask)
    if watch < 0:
        err = ctypes.get_errno()
        raise OSError(err, os.strerror(err), f"inotify_add_watch({path})")
    return watch


def _inotify_rm_watch(fd: int, wd: int) -> None:
    if _libc().inotify_rm_watch(fd, wd) < 0:
        err = ctypes.get_errno()
        if err != errno.EINVAL:
            raise OSError(err, os.strerror(err), "inotify_rm_watch")


def _iter_inotify_events(buf: bytes):
    offset = 0
    while offset + 16 <= len(buf):
        _wd, mask, _cookie, name_len = struct.unpack_from("iIII", buf, offset)
        offset += 16
        name = b""
        if name_len:
            name = buf[offset : offset + name_len].split(b"\x00", 1)[0]
            offset += name_len
        yield mask, name.decode("utf-8", errors="replace")


class FireworksRequestWatcher:
    """Daemon thread: inotify on EtherGold dir; callback on atomic fireworks-request.json write."""

    def __init__(
        self,
        on_request: Callable[[dict], None],
        *,
        request_path: str | None = None,
        poll_fallback_s: float = 0.5,
    ):
        self._on_request = on_request
        self._request_path = request_path or resolve_fireworks_request_path()
        self._watch_dir = os.path.dirname(self._request_path) or "."
        self._basename = os.path.basename(self._request_path)
        self._poll_fallback_s = max(0.1, float(poll_fallback_s))
        self._stop = threading.Event()
        self._thread: threading.Thread | None = None
        self._last_seq: int | None = None
        self._last_mtime = 0.0

    def start(self) -> None:
        if self._thread is not None:
            return
        try:
            self._last_mtime = os.path.getmtime(self._request_path)
        except OSError:
            self._last_mtime = 0.0
        payload = read_fireworks_request(self._request_path)
        if payload:
            seq = payload.get("seq")
            if isinstance(seq, int):
                self._last_seq = seq
        self._thread = threading.Thread(target=self._run, name="fireworks-inotify", daemon=True)
        self._thread.start()

    def stop(self) -> None:
        self._stop.set()

    def _dispatch_if_new(self) -> None:
        payload = read_fireworks_request(self._request_path)
        if not payload:
            return
        if not fireworks_request_is_fresh(payload):
            return
        seq = payload.get("seq")
        if seq is not None and seq == self._last_seq:
            return
        self._last_seq = seq if isinstance(seq, int) else None
        self._on_request(payload)

    def _poll_mtime_if_changed(self) -> None:
        try:
            mtime = os.path.getmtime(self._request_path)
        except OSError:
            return
        if mtime > self._last_mtime:
            self._last_mtime = mtime
            self._dispatch_if_new()

    def _event_targets_request(self, name: str) -> bool:
        if not name:
            return True
        if name == self._basename:
            return True
        return name.startswith(".fireworks-request.") and name.endswith(".tmp")

    def _run(self) -> None:
        try:
            self._run_inotify()
        except OSError as exc:
            print(f"fireworks watcher: inotify unavailable ({exc}); mtime poll only", flush=True)
            self._run_mtime_poll()

    def _run_inotify(self) -> None:
        os.makedirs(self._watch_dir, exist_ok=True)
        fd = _inotify_init()
        wd = _inotify_add_watch(fd, self._watch_dir, _DIR_MASK)
        print(
            f"fireworks watcher: inotify+mtime on {self._watch_dir} -> {self._basename}",
            flush=True,
        )
        try:
            while not self._stop.is_set():
                ready, _, _ = select.select([fd], [], [], self._poll_fallback_s)
                self._poll_mtime_if_changed()
                if not ready:
                    continue
                buf = os.read(fd, 4096)
                for event_mask, name in _iter_inotify_events(buf):
                    if not self._event_targets_request(name):
                        continue
                    if event_mask & _DIR_MASK:
                        self._poll_mtime_if_changed()
        finally:
            try:
                _inotify_rm_watch(fd, wd)
            except OSError:
                pass
            os.close(fd)

    def _run_mtime_poll(self) -> None:
        print(
            f"fireworks watcher: mtime poll on {self._request_path} "
            f"every {self._poll_fallback_s:.1f}s",
            flush=True,
        )
        while not self._stop.is_set():
            self._poll_mtime_if_changed()
            self._stop.wait(self._poll_fallback_s)


class RemoteFireworksRequestWatcher:
    """Win11 OHD poller for the office Pi's fireworks-request endpoint."""

    def __init__(
        self,
        on_request: Callable[[dict], None],
        *,
        request_url: str,
        poll_interval_s: float = 0.5,
        timeout_s: float = 1.5,
        max_age_sec: float = 45.0,
    ):
        self._on_request = on_request
        self._request_url = request_url
        self._poll_interval_s = max(0.1, float(poll_interval_s))
        self._timeout_s = max(0.1, float(timeout_s))
        self._max_age_sec = max(1.0, float(max_age_sec))
        self._stop = threading.Event()
        self._thread: threading.Thread | None = None
        self._last_seq: int | None = None
        self._had_connection_error = False

    def start(self) -> None:
        if self._thread is not None:
            return
        # Establish a baseline without replaying a request that predates OHD startup.
        payload = self._fetch()
        if payload:
            seq = payload.get("seq")
            if isinstance(seq, int):
                self._last_seq = seq
        self._thread = threading.Thread(
            target=self._run,
            name="fireworks-http-poll",
            daemon=True,
        )
        self._thread.start()
        print(
            f"fireworks watcher: HTTP poll {self._request_url} "
            f"every {self._poll_interval_s:.1f}s",
            flush=True,
        )

    def stop(self) -> None:
        self._stop.set()

    def _fetch(self) -> dict | None:
        request = urllib.request.Request(
            self._request_url,
            headers={
                "Accept": "application/json",
                "Cache-Control": "no-cache",
            },
        )
        try:
            with urllib.request.urlopen(request, timeout=self._timeout_s) as response:
                payload = json.loads(response.read().decode("utf-8"))
            self._had_connection_error = False
            return payload if isinstance(payload, dict) else None
        except urllib.error.HTTPError as exc:
            # 404 is the normal idle state: no active request file on the Pi.
            if exc.code == 404:
                self._had_connection_error = False
                return None
            self._warn_connection_once(exc)
        except (OSError, urllib.error.URLError, TimeoutError, ValueError, json.JSONDecodeError) as exc:
            self._warn_connection_once(exc)
        return None

    def _warn_connection_once(self, exc: BaseException) -> None:
        if not self._had_connection_error:
            print(f"fireworks watcher: HTTP unavailable ({exc})", flush=True)
        self._had_connection_error = True

    def _dispatch_if_new(self) -> None:
        payload = self._fetch()
        if not payload or not fireworks_request_is_fresh(
            payload, max_age_sec=self._max_age_sec
        ):
            return
        seq = payload.get("seq")
        if seq is not None and seq == self._last_seq:
            return
        self._last_seq = seq if isinstance(seq, int) else None
        self._on_request(payload)

    def _run(self) -> None:
        while not self._stop.is_set():
            self._dispatch_if_new()
            self._stop.wait(self._poll_interval_s)