merge: main → test(反向合)—— reparse bind 夹子

apps/pac-service/src/modules/sync/cold-import/cold-import.service.ts

@@ -46,6 +46,36 @@ import type { FactReject } from '../pipeline/fact-writer.service';
  * 给几条足够定位;全量在 PAC 服务端日志里(每条都有 [schema-violation] 行)。
  */
 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 {
   buildPushLookupFallbackRows,
@@ -185,10 +215,10 @@ export class ColdImportService {
 
     // 2a. dryRun:报 scope(患者数 + 各资源 txn 数),不写、不分批装配(便宜)。
     if (opts.dryRun) {
-      // ⛔ 同样分块 —— 早先这里也是整份清单塞 in,>3.2 万患者时 dry-run 直接崩,
-      //   而 --patients-file 的文档恰恰说「按受影响患者收窄是最有效的提速手段(可达 250 倍)」:
-      //   最需要先 dry-run 探一探的大清单场景,正好是它唯一不工作的场景。
-      const DRY_CHUNK = 3000;
+      // ⛔ 同样分块 —— 早先这里也是整份清单塞 in,>3.2 万患者时 dry-run 直接崩。
+      //   与实跑共用 reparseBatchSize()(见其注释:两侧用同一个夹过上限的值,
+      //   否则改一侧另一侧的墙还在)。
+      const DRY_CHUNK = reparseBatchSize();
       for (const cfg of reparseableCfgs) {
         let n = 0;
         const chunks: Array<string[] | null> = opts.patientIds?.length
@@ -209,7 +239,7 @@ export class ColdImportService {
 
     // 2b. 分批实跑:每批患者 → 重建源表(只这批 distinct rawPayload)→ transform → processSubject
     //     (transaction 幂等命中已存 → parser 重衍生 fact,版本流 supersede)。
-    const BATCH = Math.max(1, Number(process.env.PAC_REPARSE_BATCH) || 3000);
+    const BATCH = reparseBatchSize();
     const runStart = new Date();
     const aggByResource = new Map<string, PerResourceStats>();
     const seenTenants = new Set<string>();

apps/pac-service/tests/fixtures/reparse-bind-limit/assemblers/diagnosis.yaml

+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

apps/pac-service/tests/fixtures/reparse-bind-limit/manifest.yaml

+# 最小 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

apps/pac-service/tests/reparse-dry-run-bind-limit.spec.ts

+/**
+ * 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;
+    }
+  });
+});