"""Low-overhead performance traces for diagnosing slow Gateway requests."""

from __future__ import annotations

import asyncio
import faulthandler
import logging
import os
import signal
import sys
import time
from typing import Any

logger = logging.getLogger(__name__)

_VERBOSE_PATHS = {
    "/api/v1/auth/options",
    "/api/agents",
    "/api/llmwiki/knowledge-bases",
}
_STACK_DUMP_REGISTERED = False


def slow_request_threshold_ms() -> float:
    """Return the warning threshold, defaulting to half a second."""

    try:
        return max(0.0, min(float(os.getenv("DEERFLOW_SLOW_REQUEST_LOG_MS", "500")), 60_000.0))
    except (TypeError, ValueError):
        return 500.0


def verbose_request_logging() -> bool:
    return os.getenv("DEERFLOW_PERF_DIAGNOSTICS", "").strip().lower() in {"1", "true", "yes", "on"}


def is_verbose_path(path: str) -> bool:
    return path in _VERBOSE_PATHS or (
        path.startswith("/api/llmwiki/agents/")
        and path.endswith("/knowledge-bases")
    )


def should_log_request(
    *,
    elapsed_ms: float,
    headers_ms: float | None,
    first_body_ms: float | None,
    is_stream: bool,
) -> bool:
    threshold = slow_request_threshold_ms()
    if is_stream:
        return bool(
            (headers_ms is not None and headers_ms >= threshold)
            or (first_body_ms is not None and first_body_ms >= threshold)
        )
    return elapsed_ms >= threshold


def trace_request(request: Any, endpoint: str) -> dict[str, Any]:
    """Start an endpoint trace attached to the ASGI request state."""

    trace: dict[str, Any] = {
        "endpoint": endpoint,
        "started": time.perf_counter(),
        "stages": {},
        "details": {},
    }
    try:
        request.state.perf_trace = trace
    except AttributeError:
        # Direct router calls in isolated tests may use a minimal request stub.
        pass
    return trace


def trace_stage(trace: dict[str, Any], name: str, started: float) -> float:
    """Record one endpoint stage and return its completion timestamp."""

    ended = time.perf_counter()
    trace["stages"][name] = round((ended - started) * 1000, 2)
    return ended


def finish_trace(trace: dict[str, Any]) -> None:
    trace["endpoint_ms"] = round((time.perf_counter() - trace["started"]) * 1000, 2)


def register_stack_dump_signal() -> None:
    """Allow SIGUSR1 to dump all Python thread stacks to container stderr."""

    global _STACK_DUMP_REGISTERED
    if _STACK_DUMP_REGISTERED:
        return
    if os.getenv("DEERFLOW_ENABLE_STACK_DUMP", "1").strip().lower() in {"0", "false", "no", "off"}:
        logger.info("PERF Python stack dump signal disabled by configuration")
        return
    stack_signal = getattr(signal, "SIGUSR1", None)
    register = getattr(faulthandler, "register", None)
    if stack_signal is None or register is None:
        logger.info("PERF Python stack dump signal is unavailable on this platform")
        return
    try:
        register(stack_signal, file=sys.stderr, all_threads=True, chain=False)
    except (OSError, RuntimeError, ValueError):
        logger.exception("PERF unable to register Python stack dump signal")
        return
    _STACK_DUMP_REGISTERED = True
    logger.info("PERF Python stack dump ready signal=SIGUSR1 all_threads=true")


async def event_loop_watchdog(app: Any) -> None:
    """Log substantial event-loop stalls without requiring the diagnostics UI."""

    loop = asyncio.get_running_loop()
    last_warning = 0.0
    while True:
        expected = loop.time() + 1.0
        await asyncio.sleep(1.0)
        lag_ms = round(max(0.0, loop.time() - expected) * 1000, 2)
        app.state.perf_event_loop_lag_ms = lag_ms
        now = time.monotonic()
        if lag_ms >= 200 and now - last_warning >= 5:
            logger.warning("PERF event_loop_stall lag_ms=%.2f threshold_ms=200", lag_ms)
            last_warning = now