"""Webhook 子域 Router。 外部系统限界上下文的 Webhook 管理 API,覆盖订阅 CRUD / 续期 / 事件处理。 所有端点通过 ``create_use_cases_from_db`` 装配 use_cases,经 ``webhook_subscription_service`` / ``webhook_event_handler`` 端口调用用例。 路径顺序约束:静态路径 ``/subscriptions`` 和 ``/events`` 必须在动态路径 ``/subscriptions/{subscription_id}`` 前声明,避免被路径参数捕获。 特殊端点:``POST /events`` 为外部系统回调入口,无认证依赖,从请求头提取 ``subscription_slug`` / ``signature``,请求体解析为 ``event_type`` + ``payload``。 ``ExternalSystemError``(含 ``WebhookError``)由全局异常处理器 (``server.main.external_system_error_handler``)统一映射为对应 HTTP 状态码, 本端点不捕获,保证配置类错误(404 订阅不存在)与安全类错误(400 验签失败) 不被 200 响应掩盖。 """ from __future__ import annotations from datetime import datetime from typing import Any, Literal from fastapi import APIRouter, Body, Depends, Header, Path, Query, Request from pydantic import BaseModel, ConfigDict, Field, field_validator from sqlalchemy.ext.asyncio import AsyncSession from yuxi.external_systems.infrastructure.container import create_use_cases_from_db from yuxi.external_systems.use_cases.dto.webhook import ( BatchReplayEventsInput, CreateWebhookSubscriptionInput, DeleteSubscriptionInput, GetEventInput, GetFilterRulesInput, GetSubscriptionInput, GetTransformRulesInput, HandleEventInput, ListExpiringSubscriptionsInput, ListFailedEventsInput, ListRenewalFailuresInput, ListWebhookEventsInput, ListWebhookSubscriptionsInput, RenewSubscriptionInput, ReplayEventInput, TestSubscriptionInput, UpdateFilterRulesInput, UpdateTransformRulesInput, UpdateWebhookSubscriptionInput, VerifySignatureInput, ) from yuxi.storage.postgres.models_business import User from yuxi.utils.crypto import SENSITIVE_HTTP_HEADERS from server.utils.auth_middleware import get_admin_user, get_db, get_required_user webhook_router = APIRouter(prefix="/webhooks", tags=["external-systems-webhook"]) # Webhook 事件 payload 最大字节数(序列化后),防止外部系统投递超大 payload _MAX_PAYLOAD_BYTES = 256 * 1024 # 256 KB def _extract_client_ip(request: Request) -> str | None: """提取客户端真实 IP,优先解析 ``x-forwarded-for``(与访问日志中间件一致)。 截断到 64 字符以对齐 ``ExternalWebhookEvent.source_ip`` 列 ``String(64)``。 """ forwarded_for = request.headers.get("x-forwarded-for") if forwarded_for: ip = forwarded_for.split(",")[0].strip() return ip[:64] if ip else None if request.client: return request.client.host[:64] return None def _sanitize_headers(headers: Any) -> dict[str, Any]: """过滤敏感请求头,避免 ``authorization`` / ``cookie`` 等泄露到持久化层。 敏感字段集合统一引用 ``yuxi.utils.crypto.SENSITIVE_HTTP_HEADERS``(单一事实源)。 """ return {key: value for key, value in headers.items() if key.lower() not in SENSITIVE_HTTP_HEADERS} # ---------------- Request Schemas ---------------- class CreateWebhookSubscriptionRequest(BaseModel): """创建 Webhook 订阅请求体。字段对齐 ``CreateWebhookSubscriptionInput``。 字段长度约束对齐 ``ExternalWebhookSubscription`` ORM 列定义,在边界层 拦截非法输入:``slug`` 对齐 String(128) 且与 system_router 共用 slug 正则;``env_key`` 对齐 String(32);``source_event_type`` 对齐 String(128); ``target_handler_type`` 对齐 String(32);``target_handler_id`` 对齐 String(64); ``callback_path`` 对齐 String(256);``secret_algorithm`` 对齐 String(32); ``retention_days`` 对齐 Integer 且限制合理范围。 """ model_config = ConfigDict(frozen=True) slug: str = Field(..., min_length=1, max_length=128, pattern=r"^[a-zA-Z_][a-zA-Z0-9_-]{0,127}$") name: str = Field(..., min_length=1, max_length=128) system_id: int = Field(..., ge=1) source_event_type: str = Field(..., min_length=1, max_length=128) target_handler_type: str = Field(..., min_length=1, max_length=32) callback_path: str = Field(..., min_length=1, max_length=256) description: str = "" env_key: str = Field(default="default", max_length=32) target_handler_id: str | None = Field(default=None, max_length=64) secret_algorithm: str = Field(default="hmac_sha256", max_length=32) filter_rules: dict[str, Any] = Field(default_factory=dict) transform_rules: dict[str, Any] = Field(default_factory=dict) enabled: bool = True expires_at: datetime | None = None auto_renew: bool = True retention_days: int = Field(default=30, ge=1, le=3650) class UpdateWebhookSubscriptionRequest(BaseModel): """更新 Webhook 订阅请求体。``id`` 由路径提供,``updated_by`` 由认证用户填充。 字段长度约束对齐 ``ExternalWebhookSubscription`` ORM 列定义,仅透传客户端 显式设置的字段(通过 ``exclude_unset=True``),未设置字段保持 ``None`` 以 保留部分更新语义。``status`` 允许管理员手动切换订阅状态(如 paused/active)。 """ model_config = ConfigDict(frozen=True) name: str | None = Field(default=None, min_length=1, max_length=128) description: str | None = None target_handler_id: str | None = Field(default=None, max_length=64) filter_rules: dict[str, Any] | None = None transform_rules: dict[str, Any] | None = None enabled: bool | None = None status: Literal["pending", "active", "paused", "expired", "disabled"] | None = None expires_at: datetime | None = None auto_renew: bool | None = None retention_days: int | None = Field(default=None, ge=1, le=3650) class HandleEventRequest(BaseModel): """Webhook 事件投递请求体。``subscription_slug`` / ``signature`` 由请求头提供。 ``event_type`` 长度对齐 ``ExternalWebhookEvent.event_type`` String(128)。 ``payload`` 序列化后不得超过 ``_MAX_PAYLOAD_BYTES``,防止超大投递。 """ model_config = ConfigDict(frozen=True) event_type: str = Field(..., min_length=1, max_length=128) payload: dict[str, Any] = Field(default_factory=dict) external_event_id: str | None = Field(default=None, max_length=256) @field_validator("payload") @classmethod def _validate_payload_size(cls, v: dict[str, Any]) -> dict[str, Any]: import json encoded = json.dumps(v, ensure_ascii=False) if len(encoded.encode("utf-8")) > _MAX_PAYLOAD_BYTES: raise ValueError( f"payload 序列化后超过 {_MAX_PAYLOAD_BYTES} 字节限制", ) return v class BatchReplayEventsRequest(BaseModel): """批量重放 Webhook 事件请求体。""" model_config = ConfigDict(frozen=True) event_ids: list[int] = Field(..., min_length=1, max_length=100) subscription_id: int | None = Field(default=None, ge=1) class TestWebhookSubscriptionRequest(BaseModel): """测试 Webhook 订阅请求体。 ``dry_run=True`` 时仅验证签名与规则配置,不持久化事件、不分派到处理方, 避免测试产生真实副作用。 """ model_config = ConfigDict(frozen=True) payload: dict[str, Any] | None = None event_type: str | None = Field(default=None, max_length=128) signature: str | None = None dry_run: bool = True class VerifySignatureRequest(BaseModel): """验证 Webhook 签名请求体。 ``payload`` 使用 ``dict`` 类型(与实际 Webhook 投递一致),验签时按 ``json.dumps(sort_keys=True, ensure_ascii=False)`` 编码后计算 HMAC, 确保与 ``handle_event`` 中的验签基准一致。 """ model_config = ConfigDict(frozen=True) payload: dict[str, Any] = Field(default_factory=dict) signature: str = Field(..., min_length=1) class UpdateFilterRulesRequest(BaseModel): """更新 Webhook 订阅过滤规则请求体。""" model_config = ConfigDict(frozen=True) filter_rules: dict[str, Any] class UpdateTransformRulesRequest(BaseModel): """更新 Webhook 订阅转换规则请求体。""" model_config = ConfigDict(frozen=True) transform_rules: dict[str, Any] # ---------------- 静态路径端点(必须在 /subscriptions/{subscription_id} 之前声明) ---------------- @webhook_router.get("/subscriptions", response_model=dict) async def list_subscriptions( limit: int = Query(100, ge=1, le=500), offset: int = Query(0, ge=0), system_id: int | None = Query(None, ge=1), env_key: str | None = Query(None, max_length=32), source_event_type: str | None = Query(None, max_length=128), status: Literal["pending", "active", "paused", "expired", "disabled"] | None = Query( None, description="订阅状态:pending/active/paused/expired/disabled" ), enabled: bool | None = Query(None), target_handler_type: str | None = Query(None, max_length=32), target_handler_id: str | None = Query(None, max_length=64), auto_renew: bool | None = Query(None), db: AsyncSession = Depends(get_db), current_user: User = Depends(get_required_user), ) -> dict[str, Any]: """分页列出 Webhook 订阅。""" use_cases = create_use_cases_from_db(db) input_dto = ListWebhookSubscriptionsInput( limit=limit, offset=offset, system_id=system_id, env_key=env_key, source_event_type=source_event_type, status=status, enabled=enabled, target_handler_type=target_handler_type, target_handler_id=target_handler_id, auto_renew=auto_renew, ) output = await use_cases.webhook_subscription_service.list_subscriptions(input_dto) return {"success": True, "data": output.model_dump()} @webhook_router.post("/subscriptions", response_model=dict) async def create_subscription( payload: CreateWebhookSubscriptionRequest, db: AsyncSession = Depends(get_db), current_user: User = Depends(get_admin_user), ) -> dict[str, Any]: """创建 Webhook 订阅。``created_by`` 由当前管理员填充。""" use_cases = create_use_cases_from_db(db) input_dto = CreateWebhookSubscriptionInput( **payload.model_dump(), created_by=current_user.uid, ) output = await use_cases.webhook_subscription_service.create_subscription(input_dto) return {"success": True, "data": output.model_dump()} @webhook_router.post("/events", response_model=dict) async def handle_event( request: Request, payload: HandleEventRequest, db: AsyncSession = Depends(get_db), subscription_slug: str = Header(..., alias="X-Webhook-Subscription", max_length=128), signature: str | None = Header(None, alias="X-Webhook-Signature", max_length=512), ) -> dict[str, Any]: """处理 Webhook 事件投递。 无认证端点:外部系统回调入口。从请求头提取 ``subscription_slug`` / ``signature``, 请求体解析为 ``event_type`` + ``payload``。``ExternalSystemError``(含 ``WebhookError``)由全局异常处理器统一映射为对应 HTTP 状态码,本端点不捕获, 保证配置类错误(404 订阅不存在)与安全类错误(400 验签失败)不被 200 响应掩盖。 ``signature`` 可选:订阅未配置 secret 时不要求签名;订阅配置 secret 时由 use case 层 ``WebhookEventHandler.handle_event`` 强制验签。请求头在透传到 持久化前由 ``_sanitize_headers`` 剔除 ``authorization`` / ``cookie`` 等敏感字段。 """ use_cases = create_use_cases_from_db(db) input_dto = HandleEventInput( subscription_slug=subscription_slug, event_type=payload.event_type, payload=payload.payload, headers=_sanitize_headers(request.headers), signature=signature, source_ip=_extract_client_ip(request), external_event_id=payload.external_event_id, ) output = await use_cases.webhook_event_handler.handle_event(input_dto) return {"success": True, "data": output.model_dump()} @webhook_router.get("/events", response_model=dict) async def list_events( limit: int = Query(100, ge=1, le=500), offset: int = Query(0, ge=0), subscription_id: int | None = Query(None, ge=1), subscription_slug: str | None = Query(None, max_length=128), event_type: str | None = Query(None, max_length=128), processing_status: Literal["pending", "processing", "processed", "failed", "ignored", "duplicate"] | None = Query( None, description="处理状态:pending/processing/processed/failed/ignored/duplicate" ), correlation_id: str | None = Query(None, max_length=128), triggered_execution_id: str | None = Query(None, max_length=128), signature_verified: bool | None = Query(None), source_ip: str | None = Query(None, max_length=64), start: datetime | None = Query(None, description="ISO8601 开始时间"), end: datetime | None = Query(None, description="ISO8601 结束时间"), db: AsyncSession = Depends(get_db), current_user: User = Depends(get_required_user), ) -> dict[str, Any]: """分页列出 Webhook 事件。""" use_cases = create_use_cases_from_db(db) input_dto = ListWebhookEventsInput( limit=limit, offset=offset, subscription_id=subscription_id, subscription_slug=subscription_slug, event_type=event_type, processing_status=processing_status, correlation_id=correlation_id, triggered_execution_id=triggered_execution_id, signature_verified=signature_verified, source_ip=source_ip, start=start, end=end, ) output = await use_cases.webhook_subscription_service.list_events(input_dto) return {"success": True, "data": output.model_dump()} @webhook_router.post("/events/batch-replay", response_model=dict) async def batch_replay_events( payload: BatchReplayEventsRequest, db: AsyncSession = Depends(get_db), current_user: User = Depends(get_admin_user), ) -> dict[str, Any]: """批量重放 Webhook 事件。""" use_cases = create_use_cases_from_db(db) input_dto = BatchReplayEventsInput( event_ids=payload.event_ids, subscription_id=payload.subscription_id, ) output = await use_cases.webhook_event_handler.batch_replay_events(input_dto) return {"success": True, "data": output.model_dump()} @webhook_router.get("/events/failed", response_model=dict) async def list_failed_events( max_attempts: int | None = Query(None, ge=1), limit: int = Query(100, ge=1, le=500), db: AsyncSession = Depends(get_db), current_user: User = Depends(get_required_user), ) -> dict[str, Any]: """列出处理失败的 Webhook 事件。""" use_cases = create_use_cases_from_db(db) input_dto = ListFailedEventsInput(max_attempts=max_attempts, limit=limit) output = await use_cases.webhook_subscription_service.list_failed_events(input_dto) return {"success": True, "data": output.model_dump()} @webhook_router.get("/subscriptions/expiring", response_model=dict) async def list_expiring_subscriptions( days: int = Query(7, ge=1, le=365), limit: int = Query(100, ge=1, le=500), db: AsyncSession = Depends(get_db), current_user: User = Depends(get_required_user), ) -> dict[str, Any]: """列出即将过期的 Webhook 订阅。""" use_cases = create_use_cases_from_db(db) input_dto = ListExpiringSubscriptionsInput(days=days, limit=limit) output = await use_cases.webhook_subscription_service.list_expiring_subscriptions(input_dto) return {"success": True, "data": output.model_dump()} @webhook_router.get("/subscriptions/renewal-failures", response_model=dict) async def list_renewal_failures( min_attempts: int = Query(1, ge=1), limit: int = Query(100, ge=1, le=500), db: AsyncSession = Depends(get_db), current_user: User = Depends(get_required_user), ) -> dict[str, Any]: """列出续期失败的 Webhook 订阅。""" use_cases = create_use_cases_from_db(db) input_dto = ListRenewalFailuresInput(min_attempts=min_attempts, limit=limit) output = await use_cases.webhook_subscription_service.list_renewal_failures(input_dto) return {"success": True, "data": output.model_dump()} # ---------------- 动态路径端点 /subscriptions/{subscription_id} ---------------- @webhook_router.get("/subscriptions/{subscription_id}", response_model=dict) async def get_subscription( subscription_id: int = Path(ge=1), db: AsyncSession = Depends(get_db), current_user: User = Depends(get_required_user), ) -> dict[str, Any]: """获取 Webhook 订阅详情。""" use_cases = create_use_cases_from_db(db) input_dto = GetSubscriptionInput(id=subscription_id) output = await use_cases.webhook_subscription_service.get_subscription(input_dto) return {"success": True, "data": output.model_dump()} @webhook_router.put("/subscriptions/{subscription_id}", response_model=dict) async def update_subscription( subscription_id: int = Path(ge=1), payload: UpdateWebhookSubscriptionRequest = Body(...), db: AsyncSession = Depends(get_db), current_user: User = Depends(get_admin_user), ) -> dict[str, Any]: """更新 Webhook 订阅。``updated_by`` 由当前管理员填充。仅透传客户端显式设置的字段。""" use_cases = create_use_cases_from_db(db) input_dto = UpdateWebhookSubscriptionInput( id=subscription_id, **payload.model_dump(exclude_unset=True), updated_by=current_user.uid, ) output = await use_cases.webhook_subscription_service.update_subscription(input_dto) return {"success": True, "data": output.model_dump()} @webhook_router.delete("/subscriptions/{subscription_id}", response_model=dict) async def delete_subscription( subscription_id: int = Path(ge=1), db: AsyncSession = Depends(get_db), current_user: User = Depends(get_admin_user), ) -> dict[str, Any]: """删除 Webhook 订阅。""" use_cases = create_use_cases_from_db(db) input_dto = DeleteSubscriptionInput(id=subscription_id) output = await use_cases.webhook_subscription_service.delete_subscription(input_dto) return {"success": True, "data": output.model_dump()} @webhook_router.post("/subscriptions/{subscription_id}/renew", response_model=dict) async def renew_subscription( subscription_id: int = Path(ge=1), db: AsyncSession = Depends(get_db), current_user: User = Depends(get_admin_user), ) -> dict[str, Any]: """续期 Webhook 订阅。""" use_cases = create_use_cases_from_db(db) input_dto = RenewSubscriptionInput(id=subscription_id) output = await use_cases.webhook_subscription_service.renew_subscription(input_dto) return {"success": True, "data": output.model_dump()} @webhook_router.get("/events/{event_id}/processing-status", response_model=dict) async def get_event_processing_status( event_id: int = Path(ge=1), db: AsyncSession = Depends(get_db), current_user: User = Depends(get_required_user), ) -> dict[str, Any]: """查询 Webhook 事件处理状态详情。""" use_cases = create_use_cases_from_db(db) input_dto = GetEventInput(id=event_id) output = await use_cases.webhook_subscription_service.get_event_deliveries(input_dto) return {"success": True, "data": output.model_dump()} @webhook_router.get("/events/{event_id}/deliveries", response_model=dict, deprecated=True) async def get_event_deliveries( event_id: int = Path(ge=1), db: AsyncSession = Depends(get_db), current_user: User = Depends(get_required_user), ) -> dict[str, Any]: """查询 Webhook 事件投递历史(已废弃,请使用 ``/processing-status``)。""" use_cases = create_use_cases_from_db(db) input_dto = GetEventInput(id=event_id) output = await use_cases.webhook_subscription_service.get_event_deliveries(input_dto) return {"success": True, "data": output.model_dump()} @webhook_router.post("/events/{event_id}/replay", response_model=dict) async def replay_event( event_id: int = Path(ge=1), db: AsyncSession = Depends(get_db), current_user: User = Depends(get_admin_user), ) -> dict[str, Any]: """重放 Webhook 事件。""" use_cases = create_use_cases_from_db(db) input_dto = ReplayEventInput(id=event_id) output = await use_cases.webhook_event_handler.replay_event(input_dto) return {"success": True, "data": output.model_dump()} @webhook_router.get("/subscriptions/{subscription_id}/filter-rules", response_model=dict) async def get_filter_rules( subscription_id: int = Path(ge=1), db: AsyncSession = Depends(get_db), current_user: User = Depends(get_required_user), ) -> dict[str, Any]: """获取 Webhook 订阅过滤规则。""" use_cases = create_use_cases_from_db(db) input_dto = GetFilterRulesInput(id=subscription_id) output = await use_cases.webhook_subscription_service.get_filter_rules(input_dto) return {"success": True, "data": output.model_dump()} @webhook_router.put("/subscriptions/{subscription_id}/filter-rules", response_model=dict) async def update_filter_rules( subscription_id: int = Path(ge=1), payload: UpdateFilterRulesRequest = Body(...), db: AsyncSession = Depends(get_db), current_user: User = Depends(get_admin_user), ) -> dict[str, Any]: """更新 Webhook 订阅过滤规则。``updated_by`` 由当前管理员填充。""" use_cases = create_use_cases_from_db(db) input_dto = UpdateFilterRulesInput( id=subscription_id, filter_rules=payload.filter_rules, updated_by=current_user.uid, ) output = await use_cases.webhook_subscription_service.update_filter_rules(input_dto) return {"success": True, "data": output.model_dump()} @webhook_router.get("/subscriptions/{subscription_id}/transform-rules", response_model=dict) async def get_transform_rules( subscription_id: int = Path(ge=1), db: AsyncSession = Depends(get_db), current_user: User = Depends(get_required_user), ) -> dict[str, Any]: """获取 Webhook 订阅转换规则。""" use_cases = create_use_cases_from_db(db) input_dto = GetTransformRulesInput(id=subscription_id) output = await use_cases.webhook_subscription_service.get_transform_rules(input_dto) return {"success": True, "data": output.model_dump()} @webhook_router.put("/subscriptions/{subscription_id}/transform-rules", response_model=dict) async def update_transform_rules( subscription_id: int = Path(ge=1), payload: UpdateTransformRulesRequest = Body(...), db: AsyncSession = Depends(get_db), current_user: User = Depends(get_admin_user), ) -> dict[str, Any]: """更新 Webhook 订阅转换规则。``updated_by`` 由当前管理员填充。""" use_cases = create_use_cases_from_db(db) input_dto = UpdateTransformRulesInput( id=subscription_id, transform_rules=payload.transform_rules, updated_by=current_user.uid, ) output = await use_cases.webhook_subscription_service.update_transform_rules(input_dto) return {"success": True, "data": output.model_dump()} @webhook_router.post("/subscriptions/{subscription_id}/test", response_model=dict) async def test_subscription( subscription_id: int = Path(ge=1), payload: TestWebhookSubscriptionRequest = Body(...), db: AsyncSession = Depends(get_db), current_user: User = Depends(get_admin_user), ) -> dict[str, Any]: """测试 Webhook 订阅。""" use_cases = create_use_cases_from_db(db) input_dto = TestSubscriptionInput( id=subscription_id, payload=payload.payload, event_type=payload.event_type, signature=payload.signature, dry_run=payload.dry_run, ) output = await use_cases.webhook_subscription_service.test_subscription(input_dto) return {"success": True, "data": output.model_dump()} @webhook_router.post("/subscriptions/{subscription_id}/verify-signature", response_model=dict) async def verify_signature( subscription_id: int = Path(ge=1), payload: VerifySignatureRequest = Body(...), db: AsyncSession = Depends(get_db), current_user: User = Depends(get_admin_user), ) -> dict[str, Any]: """验证 Webhook 签名。""" use_cases = create_use_cases_from_db(db) input_dto = VerifySignatureInput( id=subscription_id, payload=payload.payload, signature=payload.signature, ) output = await use_cases.webhook_subscription_service.verify_signature(input_dto) return {"success": True, "data": output.model_dump()}