Commit ee1ecaa7 by luoqi

merge: sync main into host writeback development base

parents b6cee831 9f7ec1a4
--- ---
title: FRIDAY 推送数据契约 title: FRIDAY 推送数据契约
description: FRIDAY SaaS 按形态 A 推送的 9 个 source 及字段定义;宿主推自洽的业务表(自己的引用 inline 进相关行),PAC 做跨宿主临床归一。 description: FRIDAY SaaS 按形态 A 推送的 10 个 source 及字段定义;宿主推自洽的业务表(自己的引用 inline 进相关行),PAC 做跨宿主临床归一。
icon: FileJson icon: FileJson
--- ---
...@@ -290,11 +290,57 @@ icon: FileJson ...@@ -290,11 +290,57 @@ icon: FileJson
--- ---
## 7. 对接自测(预检 + 推送记录) ## 7. 回访(1 个 source)
### `customer_return_visit` — 客户回访(任务属性已 inline)
> **回访任务 `customer_task` 不再单独推**——它是**统一任务表**,按 `task_type` 分流
> (`1`咨询 → `customer_consult`、`2`回访 → 本表、`4`预约备注、`10`…),PAC 只要回访那一支的
> 「排程日期 + 任务状态」两个属性,任务头本身不是 PAC 实体。宿主把 `customer_task` 的
> 相关列 **inline 进每条回访记录**即可(关联:`customer_return_visit.task_id = customer_task.id`,
> 实测 2552/2552 全中、零空值)。这与计划行 inline 计划头属性是同一形态。
>
> ⚠️ **`task_date` 含未来排程**——界面「设回访」就是设在未来日期。PAC 据此区分
> 「已发生的回访」与「排了但还没做的」:详情页按它倒序展示,召回话术**只把已发生的**
> 算作"联系过"。**不要为了让数据好看而把未来日期改成当天**。
| 字段 | 类型 | 必填 | 说明 |
|---|---|---|---|
| `id` | string | ✅ | 回访记录主键(= PAC externalId,幂等键)|
| `tenant_id` | string | ✅ | 品牌 GUID |
| `organization_id` | string | ✅ | 诊所 |
| `customer_id` | string | ✅ | → 患者 id |
| `task_id` | string | | → `customer_task.id`(留作溯源)|
| `task_date` | string(date) | | **任务日期(宿主 inline,取自 `customer_task`)——含未来排程** |
| `task_status` | string/number | | **任务状态(宿主 inline)**——`1`进行中 `2`已完成 `3`未完成 `4`已预约 `5`创建新回访 `6`未回访 |
| `return_visit_type` | string/number | | 回访类型——`1`咨询回访 `2`术后回访 `3`常规回访 `4`自定义/事件驱动(含「取消预约回访」)`5`系统诊断召回。⚠️ 官方注释只写了 1-3,**4/5 是实测存在的**(占 12%) |
| `return_visit_status` | string/number | | 回访状态——`1`已回访 `2`未回访 |
| `treatment_items` | string | | 关联治疗项(大类,如「种植」)|
| `treatment_items_two` | string | | 关联治疗项(子项,如「单颗种植」)。⚠️ 与上一列**成对出现**,PAC 合并成单字段展示 |
| `follow_content` | string | | 回访任务内容。⚠️ **可能含 HTML 富文本**(界面是富文本编辑器),PAC 摄入时剥标签 |
| `return_visit_result` | string | | 回访结果 |
| `task_director_id` / `task_director` | string | | **执行本次回访的人**。⚠️ 见下方说明 |
| `return_visit_method` | string/number | | 回访方式——`1`打电话 `2`发微信 `3`发短信 |
| `created_gmt_at` / `updated_gmt_at` | string(datetime) | ✅ | 创建 / 末次修改时间 |
**建议一并 inline 的 `customer_task` 列**(PAC 当前不映射,原样存进 `raw_payload` 备用,
将来要用不必再找宿主改):`suggested_return_person`(建议回访人)、`actual_return_person` /
`actual_return_time`(实际回访人/时间)、`customer_status`(客户状态)、`is_first`(初诊/复诊)。
> **`task_director` 是「该任务当时派给谁」,不是患者的专属客服。**
> PAC 用它**按诊所反推客服名册**供主管指派(宿主不提供「客服归属诊所」字段,只能按行为反推)。
> 参照宿主 jvs-dw 实测:近 12 月 367 万对回访,「任务负责人」与「患者主档专属客服」仅 **23.1%** 相同。
>
> ⚠️ **`task_director` 可能是 `System`**(系统自动生成的回访,实测 158 条、全部是
> `return_visit_type=5` 的诊断召回)。PAC 侧**不会把 `System` 计入客服名册** —— 它不是人。
> 宿主照推即可,不需要过滤。
## 8. 对接自测(预检 + 推送记录)
对接期不必盲推 —— PAC 提供两个自助工具,**测试环境** `https://pac.jarvismedical.asia`: 对接期不必盲推 —— PAC 提供两个自助工具,**测试环境** `https://pac.jarvismedical.asia`:
### 7.1 预检:`?dryRun=1`(强烈建议先跑) ### 8.1 预检:`?dryRun=1`(强烈建议先跑)
``` ```
POST /pac/v1/push/rows?dryRun=1 POST /pac/v1/push/rows?dryRun=1
...@@ -315,7 +361,7 @@ POST /pac/v1/push/rows?dryRun=1 ...@@ -315,7 +361,7 @@ POST /pac/v1/push/rows?dryRun=1
**自测通过标准**:`failed=0` 且 `mappingMisses=0` 且 `suspectFields=0`,再看 `samples` 里的字段值对不对 —— 都 OK 才去掉 `?dryRun=1` 正式推。 **自测通过标准**:`failed=0` 且 `mappingMisses=0` 且 `suspectFields=0`,再看 `samples` 里的字段值对不对 —— 都 OK 才去掉 `?dryRun=1` 正式推。
### 7.2 推送记录:事后可查(两种查法) ### 8.2 推送记录:事后可查(两种查法)
**① 免登录 —— 用已有 push 凭据验签(联调推荐)** **① 免登录 —— 用已有 push 凭据验签(联调推荐)**
``` ```
...@@ -335,7 +381,7 @@ GET /pac/v1/admin/host/self/push-logs?limit=50 ...@@ -335,7 +381,7 @@ GET /pac/v1/admin/host/self/push-logs?limit=50
`source / status / dryRun / fetched / transactionsWritten / duplicates / failed / errorMessage`。 `source / status / dryRun / fetched / transactionsWritten / duplicates / failed / errorMessage`。
当时的同步响应没留存也能事后查:**哪批错了、错在哪**。预检批次由 `dryRun:true` 标识,不与正式推送混淆。 当时的同步响应没留存也能事后查:**哪批错了、错在哪**。预检批次由 `dryRun:true` 标识,不与正式推送混淆。
### 7.3 其他可用面 ### 8.3 其他可用面
| 用途 | 地址 | | 用途 | 地址 |
|---|---| |---|---|
...@@ -350,7 +396,7 @@ GET /pac/v1/admin/host/self/push-logs?limit=50 ...@@ -350,7 +396,7 @@ GET /pac/v1/admin/host/self/push-logs?limit=50
--- ---
## 8. 语义澄清记录(2026-07,源码 + 数据双证) ## 9. 语义澄清记录(2026-07,源码 + 数据双证)
以下曾是开放问题,现已定案 —— 推送方无需再确认,列此备查: 以下曾是开放问题,现已定案 —— 推送方无需再确认,列此备查:
......
...@@ -42,6 +42,16 @@ POSTGRES_DB=pac ...@@ -42,6 +42,16 @@ POSTGRES_DB=pac
# #
# staging: postgresql://pac:<staging-pwd>@<staging-pg-host>:5532/pac?schema=public # staging: postgresql://pac:<staging-pwd>@<staging-pg-host>:5532/pac?schema=public
# production: postgresql://pac:<prod-pwd>@<prod-pg-host>:5532/pac?schema=public # production: postgresql://pac:<prod-pwd>@<prod-pg-host>:5532/pac?schema=public
#
# ⚠️ 连接池:URL 里**不写** connection_limit 时,PrismaService 会按并发旋钮自动放大
# (PAC_DB_CONCURRENCY / PAC_COHORT_CONCURRENCY / PAC_RECALL_SUBSCENARIO_CONCURRENCY 取最大,
# 池 = N×6+5 封顶 40);全都 ≤1 才落回 Prisma 默认池。
# Prisma 默认池是「**物理核**×2+1」,不是逻辑核 —— 生产 nproc=4 但物理核只有 2,默认只有 **5** 条。
# URL 里一旦显式写了 connection_limit,自动放大**整个失效**(逃生口,写了就得自己负责够用)。
# ⛔ 但 compose 模式下**在本文件里写 connection_limit 是没用的** —— compose 的 environment:
# 段会整条覆盖 DATABASE_URL。要确认到底生效的是什么,只能看
# `docker inspect <容器> --format '{{range .Config.Env}}{{println .}}{{end}}' | grep DATABASE_URL`。
# 2026-09-02 生产事故就出在这:开了召回并发却没算进池,批次占满 5 条,每日健康日报连续三天超时。
DATABASE_URL=postgresql://pac:pac@localhost:5532/pac?schema=public DATABASE_URL=postgresql://pac:pac@localhost:5532/pac?schema=public
# staging+prod: redis://<host>:6479 (BullMQ 队列用,丢了会丢未消费的 plan-asset-generate 任务) # staging+prod: redis://<host>:6479 (BullMQ 队列用,丢了会丢未消费的 plan-asset-generate 任务)
...@@ -214,6 +224,9 @@ SENTRY_RELEASE= ...@@ -214,6 +224,9 @@ SENTRY_RELEASE=
# 瓶颈是**延迟**(逐次索引探查在等 page 返回)不是磁盘吞吐,所以并发 2 就超线性(×1.50)。 # 瓶颈是**延迟**(逐次索引探查在等 page 返回)不是磁盘吞吐,所以并发 2 就超线性(×1.50)。
# ⚠️ 判据只能看 `[plan] 阶段耗时` 的**墙钟** —— 并发下单条查询的 sql= 会被争抢拉长, # ⚠️ 判据只能看 `[plan] 阶段耗时` 的**墙钟** —— 并发下单条查询的 sql= 会被争抢拉长,
# 看单条会误判成"变慢了"。详见 treatment-initiation-recall.scenario.ts 里 conc 处注释。 # 看单条会误判成"变慢了"。详见 treatment-initiation-recall.scenario.ts 里 conc 处注释。
# ⚠️ **这个值会放大数据库连接池**(见上面 DATABASE_URL 处):批次会长时间占住这么多条连接,
# 不放大池的话常驻服务的其他活(每日健康日报、API)就抢不到 —— 2026-09-02 实际发生过。
# URL 里显式写了 connection_limit 的机器不吃这套自动放大,得自己保证池够大。
PAC_RECALL_SUBSCENARIO_CONCURRENCY=4 PAC_RECALL_SUBSCENARIO_CONCURRENCY=4
# gap 计算形态:legacy(默认,逐行相关子查询)/ setbased(集合式)。 # gap 计算形态:legacy(默认,逐行相关子查询)/ setbased(集合式)。
......
# patient_return_visit — FRIDAY 客户回访(customer_return_visit → patient_return_visits 独立表)
#
# upsert 资源(无 emits,同 patient/patient_relation):落 PatientReturnVisit 表,
# **不进 transaction / fact、不参与召回**。用途:患者详情页「回访记录」展示 + 按诊所反推客服名册。
#
# ── 源表关系 ──
# 宿主是「统一任务表 + 分类型记录表」结构:
# customer_task(task_type: 1咨询 / 2回访 / 4预约备注 / 10…)
# ├─ task_type=1 → customer_consult (咨询,PAC 已单独摄入)
# └─ task_type=2 → customer_return_visit (回访,本 assembler)
# 关联 customer_return_visit.task_id = customer_task.id,实测 2552/2552 全中、零空值。
# 任务头不是 PAC 实体,故 **不单独摄入** —— 宿主把 task_date / task_status 等 inline 进回访行
# (同 customer_treat_plan_item inline 计划头的 organization_id/plan_name)。
#
# ── 枚举归一到中文,与 jvs-dw 对齐 ──
# jvs-dw 的 DW 侧本就是 *_name 中文列(常规回访/已回访/已完成…),FRIDAY 是数字码。
# 这里翻成**同样的中文**,否则同一张 PatientReturnVisit 表里两个宿主的值对不齐,
# 前端展示和按类型筛选都会分叉。
canonical: patient_return_visit
primary:
table: customer_return_visit
key: id
dedup_by: id
field_mapping:
externalId: id
patientExternalId: customer_id
clinicId: organization_id
# 排程日期(宿主 inline 自 customer_task)—— **含未来**,界面「设回访」就是设未来日期。
# PAC 据此区分「已发生」与「排了没做」:详情页倒序展示、召回话术只算已发生的。
# ⚠️ 客服名册判「在岗」用 sourceCreatedAt 而非本字段,正因为本字段含未来排程。
taskDate: task_date
taskStatus: task_status
type: return_visit_type
status: return_visit_status
# 大类·子项两列合并成单字段(同 jvs-dw)。实测:两列都有 499、只一级 257、只二级 0、都空 1796
# —— 与 jvs-dw「总是成对」同构;只有一级时合并结果尾部的分隔符由 normalizeMergedItems 归一掉。
treatmentItems: treatment_items_full
followContent: follow_content_text # 已剥 HTML(transforms;界面是富文本编辑器)
result: return_visit_result
# 执行本次回访的人(≠ 患者主档专属客服)。用于按诊所反推客服名册供主管指派。
# ⚠️ 可能是 'System'(系统自动生成的诊断召回,实测 158 条全是 return_visit_type=5),
# 不是真人 —— 名册侧已排除,见 transforms 说明。
taskDirectorId: task_director_id
taskDirectorName: task_director
# 宿主侧业务时间(≠ PAC 入库时间 created_at/updated_at,后者重摄会被刷新)
sourceCreatedAt: created_gmt_at
sourceUpdatedAt: updated_gmt_at
# 数字码 → 中文,与 jvs-dw 的 DW *_name 列取值逐字对齐
enum_mapping:
type:
'1': 咨询回访
'2': 术后回访
'3': 常规回访
# ⚠️ 官方注释只写了 1-3,但实测存在 4/5(占 12%,311/2552),不能落 _default 丢掉语义:
# 4 → 100% 带治疗项,取值含「取消预约回访」「自定义」→ 事件驱动的自定义回访
# 5 → 100% 带治疗项且全是**诊断名**(残根/龋齿/缺失牙/根尖周炎),task_director 全为 System
# → 系统按诊断自动生成的召回任务
'4': 自定义回访
'5': 系统召回
_default: ''
status:
'1': 已回访
'2': 未回访
_default: ''
taskStatus:
'1': 进行中
'2': 已完成
'3': 未完成
'4': 已预约
'5': 创建新回访
'6': 未回访
_default: ''
...@@ -146,6 +146,18 @@ mysql_csv "SELECT uuid,tenant_id,organization_id,patient_id,doctor_id,status,rec ...@@ -146,6 +146,18 @@ mysql_csv "SELECT uuid,tenant_id,organization_id,patient_id,doctor_id,status,rec
# 结算明细原表(全量含 is_refund,PAC 侧切退费明细)。JOIN 结算头:① 挂靠限定 cohort # 结算明细原表(全量含 is_refund,PAC 侧切退费明细)。JOIN 结算头:① 挂靠限定 cohort
# ② 反填真 patient_id(spec.patient_id 部分品牌是诊所本地 id,不可信 → 取结算头 ps.patient_id inline) # ② 反填真 patient_id(spec.patient_id 部分品牌是诊所本地 id,不可信 → 取结算头 ps.patient_id inline)
mysql_csv "SELECT sp.id,sp.tenant_id,sp.organization_id,ps.patient_id,sp.settlement_id,sp.cure_name,sp.service_project_name,sp.receivable_this,sp.net_receipts_this,sp.is_refund,sp.created_gmt_at,sp.updated_gmt_at FROM \`arrail-settlement-server\`.patient_settlement_spec sp JOIN \`arrail-settlement-server\`.patient_settlement ps ON ps.uuid=sp.settlement_id $(pf ps.patient_id)" patient_settlement_spec.csv mysql_csv "SELECT sp.id,sp.tenant_id,sp.organization_id,ps.patient_id,sp.settlement_id,sp.cure_name,sp.service_project_name,sp.receivable_this,sp.net_receipts_this,sp.is_refund,sp.created_gmt_at,sp.updated_gmt_at FROM \`arrail-settlement-server\`.patient_settlement_spec sp JOIN \`arrail-settlement-server\`.patient_settlement ps ON ps.uuid=sp.settlement_id $(pf ps.patient_id)" patient_settlement_spec.csv
# 客户回访 + inline 任务属性(取自统一任务表 customer_task 的 task_type=2 那一支;任务头不单独导)
# 关联 rv.task_id = t.id,实测 2552/2552 全中。task_date 含未来排程(界面「设回访」设未来日期),
# 不要在导出侧过滤掉未来行 —— PAC 靠它区分「已发生」与「排了没做」。
# 末尾几列 PAC 当前不映射,原样带出存进 raw_payload 备用(宿主多给无妨,PAC 选择性摄入)。
mysql_csv "SELECT rv.id,rv.tenant_id,rv.organization_id,rv.customer_id,rv.task_id,
t.task_date,t.task_status,
rv.return_visit_type,rv.return_visit_status,rv.treatment_items,rv.treatment_items_two,
rv.follow_content,rv.return_visit_result,rv.task_director_id,rv.task_director,
rv.return_visit_method,rv.created_gmt_at,rv.updated_gmt_at,
t.suggested_return_person,t.actual_return_person,t.actual_return_time,t.customer_status,t.is_first
FROM customer.customer_return_visit rv
LEFT JOIN customer.customer_task t ON t.id=rv.task_id AND t.task_type=2 $(pf rv.customer_id)" customer_return_visit.csv
# 转介绍圈(患者-患者关系边,双向成对存;referee_relationship 为宿主字典 id,PAC 侧统计推断解码) # 转介绍圈(患者-患者关系边,双向成对存;referee_relationship 为宿主字典 id,PAC 侧统计推断解码)
mysql_csv "SELECT id,tenant_id,organization_id,customer_id,referee_patient_id,referee_relationship,type,created_gmt_at,updated_gmt_at FROM customer.customer_referee_circle $(pf customer_id)" customer_referee_circle.csv mysql_csv "SELECT id,tenant_id,organization_id,customer_id,referee_patient_id,referee_relationship,type,created_gmt_at,updated_gmt_at FROM customer.customer_referee_circle $(pf customer_id)" customer_referee_circle.csv
# 咨询(意向 potential_treatment/未成交原因;测试库仅 45 行,结构真实) # 咨询(意向 potential_treatment/未成交原因;测试库仅 45 行,结构真实)
......
...@@ -41,6 +41,8 @@ clinic_directory: ...@@ -41,6 +41,8 @@ clinic_directory:
table: med_emr_info table: med_emr_info
id_field: organization_id id_field: organization_id
name_field: organization_name name_field: organization_name
# 按时间取最新(同 jvs-dw 的理由:诊所改名后旧名仍留在历史行里)。
time_field: updated_gmt_at
# ── 增量水位声明(2026-07 与 jvs-dw 统一;file 源放 manifest 顶层)── # ── 增量水位声明(2026-07 与 jvs-dw 统一;file 源放 manifest 顶层)──
...@@ -58,6 +60,7 @@ incremental: ...@@ -58,6 +60,7 @@ incremental:
customer_treat_plan_item: { cursor_column: updated_gmt_at } customer_treat_plan_item: { cursor_column: updated_gmt_at }
customer_referee_circle: { cursor_column: updated_gmt_at } customer_referee_circle: { cursor_column: updated_gmt_at }
customer_consult: { cursor_column: updated_gmt_at } customer_consult: { cursor_column: updated_gmt_at }
customer_return_visit: { cursor_column: updated_gmt_at }
# ⭐ 宿主推「自洽的业务表」:私有字典码 / 1:N 派生 / 头-行属性 都由宿主 within-host join 后 inline 进相关行 # ⭐ 宿主推「自洽的业务表」:私有字典码 / 1:N 派生 / 头-行属性 都由宿主 within-host join 后 inline 进相关行
# (契约见 docs/integration/friday-push-payload)。故不再摄入纯字典表(std_diag/std_check_class)、 # (契约见 docs/integration/friday-push-payload)。故不再摄入纯字典表(std_diag/std_check_class)、
...@@ -76,6 +79,9 @@ tables: ...@@ -76,6 +79,9 @@ tables:
- { table: patient_settlement_spec, file: patient_settlement_spec.csv } # 结算明细原表(全量,含 is_refund) - { table: patient_settlement_spec, file: patient_settlement_spec.csv } # 结算明细原表(全量,含 is_refund)
- { table: customer_referee_circle, file: customer_referee_circle.csv } # 转介绍圈(患者-患者边) - { table: customer_referee_circle, file: customer_referee_circle.csv } # 转介绍圈(患者-患者边)
- { table: customer_consult, file: customer_consult.csv } # 咨询(意向/未成交原因) - { table: customer_consult, file: customer_consult.csv } # 咨询(意向/未成交原因)
# 回访:宿主「统一任务表 customer_task」按 task_type 分流,2=回访 → 本表(1=咨询已由 customer_consult 接)。
# 任务头不单独摄入,task_date/task_status 等由宿主 inline 进回访行(同计划行 inline 计划头)。
- { table: customer_return_visit, file: customer_return_visit.csv } # 客户回访(含 inline task_date/task_status)
transforms: transforms:
# ── 患者 phone:宿主已按「默认号→本人→最早」挑好、inline 进 customer_basic_info 的 # ── 患者 phone:宿主已按「默认号→本人→最早」挑好、inline 进 customer_basic_info 的
...@@ -386,6 +392,28 @@ transforms: ...@@ -386,6 +392,28 @@ transforms:
op: concat op: concat
parts: ['${referee_relationship}', '|', '${referee_sex}'] parts: ['${referee_relationship}', '|', '${referee_sex}']
# ═══════════ 回访(customer_return_visit → patient_return_visits 独立表)═══════════
# J.1 治疗项两列合并成单字段(同 jvs-dw:大类「种植」+ 子项「单颗种植」)。
# 实测 2552 行:两列都有 499、只一级 257、只二级 0、都空 1796 —— 与 jvs-dw「总是成对」同构。
# 只有一级时合并出 "种植 · ",尾部分隔符由落库前的 normalizeMergedItems 归一掉;
# 两列全空合并出 " · ",同样被归一成 null(回访常无治疗项,占 70%)。
- kind: derive
input: customer_return_visit
output: customer_return_visit
fields:
treatment_items_full:
op: concat
parts: ['${treatment_items}', ' · ', '${treatment_items_two}']
# J.2 回访内容剥富文本标签 —— 宿主界面是富文本编辑器,实测 794/2552(31%)带 <p>/<ol>/<li>。
# PAC 侧该字段是纯文本语义(详情页直接展示 + 喂召回话术 LLM),带标签会原样显示成 "<p>xxx</p>"。
- kind: derive
input: customer_return_visit
output: customer_return_visit
fields:
follow_content_text:
op: strip_html
from: follow_content
assemblers: assemblers:
- { file: assemblers/patient.yaml } - { file: assemblers/patient.yaml }
- { file: assemblers/appointment.yaml } - { file: assemblers/appointment.yaml }
...@@ -402,3 +430,4 @@ assemblers: ...@@ -402,3 +430,4 @@ assemblers:
- { file: assemblers/refund_item.yaml } - { file: assemblers/refund_item.yaml }
- { file: assemblers/patient_relation.yaml } - { file: assemblers/patient_relation.yaml }
- { file: assemblers/consult.yaml } - { file: assemblers/consult.yaml }
- { file: assemblers/patient_return_visit.yaml } # 客户回访(展示 + 客服名册)→ PatientReturnVisit upsert
...@@ -64,6 +64,12 @@ clinic_directory: ...@@ -64,6 +64,12 @@ clinic_directory:
table: fact_emr_treatment_out table: fact_emr_treatment_out
id_field: organization_id id_field: organization_id
name_field: organization_name name_field: organization_name
# ⚠️ 必须按时间取最新 —— 诊所改过名,源表里同一个 id 存着历次名字,且**旧名往往行数更多**:
# 273393ed… 「正大诊所」91,053 行(2016-07~2022-08,早停用)
# 「花旗医院」3,096 行(2023-06~2023-09,过渡)
# 「瑞尔齿科上海花旗诊所」42,646 行(2023-09~至今,现名)
# 66 家里 53 家都有多个名字。取 any()/最高频都会拿到旧名。
time_field: updated_date
amount_unit: yuan amount_unit: yuan
timezone: Asia/Shanghai timezone: Asia/Shanghai
......
...@@ -25,6 +25,11 @@ import { createClient } from '@clickhouse/client'; ...@@ -25,6 +25,11 @@ import { createClient } from '@clickhouse/client';
import { AppModule } from '../app.module'; import { AppModule } from '../app.module';
import { PrismaService } from '../prisma/prisma.service'; import { PrismaService } from '../prisma/prisma.service';
import { ColdImportManifestSchema } from '../modules/sync/cold-import/manifest.schema'; import { ColdImportManifestSchema } from '../modules/sync/cold-import/manifest.schema';
import {
buildClinicNameQuery,
pickLatestNames,
type ClinicDirectoryConfig,
} from '../modules/sync/cold-import/clinic-directory';
import { disableSchedulersForCli } from './bootstrap-flags'; import { disableSchedulersForCli } from './bootstrap-flags';
// CLI 是短命进程,不需要 org-tree 启动预热(且会在 app.close() 时跟后台 warmAll 抢连接报噪音)。 // CLI 是短命进程,不需要 org-tree 启动预热(且会在 app.close() 时跟后台 warmAll 抢连接报噪音)。
...@@ -37,7 +42,7 @@ function flag(name: string): string | undefined { ...@@ -37,7 +42,7 @@ function flag(name: string): string | undefined {
async function deriveFromClickhouse( async function deriveFromClickhouse(
source: NonNullable<ReturnType<typeof ColdImportManifestSchema.parse>['sql_source']>, source: NonNullable<ReturnType<typeof ColdImportManifestSchema.parse>['sql_source']>,
dir: { table: string; id_field: string; name_field: string }, dir: ClinicDirectoryConfig,
logger: Logger, logger: Logger,
): Promise<Record<string, string>> { ): Promise<Record<string, string>> {
const url = process.env.DW_CLICKHOUSE_URL?.trim() || source.connection.url; const url = process.env.DW_CLICKHOUSE_URL?.trim() || source.connection.url;
...@@ -51,12 +56,15 @@ async function deriveFromClickhouse( ...@@ -51,12 +56,15 @@ async function deriveFromClickhouse(
const realTable = (() => { const realTable = (() => {
const sql = source.queries[dir.table]; const sql = source.queries[dir.table];
const m = sql?.match(/FROM\s+([\w.]+)/i); const m = sql?.match(/FROM\s+([\w.]+)/i);
return m ? m[1] : dir.table; return m?.[1] ?? dir.table;
})(); })();
const q = if (!dir.time_field) {
`SELECT ${dir.id_field} AS id, any(${dir.name_field}) AS name ` + logger.warn(
`FROM ${realTable} WHERE notEmpty(${dir.id_field}) AND notEmpty(${dir.name_field}) ` + `clinic_directory 未声明 time_field —— 无法判断哪个名字是现名,改名过的诊所会拿到旧名。` +
`GROUP BY ${dir.id_field}`; `请在 manifest 补 time_field(如 updated_date)。`,
);
}
const q = buildClinicNameQuery(realTable, dir);
logger.log(`CH 派生: ${q}`); logger.log(`CH 派生: ${q}`);
const rs = await client.query({ query: q, format: 'JSONEachRow' }); const rs = await client.query({ query: q, format: 'JSONEachRow' });
const rows = (await rs.json()) as Array<{ id: string; name: string }>; const rows = (await rs.json()) as Array<{ id: string; name: string }>;
...@@ -71,7 +79,8 @@ async function deriveFromClickhouse( ...@@ -71,7 +79,8 @@ async function deriveFromClickhouse(
function deriveFromFile( function deriveFromFile(
manifestDir: string, manifestDir: string,
tables: Array<{ table: string; file: string; format?: 'csv' | 'json' }>, tables: Array<{ table: string; file: string; format?: 'csv' | 'json' }>,
dir: { table: string; id_field: string; name_field: string }, dir: ClinicDirectoryConfig,
logger?: Logger,
): Record<string, string> { ): Record<string, string> {
const t = tables.find((x) => x.table === dir.table); const t = tables.find((x) => x.table === dir.table);
if (!t) throw new Error(`directory.table=${dir.table} 不在 tables[] 里`); if (!t) throw new Error(`directory.table=${dir.table} 不在 tables[] 里`);
...@@ -80,15 +89,12 @@ function deriveFromFile( ...@@ -80,15 +89,12 @@ function deriveFromFile(
const rows: Array<Record<string, unknown>> = t.file.toLowerCase().endsWith('.json') const rows: Array<Record<string, unknown>> = t.file.toLowerCase().endsWith('.json')
? JSON.parse(raw) ? JSON.parse(raw)
: parseCsv(raw, { columns: true, skip_empty_lines: true, trim: true }); : parseCsv(raw, { columns: true, skip_empty_lines: true, trim: true });
const map: Record<string, string> = {}; if (!dir.time_field) {
for (const r of rows) { logger?.warn(
const id = r[dir.id_field]; `clinic_directory 未声明 time_field —— 无法判断哪个名字是现名,改名过的诊所会拿到旧名。`,
const name = r[dir.name_field]; );
if (id != null && String(id) !== '' && name != null && String(name) !== '') {
map[String(id)] = String(name);
}
} }
return map; return pickLatestNames(rows, dir);
} }
async function main(): Promise<void> { async function main(): Promise<void> {
...@@ -111,7 +117,7 @@ async function main(): Promise<void> { ...@@ -111,7 +117,7 @@ async function main(): Promise<void> {
const clinicMap = manifest.sql_source const clinicMap = manifest.sql_source
? await deriveFromClickhouse(manifest.sql_source, cd, logger) ? await deriveFromClickhouse(manifest.sql_source, cd, logger)
: deriveFromFile(manifestDir, manifest.tables ?? [], cd); : deriveFromFile(manifestDir, manifest.tables ?? [], cd, logger);
const nc = Object.keys(clinicMap).length; const nc = Object.keys(clinicMap).length;
logger.log(`派生诊所名 ${nc} 家`); logger.log(`派生诊所名 ${nc} 家`);
if (nc === 0) { if (nc === 0) {
......
...@@ -251,6 +251,10 @@ export class TreatmentInitiationRecallScenario implements PlanScenarioPlugin { ...@@ -251,6 +251,10 @@ export class TreatmentInitiationRecallScenario implements PlanScenarioPlugin {
// ⚠️ **读数注意**:并发下**单条**查询的 sql= 会被争抢拉长(生产 caries 23.8→27.7 分, // ⚠️ **读数注意**:并发下**单条**查询的 sql= 会被争抢拉长(生产 caries 23.8→27.7 分,
// perio 12.6→16.9 分),看单条会误以为变慢了。**判据只能是阶段墙钟**, // perio 12.6→16.9 分),看单条会误以为变慢了。**判据只能是阶段墙钟**,
// 不能看各条耗时之和 —— 上面那次误判有一半就栽在这。 // 不能看各条耗时之和 —— 上面那次误判有一半就栽在这。
// ⚠️ 这个旋钮**同时是连接池的输入** —— 批次会长时间占住 conc 条连接。
// withCohortDerivedPool() 已把它算进池(N×6+5);2026-09-02 之前没算,生产池只有 5,
// 批次占 4 条 → 每日 09:00 健康日报的并发 count 抢不到连接,连续三天超时。
// ⛔ 调大它之前先确认该机 DATABASE_URL 没有显式 connection_limit(写了就绕过自动放大)。
const conc = Math.max(1, Number(process.env.PAC_RECALL_SUBSCENARIO_CONCURRENCY) || 1); const conc = Math.max(1, Number(process.env.PAC_RECALL_SUBSCENARIO_CONCURRENCY) || 1);
// ⚠️ 不能 hits.push(...subHits):spread 把每个元素当实参压栈,V8 实参上限 ~6.5万; // ⚠️ 不能 hits.push(...subHits):spread 把每个元素当实参压栈,V8 实参上限 ~6.5万;
// host 患者到 ~28 万后单子场景命中可超限 → RangeError: Maximum call stack size exceeded // host 患者到 ~28 万后单子场景命中可超限 → RangeError: Maximum call stack size exceeded
...@@ -623,7 +627,15 @@ export class TreatmentInitiationRecallScenario implements PlanScenarioPlugin { ...@@ -623,7 +627,15 @@ export class TreatmentInitiationRecallScenario implements PlanScenarioPlugin {
// ⛔ 必须用事务级 SET LOCAL,**不能全局调**:work_mem 是「每个排序/哈希节点」的上限, // ⛔ 必须用事务级 SET LOCAL,**不能全局调**:work_mem 是「每个排序/哈希节点」的上限,
// 不是每连接。生产 RDS 只有约 7GB 内存(shared_buffers 1.83GB 反推),全局设大值 // 不是每连接。生产 RDS 只有约 7GB 内存(shared_buffers 1.83GB 反推),全局设大值
// 遇上几十个并发连接同时排序会把内存吃穿。SET LOCAL 出了事务自动还原, // 遇上几十个并发连接同时排序会把内存吃穿。SET LOCAL 出了事务自动还原,
// Web API 的连接完全不受影响;且场景查询是串行的,同一时刻只有一条在跑,峰值可控。 // Web API 的连接完全不受影响。
//
// 🔴 **原文这里还有一句「且场景查询是串行的,同一时刻只有一条在跑,峰值可控」——
// 那句自 2026-08-30 起就是错的**,已删。生产从那天起设了
// PAC_RECALL_SUBSCENARIO_CONCURRENCY=4(见上方 conc 处注释),于是 RDS 上
// **同时有 4 个带 SET LOCAL work_mem='256MB' 的事务**。
// ⇒ DB 侧峰值内存 = 256MB × 并发数 × 该查询的排序/哈希节点数,不再"可控"。
// 生产 RDS 约 7GB,4 路已是能承受的上限;⛔ 谁要再调大 conc 或 work_mem,
// 必须先算这笔账,别只看场景段墙钟。(2026-09-05 事故复盘时发现这处注释失真。)
// //
// ⛔ 别再去加「信号码部分索引」——(content->>'code') 上的部分索引在测试机和生产 // ⛔ 别再去加「信号码部分索引」——(content->>'code') 上的部分索引在测试机和生产
// 都实测过,生产上 −5% 落在噪声里(同配置两次测量本身差 6%),不值得引入 // 都实测过,生产上 −5% 落在噪声里(同配置两次测量本身差 6%),不值得引入
......
...@@ -368,7 +368,11 @@ export class PlanAssignmentService { ...@@ -368,7 +368,11 @@ export class PlanAssignmentService {
const aExpires: Date[] = []; const aExpires: Date[] = [];
const aStrategy: string[] = []; const aStrategy: string[] = [];
const aDedicatedCs: (string | null)[] = []; const aDedicatedCs: (string | null)[] = [];
const aDedicatedCsLastVisit: (Date | null)[] = []; /**
* ⚠️ 存 **ISO 文本**而不是 Date —— 见下方 `::text[]::timestamptz[]` 的长注释。
* 一句话:整个数组全是 null 时,Prisma 推断不出元素类型,会退化成 integer[]。
*/
const aDedicatedCsLastVisit: (string | null)[] = [];
const aPriorityScore: number[] = []; const aPriorityScore: number[] = [];
const aSourceConfidence: (number | null)[] = []; const aSourceConfidence: (number | null)[] = [];
const aSelectionMode: string[] = []; const aSelectionMode: string[] = [];
...@@ -383,12 +387,29 @@ export class PlanAssignmentService { ...@@ -383,12 +387,29 @@ export class PlanAssignmentService {
); );
aStrategy.push(item.assignStrategy); aStrategy.push(item.assignStrategy);
aDedicatedCs.push(dcs); aDedicatedCs.push(dcs);
aDedicatedCsLastVisit.push(dcs ? (lastVisitByCs.get(dcs) ?? null) : null); aDedicatedCsLastVisit.push(dcs ? (lastVisitByCs.get(dcs)?.toISOString() ?? null) : null);
aPriorityScore.push(row.priorityScore); aPriorityScore.push(row.priorityScore);
aSourceConfidence.push(confidenceByPlan.get(row.id) ?? null); aSourceConfidence.push(confidenceByPlan.get(row.id) ?? null);
aSelectionMode.push(item.selectionMode ?? 'rank'); aSelectionMode.push(item.selectionMode ?? 'rank');
} }
/**
* 🔴 `aDedicatedCsLastVisit` 走 `::text[]::timestamptz[]`,**别改回 `::timestamptz[]`** ——
* 那样写在「本批所有患者都没有专属客服」时必 500(2026-09-21 FRIDAY 宿主实证)。
*
* 机理:Prisma 从**值**推断数组元素类型。整个数组全是 null 时推不出来,退化成
* `integer[]`,撞上显式 `::timestamptz[]` → PG 报
* `42846 cannot cast type integer[] to timestamp with time zone[]`。
* ⚠️ 空数组 `[]` 反而没事(压根走不到转换);**只有「非空且全 null」才炸**。
*
* 实测哪些目标类型会塌(2026-09-21 测试机):
* ::timestamptz[] ❌ ::uuid[] ❌
* ::text[] ✅ ::double precision[] ✅ ::int[] ✅
* ⇒ 同段里 `aDedicatedCs`(::text[])、`aSourceConfidence`(::double precision[])
* 虽然也可空,但 integer 到这些类型有转换路径,**不受影响**,别顺手一起改。
*
* ⛔ 别把这段注释写进下面的模板串里 —— 那里面的 `${…}` 会被当插值求值(踩过)。
*/
const appliedRows = await tx.$queryRaw<Array<{ id: string }>>(Prisma.sql` const appliedRows = await tx.$queryRaw<Array<{ id: string }>>(Prisma.sql`
UPDATE followup_plans AS fp SET UPDATE followup_plans AS fp SET
status = 'assigned', status = 'assigned',
...@@ -413,7 +434,7 @@ export class PlanAssignmentService { ...@@ -413,7 +434,7 @@ export class PlanAssignmentService {
${aExpires}::timestamptz[], ${aExpires}::timestamptz[],
${aStrategy}::text[], ${aStrategy}::text[],
${aDedicatedCs}::text[], ${aDedicatedCs}::text[],
${aDedicatedCsLastVisit}::timestamptz[], ${aDedicatedCsLastVisit}::text[]::timestamptz[],
${aPriorityScore}::double precision[], ${aPriorityScore}::double precision[],
${aSourceConfidence}::double precision[], ${aSourceConfidence}::double precision[],
${aSelectionMode}::text[] ${aSelectionMode}::text[]
......
/**
* 诊所名字典派生 —— 从摄入源的 (id, name) 里挑出**现名**。
*
* 🔴 这里唯一要解决的问题:**诊所会改名,源表里同一个 id 存着历次名字。**
* 2026-09-03 生产实测(jvs-dw fact_emr_treatment_out,66 家诊所):
* · **53 家有多个名字**(80%),不是个案
* · 273393ed… 「正大诊所」91,053 行(2016-07~2022-08,早停用)
* 「花旗医院」3,096 行(2023-06~2023-09,过渡)
* 「瑞尔齿科上海花旗诊所」42,646 行(2023-09~至今,**现名**)
* · c59bfc52… 6 个名字,其中两个带前导 `\t`
*
* ⛔ `any(name)`(原实现)任取一个 —— 不确定,且大概率给旧名。
* ⛔ "取最高频"同样错 —— 旧名用了 6 年,行数往往碾压现名。
* ✅ 只能按时间取最新。
*
* 缺 `time_field` 时退化成"字典序最大":唯一的好处是**确定性**(同样输入同样输出,
* 不会跑一次一个名字),但**不保证是现名** —— 调用方必须告警。
*/
export interface ClinicDirectoryConfig {
table: string;
id_field: string;
name_field: string;
/// 判定"哪个名字是现名"的时间列(越大越新);字符串日期也可(字典序即时序)
time_field?: string;
}
/**
* ClickHouse 派生 SQL。
* - `trimBoth` 清掉源里带前导 `\t` 的脏值(华贸诊所实测有)
* - `argMax(name, time)` 取时间最大那行的名字
* - `ifNull(toString(time), '')` 兼容 Nullable 时间列(jvs-dw 的 updated_date 是 Nullable(String))
*/
export function buildClinicNameQuery(realTable: string, dir: ClinicDirectoryConfig): string {
const nameExpr = `trimBoth(${dir.name_field})`;
const pick = dir.time_field
? `argMax(${nameExpr}, ifNull(toString(${dir.time_field}), ''))`
: `argMax(${nameExpr}, ${nameExpr})`;
return (
`SELECT ${dir.id_field} AS id, ${pick} AS name ` +
`FROM ${realTable} WHERE notEmpty(${dir.id_field}) AND notEmpty(${nameExpr}) ` +
`GROUP BY ${dir.id_field}`
);
}
/**
* 文件源派生 —— 与 ClickHouse 分支**同口径**。
* ⛔ 不能写成"遍历行、后者覆盖前者":行序是文件顺序,等于随机取名。
*/
export function pickLatestNames(
rows: Array<Record<string, unknown>>,
dir: ClinicDirectoryConfig,
): Record<string, string> {
const best = new Map<string, { name: string; at: string }>();
for (const r of rows) {
const id = r[dir.id_field];
if (id == null || String(id) === '') continue;
const nm = r[dir.name_field] == null ? '' : String(r[dir.name_field]).trim();
if (!nm) continue;
// 无 time_field → 拿名字自身当序(字典序最大);有则按时间列
const at = dir.time_field ? String(r[dir.time_field] ?? '') : nm;
const cur = best.get(String(id));
if (!cur || at >= cur.at) best.set(String(id), { name: nm, at });
}
const out: Record<string, string> = {};
for (const [id, v] of best) out[id] = v.name;
return out;
}
...@@ -46,6 +46,36 @@ import type { FactReject } from '../pipeline/fact-writer.service'; ...@@ -46,6 +46,36 @@ import type { FactReject } from '../pipeline/fact-writer.service';
* 给几条足够定位;全量在 PAC 服务端日志里(每条都有 [schema-violation] 行)。 * 给几条足够定位;全量在 PAC 服务端日志里(每条都有 [schema-violation] 行)。
*/ */
const FACT_REJECT_SAMPLE_CAP = 10; const FACT_REJECT_SAMPLE_CAP = 10;
/**
* PG 单条 prepared statement 的 bind 变量上限。
* reparse 的每条查询都带 `patientId IN (...)` —— **每个 id 占一个 bind**,
* 所以"一批多少患者"同时是内存旋钮和一堵硬墙。
*/
const PG_MAX_BIND_VARS = 32_767;
/** 患者分批粒度的硬上限:留出 where 里 hostId / subjectType 等其余 bind 的余量。 */
const REPARSE_MAX_BATCH = PG_MAX_BIND_VARS - 100;
/** 默认批大小(内存口径:每批只把这批患者的 rawPayload 拉进内存)。 */
const REPARSE_DEFAULT_BATCH = 3000;
/**
* reparse 的患者分批粒度 —— **dryRun 与实跑共用同一个值**。
*
* ⚠️ 2026-08-28 测试服实测:`--patients-file`(3.2 万+ 清单)+ `--dry-run` 直接炸
* `too many bind variables in prepared statement, expected maximum of 32767, received 62899`
* —— 当时只有 dryRun 分支把**全部** patientIds 一次塞进 `IN`,实跑分支早就分批了。
* 讽刺的是 `--patients-file` 的文档正写着「按受影响患者收窄是 reparse 最有效的提速手段
* (实测可达 250 倍)」:清单越大越该先 dry-run 探一探,而那恰恰是它唯一不工作的场景。
*
* 🔴 2026-09-03 补上另一半:dryRun 那侧当时修好了(分块),**实跑那侧的旋钮却没有上限** ——
* `PAC_REPARSE_BATCH` 之前是 `Math.max(1, …)`,只夹下限。设成 50000 就会让同一堵墙
* 从实跑那侧长回来。这里统一夹住,两侧共用。
* ⛔ 别在任何分支把整份 patientIds 一次性塞进 `IN`;也别绕过本函数直接读 env。
*/
export function reparseBatchSize(env: NodeJS.ProcessEnv = process.env): number {
const raw = Number(env.PAC_REPARSE_BATCH) || REPARSE_DEFAULT_BATCH;
return Math.min(REPARSE_MAX_BATCH, Math.max(1, raw));
}
import type { TransformOp } from '../transforms/transforms.schema'; import type { TransformOp } from '../transforms/transforms.schema';
import { import {
buildPushLookupFallbackRows, buildPushLookupFallbackRows,
...@@ -185,10 +215,10 @@ export class ColdImportService { ...@@ -185,10 +215,10 @@ export class ColdImportService {
// 2a. dryRun:报 scope(患者数 + 各资源 txn 数),不写、不分批装配(便宜)。 // 2a. dryRun:报 scope(患者数 + 各资源 txn 数),不写、不分批装配(便宜)。
if (opts.dryRun) { if (opts.dryRun) {
// ⛔ 同样分块 —— 早先这里也是整份清单塞 in,>3.2 万患者时 dry-run 直接崩, // ⛔ 同样分块 —— 早先这里也是整份清单塞 in,>3.2 万患者时 dry-run 直接崩。
// 而 --patients-file 的文档恰恰说「按受影响患者收窄是最有效的提速手段(可达 250 倍)」: // 与实跑共用 reparseBatchSize()(见其注释:两侧用同一个夹过上限的值,
// 最需要先 dry-run 探一探的大清单场景,正好是它唯一不工作的场景。 // 否则改一侧另一侧的墙还在)。
const DRY_CHUNK = 3000; const DRY_CHUNK = reparseBatchSize();
for (const cfg of reparseableCfgs) { for (const cfg of reparseableCfgs) {
let n = 0; let n = 0;
const chunks: Array<string[] | null> = opts.patientIds?.length const chunks: Array<string[] | null> = opts.patientIds?.length
...@@ -209,7 +239,7 @@ export class ColdImportService { ...@@ -209,7 +239,7 @@ export class ColdImportService {
// 2b. 分批实跑:每批患者 → 重建源表(只这批 distinct rawPayload)→ transform → processSubject // 2b. 分批实跑:每批患者 → 重建源表(只这批 distinct rawPayload)→ transform → processSubject
// (transaction 幂等命中已存 → parser 重衍生 fact,版本流 supersede)。 // (transaction 幂等命中已存 → parser 重衍生 fact,版本流 supersede)。
const BATCH = Math.max(1, Number(process.env.PAC_REPARSE_BATCH) || 3000); const BATCH = reparseBatchSize();
const runStart = new Date(); const runStart = new Date();
const aggByResource = new Map<string, PerResourceStats>(); const aggByResource = new Map<string, PerResourceStats>();
const seenTenants = new Set<string>(); const seenTenants = new Set<string>();
......
...@@ -231,11 +231,19 @@ export const ColdImportManifestSchema = z ...@@ -231,11 +231,19 @@ export const ColdImportManifestSchema = z
/// `refresh-clinic-names` CLI 据此 SELECT DISTINCT 出 id→名,写 host.clinicNames; /// `refresh-clinic-names` CLI 据此 SELECT DISTINCT 出 id→名,写 host.clinicNames;
/// /auth/session 再合并进 dictionary.clinics 下发,前端显示名字不依赖登录传。 /// /auth/session 再合并进 dictionary.clinics 下发,前端显示名字不依赖登录传。
/// sql_source 模式:table 是 CH 表名(或 queries 的 key);文件模式:table 对应 tables[].table。 /// sql_source 模式:table 是 CH 表名(或 queries 的 key);文件模式:table 对应 tables[].table。
/// 诊所名字典来源(id → 中文名),供 refresh-clinic-names 派生 host.clinicNames。
/// ⚠️ `time_field` 不是可有可无的:诊所会改名,同一个 id 在源表里存着**历次**名字。
/// 实测 jvs-dw 66 家里 **53 家有多个名字**(如 273393ed… 同时有「正大诊所」91,053 行
/// 〔2016~2022 旧名〕/「花旗医院」3,096 行 /「瑞尔齿科上海花旗诊所」42,646 行〔当前〕)。
/// 不按时间取最新,就会把早已停用的旧名当成现名 —— 而且旧名往往行数更多,
/// "取最高频"同样错。缺 time_field 时只能退化成"最高频"并告警,结果不可信。
clinic_directory: z clinic_directory: z
.object({ .object({
table: z.string().min(1), table: z.string().min(1),
id_field: z.string().min(1), id_field: z.string().min(1),
name_field: z.string().min(1), name_field: z.string().min(1),
/// 判定"哪个名字是现名"的时间列(越大越新)。字符串日期也可(字典序即时序)。
time_field: z.string().min(1).optional(),
}) })
.optional(), .optional(),
......
...@@ -97,6 +97,11 @@ function evalExpr(expr: DeriveExpr, row: Row): unknown { ...@@ -97,6 +97,11 @@ function evalExpr(expr: DeriveExpr, row: Row): unknown {
if (typeof v !== 'string') return v ?? null; if (typeof v !== 'string') return v ?? null;
return v.trim().replace(TRAILING_NOISE, '').trim(); return v.trim().replace(TRAILING_NOISE, '').trim();
} }
case 'strip_html': {
const v = row[expr.from];
if (typeof v !== 'string') return v ?? null;
return stripHtml(v);
}
case 'normalize': { case 'normalize': {
const v = row[expr.from]; const v = row[expr.from];
if (typeof v !== 'string') return v ?? null; if (typeof v !== 'string') return v ?? null;
...@@ -116,3 +121,49 @@ function evalExpr(expr: DeriveExpr, row: Row): unknown { ...@@ -116,3 +121,49 @@ function evalExpr(expr: DeriveExpr, row: Row): unknown {
} }
} }
} }
/**
* 剥 HTML 富文本标签 → 纯文本。
*
* 宿主用富文本编辑器录入的字段(FRIDAY 回访内容实测 31% 带 <p>/<ol>/<li>/<em>)进 PAC 后
* 是**纯文本语义** —— 详情页直接展示、喂 LLM 生成话术。带标签会被原样显示成 "<p>xxx</p>",
* 也污染 LLM 上下文,所以在摄入层剥掉,而不是留给每个消费方各自处理。
*
* 口径:
* - 块级标签(</p> </li> </div> <br>)→ 换行,否则多段文字会黏成一行
* - 其余标签直接去掉
* - 常见 HTML 实体还原(&nbsp; &amp; &lt; &gt; &quot; &#39;)
* - 收尾清理连续空行/首尾空白;全部剥完只剩空白 → 返回 null(空壳不入库)
*/
export function stripHtml(input: string): string | null {
// 单轮:剥标签 → 还原实体。实体还原可能**再露出字面标签** —— 宿主实测存在双重编码:
// '<p><span…>&lt;p&gt;阿斯蒂芬撒的发生&lt;/p&gt;</span></p>'
// 外层是真标签、内层是被转义的标签,剥一轮后 &lt;p&gt; 还原成 <p> 留在结果里。
// 故做**有界多轮**:还原后若又出现标签则再剥一轮,最多 2 轮 —— 不做无界循环,
// 避免构造出的畸形输入(每轮都产生新标签)把摄入卡死。
const once = (v: string): string =>
v
// 块级结束/换行标签先转成换行,保住段落边界
.replace(/<\s*br\s*\/?\s*>/gi, '\n')
.replace(/<\/\s*(p|div|li|tr|h[1-6]|blockquote)\s*>/gi, '\n')
// 其余标签一律剥掉
.replace(/<[^>]*>/g, '')
// 实体还原(&amp; 放最后,避免 &amp;lt; 被二次解码成 <)
.replace(/&nbsp;/gi, ' ')
.replace(/&lt;/gi, '<')
.replace(/&gt;/gi, '>')
.replace(/&quot;/gi, '"')
.replace(/&#39;/g, "'")
.replace(/&amp;/gi, '&');
let text = once(input);
// 第 2 轮只在确实又露出标签时才跑(绝大多数行一轮即净,不做无谓开销)
if (/<[^>]+>/.test(text)) text = once(text);
text = text
// 连续空行压成一个,首尾清干净
.replace(/[ \t]+\n/g, '\n')
.replace(/\n{2,}/g, '\n')
.trim();
return text === '' ? null : text;
}
...@@ -125,6 +125,7 @@ export type SplitJsonArrayOp = z.infer<typeof SplitJsonArrayOpSchema>; ...@@ -125,6 +125,7 @@ export type SplitJsonArrayOp = z.infer<typeof SplitJsonArrayOpSchema>;
* - normalize: { op: normalize, from: field } — trim 升级版 + CJK→ASCII 中段标点 * - normalize: { op: normalize, from: field } — trim 升级版 + CJK→ASCII 中段标点
* (中括号()/中逗号,/中分号;/中尖括号 < >)统一 ASCII,中文宿主 enum 匹配必备 * (中括号()/中逗号,/中分号;/中尖括号 < >)统一 ASCII,中文宿主 enum 匹配必备
* - coalesce: { op: coalesce, from: [f1, f2, ...] } — 取第一个非空字段值(诊断码兜底:stdCode→message) * - coalesce: { op: coalesce, from: [f1, f2, ...] } — 取第一个非空字段值(诊断码兜底:stdCode→message)
* - strip_html: { op: strip_html, from: field } — 剥富文本标签取纯文本(宿主用富文本编辑器时)
* *
* concat parts 里 `${fieldName}` 是字段引用,其他视为字面量。 * concat parts 里 `${fieldName}` 是字段引用,其他视为字面量。
* 限定不允许任意表达式(canonical-fact-layer.md 原则 1)。 * 限定不允许任意表达式(canonical-fact-layer.md 原则 1)。
...@@ -162,6 +163,15 @@ export const DeriveTrimSchema = z.object({ ...@@ -162,6 +163,15 @@ export const DeriveTrimSchema = z.object({
from: z.string().min(1), from: z.string().min(1),
}); });
/// 剥 HTML 富文本标签 → 纯文本。宿主侧用富文本编辑器录入时,原值会带 <p>/<ol>/<li>/<em> 等
/// (FRIDAY 回访内容实测 794/2552 = 31% 含标签)。PAC 的这些字段是**纯文本语义**
/// (详情页直接展示、喂 LLM 生成话术),带标签进来会被原样显示成 "<p>xxx</p>",
/// 也会污染 LLM 上下文。故在摄入层剥掉,而不是留给每个消费方各自处理。
export const DeriveStripHtmlSchema = z.object({
op: z.literal('strip_html'),
from: z.string().min(1),
});
/// W4 末加:CJK→ASCII 标点 + trim 一站式归一(给中文宿主 enum_mapping 字符串匹配用) /// W4 末加:CJK→ASCII 标点 + trim 一站式归一(给中文宿主 enum_mapping 字符串匹配用)
/// trim 是子集(只去首尾噪音);normalize 还做"中文括号/逗号/分号 → ASCII"中段替换 /// trim 是子集(只去首尾噪音);normalize 还做"中文括号/逗号/分号 → ASCII"中段替换
/// 适用任何中文宿主的脏字段归一(不是 jvs-dw 特化),所以是通用 op /// 适用任何中文宿主的脏字段归一(不是 jvs-dw 特化),所以是通用 op
...@@ -191,6 +201,7 @@ export const DeriveExprSchema = z.discriminatedUnion('op', [ ...@@ -191,6 +201,7 @@ export const DeriveExprSchema = z.discriminatedUnion('op', [
DeriveUpperSchema, DeriveUpperSchema,
DeriveDefaultSchema, DeriveDefaultSchema,
DeriveTrimSchema, DeriveTrimSchema,
DeriveStripHtmlSchema,
DeriveNormalizeSchema, DeriveNormalizeSchema,
DeriveCoalesceSchema, DeriveCoalesceSchema,
DeriveRecodeSchema, DeriveRecodeSchema,
......
...@@ -14,16 +14,28 @@ import { tenantGuardExtension } from './tenant-guard.extension'; ...@@ -14,16 +14,28 @@ import { tenantGuardExtension } from './tenant-guard.extension';
* *
* ⭐ 2026-07-26 修:原来只认 `PAC_COHORT_CONCURRENCY`(**摄入**的旋钮),于是 * ⭐ 2026-07-26 修:原来只认 `PAC_COHORT_CONCURRENCY`(**摄入**的旋钮),于是
* `recompute-persona --concurrency=N` 这类 CLI 的旋钮**不会放大池**。 * `recompute-persona --concurrency=N` 这类 CLI 的旋钮**不会放大池**。
* 改成读**通用**的 PAC_DB_CONCURRENCY,由各 CLI 在 Nest 启动前按自己的 --concurrency 设进来;
* PAC_COHORT_CONCURRENCY 保留为向后兼容的别名,取最大者。
* *
* ⚠️ 这是**潜在**缺陷,咬不咬人取决于该机 .env 有没有显式写 connection_limit(核对过两台): * 🔴 2026-09-02 再修:**常驻服务自己也有并发旋钮,一样没被算进来** —— 生产事故。
* 测试服 47.251.104.47 URL 里显式 `connection_limit=30` → 本函数提前返回,池=30,不是瓶颈 * `PAC_RECALL_SUBSCENARIO_CONCURRENCY=4`(召回子场景并行,2026-08-30 在生产开启)让批次
* (那里观测到的 9 条连接 = 8 个 worker + 1 空闲,不是池上限 —— 别再据此推断) * 长时间占住 4 条连接,而常驻进程走的是 Prisma 默认池。结果:每日 09:00 的健康日报
* 生产 47.99.62.30 URL **没写** → 走 Prisma 默认 `核数×2+1`,4 核机 = **9** * (6 个并发 count)抢不到连接,**连续三天在 09:00:10 整点超时**,企微收不到日报;
* → `--concurrency` 超过 8 就会被卡在 9,多出来的 worker 排队等连接 * 同一根因还打掉过一条 plan upsert(`Unable to start a transaction in the given time`)。
* 也就是说:按 --concurrency=8 跑生产,9 条刚好够,本修复不产生收益;它的价值是**解锁更高并发**。 * 本函数开头那句"一调并发就得手动调池"说的就是这个病,只是当时没把常驻服务的旋钮算进来。
* *
* 改成读**通用**的 PAC_DB_CONCURRENCY,由各 CLI 在 Nest 启动前按自己的 --concurrency 设进来; * ⚠️ **Prisma 默认池是「物理核」×2+1,不是逻辑核** —— 这里以前写错过,导致低估了生产的池:
* PAC_COHORT_CONCURRENCY 保留为向后兼容的别名,两者取大。 * 生产 47.99.62.30:`nproc`=4 但 `Core(s) per socket`=2 / `Thread(s) per core`=2,
* 物理核只有 2 → 默认池 = 2×2+1 = **5**(不是按 4 核算的 9)。5 条连接扛不住 4 路并发 + 日报。
* ⛔ **别再说「测试服 .env 里写了 connection_limit=30 所以不受影响」** —— 这话流传了很久,是错的:
* compose 会在 `environment:` 段覆盖 .env 里的 DATABASE_URL,容器里实际拿到的是
* `postgresql://…@postgres:5432/pac?schema=public`,**没有 connection_limit**
* (2026-09-02 `docker inspect` 实测两台都如此)。测试服同样是默认池,
* 09-01 那天它的健康日报也报了同一句 `connection limit: 5`。
* 判断逃生口生不生效,要看**容器实际环境变量**,不是看 .env 文件。
*
* 生产 RDS `max_connections=820`(2026-09-02 实测,当时全库仅 23 条在用),
* 封顶 40 对它绰绰有余;封顶注释里那个 100 是早年自建 PG 的值,保守留着。
*/ */
export function withCohortDerivedPool(rawUrl: string | undefined): string | undefined { export function withCohortDerivedPool(rawUrl: string | undefined): string | undefined {
if (!rawUrl) return rawUrl; if (!rawUrl) return rawUrl;
...@@ -35,7 +47,15 @@ export function withCohortDerivedPool(rawUrl: string | undefined): string | unde ...@@ -35,7 +47,15 @@ export function withCohortDerivedPool(rawUrl: string | undefined): string | unde
} }
if (url.searchParams.has('connection_limit')) return rawUrl; if (url.searchParams.has('connection_limit')) return rawUrl;
const num = (v: string | undefined) => Math.max(1, parseInt(v ?? '1', 10) || 1); const num = (v: string | undefined) => Math.max(1, parseInt(v ?? '1', 10) || 1);
const n = Math.max(num(process.env.PAC_DB_CONCURRENCY), num(process.env.PAC_COHORT_CONCURRENCY)); // 三个旋钮取最大:
// PAC_DB_CONCURRENCY 通用(各 CLI 按自己的 --concurrency 在 Nest 启动前设)
// PAC_COHORT_CONCURRENCY 摄入的历史别名,向后兼容
// PAC_RECALL_SUBSCENARIO_CONCURRENCY **常驻服务**的召回并行度(批次会长时间占住这么多条)
const n = Math.max(
num(process.env.PAC_DB_CONCURRENCY),
num(process.env.PAC_COHORT_CONCURRENCY),
num(process.env.PAC_RECALL_SUBSCENARIO_CONCURRENCY),
);
if (n <= 1) return rawUrl; if (n <= 1) return rawUrl;
url.searchParams.set('connection_limit', String(Math.min(n * 6 + 5, 40))); url.searchParams.set('connection_limit', String(Math.min(n * 6 + 5, 40)));
return url.toString(); return url.toString();
......
...@@ -253,7 +253,7 @@ export class DailyHealthReportService { ...@@ -253,7 +253,7 @@ export class DailyHealthReportService {
} }
if (pac.kind === 'return_visit') { if (pac.kind === 'return_visit') {
const r = await this.prisma.$queryRaw<Array<{ c: bigint }>>` const r = await this.prisma.$queryRaw<Array<{ c: bigint }>>`
SELECT count(distinct patient_id) AS c FROM patient_return_visit WHERE host_id = ${hostId}::uuid`; SELECT count(distinct patient_id) AS c FROM patient_return_visits WHERE host_id = ${hostId}::uuid`;
return Number(r[0]?.c ?? 0); return Number(r[0]?.c ?? 0);
} }
if (pac.kind === 'image_ai') { if (pac.kind === 'image_ai') {
......
...@@ -179,6 +179,11 @@ describe('canonical-fact-layer | 闸 4:yaml enum_mapping 目标 ∈ canonical-co ...@@ -179,6 +179,11 @@ describe('canonical-fact-layer | 闸 4:yaml enum_mapping 目标 ∈ canonical-co
for (const { host, file, config } of yamls) { for (const { host, file, config } of yamls) {
const mappings = config.enum_mapping ?? {}; const mappings = config.enum_mapping ?? {};
for (const [canonicalField, valueMap] of Object.entries(mappings)) { for (const [canonicalField, valueMap] of Object.entries(mappings)) {
// patient_return_visit 特例:它的 status/type 是**展示用自由文本**,不是受控码。
// closedSets.status 装的是 PACTreatmentStatuses(治疗状态),与回访的"已回访/未回访"同名不同义。
// PAC 侧这几列直接落 PatientReturnVisit 原样展示(不进 fact、不参与召回),
// 取值口径是"与 jvs-dw 的 DW *_name 中文列逐字对齐",而非 canonical-codes 闭集。
if (config.canonical === 'patient_return_visit') continue;
// appointment.status 特例:用 host-side enum // appointment.status 特例:用 host-side enum
if (canonicalField === 'status' && config.canonical === 'appointment') { if (canonicalField === 'status' && config.canonical === 'appointment') {
test(`${host}/${file} appointment.status enum_mapping 目标在 [scheduled/arrived/cancelled/no_show]`, () => { test(`${host}/${file} appointment.status enum_mapping 目标在 [scheduled/arrived/cancelled/no_show]`, () => {
......
/**
* 诊所名字典:必须取**现名**,不是任取 / 不是最高频。
*
* 🔴 起因(2026-09-03 生产实测):原实现 `any(organization_name)` 把
* `273393edd04e4aa4afe9347bb8a2da21` 解析成「正大诊所」—— 那是 2022-08 就停用的旧名,
* 现名是「瑞尔齿科上海花旗诊所」。DW 里 66 家诊所有 **53 家**存着多个历史名字,
* 而且旧名用了六七年、行数往往碾压现名,所以"取最高频"跟 any() 一样错。
*
* 下面的样例数据直接取自生产 DW 的真实形态(含带前导 \t 的脏值)。
*/
import {
buildClinicNameQuery,
pickLatestNames,
type ClinicDirectoryConfig,
} from '../src/modules/sync/cold-import/clinic-directory';
const DIR: ClinicDirectoryConfig = {
table: 'fact_emr_treatment_out',
id_field: 'organization_id',
name_field: 'organization_name',
time_field: 'updated_date',
};
// 上海花旗:旧名行数(91,053)远超现名(42,646)—— 最高频口径会选错
const SHANGHAI_CITI = [
...rows('273393ed', '正大诊所', '2022-08-17 14:16:26', 3),
...rows('273393ed', '花旗医院', '2023-09-07 15:01:58', 2),
...rows('273393ed', '瑞尔齿科上海花旗诊所', '2026-09-03 08:55:18', 1),
];
function rows(id: string, name: string, at: string, n: number) {
return Array.from({ length: n }, () => ({
organization_id: id,
organization_name: name,
updated_date: at,
}));
}
describe('pickLatestNames —— 取现名而不是旧名', () => {
test('⭐ 旧名行数更多也要选最新的那个(上海花旗真实案例)', () => {
expect(pickLatestNames(SHANGHAI_CITI, DIR)).toEqual({
'273393ed': '瑞尔齿科上海花旗诊所',
});
});
test('行序打乱不影响结果(文件源的行序等于随机)', () => {
const shuffled = [...SHANGHAI_CITI].reverse();
expect(pickLatestNames(shuffled, DIR)['273393ed']).toBe('瑞尔齿科上海花旗诊所');
});
test('清掉源里的前导/尾随空白(华贸诊所实测有带 \\t 的行)', () => {
const r = pickLatestNames(
[
{ organization_id: 'c59bfc52', organization_name: '华贸诊所', updated_date: '2020-01-01' },
{ organization_id: 'c59bfc52', organization_name: '\t瑞尔齿科北京华贸诊所A诊区', updated_date: '2026-01-01' },
],
DIR,
);
expect(r['c59bfc52']).toBe('瑞尔齿科北京华贸诊所A诊区');
});
test('空名 / 空 id / 全空白名 一律跳过,不写进字典', () => {
const r = pickLatestNames(
[
{ organization_id: 'a', organization_name: '有效诊所', updated_date: '2020-01-01' },
{ organization_id: 'a', organization_name: ' ', updated_date: '2026-01-01' },
{ organization_id: '', organization_name: '无 id', updated_date: '2026-01-01' },
{ organization_id: 'b', organization_name: null, updated_date: '2026-01-01' },
],
DIR,
);
expect(r).toEqual({ a: '有效诊所' }); // 空白名不能顶掉有效名,b 不该出现
});
test('时间列缺值的行不会盖掉有时间的行', () => {
const r = pickLatestNames(
[
{ organization_id: 'a', organization_name: '现名', updated_date: '2026-01-01' },
{ organization_id: 'a', organization_name: '无时间旧名', updated_date: null },
],
DIR,
);
expect(r['a']).toBe('现名');
});
test('未配 time_field → 退化成字典序最大(只保证确定性,不保证是现名)', () => {
const noTime: ClinicDirectoryConfig = { ...DIR, time_field: undefined };
const a = pickLatestNames(SHANGHAI_CITI, noTime);
const b = pickLatestNames([...SHANGHAI_CITI].reverse(), noTime);
expect(a).toEqual(b); // 同样输入同样输出,不会跑一次一个名字
});
});
describe('buildClinicNameQuery —— ClickHouse 侧同口径', () => {
test('⭐ 用 argMax 按时间取最新,⛔ 不再是 any()', () => {
const q = buildClinicNameQuery('dw_group.fact_emr_treatment_out', DIR);
expect(q).toContain("argMax(trimBoth(organization_name), ifNull(toString(updated_date), ''))");
expect(q).not.toMatch(/\bany\(/);
});
test('trimBoth 同时用在取值和非空判定上(否则纯 \\t 的行会被当成有效名)', () => {
const q = buildClinicNameQuery('t', DIR);
expect(q).toContain('notEmpty(trimBoth(organization_name))');
});
test('未配 time_field 也不退回 any()(保持确定性)', () => {
const q = buildClinicNameQuery('t', { ...DIR, time_field: undefined });
expect(q).not.toMatch(/\bany\(/);
expect(q).toContain('argMax(trimBoth(organization_name), trimBoth(organization_name))');
});
});
describe('manifest 必须声明 time_field(缺了就会解析成旧名)', () => {
const yaml = require('js-yaml');
const { readFileSync } = require('node:fs');
const { join } = require('node:path');
for (const host of ['jvs-dw', 'friday']) {
test(`${host}: clinic_directory.time_field 已声明`, () => {
const m = yaml.load(
readFileSync(join(__dirname, `../data/${host}/manifest.yaml`), 'utf-8'),
) as { clinic_directory?: ClinicDirectoryConfig };
expect(m.clinic_directory?.time_field).toBeTruthy();
});
}
});
canonical: diagnosis
emits:
action: diagnosis_recorded
subjectType: diagnosis
occurredAtField: diagnosedAt
primary:
table: diagnosis_rows
key: id
field_mapping:
diagnosisId: id
patientId: patient_id
code: diag_code
diagnosedAt: diagnosed_at
# 最小 manifest —— 只为让 reparse 走到 dryRun 的 count 分支。
# 关键:transforms 让 diagnosis_rows 能回溯到源表 raw_emr,否则 reparse 会把该资源判为
# 「非 transform 产出」直接跳过,count 一次都不发,测试就测了个空。
# 数据文件不存在也没关系:reparse 只读 rawPayload,不读 tables[].file。
host_name: fixture-host
tenant_id: fixture-tenant
amount_unit: yuan
timezone: Asia/Shanghai
tables:
- table: raw_emr
file: raw_emr.csv
transforms:
- kind: derive
input: raw_emr
output: diagnosis_rows
fields:
diag_code:
op: trim
from: code
assemblers:
- file: assemblers/diagnosis.yaml
import { readFileSync } from 'node:fs';
import { join } from 'node:path';
import * as yaml from 'js-yaml';
import { TransformEngine } from '../src/modules/sync/transforms/transform-engine';
import type { TransformsConfig } from '../src/modules/sync/transforms/transforms.schema';
import { stripHtml } from '../src/modules/sync/transforms/operators/derive.op';
/**
* FRIDAY 回访摄入(customer_return_visit → PatientReturnVisit)。
*
* 【源表结构:统一任务表 + 分类型记录表】
* customer_task(task_type: 1咨询 / 2回访 / 4预约备注 / 10…)
* ├─ 1 → customer_consult (咨询,PAC 早已单独摄入)
* └─ 2 → customer_return_visit (回访,本次)
* 实测零交叉:consult 的 64 条 task_id 全指向 type=1,回访 2552 条全指向 type=2。
* 关联 customer_return_visit.task_id = customer_task.id,2552/2552 全中、零空值。
*
* 【为什么枚举要翻成中文】jvs-dw 的 DW 侧本就是 *_name 中文列(常规回访/已回访/已完成),
* FRIDAY 是数字码。同一张 PatientReturnVisit 表里两个宿主的值必须逐字对齐,
* 否则前端展示和按类型筛选会分叉。
*
* 【task_date 含未来】界面「设回访」就是设未来日期。PAC 靠它区分「已发生」与「排了没做」:
* 详情页倒序展示、召回话术只把已发生的算作"联系过";客服名册判在岗则用 sourceCreatedAt,
* 正因为 task_date 含未来排程(jvs-dw 实测最远 2033)。
*/
describe('FRIDAY 回访摄入', () => {
const dir = join(__dirname, '../data/friday');
const manifest = yaml.load(readFileSync(join(dir, 'manifest.yaml'), 'utf8')) as {
tables: Array<{ table: string; file: string }>;
incremental: { per_query: Record<string, unknown> };
transforms: unknown[];
assemblers: Array<{ file: string }>;
};
const asm = yaml.load(
readFileSync(join(dir, 'assemblers/patient_return_visit.yaml'), 'utf8'),
) as {
canonical: string;
primary: { table: string };
field_mapping: Record<string, string>;
enum_mapping: Record<string, Record<string, string>>;
};
describe('① manifest 接线完整', () => {
test('源表已声明,文件名 .csv', () => {
const t = manifest.tables.find((x) => x.table === 'customer_return_visit');
expect(t?.file).toBe('customer_return_visit.csv');
});
test('增量游标已配(push / 增量都要)', () => {
expect(manifest.incremental.per_query).toHaveProperty('customer_return_visit');
});
test('assembler 已注册', () => {
expect(manifest.assemblers.map((a) => a.file)).toContain(
'assemblers/patient_return_visit.yaml',
);
});
test('⭐ 任务头 customer_task 不单独摄入 —— 属性由宿主 inline', () => {
expect(manifest.tables.map((t) => t.table)).not.toContain('customer_task');
});
});
describe('② assembler 映射', () => {
test('是 upsert 资源(无 emits,不进 transaction/fact/召回)', () => {
expect(asm.canonical).toBe('patient_return_visit');
expect(asm).not.toHaveProperty('emits');
});
test.each([
['externalId', 'id'],
['patientExternalId', 'customer_id'],
['clinicId', 'organization_id'],
['taskDate', 'task_date'],
['taskStatus', 'task_status'],
['taskDirectorId', 'task_director_id'],
['sourceCreatedAt', 'created_gmt_at'],
])('%s ← %s', (canonical, host) => {
expect(asm.field_mapping[canonical]).toBe(host);
});
test('⭐ 治疗项走合并列、回访内容走剥标签列(而非原列)', () => {
expect(asm.field_mapping.treatmentItems).toBe('treatment_items_full');
expect(asm.field_mapping.followContent).toBe('follow_content_text');
});
});
describe('③ 枚举翻中文,与 jvs-dw 逐字对齐', () => {
test.each([
['1', '咨询回访'],
['2', '术后回访'],
['3', '常规回访'],
])('type 码%s → %s', (code, zh) => {
expect(asm.enum_mapping.type[code]).toBe(zh);
});
test('⭐ 码 4/5 必须显式配置 —— 官方注释只写 1-3,实测占 12% 不能丢', () => {
expect(asm.enum_mapping.type['4']).toBeTruthy();
expect(asm.enum_mapping.type['5']).toBeTruthy();
});
test('status / taskStatus 全码覆盖', () => {
expect(asm.enum_mapping.status['1']).toBe('已回访');
expect(asm.enum_mapping.status['2']).toBe('未回访');
for (const c of ['1', '2', '3', '4', '5', '6']) {
expect(`taskStatus[${c}]=${asm.enum_mapping.taskStatus[c]}`).not.toContain('undefined');
}
});
});
describe('④ transforms:两列合并 + 剥富文本', () => {
const engine = new TransformEngine();
const rows = [
{
id: '1',
treatment_items: '种植',
treatment_items_two: '单颗种植',
follow_content: '<p>骨灰盒或或</p>',
},
{
id: '2',
treatment_items: '诊后回访',
treatment_items_two: '',
follow_content: '<ol><li>甲</li><li>乙</li></ol>',
},
{ id: '3', treatment_items: '', treatment_items_two: '', follow_content: '纯文本' },
];
const out = engine.run({
tables: { customer_return_visit: rows },
transforms: manifest.transforms as TransformsConfig,
});
const got = (out['customer_return_visit'] ?? []) as Array<Record<string, unknown>>;
test('两列都有 → 用 · 连接', () => {
expect(got[0]!.treatment_items_full).toBe('种植 · 单颗种植');
});
test('只有一级 → 尾部留分隔符(落库前由 normalizeMergedItems 归一)', () => {
expect(String(got[1]!.treatment_items_full).startsWith('诊后回访')).toBe(true);
});
test('⭐ 富文本剥成纯文本,列表项保住段落边界', () => {
expect(got[0]!.follow_content_text).toBe('骨灰盒或或');
expect(got[1]!.follow_content_text).toBe('甲\n乙');
});
test('纯文本原样通过', () => {
expect(got[2]!.follow_content_text).toBe('纯文本');
});
});
describe('⑤ stripHtml 算子', () => {
test.each([
['<p>骨灰盒</p>', '骨灰盒'],
['<ol><li>甲</li><li>乙</li></ol>', '甲\n乙'],
['a<br>b', 'a\nb'],
['a&nbsp;b &amp; c', 'a b & c'],
['纯文本', '纯文本'],
])('%s → %s', (input, expected) => {
expect(stripHtml(input)).toBe(expected);
});
test('⭐ 全是标签/空白 → null(空壳不入库)', () => {
expect(stripHtml('<p> </p>')).toBeNull();
expect(stripHtml('')).toBeNull();
});
test('&amp;lt; 不被二次解码成 <(实体还原顺序)', () => {
expect(stripHtml('&amp;lt;')).toBe('&lt;');
});
test('⭐ 双重编码:剥完外层标签、还原实体后又露出字面标签 → 再剥一轮', () => {
// 线上真实数据(customer_return_visit id=266151):外层是真标签,内层是被转义的标签。
// 首轮导入时这条**没被剥干净**,落库后仍是 "<p>阿斯蒂芬撒的发生</p>" —— 由此加的有界第二轮。
const real =
'<p><span style="color:#000000"><span style="font-size:13px">' +
'<span style="background-color:#ffffff">&lt;p&gt;阿斯蒂芬撒的发生&lt;/p&gt;</span></span></span></p>';
expect(stripHtml(real)).toBe('阿斯蒂芬撒的发生');
});
test('只剥两轮,不做无界循环(畸形输入不能卡死摄入)', () => {
// 每轮都能再生出标签的构造输入:两轮后仍有残留是可接受的,关键是**必然终止**
const nested = '&amp;amp;lt;p&amp;amp;gt;x';
expect(typeof stripHtml(nested)).toBe('string');
});
});
});
import { readdirSync, readFileSync, statSync } from 'node:fs';
import { join } from 'node:path';
/**
* 「可空数组 + 显式 PG 类型转换」的地雷守卫。
*
* ═══ 这条规则从哪来 ═══════════════════════════════════════════════
* 2026-09-21 FRIDAY 宿主接入后,点「确认分配」必 500:
* 42846 cannot cast type integer[] to timestamp with time zone[]
*
* 机理:Prisma 从**值**推断数组元素类型。整个数组**非空且全是 null** 时推不出来,
* 退化成 `integer[]`,撞上显式 `::timestamptz[]` 就报 42846。
* ⚠️ 空数组 `[]` 反而安全(走不到转换);**只有「非空且全 null」才炸**。
*
* 为什么瑞尔一直没事、FRIDAY 一上来就炸:那一列是「该患者专属客服的最近到诊时间」,
* 瑞尔有回访历史 → 多数非 null;FRIDAY 是新宿主 → 整批全 null。
* ⇒ **这不是 FRIDAY 特有的**,瑞尔只要碰上「这批人恰好都没有专属客服」一样炸。
*
* ═══ 只有这两种目标类型会塌(2026-09-21 测试机实测)═══════════════
* ::timestamptz[] ❌ ::uuid[] ❌
* ::text[] ✅ ::double precision[] ✅ ::int[] ✅
* integer 到 text/float/int 有转换路径,到 timestamptz/uuid 没有。
* ⇒ ⛔ 别把同段里的 `::text[]` / `::double precision[]` 参数也"顺手修了",它们没病。
*
* ═══ 断言什么 ═════════════════════════════════════════════════════
* 凡是 `${x}::timestamptz[]` 或 `${x}::uuid[]`,若 `x` 的声明类型里有 `| null`,
* 就是地雷 —— 必须先走 `::text[]` 再整体转(见 plan-assignment.service.ts 的长注释)。
* ⚠️ 只锁「声明可空 ⇒ 不许裸转」这一条,⛔ 不锁具体写法/行号/措辞。
*/
const SRC = join(__dirname, '../src');
function walk(dir: string): string[] {
const out: string[] = [];
for (const name of readdirSync(dir)) {
const p = join(dir, name);
if (statSync(p).isDirectory()) out.push(...walk(p));
else if (p.endsWith('.ts')) out.push(p);
}
return out;
}
/** `${ident}::timestamptz[]` / `${ident}::uuid[]` —— 但排除已经过 text[] 中转的写法 */
const RISKY = /\$\{(\w+)\}::(timestamptz|uuid)\[\]/g;
describe('可空数组不许裸转 timestamptz[] / uuid[]', () => {
const files = walk(SRC);
it('仓库里没有「声明可空却裸转」的参数', () => {
const offenders: string[] = [];
for (const file of files) {
const src = readFileSync(file, 'utf8');
for (const m of src.matchAll(RISKY)) {
const ident = m[1];
const target = m[2];
// 找同文件里该变量的声明
const decl = new RegExp(`const\\s+${ident}\\s*:\\s*([^=]+)=`).exec(src)?.[1] ?? '';
if (/\|\s*null/.test(decl)) {
offenders.push(
`${file.replace(SRC, 'src')}: \${${ident}}::${target}[] —— 声明为 ${decl.trim()},` +
` 全 null 时会退化成 integer[] 并报 42846。改成 ::text[]::${target}[] 并传 ISO 文本/字符串。`,
);
}
}
}
expect(offenders).toEqual([]);
});
it('🔴 那个出过事的参数确实走了 text[] 中转', () => {
const f = readFileSync(join(SRC, 'modules/plan/plan-assignment.service.ts'), 'utf8');
expect(f).toContain('${aDedicatedCsLastVisit}::text[]::timestamptz[]');
expect(f).not.toMatch(/\$\{aDedicatedCsLastVisit\}::timestamptz\[\]/);
});
it('⛔ 没病的那两个没被顺手改坏(text[] / double precision[] 保持原样)', () => {
const f = readFileSync(join(SRC, 'modules/plan/plan-assignment.service.ts'), 'utf8');
expect(f).toContain('${aDedicatedCs}::text[]');
expect(f).toContain('${aSourceConfidence}::double precision[]');
});
it('数组里存的是 ISO 文本而不是 Date(否则 text[] 那步会塞进 Date 对象)', () => {
const f = readFileSync(join(SRC, 'modules/plan/plan-assignment.service.ts'), 'utf8');
expect(f).toMatch(/aDedicatedCsLastVisit:\s*\(string \| null\)\[\]/);
// ⚠️ 必须锁**这一行**的 toISOString —— 只写 toContain('toISOString()') 没牙:
// 文件别处也有它,把这行的删掉照样绿(2026-09-21 变异测试当场抓到)。
const push = /aDedicatedCsLastVisit\.push\([^;]*\);/.exec(f)?.[0] ?? '';
expect(push).toContain('toISOString()');
});
});
...@@ -10,9 +10,13 @@ import { withCohortDerivedPool } from '../src/prisma/prisma.service'; ...@@ -10,9 +10,13 @@ import { withCohortDerivedPool } from '../src/prisma/prisma.service';
* ① 批次栅栏 → worker pool:缺口正好是 E[8 次抽样最大值](≈p90)与均值之比; * ① 批次栅栏 → worker pool:缺口正好是 E[8 次抽样最大值](≈p90)与均值之比;
* 那个 75.7 秒的患者会把同批 7 个 worker 一起冻住。 * 那个 75.7 秒的患者会把同批 7 个 worker 一起冻住。
* ② --concurrency 没放大连接池 —— **潜在**缺陷,咬不咬人看该机 .env: * ② --concurrency 没放大连接池 —— **潜在**缺陷,咬不咬人看该机 .env:
* 测试服 URL 显式 `connection_limit=30` → 不受影响 * (原文说"测试服 URL 显式 connection_limit=30 所以不受影响" —— 2026-09-02 证伪:
* 生产 URL 没写 → Prisma 默认 核数×2+1,4 核机 = 9 → --concurrency>8 会被卡住 * compose 覆盖了 DATABASE_URL,两台容器里都没有 connection_limit,都是默认池)
* 所以按 --concurrency=8 跑生产时本修复无收益,价值在于解锁更高并发。 *
* 🔴 2026-09-02 补:上面那句"4 核机 = 9"是**错的**,已在实现里更正 ——
* Prisma 默认池按**物理核**算(×2+1),生产 nproc=4 但物理核只有 2 → 池 = **5**。
* 而常驻服务的 `PAC_RECALL_SUBSCENARIO_CONCURRENCY=4` 当时压根没被算进池 →
* 批次占住 4 条,每日健康日报连续三天在 09:00:10 超时。下面 describe 里锁的就是这条。
*/ */
describe('runPool —— 连续调度,不做批次栅栏', () => { describe('runPool —— 连续调度,不做批次栅栏', () => {
...@@ -66,12 +70,17 @@ describe('runPool —— 连续调度,不做批次栅栏', () => { ...@@ -66,12 +70,17 @@ describe('runPool —— 连续调度,不做批次栅栏', () => {
describe('连接池随并发放大 —— PAC_DB_CONCURRENCY', () => { describe('连接池随并发放大 —— PAC_DB_CONCURRENCY', () => {
const URL_ = 'postgresql://u:p@db:5432/pac?schema=public'; const URL_ = 'postgresql://u:p@db:5432/pac?schema=public';
const saved = { db: process.env.PAC_DB_CONCURRENCY, cohort: process.env.PAC_COHORT_CONCURRENCY }; const saved = {
const set = (db?: string, cohort?: string) => { db: process.env.PAC_DB_CONCURRENCY,
cohort: process.env.PAC_COHORT_CONCURRENCY,
recall: process.env.PAC_RECALL_SUBSCENARIO_CONCURRENCY,
};
const set = (db?: string, cohort?: string, recall?: string) => {
if (db === undefined) delete process.env.PAC_DB_CONCURRENCY; else process.env.PAC_DB_CONCURRENCY = db; if (db === undefined) delete process.env.PAC_DB_CONCURRENCY; else process.env.PAC_DB_CONCURRENCY = db;
if (cohort === undefined) delete process.env.PAC_COHORT_CONCURRENCY; else process.env.PAC_COHORT_CONCURRENCY = cohort; if (cohort === undefined) delete process.env.PAC_COHORT_CONCURRENCY; else process.env.PAC_COHORT_CONCURRENCY = cohort;
if (recall === undefined) delete process.env.PAC_RECALL_SUBSCENARIO_CONCURRENCY; else process.env.PAC_RECALL_SUBSCENARIO_CONCURRENCY = recall;
}; };
afterAll(() => set(saved.db, saved.cohort)); afterAll(() => set(saved.db, saved.cohort, saved.recall));
const limitOf = (u: string | undefined) => const limitOf = (u: string | undefined) =>
u ? new URL(u).searchParams.get('connection_limit') : null; u ? new URL(u).searchParams.get('connection_limit') : null;
...@@ -101,9 +110,30 @@ describe('连接池随并发放大 —— PAC_DB_CONCURRENCY', () => { ...@@ -101,9 +110,30 @@ describe('连接池随并发放大 —— PAC_DB_CONCURRENCY', () => {
}); });
test('都没设 / =1(常驻 API 进程)→ 不动 URL,走 Prisma 默认池', () => { test('都没设 / =1(常驻 API 进程)→ 不动 URL,走 Prisma 默认池', () => {
set(undefined, undefined); set(undefined, undefined, undefined);
expect(withCohortDerivedPool(URL_)).toBe(URL_);
set('1', '1', '1');
expect(withCohortDerivedPool(URL_)).toBe(URL_); expect(withCohortDerivedPool(URL_)).toBe(URL_);
set('1', '1'); });
// ── 🔴 2026-09-02 生产事故:常驻服务的并发旋钮没被算进池 ──
// PAC_RECALL_SUBSCENARIO_CONCURRENCY=4 让批次长期占住 4 条连接,而常驻进程用的是
// Prisma 默认池(生产 2 物理核 → 5)。每日 09:00 健康日报的 6 个并发 count 抢不到连接,
// 连续三天在 09:00:10(= 10s pool_timeout)整点失败,企微收不到日报。
test('⭐ 常驻服务的召回并发旋钮也要放大池(否则批次占满,日报/API 抢不到连接)', () => {
set(undefined, undefined, '4');
expect(limitOf(withCohortDerivedPool(URL_))).toBe(expected(4)); // 29,不再是默认的 5
});
test('召回旋钮与另外两个一起取最大,不会互相盖掉', () => {
set('2', '3', '6');
expect(limitOf(withCohortDerivedPool(URL_))).toBe(expected(6)); // 41 → 封顶 40
set('9', '3', '4');
expect(limitOf(withCohortDerivedPool(URL_))).toBe(expected(9)); // CLI 旋钮更大时以它为准
});
test('召回旋钮 =1(默认串行)→ 行为不变,仍走 Prisma 默认池', () => {
set(undefined, undefined, '1');
expect(withCohortDerivedPool(URL_)).toBe(URL_); expect(withCohortDerivedPool(URL_)).toBe(URL_);
}); });
......
import { PlanEngineService } from '../src/modules/plan/engine/plan-engine.service'; import { PlanEngineService, resolvePlanBatchSize } from '../src/modules/plan/engine/plan-engine.service';
import type { ScenarioHit } from '../src/modules/plan/engine/scenario.interface'; import type { ScenarioHit } from '../src/modules/plan/engine/scenario.interface';
/** /**
...@@ -1156,3 +1156,120 @@ describe('归因继承的边界 — 判据是「客服碰过没」', () => { ...@@ -1156,3 +1156,120 @@ describe('归因继承的边界 — 判据是「客服碰过没」', () => {
for (const [args] of called) expect(args.where.planId.in).not.toContain('p-g'); for (const [args] of called) expect(args.where.planId.in).not.toContain('p-g');
}); });
}); });
/**
* 🔴 2026-09-04 端到端分批改造的**核心判据**:结果与批大小无关。
*
* 事故背景:生产 pac-service 挂死约 11.8 小时(2026-09-04 20:06 → 次日 07:52 硬重启)。
* ⚠️ 09-05 更正:当晚**没有** OOM kill(那条内核日志是 08-29 的陈旧 tty 回显),
* 机器也**没有 swap** —— 真正形态是无 swap 下的 page-cache refault 活锁,
* 因为始终有文件页可回收,out_of_memory() 从未被调用,所以没人被杀、也就没有自愈。
* 详见 plan-engine.service.ts 里 resolvePlanBatchSize 上方那段更正说明。
* 堆账里 latestByPatient(1.93GB)+ hitsByPatient(1.06GB)是大头,而它们原先**全程不释放**
* —— 取数早就是 2000 一块,但产物累积在跨全量存活的 Map 里,runPool 的闭包又把它们
* context-allocate 到整个 runAllForHost frame 结束。改造把取数分块升格成端到端分批。
*
* 为什么这组用例是必须的:上面那 20+ 条既有用例**一条都跑不到分批路径的差异** ——
* 它们每个只有 1~4 个患者,默认批大小 2000,永远只有一批。分批写错了它们全绿。
* 这里用 describe.each 把同一份 fixture 在不同批大小下各跑一遍,直接编码
* 「结果与批大小无关」这条不变式。
*
* ⚠️ batchSize=1 是最凶的一档:每批一个患者,同时压测三件事 ——
* ① 跨批不误关(stale-close 拿的是跨批累加的 hitPatientIds,不是被逐批清空的 hitsByPatient)
* ② touchedPlanIds 按批算与全量算等价
* ③ snooze 提在循环外之后仍能被每一批查到
*/
describe('端到端分批 — 结果必须与 PAC_PLAN_BATCH_SIZE 无关', () => {
const prev = process.env.PAC_PLAN_BATCH_SIZE;
afterEach(() => {
if (prev === undefined) delete process.env.PAC_PLAN_BATCH_SIZE;
else process.env.PAC_PLAN_BATCH_SIZE = prev;
});
/// 与上面「混合一批」同一份 fixture:5 种结局各一,外加一个 0 命中的患者压 stale-close
const runMixed = async () => {
const { prisma, plans, logs } = makeStore({
plans: [
{ id: 'u-old', patientId: 'p2', status: 'active', priorityScore: 50, reasons: [{ scenario: SCEN, subKey: 'k@1' }] },
{ id: 's-old', patientId: 'p3', status: 'active', reasons: [{ scenario: SCEN, subKey: 'k@old' }] },
{
id: 't-term',
patientId: 'p4',
status: 'abandoned',
snoozedUntil: new Date('2026-12-01T00:00:00Z'),
reasons: [{ scenario: SCEN, subKey: 'k@4' }],
},
// ⭐ p9 本轮 0 命中且有 active plan → 必须被 stale-close 关掉。
// 分批写错(第 3 步进了循环 / 用了被清空的 hitsByPatient)时,
// 这里会连 p1~p4 一起关掉,plansClosed 从 1 变 5 —— 这条就是照妖镜。
{ id: 'stale-old', patientId: 'p9', status: 'active', reasons: [{ scenario: SCEN, subKey: 'k@gone' }] },
],
});
const res = await engine(
prisma,
makeScenario([
hit('p1', 'k@1'),
hit('p2', 'k@1', 50),
hit('p3', 'k@new'),
hit('p4', 'k@4'),
]),
).runAllForHost({ hostId: HOST, tenantId: TENANT, now: NOW });
return { res, plans, logs };
};
/// 把最终状态压成一个可比对的快照(引擎返回值 + 落库结果),批大小不该改变它任何一位
const snapshot = (r: Awaited<ReturnType<typeof runMixed>>) => ({
result: {
scenariosRun: r.res.scenariosRun,
patientsHit: r.res.patientsHit,
plansCreated: r.res.plansCreated,
plansSuperseded: r.res.plansSuperseded,
plansUnchanged: r.res.plansUnchanged,
plansSuppressed: r.res.plansSuppressed,
plansClosed: r.res.plansClosed,
plansSkippedAssigned: r.res.plansSkippedAssigned,
},
// plan 终态:按 (patientId, version) 排序,避免并发写入顺序带来的假差异
plans: r.plans
.map((p) => `${p.patientId}|v${p.version}|${p.status}|${p.priorityScore}`)
.sort(),
logs: r.logs.map((l) => `${l.patientId}|${l.status}`).sort(),
});
const SIZES = [1, 2, 3, 2000];
let baseline: ReturnType<typeof snapshot> | null = null;
test.each(SIZES)('批大小 %s → 结果与基线逐字段一致', async (size) => {
process.env.PAC_PLAN_BATCH_SIZE = String(size);
const snap = snapshot(await runMixed());
// 先自证这份 fixture 真的走到了 5 种结局,否则"全相等"可能只是都没跑
expect(snap.result).toMatchObject({
patientsHit: 4,
plansCreated: 1,
plansUnchanged: 1,
plansSuperseded: 1,
plansSuppressed: 1,
plansClosed: 1, // ⭐ 只关 p9;若变成 5 说明分批把其他批的患者误判成"信号消失"
});
if (baseline === null) baseline = snap;
else expect(snap).toEqual(baseline);
});
/**
* 🔴 2026-09-04 测试机实跑炸出来的:初版有个「0 = single-shot 不分批」的回退档,
* 它**绕过了 20000 上限** → 88,728 个患者一次塞进 `patientId: { in: ids }` →
* `too many bind variables ... expected maximum of 32767, received 32769`。
* 而且它压根不等价于改造前(改造前内部有 CHUNK=2000,取数从没超过 2000 个 id)。
* 已删除。这条用例钉住「任何取值都必须落在 [1,20000]」,防止有人再加回来。
*/
test('⛔ 0 也必须被夹进 [1,20000] —— 不许存在绕过 bind 上限的档位', () => {
expect(resolvePlanBatchSize({ PAC_PLAN_BATCH_SIZE: '0' } as NodeJS.ProcessEnv)).toBe(2000);
});
test('⛔ 批大小夹在 [1, 20000] —— 上限守 PG 32767 bind(latest/persona 走 patientId in ids)', () => {
expect(resolvePlanBatchSize({ PAC_PLAN_BATCH_SIZE: '999999' } as NodeJS.ProcessEnv)).toBe(20_000);
expect(resolvePlanBatchSize({ PAC_PLAN_BATCH_SIZE: '-5' } as NodeJS.ProcessEnv)).toBe(1);
expect(resolvePlanBatchSize({} as NodeJS.ProcessEnv)).toBe(2000);
expect(resolvePlanBatchSize({ PAC_PLAN_BATCH_SIZE: 'abc' } as NodeJS.ProcessEnv)).toBe(2000);
});
});
/**
* reparse --dry-run 的 bind 变量上限回归。
*
* ═══ 事故(2026-08-28 测试服实测)═════════════════════════════════════
* pnpm reparse:prod -- --host=jvs-dw --subject-type=treatment \
* --patients-file=/tmp/reparse-ids.txt --dry-run
* → Assertion violation on the database:
* `too many bind variables in prepared statement, expected maximum of 32767, received 62899`
*
* 根因是**写法**不是业务:dryRun 分支的 count 把整份 patientIds 一次塞进 `patientId IN (...)`,
* 每个 id 占一个 bind,PG 单条 prepared statement 上限 32767。实跑分支早就按 BATCH 切了片,
* 所以**只有 dry-run 会崩**。
*
* 讽刺点值得钉住:`--patients-file` 的文档写着「按受影响患者收窄是 reparse 最有效的提速手段
* (实测可达 250 倍)」—— 清单越大越该先 dry-run 探一探,而那恰恰是它唯一不工作的场景。
*
* 跑:pnpm --filter @pac/service test -- reparse-dry-run-bind-limit
*/
import { ColdImportService } from '../src/modules/sync/cold-import/cold-import.service';
import * as path from 'node:path';
const FIXTURE_DIR = path.join(__dirname, 'fixtures', 'reparse-bind-limit');
/** PG 单条 prepared statement 的 bind 上限 —— 事故里就是被这堵墙拦下的 */
const BIND_MAX = 32_767;
type CountArgs = { where: { patientId?: { in: string[] } } };
/**
* 假 prisma:**照 PG 的规矩发脾气** —— 单条 count 的 bind 数超上限就抛,
* 跟真库同样的报错。不这么做的话,不分批也「测过了」。
*/
function makePrisma(seen: number[]) {
return {
host: { findFirst: async () => ({ id: 'host-1', name: 'fixture-host' }) },
patientTransaction: {
count: async (args: CountArgs) => {
const ids = args.where.patientId?.in ?? [];
// where 里除 IN 之外还有 hostId / subjectType 两个 bind
const binds = ids.length + 2;
if (binds > BIND_MAX) {
throw new Error(
`Assertion violation on the database: too many bind variables in prepared statement, ` +
`expected maximum of ${BIND_MAX}, received ${binds}`,
);
}
seen.push(ids.length);
return ids.length; // 每个患者算一条 txn,便于断言累加没丢
},
findMany: async () => [],
},
};
}
function makeService(prisma: unknown): ColdImportService {
// dryRun 只用到 prisma + 磁盘上的 manifest/assembler,其余依赖走不到
const nope = null as never;
return new ColdImportService(prisma as never, nope, nope, nope, nope, nope, nope);
}
const ids = (n: number) => Array.from({ length: n }, (_, i) => `p-${i}`);
describe('reparse --dry-run 不许把整份 patientIds 一次塞进 IN', () => {
it('🔴 3.2 万+ 患者清单(事故量级)dry-run 不抛错', async () => {
const seen: number[] = [];
const svc = makeService(makePrisma(seen));
await expect(
svc.reparseFromTransactions({
dir: FIXTURE_DIR,
hostName: 'fixture-host',
patientIds: ids(62_899),
dryRun: true,
}),
).resolves.toBeDefined();
// 真发出去了(不是被「非 transform 产出」静默跳过 → 一次 count 都没发的假绿)
expect(seen.length).toBeGreaterThan(1);
expect(Math.max(...seen)).toBeLessThanOrEqual(BIND_MAX - 2);
// 分片不重不漏
expect(seen.reduce((a, b) => a + b, 0)).toBe(62_899);
});
it('⭐ 不给患者清单时走全量 count(没有 IN,不占 bind)', async () => {
const seen: number[] = [];
const svc = makeService(makePrisma(seen));
await svc.reparseFromTransactions({
dir: FIXTURE_DIR,
hostName: 'fixture-host',
dryRun: true,
});
expect(seen).toEqual([0]); // 一条 count,where 里没有 patientId IN
});
it('⛔ PAC_REPARSE_BATCH 调过 bind 上限也不许把墙放回来', async () => {
const prev = process.env.PAC_REPARSE_BATCH;
process.env.PAC_REPARSE_BATCH = '100000'; // 有人为了「跑快点」把批调大
try {
const seen: number[] = [];
const svc = makeService(makePrisma(seen));
await expect(
svc.reparseFromTransactions({
dir: FIXTURE_DIR,
hostName: 'fixture-host',
patientIds: ids(50_000),
dryRun: true,
}),
).resolves.toBeDefined();
expect(Math.max(...seen)).toBeLessThanOrEqual(BIND_MAX - 2);
} finally {
if (prev === undefined) delete process.env.PAC_REPARSE_BATCH;
else process.env.PAC_REPARSE_BATCH = prev;
}
});
});
...@@ -45,8 +45,14 @@ describe('闸门覆盖范围 — 只关写侧', () => { ...@@ -45,8 +45,14 @@ describe('闸门覆盖范围 — 只关写侧', () => {
test('⭐ 增量摄入(会改数据)受闸控,且连僵尸锁回收一并跳过', () => { test('⭐ 增量摄入(会改数据)受闸控,且连僵尸锁回收一并跳过', () => {
const s = read('sync-incremental.scheduler.ts'); const s = read('sync-incremental.scheduler.ts');
expect(s).toContain('schedulerDisabled()'); expect(s).toContain('schedulerDisabled()');
// 闸判定必须在 reapStaleRunningLocks 之前 —— 共库场景下回收会误清老实例正在跑的锁 // 闸判定必须在**第一次** reapStaleRunningLocks 调用之前 —— 共库场景下回收会误清老实例正在跑的锁。
expect(s.indexOf('schedulerDisabled()')).toBeLessThan(s.indexOf('await this.reapStaleRunningLocks()')); // ⚠️ 2026-09-04:调用点从 `reapStaleRunningLocks()` 变成带参数的两处(启动态传
// PROCESS_STARTED_AT,周期态在 runHostSafe 里传 undefined+前缀),写死字面量会假失败。
// 周期态那处是**传递性受闸**的:闸命中时 onModuleInit 提前 return,cron 根本不注册,
// runHostSafe 无从被调用 —— 所以这里只需锁住"闸在第一次回收之前"。
const firstReap = s.indexOf('this.reapStaleRunningLocks(');
expect(firstReap).toBeGreaterThan(-1);
expect(s.indexOf('schedulerDisabled()')).toBeLessThan(firstReap);
}); });
test('⭐ persona stale-scan(会 enqueue 重算)受闸控', () => { test('⭐ persona stale-scan(会 enqueue 重算)受闸控', () => {
......
...@@ -57,6 +57,10 @@ describe('runHostSafe 防重入', () => { ...@@ -57,6 +57,10 @@ describe('runHostSafe 防重入', () => {
const { calls, run, release } = makeService(); const { calls, run, release } = makeService();
const a = run('jvs-dw'); const a = run('jvs-dw');
const b = run('friday'); const b = run('friday');
// ⚠️ 2026-09-04:runHostSafe 在进 runOne 之前多了一次 `await 周期回收僵尸锁`,
// 于是 runOne 不再在同一微任务里被调到 —— 断言前必须把微任务队列放干,
// 否则 calls 还是空的(这是真实的行为变化,不是测试写错)。
await new Promise((r) => setImmediate(r));
// runOne 收到的是 path.join(dataDir, host) 的完整路径,按后缀断言 // runOne 收到的是 path.join(dataDir, host) 的完整路径,按后缀断言
expect(calls.map((d) => d.split('/').pop())).toEqual(['jvs-dw', 'friday']); expect(calls.map((d) => d.split('/').pop())).toEqual(['jvs-dw', 'friday']);
release(); release();
......
...@@ -10,6 +10,7 @@ ...@@ -10,6 +10,7 @@
|---|---|---| |---|---|---|
| `deploy/gen-env.sh` | 生成 `.env`(密钥随机 + 生产开关 + 业务值注入 + 缺失报告) | 拒覆盖已有(`FORCE=1` 强制) | | `deploy/gen-env.sh` | 生成 `.env`(密钥随机 + 生产开关 + 业务值注入 + 缺失报告) | 拒覆盖已有(`FORCE=1` 强制) |
| `deploy/deploy-prod.sh` | 拉码 + build + 重建容器 + 硬验证(镜像/迁移/health) | ✅ 反复跑 | | `deploy/deploy-prod.sh` | 拉码 + build + 重建容器 + 硬验证(镜像/迁移/health) | ✅ 反复跑 |
| `deploy/deploy-docs.sh` | **文档站单独部署**(pac-docs 不在 deploy-prod.sh 里)+ 硬验证 OpenAPI Server URL | ✅ 反复跑 |
| `deploy/first-import.sh` | 数据首灌:seed + 存量 + 重算画像/计划 | ✅ 可重跑 | | `deploy/first-import.sh` | 数据首灌:seed + 存量 + 重算画像/计划 | ✅ 可重跑 |
--- ---
...@@ -34,6 +35,25 @@ build 阶段不影响服务(老容器一直跑),停机只发生在重建那一 ...@@ -34,6 +35,25 @@ build 阶段不影响服务(老容器一直跑),停机只发生在重建那一
> 改了 `NEXT_PUBLIC_*`(前端 build-time 变量)必须走本脚本重 build;只改后端 `.env` 用 `docker compose -f docker-compose.prod.yml restart pac-service` 即可。 > 改了 `NEXT_PUBLIC_*`(前端 build-time 变量)必须走本脚本重 build;只改后端 `.env` 用 `docker compose -f docker-compose.prod.yml restart pac-service` 即可。
### 改了文档(`apps/pac-docs/`)要另外跑一条
`deploy-prod.sh` **只构建 pac-migrate / pac-service / pac-web,不含 pac-docs** —— 文档站要单独部署:
```bash
bash deploy/deploy-docs.sh
```
**为什么分开**:pac-docs 的 Dockerfile 是 `COPY . .`(整个仓库),任何代码改动都会让它的
build 缓存失效、Next.js SSG 全量重跑。实测(2026-09-03 测试服):什么都没改 **3 秒**,
改了后端一行代码 **111 秒**。而绝大多数部署都是改后端、文档没动 —— 并进主流程等于每次白花约 2 分钟。
⚠️ **不要手敲 `docker compose build pac-docs`**。docs 镜像里 OpenAPI 的「Server URL」来自
build arg `DOCS_API_URL`,它回落到 `NEXT_PUBLIC_API_BASE_URL`,而后者在 **`apps/pac-web/.env`**
里(不在 `pac-service/.env`)。少传一个 `--env-file` 就会变成 `http://localhost:3101`,
**且构建照样成功、不报任何错** —— 只有点开 API 参考页才看得出来,对接方照着填会打不通
(2026-09-03 实际踩到,线上挂了几天没人发现)。`deploy-docs.sh` 固化了两个 env-file
并在构建后硬校验 servers 地址,写错立刻失败。
--- ---
## 2. 新环境从零上线(换服务器 / 首次生产) ## 2. 新环境从零上线(换服务器 / 首次生产)
......
#!/usr/bin/env bash
# PAC 文档站(pac-docs)单独部署
#
# 【为什么不并进 deploy-prod.sh】
# pac-docs 的 Dockerfile 里是 `COPY . .`(整个仓库),所以**任何**代码改动都会让它的
# build 缓存失效、Next.js SSG 全量重跑。实测(2026-09-03,测试服):
# 什么都没改 → 3 秒 (17/17 层缓存命中)
# 改了后端代码一行 → 111 秒(COPY 之后全失效)
# 而绝大多数部署都是改后端、文档没动 —— 并进主流程等于每次白白多花约 2 分钟。
# 故保持分离:改了 apps/pac-docs/ 下的东西才跑本脚本。
#
# 【为什么要有这个脚本,而不是手敲 docker compose】
# docs 镜像里的 OpenAPI「Server URL」来自 build arg DOCS_API_URL,而 compose 里它是
# ${DOCS_API_URL:-${NEXT_PUBLIC_API_BASE_URL:-http://localhost:3101}}
# NEXT_PUBLIC_API_BASE_URL 在 **apps/pac-web/.env** 里(不在 pac-service/.env)。
# 手敲时少传一个 --env-file 就会一路回落到 localhost:3101 —— 而且**构建照样成功、
# 不报任何错**,只有点开 API 参考页才看得出来,对接方照着填就打不通。
# 2026-09-03 实际踩到:之前几次手工重建 docs 都只传了 pac-service/.env,
# 线上 API 参考页的 Server URL 一直是 localhost:3101。
# → 本脚本固化两个 env-file,并在构建后**硬校验** servers 地址,写错立刻失败而非静默上线。
#
# 【自更新】脚本会 git pull 更新自己。整个逻辑包在 main() 里防"字节偏移错位"(同 deploy-prod.sh),
# 但那还不够:本次进程跑的是**内存里已载入的旧版函数**,新逻辑这一轮不生效。
# 故 pull 后若本文件有变化,exec 新版重跑一次(带 --no-pull 防递归)。
#
# 用法(服务器上,~/pac):
# bash deploy/deploy-docs.sh # 部署文档站
# COMPOSE_MANAGED=1 bash deploy/deploy-docs.sh # 托管 DB 形态的机器(同 deploy-prod.sh)
# bash deploy/deploy-docs.sh --no-pull # 跳过 git 拉取(代码已就位时)
set -euo pipefail
log() { printf '\n\033[1;36m== %s ==\033[0m\n' "$*"; }
die() { printf '\033[1;31mFAIL: %s\033[0m\n' "$*" >&2; exit 1; }
main() {
cd "$(dirname "$0")/.."
if [[ "${1:-}" != "--no-pull" ]]; then
log "git 拉取($(git rev-parse --abbrev-ref HEAD))"
# ⚠️ 脚本会拉取更新**自己**。包在 main() 里能防"字节偏移错位",但防不了另一件事:
# 本次进程跑的仍是**内存里已载入的旧版函数**,拉下来的新逻辑这一轮不生效。
# 2026-09-03 实测:文件已是新版(curl -sL),报错文案却还是旧版的,排查了一轮才明白。
# → pull 后若脚本自身变了,用新版 exec 重跑一次(带 --no-pull 防无限递归)。
local before after
before=$(git rev-parse HEAD:deploy/deploy-docs.sh 2>/dev/null || echo none)
git pull --ff-only
after=$(git rev-parse HEAD:deploy/deploy-docs.sh 2>/dev/null || echo none)
if [[ "$before" != "$after" ]]; then
log "本脚本已更新 → 用新版重跑"
exec bash "$0" --no-pull
fi
fi
# compose 组装与 deploy-prod.sh 保持一致 —— 两个 env-file 缺一不可(见文件头说明)
local COMPOSE=(docker compose -f docker-compose.prod.yml)
[[ "${COMPOSE_MANAGED:-0}" == "1" ]] && COMPOSE+=(-f docker-compose.managed.yml)
COMPOSE+=(--env-file apps/pac-service/.env --env-file apps/pac-web/.env)
# 构建前先把将要注入的地址打出来,肉眼可核
local api_url
api_url=$("${COMPOSE[@]}" config 2>/dev/null | grep -m1 'DOCS_API_URL:' | sed 's/.*DOCS_API_URL: *//')
log "OpenAPI Server URL 将注入为:${api_url:-(空)}"
[[ -n "$api_url" ]] || die "解析不到 DOCS_API_URL —— 检查 apps/pac-web/.env 是否有 NEXT_PUBLIC_API_BASE_URL"
case "$api_url" in
*localhost*|*127.0.0.1*)
die "DOCS_API_URL 解析成了本地地址($api_url)。
多半是 apps/pac-web/.env 缺 NEXT_PUBLIC_API_BASE_URL,或该文件不存在。
这么构建出来的 API 参考页,对接方照着填会打不通。" ;;
esac
log "build pac-docs(显式,不和 up 混 —— 同 deploy-prod.sh 的理由)"
"${COMPOSE[@]}" build pac-docs
log "force-recreate(不信 compose 的重建判定)"
"${COMPOSE[@]}" up -d --force-recreate pac-docs
# ── 验证 1/2:镜像里的 spec 确实带对了地址(而不是只看构建日志) ──
log "验证 1/2:容器内 openapi spec 的 servers"
local got
got=$(docker exec "$("${COMPOSE[@]}" ps -q pac-docs)" \
node -e 'const d=require("/app/apps/pac-docs/openapi/pac.json");process.stdout.write(String((d.servers&&d.servers[0]&&d.servers[0].url)||""))' 2>/dev/null || true)
[[ "$got" == "$api_url" ]] \
|| die "spec 里的 servers 是「$got」,期望「$api_url」"
printf ' servers OK (%s)\n' "$got"
# ── 验证 2/2:站点起得来 ──
# ⚠️ 必须 -L 跟随重定向:根路径是 307 → /docs(fumadocs 的正常行为),
# 直接判 200 会把健康的站误报成故障(2026-09-03 首次跑本脚本即踩到)。
log "验证 2/2:文档站可访问"
local code
for _ in $(seq 1 20); do
code=$(curl -sL -o /dev/null -w '%{http_code}' http://localhost:3102/ || true)
[[ "$code" == "200" ]] && break
sleep 2
done
[[ "$code" == "200" ]] || die "文档站未就绪(跟随重定向后 HTTP $code)"
printf ' docs 200(loopback)\n'
log "文档站部署成功 ✅ $(git log --oneline -1)"
}
main "$@"
...@@ -24,6 +24,22 @@ log() { printf '\n\033[1;36m== %s ==\033[0m\n' "$*"; } ...@@ -24,6 +24,22 @@ log() { printf '\n\033[1;36m== %s ==\033[0m\n' "$*"; }
warn() { printf '\033[1;33mWARN: %s\033[0m\n' "$*" >&2; } warn() { printf '\033[1;33mWARN: %s\033[0m\n' "$*" >&2; }
die() { printf '\033[1;31mFAIL: %s\033[0m\n' "$*" >&2; exit 1; } die() { printf '\033[1;31mFAIL: %s\033[0m\n' "$*" >&2; exit 1; }
# 探活。⚠️ 两个坑,都踩过:
# ① **路由**:真实健康路由是 `/health` —— 它在 GLOBAL_PREFIX_EXCLUDE 里,**不带** /pac/v1 前缀。
# ② **判据**:全局响应拦截器把 404 也包成 HTTP 200 ——
# /pac/v1/health → 200 {"code":10004,"msg":"Cannot GET /pac/v1/health"}
# 于是"只看状态码"的探活**恒真**。本脚本此前探的正是这个不存在的路由(2026-09-05 发现),
# 三项硬验证里的 health 那项一直只证明了"Nest 的 HTTP 层还在应答",没证明应用健康。
# ⛔ 以后改这里,判据必须落在 body 上,别退回只看 %{http_code}。
probe_health() { # $1=base(如 http://127.0.0.1:3101) → stdout "<code> <body截断>";健康时返回 0
local out code body
out=$(curl -s -m 5 -w '\n%{http_code}' "$1/health" 2>/dev/null) || out=$'\n000'
code=${out##*$'\n'}
body=${out%$'\n'*}
printf '%s %s' "$code" "${body:0:120}"
[[ "$code" == "200" && "$body" == *'"status":"ok"'* ]]
}
main() { main() {
cd "$(dirname "$0")/.." cd "$(dirname "$0")/.."
...@@ -135,13 +151,12 @@ main() { ...@@ -135,13 +151,12 @@ main() {
echo " migrate OK" echo " migrate OK"
log "验证 3/3:health + web" log "验证 3/3:health + web"
local i code webcode local i hres webcode ok=0
for i in $(seq 1 30); do for i in $(seq 1 30); do
code=$(curl -s -m 5 -o /dev/null -w '%{http_code}' http://127.0.0.1:3101/pac/v1/health || true) if hres=$(probe_health http://127.0.0.1:3101); then ok=1; break; fi
[[ "$code" == "200" ]] && break
sleep 2 sleep 2
done done
[[ "${code:-}" == "200" ]] || die "service health=$code" [[ "$ok" == 1 ]] || die "service /health 不健康(重试 30 次):${hres:-无响应}"
# / 现为 307→/plans(next.config redirects),直接验 /plans 的 200 # / 现为 307→/plans(next.config redirects),直接验 /plans 的 200
webcode=$(curl -s -m 30 -o /dev/null -w '%{http_code}' http://127.0.0.1:3100/plans || true) webcode=$(curl -s -m 30 -o /dev/null -w '%{http_code}' http://127.0.0.1:3100/plans || true)
[[ "$webcode" == "200" ]] || die "web http=$webcode" [[ "$webcode" == "200" ]] || die "web http=$webcode"
...@@ -155,8 +170,7 @@ main() { ...@@ -155,8 +170,7 @@ main() {
local priv phealth pweb local priv phealth pweb
priv=$(ip route get 1 2>/dev/null | awk '{for(i=1;i<=NF;i++) if($i=="src"){print $(i+1);exit}}') priv=$(ip route get 1 2>/dev/null | awk '{for(i=1;i<=NF;i++) if($i=="src"){print $(i+1);exit}}')
[[ -n "$priv" ]] || die "拿不到私网 IP(ip route get 1 解析失败)" [[ -n "$priv" ]] || die "拿不到私网 IP(ip route get 1 解析失败)"
phealth=$(curl -s -m 5 -o /dev/null -w '%{http_code}' "http://$priv:3101/pac/v1/health" || true) phealth=$(probe_health "http://$priv:3101") || die "私网 $priv:3101 /health 不健康($phealth)— 应用可能绑成 loopback,网关够不到(会 502)。托管环境需 web/service 绑 0.0.0.0(见 docker-compose.managed.yml)"
[[ "$phealth" == "200" ]] || die "私网 $priv:3101 health=$phealth — 应用可能绑成 loopback,网关够不到(会 502)。托管环境需 web/service 绑 0.0.0.0(见 docker-compose.managed.yml)"
pweb=$(curl -s -m 30 -o /dev/null -w '%{http_code}' "http://$priv:3100/plans" || true) pweb=$(curl -s -m 30 -o /dev/null -w '%{http_code}' "http://$priv:3100/plans" || true)
[[ "$pweb" == "200" ]] || die "私网 $priv:3100 web=$pweb — 同上(loopback 绑定 → 网关 502)" [[ "$pweb" == "200" ]] || die "私网 $priv:3100 web=$pweb — 同上(loopback 绑定 → 网关 502)"
echo " 私网 $priv health 200 / web 200(网关路径 OK)" echo " 私网 $priv health 200 / web 200(网关路径 OK)"
......
...@@ -91,6 +91,80 @@ services: ...@@ -91,6 +91,80 @@ services:
dockerfile: apps/pac-service/Dockerfile dockerfile: apps/pac-service/Dockerfile
target: prod target: prod
restart: always restart: always
# 🔴 **容器内存上限** —— ⛔ 别删。它不省内存,它决定「出事时炸多大」。
#
# 【为什么加】2026-09-04 生产宕 11.8 小时。同一台机、同样吃到 ~6GB,对照 08-29:
# 08-29 内核 OOM kill → 进程死 → restart:always 拉起 → 分钟级自愈
# 09-04 三道兜底全被绕过 → **整机活锁** → 挂 11.8h,要人硬重启
# 三道兜底怎么被绕过的:
# ① V8 abort —— 天花板 08-26 抬到 8G,堆没顶到就先把机器吃穷了
# ② cgroup OOM —— 容器没上限(HostConfig.Memory=0),这条压根不存在 ← 本行修的就是它
# ③ 内核 OOM killer —— 机器**没有 swap**,内核宁可反复丢/重读 page cache 也不判 OOM
# (file page 永远"可回收"),于是没有任何一次 out_of_memory() 被触发 → refault 活锁。
# 所以本行买的不是"不涨",是**把爆炸半径从整机缩回单容器**:越界 → SIGKILL
# → restart:always 拉起 → 回到 08-29 那种分钟级自愈。
# ✅ 2026-09-05 在测试机用**同一个 pac-service 镜像**、256MB 上限的探针实证过整条链:
# 内核 `oom-kill:constraint=CONSTRAINT_MEMCG` → docker 自动重启(RestartCount 累加)。
# 同一次实测还记到 rss=241MB 时 heapUsed 只有 4MB —— **堆上限管不到堆外**,
# 这正是 max-old-space-size 当不了兜底、必须有本行的原因。
# ⚠️ 事后查证别看 `docker inspect` 的 State.ExitCode / OOMKilled:restart:always 下
# 它们显示的是**重启后那条命**(实测内核杀了 2 次,inspect 仍报 ExitCode=0 OOMKilled=false)。
# 判据是 RestartCount 变大 + `dmesg -T | grep CONSTRAINT_MEMCG`。
#
# 【8g 这个数怎么来的】(生产 pac-friday 实测)
# 全机 15.36 GB
# OS + page cache 必须留 ~2.5 GB ← 无 swap,这块被挤掉就是活锁本身
# dockerd(闲时 0.4G,08-29 压力下实测 2.7G) ~2.7 GB ← ⚠️ 它在容器外,本行限不住
# pac-web + pac-docs(实测 49+64 MB) ~0.2 GB
# ────────────────────────────────────────────────
# 可给 pac-service ~10 GB → 取 8G,余下留给 OS/dockerd
#
# ✅ 2026-09-05 生产实测两整轮(550,481 召回池),RestartCount=0 全程没被杀:
# cgroup memory.peak = 4760 MiB ← **这才是 mem_limit 管的数**,58% of 8192,1.7 倍余量
# 进程 rss 峰 = 2936 MB heapUsed 峰 = 2544 MB
# ⛔ **量余量必须看 cgroup,不能看 `docker stats` 或进程 rss。**
# 三者差 1.6 倍:cgroup 还计入 page cache 与堆外分配,进程 rss 不含前者;
# `docker stats` 是采样,峰值会漏在采样之间(实测采样只看到 3.53GiB,真实 4.65GiB)。
# 正确量法:`docker exec pac-pac-service-1 cat /sys/fs/cgroup/memory.peak`
# (高水位自容器创建起累计,重建即清零 —— 要在一轮跑完之后、下次重建之前读)
# ⚠️ 4760 里有相当部分是**可回收的** page cache(实测跑完后 file 段只剩 2 MiB),
# 撑到上限时内核会先回收它再谈 OOM,所以真实的不可压缩需求低于这个数。
# ⚠️ 定这个数时手上只有 6.03G 那个**分批改造之前**的旧峰值。⛔ 以后调必须重新量。
#
# 【与下面 NODE_OPTIONS 的关系】RSS 恒大于 V8 堆(堆外还有 Buffer/Prisma engine/源码映射,
# 实测约 560MB)。两者若相等,撞顶时**一定是先被 cgroup SIGKILL(零日志零堆栈)**,
# 而不是 V8 抛 heap limit(有 JS 堆栈)。所以堆上限必须**低于** mem_limit ——
# 09-05 拿到实测峰值后已把它从 8192 下调到 6144,理由写在那一行旁边。
#
# ⚠️ **swap 语义**:docker 不显式设 memswap_limit 时,总量 = 2×mem_limit(8G RAM + 8G swap)。
# 生产**没有 swap** → 就是硬 8G,精确。测试机有 16G swapfile → 那边实际是 8G+8G。
# ⛔ 不设 memswap_limit 是刻意的:测试机靠 swap 撑着三十来个容器,一刀切断会造出
# **生产根本不存在的**失败模式,把验证信号污染掉。
#
# ⚠️ 被 SIGKILL 会在 sync_logs 留下 status='running' 的僵尸锁 —— 靠 09-04 那条
# 「下一轮开跑前周期回收」自愈(sync-incremental.scheduler.ts reapStaleRunningLocks),
# 回收时会推一条 warning 告警(09-05 加)。
# ⛔ 删那条回收 = 这里每杀一次就永久卡死增量。三者(限制/回收/告警)是配套的。
#
# 🔴 **本行引入的新风险,必读:`docker exec` 跑的 CLI 和服务共享这 8G。**
# 补摄 / recompute-plans 一直是 `docker exec pac-pac-service-1 node --max-old-space-size=8192 …`,
# **同一个容器 = 同一个 cgroup**。加限制前:服务 4G + CLI 4.9G = 8.9G,宿主 15G 装得下;
# 加限制后:两者之和一旦过 8G,内核会在**这个 cgroup 内部**挑一个杀 —— 有可能杀的是服务。
# ⇒ 规则没变(本来就写着"别手工并发跑全量 CLI"),但**违反的代价从'也许撑得住'变成'必然被杀'**。
# ✅ 正确做法:重活单开一个**带自己上限**的容器,别 exec 进服务里 ——
# docker run --rm --memory=6g --network=pac_default \
# --env-file apps/pac-service/.env \
# --entrypoint node pac-pac-service \
# --max-old-space-size=5120 dist/cli/<名字>.cli.js …
# CLI 撑爆只杀它自己,服务那 8G 一点不受影响。
# (2026-09-05 在测试机实跑验证过:新容器 cgroup memory.max=自己的值、连库正常、
# 同时 `docker stats` 里服务仍是 8GiB 分母,互不干扰。)
# ⚠️ DATABASE_URL 分形态:
# · 托管 DB(本生产机 friday)—— .env 里就是 RDS 地址,`--env-file` 足够;
# · 自带 PG(测试机)—— .env 里是 localhost,单开的容器连不上,须另加
# `-e DATABASE_URL=postgresql://$POSTGRES_USER:$POSTGRES_PASSWORD@postgres:5432/$POSTGRES_DB?schema=public`
# ⚠️ 真要 exec 进服务容器,先确认没在跑轮次(cron `15 0,8-22/2 * * *` 沪),并盯 `docker stats`。
mem_limit: 8g
depends_on: depends_on:
postgres: postgres:
condition: service_healthy condition: service_healthy
...@@ -123,10 +197,37 @@ services: ...@@ -123,10 +197,37 @@ services:
# ⛔ 但**别手工并发跑** `recompute-plans --host=<全量>`(它也吃 8G)—— # ⛔ 但**别手工并发跑** `recompute-plans --host=<全量>`(它也吃 8G)——
# 那条 CLI **不走 sync 锁**,与服务自己那轮全量撞上时才真有可能把宿主吃满。 # 那条 CLI **不走 sync 锁**,与服务自己那轮全量撞上时才真有可能把宿主吃满。
# #
# ⚠️ 这是**止血不是根治**:池子还在涨,8G 迟早也会到顶。 # ⚠️ 抬天花板只是**止血**。根治是「增量之后别跑全量」——
# 根治是「增量之后别跑全量」—— `recompute-plans` 已经有 `--clinics` 收窄能力, # `recompute-plans` 已经有 `--clinics` 收窄能力,同样的思路搬到
# 同样的思路搬到 SyncIncrementalSchedulerService 那一处即可。 # SyncIncrementalSchedulerService 那一处即可。(2026-09-05 的端到端分批
NODE_OPTIONS: --max-old-space-size=8192 # 把峰值压掉一半多,但"每轮都对全部患者跑一遍召回"这个行为本身没变。)
#
# 🔵 2026-09-05 由 8192 下调到 6144 —— **不是为了省内存,是为了让失败可诊断**。
#
# 加了 mem_limit(8g)之后,堆上限还留在 8192 会形成一个坏顺序:RSS 恒大于堆
# (实测 rss=2873MB 时 heapTotal=2309MB,堆外常驻约 560MB),
# 两者相等时**一定是先撞 cgroup 被 SIGKILL(exit 137,零日志零堆栈)**,
# 而不是 V8 抛 `Reached heap limit`(有 JS 堆栈,能直接定位是哪段吃的)。
# 等于把唯一一次能拿到现场的机会浪费掉 —— 而 09-04 整件事就是败在没有现场。
#
# 6144 的依据(2026-09-05 生产连续两整轮实测,550,481 池):
# 峰值 heapUsed=2544MB,之后单调降到 199MB;进程 rss 峰=2936MB。6144 留 2.4 倍余量。
#
# ⚠️ **这个"V8 先抛"不是硬保证,是大概率。** cgroup 高水位 4760MiB 比堆峰值 2544MB
# 多出约 2.2G(堆外分配 + page cache)。若堆真涨到 6144,总量可能逼近 8192,
# 届时到底是 V8 先抛还是 cgroup 先杀,取决于当时有多少 page cache 可回收。
# ⇒ 想要**硬保证**就把这里降到 5120(5120+2200 < 8192);代价是相对当前峰值只剩 2.0 倍。
# 现在取 6144 是权衡:常见的堆失控场景下 V8 会先抛(page cache 会被先回收掉),
# 而 2.4 倍余量不至于把本来能跑完的轮次变成必崩。
#
# ⚠️ 不影响 CLI:补摄/recompute-plans 走 `node --max-old-space-size=8192 …`,
# 命令行 V8 flag 覆盖 NODE_OPTIONS(Node 把 NODE_OPTIONS 当作排在命令行**之前**),
# 重复 flag 取最后一个。⛔ 但别忘了它们和服务同 cgroup,见上面 mem_limit 那段。
#
# ⚠️ 什么时候该往回调:池子涨到让 heapUsed 峰值逼近 4G(即 6144 的 2/3)时。
# 峰值主要由 selectHits→hitsByPatient 那份双拷贝决定,它随**命中数**涨,
# 不随分批变小 —— 那才是下一个要动的地方,不是这个数。
NODE_OPTIONS: --max-old-space-size=6144
# 覆盖 .env 的 localhost URL → 走 docker 内部网络 # 覆盖 .env 的 localhost URL → 走 docker 内部网络
DATABASE_URL: postgresql://${POSTGRES_USER:-pac}:${POSTGRES_PASSWORD:-pac}@postgres:5432/${POSTGRES_DB:-pac}?schema=public DATABASE_URL: postgresql://${POSTGRES_USER:-pac}:${POSTGRES_PASSWORD:-pac}@postgres:5432/${POSTGRES_DB:-pac}?schema=public
REDIS_URL: redis://redis:6379 REDIS_URL: redis://redis:6379
......
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