fix(sync): 定向重摄不推进增量游标 + 大名单走文件(污染修复前置)

背景:DW 病历表 patient_id 错位污染(高逸铭案例),需按名单重摄 3 万患者。
沿用现有 PAC_COHORT_ONLY_PATIENT(cohort 收窄)时发现两个阻塞问题:

1) ⭐ 静默丢数据:只要 manifest 配了 incremental.per_query(jvs-dw 配了),
   cold-import 任何模式都写 cursor_after=run_start。定向轮只覆盖名单内患者,
   却推进全局水位 → 名单外患者在本轮期间的 DW 新写入被永久埋在游标下方。
   (与本次排查发现的"12,595 个患者符合 cohort 却从未摄入"是同类机制)
   修:定向模式下 cursor 保持上次水位不推进,daily 增量照常从原点接力。

2) 大名单 E2BIG:3 万个 id 的逗号串约 275KB > 环境变量单值 128KB(MAX_ARG_STRLEN)。
   修:支持 PAC_COHORT_ONLY_PATIENT=@/path/to/file(每行一个 id);逗号列表照旧兼容。

顺带:定向生效时打印命中患者数,便于核对名单规模。
测试 7 例:空/空白不误判定向、逗号列表、@file(含 3 万行)、文件缺失须抛错不静默退化。

生产验证(高逸铭组 7 人小批):删除→重摄后
  高逸铭 62 事实/6 诊所 → 21/1(上海时代),别人的数据清光
  刘晓旭 13 → 25、徐嘉悦 126 → 130,被吞的数据补回,诊所归属全部正确

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

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

 import { Injectable, Logger } from '@nestjs/common';
+import * as fs from 'node:fs';
 import { createClient, type ClickHouseClient } from '@clickhouse/client';
 import type { ClickHouseSource } from './manifest.schema';
 
 /**
+ * 解析 PAC_COHORT_ONLY_PATIENT —— 定向重摄的患者 id 集合(cohort 收窄用)。
+ *   `id1,id2,...`      逗号列表(小名单)
+ *   `@/path/to/file`   从文件读,每行一个 id(大名单;环境变量单值有 128KB 上限,
+ *                      上万个 id 的逗号串会 E2BIG,必须走文件)
+ * 未设置 / 空 → 返回 [](调用方据此判断"是否定向模式",影响 cohort WHERE 与 cursor 推进)。
+ */
+export function resolveOnlyPatientIds(): string[] {
+  const raw = process.env.PAC_COHORT_ONLY_PATIENT?.trim();
+  if (!raw) return [];
+  if (raw.startsWith('@')) {
+    const file = raw.slice(1);
+    if (!fs.existsSync(file)) {
+      throw new Error(`PAC_COHORT_ONLY_PATIENT 指向的文件不存在:${file}`);
+    }
+    return fs
+      .readFileSync(file, 'utf-8')
+      .split('\n')
+      .map((s) => s.trim())
+      .filter(Boolean);
+  }
+  return raw.split(',').map((s) => s.trim()).filter(Boolean);
+}
+
+/**
  * ClickHouseSourceService — 把 ClickHouse SQL 查询结果作为 cold-import 数据源。
  *
  * 形态:执行 manifest.sql_source.queries 里每个 SQL,把结果以
@@ -386,11 +411,13 @@ export class ClickHouseSourceService {
       }
       // dev/ops:只摄入指定患者(PAC_COHORT_ONLY_PATIENT=<patient_id> 或逗号列表 id1,id2,...)
       //   单患者复现 / 定向重摄受影响子集(分类修复后只重摄 reclassify 的患者)用。
-      const onlyPatient = process.env.PAC_COHORT_ONLY_PATIENT?.trim();
-      if (onlyPatient) {
-        const ids = onlyPatient.split(',').map((s) => s.trim()).filter(Boolean);
+      //   ⭐ 大名单(上万个 id)用 `@/path/to/file` 从文件读(每行一个 id):环境变量单值有
+      //   128KB(MAX_ARG_STRLEN)上限,3 万个 id 的逗号串约 275KB 会直接 E2BIG。
+      const ids = resolveOnlyPatientIds();
+      if (ids.length > 0) {
         const quoted = ids.map((id) => `'${id.replace(/'/g, "''")}'`).join(', ');
         whereParts.push(ids.length === 1 ? `${patient_key_column} = ${quoted}` : `${patient_key_column} IN (${quoted})`);
+        this.logger.log(`[clickhouse·cohort] PAC_COHORT_ONLY_PATIENT 定向:${ids.length} 个患者`);
       }
       // --clinics=X,Y:把 cohort 收窄到「在这些诊所看过」的患者。诊所过滤挂在有 org 的事实表
       //   (cohort.clinic_scope),用与 cohort 相同的患者键求交集。⚠️ 只收窄「列哪些患者」,

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

@@ -31,6 +31,7 @@ import {
 } from './manifest.schema';
 import {
   ClickHouseSourceService,
+  resolveOnlyPatientIds,
   type CohortKey,
   type IncrementalConfig,
   type PatientScope,
@@ -971,6 +972,18 @@ export class ColdImportService {
           this.logger.log(
             `Cursor 不推进(增量空跑,保留上次水位):${cursorAfterJson ?? '(首跑无 cursor)'}`,
           );
+        } else if (resolveOnlyPatientIds().length > 0) {
+          // ⭐ 定向重摄(PAC_COHORT_ONLY_PATIENT)**绝不推进游标**:本轮只覆盖名单内患者,
+          //   推进全局水位会把名单外患者在本轮期间的 DW 新写入永久埋在游标下方(静默漏拉)。
+          //   保留上次水位 → daily 增量照常从原点接力。
+          cursorAfterJson =
+            Object.keys(incrementalBaselineCursor).length > 0
+              ? JSON.stringify(incrementalBaselineCursor)
+              : null;
+          this.logger.log(
+            `Cursor 不推进(定向重摄 PAC_COHORT_ONLY_PATIENT,只覆盖名单内患者):` +
+              `${cursorAfterJson ?? '(首跑无 cursor)'}`,
+          );
         } else {
           const runStartIso = runStart.toISOString();
           const oldCursors: Record<string, string> = JSON.parse(cursorBeforeJson ?? '{}');

apps/pac-service/tests/cohort-only-patient.spec.ts

+import * as fs from 'node:fs';
+import * as os from 'node:os';
+import * as path from 'node:path';
+import { resolveOnlyPatientIds } from '../src/modules/sync/cold-import/clickhouse-source.service';
+
+/**
+ * PAC_COHORT_ONLY_PATIENT(定向重摄名单)解析。
+ *
+ * 背景:污染修复要按名单重摄 3 万个患者,逗号串约 275KB > 环境变量 128KB(MAX_ARG_STRLEN)
+ *   上限会 E2BIG,故支持 `@file`(每行一个 id)。
+ * ⚠️ 本函数还是"是否定向模式"的判定源 —— cold-import 据它决定**不推进增量游标**
+ *   (定向轮只覆盖名单内患者,推进全局水位会把名单外患者的 DW 新写入永久埋掉)。
+ */
+describe('PAC_COHORT_ONLY_PATIENT 解析', () => {
+  const ORIG = process.env.PAC_COHORT_ONLY_PATIENT;
+  let tmpFile = '';
+
+  afterEach(() => {
+    if (ORIG === undefined) delete process.env.PAC_COHORT_ONLY_PATIENT;
+    else process.env.PAC_COHORT_ONLY_PATIENT = ORIG;
+    if (tmpFile && fs.existsSync(tmpFile)) fs.unlinkSync(tmpFile);
+    tmpFile = '';
+  });
+
+  test('未设置 → 空数组(非定向模式,游标照常推进)', () => {
+    delete process.env.PAC_COHORT_ONLY_PATIENT;
+    expect(resolveOnlyPatientIds()).toEqual([]);
+  });
+
+  test('空串 / 纯空白 → 空数组(不误判成定向模式)', () => {
+    process.env.PAC_COHORT_ONLY_PATIENT = '   ';
+    expect(resolveOnlyPatientIds()).toEqual([]);
+  });
+
+  test('单个 id', () => {
+    process.env.PAC_COHORT_ONLY_PATIENT = '1855960';
+    expect(resolveOnlyPatientIds()).toEqual(['1855960']);
+  });
+
+  test('逗号列表 → 去空白、丢空项', () => {
+    process.env.PAC_COHORT_ONLY_PATIENT = '1855960, 1855959 ,,1855956,';
+    expect(resolveOnlyPatientIds()).toEqual(['1855960', '1855959', '1855956']);
+  });
+
+  test('@file → 逐行读(大名单绕开 128KB 环境变量上限)', () => {
+    tmpFile = path.join(os.tmpdir(), `pids-${Date.now()}.txt`);
+    fs.writeFileSync(tmpFile, '1855960\n1855959\n\n  1855956  \n');
+    process.env.PAC_COHORT_ONLY_PATIENT = `@${tmpFile}`;
+    expect(resolveOnlyPatientIds()).toEqual(['1855960', '1855959', '1855956']);
+  });
+
+  test('@file 三万行量级 → 全部读出(真实修复场景规模)', () => {
+    tmpFile = path.join(os.tmpdir(), `pids-big-${Date.now()}.txt`);
+    const ids = Array.from({ length: 30626 }, (_, i) => String(1000000 + i));
+    fs.writeFileSync(tmpFile, ids.join('\n'));
+    process.env.PAC_COHORT_ONLY_PATIENT = `@${tmpFile}`;
+    const got = resolveOnlyPatientIds();
+    expect(got).toHaveLength(30626);
+    expect(got[0]).toBe('1000000');
+    expect(got[30625]).toBe('1030625');
+  });
+
+  test('@file 文件不存在 → 抛错(不静默退化成全量重摄)', () => {
+    process.env.PAC_COHORT_ONLY_PATIENT = '@/no/such/pids.txt';
+    expect(() => resolveOnlyPatientIds()).toThrow(/不存在/);
+  });
+});