eth_hedge_sim/control/backend/app/api/nodes.py

from __future__ import annotations

import asyncio
import json
import secrets
import time
from typing import Annotated, Any, AsyncIterator

from fastapi import APIRouter, Depends, HTTPException, Request
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field

from ..auth import require_control_user
from ..config import get_control_settings
from ..crypto import seal
from ..db import get_control_db
from ..proxy import call_node, probe_health

router = APIRouter(prefix="/api/nodes", tags=["nodes"])


def _public_node(row: dict[str, Any]) -> dict[str, Any]:
    return {
        "id": row["id"],
        "name": row["name"],
        "base_url": row["base_url"],
        "token_configured": bool((row.get("token_sealed") or "").strip()),
        "created_at_ms": row["created_at_ms"],
        "updated_at_ms": row["updated_at_ms"],
    }


def _http_detail(data: Any) -> str:
    if isinstance(data, dict):
        d = data.get("detail", data)
        return d if isinstance(d, str) else str(d)
    return str(data)


def _raise_node_error(code: int, data: Any) -> None:
    """策略机错误不得用 401 回传，否则中控前端会误清登录态。"""
    detail = _http_detail(data)
    if code in (401, 403):
        raise HTTPException(
            status_code=502,
            detail=f"策略机鉴权失败（{detail}）。请在中控重新生成 Token，并到策略机「系统设置→登录账户」保存同一 Token。",
        )
    raise HTTPException(
        status_code=code if 400 <= code < 600 else 502,
        detail=detail,
    )


async def _collect_one_status(
    node: dict[str, Any],
    *,
    timeout: float | None = None,
) -> dict[str, Any]:
    """有 Token 时只打 fleet/status（含在线+持仓），避免 /health 再多一跳。"""
    base = _public_node(node)
    if not node.get("token_sealed"):
        probe = await probe_health(node, timeout=timeout)
        return {
            **base,
            **probe,
            "fleet_ok": False,
            "fleet_error": "未生成 Token",
        }

    code, data = await call_node(
        node, "GET", "/api/fleet/status", timeout=timeout
    )
    if code == 200 and isinstance(data, dict):
        return {
            **base,
            "online": True,
            "health": None,
            "error": None,
            "from_fleet": True,
            "fleet": data,
            "fleet_ok": True,
        }

    # fleet 失败时再探 /health，区分离线 vs Token 错误
    probe = await probe_health(node, timeout=timeout)
    item: dict[str, Any] = {
        **base,
        **probe,
        "fleet_ok": False,
        "fleet_error": _http_detail(data) if data else f"HTTP {code}",
    }
    return item


async def collect_all_status(*, timeout: float | None = None) -> list[dict[str, Any]]:
    db = get_control_db()
    nodes = db.list_nodes()
    if not nodes:
        return []
    return list(
        await asyncio.gather(
            *[_collect_one_status(n, timeout=timeout) for n in nodes]
        )
    )


class NodeCreate(BaseModel):
    name: str = Field(min_length=1, max_length=64)
    base_url: str = Field(min_length=8, max_length=256)


class NodeUpdate(BaseModel):
    name: str | None = Field(default=None, min_length=1, max_length=64)
    base_url: str | None = Field(default=None, min_length=8, max_length=256)


@router.get("/")
async def list_nodes(_user: Annotated[str, Depends(require_control_user)]) -> dict:
    db = get_control_db()
    nodes = [_public_node(n) for n in db.list_nodes()]
    return {"nodes": nodes}


@router.post("/")
async def create_node(
    body: NodeCreate,
    _user: Annotated[str, Depends(require_control_user)],
) -> dict:
    db = get_control_db()
    try:
        row = db.create_node(body.name, body.base_url)
    except Exception as e:
        raise HTTPException(status_code=400, detail=f"创建失败: {e}") from e
    return _public_node(row)


@router.get("/status/all")
async def status_all(_user: Annotated[str, Depends(require_control_user)]) -> dict:
    items = await collect_all_status()
    return {"nodes": items}


@router.get("/status/stream")
async def status_stream(
    request: Request,
    _user: Annotated[str, Depends(require_control_user)],
) -> StreamingResponse:
    """SSE：中控约每秒并行拉取策略机状态并推送到浏览器。"""
    settings = get_control_settings()
    interval = max(0.5, float(settings.control_sse_interval_sec))
    probe_timeout = max(0.5, float(settings.control_sse_probe_timeout_sec))

    async def event_gen() -> AsyncIterator[str]:
        last_payload = ""
        last_heartbeat = 0.0
        yield f": connected interval={interval}\n\n"
        while True:
            if await request.is_disconnected():
                break
            started = time.monotonic()
            try:
                items = await collect_all_status(timeout=probe_timeout)
                payload = json.dumps(
                    {"nodes": items, "ts_ms": int(time.time() * 1000)},
                    ensure_ascii=False,
                    separators=(",", ":"),
                    default=str,
                )
                now = time.monotonic()
                if payload != last_payload:
                    last_payload = payload
                    yield f"event: nodes\ndata: {payload}\n\n"
                    last_heartbeat = now
                elif now - last_heartbeat >= 5.0:
                    yield f": heartbeat {int(time.time())}\n\n"
                    last_heartbeat = now
            except Exception as e:
                err = json.dumps(
                    {"detail": str(e)},
                    ensure_ascii=False,
                    separators=(",", ":"),
                )
                yield f"event: error\ndata: {err}\n\n"
            elapsed = time.monotonic() - started
            await asyncio.sleep(max(0.05, interval - elapsed))

    return StreamingResponse(
        event_gen(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no",
        },
    )


@router.post("/update-batch")
async def update_batch(
    body: dict,
    _user: Annotated[str, Depends(require_control_user)],
) -> dict:
    ids = body.get("ids") or []
    if not isinstance(ids, list) or not ids:
        raise HTTPException(status_code=400, detail="ids 不能为空")
    db = get_control_db()
    results = []
    for nid in ids:
        node = db.get_node(int(nid))
        if not node:
            results.append({"id": nid, "ok": False, "detail": "不存在"})
            continue
        code, data = await call_node(node, "POST", "/api/fleet/update")
        results.append(
            {
                "id": nid,
                "ok": code < 400,
                "status": code,
                "result": data,
            }
        )
    return {"results": results}


@router.patch("/{node_id}")
async def update_node(
    node_id: int,
    body: NodeUpdate,
    _user: Annotated[str, Depends(require_control_user)],
) -> dict:
    db = get_control_db()
    row = db.update_node(node_id, name=body.name, base_url=body.base_url)
    if not row:
        raise HTTPException(status_code=404, detail="节点不存在")
    return _public_node(row)


@router.delete("/{node_id}")
async def delete_node(
    node_id: int,
    _user: Annotated[str, Depends(require_control_user)],
) -> dict:
    db = get_control_db()
    if not db.delete_node(node_id):
        raise HTTPException(status_code=404, detail="节点不存在")
    return {"ok": True}


@router.post("/{node_id}/generate-token")
async def generate_token(
    node_id: int,
    _user: Annotated[str, Depends(require_control_user)],
) -> dict:
    """生成新 Token，加密存中控；明文仅返回一次，需粘贴到策略机设置。"""
    db = get_control_db()
    node = db.get_node(node_id)
    if not node:
        raise HTTPException(status_code=404, detail="节点不存在")
    plain = secrets.token_urlsafe(32)
    sealed = seal(plain, get_control_settings().control_auth_secret)
    db.update_node(node_id, token_sealed=sealed)
    return {
        "ok": True,
        "token": plain,
        "msg": "请立即复制并到策略机「系统设置 → 登录账户 → 中控 API Token」保存",
    }


@router.get("/{node_id}/status")
async def node_status(
    node_id: int,
    _user: Annotated[str, Depends(require_control_user)],
) -> dict:
    db = get_control_db()
    node = db.get_node(node_id)
    if not node:
        raise HTTPException(status_code=404, detail="节点不存在")
    return await _collect_one_status(node)


@router.get("/{node_id}/stats")
async def node_stats(
    node_id: int,
    _user: Annotated[str, Depends(require_control_user)],
) -> dict:
    """代理策略机整体统计（点击详情时按需拉取，不进 SSE）。"""
    db = get_control_db()
    node = db.get_node(node_id)
    if not node:
        raise HTTPException(status_code=404, detail="节点不存在")
    if not node.get("token_sealed"):
        raise HTTPException(status_code=400, detail="未生成 Token")
    code, data = await call_node(node, "GET", "/api/fleet/stats", timeout=15.0)
    if code >= 400:
        _raise_node_error(code, data)
    if not isinstance(data, dict):
        raise HTTPException(status_code=502, detail="策略机统计返回异常")
    return data


@router.post("/{node_id}/start")
async def node_start(
    node_id: int,
    _user: Annotated[str, Depends(require_control_user)],
) -> dict:
    db = get_control_db()
    node = db.get_node(node_id)
    if not node:
        raise HTTPException(status_code=404, detail="节点不存在")
    code, data = await call_node(node, "POST", "/api/fleet/start")
    if code >= 400:
        _raise_node_error(code, data)
    return {"ok": True, "result": data}


@router.post("/{node_id}/pause")
async def node_pause(
    node_id: int,
    _user: Annotated[str, Depends(require_control_user)],
) -> dict:
    db = get_control_db()
    node = db.get_node(node_id)
    if not node:
        raise HTTPException(status_code=404, detail="节点不存在")
    code, data = await call_node(node, "POST", "/api/fleet/pause")
    if code >= 400:
        _raise_node_error(code, data)
    return {"ok": True, "result": data}


@router.post("/{node_id}/update")
async def node_update(
    node_id: int,
    _user: Annotated[str, Depends(require_control_user)],
) -> dict:
    db = get_control_db()
    node = db.get_node(node_id)
    if not node:
        raise HTTPException(status_code=404, detail="节点不存在")
    code, data = await call_node(node, "POST", "/api/fleet/update")
    if code >= 400:
        _raise_node_error(code, data)
    return {"ok": True, "result": data}


@router.post("/{node_id}/login-url")
async def node_login_url(
    node_id: int,
    _user: Annotated[str, Depends(require_control_user)],
) -> dict:
    """用 Fleet Token 向策略机签发一次性登录票，返回可打开的 URL。"""
    db = get_control_db()
    node = db.get_node(node_id)
    if not node:
        raise HTTPException(status_code=404, detail="节点不存在")
    code, data = await call_node(node, "POST", "/api/fleet/issue-login")
    if code >= 400:
        _raise_node_error(code, data)
    path = ""
    if isinstance(data, dict):
        path = str(data.get("login_path") or "")
    if not path:
        raise HTTPException(status_code=502, detail="策略机未返回 login_path")
    base = str(node["base_url"]).rstrip("/")
    return {
        "ok": True,
        "url": f"{base}{path}",
        "expires_in": data.get("expires_in") if isinstance(data, dict) else None,
    }