ForcePilot/backend/server/routers/external_systems/webhook_router.py
Kris 6dd16ad3f8 refactor(external-systems-routers): 统一分页参数格式并完善各路由文档与校验
本次提交对多个外部系统路由进行了多维度优化:
1.  统一分页参数:将所有路由的`page = offset//limit +1`、`page_size=limit`替换为标准的`limit`+`offset`分页格式
2.  完善接口文档:补充多个端点的功能说明、参数含义与返回字段解释
3.  增强参数校验:新增字段长度限制、正则校验、枚举类型约束与业务逻辑校验
4.  优化代码复用:提取重复逻辑为辅助函数,减少样板代码
5.  修复接口问题:修正工具健康检查端点路径参数类型,优化导出接口响应格式
6.  补充异常处理:为批量操作添加异常捕获与日志记录,避免流程中断
2026-07-11 06:57:24 +08:00

599 lines
24 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""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()}