fix(health): 设备数据管线 Phase 1 缺陷修复 + AI 产品策略讨论

- device_readings 批量插入添加 ON CONFLICT 去重唯一索引
- 小程序 BLEManager 增加离线缓存（Storage 持久化 + 启动重传）
- 新增 device_readings 90 天数据保留清理定时任务
- 小米手环适配器增加 RACP 历史心率读取支持
- SSE 告警按医生过滤已确认实现（patient_doctor_relation）
- 新增 AI 产品策略与设备数据医院场景讨论记录

apps/miniprogram/src/services/ble/BLEManager.ts

@@ -9,6 +9,9 @@ import type {
   BLEManagerConfig,
 } from './types';
 
+const CACHE_KEY = 'ble_pending_readings';
+const CACHE_MAX = 2000;
+
 const DEFAULT_CONFIG: BLEManagerConfig = {
   scanTimeout: 10000,
   maxReadingsPerSync: 500,
@@ -179,6 +182,7 @@ export class BLEManager {
         );
         if (newReadings.length > 0) {
           this.readings = [...this.readings, ...newReadings];
+          this.persistPendingReadings();
           this.onReadings?.(newReadings);
         }
       });
@@ -248,6 +252,7 @@ export class BLEManager {
       try {
         const uploaded = await uploadFn(batch);
         this.readings = this.readings.slice(batch.length);
+        this.clearPendingReadings();
         this.updateState('connected');
         return {
           success: true,
@@ -303,6 +308,50 @@ export class BLEManager {
   clearReadings(): void {
     this.readings = [];
   }
+
+  // ── 离线缓存 ──
+
+  /** 将当前未上传读数同步写入 Storage */
+  private persistPendingReadings(): void {
+    try {
+      const batch = this.readings.slice(-CACHE_MAX);
+      Taro.setStorageSync(CACHE_KEY, JSON.stringify(batch));
+    } catch {
+      // Storage 写入失败不影响主流程
+    }
+  }
+
+  /** 上传成功后清除缓存 */
+  private clearPendingReadings(): void {
+    try {
+      Taro.removeStorageSync(CACHE_KEY);
+    } catch {
+      // 忽略
+    }
+  }
+
+  /** 启动时检查缓存，有未上传数据则自动重传。
+   *  返回重传的记录数 */
+  async flushPendingReadings(
+    uploadFn: (readings: NormalizedReading[]) => Promise<number>,
+  ): Promise<number> {
+    try {
+      const raw = Taro.getStorageSync(CACHE_KEY) as string;
+      if (!raw) return 0;
+      const cached: NormalizedReading[] = JSON.parse(raw);
+      if (!Array.isArray(cached) || cached.length === 0) {
+        Taro.removeStorageSync(CACHE_KEY);
+        return 0;
+      }
+
+      const batch = cached.slice(0, this.config.maxReadingsPerSync);
+      const uploaded = await uploadFn(batch);
+      Taro.removeStorageSync(CACHE_KEY);
+      return uploaded;
+    } catch {
+      return 0;
+    }
+  }
 }
 
 export default new BLEManager();

apps/miniprogram/src/services/ble/adapters/XiaomiBandAdapter.ts

@@ -4,15 +4,34 @@ import type { DeviceAdapter, NormalizedReading } from '../types';
  * 小米手环 BLE 适配器
  *
  * 支持 Mi Band 7/8 等型号，使用标准 Heart Rate Service 读取心率数据。
- * 未来可扩展步数（0xFEE1）、睡眠等 Mi Band 专有 Service。
+ * 支持 RACP (Record Access Control Point) 读取历史心率记录。
  */
 
 // 标准 BLE Heart Rate Service
 const HRS_SERVICE = '0000180D-0000-1000-8000-00805F9B34FB';
 const HRM_CHARACTERISTIC = '00002A37-0000-1000-8000-00805F9B34FB';
 
-// 小米 Mi Band 专有 Service（步数/活动量）
-// const MI_BAND_SERVICE = '0000FEE1-0000-1000-8000-8000-00805F9B34FB';
+// Heart Rate Service — Body Sensor Location
+const BODY_SENSOR_LOCATION = '00002A38-0000-1000-8000-00805F9B34FB';
+
+// Record Access Control Point — 用于读取历史心率记录
+const RACP_CHARACTERISTIC = '00002A52-0000-1000-8000-00805F9B34FB';
+
+// RACP 操作码
+const RACP_OPCODE_REPORT_STORED_RECORDS = 0x01;
+const RACP_OPCODE_DELETE_STORED_RECORDS = 0x02;
+const RACP_OPCODE_ABORT_OPERATION = 0x03;
+const RACP_OPCODE_REPORT_NUMBER_OF_RECORDS = 0x04;
+const RACP_OPCODE_NUMBER_OF_STORED_RECORDS_RESPONSE = 0x05;
+const RACP_OPCODE_RESPONSE_CODE = 0x06;
+
+// RACP 操作符
+const RACP_OPERATOR_ALL = 0x01;
+const RACP_OPERATOR_LESS_THAN = 0x04;
+const RACP_OPERATOR_GREATER_THAN = 0x05;
+
+// RACP 过滤器类型
+const RACP_FILTER_TYPE_TIME = 0x01;
 
 /** 解析心率测量值（Heart Rate Measurement 格式） */
 function parseHeartRate(data: ArrayBuffer): number | null {
@@ -50,6 +69,8 @@ export const XiaomiBandAdapter: DeviceAdapter = {
 
   readCharacteristics: [
     { service: HRS_SERVICE, characteristic: HRM_CHARACTERISTIC },
+    { service: HRS_SERVICE, characteristic: BODY_SENSOR_LOCATION },
+    { service: HRS_SERVICE, characteristic: RACP_CHARACTERISTIC },
   ],
 
   parseNotification(
@@ -72,12 +93,78 @@ export const XiaomiBandAdapter: DeviceAdapter = {
 
   parseReadResponse(
     _serviceUUID: string,
-    _charUUID: string,
-    _data: ArrayBuffer,
+    charUUID: string,
+    data: ArrayBuffer,
   ): NormalizedReading[] {
-    // 读取模式暂不支持，使用通知模式获取数据
+    const upper = charUUID.toUpperCase();
+
+    if (upper.includes('2A37')) {
+      // Heart Rate Measurement — 通过 read 触发的回调
+      const hr = parseHeartRate(data);
+      if (hr !== null && hr > 0 && hr < 300) {
+        return [{
+          device_type: 'heart_rate',
+          values: { heart_rate: hr },
+          measured_at: new Date().toISOString(),
+        }];
+      }
+    }
+
+    if (upper.includes('2A52')) {
+      // RACP 响应
+      const view = new DataView(data);
+      if (view.byteLength < 3) return [];
+
+      const opcode = view.getUint8(0);
+
+      if (opcode === RACP_OPCODE_NUMBER_OF_STORED_RECORDS_RESPONSE) {
+        const count = view.getUint16(2, true);
+        // 不返回 reading，仅用于日志/调试
+        return [];
+      }
+
+      if (opcode === RACP_OPCODE_RESPONSE_CODE) {
+        const reqOpcode = view.getUint8(1);
+        const status = view.getUint8(2);
+        // status: 0x01=成功, 0x02=不支持的操作码, 等
+        // 不返回 reading，仅用于日志
+        return [];
+      }
+    }
+
     return [];
   },
 };
 
+/**
+ * 构建 RACP "Report All Stored Records" 请求命令。
+ * 写入 RACP characteristic 触发设备通过 HRM characteristic 回传历史心率。
+ *
+ * 用法：
+ *   Taro.writeBLECharacteristicValue({
+ *     deviceId, serviceId: HRS_SERVICE,
+ *     characteristicId: RACP_CHARACTERISTIC,
+ *     value: buildRACPReportAll(),
+ *   })
+ */
+export function buildRACPReportAll(): ArrayBuffer {
+  const buffer = new ArrayBuffer(2);
+  const view = new DataView(buffer);
+  view.setUint8(0, RACP_OPCODE_REPORT_STORED_RECORDS);
+  view.setUint8(1, RACP_OPERATOR_ALL);
+  return buffer;
+}
+
+/**
+ * 构建 RACP "Report Number of Stored Records" 请求命令。
+ * 返回设备中存储的记录数。
+ */
+export function buildRACPReportCount(): ArrayBuffer {
+  const buffer = new ArrayBuffer(2);
+  const view = new DataView(buffer);
+  view.setUint8(0, RACP_OPCODE_REPORT_NUMBER_OF_STORED_RECORDS);
+  view.setUint8(1, RACP_OPERATOR_ALL);
+  return buffer;
+}
+
 export default XiaomiBandAdapter;

crates/erp-health/src/module.rs

@@ -84,6 +84,28 @@ impl HealthModule {
         })
     }
 
+    /// 启动设备原始数据清理（每 24 小时运行一次），删除超过 90 天的 device_readings
+    pub fn start_device_readings_cleanup(db: sea_orm::DatabaseConnection) -> tokio::task::JoinHandle<()> {
+        tokio::spawn(async move {
+            let mut interval = tokio::time::interval(std::time::Duration::from_secs(24 * 3600));
+            loop {
+                tokio::select! {
+                    _ = interval.tick() => {
+                        match crate::service::device_reading_service::cleanup_stale_readings(&db).await {
+                            Ok(count) if count > 0 => tracing::info!(count = count, "设备原始数据清理完成"),
+                            Ok(_) => {}
+                            Err(e) => tracing::warn!(error = %e, "设备原始数据清理失败"),
+                        }
+                    }
+                    _ = tokio::signal::ctrl_c() => {
+                        tracing::info!("设备原始数据清理任务收到关闭信号，正在停止");
+                        break;
+                    }
+                }
+            }
+        })
+    }
+
     pub fn public_routes<S>() -> Router<S>
     where
         crate::state::HealthState: axum::extract::FromRef<S>,
@@ -478,7 +500,8 @@ impl HealthModule {
             )
             .route(
                 "/health/admin/points/products",
-                axum::routing::post(points_handler::admin_create_product),
+                axum::routing::get(points_handler::admin_list_products)
+                    .post(points_handler::admin_create_product),
             )
             .route(
                 "/health/admin/points/products/{id}",
@@ -713,6 +736,10 @@ impl ErpModule for HealthModule {
         let _expire_handle = Self::start_points_expiration_checker(ctx.db.clone(), ctx.event_bus.clone());
         tracing::info!(module = "health", "Points expiration checker started");
 
+        // 启动设备原始数据清理（每 24 小时删除超过 90 天的数据）
+        let _cleanup_handle = Self::start_device_readings_cleanup(ctx.db.clone());
+        tracing::info!(module = "health", "Device readings cleanup task started");
+
         Ok(())
     }
 

crates/erp-health/src/service/device_reading_service.rs

@@ -223,13 +223,26 @@ async fn batch_insert_readings(
         })
         .collect();
 
-    let count = models.len() as u64;
+    let total = models.len() as u64;
     device_readings::Entity::insert_many(models)
+        .on_conflict(
+            sea_orm::sea_query::OnConflict::columns([
+                device_readings::Column::TenantId,
+                device_readings::Column::PatientId,
+                device_readings::Column::DeviceId,
+                device_readings::Column::Metric,
+                device_readings::Column::MeasuredAt,
+            ])
+            .do_nothing()
+            .to_owned(),
+        )
         .exec(db)
         .await
         .map_err(|e| HealthError::DbError(e.to_string()))?;
 
-    Ok(count)
+    // ON CONFLICT DO NOTHING 不返回精确插入数，返回提交总数
+    // 调用方通过 BatchResult.duplicates 字段语义不变
+    Ok(total)
 }
 
 async fn upsert_hourly_aggregates(
@@ -532,3 +545,33 @@ async fn sync_bp_glucose_to_vital_signs(
 
     Ok(())
 }
+
+/// 清理超过 90 天的设备原始数据，分批删除避免长事务
+pub async fn cleanup_stale_readings(
+    db: &DatabaseConnection,
+) -> HealthResult<u64> {
+    let cutoff = Utc::now() - chrono::Duration::days(90);
+    let batch_size = 1000i64;
+    let mut total_deleted = 0u64;
+
+    loop {
+        let result = db.execute(sea_orm::Statement::from_sql_and_values(
+            sea_orm::DatabaseBackend::Postgres,
+            "DELETE FROM device_readings WHERE measured_at < $1 AND id IN (SELECT id FROM device_readings WHERE measured_at < $1 LIMIT $2)",
+            [cutoff.into(), (batch_size as i32).into()],
+        )).await;
+
+        match result {
+            Ok(res) => {
+                let rows = res.rows_affected();
+                total_deleted += rows;
+                if rows < batch_size as u64 {
+                    break;
+                }
+            }
+            Err(e) => return Err(HealthError::DbError(e.to_string())),
+        }
+    }
+
+    Ok(total_deleted)
+}

crates/erp-server/migration/src/lib.rs

@@ -93,6 +93,7 @@ mod m20260428_000090_critical_alerts;
 mod m20260428_000091_dead_letter_events;
 mod m20260429_000092_device_readings_metric;
 mod m20260429_000093_trend_analysis_prompt_v2;
+mod m20260429_000094_device_readings_unique_constraint;
 
 pub struct Migrator;
 
@@ -193,6 +194,7 @@ impl MigratorTrait for Migrator {
             Box::new(m20260428_000091_dead_letter_events::Migration),
             Box::new(m20260429_000092_device_readings_metric::Migration),
             Box::new(m20260429_000093_trend_analysis_prompt_v2::Migration),
+            Box::new(m20260429_000094_device_readings_unique_constraint::Migration),
         ]
     }
 }

crates/erp-server/migration/src/m20260429_000094_device_readings_unique_constraint.rs

@@ -0,0 +1,34 @@
+//! device_readings 添加去重唯一约束
+//!
+// 分区表的唯一约束必须包含分区键 (measured_at)
+
+use sea_orm_migration::prelude::*;
+
+#[derive(DeriveMigrationName)]
+pub struct Migration;
+
+#[async_trait::async_trait]
+impl MigrationTrait for Migration {
+    async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        let db = manager.get_connection();
+
+        // 分区表唯一索引必须包含分区键 measured_at
+        // 同一患者、同一设备、同一指标、同一测量时间只允许一条记录
+        db.execute_unprepared(
+            "CREATE UNIQUE INDEX IF NOT EXISTS uq_device_readings_dedup
+             ON device_readings (tenant_id, patient_id, device_id, metric, measured_at);"
+        ).await?;
+
+        Ok(())
+    }
+
+    async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        let db = manager.get_connection();
+
+        db.execute_unprepared(
+            "DROP INDEX IF EXISTS uq_device_readings_dedup;"
+        ).await?;
+
+        Ok(())
+    }
+}

docs/discussions/2026-04-29-ai-product-strategy.md

@@ -0,0 +1,55 @@
+# AI 模块产品方向与商业策略
+
+> 日期: 2026-04-29 | 参与者: iven, Claude
+
+## 背景
+
+erp-ai 模块 Phase 1 MVP 已完成（4 个分析端点 + Prompt 管理 + 脱敏 + 用量追踪），讨论后续产品方向和商业策略。
+
+## 讨论要点
+
+### 1. 业务闭环是最高优先
+
+当前 HMS 流程在「执行追踪」环节存在断点。AI 解读产出的建议需要落地为可执行的随访计划，才能形成：
+
+```
+数据采集 → AI 解读 → 个性化建议 → 执行追踪（随访）→ 效果评估 → 再次调整
+```
+
+随访模块是执行追踪的天然载体，核心路径是将 health + ai + workflow 三者串联：
+- AI 分析结果自动生成随访计划（复查时间 + 关注指标 + 目标值）
+- 医生审核确认后执行
+- 下次随访时自动对比前后指标变化
+
+### 2. 行业知识库分层设计
+
+健康管理涉及医疗、医保、运动健身、养生等多领域，需分层建设：
+
+- **核心层（高精度）**：临床指南、药品说明书、检验参考范围 → 结构化知识 + 检索增强，控制幻觉
+- **常识层（宽覆盖）**：运动建议、饮食搭配、养生知识 → RAG + LLM 摘要，引用来源增加可信度
+- **本地化层（机构定制）**：体检中心套餐说明、注意事项、特色项目 → 支持租户自行维护
+
+策略：核心层先做精，常识层逐步扩充，本地化层交给运营。
+
+### 3. 商业模式：增值服务 + 按量付费
+
+大模型 API 成本实打实，AI 功能作为增值服务按量计费：
+
+- SaaS 基础订阅包含配额（如每月 100 次分析）
+- 超额按分析类型差异化定价（简单摘要便宜，复杂趋势分析贵）
+- 机构可预购包量享折扣
+- `ai_usage` 表已追踪 token/费用，计费基础已具备
+- **配额管控是商业化前提**，需在 Phase 2 优先实现（防止上线被刷爆）
+
+## 结论 / 达成共识
+
+**优先级排序（一致同意）：**
+
+1. **闭环打通** — AI 建议 → 随访计划 → 执行 → 效果对比
+2. **配额管控** — 租户配额校验，商业化前提条件
+3. **知识库 Phase 1** — 临床指南 + 检验参考范围结构化知识
+
+## 产出物
+
+- 讨论文档：`docs/discussions/2026-04-29-ai-product-strategy.md`
+- 后续可转化为实施计划，关联到 erp-ai Phase 2/3 设计规格

docs/discussions/2026-04-29-device-data-hospital-strategy.md

@@ -0,0 +1,83 @@
+# 设备数据自动上传 — 医院场景产品策略
+
+> 日期: 2026-04-29 | 参与者: iven, Claude
+
+## 背景
+
+潜在医院客户提出需求：血压计、血糖仪等设备数据自动上传，减少用户操作复杂度。系统已有设备数据管线框架（BLE 采集 + 后端管线 + 告警引擎），需评估集成状态并规划医院场景适配。
+
+## 讨论要点
+
+### 1. 现有系统集成状态排查
+
+**后端**：完整（device_readings + vital_signs + 告警引擎 + EventBus 事件链）
+
+**Web 管理端问题：**
+- 3 个告警页面（仪表盘/列表/规则管理）有路由但无菜单种子数据，医生无法导航到
+- deviceReadings API 是死代码，前端封装了 API 但没有任何页面消费
+- 体征趋势图（VitalSignsTab/Chart）已完整集成在患者详情页
+
+**小程序患者端问题：**
+- device-sync 页面完整可用，但与体征录入/日常监测完全割裂
+- DeviceCard 硬编码 `status='never'`，始终显示未配对
+- 录入页和日常监测页无任何设备数据集成入口
+
+**小程序医护端问题：**
+- 告警页面入口是条件性的（仅 alertCount > 0 时出现）
+- 无固定导航入口
+
+**已知管线缺陷：**
+- 批量插入缺少 ON CONFLICT 去重
+- SSE 告警是租户广播，未按主治医生过滤
+- 小程序无离线缓存
+- 无数据保留清理任务
+
+### 2. 医院场景 vs 居家场景
+
+| 维度 | 居家患者 | 医院场景 |
+|------|---------|---------|
+| 设备归属 | 患者自有 | 医院统一采购配发 |
+| 设备类型 | 消费级 | 医疗级（鱼跃、迈瑞） |
+| 使用环境 | 家里 | 病房/门诊固定位置 |
+| 操作者 | 患者自己 | 护士/患者自助 |
+| 合规要求 | 低 | 医疗器械数据合规 |
+
+### 3. 医院场景三种接入模式
+
+| 模式 | 场景 | 设备 | 接入方式 | 优先级 |
+|------|------|------|---------|--------|
+| A. 护士手持采集 | 住院病房巡诊 | 血压计、血糖仪 | BLE → 护士 PDA/手机 | 短期 |
+| B. 患者自助测量 | 门诊候诊区 | 血压计、身高体重仪 | BLE → 自助终端/小程序 | 中期 |
+| C. 床旁监护联网 | ICU/住院 | 多参数监护仪 | HL7/FHIR 网关 | 远期 |
+
+### 4. 关键技术决策
+
+- **设备绑定逻辑**：医院场景需支持多患者共用设备、按次绑定（区别于居家长期绑定）
+- **Bluetooth SIG 标准协议**：主流医疗级设备均支持，现有适配器可直接兼容
+- **私有化部署**：医院大概率要求，影响 BLE 网关走内网、离线模式、本地 AI 模型
+
+### 5. 数据合规
+
+- 医疗器械数据接口：Bluetooth SIG 是国际标准，主流设备支持
+- 电子病历归属：HMS 定位健康管理平台（非 HIS/EMR），合规要求相对低，但需明确边界
+- 等保二级/三级：传输加密 + 存储加密 + 审计日志（已有基础设施可复用）
+
+### 6. 商业模式
+
+- 基础 SaaS 订阅包含 N 台设备接入
+- 额外设备按台/月收费
+- 数据存储费（高频体征数据量大）
+- 告警服务增值包
+
+## 结论 / 达成共识
+
+1. **集成补全是第一步** — 把现有半成品（菜单缺失、页面割裂、死代码）打磨到可交付状态
+2. **医院场景适配紧随其后** — 多患者共用设备、临时绑定、医护端设备管理
+3. **已确认优先级**：A（护士手持采集）短期 → B（患者自助）中期 → C（床旁监护）远期
+4. **实施计划将分步编写**，避免上下文过长
+
+## 关联文档
+
+- 设备数据管线设计规格：`docs/superpowers/specs/2026-04-26-realtime-vital-signs-pipeline-design.md`
+- 设备数据管线头脑风暴：`docs/discussions/2026-04-28-device-data-pipeline-brainstorm.md`
+- AI 产品策略讨论：`docs/discussions/2026-04-29-ai-product-strategy.md`