Skip to content

P
pac
Commit f3d7ae77 by luoqi
Options

merge: feat/friday-ingestion → main(FRIDAY 摄入契约收敛 + 推送自助反馈) 

- 摄入契约收敛为 9 张自洽表(宿主 inline 自己的引用,全路径一致)
- 结算 WHERE 迁移单一真理源 + 冷导入网页通道 + 数据对账
- push 预检(dryRun)+ 推送记录查询(HMAC 免登录)
- 修 traceRawSourceTable union 回溯(refund_full 在 push/reparse 静默跳过)

阶段性完成,业务确认可合。
parents a91f8a37 ec026cb2
Showing with 1320 additions and 47 deletions
apps/pac-docs/content/docs/data-reconciliation.mdx 0 → 100644
---
title: 数据对账(设计)
description: 独立核对 PAC 侧数据与各宿主源是否一致 —— 同闸口径、去重患者数、多宿主 provider、复用企微日报。
icon: Scale
---
<Callout type="warn">
**本文是设计文档,尚未实现**。记录对账机制的口径与方案,供评审 / 落地参照。
</Callout>
## 为什么要独立对账
现有的几层防线都是**单向补漏 / 水位监控**,抓不到"该有的数据到底在不在":
| 现有机制 | 作用 | 抓不到 |
|---|---|---|
| 48h 回看窗 | 容忍 DW 落库迟到 | 超 48h / 时间戳不跳的漏拉 |
| 主档反向强拉 | 补 cohort 的主档 | 非主档信号的积压 |
| ReconcileOrchestrator | 时间窗重拉补漏(手动) | 不告诉你"差了多少" |
| DwLagMonitor | cursor 距今 >24/48h 告警 | 只看水位新旧,不看行在不在 |
尤其前述三个已知漏点 —— [无 cursor 表积压](/docs/ingestion#21-增量-cohort-与无-cursor-表)、`updatedAt` 不跳漏拉、事务/fact 不一致 —— **都是补漏和水位监控抓不到的**。缺的是一个**双向数量核对**:定期比"源侧应有 vs PAC 实有",差超阈值告警。
## 三个难点与解法
对账看似简单(两边 count 一比),实则有三个坑,逐个解:
### 难点 1:多宿主 —— 源计数从哪来因宿主而异
PAC 是多宿主,不能绑死 DW —— 但对账**只覆盖 PAC 能直连源库的宿主**:
| 宿主类型 | 例 | 能直连源? | 对账 |
|---|---|---|---|
| **pull**(直连) | jvs-dw(ClickHouse) | ✅ | ✅ PAC 主动查源计数 |
| **push**(无直连) | friday(SaaS) | ❌ PAC 够不到其库 | **不对账** |
**push 宿主不对账**(定):PAC 够不到源库,唯一办法是宿主自报,但那要它维护一套"按 PAC 口径过滤后的计数",负担与可信度都不划算 —— 直接排除。push 宿主的数据质量靠既有的 push 路径告警(疑似改表 / 失败率超阈)兜。
抽象成:**PAC 只负责「拿到 `(scope, sourceCount)` → 对比 `pacCount` → 报 diff」;`sourceCount` 由每个可直连宿主的 provider 供给**(查询源库)。将来任何可直连的新宿主接入只需实现 provider,对账主逻辑不动。
### 难点 2:摄入是筛选子集 —— 不能数原表全量
PAC 摄入的**不是原表全量**,有大量故意筛选(结算 `status IN (1,3)`、瑞泰不摄、cohort 限有就诊、诊所范围…)。拿原表 count 比,diff 里全是"故意的差",没法用。
**解法:源计数复用摄入查询本身,只换投影。** manifest 的 `sql_source` 查询**就是"PAC 应该有什么"的定义**,把 `SELECT *` 换成 `count(distinct patient)`:
```sql
sourceCount = count(distinct patient) FROM ( «摄入用的那条 sql_source 查询,原样» )
```
两边都过了同一道筛选闸,diff 只剩**真漂移**。且筛选口径**零重复维护** —— 改了 `sql_source` 的 WHERE,对账自动跟着变(用的同一条查询),不会两份逻辑慢慢跑偏。
(push 宿主不对账,见难点 1,不涉及此。)
### 难点 3:fan-out 噪声 —— 不能比行数
一条源行可能拆成 N 条 fact(诊断+治疗+建议),加上版本流,行数根本不可比。
**解法:只比「去重患者数」,不比行数。** 干净、可比、且直接对应业务关切("多少患者的 X 信号 PAC 没跟上")。
## 对账口径:同闸 + 去重患者数 + 两层
只看总患者数**不够** —— 无 cursor 表积压时患者还在 PAC 里(只是他的回访/咨询行旧了),总量层 diff=0 会漏检。故分两层:
| 层 | sourceCount | pacCount | 抓什么 |
|---|---|---|---|
| **总量** | cohort 查询的 `count(distinct patient)` | PAC 该 host 的 `count(distinct patient)` | 整体漏人 |
| **信号** | 各资源 sql_source(含各自过滤)的 `count(distinct patient)` | PAC 里有对应 `fact_type` 的 `count(distinct patient)` | 某类信号积压(定位到无 cursor 表) |
**信号层示例**:`DW 有影像分析的患者 5.0 万 vs PAC 有 image_record 的患者 4.8 万,diff 2000` → 既可比又定位到"影像信号积压"。
**覆盖范围:全 fact 类型**(定)。不挑重点,每类都对(诊断 / 治疗 / 影像 / 回访 / 咨询 / 结算 / 转介…),一眼看全哪条信号在漂;性能由下节"每日一次 + 秒级 count"兜住,全覆盖不构成负担。
## 性能
对账要跑一堆 `count(distinct patient)`,得算清楚成本再上:
- **频率:每日一次**(不是每轮增量)。跟健康日报同一时刻(09:00 沪、DW 刷新后)跑一次即可 —— 漂移是慢变量,日粒度足够。
- **DW 侧**:每类信号一条 `count(distinct patient)`,全覆盖约 10 条。ClickHouse 的 `count(distinct)` 对 13M 行通常秒级;精度不敏感处可用 `uniq`(HyperLogLog,~0.5% 误差)换更低开销,要精确用 `uniqExact`。**上线前实测这 10 条的总耗时**,确认 < 分钟级。
- **PAC 侧**:`patient_facts` 按 `fact_type` 分组 `count(distinct patient_id)`,一条走索引的聚合查询,廉价。
- **不常驻**:对账是日报里的一个步骤,跑完即释放,无常驻连接 / 内存。
每日一次 + 秒级查询 → 全覆盖的性能开销可忽略,不影响在线摄入 / 重算。
## 基线:压掉"已知的正常差"
有些差**本来就该存在**,不压会天天误报 → 告警疲劳:
- 瑞泰品牌不摄(恒差)
- 无 cursor 表短期积压(会自愈)
- 源侧脏数据 / 时间窗边界
**基线放 manifest 配置**(每宿主声明"哪些 scope 差多少以内算正常"),只有**偏离基线**才告警,不是有差就报。
```yaml
# data/<host>/manifest.yaml(拟)
reconcile:
baseline:
total_patient: { tolerance_pct: 2 } # 总量差 2% 内正常
fact.image_record: { tolerance_pct: 5 } # 影像积压容忍高些
fact.payment_record: { tolerance_abs: 200 }
```
## 汇报:复用企微,折进每日健康日报
**不新起消息流** —— 复用 `AlertService`(企微群机器人),把「数据对账」折进已有的 [每日健康日报](/docs/monitoring)(每天 09:00 沪、DW 刷新后跑)。对账项超基线时,把日报整体级别顶成 🟡/🔴(现有逻辑:取最差项)。
```
每日 09:00 健康日报(已有)增加一段「数据对账」:
for each host:
for each scope in [总量患者, 各 fact 类型患者]:
sourceCount = provider(host, scope) # pull 查询 / push 自报
pacCount = PAC 同口径 count(distinct patient)
diff, diffPct
超 manifest.reconcile.baseline → 标 🟡/🔴
推企微(复用 AlertService,超基线才顶级别)
```
## 分期落地
**先只观察企微,不建表**(定)。
| 版本 | 做什么 | 不做 |
|---|---|---|
| **v1(先做)** | 同闸口径算 diff(全 fact 类型)+ 基线压噪 + 折进健康日报推企微 | **不建表**;趋势靠每日日报肉眼比 |
| v2(以后) | 若肉眼比不够,再建 `reconcile_checks` 表存快照 → 自动趋势 + 审计历史 | — |
先跑 v1 观察企微:口径准不准、会不会噪声、diff 稳不稳。够用就不上表。
## 已定 / 待决
**已定**:push 宿主(无直连)不对账 · 全 fact 类型覆盖 · 每日一次跑 · 先不建表只观察企微。
**待决**:基线初值怎么定 —— 建议**先不设容忍、原样把 diff 报进日报观察一两周**,摸清各 scope 的"正常差"后再回填 `manifest.reconcile.baseline`,避免拍脑袋定阈值。
apps/pac-docs/content/docs/ingestion.mdx
...@@ -93,6 +93,46 @@ pnpm sync -- --dir=./data/<host> --dry-run # 不写库,预览游标注入 ...@@ -93,6 +93,46 @@ pnpm sync -- --dir=./data/<host> --dry-run # 不写库,预览游标注入
pnpm sync -- --dir=./data/<host> --full # 忽略游标强制全量(灾后) pnpm sync -- --dir=./data/<host> --full # 忽略游标强制全量(灾后)
``` ```
### 2.1 增量 cohort 与「无 cursor 表」
增量分两步,理解成本要分开看:
1. **选 cohort**:`manifest.incremental.per_query` 声明的表,各按自己 `cursor_column`
取 `WHERE cursor > 水位` 的患者,**并集**成本轮 cohort(UNION —— 任一表有变更就纳入,
避免"预约变了但人没来"被漏)。
2. **拉资源**:对 cohort 患者,**所有**源表按 `(patient_id, brand) IN (cohort)` 拉行。
没进 `per_query` 的表(无 cursor)也在此步被拉 —— **按患者收窄,不全表扫**。
**关键推论**:无 cursor 表**不能独立把患者拉进增量**。若某患者只有无 cursor 表变了
(且 cursor 表都没动)→ 不进 cohort → 该变更**积压到该患者下次因 cursor 表再进 cohort 才补上**。
所以无 cursor 表适合**弱信号 / 展示类**(容忍延迟),不适合驱动召回的核心信号。
以 **jvs-dw(瑞尔)** 为例(2026-07 DW 实测行数):
| 源表 | 行数 | cursor | CH 排序键 | 说明 |
|---|---:|---|---|---|
| fact_client_out | 5.7M | `last_visit_time` | patient_id | 患者主档 |
| fact_emr_treatment_out | 5.1M | `updated_date` | patient_id… | 治疗/病历(召回核心) |
| fact_appointment_out | 9.1M | `updated_date` | id | 预约 |
| fact_settlement_out | 9.1M | `updated_date` | id | 结算 |
| fact_settlement_mode_out | 4.4M | `updated_date` | — | 支付通道 |
| **fact_returnvisit_out** | **13.4M** | **无** | 无 | 回访任务(展示) |
| **fact_consult_out** | **10.8M** | **无** | org… | 咨询(意向,弱信号) |
| fact_emr_image_analysis_out | 553K | 无 | emr_id | 影像 AI |
| fact_customer_referee_out | 499K | 无 | 无 | 转介 |
<Callout type="info">
**提频(如 2h 一轮)的成本**——由**空轮短路**决定,不是 cohort 大小:
- **cohort 选择每轮都跑**(含空轮):cursor 列(`updated_date` 等)通常**不是 CH 排序键
→ 无索引 → 每轮全列扫**。但只扫单列,9M 行亚秒级,几张合计几秒/轮,绝对量可忽略。
- **资源拉取只在 cohort 非空时跑**:上游多为日更 → 日内多数轮 cohort 为空 →
在选 cohort 后即**短路**(跳过资源拉取 + persona/plan 重算),不触碰上面两张最大的无 cursor 表。
- 所以提频主要增加的是**廉价的空轮探查**,重活(资源全表扫 + 重算)仍是刷新后 1–2 轮,不被放大。
真要压这点空轮扫描,治本是给 cursor 表的 `updated_date` 加 CH 数据跳过索引;收益很小,通常不值当。
</Callout>
--- ---
## 三、数据完整性 ## 三、数据完整性
...@@ -102,7 +142,15 @@ pnpm sync -- --dir=./data/<host> --full # 忽略游标强制全量(灾 ...@@ -102,7 +142,15 @@ pnpm sync -- --dir=./data/<host> --full # 忽略游标强制全量(灾
- `patient_transactions`:partial UNIQUE `(host_id, tenant_id, source_event_id) WHERE source_event_id NOT NULL` - `patient_transactions`:partial UNIQUE `(host_id, tenant_id, source_event_id) WHERE source_event_id NOT NULL`
- `patient_facts`:UNIQUE `(host_id, tenant_id, subject_id, version)` + active 行 partial UNIQUE - `patient_facts`:UNIQUE `(host_id, tenant_id, subject_id, version)` + active 行 partial UNIQUE
`source_event_id` 内含 `updatedAt`,因此源数据行级 in-place 更新会自然产生新 tx 与新 fact 版本(旧版 supersede);同行同 `updatedAt` 则幂等跳过。 `source_event_id` 内含 `updatedAt`(缺失时确定性回退 `createdAt`),因此源数据行级 in-place 更新会自然产生新 tx 与新 fact 版本(旧版 supersede);同行同 `updatedAt` 则幂等跳过。fact 层的变更判定是**内容驱动**的(比对 content hash + 状态 + 时间锚,不看 `updatedAt`),所以即使事务被幂等跳过,只要行被重新处理,内容真变了 fact 仍会升版本——变更不会因时间戳丢失而漏掉。
<Callout type="warn">
**`updatedAt` 维护缺失的两个后果**(如某些源表主档 `updated_gmt_at` 大量为 null)。当前设计有意保留 `updatedAt` 在幂等键中,**不改**;接入时按下面两点评估:
1. **pull/增量取数够不到**：增量按 `WHERE updated_date > cursor` 拉数,**内容变了但 `updatedAt` 没随之更新的行根本不会被选出来**——这道过滤在幂等之前,fact 层内容比对再准也无行可比。file 全量装载 / push(宿主自选推送)不受此限(不按 cursor 过滤)。治本靠源表维护 `updatedAt`;主档另有定期全量反拉兜底(见上「主档补全」)。
2. **事务与 fact 可能不一致 → reparse 会打回**:若内容变了但 `updatedAt` 没变,事务因 `source_event_id` 未变被幂等跳过(`rawPayload` 停在旧内容),但 fact 层仍按新内容升版本 —— 二者内容不再一致。此后 [reparse](#52-reparse--字典改了补存量无需全量重摄) 从 `transaction.rawPayload`(旧值)离线重放,会把 fact 打回旧内容。**规避**:对 `updatedAt` 不可靠的源表,不要依赖 reparse 修其存量,改走 DW 重摄 / 重推。
</Callout>
**主档补全(两条保障,确保 patient_id 不丢)**: **主档补全(两条保障,确保 patient_id 不丢)**:
...@@ -148,6 +196,7 @@ pnpm sync -- --dir=./data/<host> --full # 忽略游标强制全量(灾 ...@@ -148,6 +196,7 @@ pnpm sync -- --dir=./data/<host> --full # 忽略游标强制全量(灾
- **不连数据源**:rawPayload 在本地,比全量重摄快一个量级; - **不连数据源**:rawPayload 在本地,比全量重摄快一个量级;
- **可圈定**:按 host / subject-type / 患者集 / dry-run / 是否重算; - **可圈定**:按 host / subject-type / 患者集 / dry-run / 是否重算;
- **覆盖范围**:field/enum/keyword_mapping、transforms 算子、parser 改动;**不覆盖**改了 `sql_source` 的 SELECT 本身(需从数据源多拉列/表,那才需真重摄)。非 transform 产出的资源(如 SQL 视图类)会自动跳过。 - **覆盖范围**:field/enum/keyword_mapping、transforms 算子、parser 改动;**不覆盖**改了 `sql_source` 的 SELECT 本身(需从数据源多拉列/表,那才需真重摄)。非 transform 产出的资源(如 SQL 视图类)会自动跳过。
- **前提:rawPayload 是最新的**。reparse 只能和源行的最后一次落库一样新。若某源表 `updatedAt` 维护缺失,曾出现「内容变了但事务被幂等跳过」(见 §三 Callout),其 `rawPayload` 已滞后 —— 对这类表 reparse 会把 fact 打回旧内容,应改走重摄 / 重推而非 reparse。
```bash ```bash
# dry-run:报 scope(患者数 / txn 数),不写库 # dry-run:报 scope(患者数 / txn 数),不写库
......
apps/pac-docs/content/docs/meta.json
...@@ -14,6 +14,7 @@ ...@@ -14,6 +14,7 @@
"deployment", "deployment",
"deploy-runbook", "deploy-runbook",
"ingestion", "ingestion",
"data-reconciliation",
"monitoring", "monitoring",
"troubleshooting", "troubleshooting",
"---参考---", "---参考---",
......
apps/pac-docs/openapi/pac.json
This source diff could not be displayed because it is too large. You can view the blob instead.
apps/pac-service/data/friday/assemblers/diagnosis.yaml
# diagnosis — FRIDAY 诊断 fact(med_emr_info.diag[] 拆行) # diagnosis — FRIDAY 诊断 fact(med_emr_info.diag[] 拆行)
# 消费 transforms 输出表 diagnosis_rows(split + std_diag lookup + normalize/substring/coalesce) # 消费 transforms 输出表 diagnosis_rows(split + normalize/substring/coalesce;std_code 由宿主 inline 进 diag 元素)
# #
# code 解析规则(用户定,2026-07-18): # code 解析规则(用户定,2026-07-18):
# linkCode 非空 → std_diag.std_code K 开头 → 截 K 大类(std_code_k 直通下方共享字典 passthrough); # linkCode 非空 → std_diag.std_code K 开头 → 截 K 大类(std_code_k 直通下方共享字典 passthrough);
......
apps/pac-service/data/friday/assemblers/emr.yaml
...@@ -10,7 +10,7 @@ emits: ...@@ -10,7 +10,7 @@ emits:
occurredAtField: submittedAt occurredAtField: submittedAt
primary: primary:
table: med_emr_info table: _emr_official # transforms 产出:med_emr_info status∈{3,4}(排草稿;迁自导出侧过滤)
key: emr_sub_id # 小病历号 uuid(两侧连接键 + 临床事实锚点) key: emr_sub_id # 小病历号 uuid(两侧连接键 + 临床事实锚点)
dedup_by: emr_sub_id dedup_by: emr_sub_id
......
apps/pac-service/data/friday/assemblers/image.yaml
# image — FRIDAY 影像 metadata(MySQL emr.med_check;PAC 不持文件本体) # image — FRIDAY 影像 metadata(MySQL emr.med_check;PAC 不持文件本体)
# 影像不在 Mongo 病历文档内(全字段枚举证实),挂 emr_id(10 位病历号)回病历。 # 影像不在 Mongo 病历文档内(全字段枚举证实),挂 emr_id(10 位病历号)回病历。
# 消费 transforms 输出表 image_rows(med_check lookup std_check_class 附 class_name) # 直接消费 med_check(class_name 由宿主 inline:class_code→std_check_class 已解析);modality 经共享字典翻
canonical: image canonical: image
emits: emits:
...@@ -9,7 +9,7 @@ emits: ...@@ -9,7 +9,7 @@ emits:
occurredAtField: uploadedAt occurredAtField: uploadedAt
primary: primary:
table: image_rows table: med_check
key: id key: id
dedup_by: id dedup_by: id
...@@ -21,7 +21,7 @@ field_mapping: ...@@ -21,7 +21,7 @@ field_mapping:
clinicId: organization_id clinicId: organization_id
uploadedAt: shooting_time # 拍摄时间 uploadedAt: shooting_time # 拍摄时间
encounterExternalId: emr_id # 病历号级挂载(= med_emr_info.emr_id,非 emr_sub_id) encounterExternalId: emr_id # 病历号级挂载(= med_emr_info.emr_id,非 emr_sub_id)
modality: class_name # lookup 自 std_check_class 字典 modality: class_name # 宿主 inline(class_code→std_check_class 已解析)→ 共享字典翻
fileUrl: file_url # OSS 相对路径(passthrough) fileUrl: file_url # OSS 相对路径(passthrough)
fileName: file_name fileName: file_name
......
apps/pac-service/data/friday/assemblers/patient.yaml
# patient — 主档 upsert(不进 transaction,无 emits) # patient — 主档 upsert(不进 transaction,无 emits)
# 源:customer.customer_basic_info(原始表);phone 由 manifest.transforms.lookup # 源:customer.customer_basic_info;contacts_tel / relationship 由宿主导出时挑默认号 inline 进主档
# 从 customer_contacts 挑默认号附加进来,故这里直接映射标量 phone。 # (不再 PAC 侧 lookup customer_contacts),故这里直接映射标量。
canonical: patient canonical: patient
primary: primary:
...@@ -10,12 +10,12 @@ primary: ...@@ -10,12 +10,12 @@ primary:
field_mapping: field_mapping:
externalId: id # customer_basic_info 主键 = host 内部患者唯一 id(→ String 化) externalId: id # customer_basic_info 主键 = host 内部患者唯一 id(→ String 化)
name: name name: name
phone: phone # 由 transforms.lookup 从 customer_contacts 挑默认号附加 phone: contacts_tel # 宿主 inline 的默认号(沿用 customer_contacts 原列名)
# 该号关系码(lookup 同源带出)→ preferences.contactPhone{relationshipCode, relationship, isSelf}。 # 该号关系码(lookup 同源带出)→ preferences.contactPhone{relationshipCode, relationship, isSelf}。
# 官方枚举(源码实锤 friday-saas/customer …/enums/PhoneRelationshipEnum.java,2026-07-20): # 官方枚举(源码实锤 friday-saas/customer …/enums/PhoneRelationshipEnum.java,2026-07-20):
# 1本人 2爸爸 3妈妈 4爷爷 5奶奶 6朋友 7配偶 8子女 9其他(语义=持号人是患者的X) # 1本人 2爸爸 3妈妈 4爷爷 5奶奶 6朋友 7配偶 8子女 9其他(语义=持号人是患者的X)
phoneRelationshipCode: phone_relationship phoneRelationshipCode: relationship
phoneRelationship: phone_relationship # 同源列解码成 PAC 关系词(下方 enum_mapping) phoneRelationship: relationship # 同源列解码成 PAC 关系词(下方 enum_mapping)
gender: sex # tinyint 1/2 → enum_mapping 归一;0/空 → 空 gender: sex # tinyint 1/2 → enum_mapping 归一;0/空 → 空
birthDate: birthday # date;空值入库为 null birthDate: birthday # date;空值入库为 null
medicalRecordNumber: file_number # 档案号(病历号,客服沟通用) medicalRecordNumber: file_number # 档案号(病历号,客服沟通用)
......
apps/pac-service/data/friday/assemblers/payment.yaml
...@@ -26,7 +26,8 @@ field_mapping: ...@@ -26,7 +26,8 @@ field_mapping:
# 应收降级 receivableAmount(折扣率分母/权益客识别)。 # 应收降级 receivableAmount(折扣率分母/权益客识别)。
amount: net_receipts_this amount: net_receipts_this
receivableAmount: receivable_this receivableAmount: receivable_this
method: method # lookup 主导支付通道(现金/微信/刷卡/医保…原样入 fact) # method 暂不接:支付通道明细(settlement_modes)不摄入,金额取结算头 net_receipts_this。
# 将来要"支付方式分析"(医保占比/分次支付)再摄入 settlement_modes、恢复 method。
doctorId: doctor_id doctorId: doctor_id
encounterExternalId: registration_id # 关联接诊 encounterExternalId: registration_id # 关联接诊
settlementSerialNum: settlement_serial_num # 结算流水号(passthrough,对账用) settlementSerialNum: settlement_serial_num # 结算流水号(passthrough,对账用)
apps/pac-service/data/friday/assemblers/refund_full.yaml
...@@ -13,7 +13,7 @@ emits: ...@@ -13,7 +13,7 @@ emits:
kind: actual kind: actual
primary: primary:
table: patient_settlement_refund table: refund_full_rows # transforms 产出:status=4 ∪ (status=3 且 receivable<0)
key: uuid key: uuid
dedup_by: uuid dedup_by: uuid
......
apps/pac-service/data/friday/assemblers/refund_item.yaml
...@@ -11,7 +11,7 @@ emits: ...@@ -11,7 +11,7 @@ emits:
kind: actual kind: actual
primary: primary:
table: patient_settlement_spec_refund table: refund_item_rows # transforms 产出:patient_settlement_spec 里 is_refund=1
key: id key: id
dedup_by: id dedup_by: id
...@@ -19,6 +19,7 @@ field_mapping: ...@@ -19,6 +19,7 @@ field_mapping:
externalId: id externalId: id
updatedAt: updated_gmt_at updatedAt: updated_gmt_at
createdAt: created_gmt_at createdAt: created_gmt_at
# spec.patient_id 部分品牌原是诊所本地 id → 宿主导出时已从结算头反填真 patient_id(inline);PAC 直接用
patientExternalId: patient_id patientExternalId: patient_id
clinicId: organization_id clinicId: organization_id
refundedAt: created_gmt_at refundedAt: created_gmt_at
......
apps/pac-service/data/friday/assemblers/treatment_planned.yaml
# treatment_planned — FRIDAY 治疗计划(customer_treat_plan + _item 独立表;用户确认此表为治疗计划) # treatment_planned — FRIDAY 治疗计划(消费 customer_treat_plan_item;头属性 organization_id/plan_name 由宿主 inline)
# 消费 transforms 输出表 treatment_planned_rows(行表 lookup 头表拿 organization_id/plan_name) # 消费 transforms 输出表 treatment_planned_rows(item 过滤无诊所行 + normalize 术式名)
# 行粒度 = 计划行项(牙位 + 术式 mode_name + 价格区间);externalId = 行表主键(host 稳定) # 行粒度 = 计划行项(牙位 + 术式 mode_name + 价格区间);externalId = 行表主键(host 稳定)
canonical: treatment canonical: treatment
...@@ -19,7 +19,7 @@ field_mapping: ...@@ -19,7 +19,7 @@ field_mapping:
updatedAt: updated_gmt_at updatedAt: updated_gmt_at
createdAt: created_gmt_at createdAt: created_gmt_at
patientExternalId: customer_id patientExternalId: customer_id
clinicId: organization_id # lookup 自头表(行表无诊所) clinicId: organization_id # 宿主 inline 自计划头(行表原生无诊所)
occurredAt: created_gmt_at # 计划制定时点 occurredAt: created_gmt_at # 计划制定时点
category: mode_norm # 归一术式名 → 共享字典翻 11 类闭集 category: mode_norm # 归一术式名 → 共享字典翻 11 类闭集
subtype: mode_name # 原始术式名 subtype: mode_name # 原始术式名
......
apps/pac-service/package.json
...@@ -90,6 +90,7 @@ ...@@ -90,6 +90,7 @@
"socket.io": "^4.8.3", "socket.io": "^4.8.3",
"winston": "^3.19.0", "winston": "^3.19.0",
"ws": "^8.21.0", "ws": "^8.21.0",
"yauzl": "^3.3.0",
"zod": "^4.4.3" "zod": "^4.4.3"
}, },
"devDependencies": { "devDependencies": {
......
apps/pac-service/src/modules/admin/admin.module.ts
...@@ -7,6 +7,7 @@ import { MappingMissAdminController } from './mapping-miss.controller'; ...@@ -7,6 +7,7 @@ import { MappingMissAdminController } from './mapping-miss.controller';
import { HostsService } from './hosts.service'; import { HostsService } from './hosts.service';
import { AiInvocationsService } from './ai-invocations.service'; import { AiInvocationsService } from './ai-invocations.service';
import { MappingMissService } from './mapping-miss.service'; import { MappingMissService } from './mapping-miss.service';
import { ImportUploadService } from './import-upload.service';
@Module({ @Module({
controllers: [ controllers: [
...@@ -14,7 +15,7 @@ import { MappingMissService } from './mapping-miss.service'; ...@@ -14,7 +15,7 @@ import { MappingMissService } from './mapping-miss.service';
AiInvocationsAdminController, AiInvocationsAdminController,
MappingMissAdminController, MappingMissAdminController,
], ],
providers: [HostsService, AiInvocationsService, MappingMissService], providers: [HostsService, AiInvocationsService, MappingMissService, ImportUploadService],
exports: [HostsService, AiInvocationsService], exports: [HostsService, AiInvocationsService],
}) })
export class AdminModule {} export class AdminModule {}
apps/pac-service/src/modules/admin/dto/admin.dto.ts
import { createZodDto } from 'nestjs-zod'; import { createZodDto } from 'nestjs-zod';
import { z } from 'zod';
import { import {
ListHostsResponseSchema, ListHostsResponseSchema,
HostSummarySchema, HostSummarySchema,
...@@ -32,3 +33,35 @@ export class AiInvocationListQueryDto extends createZodDto(AiInvocationListQuery ...@@ -32,3 +33,35 @@ export class AiInvocationListQueryDto extends createZodDto(AiInvocationListQuery
export class AiInvocationDetailDto extends createZodDto(AiInvocationDetailSchema) {} export class AiInvocationDetailDto extends createZodDto(AiInvocationDetailSchema) {}
export class AiInvocationStatsDto extends createZodDto(AiInvocationStatsSchema) {} export class AiInvocationStatsDto extends createZodDto(AiInvocationStatsSchema) {}
export class ListAiInvocationsResponseDto extends createZodDto(ListAiInvocationsResponseSchema) {} export class ListAiInvocationsResponseDto extends createZodDto(ListAiInvocationsResponseSchema) {}
// Cold-import 网页上传(host-self)
const ColdImportUploadResponseSchema = z.object({ jobId: z.string() });
export class ColdImportUploadResponseDto extends createZodDto(ColdImportUploadResponseSchema) {}
// 进度:BullMQ job 状态 + 完成时的 ImportRunResult(含 totals);job 已清理/不存在 → state='not_found'
const ColdImportStatusSchema = z.object({
state: z.string(),
result: z.any().nullable(),
failedReason: z.string().nullable(),
// 运行中的真实进度(cohortDone/cohortTotal 批次);非运行态或 dryRun 为 null
progress: z.object({ done: z.number(), total: z.number() }).nullable(),
});
export class ColdImportStatusDto extends createZodDto(ColdImportStatusSchema) {}
/// 宿主推送记录(sync_logs direction=push 的对宿主视图)—— 对接期自助排查:
/// 哪一批、什么时候、收了几行、落了几条、去重几条、失败几条、错在哪。
const PushLogSchema = z.object({
id: z.string(),
source: z.string().describe('推送的源表名'),
status: z.string().describe('success / partial / failed'),
dryRun: z.boolean().describe('true = 预检批次(未落库)'),
fetched: z.number().int().describe('收到的行数'),
transactionsWritten: z.number().int().describe('实际落库事件数(与 fetched 的差 = 被 PAC 规则过滤)'),
duplicates: z.number().int().describe('幂等命中数(重推安全的证据)'),
failed: z.number().int(),
errorMessage: z.string().nullable(),
startedAt: z.string(),
endedAt: z.string().nullable(),
});
const PushLogsSchema = z.object({ items: z.array(PushLogSchema) });
export class PushLogsDto extends createZodDto(PushLogsSchema) {}
apps/pac-service/src/modules/admin/host-self.controller.ts
...@@ -6,19 +6,28 @@ import { ...@@ -6,19 +6,28 @@ import {
Param, Param,
Patch, Patch,
Post, Post,
Query,
UploadedFile,
UseInterceptors,
} from '@nestjs/common'; } from '@nestjs/common';
import { ApiBearerAuth, ApiOperation, ApiTags } from '@nestjs/swagger'; import { FileInterceptor } from '@nestjs/platform-express';
import { ApiBearerAuth, ApiConsumes, ApiOperation, ApiTags } from '@nestjs/swagger';
import { ZodResponse } from 'nestjs-zod'; import { ZodResponse } from 'nestjs-zod';
import { Permission } from '@pac/types'; import { ApiCode, Permission } from '@pac/types';
import { RequirePermission } from '../../common/decorators/permissions.decorator'; import { RequirePermission } from '../../common/decorators/permissions.decorator';
import { CurrentUser, AuthenticatedUser } from '../../common/decorators/current-user.decorator';
import { BizError } from '../../common/errors/biz-error';
import { import {
TenantScope, TenantScope,
TenantScopeContext, TenantScopeContext,
} from '../../common/decorators/tenant-scope.decorator'; } from '../../common/decorators/tenant-scope.decorator';
import { import {
ColdImportStatusDto,
ColdImportUploadResponseDto,
HostDetailDto, HostDetailDto,
HostStatsDto, HostStatsDto,
HostSummaryDto, HostSummaryDto,
PushLogsDto,
RemovePushSecretResponseDto, RemovePushSecretResponseDto,
RotateCallbackSecretResponseDto, RotateCallbackSecretResponseDto,
RotatePushSecretResponseDto, RotatePushSecretResponseDto,
...@@ -26,6 +35,16 @@ import { ...@@ -26,6 +35,16 @@ import {
UpdateHostRequestDto, UpdateHostRequestDto,
} from './dto/admin.dto'; } from './dto/admin.dto';
import { HostsService } from './hosts.service'; import { HostsService } from './hosts.service';
import { ImportUploadService } from './import-upload.service';
/// multer 内存模式的最小文件形状(不引 @types/multer,同 assistant.controller)
interface UploadedZip {
buffer: Buffer;
originalname: string;
mimetype: string;
}
/// 上传大小上限:存量分批包压缩后通常几十 MB;留 500MB 余量
const MAX_IMPORT_ZIP_BYTES = 500 * 1024 * 1024;
/** /**
* Host Self Admin — 宿主自己的 admin 用户管理"自己这个 host"的配置页。 * Host Self Admin — 宿主自己的 admin 用户管理"自己这个 host"的配置页。
...@@ -42,7 +61,10 @@ import { HostsService } from './hosts.service'; ...@@ -42,7 +61,10 @@ import { HostsService } from './hosts.service';
@Controller('admin/host/self') @Controller('admin/host/self')
@RequirePermission(Permission.PLATFORM_MANAGE) @RequirePermission(Permission.PLATFORM_MANAGE)
export class HostSelfAdminController { export class HostSelfAdminController {
constructor(private readonly hosts: HostsService) {} constructor(
private readonly hosts: HostsService,
private readonly importUpload: ImportUploadService,
) {}
@Get() @Get()
@ZodResponse({ status: 200, type: HostDetailDto }) @ZodResponse({ status: 200, type: HostDetailDto })
...@@ -58,6 +80,20 @@ export class HostSelfAdminController { ...@@ -58,6 +80,20 @@ export class HostSelfAdminController {
return this.hosts.getStats(scope.hostId); return this.hosts.getStats(scope.hostId);
} }
@Get('push-logs')
@ZodResponse({ status: 200, type: PushLogsDto })
@ApiOperation({
summary: '当前宿主的推送记录(对接自助排查)',
description:
'按时间倒序列最近 N 批 push(默认 50,上限 200)。每批含 source / 收行数 / 落库数 / ' +
'去重数 / 失败数 / 报错原因;dryRun=true 的是 ?dryRun=1 预检批次。' +
'用途:当时的同步响应没留存也能事后查——对接期定位"哪批错了、错在哪"。',
})
getPushLogs(@TenantScope() scope: TenantScopeContext, @Query('limit') limit?: string) {
const n = Number.parseInt(limit ?? '', 10);
return this.hosts.getPushLogs(scope.hostId, Number.isFinite(n) && n > 0 ? n : 50);
}
@Patch() @Patch()
@ZodResponse({ status: 200, type: HostSummaryDto }) @ZodResponse({ status: 200, type: HostSummaryDto })
@ApiOperation({ @ApiOperation({
...@@ -125,4 +161,38 @@ export class HostSelfAdminController { ...@@ -125,4 +161,38 @@ export class HostSelfAdminController {
) { ) {
return this.hosts.removeCallbackSecretBySuffix(scope.hostId, suffix); return this.hosts.removeCallbackSecretBySuffix(scope.hostId, suffix);
} }
// ─── 数据导入(上传压缩包 → 后台 cold-import,方向:宿主 → PAC)───
@Post('cold-import')
@ApiConsumes('multipart/form-data')
@UseInterceptors(
FileInterceptor('file', { limits: { fileSize: MAX_IMPORT_ZIP_BYTES } }),
)
@ZodResponse({ status: 202, type: ColdImportUploadResponseDto })
@ApiOperation({
summary: '上传数据压缩包触发整包 cold-import',
description:
'zip 内含 manifest.tables 声明的数据文件(CSV/JSON);manifest+assemblers 取自 PAC 代码库不接受上传。' +
'校验通过后落后台队列,返回 jobId 供轮询进度。',
})
async coldImport(
@TenantScope() scope: TenantScopeContext,
@CurrentUser() user: AuthenticatedUser,
@UploadedFile() file: UploadedZip | undefined,
@Body('dryRun') dryRun?: string,
) {
if (!file?.buffer?.length) {
throw new BizError(ApiCode.CLIENT_BAD_REQUEST, '未收到上传文件(字段名应为 file)');
}
// multipart 文本字段是字符串;'true' → dry-run 预检(不落库)
return this.importUpload.prepareAndEnqueue(scope, file.buffer, user.sub, dryRun === 'true');
}
@Get('cold-import/:jobId')
@ZodResponse({ status: 200, type: ColdImportStatusDto })
@ApiOperation({ summary: '查询一次 cold-import 进度(前端轮询,含运行中百分比)' })
async coldImportStatus(@Param('jobId') jobId: string) {
return this.importUpload.getProgress(jobId);
}
} }
apps/pac-service/src/modules/admin/hosts.service.ts
...@@ -17,6 +17,7 @@ import type { ...@@ -17,6 +17,7 @@ import type {
PullConfig, PullConfig,
ActionUrlsConfig, ActionUrlsConfig,
} from '@pac/types'; } from '@pac/types';
import { SyncDirection } from '@pac/types';
import { PrismaService } from '../../prisma/prisma.service'; import { PrismaService } from '../../prisma/prisma.service';
/** /**
...@@ -300,6 +301,37 @@ export class HostsService { ...@@ -300,6 +301,37 @@ export class HostsService {
}; };
} }
/**
* 宿主推送记录(自助排查)—— 列最近 N 批 push 的结果。
*
* 对接期宿主开发最需要的"事后可查":当时的同步响应若没打进日志,这里仍能看到
* 每批 source / 收行数 / 落库数 / 去重 / 失败 / 报错原因。`?dryRun=1` 的预检批次
* 由 triggeredBy 前缀 `push-dryrun:` 标识,单独打标,不与正式推送混淆。
*/
async getPushLogs(hostId: string, limit = 50) {
const take = Math.min(Math.max(limit, 1), 200);
const rows = await this.prisma.syncLog.findMany({
where: { hostId, direction: SyncDirection.PUSH },
orderBy: { startedAt: 'desc' },
take,
});
return {
items: rows.map((r) => ({
id: r.id,
source: r.resource,
status: r.status,
dryRun: (r.triggeredBy ?? '').startsWith('push-dryrun:'),
fetched: r.fetched ?? 0,
transactionsWritten: r.transactionsWritten ?? 0,
duplicates: r.duplicates ?? 0,
failed: r.failed ?? 0,
errorMessage: r.errorMessage,
startedAt: r.startedAt.toISOString(),
endedAt: r.endedAt?.toISOString() ?? null,
})),
};
}
// ───────────────────────────────────────────────────────── // ─────────────────────────────────────────────────────────
// Remove push secret(按 suffix 定位,精确移除某个 grace key) // Remove push secret(按 suffix 定位,精确移除某个 grace key)
// ───────────────────────────────────────────────────────── // ─────────────────────────────────────────────────────────
......
apps/pac-service/src/modules/sync/cold-import/clickhouse-source.service.ts
...@@ -134,6 +134,35 @@ export class ClickHouseSourceService { ...@@ -134,6 +134,35 @@ export class ClickHouseSourceService {
/// 用途:增量 fetched=0 时分辨「DW 真没新数据(静默)」vs「查询条件失效在空转(告警)」 /// 用途:增量 fetched=0 时分辨「DW 真没新数据(静默)」vs「查询条件失效在空转(告警)」
/// —— 2026-06-10 游标 ISO 格式 bug 让增量空转三天才被人工发现,此探针让它当天就暴露。 /// —— 2026-06-10 游标 ISO 格式 bug 让增量空转三天才被人工发现,此探针让它当天就暴露。
/// 探针绝不能影响同步主流程:任何异常吞掉返回 {}。 /// 探针绝不能影响同步主流程:任何异常吞掉返回 {}。
/**
* 对账用:把资源查询原样包一层,统计**去重患者数**(patient_id + brand 元组 —— 集团内
* 同 patient_id 跨品牌是两个人,必须带 brand)。复用摄入查询本身 → 筛选口径零重复维护。
* 失败返回 null(对账降级,不阻断日报)。查询超时放宽到 120s(大表 count distinct)。
*/
async countDistinctPatients(source: ClickHouseSource, innerSql: string): Promise<number | null> {
let client: ReturnType<typeof createClient> | null = null;
try {
const conn = this.resolveConnection(source);
client = createClient({
url: conn.url,
database: conn.database,
username: conn.username,
password: conn.password,
request_timeout: 120_000,
compression: { response: true, request: false },
});
const q = `SELECT uniqExact(patient_id, brand) AS c FROM (\n${innerSql}\n)`;
const rs = await client.query({ query: q, format: 'JSONEachRow' });
const rows = (await rs.json()) as Array<{ c: string | number }>;
return Number(rows[0]?.c ?? 0);
} catch (e) {
this.logger.warn(`[reconcile] countDistinctPatients 失败: ${e instanceof Error ? e.message : String(e)}`);
return null;
} finally {
if (client) await client.close();
}
}
async probeNewRowCounts( async probeNewRowCounts(
source: ClickHouseSource, source: ClickHouseSource,
perQuery: IncrementalConfig['perQuery'], perQuery: IncrementalConfig['perQuery'],
......
apps/pac-service/src/modules/sync/cold-import/cold-import.service.ts
...@@ -286,6 +286,9 @@ export class ColdImportService { ...@@ -286,6 +286,9 @@ export class ColdImportService {
source: string; // 原生源表名(= manifest tables[].table / transform 终端表) source: string; // 原生源表名(= manifest tables[].table / transform 终端表)
rows: Record<string, unknown>[]; rows: Record<string, unknown>[];
triggeredBy?: string; triggeredBy?: string;
/// 预检:跑完整管线(transforms + 装配 + 归一 + tenant 解析)但**不落任何库**,
/// 用于宿主对接自测——看 accepted/失败数与样本 canonical,确认 PAC 理解对了再正式推。
dryRun?: boolean;
}): Promise<{ }): Promise<{
syncLogId: string; syncLogId: string;
perResource: PerResourceStats[]; perResource: PerResourceStats[];
...@@ -333,7 +336,9 @@ export class ColdImportService { ...@@ -333,7 +336,9 @@ export class ColdImportService {
tenantId: 't_pending', // push 多 tenant 由行决定;finalize 时回填首个 seen tenant tenantId: 't_pending', // push 多 tenant 由行决定;finalize 时回填首个 seen tenant
direction: SyncDirection.PUSH, direction: SyncDirection.PUSH,
resource: opts.source, resource: opts.source,
triggeredBy: opts.triggeredBy ?? `push:${opts.hostName}:${opts.source}`, triggeredBy:
opts.triggeredBy ??
`${opts.dryRun ? 'push-dryrun' : 'push'}:${opts.hostName}:${opts.source}`,
status: SyncStatus.RUNNING, status: SyncStatus.RUNNING,
}, },
}); });
...@@ -362,19 +367,19 @@ export class ColdImportService { ...@@ -362,19 +367,19 @@ export class ColdImportService {
if (cfg.canonical === 'patient') { if (cfg.canonical === 'patient') {
perResource.push( perResource.push(
await this.processPatients( await this.processPatients(
transformed, cfg, host.id, tenantResolver, seenTenants, normalize, false, true, transformed, cfg, host.id, tenantResolver, seenTenants, normalize, !!opts.dryRun, true,
), ),
); );
} else if (cfg.canonical === 'patient_relation') { } else if (cfg.canonical === 'patient_relation') {
perResource.push( perResource.push(
await this.processPatientRelations( await this.processPatientRelations(
transformed, cfg, host.id, tenantResolver, normalize, false, transformed, cfg, host.id, tenantResolver, normalize, !!opts.dryRun,
), ),
); );
} else if (cfg.canonical === 'patient_return_visit') { } else if (cfg.canonical === 'patient_return_visit') {
perResource.push( perResource.push(
await this.processPatientReturnVisits( await this.processPatientReturnVisits(
transformed, cfg, host.id, tenantResolver, normalize, false, transformed, cfg, host.id, tenantResolver, normalize, !!opts.dryRun,
), ),
); );
} }
...@@ -388,7 +393,7 @@ export class ColdImportService { ...@@ -388,7 +393,7 @@ export class ColdImportService {
seenTenants, seenTenants,
normalize, normalize,
syncLog.id, syncLog.id,
false, !!opts.dryRun,
); );
perResource.push(stats); perResource.push(stats);
} }
...@@ -570,6 +575,37 @@ export class ColdImportService { ...@@ -570,6 +575,37 @@ export class ColdImportService {
} }
/** /**
* 数据对账 —— DW 源侧「去重患者数」(设计见 /docs/data-reconciliation)。
*
* 复用摄入用的 sql_source 查询本身(同一道筛选闸)→ diff 只剩真漂移,零重复维护。
* 只算 DW 侧;PAC 侧计数 + diff + 渲染由调用方(日报)做(它持 prisma + 按 fact_type)。
*
* @param queryKeys 要对账的 sql_source.queries key(如 fact_emr_treatment_out / image_finding_rows)
* @returns { queryKey: 去重患者数 | null(查失败) };无 sql_source(file/push 宿主)→ 空 {}
*/
async reconcileDwPatientCounts(
hostName: string,
queryKeys: string[],
): Promise<Record<string, number | null>> {
const out: Record<string, number | null> = {};
let manifest: ColdImportManifest;
try {
manifest = this.readManifest(this.resolveHostDir(hostName));
} catch (e) {
this.logger.warn(`[reconcile] 读 manifest 失败 host=${hostName}: ${e instanceof Error ? e.message : String(e)}`);
return out;
}
const source = manifest.sql_source;
if (!source) return out; // file/push 宿主无直连源,不对账
for (const key of queryKeys) {
const sql = source.queries[key];
if (!sql) { out[key] = null; continue; }
out[key] = await this.chSource.countDistinctPatients(source, sql);
}
return out;
}
/**
* raw 层列漂移检测(form A):本次推送行的列集 vs 历史同源 rawPayload 的列集。 * raw 层列漂移检测(form A):本次推送行的列集 vs 历史同源 rawPayload 的列集。
* - 历史有、本次没有 → column_removed(疑似宿主删列/改名) * - 历史有、本次没有 → column_removed(疑似宿主删列/改名)
* - 本次有、历史没有 → column_added(疑似宿主加列/改名) * - 本次有、历史没有 → column_added(疑似宿主加列/改名)
...@@ -596,10 +632,25 @@ export class ColdImportService { ...@@ -596,10 +632,25 @@ export class ColdImportService {
orderBy: { createdAt: 'desc' }, orderBy: { createdAt: 'desc' },
}); });
if (sample.length < MIN_BASELINE) return []; if (sample.length < MIN_BASELINE) return [];
const baseline = new Set<string>(); // 同 subjectType 可能混多源(如 refund = 结算头 patient_settlement ∪ 结算明细
for (const s of sample) { // patient_settlement_spec,列集不同)。按「与本批列签名的 Jaccard 相似度」自选**同源**样本,
// 否则另一源的专属列会被误报成"本源删列"(suspectFields 虚警)。跨源如 settlement/spec 交并比
// 仅 ~0.35,≥0.5 阈值可干净分离;同源 ≈1.0。
const pushed = args.pushedCols;
const sameSource = sample.filter((s) => {
const rp = s.rawPayload as Record<string, unknown> | null; const rp = s.rawPayload as Record<string, unknown> | null;
if (rp) for (const k of Object.keys(rp)) if (!k.startsWith('_')) baseline.add(k); if (!rp) return false;
const cols = Object.keys(rp).filter((k) => !k.startsWith('_'));
if (cols.length === 0) return false;
const inter = cols.filter((c) => pushed.has(c)).length;
const union = new Set([...cols, ...pushed]).size;
return union > 0 && inter / union >= 0.5;
});
if (sameSource.length < MIN_BASELINE) return []; // 同源基线不足 → 不可靠,宁可不报
const baseline = new Set<string>();
for (const s of sameSource) {
const rp = s.rawPayload as Record<string, unknown>;
for (const k of Object.keys(rp)) if (!k.startsWith('_')) baseline.add(k);
} }
const out: SuspectField[] = []; const out: SuspectField[] = [];
...@@ -2150,16 +2201,28 @@ export class ColdImportService { ...@@ -2150,16 +2201,28 @@ export class ColdImportService {
'manifest 缺数据源:既无 sql_source 也无 tables[](schema refine 本该拦住,兜底报错)', 'manifest 缺数据源:既无 sql_source 也无 tables[](schema refine 本该拦住,兜底报错)',
); );
} }
// 宽容装载:声明表文件缺失 → warn + 跳过(不 throw)。
// PAC 各主体 assembler 独立:缺某表只少那部分 facts。下游已宽容——
// transform lookup/from 缺表当空表、project input 缺表跳过、assembler primary.table
// 缺表产 0 行。成组主体的完整性由上传层(ImportUploadService)按 manifest group 声明前置守卫,
// 不在此处判(此处也服务 CLI 全量导入,不宜硬拦)。
const tables: Record<string, unknown[]> = {}; const tables: Record<string, unknown[]> = {};
const skipped: string[] = [];
for (const t of manifest.tables) { for (const t of manifest.tables) {
const filePath = path.join(dir, t.file); const filePath = path.join(dir, t.file);
if (!fs.existsSync(filePath)) { if (!fs.existsSync(filePath)) {
throw new Error(`Raw table file not found: ${filePath}`); skipped.push(t.table);
continue;
} }
const format = t.format ?? inferFileFormat(t.file); const format = t.format ?? inferFileFormat(t.file);
tables[t.table] = this.readRows(filePath, format); tables[t.table] = this.readRows(filePath, format);
this.logger.log(` loaded raw table "${t.table}" (${tables[t.table]!.length} rows)`); this.logger.log(` loaded raw table "${t.table}" (${tables[t.table]!.length} rows)`);
} }
if (skipped.length > 0) {
this.logger.warn(
` ${skipped.length} 张声明表文件缺失,已跳过(依赖它们的 assembler/transform 将当空表处理): ${skipped.join(', ')}`,
);
}
return tables; return tables;
} }
...@@ -2262,24 +2325,46 @@ interface TotalsBlock { ...@@ -2262,24 +2325,46 @@ interface TotalsBlock {
/// 沿 transforms 的 output→input 链回溯,找到非任何 transform 产出的"原始源表"名。 /// 沿 transforms 的 output→input 链回溯,找到非任何 transform 产出的"原始源表"名。
/// 例:diagnosis_rows ← _diagnosis_norm ← _diagnosis_raw ← fact_emr_treatment_out(终点=源表)。 /// 例:diagnosis_rows ← _diagnosis_norm ← _diagnosis_raw ← fact_emr_treatment_out(终点=源表)。
/// 用于 reparse:transaction.rawPayload 存的就是这张源表的行。 /// 用于 reparse(rawPayload 存的就是这张源表的行)+ push(source→匹配 assembler)。
function traceRawSourceTable(primaryTable: string, transforms: ReadonlyArray<unknown>): string { ///
const byOutput = new Map<string, string>(); // output 表名 → input 表名 /// union 多输入:各输入独立回溯,**全部收敛到同一根表**才返回该根(如 refund_full_rows
/// ← union[_refund_status4, _refund_neg3] 两路都回到 patient_settlement → 返回它);
/// 若发散到多个根(语义歧义)→ 停在 union 输出本身(维持旧行为,不误判)。
/// 修复前 union 不被追 → refund_full 在 push/reparse 里回溯不到源表被静默跳过(丢财务数据)。
export function traceRawSourceTable(primaryTable: string, transforms: ReadonlyArray<unknown>): string {
const byOutput = new Map<string, string>(); // output → input(单输入算子)
const unionInputs = new Map<string, string[]>(); // union output → inputs[]
for (const tf of transforms) { for (const tf of transforms) {
const t = tf as { input?: string; output?: string; outputs?: Array<{ output?: string }> }; const t = tf as {
kind?: string;
input?: string;
output?: string;
inputs?: string[];
outputs?: Array<{ output?: string }>;
};
if (t.kind === 'union' && t.output && Array.isArray(t.inputs)) {
unionInputs.set(t.output, t.inputs);
continue;
}
if (t.output && t.input) byOutput.set(t.output, t.input); if (t.output && t.input) byOutput.set(t.output, t.input);
// route_by_pattern 多 output:每个 output 都回到同一 input // route_by_pattern 多 output:每个 output 都回到同一 input
if (Array.isArray(t.outputs) && t.input) { if (Array.isArray(t.outputs) && t.input) {
for (const o of t.outputs) if (o?.output) byOutput.set(o.output, t.input); for (const o of t.outputs) if (o?.output) byOutput.set(o.output, t.input);
} }
} }
let cur = primaryTable; const resolve = (tbl: string, seen: Set<string>): string => {
const seen = new Set<string>(); if (seen.has(tbl)) return tbl; // 环保护
while (byOutput.has(cur) && !seen.has(cur)) { seen.add(tbl);
seen.add(cur); if (byOutput.has(tbl)) return resolve(byOutput.get(tbl)!, seen);
cur = byOutput.get(cur)!; const ins = unionInputs.get(tbl);
} if (ins && ins.length > 0) {
return cur; // 各输入独立回溯(各带 seen 副本,兄弟分支互不干扰);全一致才收敛,否则停在 union 输出
const roots = new Set(ins.map((i) => resolve(i, new Set(seen))));
return roots.size === 1 ? [...roots][0]! : tbl;
}
return tbl; // 已是根
};
return resolve(primaryTable, new Set<string>());
} }
export interface PerResourceStats extends TotalsBlock { export interface PerResourceStats extends TotalsBlock {
......
apps/pac-service/src/modules/sync/cold-import/manifest.schema.ts
...@@ -26,6 +26,12 @@ export const RawTableSchema = z.object({ ...@@ -26,6 +26,12 @@ export const RawTableSchema = z.object({
file: z.string().min(1).describe('e.g. tb_encounter.csv / tb_patient.json'), file: z.string().min(1).describe('e.g. tb_encounter.csv / tb_patient.json'),
/// 文件格式(默认按扩展名推断) /// 文件格式(默认按扩展名推断)
format: z.enum(['csv', 'json']).optional(), format: z.enum(['csv', 'json']).optional(),
/// 成组主体标签(宿主特有,可选)。同 group 的表在**网页上传**时要么全在、要么整组跳过
/// —— 由 ImportUploadService 通用地读此声明做完整性校验。用于"半组产不出主体"的强依赖
/// (如 treatment_planned = customer_treat_plan[头] + _item[行],缺任一整个资源产 0 行)。
/// 不标 = 独立表:缺失仅少那部分 facts(装载层已宽容跳过)。
/// 注:CLI 全量导入不受此约束(只 warn 跳过);仅上传路径按 group 前置守卫。
group: z.string().min(1).optional(),
}); });
export type RawTable = z.infer<typeof RawTableSchema>; export type RawTable = z.infer<typeof RawTableSchema>;
......
apps/pac-service/src/modules/sync/push/push-receiver.service.ts
...@@ -49,14 +49,18 @@ export class PushReceiverService { ...@@ -49,14 +49,18 @@ export class PushReceiverService {
hostName: string; hostName: string;
source: string; source: string;
rows: Record<string, unknown>[]; rows: Record<string, unknown>[];
/// 预检(?dryRun=1):跑完整管线但不落库,也不触发重算。宿主对接自测用。
dryRun?: boolean;
}): Promise<PushRowsResponse> { }): Promise<PushRowsResponse> {
const { hostId, hostName, source, rows } = input; const { hostId, hostName, source, rows } = input;
const dryRun = !!input.dryRun;
const r = await this.coldImport.ingestRawTables({ const r = await this.coldImport.ingestRawTables({
hostName, hostName,
source, source,
rows, rows,
triggeredBy: `push:${hostName}:${source}`, dryRun,
triggeredBy: `${dryRun ? 'push-dryrun' : 'push'}:${hostName}:${source}`,
}); });
const agg = r.perResource.reduce( const agg = r.perResource.reduce(
...@@ -81,8 +85,9 @@ export class PushReceiverService { ...@@ -81,8 +85,9 @@ export class PushReceiverService {
// 事件驱动:touched 患者入队 persona-recompute(去抖:存量批次重放时窗内合并,见 QueueProducer) // 事件驱动:touched 患者入队 persona-recompute(去抖:存量批次重放时窗内合并,见 QueueProducer)
// ⚠️ plan 不在此触发(persona 完成后链式 enqueue),与单患者刷新路径有意区分。 // ⚠️ plan 不在此触发(persona 完成后链式 enqueue),与单患者刷新路径有意区分。
// 预检不产生任何副作用 → 不入队(touched 本就为空,此处显式短路,语义更清楚)。
let personaEnqueued = 0; let personaEnqueued = 0;
for (const { patientId, tenantId } of r.touched) { for (const { patientId, tenantId } of dryRun ? [] : r.touched) {
try { try {
await this.queue.enqueuePersonaRecompute( await this.queue.enqueuePersonaRecompute(
{ hostId, tenantId, patientId, triggeredBy: `push:${hostName}` }, { hostId, tenantId, patientId, triggeredBy: `push:${hostName}` },
...@@ -101,6 +106,13 @@ export class PushReceiverService { ...@@ -101,6 +106,13 @@ export class PushReceiverService {
`txn=${agg.txn} dup=${agg.dup} failed=${agg.failed} personaEnqueued=${personaEnqueued}`, `txn=${agg.txn} dup=${agg.dup} failed=${agg.failed} personaEnqueued=${personaEnqueued}`,
); );
// 预检:回样本 canonical(各资源首行),供宿主核对 PAC 把原生行翻译成了什么
const samples = dryRun
? r.perResource
.filter((s) => (s.sampleCanonical?.length ?? 0) > 0)
.map((s) => ({ resource: s.resource, canonical: s.sampleCanonical[0] }))
: [];
return { return {
syncLogId: r.syncLogId, syncLogId: r.syncLogId,
source, source,
...@@ -111,6 +123,8 @@ export class PushReceiverService { ...@@ -111,6 +123,8 @@ export class PushReceiverService {
personaEnqueued, personaEnqueued,
mappingMisses: r.mappingMisses.length, mappingMisses: r.mappingMisses.length,
suspectFields: r.suspectFields.length, suspectFields: r.suspectFields.length,
dryRun,
samples,
}; };
} }
......
apps/pac-service/src/modules/sync/push/push.controller.ts
import { import {
Body, Body,
Controller, Controller,
Get,
Headers, Headers,
HttpCode, HttpCode,
Post, Post,
Query,
Req, Req,
} from '@nestjs/common'; } from '@nestjs/common';
import type { Request } from 'express'; import type { Request } from 'express';
import { ApiOperation, ApiTags } from '@nestjs/swagger'; import { ApiOperation, ApiQuery, ApiTags } from '@nestjs/swagger';
import { ApiCode } from '@pac/types'; import { ApiCode } from '@pac/types';
import { BizError } from '../../../common/errors/biz-error'; import { BizError } from '../../../common/errors/biz-error';
import { Public } from '../../../common/decorators/public.decorator'; import { Public } from '../../../common/decorators/public.decorator';
import { HmacVerifier } from './hmac-verifier.service'; import { HmacVerifier } from './hmac-verifier.service';
import { PushReceiverService } from './push-receiver.service'; import { PushReceiverService } from './push-receiver.service';
import { HostsService } from '../../admin/hosts.service';
import { import {
PushEventsRequestSchema, PushEventsRequestSchema,
type PushEventsRequest, type PushEventsRequest,
...@@ -44,6 +47,7 @@ export class PushController { ...@@ -44,6 +47,7 @@ export class PushController {
constructor( constructor(
private readonly verifier: HmacVerifier, private readonly verifier: HmacVerifier,
private readonly receiver: PushReceiverService, private readonly receiver: PushReceiverService,
private readonly hosts: HostsService,
) {} ) {}
/** /**
...@@ -137,14 +141,23 @@ export class PushController { ...@@ -137,14 +141,23 @@ export class PushController {
@Public() @Public()
@HttpCode(200) @HttpCode(200)
@ApiOperation({ @ApiOperation({
summary: '宿主推送原生表行 webhook(形态 A,HMAC 验签)', summary: '宿主推送原生表行 webhook(形态 A,HMAC 验签;?dryRun=1 预检不落库)',
description: 'Headers 必填 X-PAC-Host-Id / X-PAC-Timestamp / X-PAC-Signature。Body: { tenantId?, source, rows[] }。', description:
'Headers 必填 X-PAC-Host-Id / X-PAC-Timestamp / X-PAC-Signature。Body: { source, rows[] }。' +
' 加 ?dryRun=1 走**预检**:跑完整管线(transforms+装配+归一)但不落库、不触发重算,' +
'响应带各资源样本 canonical —— 对接自测先预检到 failed=0 且 mappingMisses/suspectFields 为 0,再正式推。',
})
@ApiQuery({
name: 'dryRun',
required: false,
description: "预检开关:'1' / 'true' / 'yes' = 跑完整管线但不落库、不触发重算,响应带样本 canonical",
}) })
async receiveRows( async receiveRows(
@Req() req: Request & { rawBody?: Buffer }, @Req() req: Request & { rawBody?: Buffer },
@Headers('x-pac-host-id') hostIdHeader: string | undefined, @Headers('x-pac-host-id') hostIdHeader: string | undefined,
@Headers('x-pac-timestamp') timestampHeader: string | undefined, @Headers('x-pac-timestamp') timestampHeader: string | undefined,
@Headers('x-pac-signature') signatureHeader: string | undefined, @Headers('x-pac-signature') signatureHeader: string | undefined,
@Query('dryRun') dryRunQuery: string | undefined,
@Body() body: PushRowsRequest, @Body() body: PushRowsRequest,
) { ) {
const rawBody = req.rawBody ? req.rawBody.toString('utf8') : JSON.stringify(body); const rawBody = req.rawBody ? req.rawBody.toString('utf8') : JSON.stringify(body);
...@@ -158,13 +171,53 @@ export class PushController { ...@@ -158,13 +171,53 @@ export class PushController {
const parsed = this.parseOrReject(() => PushRowsRequestSchema.parse(body)); const parsed = this.parseOrReject(() => PushRowsRequestSchema.parse(body));
// ?dryRun=1 / true / yes 均视为预检(宿主客户端拼 query 习惯不一)
const dryRun = ['1', 'true', 'yes'].includes((dryRunQuery ?? '').trim().toLowerCase());
return this.withHostLock(hostId, hostName, () => return this.withHostLock(hostId, hostName, () =>
this.receiver.receiveRows({ this.receiver.receiveRows({
hostId, hostId,
hostName, hostName,
source: parsed.source, source: parsed.source,
rows: parsed.rows, rows: parsed.rows,
dryRun,
}), }),
); );
} }
/**
* 推送记录查询 —— **用已有 push 凭据验签,无需另外登录**(对接自测友好)。
*
* 与登录态的 `GET /admin/host/self/push-logs` 返回同一份数据(同一实现),区别只在鉴权方式:
* 这里复用宿主推送时已经在用的 HMAC 三件套,故不必为联调再开账号;
* 也**不是公开接口** —— 没有正确签名拿不到任何数据,不会泄露宿主推送情况。
*
* GET 无 body,签名对**空串**计算:`hex(HMAC-SHA256(`${timestamp}.`, secret))`。
*/
@Get('logs')
@Public()
@HttpCode(200)
@ApiOperation({
summary: '查询本宿主推送记录(HMAC 验签,免登录)',
description:
'Headers 同 push:X-PAC-Host-Id / X-PAC-Timestamp / X-PAC-Signature。' +
'GET 无 body,签名对空串算:hex(HMAC-SHA256(`<timestamp>.`, secret))。' +
'返回最近 N 批(默认 50、上限 200):source / status / dryRun / 收行数 / 落库数 / 去重 / 失败 / 报错。',
})
@ApiQuery({ name: 'limit', required: false, description: '返回条数,默认 50、上限 200' })
async logs(
@Headers('x-pac-host-id') hostIdHeader: string | undefined,
@Headers('x-pac-timestamp') timestampHeader: string | undefined,
@Headers('x-pac-signature') signatureHeader: string | undefined,
@Query('limit') limit?: string,
) {
const { hostId } = await this.verifier.verify({
hostIdHeader,
timestampHeader,
signatureHeader,
rawBody: '', // GET 无 body
});
const n = Number.parseInt(limit ?? '', 10);
return this.hosts.getPushLogs(hostId, Number.isFinite(n) && n > 0 ? n : 50);
}
} }
apps/pac-service/src/modules/sync/push/push.schema.ts
...@@ -87,9 +87,15 @@ export const PushRowsResponseSchema = z.object({ ...@@ -87,9 +87,15 @@ export const PushRowsResponseSchema = z.object({
duplicates: z.number().int(), duplicates: z.number().int(),
failed: z.number().int(), failed: z.number().int(),
personaEnqueued: z.number().int().describe('入队画像重算的患者数(plan 不在此触发,由定时任务保)'), personaEnqueued: z.number().int().describe('入队画像重算的患者数(plan 不在此触发,由定时任务保)'),
/// 宿主自检信号(代替 cold-import 的 dry-run):样本批推完看这两个数,>0 先停下联系 PAC。 /// 宿主自检信号:样本批推完看这两个数,>0 先停下联系 PAC。
mappingMisses: z.number().int().describe('映射覆盖缺口种类数(有原值落 _default)'), mappingMisses: z.number().int().describe('映射覆盖缺口种类数(有原值落 _default)'),
suspectFields: z.number().int().describe('疑似宿主改表项数(列缺席/新增,vs 历史基线)'), suspectFields: z.number().int().describe('疑似宿主改表项数(列缺席/新增,vs 历史基线)'),
/// 预检模式(?dryRun=1):跑完整管线(transforms+装配+归一+tenant 解析)但**不落任何库**,
/// 数字表示"若正式推会发生什么"。对接自测用:先预检到 failed=0 且两个自检信号为 0,再正式推。
dryRun: z.boolean().describe('true = 本次为预检,未落库'),
samples: z
.array(z.object({ resource: z.string(), canonical: z.unknown() }))
.describe('预检样本:各 canonical 资源首行的翻译结果(仅 dryRun 返回,供宿主核对 PAC 是否理解正确)'),
}); });
export type PushRowsResponse = z.infer<typeof PushRowsResponseSchema>; export type PushRowsResponse = z.infer<typeof PushRowsResponseSchema>;
apps/pac-service/src/modules/sync/sync.module.ts
...@@ -32,8 +32,11 @@ import { HmacVerifier } from './push/hmac-verifier.service'; ...@@ -32,8 +32,11 @@ import { HmacVerifier } from './push/hmac-verifier.service';
* - AssemblerModule(raw → canonical;真实 HTTP adapter 才需要;mock/push 跳过) * - AssemblerModule(raw → canonical;真实 HTTP adapter 才需要;mock/push 跳过)
* - PipelineModule(synthesizer + parser pipeline) * - PipelineModule(synthesizer + parser pipeline)
*/ */
import { AdminModule } from '../admin/admin.module';
@Module({ @Module({
imports: [ imports: [
AdminModule, // push/logs 复用 HostsService.getPushLogs(与登录态自助页同一实现)
FactsModule, FactsModule,
AssemblerModule, AssemblerModule,
TransformsModule, TransformsModule,
......
apps/pac-service/src/modules/sync/transforms/operators/filter.op.ts
...@@ -10,6 +10,7 @@ import { isEmptyText, type Row } from '../row'; ...@@ -10,6 +10,7 @@ import { isEmptyText, type Row } from '../row';
* - not_empty: true — 字段非 null/undefined/空字符串 * - not_empty: true — 字段非 null/undefined/空字符串
* - in: [...] — 字段值在列表中 * - in: [...] — 字段值在列表中
* - equals: x — 字段值等于 x * - equals: x — 字段值等于 x
* - lt/lte/gt/gte: n — 数值比较(字段解析为数字;非数值/空 → 不通过)
*/ */
export function runFilter(op: FilterOp, rows: Row[]): { rows: Row[]; dropped: number } { export function runFilter(op: FilterOp, rows: Row[]): { rows: Row[]; dropped: number } {
const out: Row[] = []; const out: Row[] = [];
...@@ -127,7 +128,26 @@ function passAll(row: Row, where: Record<string, WhereRule>): boolean { ...@@ -127,7 +128,26 @@ function passAll(row: Row, where: Record<string, WhereRule>): boolean {
if (!set.has(String(v))) return false; if (!set.has(String(v))) return false;
} else if ('equals' in rule) { } else if ('equals' in rule) {
if (String(v) !== String(rule.equals)) return false; if (String(v) !== String(rule.equals)) return false;
} else if ('lt' in rule) {
const n = toNumber(v);
if (n === null || n >= rule.lt) return false;
} else if ('lte' in rule) {
const n = toNumber(v);
if (n === null || n > rule.lte) return false;
} else if ('gt' in rule) {
const n = toNumber(v);
if (n === null || n <= rule.gt) return false;
} else if ('gte' in rule) {
const n = toNumber(v);
if (n === null || n < rule.gte) return false;
} }
} }
return true; return true;
} }
/// 数值解析:空/非数字 → null(数值比较一律不通过)。CSV 字段是字符串,金额如 "-30.00"。
function toNumber(v: unknown): number | null {
if (v === null || v === undefined || (typeof v === 'string' && v.trim() === '')) return null;
const n = Number(v);
return Number.isFinite(n) ? n : null;
}
apps/pac-service/src/modules/sync/transforms/transforms.schema.ts
...@@ -69,6 +69,8 @@ export type ProjectOp = z.infer<typeof ProjectOpSchema>; ...@@ -69,6 +69,8 @@ export type ProjectOp = z.infer<typeof ProjectOpSchema>;
* - `not_empty: true` — 字符串非空 / null 过滤 * - `not_empty: true` — 字符串非空 / null 过滤
* - `in: [...]` — 值在列表中 * - `in: [...]` — 值在列表中
* - `equals: x` — 值等于 x * - `equals: x` — 值等于 x
* - `lt/lte/gt/gte: n` — 数值比较(字段解析为数字;非数值/空 → 不通过)。
* 让退费正负切分等"业务口径 WHERE"能在 PAC 侧做(宿主导原表,PAC transforms 过滤)。
*/ */
export const WhereRuleSchema = z.union([ export const WhereRuleSchema = z.union([
z.object({ not_empty: z.literal(true) }), z.object({ not_empty: z.literal(true) }),
...@@ -82,6 +84,11 @@ export const WhereRuleSchema = z.union([ ...@@ -82,6 +84,11 @@ export const WhereRuleSchema = z.union([
z.object({ real_no_tooth: z.array(z.union([z.string(), z.number()])).min(1) }), z.object({ real_no_tooth: z.array(z.union([z.string(), z.number()])).min(1) }),
z.object({ in: z.array(z.union([z.string(), z.number()])).min(1) }), z.object({ in: z.array(z.union([z.string(), z.number()])).min(1) }),
z.object({ equals: z.union([z.string(), z.number()]) }), z.object({ equals: z.union([z.string(), z.number()]) }),
// 数值比较:字段解析为数字后比较;非数值/空值一律不通过(退费 receivable_this < 0 等)。
z.object({ lt: z.number() }),
z.object({ lte: z.number() }),
z.object({ gt: z.number() }),
z.object({ gte: z.number() }),
]); ]);
export type WhereRule = z.infer<typeof WhereRuleSchema>; export type WhereRule = z.infer<typeof WhereRuleSchema>;
......
apps/pac-service/src/queues/job-payloads.ts
...@@ -40,3 +40,18 @@ export interface ExecutionCallbackJob { ...@@ -40,3 +40,18 @@ export interface ExecutionCallbackJob {
executionId: string; executionId: string;
triggeredBy: string; triggeredBy: string;
} }
export interface ColdImportJob {
hostId: string;
tenantId: string;
/** manifest.host_name;仅日志用(importDirectory 从 dir 内 manifest 自定位 host) */
hostName: string;
/** cold-import 的 --dir:迷你 data 根下的 <host>/ 子目录(含 manifest+assemblers+数据文件) */
dir: string;
/** 清理根:迷你 data 根(dir 的父,含 <host>/ 和 _shared/);worker 跑完整个删掉 */
cleanupDir: string;
/** true = dry-run 预检:跑完整校验+翻译但不落库,返回 totals(failed/mappingMiss)供确认 */
dryRun?: boolean;
/** `upload:<userId>` */
triggeredBy: string;
}
apps/pac-service/src/queues/processors/cold-import.processor.ts 0 → 100644
import { Logger } from '@nestjs/common';
import { Processor, WorkerHost } from '@nestjs/bullmq';
import { Job } from 'bullmq';
import * as fs from 'node:fs';
import { QueueName } from '../queue-names';
import type { ColdImportJob } from '../job-payloads';
import { ColdImportService } from '../../modules/sync/cold-import/cold-import.service';
/**
* cold-import worker — 网页上传压缩包触发的整包导入。
*
* 流程:
* 1. 调 ColdImportService.importDirectory(job.dir) — 与 CLI `pnpm cold-import` 同一入口;
* 内部自建 syncLog(审计/最近事件)、cohort 分批、幂等落库。返回 ImportRunResult。
* 2. finally 清理临时解压目录(含 PII 的上传数据文件不残留);失败时保留供排查。
*
* concurrency=1:整包导入是重任务,且 importDirectory 有 partial UNIQUE(host_id)
* WHERE status='running' 并发锁 —— 串行天然避免撞锁。attempts=1 不自动重试。
*
* 返回值即 job.returnvalue,前端按 jobId 轮询时读其中 totals 展示结果。
*/
@Processor(QueueName.COLD_IMPORT, { concurrency: 1 })
export class ColdImportProcessor extends WorkerHost {
private readonly logger = new Logger(ColdImportProcessor.name);
constructor(private readonly coldImport: ColdImportService) {
super();
}
async process(job: Job<ColdImportJob>) {
const { hostName, dir, cleanupDir, dryRun, triggeredBy } = job.data;
this.logger.log(
`cold-import START job=${job.id} host=${hostName} dir=${dir} dryRun=${!!dryRun} triggeredBy=${triggeredBy}`,
);
try {
const result = await this.coldImport.importDirectory(dir, { dryRun: !!dryRun });
this.logger.log(
`cold-import DONE job=${job.id} host=${hostName} status=${result.status} ` +
`patients=${result.totals.patientsUpserted} txn=${result.totals.transactionsWritten} ` +
`dup=${result.totals.duplicates} failed=${result.totals.failed} ` +
`factsCreated=${result.totals.factsCreated} factsSuperseded=${result.totals.factsSuperseded}`,
);
return result;
} finally {
// 成功/失败都清理整个迷你 data 根(cleanupDir,含 <host>/ 数据 + _shared/ 副本)。
// 始终清理(PII 优先),失败原因已在 syncLog.errorMessage + job.failedReason 里。
const target = cleanupDir ?? dir; // 兼容旧 job(无 cleanupDir)
try {
await fs.promises.rm(target, { recursive: true, force: true });
this.logger.debug(`cold-import 清理临时目录 job=${job.id} dir=${target}`);
} catch (err) {
this.logger.warn(
`cold-import 清理临时目录失败 job=${job.id} dir=${target}: ${err instanceof Error ? err.message : err}`,
);
}
}
}
}
apps/pac-service/src/queues/queue-names.ts
...@@ -15,6 +15,8 @@ export const QueueName = { ...@@ -15,6 +15,8 @@ export const QueueName = {
PLAN_ASSET_GENERATE: 'plan-asset-generate', PLAN_ASSET_GENERATE: 'plan-asset-generate',
/// 执行结果回执投递(PAC → 宿主 callbackUrl,HMAC 签名 + 退避重试;jobName = executionId) /// 执行结果回执投递(PAC → 宿主 callbackUrl,HMAC 签名 + 退避重试;jobName = executionId)
EXECUTION_CALLBACK: 'execution-callback', EXECUTION_CALLBACK: 'execution-callback',
/// 网页上传压缩包触发的整包 cold-import(长任务,jobName = jobId;concurrency=1 串行)
COLD_IMPORT: 'cold-import',
} as const; } as const;
export type QueueName = (typeof QueueName)[keyof typeof QueueName]; export type QueueName = (typeof QueueName)[keyof typeof QueueName];
...@@ -53,4 +55,14 @@ export const QueueDefaults = { ...@@ -53,4 +55,14 @@ export const QueueDefaults = {
removeOnComplete: { count: 500 }, removeOnComplete: { count: 500 },
removeOnFail: { count: 5_000 }, removeOnFail: { count: 5_000 },
}, },
[QueueName.COLD_IMPORT]: {
/**
* 整包导入是几十分钟的重任务,且 importDirectory 有 partial UNIQUE(host_id) WHERE status='running'
* 并发锁(重试会撞锁)→ **不自动重试**(attempts:1),失败就留 failed 供人工排查。
* 完成的 job 保留久一点(前端靠 jobId 轮询进度,别在读到前被清)。
*/
attempts: 1,
removeOnComplete: { age: 24 * 3_600, count: 100 },
removeOnFail: { count: 1_000 },
},
} as const; } as const;
apps/pac-service/src/queues/queue-producer.service.ts
import { Injectable, Logger } from '@nestjs/common'; import { Injectable, Logger } from '@nestjs/common';
import { InjectQueue } from '@nestjs/bullmq'; import { InjectQueue } from '@nestjs/bullmq';
import { randomUUID } from 'node:crypto';
import type { Queue } from 'bullmq'; import type { Queue } from 'bullmq';
import { QueueName, QueueDefaults } from './queue-names'; import { QueueName, QueueDefaults } from './queue-names';
import type { import type {
...@@ -7,6 +8,7 @@ import type { ...@@ -7,6 +8,7 @@ import type {
PlanRecomputeJob, PlanRecomputeJob,
PlanAssetGenerateJob, PlanAssetGenerateJob,
ExecutionCallbackJob, ExecutionCallbackJob,
ColdImportJob,
} from './job-payloads'; } from './job-payloads';
/** /**
...@@ -29,6 +31,7 @@ export class QueueProducer { ...@@ -29,6 +31,7 @@ export class QueueProducer {
@InjectQueue(QueueName.PLAN_RECOMPUTE) private readonly planQ: Queue, @InjectQueue(QueueName.PLAN_RECOMPUTE) private readonly planQ: Queue,
@InjectQueue(QueueName.PLAN_ASSET_GENERATE) private readonly assetQ: Queue, @InjectQueue(QueueName.PLAN_ASSET_GENERATE) private readonly assetQ: Queue,
@InjectQueue(QueueName.EXECUTION_CALLBACK) private readonly callbackQ: Queue, @InjectQueue(QueueName.EXECUTION_CALLBACK) private readonly callbackQ: Queue,
@InjectQueue(QueueName.COLD_IMPORT) private readonly coldImportQ: Queue,
) {} ) {}
async enqueuePersonaRecompute( async enqueuePersonaRecompute(
...@@ -90,6 +93,41 @@ export class QueueProducer { ...@@ -90,6 +93,41 @@ export class QueueProducer {
); );
} }
/**
* 整包 cold-import 入队。返回 jobId 作前端轮询句柄(不落 syncLog —— importDirectory 自己会写
* 自己的 syncLog 供审计/最近事件;前端进度直接查 BullMQ job 的 state + returnvalue)。
* concurrency=1 串行 + importDirectory 自带 running 锁 → 同 host 不会并发跑。
*/
async enqueueColdImport(job: ColdImportJob): Promise<{ jobId: string }> {
const jobId = randomUUID();
await this.coldImportQ.add(job.hostName, job, {
jobId,
...QueueDefaults[QueueName.COLD_IMPORT],
});
this.logger.log(
`enqueue cold-import job=${jobId} host=${job.hostName} triggeredBy=${job.triggeredBy}`,
);
return { jobId };
}
/** 查一次 cold-import job 进度(前端轮询用)。job 不存在返回 null(可能已被清理)。 */
async getColdImportJob(jobId: string): Promise<{
state: string;
hostId: string | null;
returnValue: unknown;
failedReason: string | null;
} | null> {
const job = await this.coldImportQ.getJob(jobId);
if (!job) return null;
const state = await job.getState();
return {
state,
hostId: (job.data as { hostId?: string } | undefined)?.hostId ?? null,
returnValue: job.returnvalue ?? null,
failedReason: job.failedReason ?? null,
};
}
async enqueueExecutionCallback(job: ExecutionCallbackJob): Promise<void> { async enqueueExecutionCallback(job: ExecutionCallbackJob): Promise<void> {
// jobId = executionId(append-only 唯一)→ 同一 execution 不会重复入队 // jobId = executionId(append-only 唯一)→ 同一 execution 不会重复入队
await this.callbackQ.add(job.executionId, job, { await this.callbackQ.add(job.executionId, job, {
......
apps/pac-service/src/queues/queues.module.ts
...@@ -17,6 +17,7 @@ import { PersonaRecomputeProcessor } from './processors/persona-recompute.proces ...@@ -17,6 +17,7 @@ import { PersonaRecomputeProcessor } from './processors/persona-recompute.proces
import { PlanRecomputeProcessor } from './processors/plan-recompute.processor'; import { PlanRecomputeProcessor } from './processors/plan-recompute.processor';
import { PlanAssetGenerateProcessor } from './processors/plan-asset-generate.processor'; import { PlanAssetGenerateProcessor } from './processors/plan-asset-generate.processor';
import { ExecutionCallbackProcessor } from './processors/execution-callback.processor'; import { ExecutionCallbackProcessor } from './processors/execution-callback.processor';
import { ColdImportProcessor } from './processors/cold-import.processor';
/** /**
* Queues — BullMQ 三个队列的统一注册入口。 * Queues — BullMQ 三个队列的统一注册入口。
...@@ -56,6 +57,7 @@ import { ExecutionCallbackProcessor } from './processors/execution-callback.proc ...@@ -56,6 +57,7 @@ import { ExecutionCallbackProcessor } from './processors/execution-callback.proc
{ name: QueueName.PLAN_RECOMPUTE }, { name: QueueName.PLAN_RECOMPUTE },
{ name: QueueName.PLAN_ASSET_GENERATE }, { name: QueueName.PLAN_ASSET_GENERATE },
{ name: QueueName.EXECUTION_CALLBACK }, { name: QueueName.EXECUTION_CALLBACK },
{ name: QueueName.COLD_IMPORT },
), ),
PersonaModule, PersonaModule,
PlanModule, PlanModule,
...@@ -73,6 +75,7 @@ import { ExecutionCallbackProcessor } from './processors/execution-callback.proc ...@@ -73,6 +75,7 @@ import { ExecutionCallbackProcessor } from './processors/execution-callback.proc
PlanRecomputeProcessor, PlanRecomputeProcessor,
PlanAssetGenerateProcessor, PlanAssetGenerateProcessor,
ExecutionCallbackProcessor, ExecutionCallbackProcessor,
ColdImportProcessor,
], ],
exports: [BullModule, QueueProducer, StaleScanService, SyncIncrementalSchedulerService], exports: [BullModule, QueueProducer, StaleScanService, SyncIncrementalSchedulerService],
}) })
......
apps/pac-service/src/types/yauzl.d.ts 0 → 100644
// yauzl 无官方 @types —— 遵循仓库"不引额外类型包"的先例(见 assistant.controller、mcp-sdk.d.ts),
// 只声明本项目实际用到的最小 API 面。
declare module 'yauzl' {
import type { EventEmitter } from 'node:events';
import type { Readable } from 'node:stream';
export interface Entry {
fileName: string;
}
export interface ZipFile extends EventEmitter {
readEntry(): void;
openReadStream(
entry: Entry,
callback: (err: Error | null, stream?: Readable) => void,
): void;
}
export interface OpenOptions {
lazyEntries?: boolean;
}
export function open(
path: string,
options: OpenOptions,
callback: (err: Error | null, zipfile?: ZipFile) => void,
): void;
}
apps/pac-service/tests/cold-import-extract.spec.ts 0 → 100644
/**
* cold-import 上传解压 — extractWhitelisted 白名单 + zip-slip 防护
*
* 覆盖:
* - 白名单过滤:zip 里白名单外的文件(c.csv)不提取
* - 子目录条目按 basename 落平(sub/nested.csv → nested.csv,若在白名单)
* - 缺文件:zip 缺白名单声明的文件 → extracted 集合据此可算出缺哪些
* - zip-slip:含 `../evil.csv` 路径穿越条目的包被 yauzl fail-closed **整包拒绝**(reject)
*
* fixtures 由 tests/fixtures/cold-import/*.zip 提供(python3 zipfile 预造,含合法 arcname `../`)。
*
* 跑:pnpm --filter @pac/service test -- cold-import-extract
*/
import * as fs from 'node:fs';
import * as os from 'node:os';
import * as path from 'node:path';
import { extractWhitelisted } from '../src/modules/admin/import-upload.service';
const FIX = path.join(__dirname, 'fixtures', 'cold-import');
function tmpDir(): string {
return fs.mkdtempSync(path.join(os.tmpdir(), 'cold-import-test-'));
}
describe('extractWhitelisted', () => {
it('只提取白名单命中的文件,忽略白名单外条目', async () => {
const dest = tmpDir();
try {
const got = await extractWhitelisted(
path.join(FIX, 'good.zip'),
dest,
new Set(['a.csv', 'b.csv']),
);
expect([...got].sort()).toEqual(['a.csv', 'b.csv']);
expect(fs.existsSync(path.join(dest, 'a.csv'))).toBe(true);
expect(fs.existsSync(path.join(dest, 'b.csv'))).toBe(true);
} finally {
fs.rmSync(dest, { recursive: true, force: true });
}
});
it('缺文件:zip 缺声明文件 → extracted 不含它(供缺文件校验)', async () => {
const dest = tmpDir();
try {
const whitelist = new Set(['a.csv', 'b.csv']);
const got = await extractWhitelisted(path.join(FIX, 'partial.zip'), dest, whitelist);
const missing = [...whitelist].filter((f) => !got.has(f));
expect(missing).toEqual(['b.csv']);
} finally {
fs.rmSync(dest, { recursive: true, force: true });
}
});
it('zip-slip:含 `../` 路径穿越条目的包被整包拒绝,且父目录零写入', async () => {
const parent = tmpDir();
const dest = path.join(parent, 'inner');
fs.mkdirSync(dest);
try {
await expect(
extractWhitelisted(path.join(FIX, 'evil.zip'), dest, new Set(['a.csv', 'evil.csv'])),
).rejects.toThrow(/relative path|invalid/i);
// 关键:父目录绝无 evil.csv 写入(未逃逸)
expect(fs.existsSync(path.join(parent, 'evil.csv'))).toBe(false);
} finally {
fs.rmSync(parent, { recursive: true, force: true });
}
});
it('白名单外文件(c.csv)不落盘,子目录条目按 basename 落平', async () => {
const dest = tmpDir();
try {
const got = await extractWhitelisted(
path.join(FIX, 'extra.zip'),
dest,
new Set(['a.csv', 'nested.csv']),
);
expect([...got].sort()).toEqual(['a.csv', 'nested.csv']);
expect(fs.existsSync(path.join(dest, 'a.csv'))).toBe(true);
expect(fs.existsSync(path.join(dest, 'nested.csv'))).toBe(true); // sub/nested.csv → nested.csv
expect(fs.existsSync(path.join(dest, 'c.csv'))).toBe(false); // 白名单外
} finally {
fs.rmSync(dest, { recursive: true, force: true });
}
});
});
apps/pac-service/tests/cold-import-groups.spec.ts 0 → 100644
/**
* cold-import 成组完整性 — findIncompleteGroups
*
* 规则(用户定):同 manifest group 的表在网页上传时要么全在、要么整组不在;部分在 → 拒(半组无意义)。
* 独立表(无 group)缺失不影响(装载层宽容跳过)。
*
* 跑:pnpm --filter @pac/service test -- cold-import-groups
*/
import { findIncompleteGroups } from '../src/modules/admin/import-upload.service';
// 模拟 FRIDAY manifest.tables 的关键子集:treatment_plan 组(头+行)+ 若干独立表
const TABLES = [
{ file: 'customer_basic_info.csv' }, // 独立
{ file: 'appointment_base.csv' }, // 独立
{ file: 'customer_treat_plan.csv', group: 'treatment_plan' },
{ file: 'customer_treat_plan_item.csv', group: 'treatment_plan' },
];
describe('findIncompleteGroups', () => {
it('组内全在 → 不报(通过)', () => {
const got = findIncompleteGroups(
TABLES,
new Set(['customer_basic_info.csv', 'customer_treat_plan.csv', 'customer_treat_plan_item.csv']),
);
expect(got).toEqual([]);
});
it('整组都不在 → 不报(宽容:该主体不导)', () => {
const got = findIncompleteGroups(TABLES, new Set(['customer_basic_info.csv']));
expect(got).toEqual([]);
});
it('组内部分在(缺行表)→ 报不完整', () => {
const got = findIncompleteGroups(
TABLES,
new Set(['customer_basic_info.csv', 'customer_treat_plan.csv']), // 缺 _item
);
expect(got).toHaveLength(1);
expect(got[0]!.group).toBe('treatment_plan');
expect(got[0]!.missing).toEqual(['customer_treat_plan_item.csv']);
expect(got[0]!.present).toEqual(['customer_treat_plan.csv']);
});
it('独立表缺失不影响(无 group 不参与成组)', () => {
const got = findIncompleteGroups(
TABLES,
new Set(['customer_treat_plan.csv', 'customer_treat_plan_item.csv']), // 缺两张独立表,但组齐
);
expect(got).toEqual([]);
});
});
apps/pac-service/tests/filter-numeric.spec.ts 0 → 100644
/**
* filter 数值比较算子 — lt/lte/gt/gte
*
* 动机:让"业务口径 WHERE"(如退费正负切分 receivable_this < 0)能在 PAC transforms 做,
* 宿主只需导原表。非数值/空值一律不通过(空金额不算负数)。
*
* 跑:pnpm --filter @pac/service test -- filter-numeric
*/
import { runFilter } from '../src/modules/sync/transforms/operators/filter.op';
import type { FilterOp } from '../src/modules/sync/transforms/transforms.schema';
const rows = [
{ m: '-30.00' },
{ m: '0' },
{ m: '5' },
{ m: '' }, // 空
{ m: 'abc' }, // 非数值
{ m: null as unknown as string },
];
function filt(where: FilterOp['where']) {
const op = { kind: 'filter', input: 'x', output: 'y', where } as FilterOp;
return runFilter(op, rows).rows.map((r) => r.m);
}
describe('filter 数值算子', () => {
it('lt:< 0 只保留负数,空/非数值不通过', () => {
expect(filt({ m: { lt: 0 } })).toEqual(['-30.00']);
});
it('gte:>= 0 保留 0 和正数', () => {
expect(filt({ m: { gte: 0 } })).toEqual(['0', '5']);
});
it('gt:> 0 只保留正数', () => {
expect(filt({ m: { gt: 0 } })).toEqual(['5']);
});
it('lte:<= 0 保留 0 和负数', () => {
expect(filt({ m: { lte: 0 } })).toEqual(['-30.00', '0']);
});
it('空值 / 非数值 / null 一律不通过数值比较', () => {
// 用一个恒真下界也不会把空/非数值放进来
expect(filt({ m: { gte: -1e9 } })).toEqual(['-30.00', '0', '5']);
});
});
apps/pac-service/tests/friday-settlement-where.spec.ts 0 → 100644
/**
* FRIDAY 结算/病历 WHERE 迁移到 PAC transforms —— 验证"导原表 → PAC 侧 filter 切分"正确。
*
* 背景:业务 WHERE 从 export.sh 全部搬进 manifest transforms(单一真理源)。
* 本测用合成的**原表**数据(全 status + 正负金额)喂 TransformEngine,断言四处切分:
* ① 消费 status∈{1,3} 且 receivable≥0(负额被 gte 挡)
* ② 退费① status=4 ∪ (status=3 且 receivable<0) ← filter×2 + union
* ③ 退费明细 is_refund=1
* ④ 病历 status∈{3,4} 排草稿
*
* 跑:pnpm --filter @pac/service test -- friday-settlement-where
*/
import { TransformEngine } from '../src/modules/sync/transforms/transform-engine';
import type { TransformsConfig } from '../src/modules/sync/transforms/transforms.schema';
const engine = new TransformEngine();
// 合成原表(CSV 源天然全字符串;med_emr_info 是 JSON,status 用数字测 in 的 String 归一)
const patient_settlement = [
{ uuid: 's1', status: '1', receivable_this: '100', patient_id: '666716' }, // 消费
{ uuid: 's3', status: '3', receivable_this: '50', patient_id: '666716' }, // 消费(status3 正额)
{ uuid: 's1neg', status: '1', receivable_this: '-5', patient_id: '666716' }, // 消费但金额<0 → 被 gte 挡
{ uuid: 's4', status: '4', receivable_this: '-30', patient_id: '666716' }, // 退费(status4 整单反向)
{ uuid: 's3neg', status: '3', receivable_this: '-20', patient_id: '666716' }, // 退费(status3 负额表达)
{ uuid: 's2', status: '2', receivable_this: '10', patient_id: '666716' }, // 草稿/其他 → 都不收
];
// spec.patient_id 部分品牌原是「诊所本地垃圾 id」→ 宿主导出时已从结算头反填真 patient_id(inline);PAC 直用
const patient_settlement_spec = [
{ id: 'sp1', is_refund: '1', settlement_id: 's3', patient_id: '666716' }, // 退费明细(宿主已反填真 patient_id)
{ id: 'sp2', is_refund: '0', settlement_id: 's1', patient_id: '666716' }, // 非退费
];
const med_emr_info = [
{ emr_sub_id: 'e3', status: 3 }, // 正式
{ emr_sub_id: 'e4', status: 4 }, // 正式
{ emr_sub_id: 'e2', status: 2 }, // 草稿 → 排除
];
const transforms = [
{ kind: 'filter', input: 'patient_settlement', output: '_pay_hdr', where: { status: { in: ['1', '3'] }, receivable_this: { gte: 0 } } },
{ kind: 'filter', input: 'patient_settlement', output: '_refund_status4', where: { status: { equals: '4' } } },
{ kind: 'filter', input: 'patient_settlement', output: '_refund_neg3', where: { status: { equals: '3' }, receivable_this: { lt: 0 } } },
{ kind: 'union', inputs: ['_refund_status4', '_refund_neg3'], output: 'refund_full_rows' },
{ kind: 'filter', input: 'patient_settlement_spec', output: 'refund_item_rows', where: { is_refund: { equals: '1' } } },
{ kind: 'filter', input: 'med_emr_info', output: '_emr_official', where: { status: { in: ['3', '4'] } } },
] as unknown as TransformsConfig;
describe('FRIDAY 结算/病历 WHERE 迁移(导原表 + PAC 切分)', () => {
const out = engine.run({
tables: { patient_settlement, patient_settlement_spec, med_emr_info },
transforms,
});
const pick = (rows: unknown[], key: string) =>
(rows as Array<Record<string, unknown>>).map((r) => r[key]).sort();
it('① 消费:status∈{1,3} 且 receivable≥0(负额 s1neg 被挡,s2/s4 不在)', () => {
expect(pick(out['_pay_hdr'] ?? [], 'uuid')).toEqual(['s1', 's3']);
});
it('② 退费①:status=4 ∪ (status=3 且金额<0)', () => {
expect(pick(out['refund_full_rows'] ?? [], 'uuid')).toEqual(['s3neg', 's4']);
});
it('③ 退费明细:is_refund=1', () => {
expect(pick(out['refund_item_rows'] ?? [], 'id')).toEqual(['sp1']);
});
it('③.1 退费明细患者 = 宿主反填的真 patient_id(spec.patient_id 直用,不再跨表 lookup)', () => {
const rows = (out['refund_item_rows'] ?? []) as Array<Record<string, unknown>>;
expect(rows.map((r) => r['patient_id'])).toEqual(['666716']);
});
it('④ 病历:status∈{3,4} 排草稿 e2', () => {
expect(pick(out['_emr_official'] ?? [], 'emr_sub_id')).toEqual(['e3', 'e4']);
});
});
apps/pac-service/tests/trace-raw-source.spec.ts 0 → 100644
/**
* traceRawSourceTable —— 回溯 assembler primary 表到「原始源表」,用于 push(source→匹配 assembler)
* 与 reparse(rawPayload 属哪张源表)。重点回归:**union 多输入必须能收敛到同一根**,
* 否则 union-primary 资源(refund_full)在 push/reparse 里被静默跳过(丢财务数据)。
*
* 跑:pnpm --filter @pac/service test -- trace-raw-source
*/
import { traceRawSourceTable } from '../src/modules/sync/cold-import/cold-import.service';
// 结算链最小 transforms:消费(filter 链)+ 退费整单(filter×2 + union)+ 退费明细(filter+lookup)
const transforms = [
{ kind: 'filter', input: 'patient_settlement', output: '_pay_hdr', where: {} },
{ kind: 'lookup', input: '_pay_hdr', output: 'payment_rows', from: 'settlement_modes', select: {} },
{ kind: 'filter', input: 'patient_settlement', output: '_refund_status4', where: {} },
{ kind: 'filter', input: 'patient_settlement', output: '_refund_neg3', where: {} },
{ kind: 'union', inputs: ['_refund_status4', '_refund_neg3'], output: 'refund_full_rows' },
{ kind: 'filter', input: 'patient_settlement_spec', output: '_refund_item_raw', where: {} },
{ kind: 'lookup', input: '_refund_item_raw', output: 'refund_item_rows', from: 'patient_settlement', select: {} },
// route_by_pattern 多输出回同一 input
{ kind: 'route_by_pattern', input: '_treat_raw', outputs: [{ output: '_actual_raw' }, { output: '_rec_raw' }] },
];
describe('traceRawSourceTable', () => {
it('单输入链:payment_rows → patient_settlement', () => {
expect(traceRawSourceTable('payment_rows', transforms)).toBe('patient_settlement');
});
it('union 收敛:refund_full_rows 两路都回到 patient_settlement(修复点)', () => {
expect(traceRawSourceTable('refund_full_rows', transforms)).toBe('patient_settlement');
});
it('lookup 链取 input(不取 from):refund_item_rows → patient_settlement_spec', () => {
expect(traceRawSourceTable('refund_item_rows', transforms)).toBe('patient_settlement_spec');
});
it('route 多输出回同一 input:_actual_raw → _treat_raw', () => {
expect(traceRawSourceTable('_actual_raw', transforms)).toBe('_treat_raw');
});
it('已是根表:原样返回', () => {
expect(traceRawSourceTable('patient_settlement', transforms)).toBe('patient_settlement');
});
it('union 输入发散到多根 → 停在 union 输出(不误判)', () => {
const diverge = [
{ kind: 'filter', input: 'table_a', output: '_a', where: {} },
{ kind: 'filter', input: 'table_b', output: '_b', where: {} },
{ kind: 'union', inputs: ['_a', '_b'], output: 'merged' },
];
expect(traceRawSourceTable('merged', diverge)).toBe('merged');
});
});
apps/pac-web/src/components/host-admin/cold-import-card.tsx 0 → 100644
'use client';
import { useEffect, useRef, useState } from 'react';
import { Upload, Loader2 } from 'lucide-react';
import { toast } from 'sonner';
import { Card, CardContent, CardDescription, CardHeader, CardTitle } from '@/components/ui/card';
import { Button } from '@/components/ui/button';
import { hostApi, type ColdImportStatus } from './host-api';
/** BullMQ 终态:轮询到这些状态就停。 */
const TERMINAL = new Set(['completed', 'failed', 'not_found']);
const POLL_MS = 3000;
const STATE_LABEL: Record<string, string> = {
waiting: '排队中',
delayed: '排队中',
active: '导入中',
completed: '完成',
failed: '失败',
not_found: '已结束',
};
function fmtElapsed(sec: number): string {
const m = Math.floor(sec / 60);
const s = sec % 60;
return m > 0 ? `${m}分${s}秒` : `${s}秒`;
}
/**
* ColdImportCard — 宿主自助上传数据压缩包触发 cold-import。
*
* 流程:选 zip → 上传拿 jobId → 每 3s 轮询进度 → 终态停 + 展示 totals + 刷新最近事件。
* manifest/assemblers 取自 PAC 代码库,上传包只含数据文件(见后端 ImportUploadService)。
*/
export function ColdImportCard({ onDone }: { onDone: () => void }) {
const inputRef = useRef<HTMLInputElement>(null);
const [file, setFile] = useState<File | null>(null);
const [dryRun, setDryRun] = useState(true); // 默认先预检,更安全
const [uploading, setUploading] = useState(false);
const [jobId, setJobId] = useState<string | null>(null);
const [status, setStatus] = useState<ColdImportStatus | null>(null);
const [startedAt, setStartedAt] = useState<number | null>(null);
const [elapsed, setElapsed] = useState(0);
// onDone 存 ref,避免父组件重渲染导致轮询 effect 被重置
const onDoneRef = useRef(onDone);
useEffect(() => {
onDoneRef.current = onDone;
}, [onDone]);
// 轮询:仅依赖 jobId
useEffect(() => {
if (!jobId) return;
let timer: ReturnType<typeof setInterval> | null = null;
let stopped = false;
const poll = async () => {
try {
const s = await hostApi.getImportStatus(jobId);
if (stopped) return;
setStatus(s);
if (TERMINAL.has(s.state)) {
if (timer) clearInterval(timer);
timer = null;
if (s.state === 'completed') {
if (s.result?.dryRun) {
// 预检:未落库,不刷新最近事件
toast.success('预检完成(未落库),检查下方 failed / 数量再正式导入');
} else {
toast.success('导入完成');
onDoneRef.current();
}
} else if (s.state === 'failed') {
toast.error(`导入失败:${s.failedReason ?? '未知原因'}`);
} else {
toast.message('任务已不在队列(可能已完成并被清理)');
}
}
} catch {
// 单次轮询失败不打断,等下一轮
}
};
void poll();
timer = setInterval(() => void poll(), POLL_MS);
return () => {
stopped = true;
if (timer) clearInterval(timer);
};
}, [jobId]);
const running = jobId != null && (!status || !TERMINAL.has(status.state));
// 已用时长:running 时每秒 tick
useEffect(() => {
if (!running || startedAt == null) return;
const t = setInterval(() => setElapsed(Math.floor((Date.now() - startedAt) / 1000)), 1000);
return () => clearInterval(t);
}, [running, startedAt]);
const onUpload = async () => {
if (!file) return;
setUploading(true);
setStatus(null);
setJobId(null);
try {
const form = new FormData();
form.append('file', file);
form.append('dryRun', String(dryRun));
const { jobId: id } = await hostApi.importArchive(form);
setJobId(id);
setStartedAt(Date.now());
setElapsed(0);
toast.success(dryRun ? '已上传,开始预检(不落库)' : '已上传,开始后台导入');
} catch (e) {
toast.error(e instanceof Error ? e.message : '上传失败');
} finally {
setUploading(false);
}
};
const t = status?.result?.totals;
return (
<Card>
<CardHeader>
<CardTitle>数据导入</CardTitle>
<CardDescription>
上传数据压缩包(zip,含 manifest 声明的 CSV/JSON)触发整包 cold-import。manifest / assemblers
取自 PAC 侧无需上传。重复上传同一批数据是安全的(幂等空转:去重高、fact 新建≈0)。
</CardDescription>
</CardHeader>
<CardContent className="space-y-3">
<input
ref={inputRef}
type="file"
accept=".zip"
className="hidden"
onChange={(e) => setFile(e.target.files?.[0] ?? null)}
/>
<div className="flex flex-wrap items-center gap-3">
<Button
variant="outline"
onClick={() => inputRef.current?.click()}
disabled={uploading || running}
>
选择 zip
</Button>
<span className="min-w-0 flex-1 truncate text-sm text-muted-foreground">
{file ? file.name : '未选择文件'}
</span>
<label className="flex items-center gap-1.5 text-sm text-muted-foreground">
<input
type="checkbox"
checked={dryRun}
onChange={(e) => setDryRun(e.target.checked)}
disabled={uploading || running}
/>
仅预检(不落库)
</label>
<Button onClick={onUpload} disabled={!file || uploading || running}>
{uploading ? (
<Loader2 className="mr-2 h-4 w-4 animate-spin" />
) : (
<Upload className="mr-2 h-4 w-4" />
)}
{dryRun ? '预检' : '上传并导入'}
</Button>
</div>
{jobId && (
<div className="rounded-md border p-3 text-sm">
<div className="flex items-center gap-2">
{running && <Loader2 className="h-4 w-4 animate-spin" />}
<span>
状态:{STATE_LABEL[status?.state ?? ''] ?? status?.state ?? '排队中'}
{status?.result?.dryRun ? '(预检)' : ''}
</span>
{(running || elapsed > 0) && (
<span className="text-xs text-muted-foreground">· 用时 {fmtElapsed(elapsed)}</span>
)}
</div>
{/* 运行中真实进度条(cohortDone/cohortTotal 批次) */}
{status?.progress && status.progress.total > 0 && (
<div className="mt-2">
<div className="h-1.5 w-full overflow-hidden rounded bg-muted">
<div
className="h-full bg-primary transition-all"
style={{
width: `${Math.round((status.progress.done / status.progress.total) * 100)}%`,
}}
/>
</div>
<div className="mt-1 text-xs text-muted-foreground">
已处理 {status.progress.done}/{status.progress.total} 批
</div>
</div>
)}
{/* 长任务提示:后台跑,可离开 */}
{running && (
<p className="mt-1.5 text-xs text-muted-foreground">
导入在后台进行,可离开此页,稍后回来查看结果。
</p>
)}
{status?.state === 'failed' && status.failedReason && (
<p className="mt-1 text-destructive">失败原因:{status.failedReason}</p>
)}
{t && (
<div className="mt-2 grid grid-cols-2 gap-x-4 gap-y-1 text-muted-foreground">
<span>患者 upsert:{t.patientsUpserted}</span>
<span>事务写入:{t.transactionsWritten}</span>
<span>幂等去重:{t.duplicates}</span>
<span>失败:{t.failed}</span>
<span>fact 新建:{t.factsCreated}</span>
<span>fact 更新:{t.factsSuperseded}</span>
<span>fact 未变:{t.factsUnchanged}</span>
<span>证据追加:{t.factsEvidenceAppended}</span>
</div>
)}
</div>
)}
</CardContent>
</Card>
);
}
apps/pac-web/src/components/host-admin/host-admin-app.tsx
...@@ -24,6 +24,7 @@ import { isPostMessageSentinel } from '@/lib/host-message'; ...@@ -24,6 +24,7 @@ import { isPostMessageSentinel } from '@/lib/host-message';
import { hostApi } from './host-api'; import { hostApi } from './host-api';
import { useConfirm } from './use-confirm'; import { useConfirm } from './use-confirm';
import { SecretRevealDialog } from './secret-reveal-dialog'; import { SecretRevealDialog } from './secret-reveal-dialog';
import { ColdImportCard } from './cold-import-card';
/** /**
* HostAdminApp — 宿主自己的 admin 配置页主视图。 * HostAdminApp — 宿主自己的 admin 配置页主视图。
...@@ -248,6 +249,9 @@ export function HostAdminApp({ data, onRefresh }: { data: HostDetail; onRefresh: ...@@ -248,6 +249,9 @@ export function HostAdminApp({ data, onRefresh }: { data: HostDetail; onRefresh:
<StatsCard data={data} /> <StatsCard data={data} />
{/* 数据导入(上传压缩包触发 cold-import) */}
<ColdImportCard onDone={onRefresh} />
{/* 凭据 */} {/* 凭据 */}
<Card> <Card>
<CardHeader> <CardHeader>
......
apps/pac-web/src/components/host-admin/host-api.ts
...@@ -13,6 +13,28 @@ import { api } from '@/lib/api-client'; ...@@ -13,6 +13,28 @@ import { api } from '@/lib/api-client';
const BASE = '/pac/v1/admin/host/self'; const BASE = '/pac/v1/admin/host/self';
export interface ColdImportTotals {
patientsUpserted: number;
transactionsWritten: number;
duplicates: number;
failed: number;
factsCreated: number;
factsSuperseded: number;
factsUnchanged: number;
factsEvidenceAppended: number;
factsFailed: number;
}
export interface ColdImportStatus {
/** BullMQ job 状态:waiting/active/completed/failed/...;'not_found' = 已清理或不存在 */
state: string;
/** 完成时的 ImportRunResult(含 totals + dryRun 标志);进行中为 null */
result: { status?: string; dryRun?: boolean; totals?: ColdImportTotals } | null;
failedReason: string | null;
/** 运行中的真实进度(cohortDone/cohortTotal 批次);非运行态或 dryRun 为 null */
progress: { done: number; total: number } | null;
}
export const hostApi = { export const hostApi = {
getSelf: () => api.get<HostDetail>(BASE), getSelf: () => api.get<HostDetail>(BASE),
getStats: () => api.get<HostStats>(`${BASE}/stats`), getStats: () => api.get<HostStats>(`${BASE}/stats`),
...@@ -28,4 +50,9 @@ export const hostApi = { ...@@ -28,4 +50,9 @@ export const hostApi = {
api.delete<{ totalKeysRemaining: number }>( api.delete<{ totalKeysRemaining: number }>(
`${BASE}/callback-secrets/${encodeURIComponent(suffix)}`, `${BASE}/callback-secrets/${encodeURIComponent(suffix)}`,
), ),
// 数据导入:上传压缩包(FormData 字段名 file)→ 返回 jobId;按 jobId 轮询进度
importArchive: (form: FormData) => api.post<{ jobId: string }>(`${BASE}/cold-import`, form),
getImportStatus: (jobId: string) =>
api.get<ColdImportStatus>(`${BASE}/cold-import/${encodeURIComponent(jobId)}`),
}; };
pnpm-lock.yaml
...@@ -235,6 +235,9 @@ importers: ...@@ -235,6 +235,9 @@ importers:
ws: ws:
specifier: ^8.21.0 specifier: ^8.21.0
version: 8.21.0 version: 8.21.0
yauzl:
specifier: ^3.3.0
version: 3.3.0
zod: zod:
specifier: ^4.4.3 specifier: ^4.4.3
version: 4.4.3 version: 4.4.3
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment