"""VNC 反向代理。

设计：
- HTTP POST /v1/instances/{id}/vnc-ticket
  鉴权后向上游 (POST /clouds/{upstream_id}/vnc) 取凭据，把上游 wss/凭据存 Redis（一次性 ticket, TTL 60s）。
  返回前端 {ticket, vnc_pass, password, hostname, ...} —— vnc_url 不直接返回，前端只用 ticket 连本站 ws。

- WS  /v1/vnc/{ticket}
  浏览器连这里。我们消费 ticket（取出 + 立即删除），按上游 url 建立第二条 ws，双向转发字节流。
  浏览器不需要也无法在 WS 中带 JWT；一次性 ticket 即为身份凭证。

为啥不让前端直连上游 wss：
- 上游 wss 可能是 ws://（不安全），混合内容会被浏览器拦
- 暴露真实上游 IP/Token
- 跨域 / 证书问题
"""
from __future__ import annotations

import asyncio
import json
import logging
import secrets
import ssl
from typing import Any

import aiohttp
from fastapi import APIRouter, Depends, HTTPException, WebSocket, WebSocketDisconnect
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from starlette.websockets import WebSocketState

from app.core.deps import get_current_user
from app.database import get_db
from app.models.instance import Instance
from app.models.user import User
from app.redis_client import k, redis_client
from app.schemas.common import ApiResponse
from app.services.dcim_cloud_client import DcimCloudError
from app.services import zjmf_instance_adapter

log = logging.getLogger("qycw.vnc")

router = APIRouter()

TICKET_TTL_SECONDS = 60


# 给上游 wss 用的"宽松 SSL"上下文：
# - 不校验证书链（上游可能是自签或域名错配）
# - 不校验 hostname
# - 明确启用 TLS 1.2+；某些上游对 TLS 1.3 ALPN 协商不友好，留个默认
def _lax_ssl_context() -> ssl.SSLContext:
    ctx = ssl.create_default_context()
    ctx.check_hostname = False
    ctx.verify_mode = ssl.CERT_NONE
    # 让 TCP 立刻发出小包，避免 Nagle 延迟
    return ctx


def _ticket_key(ticket: str) -> str:
    return k("vnc_ticket", ticket)


@router.post("/instances/{instance_id}/vnc-ticket", response_model=ApiResponse)
async def issue_vnc_ticket(
    instance_id: int,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """签发一次性 VNC ticket，前端用它连 /v1/vnc/{ticket}。"""
    # 局部 import 避免循环依赖
    from app.routers.instances import _get_dcim_client

    res = await db.execute(
        select(Instance).where(
            Instance.id == instance_id, Instance.user_id == current_user.id
        )
    )
    instance = res.scalar_one_or_none()
    if not instance:
        raise HTTPException(status_code=404, detail="实例不存在")
    if not instance.upstream_id:
        raise HTTPException(status_code=400, detail="实例尚未开通，无法获取 VNC")
    if not instance.cloud_server_id:
        raise HTTPException(status_code=400, detail="未绑定云节点")

    if await zjmf_instance_adapter.get_zjmf_server_for_instance(db, instance):
        payload = await zjmf_instance_adapter.get_vnc_info(db, instance)
        vnc_url = payload.get("vnc_url")
        if not vnc_url:
            raise HTTPException(status_code=502, detail="魔方财务未返回 VNC 地址")
        if str(vnc_url).startswith(("http://", "https://")):
            return ApiResponse(data={
                "ticket": "",
                "expires_in": 0,
                "vnc_pass": payload.get("vnc_pass"),
                "password": payload.get("password"),
                "hostname": instance.hostname,
                "instance_id": instance.id,
                "direct_url": vnc_url,
                "provider": "zjmf_finance",
                "connection_type": "direct_url",
            })

        ticket = secrets.token_urlsafe(24)
        ticket_data = {
            "user_id": current_user.id,
            "instance_id": instance.id,
            "upstream_id": instance.upstream_id,
            "vnc_url": vnc_url,
            "vnc_pass": payload.get("vnc_pass") or "",
        }
        await redis_client.set(_ticket_key(ticket), json.dumps(ticket_data), ex=TICKET_TTL_SECONDS)
        return ApiResponse(data={
            "ticket": ticket,
            "expires_in": TICKET_TTL_SECONDS,
            "vnc_pass": payload.get("vnc_pass"),
            "password": payload.get("password"),
            "hostname": instance.hostname,
            "instance_id": instance.id,
            "provider": "zjmf_finance",
            "connection_type": "websocket",
        })

    client, _ = await _get_dcim_client(instance.cloud_server_id, db)
    try:
        code, text, body = await client.request(
            f"/clouds/{instance.upstream_id}/vnc", method="POST", timeout=10.0
        )
    except DcimCloudError as e:
        raise HTTPException(status_code=502, detail=f"上游调用失败：{e}")
    if not (200 <= code < 300):
        raise HTTPException(status_code=502, detail=f"上游返回异常（HTTP {code}）：{text[:200]}")
    payload = (body or {}).get("data") or body or {}
    if not isinstance(payload, dict):
        raise HTTPException(status_code=502, detail="上游返回数据格式异常")

    vnc_url = payload.get("vnc_url")
    if not vnc_url:
        raise HTTPException(status_code=502, detail="上游未返回 vnc_url")

    ticket = secrets.token_urlsafe(24)
    ticket_data = {
        "user_id": current_user.id,
        "instance_id": instance.id,
        "upstream_id": instance.upstream_id,
        "vnc_url": vnc_url,
        "vnc_pass": payload.get("vnc_pass") or "",
    }
    await redis_client.set(_ticket_key(ticket), json.dumps(ticket_data), ex=TICKET_TTL_SECONDS)

    return ApiResponse(data={
        "ticket": ticket,
        "expires_in": TICKET_TTL_SECONDS,
        "vnc_pass": payload.get("vnc_pass"),
        "password": payload.get("password"),
        "hostname": instance.hostname,
        "instance_id": instance.id,
    })


async def _consume_ticket(ticket: str) -> dict[str, Any] | None:
    """取出 ticket 并立即删除（原子）。Redis GETDEL 保证并发时只有一个调用拿到。"""
    try:
        # redis-py 4.0+ 提供 getdel
        raw = await redis_client.getdel(_ticket_key(ticket))
    except AttributeError:
        # 老版本 fallback：pipeline 模拟原子
        async with redis_client.pipeline(transaction=True) as pipe:
            pipe.get(_ticket_key(ticket))
            pipe.delete(_ticket_key(ticket))
            results = await pipe.execute()
        raw = results[0] if results else None
    if not raw:
        return None
    try:
        return json.loads(raw)
    except json.JSONDecodeError:
        return None


@router.websocket("/vnc/{ticket}")
async def vnc_bridge(ws: WebSocket, ticket: str):
    """浏览器 ↔ 我们 ↔ 上游 wss 的双向二进制转发。"""
    # 上游 RFB 需要 binary 子协议
    requested_protocols = ws.headers.get("sec-websocket-protocol", "")
    subprotocols = [p.strip() for p in requested_protocols.split(",") if p.strip()]
    accept_proto = "binary" if "binary" in subprotocols else (subprotocols[0] if subprotocols else None)

    ticket_data = await _consume_ticket(ticket)
    if not ticket_data:
        await ws.close(code=4401)  # 未授权 / 票据过期
        return

    upstream_url = ticket_data.get("vnc_url")
    if not upstream_url:
        await ws.close(code=4400)
        return

    await ws.accept(subprotocol=accept_proto)
    log.info("vnc bridge open: instance=%s url=%s requested_proto=%s accept_proto=%s",
             ticket_data.get("instance_id"), upstream_url, subprotocols, accept_proto)

    # 大缓冲 + 长心跳：减少调度次数和 keepalive 开销
    # 上游可能是 wss + 自签证书 / 错配域名，用宽松 SSL 上下文
    ssl_ctx = _lax_ssl_context()
    connector = aiohttp.TCPConnector(ssl=ssl_ctx)
    session = aiohttp.ClientSession(connector=connector)
    try:
        # 上游 RFB 一般用 binary 子协议；少数实现不带，留个空 fallback。
        upstream_protocols = ("binary",) if accept_proto == "binary" else ()
        log.info("vnc upstream connecting: protocols=%s", upstream_protocols)
        try:
            # 关键：限定握手 timeout，避免上游阻塞导致浏览器永久卡 connecting
            upstream = await asyncio.wait_for(
                session.ws_connect(
                    upstream_url,
                    protocols=upstream_protocols,
                    heartbeat=60.0,
                    max_msg_size=0,
                    receive_timeout=None,
                    autoclose=True,
                    autoping=True,
                    compress=0,
                ),
                timeout=8.0,
            )
        except asyncio.TimeoutError:
            log.warning("vnc upstream handshake timeout: %s", upstream_url)
            if ws.client_state == WebSocketState.CONNECTED:
                await ws.close(code=1011)
            return
        except aiohttp.WSServerHandshakeError as e:
            log.warning("vnc upstream handshake with protocols=%s failed: %s; retrying without protocols", upstream_protocols, e)
            try:
                upstream = await asyncio.wait_for(
                    session.ws_connect(
                        upstream_url,
                        heartbeat=60.0,
                        max_msg_size=0,
                        receive_timeout=None,
                        autoclose=True,
                        autoping=True,
                        compress=0,
                    ),
                    timeout=8.0,
                )
            except (asyncio.TimeoutError, aiohttp.WSServerHandshakeError) as e2:
                log.warning("vnc upstream handshake retry failed: %s", e2)
                if ws.client_state == WebSocketState.CONNECTED:
                    await ws.close(code=1011)
                return
        log.info("vnc upstream connected: subprotocol=%s response_headers=%s",
                 getattr(upstream, "protocol", None),
                 dict(getattr(upstream, "_response", None).headers) if getattr(upstream, "_response", None) else None)

        try:
            async def browser_to_upstream():
                send_bytes = upstream.send_bytes
                send_str = upstream.send_str
                try:
                    while True:
                        msg = await ws.receive()
                        mtype = msg.get("type")
                        if mtype == "websocket.disconnect":
                            break
                        b = msg.get("bytes")
                        if b is not None:
                            await send_bytes(b)
                            continue
                        t = msg.get("text")
                        if t is not None:
                            await send_str(t)
                except WebSocketDisconnect:
                    pass
                except Exception as e:
                    log.debug("vnc browser->upstream end: %s", e)
                finally:
                    try:
                        await upstream.close()
                    except Exception:
                        pass

            async def upstream_to_browser():
                send_bytes = ws.send_bytes
                send_text = ws.send_text
                BINARY = aiohttp.WSMsgType.BINARY
                TEXT = aiohttp.WSMsgType.TEXT
                CLOSED = aiohttp.WSMsgType.CLOSED
                ERROR = aiohttp.WSMsgType.ERROR
                first_seen = False
                try:
                    async for msg in upstream:
                        t = msg.type
                        if not first_seen:
                            first_seen = True
                            preview = msg.data[:32] if isinstance(msg.data, (bytes, bytearray)) else str(msg.data)[:64]
                            log.info("vnc upstream first frame: type=%s size=%s preview=%r",
                                     t.name, len(msg.data) if hasattr(msg.data, "__len__") else "?", preview)
                        if t is BINARY:
                            if ws.client_state == WebSocketState.CONNECTED:
                                await send_bytes(msg.data)
                        elif t is TEXT:
                            if ws.client_state == WebSocketState.CONNECTED:
                                await send_text(msg.data)
                        elif t is CLOSED or t is ERROR:
                            log.info("vnc upstream sent close: code=%s extra=%s", upstream.close_code, getattr(msg, "extra", None))
                            break
                except Exception as e:
                    log.debug("vnc upstream->browser end: %s", e)
                finally:
                    if ws.client_state == WebSocketState.CONNECTED:
                        try:
                            await ws.close()
                        except Exception:
                            pass

            await asyncio.gather(
                browser_to_upstream(),
                upstream_to_browser(),
            )
        finally:
            try:
                await upstream.close()
            except Exception:
                pass
    except aiohttp.ClientError as e:
        log.warning("vnc upstream connect failed: %s", e)
        if ws.client_state == WebSocketState.CONNECTED:
            await ws.close(code=1011)
    finally:
        await session.close()
        log.info("vnc bridge closed: instance=%s", ticket_data.get("instance_id"))
