cfc/cfc-langgraph/app/middleware.py

"""请求日志中间件 — 记录最近 MAX_ENTRIES 条请求到内存

实现为纯 ASGI 中间件（而非 BaseHTTPMiddleware），原因：
- BaseHTTPMiddleware 中先 `await request.body()` 再 `call_next(request)` 会消费请求体，
  导致下游 handler 再次 receive 时拿不到 body，POST 请求死锁/挂起；
- 单 worker 部署下，一个请求挂起会卡死整个 worker，所有请求（含 /api/v1/logs）超时。

纯 ASGI 中间件通过包装 receive/send 通道缓存请求体与响应状态，既记录日志又不破坏下游请求。
"""
import time
import uuid
from collections import deque
import logging

logger = logging.getLogger(__name__)

MAX_ENTRIES = 500
_request_log: deque = deque(maxlen=MAX_ENTRIES)


def get_request_log() -> list:
    """返回请求日志副本（最近在前）"""
    return list(_request_log)[::-1]


def clear_request_log() -> int:
    """清空日志，返回清空条数"""
    n = len(_request_log)
    _request_log.clear()
    return n


def _decode_body(raw: bytes) -> str:
    """将请求体字节解码为可读文本（截断超长内容）"""
    if not raw:
        return ""
    try:
        text = raw.decode("utf-8")
        if len(text) > 2000:
            text = text[:2000] + "...(truncated)"
        return text
    except Exception:
        return f"<binary {len(raw)} bytes>"


class RequestLogMiddleware:
    """纯 ASGI 请求日志中间件。

    通过包装 receive 通道缓存请求体（不消费、不阻塞下游），
    通过包装 send 通道捕获响应状态码与响应体，避免破坏原请求。
    """

    def __init__(self, app):
        self.app = app

    async def __call__(self, scope, receive, send):
        if scope["type"] != "http":
            await self.app(scope, receive, send)
            return

        start_time = time.perf_counter()
        method = scope.get("method", "GET")
        path = scope.get("path", "")
        client = scope.get("client")
        client_ip = client[0] if client else "-"

        # 读取或生成 X-Request-ID（Java AiGateway 在每次调用时都会注入 UUID）
        request_id = None
        for name, value in scope.get("headers", []):
            if name == b"x-request-id":
                request_id = value.decode()
                break
        request_id = request_id or str(uuid.uuid4())

        # 带 trace_id 的 adapter：JsonFormatter 会把它写为顶层字段
        req_logger = logging.LoggerAdapter(logger, {"trace_id": request_id})

        # 缓存请求体（仅对带 body 的方法），不影响下游 receive
        body_bytes = b""
        body_cached = False

        async def receive_wrapper():
            nonlocal body_bytes, body_cached
            message = await receive()
            if message["type"] == "http.request":
                body_bytes += message.get("body", b"")
                if not message.get("more_body", False):
                    body_cached = True
            return message

        # 捕获响应状态码与响应体
        status_code = None
        response_chunks = []
        sent_response_start = False

        async def send_wrapper(message):
            nonlocal status_code, response_chunks, sent_response_start
            if message["type"] == "http.response.start":
                status_code = message.get("status", 500)
                sent_response_start = True
            elif message["type"] == "http.response.body":
                body = message.get("body", b"")
                if body:
                    response_chunks.append(body)
                    total = sum(len(c) for c in response_chunks)
                    if total > 1000:
                        response_chunks[:] = [b"".join(response_chunks)[:1000]]
            return await send(message)

        try:
            await self.app(scope, receive_wrapper, send_wrapper)
        except Exception as exc:
            req_logger.error("REQUEST EXCEPTION: %s %s %.2fs error=%r",
                          method, path, time.perf_counter() - start_time, exc)
            if not sent_response_start:
                from starlette.responses import JSONResponse
                resp = JSONResponse(status_code=500, content={"detail": str(exc)})
                await resp(scope, receive, send)
            status_code = status_code or 500

        elapsed = time.perf_counter() - start_time
        final_status = status_code if status_code is not None else 500

        entry = {
            "ts": time.strftime("%H:%M:%S"),
            "method": method,
            "path": path,
            "status": final_status,
            "duration_ms": round(elapsed * 1000),
            "client_ip": client_ip,
            "trace_id": request_id,
            "body": _decode_body(body_bytes) if body_cached else None,
            "response": _decode_body(b"".join(response_chunks)[:1000]) if response_chunks else "",
        }
        _request_log.appendleft(entry)

        if elapsed > 5:
            req_logger.warning("SLOW_REQUEST: %s %s %.2fs", method, path, elapsed)
        else:
            req_logger.debug("REQUEST: %s %s %.2fs", method, path, elapsed)