From 12ec5af0de626393f18ec4590a93ccac4076e6a4 Mon Sep 17 00:00:00 2001 From: Kris <2893855659@qq.com> Date: Tue, 7 Jul 2026 16:25:47 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=96=B0=E5=A2=9E=E5=85=A8=E9=93=BE?= =?UTF-8?q?=E8=B7=AF=E8=BF=BD=E8=B8=AA=E3=80=81=E4=BC=98=E9=9B=85=E5=85=B3?= =?UTF-8?q?=E5=81=9C=E6=94=AF=E6=8C=81=E4=B8=8E=E7=BB=9F=E4=B8=80=E5=BC=82?= =?UTF-8?q?=E5=B8=B8=E5=A4=84=E7=90=86=EF=BC=8C=E4=BC=98=E5=8C=96=E6=B8=A0?= =?UTF-8?q?=E9=81=93=E6=9F=A5=E8=AF=A2=E6=8E=A5=E5=8F=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. 新增TraceId中间件实现全链路请求追踪 2. 新增在途请求计数器与中间件支持优雅关停 3. 重构全局异常处理器，统一三模块异常响应格式 4. 优化渠道查询接口，新增多维度过滤与模糊匹配 5. 重构channels模块数据库会话管理，避免事务冲突 6. 新增目录搜索建议与目录导出功能 --- backend/server/main.py | 82 ++++----- backend/server/routers/channels/__init__.py | 26 +-- .../routers/channels/directory_router.py | 161 ++++++++++++++++-- .../server/routers/channels/message_router.py | 37 +++- .../server/routers/channels/outbox_router.py | 41 ++++- backend/server/utils/error_handlers.py | 92 ++++++++++ backend/server/utils/error_response.py | 122 +++++++++++++ backend/server/utils/inflight_tracker.py | 78 +++++++++ backend/server/utils/lifespan.py | 4 + backend/server/utils/trace_middleware.py | 37 ++++ 10 files changed, 595 insertions(+), 85 deletions(-) create mode 100644 backend/server/utils/error_handlers.py create mode 100644 backend/server/utils/error_response.py create mode 100644 backend/server/utils/inflight_tracker.py create mode 100644 backend/server/utils/trace_middleware.py diff --git a/backend/server/main.py b/backend/server/main.py index b950974a..85a91cde 100644 --- a/backend/server/main.py +++ b/backend/server/main.py @@ -16,17 +16,26 @@ from collections import defaultdict, deque import uvicorn from fastapi import FastAPI, Request, status +from fastapi.exceptions import RequestValidationError from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse from starlette.middleware.base import BaseHTTPMiddleware from yuxi.external_systems.exceptions import ExternalSystemError from yuxi.scheduler.exceptions import SchedulerError +from yuxi.channels.contract.errors import Error as ChannelError from server.routers import router +from server.utils.error_handlers import ( + unified_error_handler, + unhandled_exception_handler, + request_validation_error_handler, +) from server.utils.lifespan import lifespan from server.utils.auth_middleware import is_public_path from server.utils.common_utils import setup_logging from server.utils.access_log_middleware import AccessLogMiddleware +from server.utils.trace_middleware import TraceIdMiddleware +from server.utils.inflight_tracker import InflightTrackingMiddleware # 设置日志配置 setup_logging() @@ -44,47 +53,12 @@ app = FastAPI(lifespan=lifespan) app.include_router(router, prefix="/api") -@app.exception_handler(ExternalSystemError) -async def external_system_error_handler(_request: Request, exc: ExternalSystemError) -> JSONResponse: - """外部系统限界上下文异常统一映射为 HTTP 响应。 - - - ``status_code`` 来自异常类。 - - ``details`` 透传到响应体，供前端结构化处理。 - - 429/503 且 ``details`` 含 ``retry_after`` 时，自动设置 ``Retry-After`` 响应头。 - """ - headers: dict[str, str] = {} - if exc.status_code in (429, 503): - retry_after = exc.details.get("retry_after") - if retry_after is not None: - headers["Retry-After"] = str(retry_after) - return JSONResponse( - status_code=exc.status_code, - content={"detail": exc.message, "details": exc.details}, - headers=headers, - ) - - -@app.exception_handler(SchedulerError) -async def scheduler_error_handler(_request: Request, exc: SchedulerError) -> JSONResponse: - """定时任务调度限界上下文异常统一映射为 HTTP 响应。 - - - ``status_code`` 来自异常类（400 / 404 / 409 / 413 / 500）。 - - ``details`` 透传到响应体，供前端结构化处理。 - """ - return JSONResponse( - status_code=exc.status_code, - content={"detail": exc.message, "details": exc.details}, - ) - - -# CORS 设置 -app.add_middleware( - CORSMiddleware, - allow_origins=["*"], - allow_credentials=True, - allow_methods=["*"], - allow_headers=["*"], -) +# 三模块异常统一注册到同一个 handler +app.add_exception_handler(ChannelError, unified_error_handler) +app.add_exception_handler(ExternalSystemError, unified_error_handler) +app.add_exception_handler(SchedulerError, unified_error_handler) +app.add_exception_handler(RequestValidationError, request_validation_error_handler) +app.add_exception_handler(Exception, unhandled_exception_handler) def _extract_client_ip(request: Request) -> str: @@ -115,7 +89,15 @@ class LoginRateLimitMiddleware(BaseHTTPMiddleware): retry_after = int(max(1, RATE_LIMIT_WINDOW_SECONDS - (now - attempt_history[0]))) return JSONResponse( status_code=status.HTTP_429_TOO_MANY_REQUESTS, - content={"detail": "登录尝试过于频繁，请稍后再试"}, + content={ + "success": False, + "error": { + "code": "RATE_LIMIT", + "message": "登录尝试过于频繁，请稍后再试", + "trace_id": None, + "details": {"retry_after": retry_after}, + }, + }, headers={"Retry-After": str(retry_after)}, ) @@ -165,12 +147,20 @@ class AuthMiddleware(BaseHTTPMiddleware): return await call_next(request) -# 添加访问日志中间件（记录请求处理时间） -app.add_middleware(AccessLogMiddleware) - -# 添加鉴权中间件 +# 中间件注册顺序为 LIFO（后注册先执行）： +# CORSMiddleware 最内层，InflightTrackingMiddleware 最外层最先执行 +app.add_middleware( + CORSMiddleware, + allow_origins=["*"], + allow_credentials=True, + allow_methods=["*"], + allow_headers=["*"], +) app.add_middleware(LoginRateLimitMiddleware) app.add_middleware(AuthMiddleware) +app.add_middleware(AccessLogMiddleware) +app.add_middleware(TraceIdMiddleware) +app.add_middleware(InflightTrackingMiddleware) if __name__ == "__main__": # uvicorn.run(app, host="0.0.0.0", port=5050, threads=10, workers=10, reload=True) diff --git a/backend/server/routers/channels/__init__.py b/backend/server/routers/channels/__init__.py index 10da9ce1..61ac59ef 100644 --- a/backend/server/routers/channels/__init__.py +++ b/backend/server/routers/channels/__init__.py @@ -16,7 +16,6 @@ from enum import Enum from typing import Any, Literal from fastapi import APIRouter, Depends, Request -from sqlalchemy.ext.asyncio import AsyncSession from yuxi.channels.contract.dtos.common import Operator, OperatorRole from yuxi.channels.contract.dtos.control import ControlResult from yuxi.channels.contract.errors import ( @@ -51,13 +50,12 @@ from yuxi.channels.infrastructure.channel_use_cases import ( ChannelUseCases, create_channel_use_cases, ) +from yuxi.storage.postgres.manager import pg_manager from yuxi.storage.postgres.models_business import User from yuxi.utils.crypto import SENSITIVE_HTTP_HEADERS from yuxi.utils.datetime_utils import coerce_any_to_utc_datetime from yuxi.utils.trace_context import get_trace_id -from server.utils.auth_middleware import get_db - # 时间粒度查询参数类型（analytics / dashboard / pairing / content_review / outbox 共享） Granularity = Literal["hour", "day", "week"] @@ -86,23 +84,29 @@ async def get_channel_di_container(request: Request) -> DependencyInjectionConta async def get_channel_use_cases( - db: AsyncSession = Depends(get_db), di_container: DependencyInjectionContainer = Depends(get_channel_di_container), ) -> ChannelUseCases: - """装配 channels 控制面 UseCase 集合。 + """装配 channels 控制面 UseCase 集合（请求级独立 session）。 - 将请求级 DB 会话与 DI 容器传入 ``create_channel_use_cases``，构造本次 - 请求可用的 ``ChannelUseCases``。UseCase 内部端口方法在控制面管道事务中 - 执行。 + channels 模块 **不复用** 鉴权依赖（``get_db``）持有的 ``AsyncSession``， + 而是通过 ``pg_manager.get_async_session_context()`` 开启独立的请求级 + 会话。原因：鉴权阶段执行的 ``SELECT`` 会触发 SQLAlchemy 2.0 autobegin + 语义开启隐式事务，若 channels 复用同一 session，控制面 + ``TransactionPort.begin()`` 会在已存在的事务上再次 ``begin()``， + 抛出 ``InvalidRequestError: A transaction is already begun``。 + + 独立 session 使事务边界由 channels 应用层独占控制，与鉴权会话隔离， + 避免事务状态互相污染。会话生命周期限制在本次请求内，请求结束自动 + commit（无异常）或 rollback（有异常）后关闭。 Args: - db: 请求级异步 DB 会话。 di_container: channels 模块 DI 容器。 - Returns: + Yields: 已装配的 ``ChannelUseCases``，含 ``session_management`` 等端口字段。 """ - return create_channel_use_cases(db, di_container) + async with pg_manager.get_async_session_context() as db: + yield create_channel_use_cases(db, di_container) def build_operator(user: User, request: Request) -> Operator: diff --git a/backend/server/routers/channels/directory_router.py b/backend/server/routers/channels/directory_router.py index 3aaf6274..e011f57f 100644 --- a/backend/server/routers/channels/directory_router.py +++ b/backend/server/routers/channels/directory_router.py @@ -1,38 +1,51 @@ -"""目录查询域 Router（FR-14，DIR-QUERY-01~04 + DIR-01 + DIR-CACHE-CLEAR）。 +"""目录查询域 Router（FR-14，DIR-QUERY-01~04 + DIR-01 + DIR-CACHE-CLEAR + Task 5）。 本 router 实现渠道通讯录目录查询的全部 HTTP 端点，覆盖用户搜索 / 群组搜索 / -用户资料 / 群组详情 / 群组成员 / 缓存清理共 6 个操作。所有端点统一采用模板 A -（控制面端口路由），通过 ``get_channel_use_cases`` 装配 ``ChannelUseCases``， -经 ``directory`` 端口调用类型化方法，由 ``ChannelControlService._executeControl`` -内部走控制面管道（auth → permission → rate_limit → dispatch → audit）。 +用户资料 / 群组详情 / 群组成员 / 缓存清理 / 目录导出共 7 个操作。除导出端点 +返回 ``StreamingResponse`` 文件流外，其余端点统一采用模板 A（控制面端口路由）， +通过 ``get_channel_use_cases`` 装配 ``ChannelUseCases``，经 ``directory`` 端口 +调用类型化方法，由 ``ChannelControlService._executeControl`` 内部走控制面管道 +（auth → permission → rate_limit → dispatch → audit）。导出端点同样经控制面 +管道执行，handler 返回内容字符串，router 包装为文件流以保持审计链路。 鉴权策略：全部端点使用 ``get_admin_user`` 依赖，要求管理员或超级管理员角色。 角色校验由依赖函数完成，router 内不做角色判断（规范 §4）。 -路径设计：静态后缀路径（``/search``）先于动态单段路径（``/{peer_id}`` / -``/{group_id}``）声明，动态单段先于多段子资源路径（``/{group_id}/members``） -声明（规范 §6.5），避免静态路径被动态参数捕获。子 router 不自行设置 prefix， -根前缀 ``/channels`` 由 ``channels_router`` 聚合 router 统一追加。 +路径设计：静态后缀路径（``/search`` / ``/export``）先于动态单段路径 +（``/{peer_id}`` / ``/{group_id}``）声明，动态单段先于多段子资源路径 +（``/{group_id}/members``）声明（规范 §6.5），避免静态路径被动态参数捕获。 +子 router 不自行设置 prefix，根前缀 ``/channels`` 由 ``channels_router`` 聚合 +router 统一追加。 -端点清单（对应《05-目录查询域设计方案 v1.1》§2.3）： - - GET /{channel_type}/{account_id}/directory/users/search DIR-QUERY-01 search_directory_users - - GET /{channel_type}/{account_id}/directory/groups/search DIR-QUERY-02 search_directory_groups - - GET /{channel_type}/{account_id}/directory/users/{peer_id} DIR-QUERY-03 get_directory_user_profile - - GET /{channel_type}/{account_id}/directory/groups/{group_id} DIR-01 get_directory_group_detail - - GET /{channel_type}/{account_id}/directory/groups/{group_id}/members DIR-QUERY-04 get_directory_group_members - - DELETE /{channel_type}/{account_id}/directory/cache DIR-CACHE-CLEAR clear_directory_cache +端点清单（对应《05-目录查询域设计方案 v1.1》§2.3 + Task 5）： + - GET /{channel_type}/{account_id}/directory/users/search + DIR-QUERY-01 search_directory_users + - GET /{channel_type}/{account_id}/directory/groups/search + DIR-QUERY-02 search_directory_groups + - GET /{channel_type}/{account_id}/directory/users/{peer_id} + DIR-QUERY-03 get_directory_user_profile + - GET /{channel_type}/{account_id}/directory/groups/{group_id} + DIR-01 get_directory_group_detail + - GET /{channel_type}/{account_id}/directory/groups/{group_id}/members + DIR-QUERY-04 get_directory_group_members + - DELETE /{channel_type}/{account_id}/directory/cache + DIR-CACHE-CLEAR clear_directory_cache + - POST /{channel_type}/{account_id}/directory/export + TASK-5 export_directory """ from __future__ import annotations -from typing import Any +from typing import Any, Literal from fastapi import APIRouter, Body, Depends, Query, Request +from fastapi.responses import StreamingResponse from pydantic import BaseModel, ConfigDict, Field from yuxi.channels.contract.dtos.channel import ChannelType from yuxi.channels.contract.dtos.directory import ( BatchDirectoryProfilesQuery, ClearDirectoryCacheCmd, + DirectoryEntryType, DirectoryQuery, ) from yuxi.storage.postgres.models_business import User @@ -83,6 +96,38 @@ async def search_directory_users( return {"success": True, "data": serialize_control_data(result.data)} +@directory_router.get( + "/{channel_type}/{account_id}/directory/suggestions", + response_model=dict, +) +async def search_directory_suggestions( + channel_type: ChannelType, + account_id: str, + request: Request, + keyword: str = Query(..., min_length=1, description="搜索关键词"), + limit: int = Query(default=10, ge=1, le=20, description="建议数量"), + use_cases=Depends(get_channel_use_cases), + current_user: User = Depends(get_admin_user), +) -> dict[str, Any]: + """搜索渠道目录建议（FR-14）。 + + 根据关键词搜索渠道侧目录建议，用于输入联想。5min TTL 缓存。对应控制 + 面操作 ``directory/search_suggestions``（由 + ``ChannelControlService.searchDirectorySuggestions`` 内部构造 + ``ControlCmd`` 并委托 ``_executeControl`` 执行控制面管道）。 + """ + operator = build_operator(current_user, request) + query = DirectoryQuery(keyword=keyword, limit=limit) + result = await use_cases.directory.searchDirectorySuggestions( + channel_type=channel_type, + account_id=account_id, + query=query, + operator=operator, + ) + raiseOnControlFailure(result) + return {"success": True, "data": serialize_control_data(result.data)} + + @directory_router.get( "/{channel_type}/{account_id}/directory/groups/search", response_model=dict, @@ -157,6 +202,35 @@ class BatchProfilesRequest(BaseModel): min_length=1, description="待查询的对端 ID 列表（可能是用户 ID 或群组 ID）", ) + types: list[Literal["user", "group"]] | None = Field( + default=None, + description="按条目类型过滤（user / group），不传入则查询全部类型", + ) + + +class ExportDirectoryRequest(BaseModel): + """目录导出请求（Task 5）。""" + + model_config = ConfigDict(frozen=True) + + scope: Literal["users", "groups"] = Field( + default="users", + description="导出范围（users 或 groups）", + ) + keyword: str | None = Field( + default=None, + description="搜索关键词", + ) + format: Literal["csv", "json"] = Field( + default="csv", + description="导出格式（csv 或 json）", + ) + limit: int = Field( + default=1000, + ge=1, + le=1000, + description="导出数量限制（1~1000）", + ) @directory_router.post( @@ -182,7 +256,11 @@ async def batch_directory_profiles( ``ControlCmd`` 并委托 ``_executeControl`` 执行控制面管道）。 """ operator = build_operator(current_user, request) - query = BatchDirectoryProfilesQuery(peer_ids=tuple(payload.peer_ids)) + types = tuple(DirectoryEntryType(t) for t in payload.types) if payload.types is not None else None + query = BatchDirectoryProfilesQuery( + peer_ids=tuple(payload.peer_ids), + types=types, + ) result = await use_cases.directory.batchGetDirectoryProfiles( channel_type=channel_type, account_id=account_id, @@ -193,6 +271,53 @@ async def batch_directory_profiles( return {"success": True, "data": serialize_control_data(result.data)} +@directory_router.post( + "/{channel_type}/{account_id}/directory/export", +) +async def export_directory( + channel_type: ChannelType, + account_id: str, + request: Request, + payload: ExportDirectoryRequest = Body(...), + use_cases=Depends(get_channel_use_cases), + current_user: User = Depends(get_admin_user), +) -> StreamingResponse: + """导出目录数据（FR-14，Task 5）。 + + 搜索渠道侧用户或群组目录，将结果格式化为 CSV 或 JSON 文件流返回。 + 经控制面管道执行 ``directory/export`` 操作，保持审计链路；handler 返回 + ``{"content": str, "format": str, "filename": str}``，router 包装为 + ``StreamingResponse`` 触发浏览器下载。 + """ + operator = build_operator(current_user, request) + result = await use_cases.directory.exportDirectory( + channel_type=channel_type, + account_id=account_id, + scope=payload.scope, + keyword=payload.keyword, + format=payload.format, + limit=payload.limit, + operator=operator, + ) + raiseOnControlFailure(result) + data = serialize_control_data(result.data) + content: str = data["content"] + fmt: str = data["format"] + filename: str = data["filename"] + media_type = "text/csv; charset=utf-8" if fmt == "csv" else "application/json; charset=utf-8" + + async def _stream(): + yield content.encode("utf-8") + + return StreamingResponse( + _stream(), + media_type=media_type, + headers={ + "Content-Disposition": f"attachment; filename={filename}", + }, + ) + + @directory_router.get( "/{channel_type}/{account_id}/directory/groups/{group_id}", response_model=dict, diff --git a/backend/server/routers/channels/message_router.py b/backend/server/routers/channels/message_router.py index fa4d4bb7..d8814fd6 100644 --- a/backend/server/routers/channels/message_router.py +++ b/backend/server/routers/channels/message_router.py @@ -145,10 +145,19 @@ async def list_admin_messages( channel_type: ChannelType | None = Query(default=None, description="按渠道过滤"), start_time: str | None = Query(default=None, description="起始时间 ISO 8601"), end_time: str | None = Query(default=None, description="截止时间 ISO 8601"), - channel_status: Literal["sent", "delivered", "read", "recalled", "edited"] | None = Query( + channel_status: str | None = Query( default=None, - description="按渠道侧状态过滤（sent/delivered/read/recalled/edited）", + description="按渠道侧状态过滤，支持逗号分隔多值（如 sent,delivered）", ), + role: str | None = Query( + default=None, + description="按角色过滤，支持逗号分隔多值（如 admin,assistant）", + ), + account_id: str | None = Query(default=None, description="按渠道账户业务 ID 模糊匹配"), + peer_id: str | None = Query(default=None, description="按对端 ID 模糊匹配"), + message_id: str | None = Query(default=None, description="按消息业务 ID 模糊匹配"), + channel_msg_id: str | None = Query(default=None, description="按渠道消息 ID 模糊匹配"), + channel_session_id: str | None = Query(default=None, description="按渠道会话 ID 模糊匹配"), limit: int = Query(default=50, ge=1, le=200, description="分页大小"), offset: int = Query(default=0, ge=0, description="分页偏移"), use_cases=Depends(get_channel_use_cases), @@ -169,6 +178,12 @@ async def list_admin_messages( start_time=start_time_dt, end_time=end_time_dt, channel_status=channel_status, + role=role, + account_id=account_id, + peer_id=peer_id, + message_id=message_id, + channel_msg_id=channel_msg_id, + channel_session_id=channel_session_id, limit=limit, offset=offset, operator=operator, @@ -182,8 +197,18 @@ async def search_messages( request: Request, keyword: str = Query(..., min_length=2, description="搜索关键词"), channel_type: ChannelType | None = Query(default=None, description="按渠道过滤"), + channel_status: str | None = Query( + default=None, + description="按渠道侧状态过滤，支持逗号分隔多值（如 sent,delivered）", + ), + role: str | None = Query( + default=None, + description="按角色过滤，支持逗号分隔多值（如 admin,assistant）", + ), channel_session_id: str | None = Query(default=None, description="按会话 ID 过滤"), peer_id: str | None = Query(default=None, description="按对端 ID 过滤"), + message_id: str | None = Query(default=None, description="按消息业务 ID 模糊匹配"), + channel_msg_id: str | None = Query(default=None, description="按渠道消息 ID 模糊匹配"), start_time: str | None = Query(default=None, description="起始时间（ISO 8601）"), end_time: str | None = Query(default=None, description="截止时间（ISO 8601）"), limit: int = Query(default=20, ge=1, le=100, description="分页大小"), @@ -193,8 +218,8 @@ async def search_messages( ) -> dict[str, Any]: """全文搜索消息（MSG-SEARCH-01）。 - 按关键词搜索消息内容，支持渠道、会话、对端 ID、时间范围过滤。 - 对应控制面操作 ``message/search``。 + 按关键词搜索消息内容，支持渠道、状态、角色、会话、对端 ID、消息 ID、 + 渠道消息 ID、时间范围过滤。对应控制面操作 ``message/search``。 """ operator = build_operator(current_user, request) start_time_dt = parse_datetime("start_time", start_time) @@ -205,8 +230,12 @@ async def search_messages( offset=offset, operator=operator, channel_type=channel_type, + channel_status=channel_status, + role=role, channel_session_id=channel_session_id, peer_id=peer_id, + message_id=message_id, + channel_msg_id=channel_msg_id, start_time=start_time_dt, end_time=end_time_dt, ) diff --git a/backend/server/routers/channels/outbox_router.py b/backend/server/routers/channels/outbox_router.py index 063bdb21..bd5854c0 100644 --- a/backend/server/routers/channels/outbox_router.py +++ b/backend/server/routers/channels/outbox_router.py @@ -82,11 +82,12 @@ class UpdateRetryPolicyRequest(BaseModel): async def list_outbox_messages( request: Request, channel_type: ChannelType | None = Query(default=None, description="按渠道类型过滤"), - channel_account_id: str | None = Query(default=None, description="按渠道账户 ID 过滤"), + channel_account_id: str | None = Query(default=None, description="按渠道账户 ID 过滤，支持模糊搜索"), status: OutboxStatus | None = Query(default=None, description="按状态过滤"), - message_id: str | None = Query(default=None, description="按消息 ID 过滤"), - channel_msg_id: str | None = Query(default=None, description="按渠道侧消息 ID 过滤"), + message_id: str | None = Query(default=None, description="按消息 ID 模糊搜索"), + channel_msg_id: str | None = Query(default=None, description="按渠道侧消息 ID 模糊搜索"), channel_session_id: str | None = Query(default=None, description="按渠道会话 ID 过滤"), + last_error: str | None = Query(default=None, description="按错误关键词筛选"), created_after: str | None = Query(default=None, description="创建时间下界（ISO 8601）"), created_before: str | None = Query(default=None, description="创建时间上界（ISO 8601）"), retry_count_min: int | None = Query( @@ -103,14 +104,15 @@ async def list_outbox_messages( created_before_dt = parse_datetime("created_before", created_before) query_filter = OutboxQueryFilter( channel_type=channel_type, - channel_account_id=channel_account_id, status=status, - message_id=message_id, - channel_msg_id=channel_msg_id, channel_session_id=channel_session_id, created_after=created_after_dt, created_before=created_before_dt, retry_count_min=retry_count_min, + channel_account_id_like=channel_account_id, + message_id_like=message_id, + channel_msg_id_like=channel_msg_id, + last_error_like=last_error, ) result = await use_cases.outbox_management.listOutboxEntries( query_filter, @@ -147,6 +149,7 @@ async def get_outbox_trend( start_time: str = Query(..., description="起始时间（ISO 8601，含）"), end_time: str = Query(..., description="结束时间（ISO 8601，含）"), channel_type: ChannelType | None = Query(default=None, description="按渠道类型过滤"), + channel_account_id: str | None = Query(default=None, description="按渠道账户 ID 过滤"), granularity: Literal["minute", "hour", "day"] = Query(default="hour", description="时间粒度"), metric: Literal["queue_depth", "retry_count", "dead_count"] = Query(default="queue_depth", description="度量指标"), use_cases=Depends(get_channel_use_cases), @@ -164,6 +167,7 @@ async def get_outbox_trend( start_time=start_time_dt, end_time=end_time_dt, channel_type=channel_type, + channel_account_id=channel_account_id, granularity=granularity, metric=metric, ) @@ -175,6 +179,17 @@ async def get_outbox_trend( @outbox_router.get("/dead-letter", response_model=dict) async def list_dead_letter_outbox( request: Request, + channel_type: ChannelType | None = Query(default=None, description="按渠道类型过滤"), + channel_account_id: str | None = Query(default=None, description="按渠道账户 ID 过滤，支持模糊搜索"), + message_id: str | None = Query(default=None, description="按消息 ID 模糊搜索"), + channel_msg_id: str | None = Query(default=None, description="按渠道侧消息 ID 模糊搜索"), + channel_session_id: str | None = Query(default=None, description="按渠道会话 ID 过滤"), + last_error: str | None = Query(default=None, description="按错误关键词筛选"), + created_after: str | None = Query(default=None, description="创建时间下界（ISO 8601）"), + created_before: str | None = Query(default=None, description="创建时间上界（ISO 8601）"), + retry_count_min: int | None = Query( + default=None, ge=0, description="最小重试次数下界（含），用于筛选已发生重试的条目" + ), limit: int = Query(default=100, ge=1, le=200, description="分页大小"), offset: int = Query(default=0, ge=0, description="分页偏移"), use_cases=Depends(get_channel_use_cases), @@ -182,7 +197,21 @@ async def list_dead_letter_outbox( ) -> dict[str, Any]: """查询死信队列（OBX-04）。对应控制面操作 outbox/dead_letter_list。""" operator = build_operator(current_user, request) + created_after_dt = parse_datetime("created_after", created_after) + created_before_dt = parse_datetime("created_before", created_before) + query_filter = OutboxQueryFilter( + channel_type=channel_type, + channel_session_id=channel_session_id, + created_after=created_after_dt, + created_before=created_before_dt, + retry_count_min=retry_count_min, + channel_account_id_like=channel_account_id, + message_id_like=message_id, + channel_msg_id_like=channel_msg_id, + last_error_like=last_error, + ) result = await use_cases.outbox_management.listDeadLetterOutboxEntries( + query_filter, operator=operator, limit=limit, offset=offset, diff --git a/backend/server/utils/error_handlers.py b/backend/server/utils/error_handlers.py new file mode 100644 index 00000000..7e25a418 --- /dev/null +++ b/backend/server/utils/error_handlers.py @@ -0,0 +1,92 @@ +"""全局异常处理器集中实现。 + +3 个 handler： +- unified_error_handler：处理三模块异常（满足 UnifiedError 协议） +- request_validation_error_handler：处理 Pydantic 请求体校验失败 +- unhandled_exception_handler：兜底原生 Exception +""" + +from __future__ import annotations + +import logging + +from fastapi import Request +from fastapi.exceptions import RequestValidationError +from fastapi.responses import JSONResponse +from yuxi.channels.contract.errors import Error as ChannelError +from yuxi.external_systems.exceptions import ExternalSystemError +from yuxi.scheduler.exceptions import SchedulerError + +from server.utils.error_response import build_error_response, build_error_response_from_exc + + +async def unified_error_handler( + _request: Request, + exc: ChannelError | ExternalSystemError | SchedulerError, +) -> JSONResponse: + """三模块异常统一处理器。 + + 三模块异常基类都满足 UnifiedError 协议（status_code/error_code/message/ + details/trace_id），通过 build_error_response_from_exc() 统一构建响应。 + 不再有 isinstance 特判，不再有查表/类属性分支，逻辑完全相同。 + + Args: + _request: FastAPI 请求对象（未使用）。 + exc: 三模块异常之一。 + + Returns: + 统一格式的错误响应。 + """ + return build_error_response_from_exc(exc) + + +async def unhandled_exception_handler(_request: Request, exc: Exception) -> JSONResponse: + """原生异常兜底处理器。 + + 所有未被特定异常处理器捕获的 Exception 在此统一兜底， + 返回结构化 500 响应。trace_id 自动从 ContextVar 获取。 + + Args: + _request: FastAPI 请求对象（未使用）。 + exc: 未被捕获的原生异常。 + + Returns: + 500 内部服务器错误响应。 + """ + logging.getLogger("uvicorn.error").exception("unhandled exception", exc_info=exc) + return build_error_response( + status_code=500, + code="INTERNAL", + message="Internal server error", + ) + + +async def request_validation_error_handler( + _request: Request, + exc: RequestValidationError, +) -> JSONResponse: + """Pydantic 请求体校验失败统一处理器。 + + 将 FastAPI 默认的 422 + ``{"detail": [...]}`` 转换为项目统一格式 + 422 + ``{"success": false, "error": {...}}``，确保响应格式一致。 + + Pydantic v2 的 ``model_validator`` / ``field_validator`` 中抛出的 + ``ValueError`` 子类（含项目的 ``ValidationError``）会被包装为 + ``pydantic.ValidationError``，再由 FastAPI 转为 ``RequestValidationError``， + 本处理器统一捕获并转换为项目格式。状态码保持 422（FastAPI 标准）， + 仅统一响应结构。 + """ + errors = exc.errors() + first = errors[0] if errors else {} + loc = first.get("loc", []) + field = ".".join(str(part) for part in loc) if loc else "" + message = first.get("msg", "request validation failed") + # Pydantic 对 ValueError 子类的包装会加 "Value error, " 前缀，去除以还原原始 message + if message.startswith("Value error, "): + message = message[len("Value error, ") :] + return build_error_response( + status_code=422, + code="VALIDATION_ERROR", + message=message, + details={"field": field, "errors": errors}, + ) diff --git a/backend/server/utils/error_response.py b/backend/server/utils/error_response.py new file mode 100644 index 00000000..1ada369e --- /dev/null +++ b/backend/server/utils/error_response.py @@ -0,0 +1,122 @@ +"""统一错误响应构建器。 + +定义 ``UnifiedError`` 协议（三模块异常基类都实现此协议）， +提供 ``build_error_response()`` 和 ``build_error_response_from_exc()`` +统一构建错误响应，确保所有全局异常处理器输出格式一致。 +""" + +from __future__ import annotations + +from typing import Any, Protocol + +from fastapi.responses import JSONResponse +from yuxi.utils.trace_context import get_trace_id + + +class UnifiedError(Protocol): + """统一异常协议。 + + 三模块异常基类（ChannelError / ExternalSystemError / SchedulerError） + 都实现此协议，允许统一 handler 处理。 + + 字段： + status_code: HTTP 状态码。 + error_code: 稳定错误码（字符串，如 "VALIDATION_ERROR"）。 + message: 人类可读错误信息。 + details: 业务字段字典（如 references, retry_after 等）。 + trace_id: 链路追踪 ID。 + """ + + status_code: int + error_code: str + message: str + details: dict[str, Any] + trace_id: str | None + + +def _extract_retry_after(exc: UnifiedError) -> dict[str, str]: + """统一提取 Retry-After 响应头。 + + 429/503 且 details 含 retry_after 时设置。 + 503 且无 retry_after 时设置默认 60 秒（ChannelDegradedError 等降级场景）。 + channels 的 RateLimitError 已将 retry_after 写入 details（通过 to_dict()）。 + external_systems 的 RateLimitExceededError 已将 retry_after 写入 details。 + + Args: + exc: 满足 UnifiedError 协议的异常对象。 + + Returns: + 含 Retry-After 头的字典（无需要时为空字典）。 + """ + headers: dict[str, str] = {} + if exc.status_code in (429, 503): + retry_after = exc.details.get("retry_after") + if retry_after is not None: + headers["Retry-After"] = str(retry_after) + elif exc.status_code == 503: + # 503 降级场景无明确 retry_after 时设置默认 60 秒 + headers["Retry-After"] = "60" + return headers + + +def build_error_response( + status_code: int, + code: str, + message: str, + details: dict[str, Any] | None = None, + trace_id: str | None = None, + headers: dict[str, str] | None = None, +) -> JSONResponse: + """构建统一错误响应。 + + 所有全局异常处理器调用此函数，确保响应格式一致。 + trace_id 优先使用传入值，其次从 ContextVar 获取。 + + Args: + status_code: HTTP 状态码。 + code: 稳定错误码。 + message: 人类可读错误信息。 + details: 业务字段字典。 + trace_id: 链路追踪 ID（为 None 时从 ContextVar 获取）。 + headers: 额外响应头（如 Retry-After）。 + + Returns: + 统一格式的 JSONResponse。 + """ + resolved_trace_id = trace_id or get_trace_id() + return JSONResponse( + status_code=status_code, + content={ + "success": False, + "error": { + "code": code, + "message": message, + "trace_id": resolved_trace_id, + "details": details or {}, + }, + }, + headers=headers or {}, + ) + + +def build_error_response_from_exc(exc: UnifiedError) -> JSONResponse: + """从统一异常对象构建错误响应（统一入口）。 + + 所有满足 UnifiedError 协议的异常都通过此函数构建响应， + 确保 handler 逻辑完全一致，不再有 isinstance 特判。 + + Args: + exc: 满足 UnifiedError 协议的异常对象。 + + Returns: + 统一格式的 JSONResponse。 + """ + headers = _extract_retry_after(exc) + return build_error_response( + status_code=exc.status_code, + code=exc.error_code, + message=exc.message, + details=exc.details, + trace_id=exc.trace_id, + headers=headers, + ) diff --git a/backend/server/utils/inflight_tracker.py b/backend/server/utils/inflight_tracker.py new file mode 100644 index 00000000..d9cb91f3 --- /dev/null +++ b/backend/server/utils/inflight_tracker.py @@ -0,0 +1,78 @@ +"""在途请求计数器与中间件，用于优雅关停时排空在途请求。 + +``InflightRequestTracker`` 通过 ``increment`` / ``decrement`` 维护在途请求 +计数，计数归零时 ``wait_drained`` 立即返回；超时未归零则抛 ``TimeoutError``， +由 ``HostShutdown._waitForInflightRequests`` 捕获并记录告警后强制进入下一步 +（§11.9）。 + +``InflightTrackingMiddleware`` 在每个 HTTP 请求进入时 ``increment``、结束时 +``decrement``（``finally`` 保证异常路径也执行），确保计数准确。 +""" + +from __future__ import annotations + +import asyncio + +from starlette.middleware.base import BaseHTTPMiddleware +from starlette.requests import Request + +__all__ = ["InflightRequestTracker", "InflightTrackingMiddleware"] + + +class InflightRequestTracker: + """在途请求计数器，用于优雅关停时排空在途请求。 + + 关键不变量： + - ``_count`` 归零时 ``_zero_event`` 被 set，``wait_drained`` 立即返回。 + - ``decrement`` 防御性归零，避免计数变为负数。 + """ + + def __init__(self) -> None: + self._count = 0 + self._zero_event = asyncio.Event() + self._zero_event.set() # 初始为 0，已触发 + + @property + def count(self) -> int: + """当前在途请求数。""" + return self._count + + def increment(self) -> None: + """在途请求计数 +1，清除归零事件。""" + self._count += 1 + self._zero_event.clear() + + def decrement(self) -> None: + """在途请求计数 -1，归零时触发归零事件。""" + self._count -= 1 + if self._count <= 0: + self._count = 0 + self._zero_event.set() + + async def wait_drained(self, timeout: float) -> None: + """等待在途请求排空。 + + 参数： + timeout: 等待超时（秒），超时后抛 ``TimeoutError``。 + + Raises: + TimeoutError: 超时未排空。 + """ + await asyncio.wait_for(self._zero_event.wait(), timeout=timeout) + + +class InflightTrackingMiddleware(BaseHTTPMiddleware): + """在途请求计数中间件。 + + 每个请求进入时 ``increment``、结束时 ``decrement``（``finally`` 保证 + 异常路径也执行）。注册时应最后注册（LIFO 顺序中最先执行），确保覆盖 + 所有后续中间件与路由处理。 + """ + + async def dispatch(self, request: Request, call_next): + tracker: InflightRequestTracker = request.app.state.inflight_tracker + tracker.increment() + try: + return await call_next(request) + finally: + tracker.decrement() diff --git a/backend/server/utils/lifespan.py b/backend/server/utils/lifespan.py index 51e424e5..56ac17b8 100644 --- a/backend/server/utils/lifespan.py +++ b/backend/server/utils/lifespan.py @@ -15,11 +15,15 @@ from yuxi.knowledge import knowledge_base from yuxi.utils import logger from yuxi.agents.backends.sandbox import init_sandbox_provider, shutdown_sandbox_provider from yuxi import get_version +from server.utils.inflight_tracker import InflightRequestTracker @asynccontextmanager async def lifespan(app: FastAPI): + # 初始化在途请求计数器（InflightTrackingMiddleware 依赖） + app.state.inflight_tracker = InflightRequestTracker() + # 注册外部系统适配器（避免模块导入副作用） try: register_adapters() diff --git a/backend/server/utils/trace_middleware.py b/backend/server/utils/trace_middleware.py new file mode 100644 index 00000000..88126de0 --- /dev/null +++ b/backend/server/utils/trace_middleware.py @@ -0,0 +1,37 @@ +"""TraceId 中间件：为每个请求生成/注入 trace_id。 + +优先从 X-Request-Id header 读取（支持上游链路传递）， +否则生成 uuid4。写入 ContextVar 供下游使用。 +""" + +from __future__ import annotations + +from starlette.middleware.base import BaseHTTPMiddleware +from starlette.requests import Request +from starlette.responses import Response +from yuxi.utils.trace_context import ( + generate_trace_id, + reset_trace_id, + set_trace_id, +) + +_TRACE_ID_HEADER = "X-Request-Id" + + +class TraceIdMiddleware(BaseHTTPMiddleware): + """为每个请求注入 trace_id 的中间件。 + + 注册时应最后注册（LIFO 顺序中最先执行），确保所有后续中间件、 + 异常处理器都能从 ContextVar 读取 trace_id。 + """ + + async def dispatch(self, request: Request, call_next): + trace_id = request.headers.get(_TRACE_ID_HEADER) or generate_trace_id() + set_trace_id(trace_id) + try: + response: Response = await call_next(request) + # 回写 trace_id 到响应头，前端可关联 + response.headers[_TRACE_ID_HEADER] = trace_id + return response + finally: + reset_trace_id()