"""Bounded pre-trigger observer for the existing Task26 gateway window."""

from __future__ import annotations

import ctypes
import json
import os
import select
import struct
import subprocess
import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path

from gateway.platforms.telegram_customer_bootstrap import (
    BootstrapState,
    RoomBootstrapStore,
    room_bootstrap_state_dir,
)


_IN_CLOSE_WRITE = 0x00000008
_IN_MOVED_TO = 0x00000080


def _emit(pipe_path: Path, payload: dict[str, object]) -> None:
    with pipe_path.open("w", encoding="utf-8") as stream:
        stream.write(json.dumps(payload, sort_keys=True, separators=(",", ":")) + "\n")
        stream.flush()


def _names(payload: bytes) -> set[bytes]:
    names: set[bytes] = set()
    offset = 0
    while offset + 16 <= len(payload):
        _watch, _mask, _cookie, length = struct.unpack_from("iIII", payload, offset)
        name = payload[offset + 16 : offset + 16 + length].split(b"\0", 1)[0]
        if name:
            names.add(name)
        offset += 16 + length
    return names


def _state(profile: Path) -> dict[str, object]:
    value = json.loads((profile / "gateway_state.json").read_text(encoding="utf-8"))
    return value if isinstance(value, dict) else {}


def _connected(value: dict[str, object]) -> bool:
    platforms = value.get("platforms")
    telegram = platforms.get("telegram") if isinstance(platforms, dict) else None
    return (
        value.get("gateway_state") == "running"
        and isinstance(telegram, dict)
        and telegram.get("state") == "connected"
    )


def main() -> int:
    profile = Path(sys.argv[1])
    session_id = sys.argv[2]
    pipe_path = Path(sys.argv[3])
    timeout_seconds = int(sys.argv[4])
    service = "hermes-gateway-dualcoachtest.service"
    store = RoomBootstrapStore(room_bootstrap_state_dir(profile))
    session = store.get(session_id)
    if session.state is not BootstrapState.PREPARED or session.generation != 1:
        _emit(pipe_path, {"kind": "INITIAL_STATE_FAILURE", "state": session.state.value, "generation": session.generation})
        return 2
    if session.expires_at <= datetime.now(timezone.utc):
        _emit(pipe_path, {"kind": "INITIAL_STATE_FAILURE", "state": "EXPIRED_BY_CLOCK"})
        return 2

    journal = subprocess.Popen(
        ["journalctl", "--user", "--follow", "--no-pager", "-o", "short-iso", "-u", service, "--since", "now"],
        stdout=subprocess.PIPE,
        stderr=subprocess.DEVNULL,
        bufsize=0,
    )
    libc = ctypes.CDLL(None, use_errno=True)
    descriptor = libc.inotify_init1(os.O_CLOEXEC)
    if descriptor < 0:
        journal.terminate()
        raise OSError(ctypes.get_errno(), "inotify_init1")
    try:
        if libc.inotify_add_watch(
            descriptor,
            os.fsencode(profile),
            _IN_CLOSE_WRITE | _IN_MOVED_TO,
        ) < 0:
            raise OSError(ctypes.get_errno(), "inotify_add_watch")
        _emit(pipe_path, {"kind": "SUBSCRIBED", "state_path": str(profile / "gateway_state.json"), "journal_unit": service})
        deadline = min(
            datetime.now(timezone.utc) + timedelta(seconds=timeout_seconds),
            session.expires_at,
        )
        journal_started = False
        journal_buffer = b""
        while True:
            remaining = (deadline - datetime.now(timezone.utc)).total_seconds()
            if remaining <= 0:
                _emit(pipe_path, {"kind": "TIMEOUT"})
                return 3
            watched = [descriptor]
            if journal.stdout is not None:
                watched.append(journal.stdout.fileno())
            readable, _, _ = select.select(watched, [], [], remaining)
            if not readable:
                continue
            if descriptor in readable:
                changed = _names(os.read(descriptor, 65536))
                if b"gateway_state.json" in changed:
                    try:
                        current = _state(profile)
                    except (OSError, ValueError):
                        continue
                    if _connected(current):
                        _emit(pipe_path, {"kind": "CONNECTED", "pid": current.get("pid")})
                        return 0
                    if current.get("gateway_state") == "startup_failed":
                        _emit(pipe_path, {"kind": "STARTUP_FAILURE", "gateway_state": "startup_failed"})
                        return 4
            if journal.stdout is not None and journal.stdout.fileno() in readable:
                chunk = os.read(journal.stdout.fileno(), 65536)
                if not chunk:
                    continue
                journal_buffer += chunk
                lines = journal_buffer.split(b"\n")
                journal_buffer = lines.pop()
                for raw in lines:
                    line = raw.decode("utf-8", "replace")
                    if ": Started hermes-gateway-dualcoachtest.service" in line:
                        journal_started = True
                    if journal_started and (
                        ": Failed to start hermes-gateway-dualcoachtest.service" in line
                        or ": Stopped hermes-gateway-dualcoachtest.service" in line
                    ):
                        _emit(pipe_path, {"kind": "STARTUP_FAILURE", "journal": line})
                        return 4
    finally:
        os.close(descriptor)
        if journal.poll() is None:
            journal.terminate()
        try:
            journal.wait(timeout=2)
        except subprocess.TimeoutExpired:
            journal.kill()
            journal.wait()


if __name__ == "__main__":
    raise SystemExit(main())