"""bbuilder host — supervises Nerimity bots built in the bbuilder desktop app.

Runs as a Hugging Face Space. FastAPI serves the control API that the Godot app talks to;
a small Gradio dashboard is mounted at / so the Space is also useful in a browser.

Every /api/* route requires the X-BB-Key header to match the APP_KEY Space secret. /health
is deliberately open — the self-ping uses it, and it leaks nothing but bot names and status.
"""
from __future__ import annotations

import asyncio
import hmac
import json
import os
import queue
import threading
import time
from pathlib import Path

import gradio as gr
import uvicorn

# ZeroGPU refuses to start a Space with no @spaces.GPU function ("No @spaces.GPU function
# detected during startup"). We're on ZeroGPU only because free cpu-basic Gradio Spaces now
# require PRO — the bots themselves are pure CPU. gpu_probe() below satisfies the check and
# doubles as a diagnostic; nothing in the hosting path calls it.
try:
    import spaces

    HAS_SPACES = True
except ImportError:  # running locally, outside a ZeroGPU container
    HAS_SPACES = False
from fastapi import Depends, FastAPI, Header, HTTPException, Request, UploadFile
from fastapi.responses import JSONResponse
from fastapi import WebSocket, WebSocketDisconnect

import noderuntime
import store
from compiler import CompileError, compile_project
from supervisor import BOTS_DIR, DATA_DIR, Supervisor

## The runtime shipped into every bot. Host-owned, never uploaded by a client.
RUNTIME_SRC = Path(__file__).parent / "bb-runtime.js"

APP_KEY = os.environ.get("APP_KEY", "")
SELF_URL = os.environ.get("BB_SELF_URL", "https://joddabod-bbuilder-host.hf.space")
PING_INTERVAL = 300          # 5 minutes
BOOT_TIME = time.time()

STATE: dict = {"node": None, "node_version": "?", "sdk_ready": False, "boot_log": []}


def _catalog_blocks() -> dict:
    """Block catalogue the compiler builds from; also a cheap liveness signal."""
    try:
        return json.loads((Path(__file__).parent / "catalog.json").read_text())["blocks"]
    except Exception:  # noqa: BLE001
        return {}


def boot_log(msg: str) -> None:
    print(f"[boot] {msg}", flush=True)
    STATE["boot_log"].append(msg)


sup = Supervisor(on_change=lambda bot: store.push_bot_async(bot.id))
app = FastAPI(title="bbuilder host", docs_url=None, redoc_url=None)


# --------------------------------------------------------------------------- auth

def require_key(x_bb_key: str = Header(default="")) -> None:
    if not APP_KEY:
        raise HTTPException(503, "host has no APP_KEY configured")
    if x_bb_key != APP_KEY:
        raise HTTPException(401, "bad or missing X-BB-Key")


Auth = Depends(require_key)


# --------------------------------------------------------------------------- open

## Bump when the API contract changes. /health reports it so a deploy can tell whether the
## new code is actually live — the old process answers /health perfectly well while the new
## one is still building, which otherwise makes a deploy look successful when it isn't.
API_VERSION = 4


@app.get("/health")
def health() -> dict:
    """Open endpoint. Used by the self-ping and by the app's connection check."""
    return {
        "ok": True,
        "service": "bbuilder-host",
        "api_version": API_VERSION,
        "blocks": len(_catalog_blocks()),
        "uptime": round(time.time() - BOOT_TIME, 1),
        "node": STATE["node_version"],
        "sdk_ready": STATE["sdk_ready"],
        "store": store.available(),
        "bots": [
            {"id": b.id, "name": b.name, "status": b.status, "enabled": b.enabled}
            for b in sup.bots.values()
        ],
    }


# --------------------------------------------------------------------------- api

@app.get("/api/bots", dependencies=[Auth])
def list_bots() -> dict:
    return {"bots": [b.public() for b in sup.bots.values()]}


@app.post("/api/bots/{bot_id}/deploy", dependencies=[Auth])
async def deploy(bot_id: str, request: Request, project: UploadFile) -> dict:
    """Deploy a project.

    Takes **block structure only** — the client never sends JavaScript. The host compiles
    it against its own catalog.json, so the worst a caller can do is describe a bot out of
    blocks the catalogue already offers. bb-runtime.js is the host's copy, not an upload.
    """
    if not bot_id.replace("-", "").replace("_", "").isalnum():
        raise HTTPException(400, "bot id must be alphanumeric/dash/underscore")

    form = await request.form()
    name = str(form.get("name") or bot_id)
    token = str(form.get("token") or "")
    autostart = str(form.get("autostart") or "").lower() in ("1", "true", "yes")

    raw = await project.read()
    if len(raw) > 4 * 1024 * 1024:
        raise HTTPException(413, "project is too large")
    try:
        project_data = json.loads(raw)
    except json.JSONDecodeError as exc:
        raise HTTPException(400, f"project is not valid JSON: {exc}") from exc
    if not isinstance(project_data, dict):
        raise HTTPException(400, "project must be a JSON object")

    try:
        source, warnings = compile_project(project_data, name)
    except CompileError as exc:
        raise HTTPException(400, f"could not build this project: {exc}") from exc

    if not RUNTIME_SRC.exists():
        raise HTTPException(500, "host is missing bb-runtime.js")

    bot = sup.ensure(bot_id, name)
    was_running = bot.status == "running"
    if was_running:
        sup.stop(bot_id, disable=False)

    (bot.dir / "bot.js").write_text(source)
    (bot.dir / "bb-runtime.js").write_text(RUNTIME_SRC.read_text())
    (bot.dir / "project.bbproj").write_bytes(raw)
    (bot.dir / "package.json").write_text(json.dumps({
        "name": f"bb-{bot_id}", "private": True, "type": "commonjs",
    }, indent=2))

    if token:
        secret = bot.dir / "secret.json"
        secret.write_text(json.dumps({"token": token}))
        secret.chmod(0o600)

    sup.write_meta(bot)
    sup.log(bot, f"compiled {len(source.splitlines())} lines from blocks", "system")
    for w in warnings:
        sup.log(bot, w, "system")
    store.push_bot_async(bot_id)

    if was_running or autostart:
        ok, msg = sup.start(bot_id)
        return {"ok": True, "deployed": True, "started": ok, "message": msg,
                "warnings": warnings, "bot": bot.public()}
    return {"ok": True, "deployed": True, "started": False,
            "warnings": warnings, "bot": bot.public()}


@app.post("/api/bots/{bot_id}/start", dependencies=[Auth])
def start_bot(bot_id: str) -> dict:
    ok, msg = sup.start(bot_id)
    if not ok:
        raise HTTPException(400, msg)
    return {"ok": True, "message": msg, "bot": sup.get(bot_id).public()}


@app.post("/api/bots/{bot_id}/stop", dependencies=[Auth])
def stop_bot(bot_id: str) -> dict:
    ok, msg = sup.stop(bot_id)
    if not ok:
        raise HTTPException(404, msg)
    return {"ok": True, "message": msg, "bot": sup.get(bot_id).public()}


@app.post("/api/bots/{bot_id}/restart", dependencies=[Auth])
def restart_bot(bot_id: str) -> dict:
    ok, msg = sup.restart(bot_id)
    if not ok:
        raise HTTPException(400, msg)
    return {"ok": True, "message": msg, "bot": sup.get(bot_id).public()}


def _bot_id_for_token(token: str) -> str | None:
    """Find the bot whose stored Nerimity token matches.

    Holding a bot's token is proof of ownership of that bot, so this lets someone manage
    their bot from any machine without knowing the id it was deployed under. Compared with
    compare_digest so the endpoint can't be used as a timing oracle to recover a token.
    """
    token = token.strip()
    if not token:
        return None
    for bot in sup.bots.values():
        secret = bot.dir / "secret.json"
        if not secret.exists():
            continue
        try:
            stored = str(json.loads(secret.read_text()).get("token", ""))
        except (json.JSONDecodeError, OSError):
            continue
        if stored and hmac.compare_digest(stored, token):
            return bot.id
    return None


# Deliberately NOT under /api/bots/… : "/api/bots/by-token/start" would be matched by
# "/api/bots/{bot_id}/start" with bot_id="by-token", because that route is registered first.
@app.post("/api/token/{action}", dependencies=[Auth])
async def by_token(action: str, request: Request) -> dict:
    """Start, stop, restart or delete a bot identified only by its Nerimity token."""
    if action not in ("start", "stop", "restart", "delete"):
        raise HTTPException(400, "unknown action")

    form = await request.form()
    bot_id = _bot_id_for_token(str(form.get("token") or ""))
    if bot_id is None:
        raise HTTPException(404, "no bot on this host is using that token")

    if action == "delete":
        sup.delete(bot_id)
        store.delete_bot(bot_id)
        return {"ok": True, "id": bot_id, "message": "deleted"}

    handler = {"start": sup.start, "stop": sup.stop, "restart": sup.restart}[action]
    ok, msg = handler(bot_id)
    if not ok:
        raise HTTPException(400, msg)
    return {"ok": True, "id": bot_id, "message": msg, "bot": sup.get(bot_id).public()}


@app.post("/api/reap", dependencies=[Auth])
def reap() -> dict:
    """Kill any bot process the registry has lost track of.

    Exists because an orphaned bot is otherwise unreachable — it keeps running and keeps
    talking to Nerimity with no endpoint able to stop it.
    """
    killed = sup.reap_orphans()
    return {"ok": True, "killed": killed, "count": len(killed)}


@app.post("/api/token", dependencies=[Auth])
async def resolve_token(request: Request) -> dict:
    """Look up which bot a token belongs to, so the editor can attach to its log stream."""
    form = await request.form()
    bot_id = _bot_id_for_token(str(form.get("token") or ""))
    if bot_id is None:
        raise HTTPException(404, "no bot on this host is using that token")
    return {"ok": True, "id": bot_id, "bot": sup.get(bot_id).public()}


@app.get("/api/bots/{bot_id}/logs", dependencies=[Auth])
def get_logs(bot_id: str, since: int = 0) -> dict:
    bot = sup.get(bot_id)
    if bot is None:
        raise HTTPException(404, "no such bot")
    return {"bot": bot.public(), "lines": sup.logs_since(bot, since)}


@app.delete("/api/bots/{bot_id}", dependencies=[Auth])
def delete_bot(bot_id: str) -> dict:
    if not sup.delete(bot_id):
        raise HTTPException(404, "no such bot")
    store.delete_bot(bot_id)
    return {"ok": True}


@app.websocket("/api/bots/{bot_id}/stream")
async def stream(ws: WebSocket, bot_id: str) -> None:
    """Live log + block-step feed. Key is passed as ?key= since browsers/Godot can't set
    headers on a WebSocket handshake."""
    key = ws.query_params.get("key", "")
    if not APP_KEY or key != APP_KEY:
        await ws.close(code=4401)
        return
    bot = sup.get(bot_id)
    if bot is None:
        await ws.close(code=4404)
        return

    await ws.accept()
    q: queue.Queue = queue.Queue(maxsize=1000)
    sup.subscribe(bot, q)
    try:
        since = int(ws.query_params.get("since", "0"))
        for line in sup.logs_since(bot, since):
            await ws.send_text(json.dumps(line))
        loop = asyncio.get_running_loop()
        while True:
            try:
                item = await loop.run_in_executor(None, q.get, True, 25)
            except queue.Empty:
                await ws.send_text(json.dumps({"t": "ping", "status": bot.status}))
                continue
            await ws.send_text(json.dumps(item))
    except (WebSocketDisconnect, RuntimeError):
        pass
    finally:
        sup.unsubscribe(bot, q)


@app.exception_handler(HTTPException)
async def http_error(_: Request, exc: HTTPException) -> JSONResponse:
    return JSONResponse({"ok": False, "error": exc.detail}, status_code=exc.status_code)


# --------------------------------------------------------------------------- boot

def bootstrap() -> None:
    boot_log(store.pull_all())
    sup.load_from_disk()
    boot_log(f"registry has {len(sup.bots)} bot(s)")

    node = noderuntime.ensure_node()
    STATE["node"] = node
    STATE["node_version"] = noderuntime.version(node)
    sup.node_bin = node
    boot_log(f"node {STATE['node_version']}")

    STATE["sdk_ready"] = noderuntime.ensure_nerimity(node)
    boot_log(f"nerimity.js ready: {STATE['sdk_ready']}")

    sup.start_enabled()
    boot_log("boot complete")


def self_ping() -> None:
    """Keep the Space awake. Defeats the inactivity GC; does not prevent HF-side rebuilds,
    which is why bootstrap() restores from the dataset."""
    import urllib.request
    while True:
        time.sleep(PING_INTERVAL)
        try:
            with urllib.request.urlopen(f"{SELF_URL}/health", timeout=30) as r:
                r.read(1)
        except Exception as exc:  # noqa: BLE001
            print(f"[ping] failed: {exc}", flush=True)


# --------------------------------------------------------------------------- ui

def dashboard_rows() -> list[list]:
    return [
        [b.id, b.name, b.status, "yes" if b.enabled else "no", b.restarts, b.last_error or ""]
        for b in sup.bots.values()
    ] or [["—", "no bots deployed yet", "", "", "", ""]]


if HAS_SPACES:

    @spaces.GPU(duration=10)
    def gpu_probe() -> str:
        """Required by ZeroGPU, and genuinely useful as a one-click sanity check."""
        try:
            import torch

            if torch.cuda.is_available():
                return f"GPU reachable: {torch.cuda.get_device_name(0)} (unused by bot hosting)"
            return "No CUDA device visible (fine — bots are CPU-only)"
        except ImportError:
            return "torch not installed; GPU unused by design — bots are CPU-only"

else:

    def gpu_probe() -> str:
        return "not running on ZeroGPU"


def status_text() -> str:
    return (
        f"**node** `{STATE['node_version']}` · **nerimity.js** "
        f"{'ready' if STATE['sdk_ready'] else 'not installed'} · "
        f"**durable store** {'on' if store.available() else 'OFF'} · "
        f"**uptime** {round((time.time() - BOOT_TIME) / 60)} min"
    )


with gr.Blocks(title="bbuilder host") as demo:
    gr.Markdown("# bbuilder host\nHosting for Nerimity bots built with the bbuilder desktop app.")
    status_md = gr.Markdown(status_text())
    table = gr.Dataframe(
        headers=["id", "name", "status", "enabled", "restarts", "last error"],
        value=dashboard_rows,
        interactive=False,
        wrap=True,
    )
    with gr.Row():
        refresh = gr.Button("Refresh", variant="primary")
        probe = gr.Button("GPU probe")
    probe_out = gr.Textbox(label="probe", interactive=False, visible=True)
    refresh.click(lambda: (status_text(), dashboard_rows()), outputs=[status_md, table])
    probe.click(gpu_probe, outputs=probe_out)
    gr.Markdown(
        "Control endpoints live under `/api/` and require the `X-BB-Key` header. "
        "This dashboard is read-only."
    )

def serve_via_gradio() -> None:
    """Let Gradio own the server, then graft our API routes onto its FastAPI app.

    ZeroGPU only notices @spaces.GPU functions when the app starts through
    demo.launch(); calling uvicorn.run() on our own app bypasses that hook and the Space is
    killed with "No @spaces.GPU function detected during startup". So Gradio launches, and
    we insert our routes at the *front* of its router afterwards — position matters because
    Gradio registers a catch-all for SPA routing that would otherwise shadow /health.
    """
    demo.queue()
    demo.launch(
        server_name="0.0.0.0",
        server_port=int(os.environ.get("GRADIO_SERVER_PORT") or 7860),
        prevent_thread_lock=True,
        show_api=False,
        # Gradio 5 defaults to running a Node SSR server in front of the Python app. It
        # answers any path it doesn't recognise with the SPA shell, which swallows /health
        # and /api/* before FastAPI sees them. We need FastAPI in front, so: no SSR.
        ssr_mode=False,
    )
    target = demo.app
    for route in reversed(app.router.routes):
        if getattr(route, "path", None) in ("/openapi.json",):
            continue
        target.router.routes.insert(0, route)
    print(f"[boot] grafted {len(app.router.routes)} API routes onto the Gradio app", flush=True)

    while True:
        time.sleep(3600)


def serve_standalone() -> None:
    """Local development: no ZeroGPU to appease, so run our own app directly."""
    port = int(os.environ.get("GRADIO_SERVER_PORT") or os.environ.get("BB_PORT") or 7860)
    uvicorn.run(gr.mount_gradio_app(app, demo, path="/"), host="0.0.0.0", port=port)


if __name__ == "__main__":
    DATA_DIR.mkdir(parents=True, exist_ok=True)
    BOTS_DIR.mkdir(parents=True, exist_ok=True)
    threading.Thread(target=bootstrap, daemon=True).start()
    threading.Thread(target=self_ping, daemon=True).start()
    if HAS_SPACES:
        serve_via_gradio()
    else:
        serve_standalone()