"""Observe one prepared bootstrap session without invoking Telegram or providers."""

from __future__ import annotations

import ctypes
import json
import os
import select
import socket
import struct
import sys
from datetime import datetime, 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 _notify(message: str) -> None:
    address = os.environ.get("NOTIFY_SOCKET")
    if not address:
        return
    if address.startswith("@"):
        address = "\0" + address[1:]
    with socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM) as channel:
        channel.connect(address)
        channel.sendall(message.encode("utf-8"))


def _gateway_is_connected(profile: Path) -> bool:
    try:
        state = json.loads((profile / "gateway_state.json").read_text("utf-8"))
    except (OSError, ValueError):
        return False
    telegram = state.get("platforms", {}).get("telegram", {})
    return (
        state.get("gateway_state") == "running"
        and telegram.get("state") == "connected"
    )


def _event_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 main() -> int:
    profile = Path(sys.argv[1])
    session_id = sys.argv[2]
    expires_at = datetime.fromisoformat(sys.argv[3]).astimezone(timezone.utc)
    store = RoomBootstrapStore(room_bootstrap_state_dir(profile))
    current = store.get(session_id)
    if current.state is not BootstrapState.PREPARED or current.generation != 1:
        return 2
    libc = ctypes.CDLL(None, use_errno=True)
    descriptor = libc.inotify_init1(os.O_CLOEXEC)
    if descriptor < 0:
        raise OSError(ctypes.get_errno(), "inotify_init1")
    try:
        for path in (store.state_dir, profile):
            if libc.inotify_add_watch(
                descriptor,
                os.fsencode(path),
                _IN_CLOSE_WRITE | _IN_MOVED_TO,
            ) < 0:
                raise OSError(ctypes.get_errno(), f"inotify_add_watch:{path}")
        _notify("READY=1\nSTATUS=claim watcher subscribed to canonical ledger")
        while True:
            remaining = (expires_at - datetime.now(timezone.utc)).total_seconds()
            if remaining <= 0:
                return 3
            readable, _, _ = select.select([descriptor], [], [], remaining)
            if not readable:
                continue
            names = _event_names(os.read(descriptor, 65536))
            if b"ledger.json" in names:
                current = store.get(session_id)
                if current.state is not BootstrapState.PREPARED:
                    return 0
            if b"gateway_state.json" in names and _gateway_is_connected(profile):
                _notify("STATUS=claim watcher subscribed; gateway telegram connected")
    finally:
        os.close(descriptor)


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