背景与客户场景
客户的不同部门关注不同的 IQC（进料检验）、OQC（出货检验）报表。为减少逐份检查，系统提供自主订阅：本次新增或修改的数据所在批次判异异常时，汇总异常项目并通知订阅者。
例如，小王和小李订阅了 IQC 报表 A。供应商数据导入后，本次变化涉及的“长度”“宽度”项目出现异常，系统生成一个事件，两人各收到一条消息，独立管理已读和删除状态。
产品借鉴
这里只参考产品行为，不推断它们的内部架构。
核心模型
实现使用 Python、SQLAlchemy 和 PostgreSQL，沿用现有判异计算和定时任务。核心是三张表：
subject_type + subject_id 定位报表，例如 IQC_REPORT + 101，避免不同模块的同 ID 报表混淆。
event_type 标识变化种类，如 iqc.supplier.sample.imported 表示 IQC 供应商样本导入。导入、手动新增、手动修改分别编码，页面统一显示“数据已更新”。用户订阅报表时，后端将该模块支持的事件类型写入 event_types。
拆开保存后，业务侧只负责记录变化，派发逻辑负责匹配接收人，消息表负责个人处理状态。取消订阅不会删除历史消息。
payload 保存派发依据：变化标识、变化时间、报表信息、来源和异常项目；content 只复制展示所需的报表信息、来源、异常项目 ID 和名称。两者都是 JSON 字段。
如何生成事件
只看项目的“异常”状态会把历史异常也算进去。当前用 Change 记录本次实际变化的样本 ID，再匹配计算结果中的异常批次。原值不变的编辑不记入变化。
# 示意：changed_sample_ids 是本次变化的样本 ID 集合
has_abnormal_change = any(
sample.id in changed_sample_ids
and batch_key(sample.detect_time, sample.batch) in abnormal_batches
for sample in samples
)
例如，历史批次 B1 异常，本次新增到 B2：计算后只有 B1 异常，不通知；B2 也异常，则通知。修改后仍命中异常批次，也会再次通知。
这个条件判断的是变化样本所在批次是否异常，不证明某个数值单独造成了异常。
flowchart TD
A[新增、修改或导入样本] --> B{有实际变化的样本吗}
B -->|否| X[不生成通知事件]
B -->|是| C[记录样本 ID<br/>执行现有判异计算并保存结果]
C --> D{变化涉及的项目<br/>是否全部计算成功}
D -->|否| X
D -->|是| E{变化样本所在批次<br/>是否判异异常}
E -->|否| X
E -->|是| F[按报表汇总异常项目<br/>IQC 再按来源分组]
F --> G[保存事件，等待派发]
一次导入处理内，同一报表的异常项目汇总成一个事件，粒度不按文件划分。IQC 按供应商、天禄自检分别汇总：两种来源都有异常，就生成两个事件。数据迁移不生成通知事件。
手动单项操作走异步会话 publish_manual，导入计算子进程走同步会话 publish_import_sync；复用异常匹配、内容组装和 SQL 构造，分别实现数据库调用。
如何派发消息
事件入库后，直接调用派发方法，为符合条件的订阅者写入消息表，不经过内存队列。处理步骤如下：
查询对象匹配、订阅时间不晚于本次变化时间的订阅者。
检查账号是否有效、事件类型是否匹配，以及当前模块查看和订阅权限。
为符合条件的用户写入消息，把事件标记为已处理（
PROCESSED），一起提交事务。
事务保证这组操作全部成功或全部撤销：中途失败就回滚本轮消息写入，不会把事件误标为已完成。没有合格接收人时，也标记处理完成。
派发时查询当前订阅和权限，重试也一样。失去权限后停止通知、保留订阅；取消订阅后，不再补发尚未生成的消息。历史消息按当前模块订阅权限查看和管理。
这里的成功指消息入库，不代表用户已看到。前端小铃铛每 60 秒拉取未读消息，与后端补偿任务独立运行。
失败补偿与去重
事件有三个状态：PENDING 待处理、PROCESSED 已处理、FAILED 派发失败。每分钟扫描待处理事件，以及距离上次更新至少一个重试间隔的失败事件，每轮最多 100 条。
flowchart TD
A[事件入库 PENDING] -->|立即调用| B[派发消息]
B --> C{事务是否提交成功}
C -->|是| D[个人消息入库<br/>事件标记 PROCESSED]
C -->|否| E[回滚本轮写入<br/>另行记录 FAILED]
F[每分钟扫描] --> G[取 PENDING 和达到重试间隔的 FAILED]
G --> B
立即派发和补偿任务可能同时处理同一事件。仅“先查再插”无法防重，两个任务可能同时查到没有消息。实现依靠数据库唯一约束保证幂等，即重复处理不会重复产生消息：
订阅唯一键：
(user_id, subject_type, subject_id)，防止重复订阅。事件唯一键：对象、事件类型和
payload.change_key，防止同一次变化重复生成事件。消息唯一键：
(user_id, event_id)，防止同一事件重复通知同一用户。
插入冲突使用 ON CONFLICT DO NOTHING。更新状态时还会确认事件尚未完成，否则回滚本轮事务。这里没有额外的应用层锁，使用的是数据库自身的锁和事务。
change_key 是一次变化的随机标识，只在该次处理内沿用；重新发起请求生成新标识，不会按数据内容合并事件。
消息通过 deleted_at 软删除，记录保留但查询隐藏，唯一键仍防止重试恢复已删除的消息。
补偿范围
业务保存、事件入库分别提交，消息和事件完成状态再一起提交。通知失败不会回滚业务数据。
补偿只覆盖已入库的事件。业务已保存、事件尚未入库时进程中断，当前扫描无法补回。
以后若要消除这个间隙，可采用事务 Outbox：业务数据和待派发事件在同一事务保存，提交后再处理事件；它不一定需要消息队列。
