"""拉模式 Host Agent 的注册、认证与心跳 API。"""
from __future__ import annotations

import asyncio
import hmac
import secrets
import time
import uuid
from typing import Dict, List, Optional

from fastapi import APIRouter, Header, HTTPException, Response, status
from sqlalchemy import select

from ..config import get_settings
from ..db import session_scope
from ..enums import HostStatus
from ..models import TestHost, iso_z, utcnow
from ..schemas import (
    AgentHeartbeatBody,
    AgentLeaseBody,
    AgentRegisterBody,
    AgentStepAttemptBody,
    AgentStepCompleteBody,
    AgentStepEventsBody,
)
from ..services.agent_status import refresh_agent_host
from ..services.step_leases import (
    StepLeaseConflict,
    acknowledge_step,
    complete_step,
    record_step_events,
    renew_step,
    try_lease_next_step,
)
from .deps import agent_token_digest, require_agent

router = APIRouter(prefix="/api/v1/agent", tags=["host-agent"])


def _idempotency_key(value: str) -> str:
    if len(value) > 60:
        raise HTTPException(status_code=422, detail="Idempotency-Key 过长")
    try:
        uuid.UUID(value)
    except (ValueError, AttributeError) as exc:
        raise HTTPException(
            status_code=422, detail="Idempotency-Key 必须是 UUID"
        ) from exc
    return value


def _lease_conflict(exc: StepLeaseConflict) -> HTTPException:
    return HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc))


def _check_enrollment_token(provided: Optional[str]) -> None:
    settings = get_settings()
    required = settings.agent_enrollment_token
    if settings.env == "production" and not required:
        raise HTTPException(
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
            detail="生产环境尚未配置 Agent 注册口令",
        )
    if required and not hmac.compare_digest(provided or "", required):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Agent 注册口令无效",
        )


async def _resolve_host(body: AgentRegisterBody) -> Dict[str, object]:
    async with session_scope() as session:
        host: Optional[TestHost] = None
        if body.host_id:
            host = await session.get(TestHost, body.host_id)
            if host is None:
                raise HTTPException(status_code=404, detail="Test Host 不存在")
        elif body.host_name:
            host = await session.scalar(
                select(TestHost).where(TestHost.name == body.host_name)
            )
        else:
            host = await session.scalar(select(TestHost).order_by(TestHost.created_at))

        if host is None:
            host = TestHost(name=body.host_name or "agent-host-" + uuid.uuid4().hex[:8])
            session.add(host)
            await session.flush()

        raw_token = secrets.token_urlsafe(32)
        host.agent_id = host.agent_id or "agent_" + uuid.uuid4().hex[:20]
        host.agent_token = agent_token_digest(raw_token)
        host.agent_version = body.agent_version
        host.os_family = body.os_family
        if body.os_version is not None:
            host.os_version = body.os_version
        if body.kernel is not None:
            host.kernel = body.kernel
        if body.cpu is not None:
            host.cpu = body.cpu
        if body.motherboard is not None:
            host.motherboard = body.motherboard
        if body.bios_version is not None:
            host.bios_version = body.bios_version
        if body.ip_address is not None:
            host.ip_address = body.ip_address
        if body.toolchain:
            host.toolchain = body.toolchain
        if body.labels:
            host.labels = {**(host.labels or {}), **body.labels}
        host.status = HostStatus.ONLINE.value
        host.last_heartbeat_at = utcnow()
        payload = {
            "agent_id": host.agent_id,
            "host_id": host.id,
            "token": raw_token,
            "heartbeat_interval_s": get_settings().timing.heartbeat_interval_s,
            "server_time": iso_z(host.last_heartbeat_at),
        }
    return payload


@router.post("/register", status_code=status.HTTP_201_CREATED)
async def register_agent(
    body: AgentRegisterBody,
    enrollment_token: Optional[str] = Header(
        default=None, alias="X-FlashOps-Enrollment-Token"
    ),
) -> Dict[str, object]:
    _check_enrollment_token(enrollment_token)
    return await _resolve_host(body)


@router.post("/{agent_id}/heartbeat")
async def heartbeat(
    agent_id: str,
    body: AgentHeartbeatBody,
    authorization: Optional[str] = Header(default=None),
) -> Dict[str, object]:
    async with session_scope() as session:
        host = await require_agent(session, agent_id, authorization)
        host.status = (
            HostStatus.ONLINE.value if body.healthy else HostStatus.DEGRADED.value
        )
        host.last_heartbeat_at = utcnow()
        if body.agent_version:
            host.agent_version = body.agent_version
        if body.ip_address:
            host.ip_address = body.ip_address
        if body.toolchain:
            host.toolchain = body.toolchain
        if body.device_probe:
            host.labels = {
                **(host.labels or {}),
                "agent_probe": body.device_probe,
            }
        return {
            "ok": True,
            "server_time": iso_z(host.last_heartbeat_at),
            "heartbeat_interval_s": get_settings().timing.heartbeat_interval_s,
            "commands": [],
        }


@router.post("/{agent_id}/lease")
async def lease_step(
    agent_id: str,
    body: AgentLeaseBody,
    authorization: Optional[str] = Header(default=None),
) -> object:
    deadline = time.monotonic() + min(
        body.wait_timeout_s, get_settings().timing.lease_poll_timeout_s
    )
    while True:
        async with session_scope() as session:
            host = await require_agent(session, agent_id, authorization)
            lease = await try_lease_next_step(session, host)
            if lease is not None:
                return lease
        remaining = deadline - time.monotonic()
        if remaining <= 0:
            return Response(status_code=status.HTTP_204_NO_CONTENT)
        await asyncio.sleep(min(get_settings().engine_tick_s, remaining))


@router.post("/{agent_id}/steps/{step_id}/ack")
async def ack_step(
    agent_id: str,
    step_id: str,
    body: AgentStepAttemptBody,
    authorization: Optional[str] = Header(default=None),
    idempotency_key: str = Header(alias="Idempotency-Key"),
) -> Dict[str, object]:
    key = _idempotency_key(idempotency_key)
    try:
        async with session_scope() as session:
            host = await require_agent(session, agent_id, authorization)
            return await acknowledge_step(session, host, step_id, body.attempt, key)
    except StepLeaseConflict as exc:
        raise _lease_conflict(exc) from exc


@router.post("/{agent_id}/steps/{step_id}/renew")
async def renew_step_lease(
    agent_id: str,
    step_id: str,
    body: AgentStepAttemptBody,
    authorization: Optional[str] = Header(default=None),
    idempotency_key: str = Header(alias="Idempotency-Key"),
) -> Dict[str, object]:
    key = _idempotency_key(idempotency_key)
    try:
        async with session_scope() as session:
            host = await require_agent(session, agent_id, authorization)
            return await renew_step(session, host, step_id, body.attempt, key)
    except StepLeaseConflict as exc:
        raise _lease_conflict(exc) from exc


@router.post("/{agent_id}/steps/{step_id}/events")
async def step_events(
    agent_id: str,
    step_id: str,
    body: AgentStepEventsBody,
    authorization: Optional[str] = Header(default=None),
    idempotency_key: str = Header(alias="Idempotency-Key"),
) -> Dict[str, object]:
    key = _idempotency_key(idempotency_key)
    try:
        async with session_scope() as session:
            host = await require_agent(session, agent_id, authorization)
            return await record_step_events(
                session, host, step_id, body.attempt, body.events, key
            )
    except StepLeaseConflict as exc:
        raise _lease_conflict(exc) from exc


@router.post("/{agent_id}/steps/{step_id}/complete")
async def finish_step(
    agent_id: str,
    step_id: str,
    body: AgentStepCompleteBody,
    authorization: Optional[str] = Header(default=None),
    idempotency_key: str = Header(alias="Idempotency-Key"),
) -> Dict[str, object]:
    key = _idempotency_key(idempotency_key)
    try:
        async with session_scope() as session:
            host = await require_agent(session, agent_id, authorization)
            return await complete_step(session, host, step_id, body, key)
    except StepLeaseConflict as exc:
        raise _lease_conflict(exc) from exc


@router.get("/hosts")
async def list_agent_hosts() -> List[Dict[str, object]]:
    async with session_scope() as session:
        hosts = (
            await session.scalars(select(TestHost).order_by(TestHost.name))
        ).all()
        for host in hosts:
            refresh_agent_host(host)
        return [
            {
                "host_id": host.id,
                "host_name": host.name,
                "agent_id": host.agent_id,
                "agent_version": host.agent_version,
                "status": host.status,
                "last_heartbeat_at": iso_z(host.last_heartbeat_at),
                "ip_address": host.ip_address,
                "toolchain": host.toolchain,
            }
            for host in hosts
        ]
