"""Daemon IPC plumbing. AF_UNIX socket on POSIX, TCP loopback on Windows.

[WHO]: Provides socket path, process spawn, connect(), serve(), and endpoint cleanup helpers
[FROM]: Depends on asyncio, socket, subprocess, tempfile, pathlib for cross-platform daemon IPC
[TO]: Consumed by browser_harness.daemon, browser_harness.helpers, and browser_harness.admin
[HERE]: extensions/builtin/browser/src/browser_harness/_ipc.py within vendored Browser Harness package
"""
import asyncio, os, re, socket, subprocess, sys, tempfile
from pathlib import Path

IS_WINDOWS = sys.platform == "win32"
# BH_TMP_DIR set → caller-isolated dir, bare filenames (avoids AF_UNIX sun_path
# overrun: 104 macOS / 108 Linux). Unset → shared tmpdir, "bu-<NAME>" prefix
# disambiguates daemons. POSIX default is /tmp (gettempdir() returns long
# /var/folders/... on macOS); Windows uses TCP so any tempdir is fine.
BH_TMP_DIR = os.environ.get("BH_TMP_DIR")
_TMP = Path(BH_TMP_DIR or (tempfile.gettempdir() if IS_WINDOWS else "/tmp"))
_TMP.mkdir(parents=True, exist_ok=True)
_NAME_RE = re.compile(r"\A[A-Za-z0-9_-]{1,64}\Z")


def _check(name):  # path-traversal guard for BU_NAME
    if not _NAME_RE.match(name or ""):
        raise ValueError(f"invalid BU_NAME {name!r}: must match [A-Za-z0-9_-]{{1,64}}")
    return name


def _stem(name):  # "bu" when BH_TMP_DIR isolates us, else "bu-<NAME>"
    _check(name)
    return "bu" if BH_TMP_DIR else f"bu-{name}"


def log_path(name):   return _TMP / f"{_stem(name)}.log"
def pid_path(name):   return _TMP / f"{_stem(name)}.pid"
def port_path(name):  return _TMP / f"{_stem(name)}.port"  # Windows-only: holds the daemon's TCP port
def _sock_path(name): return _TMP / f"{_stem(name)}.sock"


def sock_addr(name):  # display-only, used in log lines
    if not IS_WINDOWS: return str(_sock_path(name))
    try: return f"127.0.0.1:{port_path(name).read_text().strip()}"
    except FileNotFoundError: return f"tcp:{_stem(name)}"


def spawn_kwargs():  # subprocess.Popen flags so the daemon detaches from this terminal
    if IS_WINDOWS:
        # CREATE_NO_WINDOW: no console window for the daemon. CREATE_NEW_PROCESS_GROUP:
        # daemon doesn't receive Ctrl-C/Ctrl-Break sent to the parent terminal, so
        # closing that terminal doesn't kill it. DETACHED_PROCESS is intentionally
        # omitted: per Win32 docs it overrides CREATE_NO_WINDOW, causing Windows to
        # allocate a fresh console for the (still console-subsystem) python.exe.
        return {"creationflags": subprocess.CREATE_NEW_PROCESS_GROUP | subprocess.CREATE_NO_WINDOW}
    return {"start_new_session": True}


def connect(name, timeout=1.0):
    """Blocking client. Raises FileNotFoundError if no daemon, TimeoutError on connect timeout."""
    if not IS_WINDOWS:
        # uv-Python on Windows lacks socket.AF_UNIX, so this branch must be gated.
        s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        s.settimeout(timeout); s.connect(str(_sock_path(name))); return s
    try: port = int(port_path(name).read_text().strip())
    except (FileNotFoundError, ValueError): raise FileNotFoundError(str(port_path(name)))
    s = socket.create_connection(("127.0.0.1", port), timeout=timeout)
    s.settimeout(timeout); return s


async def serve(name, handler):
    """Run the server until cancelled. handler(reader, writer) sees the same interface either way."""
    if not IS_WINDOWS:
        path = str(_sock_path(name))
        if os.path.exists(path): os.unlink(path)
        server = await asyncio.start_unix_server(handler, path=path)
        os.chmod(path, 0o600)
        async with server: await asyncio.Event().wait()
        return
    server = await asyncio.start_server(handler, "127.0.0.1", 0)
    pf = port_path(name)
    pf.write_text(str(server.sockets[0].getsockname()[1]))  # so clients can find us
    try:
        async with server: await asyncio.Event().wait()
    finally:
        try: pf.unlink()
        except FileNotFoundError: pass


def cleanup_endpoint(name):  # best-effort; silent if already gone
    p = _sock_path(name) if not IS_WINDOWS else port_path(name)
    try: p.unlink()
    except FileNotFoundError: pass