hms/docs/audits/v2/03-data-flow-traces.md

数据流追踪报告

审计日期: 2026-05-04 | 范围: 12 条核心数据流

---

DF1: 体征录入 → 存储 → 告警

调用链

1. health_data_handler::create_vital_signs — 权限校验 health.health-data.manage，输入消毒
2. health_data_service::vital_signs::create_vital_signs — 校验患者存在，构建 ActiveModel
3. vital_signs::Entity::insert — SeaORM 写入 vital_signs 表
4. health_data_service::alert::check_vital_signs_alert — 从 DB 加载 critical_value_threshold，逐指标比对阈值
5. event_bus.publish("health_data.critical_alert") — 发布危急值事件（含患者/医生信息）
6. audit_service::record — 审计日志

发现的问题

• 告警触发不阻塞: L129 check_vital_signs_alert 返回值被 .await 但未处理错误（fire-and-forget 语义），阈值配置缺失时静默跳过，无重试机制
• 告警双路径: 存在两条独立告警路径 — alert.rs（危急值事件）和 alert_engine.rs（规则引擎），阈值来源不同，可能产生重复告警

Mermaid

flowchart LR
  A[Handler: create_vital_signs] --> B[Service: 校验患者]
  B --> C[(Entity: vital_signs)]
  C --> D[check_vital_signs_alert]
  D --> E[(critical_value_threshold)]
  D --> F[EventBus: critical_alert]

---

DF2: BLE 设备 → 聚合 → 趋势

调用链

1. ble_gateway_handler::gateway_upload — API Key 认证（非 JWT），接收多患者批量数据
2. ble_gateway_service::gateway_upload — 校验患者-网关绑定关系
3. device_reading_service::batch_create_readings — 校验患者、绑定设备、批量插入
4. device_readings::Entity::insert_many — ON CONFLICT DO NOTHING 去重
5. sync_bp_glucose_to_vital_signs — 双写 vital_signs 表（仅血压/血糖）
6. upsert_hourly_aggregates — 按 (device_type, hour) 分组聚合写入 vital_signs_hourly
7. event_bus.publish("device.readings.synced") — 发布设备同步事件
8. vital_signs_daily_service::aggregate_daily — 定时从 hourly → daily 聚合

发现的问题

• 双写无事务保护: L124 sync_bp_glucose_to_vital_signs 错误仅 warn 不阻塞主流程，可能导致 vital_signs 与 device_readings 数据不一致
• 去重计数不准: L252 ON CONFLICT DO NOTHING 返回提交总数而非实际插入数，duplicates 字段语义不准确

Mermaid

flowchart LR
  A[Gateway: API Key 认证] --> B[校验绑定关系]
  B --> C[(device_readings)]
  C --> D[双写 vital_signs]
  C --> E[聚合 hourly]
  E --> F[(vital_signs_daily)]
  C --> G[EventBus: synced]

---

DF3: AI 分析 SSE → 建议记录 → 行动分发

调用链

1. erp-ai/handler::stream_trends — 权限校验，获取趋势数据，脱敏
2. analysis.stream_analyze — 调用 LLM（Claude），返回 SSE 流 + analysis_id
3. build_sse_stream — 逐 chunk 推送 SSE event:chunk，完成后 event:done
4. analysis.complete_analysis — 标记分析完成，存储完整内容
5. post_process::post_process_analysis — 解析双通道输出（文本+结构化）
6. SuggestionService::create_suggestions — 创建 ai_suggestion 记录
7. event_bus.publish("ai.analysis.completed") — 发布分析完成事件
8. action_inbox_service::list_action_items — UNION 查询 ai_suggestion + alerts + followup

发现的问题

• 自动分析使用 Uuid::nil(): auto_analysis.rs L108 system_user_id = Uuid::nil()，audit trail 中操作人为零值 UUID，无法追溯
• post_process 失败静默: L66 建议创建失败仅 warn，分析仍标记 completed

Mermaid

flowchart LR
  A[SSE Handler: stream_trends] --> B[LLM Stream]
  B --> C[complete_analysis]
  C --> D[post_process: 解析双通道]
  D --> E[(ai_suggestion)]
  D --> F[EventBus: analysis.completed]
  F --> G[ActionInbox: UNION 查询]

---

DF4: 透析记录 → KDIGO 风险评分 → 告警

调用链

1. dialysis_handler::create_dialysis_record — 权限校验 health.dialysis.manage
2. dialysis_service::create_dialysis_record — PII 加密（symptoms, complication_notes），写入
3. dialysis_record::Entity::insert — SeaORM 写入 dialysis_record 表
4. event_bus.publish("dialysis.record.created") — 发布透析记录创建事件
5. erp-ai/handler::assess_dialysis_risk — 接收 DialysisLabInput（Kt/V, eGFR 等）
6. DialysisRiskScorer::assess — KDIGO CKD 分期（基于 eGFR）+ 本地规则引擎评分
7. LocalRulesEngine — 评估 Kt/V、血磷、血钾、血红蛋白等规则，返回 risk_level + suggestions

发现的问题

• 透析创建与风险评分解耦: 风险评分为独立 HTTP 端点，创建透析记录后不会自动触发 KDIGO 评分，需前端手动调用
• 事件未被消费: dialysis.record.created 事件已发布但无确认的 subscriber 自动触发风险评估

Mermaid

flowchart LR
  A[Handler: create_dialysis_record] --> B[PII 加密]
  B --> C[(dialysis_record)]
  C --> D[EventBus: record.created]
  E[手动调用 assess_dialysis_risk] --> F[KDIGO 分期]
  F --> G[LocalRulesEngine]
  G --> H[RiskAssessment]

---

DF5: 告警触发 → 降噪 → 推送

调用链

1. alert_engine::evaluate_rules — 加载 alert_rules（按 device_type 过滤活跃规则）
2. 规则评估: single_threshold / consecutive / trend（基于 vital_signs_hourly）
3. alert_noise_reducer::apply_noise_reduction — 两级降噪:
   • check_patient_escalation — 30min 内 3 次低级告警 → 升级严重度
   • check_system_aggregation — 5min 内同患者重复 → 抑制通知（critical 除外）
4. alerts::Entity::insert — 写入告警记录（含升级后严重度）
5. event_bus.publish("alert.triggered") — 发布告警事件（含 notify_roles, suppressed 标记）
6. erp-message/sse_handler::message_stream — SSE 推送:
   • alert.triggered → SSE event: alert（校验管床医生关系）
   • 30s 心跳保活 + Last-Event-ID 断点续传

发现的问题

• 聚合窗口无滑动: check_system_aggregation 每次插入后都查询最近 5min 内所有告警，高并发下可能导致 N+1 查询
• SSE 无背压控制: sse_handler 使用 unbounded channel，告警风暴时可能导致内存压力

Mermaid

flowchart LR
  A[alert_engine: 规则匹配] --> B[降噪: 患者升级]
  B --> C[降噪: 系统聚合]
  C --> D[(alerts 表)]
  D --> E[EventBus: alert.triggered]
  E --> F[SSE: 管床医生过滤]
  F --> G[前端推送]

---

DF6: 护理计划 → 项目 → 预后

调用链

1. care_plan_handler::create_care_plan — 权限校验 health.care-plan.manage
2. care_plan_service::create_care_plan — 校验患者、计划类型，写入
3. care_plan::Entity::insert — 写入 care_plan 表（status=draft）
4. care_plan_item::Entity::insert — 写入护理项目（排序、类型、频次）
5. care_plan_service::create_care_plan_outcome — 写入预后评估记录
6. care_plan_outcome::Entity::insert — 写入 care_plan_outcome 表（评分、达成状态）
7. event_bus.publish("care_plan.status_changed") — 状态变更事件
8. event_bus.publish("care_plan.outcome_updated") — 预后更新事件

发现的问题

• plan → item → outcome 无事务包裹: 三张表写入分散在不同函数调用中，中间失败可能导致孤立记录
• status 流转无状态机保护: care_plan.status 为自由字符串，未使用枚举约束，可写入非法状态

Mermaid

flowchart LR
  A[Handler: create_care_plan] --> B[校验患者+类型]
  B --> C[(care_plan)]
  C --> D[(care_plan_item)]
  D --> E[(care_plan_outcome)]
  E --> F[EventBus: status_changed]

---

汇总

数据流	关键风险	严重度
DF1	告警双路径可能重复触发	中
DF2	双写无事务保护	高
DF3	自动分析 operator 为零值 UUID	低
DF4	透析创建与 KDIGO 评分未自动串联	高
DF5	SSE 无背压、聚合查询 N+1	中
DF6	三表写入无事务、状态无枚举约束	高
DF7	前端无健康摘要页面、概念混淆	中
DF8	$everything 无分页、血压 ID 重复、缺 Bundle link	高
DF9	allowed_patient_ids 越权、JWT 弱 fallback	高
DF10	签到无事件发布、连续天数计算脆弱	低
DF11	前后对比功能未实现、再分析无幂等	高
DF12	串行处理、降采样效率低、双写失败静默	中

---

DF7: 家庭代理查看 → 同意验证 → 健康摘要

调用链

1. family_proxy_handler::list_my_family_patients — 查询已授权访问的家庭患者
2. family_proxy_service::check_access — 验证 consent + access_level
3. family_proxy_handler::get_family_health_summary — 返回脱敏健康摘要

发现的问题

• 前端无健康摘要页面: 后端 get_family_health_summary 已实现但前端无调用
• 概念混淆: MP family 页面调用 listPatients(患者管理 API)而非 list_my_family_patients(家庭代理 API)
• link_user 无入口: 家庭成员绑定用户流程在小程序中无入口

Mermaid

flowchart LR
  A[MP 就诊人管理] --> B{Consent 验证}
  B -->|granted| C[family_proxy_service]
  C --> D[脱敏健康摘要]
  B -->|denied| E[403 拒绝]

---

DF8: FHIR 资源查询 → 转换器 → 标准 JSON

调用链

1. oauth_auth_middleware — Bearer Token + scope 验证
2. fhir/handler.rs — 8 种资源类型路由分发
3. fhir/converter.rs — 内部 Entity → FHIR R4 资源映射
4. fhir/types.rs — FHIR 资源类型定义

发现的问题

• $everything 无分页限制: Observations 限 200 条，但 Devices/Encounters 无限，大数据量可能 OOM
• 日期搜索不完整: 仅支持 gt/lt 前缀，不支持 FHIR 标准的 eq/ge/le
• Bundle 缺 link 字段: 不符合 FHIR R4 规范
• 血压 observation ID 重复: 拆分收缩压/舒张压时 ID 重复

Mermaid

flowchart LR
  A[外部 HIS/LIS] --> B[OAuth Bearer Token]
  B --> C[FHIR Handler]
  C --> D[Converter 转换]
  D --> E[FHIR R4 Bundle JSON]

---

DF9: OAuth 授权 → 令牌 → API 调用

调用链

1. oauth/service.rs::create_client — 管理员创建 client (Argon2 哈希 secret)
2. POST /oauth/token — Client Credentials Grant
3. oauth/service.rs::verify_client — Argon2 验证 + scope 校验
4. JWT 签发 — 含 tenant_id + scope + allowed_patient_ids

发现的问题

• [高危] allowed_patient_ids 未在查询时强制执行: 第三方应用可能越权访问非授权患者数据
• [中危] JWT Secret 硬编码 fallback: "dev-secret-key" 生产环境可能被误用
• [中危] 速率限制仅存储未执行: rate_limit_per_minute 存在但无 middleware
• 仅支持 Client Credentials，无 Authorization Code/PKCE

Mermaid

flowchart LR
  A[管理员创建 Client] --> B[第三方 POST /oauth/token]
  B --> C[Argon2 验证]
  C --> D[JWT 签发]
  D --> E[FHIR API 调用]

---

DF10: 积分签到 → 事务 → 余额更新

调用链

1. points_handler::daily_checkin — 权限校验
2. points_service::daily_checkin — 事务内: 签到记录 + 积分规则 + 流水 + 余额 CAS 更新 + 阶梯奖励

发现的问题

• 无事件发布/通知推送: 签到成功后不发布 DomainEvent
• 连续天数计算脆弱: 仅查前一天记录，DB 写入失败导致连续天数断裂且无法恢复
• 阶梯奖励 CAS 冲突: 代码已通过重新查询规避

Mermaid

flowchart LR
  A[MP 签到] --> B[事务开始]
  B --> C[签到记录]
  C --> D[积分流水]
  D --> E[余额 CAS 更新]
  E --> F[阶梯奖励]

---

DF11: 随访完成 → 再分析触发 → 前后对比

调用链

1. follow_up_service::complete_task — 事务(PII 加密+记录+状态更新)
2. EventBus::publish(FOLLOW_UP_COMPLETED) — 发布完成事件
3. 事件消费 — 反查关联 AI 建议 → 发布 ai.reanalysis.requested
4. reanalysis.rs — 仅日志，对比功能未实现

发现的问题

• [功能缺失] 前后对比未实现: reanalysis.rs 注释写明"后续在 comparison.rs 中实现完整对比逻辑"
• baseline_snapshot 无写入入口: 代码读取了该字段但未找到首次 AI 分析时的写入逻辑
• [中危] 再分析触发无幂等保护: 同一事件重复消费可能多次触发 AI 分析

Mermaid

flowchart LR
  A[护士完成随访] --> B[FOLLOW_UP_COMPLETED]
  B --> C[反查 AI 建议]
  C --> D[ai.reanalysis.requested]
  D --> E{reanalysis.rs}
  E -->|现状| F[仅日志/TODO]
  E -->|目标| G[前后对比报告]

---

DF12: BLE 网关批量上传 → 多患者批量处理

调用链

1. POST /health/gateway/upload + API Key → gateway_auth_middleware(SHA-256 验证)
2. ble_gateway_service::process_gateway_upload — 逐患者 for 循环
3. device_reading_service::batch_create_readings — 验证+批量插入+双写+降采样+事件

发现的问题

• 串行处理多患者: for 循环逐个处理，延迟为所有患者处理时间之和
• 降采样效率低: 查出全量 hourly 记录再内存匹配，未按时间范围过滤
• [中危] 双写 vital_signs 失败静默忽略: 仅 warn 日志，可能导致数据不一致
• 心跳端点无审计日志

Mermaid

flowchart LR
  A[BLE 网关 POST] --> B[API Key 验证]
  B --> C[for 患者循环]
  C --> D[batch_create_readings]
  D --> E[insert_many]
  D --> F[双写 vital_signs]
  D --> G[降采样 hourly]