"""Fixtures for the live-socket integration tests (phase 06).

`ws_server` is a **session-scoped factory**: one call spins up a real
`uvicorn.Server` on an ephemeral `127.0.0.1` port (never a routable
interface, the `tests/conftest.py` rule) with a `Settings` built from
explicit overrides, and the returned context manager tears the server down
when the test ends (a second safety close runs at session end). The app
built here is yielded too, so a test can reach
`app.state.supervisor` / `app.state.ws_registry` to assert on room state
(the seat's `connected` flag, the disconnect count, the socket census)
without a second client.

`live_server` is the same machine with one difference (phase 06 task 06):
`UNDERCOOKED_LEVELS_DIR` defaults to the **shipped** `levels/` campaign
(so a live server carries the real content the content gate validated),
and any `Settings` field is overridable (`tick_hz`, `snapshot_hz`, …)
the same way. A test that needs a playable, tunable round passes
`levels_dir=` explicitly (the task-03 `levels_playable` precedent, and
the task-06 `levels_round` campaign for the two-client round).

`ws_client` is a factory for the **scripted client** the round/abuse/
budget suites drive: `ws_client(url, code, token=None, name=None)` returns
a `WsClient` — a connected `websockets` socket with the `hello` already
sent and a background reader that keeps a merged snapshot view
(`client.view`: full frames reset, delta frames patch, the frame's `e`
events accumulated) and a frame queue the client's three verbs consume:

* `send_model(model)` — queue a JSON frame out (fire-and-forget; the
  `input` choreography cadence never blocks on the socket),
* `expect(type, wait_s)` — the next frame of `type`, the ones before it
  buffered and re-queued for a later verb,
* `collect(type, n, wait_s)` — `n` frames of `type`, the same buffering.

Settings are built the same way `tests/conftest.py` `make_settings` does
(the authoritative `config.ENV_KEYS` contract, ambient `UNDERCOOKED_*` /
`DEBUGPY*` stripped), with one deliberate difference: `UNDERCOOKED_LEVELS_DIR`
is *kept*, pointed at the campaign the factory names (the `ws_server`
default is the fixture-isolated campaign with a recipe book — `create_app`'s
content gate passes on it; the `live_server` default is the shipped
`levels/`), so a live server boots without the ambient environment leaking.
"""

from __future__ import annotations

import asyncio
import contextlib
import importlib
import json
import os
import threading
import time
from collections.abc import Callable, Iterator
from pathlib import Path
from typing import Any

import httpx
import pytest
import websockets
from websockets.exceptions import ConnectionClosed

LIVE_HOST = "127.0.0.1"  # loopback only: tests never listen on a routable interface
FIXTURE_LEVELS_DIR = Path(__file__).resolve().parents[1] / "fixtures" / "levels"
#: The shipped campaign (phase 06 task 06: `live_server` boots on it).
SHIPPED_LEVELS_DIR = Path(__file__).resolve().parents[2] / "levels"

config = importlib.import_module("undercooked.config")
app_module = importlib.import_module("undercooked.server.app")
uvicorn = importlib.import_module("uvicorn")


def _as_env_value(value: object) -> str:
    """Render a Python override value the way `load_settings` reads it."""
    if isinstance(value, bool):
        return "1" if value else "0"
    if isinstance(value, (str, os.PathLike)):
        return str(value)
    if isinstance(value, (list, tuple)):
        return ",".join(str(item) for item in value)
    return str(value)


def _wait_for_health(url: str, thread: threading.Thread) -> None:
    """Poll `/healthz` until it answers 200, failing fast if the server died."""
    timeout = float(os.environ.get("UNDERCOOKED_TEST_SOCKET_TIMEOUT", "30"))
    deadline = time.monotonic() + timeout
    probe = httpx.Client(timeout=1.0, trust_env=False)
    try:
        while time.monotonic() < deadline:
            try:
                if probe.get(f"{url}/healthz").status_code == 200:
                    return
            except httpx.HTTPError:
                pass
            thread.join(timeout=0.05)
            if not thread.is_alive():
                raise RuntimeError(f"live server on {url} died during startup")
    finally:
        probe.close()
    raise RuntimeError(f"live server on {url} never became healthy")


def _start_live_server(
    overrides: dict[str, object], default_levels_dir: Path
) -> tuple[str, Any, Any, threading.Thread]:
    """One live uvicorn server on an ephemeral loopback port.

    Returns `(url, application, server, thread)`; the caller wraps them in
    a `LiveServer`. `overrides` are `Settings` field names (validated
    against `config.ENV_KEYS`) — `levels_dir` falls back to
    `default_levels_dir` when the caller did not name it.
    """
    env = {
        key: value
        for key, value in os.environ.items()
        if not key.startswith(("UNDERCOOKED_", "DEBUGPY"))
    }
    levels_dir = overrides.get("levels_dir", str(default_levels_dir))
    env["UNDERCOOKED_LEVELS_DIR"] = str(levels_dir)
    for name, value in overrides.items():
        env[config.ENV_KEYS[name]] = _as_env_value(value)
    settings = config.load_settings(env)
    application = app_module.create_app(settings)
    uv_config = uvicorn.Config(
        application,
        host=LIVE_HOST,
        port=0,
        log_level="warning",
        lifespan="on",
    )
    sock = uv_config.bind_socket()
    url = f"http://{LIVE_HOST}:{sock.getsockname()[1]}"
    server = uvicorn.Server(uv_config)
    thread = threading.Thread(
        target=server.run, args=([sock],), name="undercooked-live-ws", daemon=True
    )
    thread.start()
    _wait_for_health(url, thread)
    return url, application, server, thread


class LiveServer:
    """One live uvicorn server + its app (a context manager)."""

    def __init__(
        self,
        url: str,
        app: Any,
        server: Any,
        thread: threading.Thread,
    ) -> None:
        self.url = url
        self.app = app
        self._server = server
        self._thread = thread
        self._closed = False

    def __enter__(self) -> tuple[str, Any]:
        return self.url, self.app

    def __exit__(self, *exc_info: object) -> None:
        self.close()

    def close(self) -> None:
        if self._closed:
            return
        self._closed = True
        self._server.should_exit = True
        self._thread.join(timeout=15)
        if self._thread.is_alive():  # pragma: no cover - only if uvicorn stops cooperating
            pytest.fail("live server did not shut down within 15 s")


def _make_live_server_factory(default_levels_dir: Path) -> Callable[..., LiveServer]:
    started: list[LiveServer] = []

    def _make(**overrides: object) -> LiveServer:
        if any(name not in config.ENV_KEYS for name in overrides):
            unknown = sorted(name for name in overrides if name not in config.ENV_KEYS)
            pytest.fail(f"unknown settings override(s): {unknown}")
        url, application, server, thread = _start_live_server(
            dict(overrides), default_levels_dir
        )
        live = LiveServer(url, application, server, thread)
        started.append(live)
        return live

    # The session-end safety close (the context managers already closed
    # each server the test used; a leaked one is closed here).
    def _close_all() -> None:
        for live in started:
            live.close()

    _make.close_all = _close_all  # type: ignore[attr-defined]
    return _make


@pytest.fixture(scope="session")
def ws_server() -> Iterator[Any]:
    """Factory: `with ws_server(max_players=2) as (url, app): …`

    Each call starts a fresh `uvicorn.Server` on an ephemeral loopback
    port (bound before the serving thread starts, so the port is known
    exactly — no log scraping, no collisions between concurrent runs)
    with a `Settings` resolved from `**overrides` (field name, the same
    names `make_settings` takes) over the fixture-isolated content.
    Shutdown is requested through `should_exit` so the lifespan stops
    cleanly.
    """
    factory = _make_live_server_factory(FIXTURE_LEVELS_DIR)
    yield factory
    factory.close_all()  # type: ignore[attr-defined]


@pytest.fixture(scope="session")
def live_server() -> Iterator[Any]:
    """Factory: `with live_server(tick_hz=20, snapshot_hz=20) as (url, app): …`

    The task-06 live server: identical to `ws_server` except that
    `UNDERCOOKED_LEVELS_DIR` points at the **shipped** `levels/` campaign
    by default (override with `levels_dir=…`), so the server boots on the
    real, gate-validated content; `tick_hz` / `snapshot_hz` (and every
    other `Settings` field) are overridable the same way.
    """
    factory = _make_live_server_factory(SHIPPED_LEVELS_DIR)
    yield factory
    factory.close_all()  # type: ignore[attr-defined]


def _uri(url: str, code: str) -> str:
    return url.replace("http://", "ws://", 1) + f"/ws/{code}"


class View:
    """The merged snapshot state a scripted client reasons about.

    Full frames (`full: true`, the `FULL_SNAPSHOT_EVERY_TICKS` cadence and
    the `needs_full` resync) reset the entity maps; delta frames patch
    them (a `{"gone": true}` tombstone removes). The frame's `e` events
    accumulate in order (a boundary's events ride that one frame — the
    encoder's contract), and the scalars (`t` / `s` / `tgt` / `tl`) are
    the latest frame's."""

    def __init__(self) -> None:
        self.players: dict[str, dict[str, Any]] = {}
        self.stations: dict[tuple[Any, Any, Any], dict[str, Any]] = {}
        self.orders: dict[Any, dict[str, Any]] = {}
        self.events: list[dict[str, Any]] = []
        self.tick = 0
        self.score = 0
        self.target = 0
        self.time_left = 0.0

    def apply(self, frame: dict[str, Any]) -> None:
        # The wire's `snap` message nests the encoder's frame under
        # `raw` (`SnapMsg.raw` — the message's own `p` is the protocol
        # version, the frame's own `p` is its players array).
        raw = frame.get("raw")
        if not isinstance(raw, dict):
            raw = frame
        if raw.get("full"):
            self.players = {
                entry["i"]: entry for entry in raw.get("p", []) if not entry.get("gone")
            }
            self.stations = {
                (entry["k"], entry["x"], entry["y"]): entry
                for entry in raw.get("st", [])
                if not entry.get("gone")
            }
            self.orders = {
                entry["id"]: entry for entry in raw.get("o", []) if not entry.get("gone")
            }
        else:
            delta = raw.get("d", {})
            for entry in delta.get("p", []):
                if entry.get("gone"):
                    self.players.pop(entry["i"], None)
                else:
                    self.players[entry["i"]] = entry
            for entry in delta.get("st", []):
                key = (entry["k"], entry["x"], entry["y"])
                if entry.get("gone"):
                    self.stations.pop(key, None)
                else:
                    self.stations[key] = entry
            for entry in delta.get("o", []):
                if entry.get("gone"):
                    self.orders.pop(entry["id"], None)
                else:
                    self.orders[entry["id"]] = entry
        self.events.extend(raw.get("e", []))
        self.tick = raw.get("t", self.tick)
        self.score = raw.get("s", self.score)
        self.target = raw.get("tgt", self.target)
        self.time_left = raw.get("tl", self.time_left)

    def player(self, index: str) -> dict[str, Any] | None:
        return self.players.get(index)

    def station(self, kind: str) -> dict[str, Any] | None:
        """The first station of `kind` (the wire's identity is `k` +
        coordinates; a level's stations of a kind sit on distinct cells)."""
        for (station_kind, _, _), entry in self.stations.items():
            if station_kind == kind:
                return entry
        return None

    def station_at(self, kind: str, x: float, y: float) -> dict[str, Any] | None:
        return self.stations.get((kind, x, y))

    def event(self, kind: str) -> dict[str, Any] | None:
        for entry in self.events:
            if entry.get("k") == kind:
                return entry
        return None


class WsClient:
    """The scripted client (phase 06 task 06): a connected socket with the
    `hello` already sent, a background reader (the merged `view` + the
    frame queue), and the three verbs the suites name.

    `send_model` is fire-and-forget by design: the choreography feeds
    inputs on a cadence (~10 Hz) faster than the socket's round trip, and
    the room's per-tick queue is what serialises the intent — a blocking
    send would add latency the sim's queue discipline is meant to absorb.
    `raw` is the malformed-frame half (the abuse suite's `not json`).
    """

    def __init__(self, ws: Any) -> None:
        self.ws = ws
        self.view = View()
        self._frames: asyncio.Queue[dict[str, Any]] = asyncio.Queue()
        self._reader: asyncio.Task[None] | None = None
        self._closed = False

    def start(self) -> None:
        self._reader = asyncio.get_running_loop().create_task(self._read_loop())

    async def _read_loop(self) -> None:
        try:
            while not self._closed:
                raw = await self.ws.recv()
                frame = json.loads(raw)
                if frame.get("type") == "snap":
                    self.view.apply(frame)
                self._frames.put_nowait(frame)
        except (ConnectionClosed, asyncio.CancelledError):
            pass

    def send_model(self, model: dict[str, Any]) -> None:
        """Queue one JSON frame out (the task's `send_model`)."""
        asyncio.get_running_loop().create_task(self.ws.send(json.dumps(model)))

    async def send(self, model: dict[str, Any]) -> None:
        await self.ws.send(json.dumps(model))

    def raw(self, text: str) -> None:
        """One raw frame out (the malformed half: `not json`, …)."""
        asyncio.get_running_loop().create_task(self.ws.send(text))

    async def expect(self, type: str, wait_s: float = 10.0) -> dict[str, Any]:
        """The next frame of `type`; the frames before it are buffered and
        re-queued (a later verb sees them, in order)."""
        loop = asyncio.get_running_loop()
        deadline = loop.time() + wait_s
        held: list[dict[str, Any]] = []
        while True:
            remaining = deadline - loop.time()
            if remaining <= 0:
                msg = f"timed out waiting for a {type!r} frame"
                raise TimeoutError(msg)
            frame = await asyncio.wait_for(self._frames.get(), remaining)
            if frame.get("type") == type:
                for buffered in held:
                    self._frames.put_nowait(buffered)
                return frame
            held.append(frame)

    async def collect(
        self, type: str, n: int, wait_s: float = 30.0
    ) -> list[dict[str, Any]]:
        """`n` frames of `type` (in arrival order); the frames in between
        are buffered and re-queued."""
        loop = asyncio.get_running_loop()
        deadline = loop.time() + wait_s
        held: list[dict[str, Any]] = []
        out: list[dict[str, Any]] = []
        while len(out) < n:
            remaining = deadline - loop.time()
            if remaining <= 0:
                msg = f"timed out after {len(out)}/{n} {type!r} frames"
                raise TimeoutError(msg)
            frame = await asyncio.wait_for(self._frames.get(), remaining)
            if frame.get("type") == type:
                out.append(frame)
            else:
                held.append(frame)
        for buffered in held:
            self._frames.put_nowait(buffered)
        return out

    async def close(self) -> None:
        if self._closed:
            return
        self._closed = True
        if self._reader is not None:
            self._reader.cancel()
            with contextlib.suppress(asyncio.CancelledError, Exception):
                await self._reader
        with contextlib.suppress(Exception):
            await self.ws.close()


@pytest.fixture(scope="session")
def ws_client() -> Callable[..., Any]:
    """Factory: `client = await ws_client(url, code, token, name=…)`.

    Connects to `/ws/{code}` on the named server, sends the `hello`
    (the `token` joins or rejoins; absent, the server mints one and the
    `welcome` echoes it) and returns the scripted `WsClient` (its reader
    already running, its `view` merging snapshots).
    """

    async def _connect(
        url: str,
        code: str,
        token: str | None = None,
        *,
        name: str | None = None,
        p: int = 1,
    ) -> WsClient:
        ws = await websockets.connect(_uri(url, code), open_timeout=10.0)
        hello: dict[str, Any] = {"type": "hello", "p": p}
        if token is not None:
            hello["token"] = token
        if name is not None:
            hello["name"] = name
        await ws.send(json.dumps(hello))
        client = WsClient(ws)
        client.start()
        return client

    return _connect