refactor(storage/postgres): 清理废弃字段与冗余模型
1. 移除ChannelAccount表的capabilities、config_version、last_message_at字段 2. 扩大service_user_uid字段长度至225 3. 移除AuditExportTask模型并清理相关代码 4. 更新ChannelOutboxEntry、ChannelIdempotency、ChannelContentReviewRecord的注释与索引 5. 为ChannelPairing添加SQLAlchemy乐观锁支持 6. 新增ChannelContentReviewRecord的决策时长统计字段
This commit is contained in:
parent
ff07cef881
commit
16f7e2d6ab
@ -71,9 +71,8 @@ class ChannelAccount(Base):
|
||||
account_id = Column(String(128), nullable=False, comment="渠道账户 ID（全局唯一，业务标识）")
|
||||
display_name = Column(String(128), nullable=False, comment="展示名称")
|
||||
|
||||
# 配置与能力
|
||||
# 配置
|
||||
config = Column(JSON_VALUE, nullable=False, default=dict, comment="渠道配置（敏感字段加密存储）")
|
||||
capabilities = Column(JSON_VALUE, nullable=False, default=dict, comment="能力声明（ChannelCapabilities 序列化）")
|
||||
|
||||
# 状态
|
||||
enabled = Column(
|
||||
@ -86,7 +85,6 @@ class ChannelAccount(Base):
|
||||
default="stopped",
|
||||
comment="插件生命周期状态（FR-32）：discovered/parsed/loaded/initialized/starting/running/paused/resuming/stopping/stopped/unloaded/failed",
|
||||
)
|
||||
config_version = Column(Integer, nullable=False, default=1, comment="配置版本号（FR-37 热更新冲突检测）")
|
||||
# 接入态状态机（与运行态 status 解耦）：pending/configured/verified/online/offline/failed
|
||||
onboarding_status = Column(
|
||||
String(32),
|
||||
@ -99,13 +97,12 @@ class ChannelAccount(Base):
|
||||
credential_ref = Column(String(128), nullable=True, comment="凭证引用，指向凭证存储中的版本化凭证记录")
|
||||
credential_version = Column(Integer, nullable=False, default=0, server_default="0", comment="凭证版本号")
|
||||
# 关联服务账号（user_type='service'）
|
||||
service_user_uid = Column(String(64), nullable=True, comment="关联服务账号 User.uid")
|
||||
service_user_uid = Column(String(225), nullable=True, comment="关联服务账号 User.uid")
|
||||
last_error = Column(Text, nullable=True, comment="最近一次错误信息（FR-35 健康检查详情）")
|
||||
last_error_at = Column(DateTime, nullable=True, comment="最近一次错误时间")
|
||||
last_health_check_at = Column(
|
||||
DateTime, nullable=True, comment="最近一次健康检查时间（FR-35，无论成功失败均更新；用于诊断检查任务是否存活）"
|
||||
)
|
||||
last_message_at = Column(DateTime, nullable=True, comment="最近一次消息时间（冗余字段）")
|
||||
transport_cursor = Column(
|
||||
String(256), nullable=False, default="", server_default="", comment="传输游标，Puller类型使用"
|
||||
)
|
||||
@ -134,11 +131,9 @@ class ChannelAccount(Base):
|
||||
"account_id": self.account_id,
|
||||
"display_name": self.display_name,
|
||||
"config": self.config or {},
|
||||
"capabilities": self.capabilities or {},
|
||||
"enabled": bool(self.enabled),
|
||||
"status": self.status,
|
||||
"plugin_status": self.plugin_status,
|
||||
"config_version": self.config_version,
|
||||
"onboarding_status": self.onboarding_status,
|
||||
"credential_ref": self.credential_ref,
|
||||
"credential_version": self.credential_version,
|
||||
@ -146,7 +141,6 @@ class ChannelAccount(Base):
|
||||
"last_error": self.last_error,
|
||||
"last_error_at": format_utc_datetime(self.last_error_at),
|
||||
"last_health_check_at": format_utc_datetime(self.last_health_check_at),
|
||||
"last_message_at": format_utc_datetime(self.last_message_at),
|
||||
"transport_cursor": self.transport_cursor or "",
|
||||
"last_rotated_at": format_utc_datetime(self.last_rotated_at),
|
||||
"version": self.version,
|
||||
@ -403,6 +397,11 @@ class ChannelPairing(Base):
|
||||
),
|
||||
)
|
||||
|
||||
# 启用 SQLAlchemy 乐观锁：ORM 实例 flush 时自动追加
|
||||
# ``WHERE id = ? AND version = ?`` 并自增 version，并发状态转换冲突时
|
||||
# 抛 ``StaleDataError``，由适配器层翻译为 ``ConflictError``。
|
||||
__mapper_args__ = {"version_id_col": version}
|
||||
|
||||
def to_dict(self) -> dict[str, Any]:
|
||||
"""序列化为字典。"""
|
||||
return {
|
||||
@ -506,69 +505,6 @@ class ChannelAuditLog(Base):
|
||||
}
|
||||
|
||||
|
||||
class AuditExportTask(Base):
|
||||
"""审计日志导出任务。
|
||||
|
||||
对应聚合根 ``AuditExportTask``（``yuxi.channels.core.model.audit_export_task``）。
|
||||
持久化异步导出任务的状态与结果，配合 ARQ worker 实现审计日志的异步导出。
|
||||
任务状态机：pending → running → completed | failed，终态不可变更。
|
||||
"""
|
||||
|
||||
__tablename__ = "audit_export_tasks"
|
||||
|
||||
# 主键
|
||||
id = Column(BigInteger, primary_key=True, autoincrement=True)
|
||||
|
||||
# 业务标识
|
||||
task_id = Column(String(64), nullable=False, unique=True, index=True, comment="导出任务 ID（UUID，业务标识）")
|
||||
status = Column(
|
||||
String(32), nullable=False, index=True, comment="任务状态：pending / running / completed / failed / cancelled"
|
||||
)
|
||||
|
||||
# 导出参数与结果
|
||||
query_snapshot = Column(JSON_VALUE, nullable=False, comment="查询条件快照（AuditQuery 序列化）")
|
||||
format = Column(String(16), nullable=False, default="json", comment="导出格式（当前仅支持 json）")
|
||||
total_records = Column(Integer, nullable=True, comment="导出记录总数（completed 状态时填充）")
|
||||
file_path = Column(String(512), nullable=True, comment="导出报告文件路径（completed 状态时填充）")
|
||||
partial_file_path = Column(String(512), nullable=True, comment="部分导出文件路径（cancelled 状态时填充）")
|
||||
error = Column(Text, nullable=True, comment="失败原因（failed 状态时填充）")
|
||||
|
||||
# 时间与操作人
|
||||
created_by = Column(String(64), nullable=False, index=True, comment="创建者用户 ID")
|
||||
created_at = Column(DateTime(timezone=True), nullable=False, comment="任务创建时间（UTC）")
|
||||
completed_at = Column(
|
||||
DateTime(timezone=True), nullable=True, comment="任务完成时间（completed / failed 状态时填充）"
|
||||
)
|
||||
|
||||
# 审计字段
|
||||
is_deleted = Column(Integer, nullable=False, default=0)
|
||||
deleted_at = Column(DateTime(timezone=True), nullable=True)
|
||||
|
||||
__table_args__ = (
|
||||
Index("ix_audit_export_tasks_status_time", "status", "created_at"),
|
||||
Index("ix_audit_export_tasks_created_by_time", "created_by", "created_at"),
|
||||
)
|
||||
|
||||
def to_dict(self) -> dict[str, Any]:
|
||||
"""序列化为字典。"""
|
||||
return {
|
||||
"id": self.id,
|
||||
"task_id": self.task_id,
|
||||
"status": self.status,
|
||||
"query_snapshot": self.query_snapshot or {},
|
||||
"format": self.format,
|
||||
"total_records": self.total_records,
|
||||
"file_path": self.file_path,
|
||||
"partial_file_path": self.partial_file_path,
|
||||
"error": self.error,
|
||||
"created_by": self.created_by,
|
||||
"created_at": format_utc_datetime(self.created_at),
|
||||
"completed_at": format_utc_datetime(self.completed_at),
|
||||
"is_deleted": self.is_deleted,
|
||||
"deleted_at": format_utc_datetime(self.deleted_at),
|
||||
}
|
||||
|
||||
|
||||
class ChannelOutboxEntry(Base):
|
||||
"""持久化投递队列，存储出站消息的持久化投递状态。
|
||||
|
||||
@ -576,10 +512,10 @@ class ChannelOutboxEntry(Base):
|
||||
仅存消息引用，不存完整消息内容（FR-22）。conversation_id / channel_session_id
|
||||
由 persistence 适配器从管道上下文注入（聚合根无这两个字段）。
|
||||
|
||||
并发投递锁：多 worker 拉取 pending 消息时通过 ``locked_by`` / ``locked_at``
|
||||
实现显式租约。worker 取走消息后写入 ``locked_by`` 与 ``locked_at``，处理完成
|
||||
后清空并更新 ``status``；若 worker 崩溃，``locked_at`` 超时后由回收任务重置
|
||||
锁并重新投递，保证 at-least-once 语义。
|
||||
并发投递锁（已废弃）：``locked_by`` / ``locked_at`` 列保留兼容历史数据，
|
||||
但并发控制已由 Redis advisory lock（``OutboxRetryWorker.acquireAdvisoryLock``）
|
||||
实现，DB 级租约机制不再使用。``delivery_mode`` / ``ack_policy`` 列保留默认值，
|
||||
投递模式为运行时管道决策不持久化，``ack_policy`` 为入站概念不在出站流程读取。
|
||||
"""
|
||||
|
||||
__tablename__ = "channel_outbox_entries"
|
||||
@ -723,6 +659,13 @@ class ChannelOutboxEntry(Base):
|
||||
"last_error": self.last_error,
|
||||
"channel_msg_id": self.channel_msg_id,
|
||||
"sent_at": format_utc_datetime(self.sent_at),
|
||||
"latency_ms": self.latency_ms,
|
||||
"funnel_node": self.funnel_node,
|
||||
"idempotency_key": self.idempotency_key,
|
||||
"channel_request_id": self.channel_request_id,
|
||||
"partial_failure": self.partial_failure,
|
||||
"stream_aborted_at_chunk": self.stream_aborted_at_chunk,
|
||||
"degraded_reason": self.degraded_reason,
|
||||
"expires_at": format_utc_datetime(self.expires_at),
|
||||
"trace_id": self.trace_id,
|
||||
"version": self.version,
|
||||
@ -786,6 +729,9 @@ class ChannelRouteBinding(Base):
|
||||
"match_value",
|
||||
unique=True,
|
||||
postgresql_where=text("is_deleted = 0"),
|
||||
# PG15+：NULL 视为相等，使 account/default 等 match_value 为 NULL 的
|
||||
# tier 也能在 (channel_type, account_id, match_source) 维度上唯一。
|
||||
postgresql_nulls_not_distinct=True,
|
||||
),
|
||||
Index(
|
||||
"ix_channel_route_bindings_list",
|
||||
@ -800,7 +746,8 @@ class ChannelIdempotency(Base):
|
||||
"""写操作幂等记录，支持重复请求重放首次响应。
|
||||
|
||||
不复用现有 ``scheduled_task_idempotency`` 表（限界上下文隔离）。
|
||||
is_deleted 恒为 0，不设 version 字段；过期后由定时任务硬删除。
|
||||
is_deleted 恒为 0，不设 version 字段；过期后由定时任务
|
||||
``ChannelIdempotencyCleanupHandler`` 硬删除。
|
||||
|
||||
幂等处理流程：
|
||||
1. 请求到达，按 ``idempotency_key`` 查询记录。
|
||||
@ -808,11 +755,12 @@ class ChannelIdempotency(Base):
|
||||
持有处理权；处理完成后更新为 ``status='completed'`` 并写入 ``response_body``。
|
||||
处理失败则更新为 ``status='failed'``（可由后续请求覆盖重试）。
|
||||
3. 存在且 ``status='in_progress'``：重复请求应拒绝（409 Conflict）或阻塞等待，
|
||||
避免并发执行同一操作。
|
||||
4. 存在且 ``status='completed'``：校验 ``request_hash``，匹配则重放
|
||||
``response_body``；不匹配说明同 key 不同参数，拒绝（409 Conflict）以防
|
||||
掩盖客户端 bug。
|
||||
5. 已过期（``expires_at < now``）：硬删除后按新请求处理。
|
||||
避免并发执行同一操作。超时的 in_progress（超过 ``in_progress_started_at``
|
||||
阈值）视为崩溃遗留，删除重建。
|
||||
4. 存在且 ``status='completed'``：重放 ``response_body``（管理员消息路径），
|
||||
或拒绝重复创建 AgentRun（入站消息路径）。
|
||||
5. 过期记录由定时任务 ``ChannelIdempotencyCleanupHandler`` 按 ``expires_at``
|
||||
硬删除，不阻塞请求处理。
|
||||
"""
|
||||
|
||||
__tablename__ = "channel_idempotency"
|
||||
@ -825,18 +773,7 @@ class ChannelIdempotency(Base):
|
||||
operation = Column(
|
||||
String(32),
|
||||
nullable=False,
|
||||
index=True,
|
||||
comment="操作类型：admin_message_send/pairing_approve/pairing_reject/config_update/whitelist_modify/session_merge",
|
||||
)
|
||||
account_id = Column(
|
||||
Integer,
|
||||
ForeignKey("channel_accounts.id"),
|
||||
nullable=True,
|
||||
index=True,
|
||||
comment="关联渠道账户（可选，部分操作如全局配置变更无账户）",
|
||||
)
|
||||
request_hash = Column(
|
||||
String(64), nullable=True, comment="请求参数 SHA256（用于检测同 key 不同参数的冲突；无参数操作可留空）"
|
||||
comment="操作类型：inbound_message（入站消息）/ message/send（管理员消息发送）",
|
||||
)
|
||||
status = Column(
|
||||
String(16), nullable=False, default="in_progress", index=True, comment="处理状态：in_progress/completed/failed"
|
||||
@ -862,11 +799,7 @@ class ChannelIdempotency(Base):
|
||||
is_deleted = Column(Integer, nullable=False, default=0, index=True)
|
||||
deleted_at = Column(DateTime, nullable=True)
|
||||
|
||||
__table_args__ = (
|
||||
Index("ix_channel_idempotency_created_at", "created_at"),
|
||||
Index("ix_channel_idempotency_expires_at", "expires_at"),
|
||||
Index("ix_channel_idempotency_status_started", "status", "in_progress_started_at"),
|
||||
)
|
||||
__table_args__ = (Index("ix_channel_idempotency_expires_at", "expires_at"),)
|
||||
|
||||
def to_dict(self) -> dict[str, Any]:
|
||||
"""序列化为字典。"""
|
||||
@ -874,8 +807,6 @@ class ChannelIdempotency(Base):
|
||||
"id": self.id,
|
||||
"idempotency_key": self.idempotency_key,
|
||||
"operation": self.operation,
|
||||
"account_id": self.account_id,
|
||||
"request_hash": self.request_hash,
|
||||
"status": self.status,
|
||||
"in_progress_started_at": format_utc_datetime(self.in_progress_started_at),
|
||||
"response_body": self.response_body,
|
||||
@ -930,11 +861,17 @@ class ChannelContentReviewRecord(Base):
|
||||
source = Column(
|
||||
String(32),
|
||||
nullable=False,
|
||||
index=True,
|
||||
comment="审核来源：manual_preview / inbound_pipeline / outbound_pipeline",
|
||||
)
|
||||
trace_id = Column(String(64), nullable=True, comment="链路追踪 ID")
|
||||
|
||||
# 决策时长统计（CR-STATS-01 avg_decision_seconds 真实化）
|
||||
# decision_started_at：审核请求开始时间（首次写入时与 reviewed_at 同值）
|
||||
# decision_completed_at：审核结论最终确定时间（人工 update_verdict 时刷新）
|
||||
# 仅对有 completed_at 的记录参与 avg 计算（首次审核即终态的不计入）
|
||||
decision_started_at = Column(DateTime, nullable=True, comment="审核决策开始时间（UTC）")
|
||||
decision_completed_at = Column(DateTime, nullable=True, comment="审核决策完成时间（UTC，人工覆盖时刷新）")
|
||||
|
||||
# 审计字段（与现有渠道模型统一）
|
||||
created_by = Column(String(64), nullable=True)
|
||||
updated_by = Column(String(64), nullable=True)
|
||||
@ -944,6 +881,8 @@ class ChannelContentReviewRecord(Base):
|
||||
deleted_at = Column(DateTime, nullable=True)
|
||||
|
||||
# 复合索引（按渠道+账户查询、按时间范围查询）
|
||||
# 注：source 不作为查询过滤条件（仅聚合 case when 使用），不建索引；
|
||||
# reviewed_at 是高频时间范围扫描字段，单列索引直接支持 stats/analytics 趋势分桶。
|
||||
__table_args__ = (
|
||||
Index(
|
||||
"ix_channel_content_review_channel_account",
|
||||
@ -951,8 +890,7 @@ class ChannelContentReviewRecord(Base):
|
||||
"account_id",
|
||||
),
|
||||
Index(
|
||||
"ix_channel_content_review_source_reviewed",
|
||||
"source",
|
||||
"ix_channel_content_review_reviewed_at",
|
||||
"reviewed_at",
|
||||
),
|
||||
)
|
||||
@ -974,6 +912,8 @@ class ChannelContentReviewRecord(Base):
|
||||
"reviewer": self.reviewer,
|
||||
"source": self.source,
|
||||
"trace_id": self.trace_id,
|
||||
"decision_started_at": format_utc_datetime(self.decision_started_at),
|
||||
"decision_completed_at": format_utc_datetime(self.decision_completed_at),
|
||||
"created_by": self.created_by,
|
||||
"updated_by": self.updated_by,
|
||||
"created_at": format_utc_datetime(self.created_at),
|
||||
@ -981,71 +921,3 @@ class ChannelContentReviewRecord(Base):
|
||||
"is_deleted": self.is_deleted,
|
||||
"deleted_at": format_utc_datetime(self.deleted_at),
|
||||
}
|
||||
|
||||
|
||||
class ChannelReport(Base):
|
||||
"""渠道报告 ORM 模型。
|
||||
|
||||
存储一次性报告的元数据与内容。报告内容由 scheduler handler
|
||||
异步生成后写入 ``content`` 字段（JSONB）。
|
||||
|
||||
设计依据：对齐《19-聚合视图域-reports-router-设计方案.md》§8.4。
|
||||
"""
|
||||
|
||||
__tablename__ = "channel_reports"
|
||||
|
||||
# 主键
|
||||
id = Column(Integer, primary_key=True, autoincrement=True)
|
||||
|
||||
# 业务标识
|
||||
report_id = Column(String(64), nullable=False, unique=True, index=True, comment="报告 ID（UUID，业务标识）")
|
||||
task_id = Column(String(64), nullable=False, unique=True, index=True, comment="关联调度任务 ID（业务标识）")
|
||||
report_type = Column(String(32), nullable=False, comment="报告类型")
|
||||
status = Column(String(16), nullable=False, default="pending", comment="状态：pending/generating/ready/failed")
|
||||
|
||||
# 内容与参数
|
||||
params = Column(JSON_VALUE, nullable=False, default=dict, comment="报告生成参数（序列化）")
|
||||
content = Column(JSON_VALUE, nullable=True, comment="报告内容（ready 状态时填充，JSONB）")
|
||||
error_message = Column(String(512), nullable=True, comment="失败原因（failed 状态时填充）")
|
||||
|
||||
# 时间与来源
|
||||
created_at = Column(DateTime, nullable=False, default=utc_now_naive, comment="创建时间")
|
||||
ready_at = Column(DateTime, nullable=True, comment="报告就绪时间（ready 状态时填充）")
|
||||
created_by = Column(String(64), nullable=False, default="system", comment="创建者（用户 ID 或 system）")
|
||||
|
||||
# 重试关联（RPT-ONEOFF-RETRY）
|
||||
retried_from = Column(String(64), nullable=True, comment="重试来源报告的 task_id（仅重试生成的新报告非空）")
|
||||
retried_at = Column(DateTime, nullable=True, comment="原报告被重试的时间（原报告标记字段）")
|
||||
|
||||
# 审计字段（与 channels 其他表一致）
|
||||
updated_by = Column(String(64), nullable=True)
|
||||
updated_at = Column(DateTime, nullable=False, default=utc_now_naive, onupdate=utc_now_naive)
|
||||
is_deleted = Column(Integer, nullable=False, default=0, index=True)
|
||||
deleted_at = Column(DateTime, nullable=True)
|
||||
|
||||
__table_args__ = (
|
||||
Index("ix_channel_reports_status_created", "status", "created_at"),
|
||||
Index("ix_channel_reports_type_created", "report_type", "created_at"),
|
||||
)
|
||||
|
||||
def to_dict(self) -> dict[str, Any]:
|
||||
"""序列化为字典。"""
|
||||
return {
|
||||
"id": self.id,
|
||||
"report_id": self.report_id,
|
||||
"task_id": self.task_id,
|
||||
"report_type": self.report_type,
|
||||
"status": self.status,
|
||||
"params": self.params or {},
|
||||
"content": self.content,
|
||||
"error_message": self.error_message,
|
||||
"created_at": format_utc_datetime(self.created_at),
|
||||
"ready_at": format_utc_datetime(self.ready_at),
|
||||
"created_by": self.created_by,
|
||||
"retried_from": self.retried_from,
|
||||
"retried_at": format_utc_datetime(self.retried_at),
|
||||
"updated_by": self.updated_by,
|
||||
"updated_at": format_utc_datetime(self.updated_at),
|
||||
"is_deleted": self.is_deleted,
|
||||
"deleted_at": format_utc_datetime(self.deleted_at),
|
||||
}
|
||||
|
||||
Loading…
Reference in New Issue
Block a user