"""多渠道网关相关的 PostgreSQL 数据模型。 表名统一以 `channel_` 为前缀；属于渠道限界上下文（channel bounded context）， 与 `external_systems` / `scheduled_task` 等上下文隔离，不跨上下文 import 实现类。所有表统一包含审计字段（created_by/updated_by/created_at/updated_at/ is_deleted/deleted_at），支持软删除与操作追溯。 本模块承载渠道网关域自身的持久化模型： - `channel_accounts`：渠道账户配置与状态 - `channel_sessions`：渠道会话映射 - `channel_user_identities`：跨渠道统一身份 - `channel_pairings`：DM 配对审批 - `channel_audit_logs`：渠道审计日志（append-only） - `channel_outbox_entries`：持久化投递队列 - `channel_idempotency`：写操作幂等记录 - `channel_reports`：一次性报告元数据与内容 设计依据：docs/vibe/V1.2/06-数据库设计方案文档.md """ from __future__ import annotations from datetime import timedelta from typing import Any from sqlalchemy import ( BigInteger, Boolean, Column, DateTime, Float, ForeignKey, Index, Integer, String, Text, UniqueConstraint, text, ) from sqlalchemy.dialects.postgresql import JSON, JSONB from yuxi.storage.postgres.models_business import Base from yuxi.utils.datetime_utils import format_utc_datetime, utc_now_naive JSON_VALUE = JSON().with_variant(JSONB, "postgresql") # 幂等记录默认过期时间（小时） IDEMPOTENCY_DEFAULT_TTL_HOURS = 24 class ChannelAccount(Base): """渠道账户配置与状态。 对应聚合根 ``ChannelAccount``（``yuxi.channels.core.model.channel_account``）。 状态机：active → disabled（手动禁用）、active → degraded → active（自动降级恢复）、 disabled → active（手动启用）。disabled 状态不得自动恢复，必须手动启用。 ``enabled`` 与 ``status`` 的语义关系： - ``enabled`` 为管理员意图（admin intent），``status`` 为运行时状态（runtime state）。 - 约束：``enabled=False`` ⇔ ``status='disabled'``；``enabled=True`` 时 ``status`` 可为 ``active`` 或 ``degraded``。应用层必须保证两者一致性，避免出现 ``enabled=False, status='active'`` 等矛盾状态。 """ __tablename__ = "channel_accounts" # 主键 id = Column(Integer, primary_key=True, autoincrement=True) # 业务标识 channel_type = Column(String(32), nullable=False, index=True, comment="渠道类型：feishu/dingtalk/wecom/webchat/...") 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="渠道配置（敏感字段加密存储）") # 状态 enabled = Column( Boolean, nullable=False, default=True, comment="管理员启用意图；与 status='disabled' 互为充要，应用层须保证一致" ) status = Column(String(32), nullable=False, default="active", comment="运行时状态机：active/disabled/degraded") plugin_status = Column( String(32), nullable=False, default="stopped", comment="插件生命周期状态（FR-32）：discovered/parsed/loaded/initialized/starting/running/paused/resuming/stopping/stopped/unloaded/failed", ) # 接入态状态机（与运行态 status 解耦）：pending/configured/verified/online/offline/failed onboarding_status = Column( String(32), nullable=False, default="pending", server_default="pending", comment="接入态状态机：pending/configured/verified/online/offline/failed", ) # 凭证引用与版本（version 化凭证存储） 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(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，无论成功失败均更新；用于诊断检查任务是否存活）" ) transport_cursor = Column( String(256), nullable=False, default="", server_default="", comment="传输游标，Puller类型使用" ) last_rotated_at = Column(DateTime, nullable=True, comment="凭据最近轮换时间") version = Column(Integer, nullable=False, default=1, comment="乐观锁版本号") # 审计字段 created_by = Column(String(64), nullable=True) updated_by = Column(String(64), nullable=True) created_at = Column(DateTime, nullable=False, default=utc_now_naive) 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__ = ( UniqueConstraint("channel_type", "account_id", name="uq_channel_accounts_type_account"), Index("ix_channel_accounts_status", "status"), Index("ix_channel_accounts_plugin_status", "plugin_status"), ) def to_dict(self) -> dict[str, Any]: """序列化为字典；config 中的敏感字段由 persistence 层解密，to_dict 不返回明文。""" return { "id": self.id, "channel_type": self.channel_type, "account_id": self.account_id, "display_name": self.display_name, "config": self.config or {}, "enabled": bool(self.enabled), "status": self.status, "plugin_status": self.plugin_status, "onboarding_status": self.onboarding_status, "credential_ref": self.credential_ref, "credential_version": self.credential_version, "service_user_uid": self.service_user_uid, "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), "transport_cursor": self.transport_cursor or "", "last_rotated_at": format_utc_datetime(self.last_rotated_at), "version": self.version, "created_by": self.created_by, "updated_by": self.updated_by, "created_at": format_utc_datetime(self.created_at), "updated_at": format_utc_datetime(self.updated_at), } class ChannelSession(Base): """渠道会话映射，关联渠道账户与内部 Conversation。 对应聚合根 ``ChannelSession``（``yuxi.channels.core.model.channel_session``）。 conversation_id 不设唯一约束，允许跨渠道关联（FR-06）。 """ __tablename__ = "channel_sessions" # 主键 id = Column(Integer, primary_key=True, autoincrement=True) # 业务标识 session_id = Column(String(64), nullable=False, unique=True, comment="渠道会话 ID（UUID，业务标识）") account_id = Column( Integer, ForeignKey("channel_accounts.id"), nullable=False, index=True, comment="关联渠道账户 ID（ORM 外键）", ) channel_type = Column(String(32), nullable=False, index=True, comment="渠道类型（冗余，便于查询）") peer_id = Column(String(256), nullable=False, comment="对端 ID（用户 ID 或群组 ID）") chat_type = Column(String(16), nullable=False, comment="会话类型：p2p / group") # 关联内部会话（可空，首次消息时创建；不设唯一约束，支持跨渠道关联 FR-06） conversation_id = Column( Integer, ForeignKey("conversations.id"), nullable=True, comment="关联内部会话 ID（可空）", ) # 跨渠道身份与路由 unified_identity_id = Column(String(64), nullable=True, index=True, comment="统一身份 ID（FR-06 跨渠道关联）") owner_peer_id = Column(String(256), nullable=True, comment="主会话所有者对端 ID（FR-26）") is_temporary = Column(Boolean, nullable=False, default=False, comment="临时会话标记（FR-27）") last_route_at = Column(DateTime, nullable=True, comment="最近路由更新时间（FR-04/FR-26）") route_match_source = Column( String(32), nullable=True, comment="最近路由匹配来源（direct/normalized_direct/parent/normalized_parent/wildcard/channel_binding）", ) last_message_at = Column( DateTime, nullable=True, comment="最近一次消息时间（冗余字段，用于会话列表排序与 inactive 临时会话清理）" ) closed_at = Column(DateTime, nullable=True, comment="会话关闭时间") version = Column(Integer, nullable=False, default=1, comment="乐观锁版本号") # 审计字段 created_by = Column(String(64), nullable=True) updated_by = Column(String(64), nullable=True) created_at = Column(DateTime, nullable=False, default=utc_now_naive) 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__ = ( # 部分唯一索引：同一账户下对端 ID 唯一（仅未软删除时），支持软删除后重建 Index( "uq_channel_sessions_account_peer_active", "account_id", "peer_id", unique=True, postgresql_where=text("is_deleted = 0"), ), Index("ix_channel_sessions_unified_identity", "unified_identity_id", "channel_type", "account_id"), Index("ix_channel_sessions_conversation", "conversation_id"), ) def to_dict(self) -> dict[str, Any]: """序列化为字典。""" return { "id": self.id, "session_id": self.session_id, "account_id": self.account_id, "channel_type": self.channel_type, "peer_id": self.peer_id, "chat_type": self.chat_type, "conversation_id": self.conversation_id, "unified_identity_id": self.unified_identity_id, "owner_peer_id": self.owner_peer_id, "is_temporary": bool(self.is_temporary), "last_route_at": format_utc_datetime(self.last_route_at), "route_match_source": self.route_match_source, "last_message_at": format_utc_datetime(self.last_message_at), "closed_at": format_utc_datetime(self.closed_at), "version": self.version, "created_by": self.created_by, "updated_by": self.updated_by, "created_at": format_utc_datetime(self.created_at), "updated_at": format_utc_datetime(self.updated_at), "is_deleted": self.is_deleted, "deleted_at": format_utc_datetime(self.deleted_at), } class ChannelUserIdentity(Base): """跨渠道统一身份映射。 对应聚合根 ``UserIdentity``（``yuxi.channels.core.model.user_identity``）。 user_id 为可空外键，未关联真实用户时为 None（走隔离策略）。 """ __tablename__ = "channel_user_identities" # 主键 id = Column(Integer, primary_key=True, autoincrement=True) # 业务标识 identity_id = Column(String(64), nullable=False, unique=True, comment="统一身份 ID（UUID，业务标识）") user_id = Column( Integer, ForeignKey("users.id"), nullable=True, index=True, comment="关联用户表 ID（可空，未关联真实用户时为 None）", ) # 身份信息 identity_type = Column(String(32), nullable=False, comment="身份类型：email/phone/employee_id") identity_value = Column(String(255), nullable=False, comment="身份值") channel_type = Column(String(32), nullable=True, comment="首次解析到的渠道类型") channel_sender_id = Column(String(255), nullable=True, comment="首次解析到的渠道发送者 ID") source = Column(String(64), nullable=True, comment="身份来源（explicit_link/phone/...）") # 绑定与合并 channel_bindings = Column( JSON_VALUE, nullable=False, default=dict, comment="渠道绑定映射：{channel_type: [peer_id, ...]}", ) merged_from = Column(JSON_VALUE, nullable=False, default=list, comment="合并来源结构化证据列表") confidence = Column(String(16), nullable=False, default="low", comment="身份置信度：low/medium/high") version = Column(Integer, nullable=False, default=1, comment="乐观锁版本号") pending_review = Column(Boolean, nullable=False, default=False, comment="低置信度合并待审批标记（H-21）") # 审计字段 created_by = Column(String(64), nullable=True) updated_by = Column(String(64), nullable=True) created_at = Column(DateTime, nullable=False, default=utc_now_naive) 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_user_identities_type_value", "identity_type", "identity_value"), # 部分唯一索引：同一身份类型 + 身份值在未软删除时唯一，从源头杜绝重复身份 Index( "uq_channel_user_identities_type_value_active", "identity_type", "identity_value", unique=True, postgresql_where=text("is_deleted = 0"), ), # GIN 索引：加速 channel_bindings / merged_from 的 @> 包含查询 Index( "ix_channel_user_identities_channel_bindings_gin", "channel_bindings", postgresql_using="gin", postgresql_ops={"channel_bindings": "jsonb_path_ops"}, ), Index( "ix_channel_user_identities_merged_from_gin", "merged_from", postgresql_using="gin", postgresql_ops={"merged_from": "jsonb_path_ops"}, ), ) def to_dict(self) -> dict[str, Any]: """序列化为字典。""" return { "id": self.id, "identity_id": self.identity_id, "user_id": self.user_id, "identity_type": self.identity_type, "identity_value": self.identity_value, "channel_type": self.channel_type, "channel_sender_id": self.channel_sender_id, "source": self.source, "channel_bindings": self.channel_bindings or {}, "merged_from": self.merged_from or [], "confidence": self.confidence, "version": self.version, "pending_review": self.pending_review, "created_by": self.created_by, "updated_by": self.updated_by, "created_at": format_utc_datetime(self.created_at), "updated_at": format_utc_datetime(self.updated_at), "is_deleted": self.is_deleted, "deleted_at": format_utc_datetime(self.deleted_at), } class ChannelPairing(Base): """DM 配对审批记录。 对应聚合根 ``PairingApproval``（``yuxi.channels.core.model.pairing_approval``）。 状态机不可逆：pending → approved | rejected | expired | revoked，终态不可变更。 配对审批表故障时必须拒绝所有 DM（fail-closed，FR-33）。 """ __tablename__ = "channel_pairings" # 主键 id = Column(Integer, primary_key=True, autoincrement=True) # 业务标识 pairing_id = Column(String(64), nullable=False, unique=True, comment="配对 ID（UUID，业务标识）") account_id = Column( Integer, ForeignKey("channel_accounts.id"), nullable=False, index=True, comment="关联渠道账户 ID（ORM 外键）", ) peer_id = Column(String(256), nullable=False, comment="渠道侧对端用户 ID") peer_name = Column(String(128), nullable=True, comment="对端名称（冗余，便于展示）") # 状态机 status = Column( String(32), nullable=False, default="pending", index=True, comment="状态：pending/approved/rejected/expired/revoked", ) approver_id = Column(String(64), nullable=True, comment="审批人用户标识") approved_at = Column(DateTime, nullable=True, comment="审批时间（approved 状态时填充）") rejected_at = Column(DateTime, nullable=True, comment="拒绝时间（rejected 状态时填充）") expired_at = Column(DateTime, nullable=True, comment="过期生效时间（expired 状态时填充，由定时任务或审批时回填）") revoked_at = Column(DateTime, nullable=True, comment="撤销时间（revoked 状态时填充，FR-33）") # 时间 requested_at = Column(DateTime, nullable=False, default=utc_now_naive, comment="申请时间") expires_at = Column(DateTime, nullable=False, index=True, comment="过期时间（默认创建时间 + 7 天）") reason = Column(Text, nullable=True, comment="审批原因 / 拒绝原因 / 撤销原因") version = Column(Integer, nullable=False, default=1, comment="乐观锁版本号") # 审计字段 created_by = Column(String(64), nullable=True) updated_by = Column(String(64), nullable=True) created_at = Column(DateTime, nullable=False, default=utc_now_naive) 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_pairings_account_peer_status", "account_id", "peer_id", "status"), # 部分唯一索引：同一账户下对端 ID 同时只允许一条 pending，防止重复申请 Index( "uq_channel_pairings_account_peer_pending", "account_id", "peer_id", unique=True, postgresql_where=text("status = 'pending' AND is_deleted = 0"), ), ) # 启用 SQLAlchemy 乐观锁：ORM 实例 flush 时自动追加 # ``WHERE id = ? AND version = ?`` 并自增 version，并发状态转换冲突时 # 抛 ``StaleDataError``，由适配器层翻译为 ``ConflictError``。 __mapper_args__ = {"version_id_col": version} def to_dict(self) -> dict[str, Any]: """序列化为字典。""" return { "id": self.id, "pairing_id": self.pairing_id, "account_id": self.account_id, "peer_id": self.peer_id, "peer_name": self.peer_name, "status": self.status, "approver_id": self.approver_id, "approved_at": format_utc_datetime(self.approved_at), "rejected_at": format_utc_datetime(self.rejected_at), "expired_at": format_utc_datetime(self.expired_at), "revoked_at": format_utc_datetime(self.revoked_at), "requested_at": format_utc_datetime(self.requested_at), "expires_at": format_utc_datetime(self.expires_at), "reason": self.reason, "version": self.version, "created_by": self.created_by, "updated_by": self.updated_by, "created_at": format_utc_datetime(self.created_at), "updated_at": format_utc_datetime(self.updated_at), "is_deleted": self.is_deleted, "deleted_at": format_utc_datetime(self.deleted_at), } class ChannelAuditLog(Base): """渠道审计日志（append-only）。 对应聚合根 ``AuditLog``（``yuxi.channels.core.model.audit_log``）。 不复用现有 ``OperationLog`` 表（语义不同）。is_deleted 恒为 0，不设 version 字段。 审计日志写入必须在事务内完成（fail-closed，FR-34）。 """ __tablename__ = "channel_audit_logs" # 主键（大整数，高写入量） id = Column(BigInteger, primary_key=True, autoincrement=True) # 业务标识 audit_log_id = Column(String(64), nullable=False, unique=True, comment="审计日志 ID（UUID，业务标识）") # 操作信息 operation = Column(String(64), nullable=False, index=True, comment="操作类型（见 AuditOperationType 枚举）") operator = Column(String(64), nullable=False, index=True, comment="操作人（用户 ID 或 system）") target_channel = Column(String(64), nullable=False, index=True, comment="目标渠道（渠道类型或 global）") target_account = Column(String(128), nullable=True, comment="目标账户 ID（可选）") result = Column(String(32), nullable=False, comment="操作结果：success / failed") # 详情 params_summary = Column( JSON_VALUE, nullable=False, default=dict, comment="参数摘要（敏感字段已脱敏）", ) trace_id = Column(String(64), nullable=True, index=True, comment="链路追踪 ID") source_ip = Column(String(64), nullable=True, comment="来源 IP") request_id = Column(String(64), nullable=True, comment="请求 ID（链路追踪）") # FR-19 管理员消息审计关联字段 message_id = Column(String(64), nullable=True, index=True, comment="关联消息 ID（FR-19）") content_summary = Column(String(256), nullable=True, comment="消息内容摘要（FR-19）") # 时间戳（业务操作时间，与 created_at 同值但语义不同） timestamp = Column(DateTime, nullable=False, default=utc_now_naive, index=True, comment="操作时间戳") # 审计字段（is_deleted 恒为 0，append-only；不设 version） created_by = Column(String(64), nullable=True) updated_by = Column(String(64), nullable=True) created_at = Column(DateTime, nullable=False, default=utc_now_naive) 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_audit_logs_operation_time", "operation", "timestamp"), Index("ix_channel_audit_logs_operator_time", "operator", "timestamp"), Index("ix_channel_audit_logs_target_time", "target_channel", "timestamp"), # 高频过滤列索引（H-8）：target_account 为审计日志查询的高频过滤条件， # 无索引会触发全表扫描。 Index("ix_channel_audit_logs_target_account", "target_account"), ) def to_dict(self) -> dict[str, Any]: """序列化为字典。""" return { "id": self.id, "audit_log_id": self.audit_log_id, "operation": self.operation, "operator": self.operator, "target_channel": self.target_channel, "target_account": self.target_account, "result": self.result, "params_summary": self.params_summary or {}, "trace_id": self.trace_id, "source_ip": self.source_ip, "request_id": self.request_id, "timestamp": format_utc_datetime(self.timestamp), "created_by": self.created_by, "updated_by": self.updated_by, "created_at": format_utc_datetime(self.created_at), "updated_at": format_utc_datetime(self.updated_at), } class ChannelOutboxEntry(Base): """持久化投递队列，存储出站消息的持久化投递状态。 对应聚合根 ``OutboxEntry``（``yuxi.channels.core.model.outbox_entry``）。 仅存消息引用，不存完整消息内容（FR-22）。conversation_id / channel_session_id 由 persistence 适配器从管道上下文注入（聚合根无这两个字段）。 并发投递锁（已废弃）：``locked_by`` / ``locked_at`` 列保留兼容历史数据， 但并发控制已由 Redis advisory lock（``OutboxRetryWorker.acquireAdvisoryLock``） 实现，DB 级租约机制不再使用。``delivery_mode`` / ``ack_policy`` 列保留默认值， 投递模式为运行时管道决策不持久化，``ack_policy`` 为入站概念不在出站流程读取。 """ __tablename__ = "channel_outbox_entries" # 主键（大整数，高写入量） id = Column(BigInteger, primary_key=True, autoincrement=True) # 业务标识 outbox_id = Column(String(64), nullable=False, unique=True, comment="发件箱 ID（UUID，业务标识）") # 关联（message_id 复用现有 Message 表，PRD §7.1.7） message_id = Column( Integer, ForeignKey("messages.id"), nullable=False, index=True, comment="关联现有 Message 表 ID" ) account_id = Column( Integer, ForeignKey("channel_accounts.id"), nullable=False, index=True, comment="目标渠道账户 ID" ) conversation_id = Column( Integer, ForeignKey("conversations.id"), nullable=False, comment="关联会话 ID（由 persistence 适配器从管道上下文注入）", ) channel_session_id = Column( Integer, ForeignKey("channel_sessions.id"), nullable=True, comment="关联渠道会话 ID（由 persistence 适配器从管道上下文注入，可空）", ) # 状态机 status = Column( String(32), nullable=False, default="pending", index=True, comment="状态机：pending/sent/suppressed/failed/sent_unconfirmed/dead", ) durability_policy = Column( String(32), nullable=False, default="required", comment="持久化策略：required/best_effort/none" ) delivery_mode = Column( String(32), nullable=False, default="persistent", comment="投递模式（FR-13/FR-22）：streaming（流式）/ persistent（持久化）", ) ack_policy = Column( String(32), nullable=False, default="after_agent_dispatch", comment="ACK 策略（FR-24）：after_record/after_agent_dispatch/after_persist/manual", ) # 并发投递锁（at-least-once 投递保障） locked_by = Column(String(64), nullable=True, comment="持有该条目的 worker ID（租约标识）") locked_at = Column(DateTime, nullable=True, comment="锁定时间；超时后由回收任务重置，使条目可被重新拉取") # 重试 retry_count = Column(Integer, nullable=False, default=0, comment="重试次数") max_retry = Column(Integer, nullable=False, default=5, comment="最大重试次数") next_retry_at = Column(DateTime, nullable=True, comment="下次重试时间（指数退避）") last_retry_at = Column(DateTime, nullable=True, comment="上次重试时间（用于诊断退避进度与 SLA）") last_error = Column(Text, nullable=True, comment="最后一次失败的错误信息") # 渠道回执 channel_msg_id = Column(String(128), nullable=True, comment="渠道返回的消息 ID") sent_at = Column(DateTime, nullable=True, comment="首次成功投递时间（用于计算投递延迟 sent_at - created_at）") latency_ms = Column(Integer, nullable=True, comment="投递延迟（毫秒），markSent 首次调用时计算") funnel_node = Column(String(32), nullable=True, comment="漏斗节点（enter/sent/suppressed/failed/dead）") # 投递原子语义与降级追踪（O-02/O-09/O-10/O-11/O-16） idempotency_key = Column( String(128), nullable=True, comment="渠道侧幂等键，markSentUnconfirmed 时写入，供恢复扫描器 queryMessageByRequestId 查询", ) channel_request_id = Column( String(128), nullable=True, comment="渠道侧请求 ID，markSentUnconfirmed 时写入，用于恢复扫描器查询投递结果", ) partial_failure = Column( Boolean, nullable=False, default=False, comment="多分片投递部分失败标记，markPartialFailure 时置 True", ) stream_aborted_at_chunk = Column( Integer, nullable=True, comment="流式投递 TTL 超时时已发送的分片数，deliver 据此调用 sendMessageContinuation 续发", ) degraded_reason = Column( Text, nullable=True, comment="降级原因摘要（BEST_EFFORT 持久化失败、markDeliveryUnconfirmedFailed 等）", ) delivered_parts = Column( JSONB, nullable=False, default=list, server_default="[]", comment="多分片投递已成功投递的分片序号列表，重试时仅发送未投递分片（H-15）", ) # 过期与追踪 expires_at = Column(DateTime, nullable=False, comment="过期时间（默认创建时间 + 24h）") trace_id = Column(String(64), nullable=True, comment="链路追踪 ID") version = Column(Integer, nullable=False, default=1, comment="乐观锁版本号") # 审计字段 created_by = Column(String(64), nullable=True) updated_by = Column(String(64), nullable=True) created_at = Column(DateTime, nullable=False, default=utc_now_naive) 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_outbox_status_retry", "status", "next_retry_at"), Index("ix_channel_outbox_expires_at", "expires_at"), # 锁超时回收扫描：回收任务查询 locked_at 早于阈值的条目 Index("ix_channel_outbox_locked_at", "locked_at"), Index("ix_channel_outbox_latency_ms", "latency_ms"), Index("ix_channel_outbox_funnel_node", "funnel_node"), # 外键列索引（H-8）：_applyOutboxQueryFilter 按 channel_session_id 过滤， # conversation_id 为高频 JOIN/级联列，无索引会触发全表扫描。 Index("ix_channel_outbox_channel_session", "channel_session_id"), Index("ix_channel_outbox_conversation", "conversation_id"), ) def to_dict(self) -> dict[str, Any]: """序列化为字典。""" return { "id": self.id, "outbox_id": self.outbox_id, "message_id": self.message_id, "account_id": self.account_id, "conversation_id": self.conversation_id, "channel_session_id": self.channel_session_id, "status": self.status, "durability_policy": self.durability_policy, "delivery_mode": self.delivery_mode, "ack_policy": self.ack_policy, "locked_by": self.locked_by, "locked_at": format_utc_datetime(self.locked_at), "retry_count": self.retry_count, "max_retry": self.max_retry, "next_retry_at": format_utc_datetime(self.next_retry_at), "last_retry_at": format_utc_datetime(self.last_retry_at), "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, "delivered_parts": self.delivered_parts, "expires_at": format_utc_datetime(self.expires_at), "trace_id": self.trace_id, "version": self.version, "created_by": self.created_by, "updated_by": self.updated_by, "created_at": format_utc_datetime(self.created_at), "updated_at": format_utc_datetime(self.updated_at), "is_deleted": self.is_deleted, "deleted_at": format_utc_datetime(self.deleted_at), } class ChannelRouteBinding(Base): """渠道路由绑定规则，描述某渠道账户在指定匹配层级下的 Agent 绑定。 对应 DTO ``RouteBindingRule``（``yuxi.channels.contract.dtos.route``）。 ``match_source`` 为 tier 名字符串（session_key/identity_id/peer_id/chat_type/ channel_session/channel_type/account/default），``match_value`` 为该层级下的 匹配值（可空表示该层级无条件命中）。同一 (channel_type, account_id, match_source, match_value) 在未软删除时唯一。 """ __tablename__ = "channel_route_bindings" # 主键 id = Column(Integer, primary_key=True, autoincrement=True) # 业务标识 binding_id = Column(String(64), nullable=False, index=True, comment="绑定 ID（UUID，业务标识）") # 关联 channel_type = Column(String(32), nullable=False, index=True, comment="渠道类型") account_id = Column(String(128), nullable=False, index=True, comment="渠道账户 ID（业务标识）") # 匹配规则 match_source = Column(String(32), nullable=False, comment="匹配层级名称（tier name）") match_value = Column(String(512), nullable=True, comment="匹配值；为空表示该层级无条件命中") # 绑定目标 agent_binding = Column(String(255), nullable=False, comment="Agent 绑定（agent slug）") # 状态 enabled = Column(Boolean, nullable=False, default=True, comment="是否启用") description = Column(String(512), nullable=True, comment="规则说明") # 审计字段 version = Column(Integer, nullable=False, default=1, comment="乐观锁版本号") created_by = Column(String(64), nullable=True) updated_by = Column(String(64), nullable=True) created_at = Column(DateTime, nullable=False, default=utc_now_naive) 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( "uq_channel_route_bindings_match", "channel_type", "account_id", "match_source", "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", "channel_type", "account_id", "enabled", ), ) class ChannelIdempotency(Base): """写操作幂等记录，支持重复请求重放首次响应。 不复用现有 ``scheduled_task_idempotency`` 表（限界上下文隔离）。 is_deleted 恒为 0，不设 version 字段；过期后由定时任务 ``ChannelIdempotencyCleanupHandler`` 硬删除。 幂等处理流程： 1. 请求到达，按 ``idempotency_key`` 查询记录。 2. 不存在：创建记录，``status='in_progress'``，``in_progress_started_at=now``， 持有处理权；处理完成后更新为 ``status='completed'`` 并写入 ``response_body``。 处理失败则更新为 ``status='failed'``（可由后续请求覆盖重试）。 3. 存在且 ``status='in_progress'``：重复请求应拒绝（409 Conflict）或阻塞等待， 避免并发执行同一操作。超时的 in_progress（超过 ``in_progress_started_at`` 阈值）视为崩溃遗留，删除重建。 4. 存在且 ``status='completed'``：重放 ``response_body``（管理员消息路径）， 或拒绝重复创建 AgentRun（入站消息路径）。 5. 过期记录由定时任务 ``ChannelIdempotencyCleanupHandler`` 按 ``expires_at`` 硬删除，不阻塞请求处理。 """ __tablename__ = "channel_idempotency" # 主键 id = Column(Integer, primary_key=True, autoincrement=True) # 幂等键 idempotency_key = Column(String(128), nullable=False, unique=True, comment="客户端传入的幂等键（UUID，PRD §7.0.3）") operation = Column( String(32), nullable=False, comment="操作类型：inbound_message（入站消息）/ message/send（管理员消息发送）", ) status = Column( String(16), nullable=False, default="in_progress", index=True, comment="处理状态：in_progress/completed/failed" ) in_progress_started_at = Column( DateTime, nullable=True, comment="开始处理时间；用于检测 in_progress 超时（疑似 worker 崩溃）并回收" ) response_body = Column( JSON_VALUE, nullable=True, comment="首次请求的响应体（status='completed' 时填充，用于重复请求重放）" ) expires_at = Column( DateTime, nullable=False, default=lambda: utc_now_naive() + timedelta(hours=IDEMPOTENCY_DEFAULT_TTL_HOURS), comment="过期时间（默认创建时间 + 24h，过期后由定时任务硬删除）", ) # 审计字段（is_deleted 恒为 0，过期后硬删除；不设 version） created_by = Column(String(64), nullable=True) updated_by = Column(String(64), nullable=True) created_at = Column(DateTime, nullable=False, default=utc_now_naive) 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_idempotency_expires_at", "expires_at"),) def to_dict(self) -> dict[str, Any]: """序列化为字典。""" return { "id": self.id, "idempotency_key": self.idempotency_key, "operation": self.operation, "status": self.status, "in_progress_started_at": format_utc_datetime(self.in_progress_started_at), "response_body": self.response_body, "expires_at": format_utc_datetime(self.expires_at), "created_by": self.created_by, "updated_by": self.updated_by, "created_at": format_utc_datetime(self.created_at), "updated_at": format_utc_datetime(self.updated_at), } class ChannelContentReviewRecord(Base): """内容审核历史记录 ORM 模型。 持久化 ContentReviewRecord DTO，承载审核结果与命中片段。 独立于 ChannelAuditLog（操作记录）与 ChannelOutboxEntry（投递记录）。 """ __tablename__ = "channel_content_review_records" # 主键（高写入量审核记录用 BigInteger） id = Column(BigInteger, primary_key=True, autoincrement=True) # 业务标识 review_id = Column( String(64), nullable=False, unique=True, comment="审核记录 ID（rev_YYYYMMDDhhmmssffffff_xxxxxxxx）" ) # 关联上下文 channel_type = Column(String(32), nullable=False, index=True, comment="渠道类型（feishu / wechat / ...）") account_id = Column(String(64), nullable=False, index=True, comment="渠道账户 ID") # 审核入参 resource_type = Column( String(32), nullable=False, comment="资源类型：message_text / message_attachment / user_profile" ) content_preview = Column(String(200), nullable=False, comment="待审核内容预览（前 200 字符）") # 审核结论 verdict = Column(String(16), nullable=False, index=True, comment="审核结论：pass / review / block") confidence = Column(Float, nullable=False, comment="置信度 [0.0, 1.0]") categories = Column(JSON_VALUE, nullable=False, default=list, comment='命中分类数组（如 ["politics", "violence"]）') detail = Column( JSON_VALUE, nullable=False, default=list, comment="命中片段数组（含 snippet / position / category / severity）" ) # 时间与人 reviewed_at = Column(DateTime, nullable=False, index=True, comment="审核时间戳（UTC）") reviewer = Column(String(64), nullable=False, comment="审核人（operator.user_id）") # 来源与追踪 source = Column( String(32), nullable=False, 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) created_at = Column(DateTime, nullable=False, default=utc_now_naive) 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) # 复合索引（按渠道+账户查询、按时间范围查询） # 注：source 不作为查询过滤条件（仅聚合 case when 使用），不建索引； # reviewed_at 是高频时间范围扫描字段，单列索引直接支持 stats/analytics 趋势分桶。 __table_args__ = ( Index( "ix_channel_content_review_channel_account", "channel_type", "account_id", ), Index( "ix_channel_content_review_reviewed_at", "reviewed_at", ), ) def to_dict(self) -> dict[str, Any]: """序列化为字典（datetime 统一用 format_utc_datetime）。""" return { "id": self.id, "review_id": self.review_id, "channel_type": self.channel_type, "account_id": self.account_id, "resource_type": self.resource_type, "content_preview": self.content_preview, "verdict": self.verdict, "confidence": self.confidence, "categories": self.categories, "detail": self.detail, "reviewed_at": format_utc_datetime(self.reviewed_at), "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), "updated_at": format_utc_datetime(self.updated_at), "is_deleted": self.is_deleted, "deleted_at": format_utc_datetime(self.deleted_at), }