"""
KanBot web view — live board over Postgres LISTEN/NOTIFY.

Local-first, dependency-light: a single asyncio server that
  - serves the static branded SPA (index.html) over HTTP,
  - LISTENs on 'kb_events' and pushes every transition to connected browsers
    over a raw WebSocket (RFC 6455, hand-rolled — no node, no npm, no build step),
  - answers a tiny JSON snapshot endpoint so a fresh page loads the full board.

Why not Next.js: this is a dashboard, not a marketing site. One file, runs on the
M4 with the same asyncpg the rest of the stack uses. Swap it for a Next app later
without touching the core — it's just another MCP-contract client.

Run:  python -m web.server   then open http://localhost:8800
"""
from __future__ import annotations

import asyncio
import base64
import hashlib
import json
import os
import struct
from pathlib import Path

import asyncpg

HOST = os.environ.get("KANBOT_WEB_HOST", "127.0.0.1")
PORT = int(os.environ.get("KANBOT_WEB_PORT", "8800"))
DSN = os.environ.get("DATABASE_URL", "postgresql://kanbot:kanbot@localhost:5432/kanbot")
WS_MAGIC = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
INDEX = (Path(__file__).parent / "index.html").read_text()

_clients: set[asyncio.StreamWriter] = set()


# --------------------------------------------------------------------------- #
# board snapshot (fresh-load) — grouped by column
# --------------------------------------------------------------------------- #
async def board_snapshot(pool: asyncpg.Pool) -> dict:
    async with pool.acquire() as c:
        rows = await c.fetch(
            'SELECT id,title,"column",agent_role,assigned_worker,attempt,'
            "requires_human,tags FROM cards ORDER BY created_at")
    cols: dict[str, list] = {k: [] for k in
                             ("backlog", "ready", "in-flight", "review", "blocked", "done")}
    for r in rows:
        cols[r["column"]].append({
            "id": r["id"], "title": r["title"], "role": r["agent_role"],
            "worker": r["assigned_worker"], "attempt": r["attempt"],
            "requires_human": r["requires_human"], "tags": json.loads(r["tags"]),
        })
    return cols


# --------------------------------------------------------------------------- #
# minimal WebSocket frame encoder (server->client text frames only)
# --------------------------------------------------------------------------- #
def _ws_frame(text: str) -> bytes:
    data = text.encode()
    n = len(data)
    header = bytearray([0x81])  # FIN + text opcode
    if n < 126:
        header.append(n)
    elif n < 65536:
        header.append(126)
        header += struct.pack(">H", n)
    else:
        header.append(127)
        header += struct.pack(">Q", n)
    return bytes(header) + data


async def _broadcast(text: str) -> None:
    frame = _ws_frame(text)
    dead = []
    for w in list(_clients):
        try:
            w.write(frame)
            await w.drain()
        except Exception:
            dead.append(w)
    for w in dead:
        _clients.discard(w)


# --------------------------------------------------------------------------- #
# single connection handler: HTTP for pages + snapshot, Upgrade for WS
# --------------------------------------------------------------------------- #
async def handle(reader: asyncio.StreamReader, writer: asyncio.StreamWriter,
                 pool: asyncpg.Pool) -> None:
    try:
        req = await reader.readuntil(b"\r\n\r\n")
    except (asyncio.IncompleteReadError, ConnectionResetError):
        writer.close(); return
    head = req.decode(errors="ignore")
    line = head.split("\r\n", 1)[0]
    try:
        _, path, _ = line.split(" ", 2)
    except ValueError:
        writer.close(); return

    # WebSocket upgrade
    if "upgrade: websocket" in head.lower():
        key = ""
        for h in head.split("\r\n"):
            if h.lower().startswith("sec-websocket-key:"):
                key = h.split(":", 1)[1].strip()
        accept = base64.b64encode(
            hashlib.sha1((key + WS_MAGIC).encode()).digest()).decode()
        writer.write(
            b"HTTP/1.1 101 Switching Protocols\r\n"
            b"Upgrade: websocket\r\nConnection: Upgrade\r\n"
            b"Sec-WebSocket-Accept: " + accept.encode() + b"\r\n\r\n")
        await writer.drain()
        _clients.add(writer)
        # push initial snapshot so a fresh tab is instantly correct
        writer.write(_ws_frame(json.dumps(
            {"type": "snapshot", "board": await board_snapshot(pool)})))
        await writer.drain()
        try:
            while not reader.at_eof():
                await asyncio.sleep(30)  # we only push; ignore client frames
        finally:
            _clients.discard(writer)
        return

    # plain HTTP
    if path == "/snapshot":
        body = json.dumps(await board_snapshot(pool)).encode()
        ctype = b"application/json"
    else:
        body = INDEX.encode()
        ctype = b"text/html; charset=utf-8"
    writer.write(b"HTTP/1.1 200 OK\r\nContent-Type: " + ctype +
                 b"\r\nContent-Length: " + str(len(body)).encode() +
                 b"\r\nConnection: close\r\n\r\n" + body)
    await writer.drain()
    writer.close()


async def main() -> None:
    pool = await asyncpg.create_pool(DSN, min_size=1, max_size=4)
    listen_conn = await asyncpg.connect(DSN)

    async def on_notify(_c, _pid, _chan, payload):
        # re-snapshot on any transition (board is small; correctness > cleverness)
        await _broadcast(json.dumps(
            {"type": "event", "event": json.loads(payload),
             "board": await board_snapshot(pool)}))

    await listen_conn.add_listener("kb_events", on_notify)

    server = await asyncio.start_server(
        lambda r, w: handle(r, w, pool), HOST, PORT)
    print(f"[web] KanBot board live at http://{HOST}:{PORT}")
    async with server:
        await server.serve_forever()


if __name__ == "__main__":
    asyncio.run(main())