Kris
1c6bed1403
本次提交包含多项核心改进： 1. 新增微信公众号插件拉取传输模式配置，完善manifest与manifest加载逻辑 2. 新增运行态状态机与幂等冲突错误体系，补充错误映射与领域错误导出 3. 优化出站与入站上下文，新增幂等键、流式中断标记等字段 4. 完善发件箱仓储与模型，新增失败条目查询、投递原子语义字段 5. 修复签名验证阶段异常捕获逻辑，防御性处理内置NotImplementedError 6. 新增出站预算释放方法，完善机器人循环预算管控 7. 优化出站管道格式阶段，新增消息长度校验逻辑 8. 完善出站打字指示器阶段，新增重复启动防御与状态同步 9. 重构出站标记失败阶段，按源状态分支处理状态转换 10. 新增入站幂等过滤阶段，修复入站路由阶段空指针问题 11. 优化出站恢复扫描器，修复状态机调用与聚合根重建逻辑 12. 完善微信公众号适配器，新增类型校验与异常包装 13. 修复数据库事务回滚逻辑，简化不必要的显式回滚操作
662 lines
25 KiB
Python
662 lines
25 KiB
Python
"""Outbox DTO。
|
||
|
||
定义发件箱相关的不可变值对象，包括发件箱 ID、发件箱状态枚举、消息
|
||
持久化策略枚举、投递回执、多分片回执与发件箱条目。所有 DTO 均为
|
||
``dataclass(frozen=True)``，仅依赖标准库与契约层内部类型，用于
|
||
持久化投递、重试控制与死信管理（FR-22）。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from dataclasses import dataclass
|
||
from datetime import datetime
|
||
from enum import StrEnum
|
||
from typing import TYPE_CHECKING, Any, Literal
|
||
|
||
from yuxi.channels.contract.dtos.channel import ChannelType
|
||
from yuxi.channels.contract.dtos.common import TrendDataPoint
|
||
from yuxi.channels.contract.errors import ValidationError
|
||
|
||
if TYPE_CHECKING:
|
||
from yuxi.channels.contract.dtos.outbound import FormattedMessage
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class OutboxId:
|
||
"""发件箱 ID。
|
||
|
||
标识一条发件箱条目的唯一 ID，用于持久化投递追踪与状态同步。
|
||
|
||
字段：
|
||
value: 发件箱 ID 字符串。
|
||
"""
|
||
|
||
value: str
|
||
|
||
|
||
class OutboxStatus(StrEnum):
|
||
"""发件箱状态。
|
||
|
||
标识发件箱条目的生命周期状态，用于投递流程控制与重试决策。继承
|
||
``str, Enum`` 以支持 JSON 序列化与字符串比较。
|
||
|
||
取值：
|
||
PENDING: 待发送。
|
||
SENT: 已发送。
|
||
SUPPRESSED: 已抑制（栅栏拦截）。
|
||
FAILED: 已失败。
|
||
SENT_UNCONFIRMED: 已发未确认。
|
||
DEAD: 死信（重试超限）。
|
||
"""
|
||
|
||
PENDING = "pending"
|
||
SENT = "sent"
|
||
SUPPRESSED = "suppressed"
|
||
FAILED = "failed"
|
||
SENT_UNCONFIRMED = "sent_unconfirmed"
|
||
DEAD = "dead"
|
||
|
||
|
||
class MessageDurabilityPolicy(StrEnum):
|
||
"""消息持久化策略。
|
||
|
||
标识消息投递的持久化等级，用于权衡投递可靠性与性能。继承
|
||
``str, Enum`` 以支持 JSON 序列化与字符串比较。
|
||
|
||
取值：
|
||
REQUIRED: 必须持久化（强可靠）。
|
||
BEST_EFFORT: 尽力持久化。
|
||
NONE: 不持久化。
|
||
"""
|
||
|
||
REQUIRED = "required"
|
||
BEST_EFFORT = "best_effort"
|
||
NONE = "none"
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class DeliveryReceipt:
|
||
"""投递回执（FR-22）。
|
||
|
||
描述一次投递的回执信息，包括发件箱 ID、状态、渠道侧消息 ID、重试
|
||
次数与下次重试时间，用于投递状态同步与重试调度。
|
||
|
||
字段：
|
||
outbox_id: 发件箱 ID。
|
||
status: 发件箱状态。
|
||
channel_msg_id: 渠道侧消息 ID（可选）。
|
||
retry_count: 重试次数（默认 0）。
|
||
next_retry_at: 下次重试时间（可选）。
|
||
"""
|
||
|
||
outbox_id: str
|
||
status: OutboxStatus
|
||
channel_msg_id: str | None = None
|
||
retry_count: int = 0
|
||
next_retry_at: datetime | None = None
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class MultiPartReceipt:
|
||
"""多分片回执（FR-22）。
|
||
|
||
描述一条消息在渠道侧拆分为多个分片投递的回执，包括平台消息 ID、
|
||
分片类型、序号、可选的话题 / 回复目标 ID 与分片投递结果，用于
|
||
多分片消息的状态同步与部分失败判定（FR-22）。
|
||
|
||
字段：
|
||
platform_msg_id: 平台消息 ID。
|
||
part_type: 分片类型（text / card / attachment）。
|
||
sequence: 序号（默认 0）。
|
||
topic_id: 话题 ID（可选）。
|
||
reply_target_id: 回复目标 ID（可选）。
|
||
success: 分片投递是否成功（默认 True）。``False`` 表示该分片
|
||
投递失败，``DeliverStage`` 据此判定部分失败。
|
||
"""
|
||
|
||
platform_msg_id: str
|
||
part_type: Literal["text", "card", "attachment"]
|
||
sequence: int = 0
|
||
topic_id: str | None = None
|
||
reply_target_id: str | None = None
|
||
success: bool = True
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class OutboxConfig:
|
||
"""Outbox 配置。
|
||
|
||
描述发件箱投递相关的可配置参数，由组合根从配置源加载并注入到应用层
|
||
阶段，解耦应用层对全局配置的依赖（§6.1 应用服务层禁止依赖具体技术
|
||
适配器）。
|
||
|
||
字段：
|
||
ttl_seconds: 发件箱条目存活时间（秒），``expires_at`` 缺失时
|
||
用于回退计算过期时间。
|
||
retry_backoff_schedule: 指数退避序列（秒），由聚合根
|
||
``markFailed`` 计算 ``next_retry_at`` 时使用。
|
||
max_retry: 最大重试次数（默认 5，镜像聚合根）。
|
||
failed_retry_enabled: 是否启用 FAILED 状态自动重试扫描（默认 True，
|
||
O-04/H-7）。禁用时恢复扫描器跳过 FAILED 条目，仅由运维通过
|
||
batch-retry 端点手动重投。
|
||
failed_retry_interval_seconds: FAILED 条目重试最小间隔（秒，默认 300）。
|
||
仅扫描 ``updated_at`` 距今超过此间隔的条目，避免刚失败的条目
|
||
被立即重试导致风暴。
|
||
"""
|
||
|
||
ttl_seconds: int
|
||
retry_backoff_schedule: tuple[int, ...]
|
||
max_retry: int = 5
|
||
failed_retry_enabled: bool = True
|
||
failed_retry_interval_seconds: int = 300
|
||
|
||
def __post_init__(self) -> None:
|
||
"""校验配置参数合法性。"""
|
||
if self.ttl_seconds <= 0:
|
||
raise ValidationError("ttl_seconds", "must be a positive integer")
|
||
if self.max_retry <= 0:
|
||
raise ValidationError("max_retry", "must be a positive integer")
|
||
if not self.retry_backoff_schedule:
|
||
raise ValidationError("retry_backoff_schedule", "must be a non-empty list")
|
||
if self.failed_retry_interval_seconds <= 0:
|
||
raise ValidationError(
|
||
"failed_retry_interval_seconds",
|
||
"must be a positive integer",
|
||
)
|
||
|
||
@classmethod
|
||
def default(cls) -> OutboxConfig:
|
||
"""构造默认 OutboxConfig 实例。
|
||
|
||
集中声明默认值，供装配期在配置缺失时使用，消除应用层
|
||
三处重复硬编码（deliver_stage / outbox_mark_failed / outbox_rollback）。
|
||
|
||
默认值：
|
||
ttl_seconds: 86400（24 小时）。
|
||
retry_backoff_schedule: (60, 120, 300, 600, 1800)（指数退避序列）。
|
||
max_retry: 5。
|
||
failed_retry_enabled: True（O-04 FAILED 自动重试默认启用）。
|
||
failed_retry_interval_seconds: 300（5 分钟最小重试间隔）。
|
||
"""
|
||
return cls(
|
||
ttl_seconds=86400,
|
||
retry_backoff_schedule=(60, 120, 300, 600, 1800),
|
||
max_retry=5,
|
||
failed_retry_enabled=True,
|
||
failed_retry_interval_seconds=300,
|
||
)
|
||
|
||
|
||
class OutboxConfigHolder:
|
||
"""OutboxConfig 可变持有器，支持运行时热更新。
|
||
|
||
``OutboxConfig`` 本身为 ``frozen=True`` 不可变值对象，无法在运行时
|
||
修改字段。本持有器封装可变引用，使 ``updateRetryPolicy`` 操作能够
|
||
热更新配置，并被同一请求范围内的 handler / pipeline / adapter 共享
|
||
读取。跨请求与跨进程一致性通过 ``ConfigPort``（Redis）持久化保证。
|
||
|
||
使用方式：
|
||
- 组合根创建 holder 实例并注入到需要读取配置的组件。
|
||
- ``OutboxHandler.updateRetryPolicy`` 调用 ``holder.update()`` 热更新。
|
||
- 其他组件通过 ``holder.current`` 读取最新配置。
|
||
"""
|
||
|
||
def __init__(self, config: OutboxConfig) -> None:
|
||
self._config = config
|
||
|
||
@property
|
||
def current(self) -> OutboxConfig:
|
||
"""返回当前持有的 OutboxConfig 实例。"""
|
||
return self._config
|
||
|
||
def update(self, config: OutboxConfig) -> None:
|
||
"""热更新持有的 OutboxConfig 实例。
|
||
|
||
参数：
|
||
config: 新的 OutboxConfig 实例。
|
||
"""
|
||
self._config = config
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class OutboxEntry:
|
||
"""发件箱条目。
|
||
|
||
描述一条发件箱条目的完整状态，包括发件箱 ID、消息 ID、渠道账户
|
||
ID、状态、持久化策略、重试信息与时间戳，用于持久化投递的全流程
|
||
管理（FR-22）。
|
||
|
||
字段：
|
||
outbox_id: 发件箱 ID。
|
||
message_id: 消息 ID。BEST_EFFORT 轻量记录允许 ``None``（消息持久化
|
||
失败但仍需追踪投递状态的降级路径，O-01/C-6）。
|
||
channel_account_id: 渠道账户 ID。
|
||
status: 发件箱状态。
|
||
durability_policy: 消息持久化策略。
|
||
retry_count: 重试次数（默认 0）。
|
||
max_retry: 最大重试次数（默认 5，镜像聚合根）。
|
||
next_retry_at: 下次重试时间（可选）。
|
||
last_error: 最近错误信息（可选）。
|
||
created_at: 创建时间（可选）。
|
||
updated_at: 更新时间（可选）。
|
||
expires_at: 过期时间（可选，默认 created_at + 24h，镜像聚合根）。
|
||
channel_msg_id: 渠道侧消息 ID（可选，投递成功时填充）。
|
||
version: 乐观锁版本号（默认 1，完整更新时用于并发控制）。
|
||
channel_session_id: 渠道会话 ID（可选，业务标识 session_id UUID）。
|
||
由 ORM ``channel_session_id``（int FK）经 mapper 转换为
|
||
``str`` 业务标识填充。重试 worker 当前仍通过 ORM 反查
|
||
``channel_session_id`` 列（int PK）定位 peer_id，本字段
|
||
供其他消费方按业务标识引用会话（FR-22）。
|
||
latency_ms: 投递延迟（毫秒，可选）。仅 ``markSent`` 首次调用时
|
||
计算，供聚合视图域 analytics 子域延迟分布查询使用。默认
|
||
``None`` 保证向后兼容历史数据。
|
||
funnel_node: 漏斗节点（可选，取值 enter / sent / suppressed /
|
||
failed / dead）。供聚合视图域 analytics 子域漏斗聚合查询
|
||
使用。默认 ``None`` 保证向后兼容历史数据。
|
||
channel_type: 渠道类型（可选）。由列表查询关联 ``channel_accounts``
|
||
表填充，供管理后台展示与级联筛选使用。
|
||
idempotency_key: 幂等键（可选）。出站投递幂等控制用，跨重试稳定，
|
||
供适配器侧去重与 ``sendMessageContinuation`` 续发定位。
|
||
channel_request_id: 渠道请求 ID（可选）。适配器 ``sendMessage`` 等
|
||
调用生成的请求标识，供 ``queryMessageByRequestId`` 反查投递结果。
|
||
partial_failure: 是否部分失败（默认 False）。多分片投递时部分分片
|
||
失败置 True，供下游阶段判定降级与重试策略。
|
||
stream_aborted_at_chunk: 流式中断时分片序号（可选）。流式投递被
|
||
中断时记录已发送分片序号，供 ``sendMessageContinuation`` 续发。
|
||
degraded_reason: 降级原因（可选）。投递降级时记录原因，供可观测性
|
||
与审计使用。
|
||
"""
|
||
|
||
outbox_id: str
|
||
message_id: str | None
|
||
channel_account_id: str
|
||
status: OutboxStatus
|
||
durability_policy: MessageDurabilityPolicy
|
||
retry_count: int = 0
|
||
max_retry: int = 5
|
||
next_retry_at: datetime | None = None
|
||
last_error: str | None = None
|
||
created_at: datetime | None = None
|
||
updated_at: datetime | None = None
|
||
expires_at: datetime | None = None
|
||
channel_msg_id: str | None = None
|
||
version: int = 1
|
||
channel_session_id: str | None = None
|
||
latency_ms: int | None = None
|
||
funnel_node: Literal["enter", "sent", "suppressed", "failed", "dead"] | None = None
|
||
channel_type: ChannelType | None = None
|
||
idempotency_key: str | None = None
|
||
channel_request_id: str | None = None
|
||
partial_failure: bool = False
|
||
stream_aborted_at_chunk: int | None = None
|
||
degraded_reason: str | None = None
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class RetryContext:
|
||
"""Outbox 重试上下文 DTO（FR-22）。
|
||
|
||
封装 ``OutboxRetryWorker`` 重试所需的渠道上下文，由
|
||
``PersistencePort.resolveRetryContext`` 解析后返回，消除应用层
|
||
对 SQLAlchemy ORM 的直接依赖（INV-1）。
|
||
|
||
字段：
|
||
channel_type: 渠道类型字符串（由 manifest 声明，如 ``"feishu"`` / ``"custom"``）。
|
||
account_id: 渠道账户业务 ID（str，非 ORM 主键）。
|
||
peer_id: 对端 ID（接收方标识）。
|
||
message: 完整的格式化消息对象（``FormattedMessage``），保留
|
||
``format`` / ``rich_message`` / ``attachments``，避免重试时
|
||
富媒体 / 卡片消息退化为纯文本。
|
||
"""
|
||
|
||
channel_type: str
|
||
account_id: str
|
||
peer_id: str
|
||
message: "FormattedMessage" # noqa: UP037
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class OutboxQueryFilter:
|
||
"""Outbox 查询过滤条件（OBX-01 / OBX-04）。
|
||
|
||
描述 outbox 条目列表查询与统计的过滤条件，所有字段可选，未提供时
|
||
表示不按该维度过滤。由驱动端口 ``listOutboxEntries`` /
|
||
``countOutboxEntries`` 消费，被驱动端口 ``OutboxRepositoryPort``
|
||
实现解析为持久化查询条件。
|
||
|
||
字段：
|
||
channel_type: 渠道类型（可选）。
|
||
channel_account_id: 渠道账户 ID（可选）。
|
||
status: 发件箱状态（可选）。
|
||
message_id: 消息 ID（可选）。
|
||
channel_msg_id: 渠道侧消息 ID（可选）。
|
||
created_after: 创建时间下界（可选，含）。
|
||
created_before: 创建时间上界（可选，含）。
|
||
channel_session_id: 渠道会话 ID（可选）。
|
||
retry_count_min: 最小重试次数下界（可选，含）。仅返回
|
||
``retry_count >= retry_count_min`` 的条目，用于筛选已发生重试
|
||
的条目（如排查反复失败的投递）。
|
||
message_id_like: 消息 ID 模糊搜索（可选，SQL ``LIKE`` 子串匹配）。
|
||
channel_msg_id_like: 渠道侧消息 ID 模糊搜索（可选，SQL ``LIKE``）。
|
||
channel_account_id_like: 渠道账户 ID 模糊搜索（可选，SQL ``LIKE``）。
|
||
last_error_like: 最近错误关键词筛选（可选，SQL ``LIKE``）。
|
||
"""
|
||
|
||
channel_type: ChannelType | None = None
|
||
channel_account_id: str | None = None
|
||
status: OutboxStatus | None = None
|
||
message_id: str | None = None
|
||
channel_msg_id: str | None = None
|
||
created_after: datetime | None = None
|
||
created_before: datetime | None = None
|
||
channel_session_id: str | None = None
|
||
retry_count_min: int | None = None
|
||
message_id_like: str | None = None
|
||
channel_msg_id_like: str | None = None
|
||
channel_account_id_like: str | None = None
|
||
last_error_like: str | None = None
|
||
|
||
def __post_init__(self) -> None:
|
||
"""校验过滤条件合法性。
|
||
|
||
- ``retry_count_min`` 非空时必须 >= 0。
|
||
- ``created_after`` 与 ``created_before`` 同时提供时，
|
||
``created_after`` 必须早于 ``created_before``。
|
||
"""
|
||
if self.retry_count_min is not None and self.retry_count_min < 0:
|
||
raise ValidationError("retry_count_min", "must be a non-negative integer")
|
||
if (
|
||
self.created_after is not None
|
||
and self.created_before is not None
|
||
and self.created_after >= self.created_before
|
||
):
|
||
raise ValidationError("time_range", "created_after must be earlier than created_before")
|
||
|
||
def to_dict(self) -> dict[str, Any]:
|
||
"""序列化为 dict，仅包含非 None 字段。
|
||
|
||
``channel_type`` 与 ``status`` 序列化为 ``.value`` 字符串，
|
||
``datetime`` 序列化为 ISO 8601 字符串，供日志输出与审计记录使用。
|
||
"""
|
||
result: dict[str, Any] = {}
|
||
if self.channel_type is not None:
|
||
result["channel_type"] = self.channel_type
|
||
if self.channel_account_id is not None:
|
||
result["channel_account_id"] = self.channel_account_id
|
||
if self.status is not None:
|
||
result["status"] = self.status.value
|
||
if self.message_id is not None:
|
||
result["message_id"] = self.message_id
|
||
if self.channel_msg_id is not None:
|
||
result["channel_msg_id"] = self.channel_msg_id
|
||
if self.created_after is not None:
|
||
result["created_after"] = self.created_after.isoformat()
|
||
if self.created_before is not None:
|
||
result["created_before"] = self.created_before.isoformat()
|
||
if self.channel_session_id is not None:
|
||
result["channel_session_id"] = self.channel_session_id
|
||
if self.retry_count_min is not None:
|
||
result["retry_count_min"] = self.retry_count_min
|
||
if self.message_id_like is not None:
|
||
result["message_id_like"] = self.message_id_like
|
||
if self.channel_msg_id_like is not None:
|
||
result["channel_msg_id_like"] = self.channel_msg_id_like
|
||
if self.channel_account_id_like is not None:
|
||
result["channel_account_id_like"] = self.channel_account_id_like
|
||
if self.last_error_like is not None:
|
||
result["last_error_like"] = self.last_error_like
|
||
return result
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class OutboxStats:
|
||
"""Outbox 统计信息（OBX-02）。
|
||
|
||
描述 outbox 条目按状态分组的计数快照与诊断指标，由 ``getOutboxStats``
|
||
返回，供管理后台展示投递健康度。所有计数仅含 ``is_deleted=0`` 条目。
|
||
|
||
字段：
|
||
total: 条目总数。
|
||
pending: 待发送条目数。
|
||
sent: 已发送条目数。
|
||
suppressed: 已抑制条目数。
|
||
failed: 已失败条目数。
|
||
sent_unconfirmed: 已发未确认条目数。
|
||
dead: 死信条目数。
|
||
top_errors: 最近错误 Top5，每项含 ``error`` 与 ``count``。
|
||
avg_latency_ms: 平均投递延迟（毫秒），无数据时为 ``None``。
|
||
oldest_pending_at: 最早 pending 条目创建时间，无 pending 时为 ``None``。
|
||
"""
|
||
|
||
total: int
|
||
pending: int
|
||
sent: int
|
||
suppressed: int
|
||
failed: int
|
||
sent_unconfirmed: int
|
||
dead: int
|
||
top_errors: tuple[dict[str, Any], ...] = ()
|
||
avg_latency_ms: float | None = None
|
||
oldest_pending_at: datetime | None = None
|
||
|
||
def to_dict(self) -> dict[str, Any]:
|
||
"""序列化为 dict，供控制面结果返回与日志输出使用。"""
|
||
return {
|
||
"total": self.total,
|
||
"pending": self.pending,
|
||
"sent": self.sent,
|
||
"suppressed": self.suppressed,
|
||
"failed": self.failed,
|
||
"sent_unconfirmed": self.sent_unconfirmed,
|
||
"dead": self.dead,
|
||
"top_errors": list(self.top_errors),
|
||
"avg_latency_ms": self.avg_latency_ms,
|
||
"oldest_pending_at": self.oldest_pending_at.isoformat() if self.oldest_pending_at else None,
|
||
}
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class BatchRetryResult:
|
||
"""批量重投死信结果（OBX-001）。
|
||
|
||
描述死信批量重投操作的执行结果，包括成功入队数量与失败条目。
|
||
|
||
字段：
|
||
retried_count: 成功入队重投的条目数。
|
||
failed_count: 重投失败的条目数。
|
||
retried_at: 重投执行时间。
|
||
"""
|
||
|
||
retried_count: int
|
||
failed_count: int
|
||
retried_at: datetime
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class BatchDeleteResult:
|
||
"""批量删除死信结果（OBX-002）。
|
||
|
||
描述死信批量删除操作的执行结果，包括成功删除数量。
|
||
|
||
字段：
|
||
deleted_count: 成功删除的条目数。
|
||
deleted_at: 删除执行时间。
|
||
"""
|
||
|
||
deleted_count: int
|
||
deleted_at: datetime
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class RetryPolicySnapshot:
|
||
"""重试策略快照（OBX-003）。
|
||
|
||
描述 Outbox 重试策略的只读快照，用于查询与更新操作返回。
|
||
|
||
字段：
|
||
max_retry: 最大重试次数。
|
||
ttl_seconds: 条目存活时间（秒）。
|
||
retry_backoff_schedule: 指数退避序列（秒）。
|
||
updated_at: 策略更新时间（可选，默认策略时为 None）。
|
||
"""
|
||
|
||
max_retry: int
|
||
ttl_seconds: int
|
||
retry_backoff_schedule: tuple[int, ...]
|
||
updated_at: datetime | None = None
|
||
|
||
def __post_init__(self) -> None:
|
||
"""校验策略参数合法性。"""
|
||
if self.max_retry <= 0:
|
||
raise ValidationError("max_retry", "must be a positive integer")
|
||
if self.ttl_seconds <= 0:
|
||
raise ValidationError("ttl_seconds", "must be a positive integer")
|
||
if not self.retry_backoff_schedule:
|
||
raise ValidationError("retry_backoff_schedule", "must be a non-empty list")
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class DeadLetterExportCmd:
|
||
"""死信导出命令（OBX-DL-EXPORT）。
|
||
|
||
描述死信导出查询条件，由 ``OutboxRepositoryPort.exportDeadLetters`` 消费。
|
||
``format`` 取值 ``json`` / ``csv``，默认 ``json``。
|
||
|
||
字段：
|
||
channel_type: 渠道类型过滤（可选）。
|
||
created_after: 创建时间下界（可选，含）。
|
||
created_before: 创建时间上界（可选，含）。
|
||
format: 导出格式（默认 ``json``，``json`` / ``csv``）。
|
||
limit: 导出条目上限（默认 10000），防止大规模死信队列 OOM。
|
||
"""
|
||
|
||
channel_type: ChannelType | None = None
|
||
created_after: datetime | None = None
|
||
created_before: datetime | None = None
|
||
format: Literal["json", "csv"] = "json"
|
||
limit: int = 10000
|
||
|
||
def __post_init__(self) -> None:
|
||
"""校验导出命令参数合法性。
|
||
|
||
- ``format`` 必须为 ``json`` 或 ``csv``。
|
||
- ``created_after`` 与 ``created_before`` 同时提供时，
|
||
``created_after`` 必须早于 ``created_before``。
|
||
- ``limit`` 必须为正整数。
|
||
"""
|
||
if self.format not in ("json", "csv"):
|
||
raise ValidationError("format", f"must be 'json' or 'csv', got '{self.format}'")
|
||
if (
|
||
self.created_after is not None
|
||
and self.created_before is not None
|
||
and self.created_after >= self.created_before
|
||
):
|
||
raise ValidationError("time_range", "created_after must be earlier than created_before")
|
||
if not isinstance(self.limit, int) or self.limit <= 0:
|
||
raise ValidationError("limit", f"must be a positive integer, got {self.limit!r}")
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class DeadLetterExportResult:
|
||
"""死信导出结果（OBX-DL-EXPORT）。
|
||
|
||
描述死信导出的返回内容，``content`` 在 ``format=json`` 时为记录元组，
|
||
在 ``format=csv`` 时为 CSV 字符串。
|
||
|
||
字段：
|
||
format: 导出格式（``json`` / ``csv``）。
|
||
filename: 建议的文件名（``dead-letter-{timestamp}.{format}``）。
|
||
total_records: 导出记录总数。
|
||
content: 导出内容（json 为记录元组，csv 为字符串）。
|
||
"""
|
||
|
||
format: str
|
||
filename: str
|
||
total_records: int
|
||
content: tuple[dict[str, Any], ...] | str
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class OutboxTrendQuery:
|
||
"""投递积压趋势查询（OBX-TREND）。
|
||
|
||
描述 outbox 积压趋势的查询条件，由 ``OutboxRepositoryPort.getTrend`` 消费。
|
||
``granularity`` 取值 ``minute`` / ``hour`` / ``day``，默认 ``hour``；
|
||
``metric`` 取值 ``queue_depth`` / ``retry_count`` / ``dead_count``，
|
||
默认 ``queue_depth``。
|
||
|
||
字段：
|
||
start_time: 起始时间（必填，含）。
|
||
end_time: 结束时间（必填，含）。
|
||
channel_type: 渠道类型过滤（可选）。
|
||
channel_account_id: 渠道账户业务 ID 过滤（可选）。
|
||
granularity: 时间粒度（默认 ``hour``）。
|
||
metric: 度量指标（默认 ``queue_depth``）。
|
||
"""
|
||
|
||
start_time: datetime
|
||
end_time: datetime
|
||
channel_type: ChannelType | None = None
|
||
channel_account_id: str | None = None
|
||
granularity: Literal["minute", "hour", "day"] = "hour"
|
||
metric: Literal["queue_depth", "retry_count", "dead_count"] = "queue_depth"
|
||
|
||
def __post_init__(self) -> None:
|
||
"""校验趋势查询参数合法性。
|
||
|
||
- ``start_time`` 必须早于 ``end_time``。
|
||
- ``granularity`` 必须为 ``minute`` / ``hour`` / ``day``。
|
||
- ``metric`` 必须为 ``queue_depth`` / ``retry_count`` / ``dead_count``。
|
||
"""
|
||
if self.start_time >= self.end_time:
|
||
raise ValidationError("time_range", "start_time must be earlier than end_time")
|
||
if self.granularity not in ("minute", "hour", "day"):
|
||
raise ValidationError(
|
||
"granularity",
|
||
f"must be 'minute', 'hour' or 'day', got '{self.granularity}'",
|
||
)
|
||
if self.metric not in ("queue_depth", "retry_count", "dead_count"):
|
||
raise ValidationError(
|
||
"metric",
|
||
f"must be 'queue_depth', 'retry_count' or 'dead_count', got '{self.metric}'",
|
||
)
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class OutboxTrendResult:
|
||
"""投递积压趋势结果（OBX-TREND）。
|
||
|
||
描述按时间粒度聚合后的趋势序列，``series`` 使用通用 ``TrendDataPoint``。
|
||
|
||
字段：
|
||
metric: 度量指标。
|
||
granularity: 时间粒度。
|
||
series: 趋势数据点元组。
|
||
"""
|
||
|
||
metric: str
|
||
granularity: str
|
||
series: tuple[TrendDataPoint, ...]
|
||
|
||
|
||
__all__ = [
|
||
"BatchDeleteResult",
|
||
"BatchRetryResult",
|
||
"DeadLetterExportCmd",
|
||
"DeadLetterExportResult",
|
||
"DeliveryReceipt",
|
||
"MessageDurabilityPolicy",
|
||
"MultiPartReceipt",
|
||
"OutboxConfig",
|
||
"OutboxConfigHolder",
|
||
"OutboxEntry",
|
||
"OutboxId",
|
||
"OutboxQueryFilter",
|
||
"OutboxStats",
|
||
"OutboxStatus",
|
||
"OutboxTrendQuery",
|
||
"OutboxTrendResult",
|
||
"RetryContext",
|
||
"RetryPolicySnapshot",
|
||
]
|