"""HTTP + WebSocket 接口。前端静态文件也由这里托管（同源，门户 vhost 直接反代到本进程）。

身份：统一门户在 vhost 里**覆盖式**注入 `X-Auth-User`（gen-vhosts.py），本进程只绑 127.0.0.1，
所以这个头只可能来自门户 —— 与 evse 一样按 L2 信任它，用来记审计「是谁按的」。
直接在服务器本机 curl（没有这个头）记作 `local`。
⚠️ 这个信任**只在绑 127.0.0.1 时成立**。哪天改绑 0.0.0.0，外面加一个 `X-Auth-User: admin`
就能冒名写审计 —— 所以 main.py 拒绝非回环地址的绑定。
"""
import asyncio
import json
from contextlib import asynccontextmanager
from typing import Awaitable, Callable, List, Optional

from fastapi import FastAPI, HTTPException, Request, WebSocket, WebSocketDisconnect
from fastapi.responses import FileResponse, JSONResponse
from fastapi.staticfiles import StaticFiles

from . import protocol
from .db import now_ms
from .hub import CmdTimeout, Hub
from .recovery import Params, RecoveryManager


def _user(req: Request) -> str:
    return req.headers.get("x-auth-user") or "local"


def _board_or_404(hub: Hub, board: str):
    if board not in hub.latest:
        raise HTTPException(404, "没有这块板子：%s" % board)


Hook = Callable[[], Awaitable[None]]


def create_app(hub: Hub, rec: RecoveryManager, static_dir: Optional[str] = None,
               startup: Optional[List[Hook]] = None, shutdown: Optional[List[Hook]] = None) -> FastAPI:
    # 用 lifespan 而不是 on_event：后者在新版 Starlette 里已经删了，服务器上的旧版两种都认。
    @asynccontextmanager
    async def lifespan(app):
        for fn in startup or []:
            await fn()
        yield
        for fn in shutdown or []:
            await fn()

    app = FastAPI(title="PhoneRail 机架管理", docs_url="/api/docs", openapi_url="/api/openapi.json",
                  lifespan=lifespan)

    @app.get("/api/health")
    def health():
        return {"ok": True, "mqtt": hub.mqtt_up, "boards": len(hub.latest), "ts": now_ms()}

    @app.get("/api/me")
    def me(req: Request):
        return {"user": _user(req), "roles": req.headers.get("x-auth-roles", "")}

    # ---- 板子 ----------------------------------------------------------------
    @app.get("/api/boards")
    def boards():
        return hub.boards()

    @app.get("/api/boards/{board}")
    def board(board: str):
        _board_or_404(hub, board)
        return hub.snapshot(board)

    @app.patch("/api/boards/{board}")
    async def patch_board(board: str, req: Request):
        _board_or_404(hub, board)
        body = await req.json()
        for k in ("name", "location"):
            if k in body:
                hub.db.x("UPDATE boards SET %s=? WHERE id=?" % k, (str(body[k])[:64], board))
        snap = hub.snapshot(board)
        hub.broadcast({"t": "board", "board": snap})
        return snap

    @app.patch("/api/boards/{board}/rails/{rail}")
    async def patch_phone(board: str, rail: int, req: Request):
        _board_or_404(hub, board)
        if not 1 <= rail <= protocol.RAILS:
            raise HTTPException(400, "rail 必须是 1~5")
        body = await req.json()
        for k in ("label", "adb_addr", "notes"):
            if k in body:
                hub.db.x("UPDATE phones SET %s=? WHERE board_id=? AND rail=?" % k,
                         (str(body[k])[:128], board, rail))
        if "auto_recover" in body:
            hub.db.x("UPDATE phones SET auto_recover=? WHERE board_id=? AND rail=?",
                     (1 if body["auto_recover"] else 0, board, rail))
            rec.reload()
        snap = hub.snapshot(board)
        hub.broadcast({"t": "board", "board": snap})
        return snap

    @app.post("/api/boards/{board}/cmd")
    async def command(board: str, req: Request):
        _board_or_404(hub, board)
        try:
            cmd = protocol.build_command(await req.json())
        except (protocol.CmdError, ValueError) as e:
            raise HTTPException(400, str(e))
        try:
            return await hub.call(board, cmd, _user(req))
        except CmdTimeout as e:
            raise HTTPException(504, str(e))

    @app.get("/api/boards/{board}/rails/{rail}/history")
    def history(board: str, rail: int, start: int, end: Optional[int] = None):
        _board_or_404(hub, board)
        return hub.db.history(board, rail, start, end or now_ms())

    # ---- 事件 / 审计 / 恢复 ----------------------------------------------------
    @app.get("/api/events")
    def events(board: Optional[str] = None, limit: int = 200, before: Optional[int] = None):
        sql = "SELECT * FROM events WHERE 1=1"
        args = []
        if board:
            sql += " AND board_id=?"
            args.append(board)
        if before:
            sql += " AND received<?"
            args.append(before)
        sql += " ORDER BY received DESC, seq DESC LIMIT ?"
        args.append(min(limit, 1000))
        out = []
        for r in hub.db.q(sql, args):
            d = dict(r)
            code, zh = protocol.EVENT_TYPES.get(d["type"], ("?", "未知事件 %d" % d["type"]))
            d.update(code=code, name=zh, severe=d["type"] in protocol.EVENT_SEVERE)
            if d["type"] == 13:
                d["name"] += "（%s）" % protocol.CMD_NAMES.get(d["arg"], d["arg"])
            out.append(d)
        return out

    @app.get("/api/audit")
    def audit(limit: int = 200, board: Optional[str] = None):
        sql, args = "SELECT * FROM audit", []
        if board:
            sql, args = sql + " WHERE board_id=?", [board]
        rows = hub.db.q(sql + " ORDER BY id DESC LIMIT ?", args + [min(limit, 1000)])
        return [dict(r, request=json.loads(r["request"]), response=json.loads(r["response"]))
                for r in rows]

    @app.get("/api/recovery")
    def recovery():
        from dataclasses import asdict
        log = hub.db.q("SELECT * FROM recovery_log ORDER BY id DESC LIMIT 200")
        return {"params": asdict(rec.params), "states": rec.states(), "log": [dict(r) for r in log]}

    @app.put("/api/recovery/params")
    async def recovery_params(req: Request):
        from dataclasses import asdict
        p = Params.load(json.dumps(await req.json()))
        rec.set_params(p)
        return asdict(p)

    @app.post("/api/recovery/{board}/{rail}/reset")
    def recovery_reset(board: str, rail: int):
        rec.reset(board, rail)
        return {"ok": True}

    # ---- 实时推送 --------------------------------------------------------------
    @app.websocket("/api/ws")
    async def ws(sock: WebSocket):
        await sock.accept()
        q = hub.subscribe_client()
        try:
            await sock.send_json({"t": "hello", "boards": hub.boards(), "mqtt": hub.mqtt_up,
                                  "user": sock.headers.get("x-auth-user") or "local"})
            while hub.is_subscribed(q):
                try:
                    msg = await asyncio.wait_for(q.get(), 20)
                except asyncio.TimeoutError:
                    msg = {"t": "ping", "ts": now_ms()}  # 保活，顺带让前端判断连接还活着
                await sock.send_json(msg)
        except WebSocketDisconnect:
            pass
        finally:
            hub.unsubscribe_client(q)

    # ---- 前端 ------------------------------------------------------------------
    if static_dir:
        app.mount("/assets", StaticFiles(directory=static_dir + "/assets"), name="assets")

        @app.get("/{path:path}")
        def spa(path: str):
            # index.html 不许缓存（home-server-deploy §关键约定 8 的教训）；assets 带哈希，可以长缓存
            resp = FileResponse(static_dir + "/index.html")
            resp.headers["Cache-Control"] = "no-cache"
            return resp

    return app