fix(sync): source_unit 解析器不再挂单例实例字段 —— 根治跨宿主串味建出的重复主档

ColdImportService 是 Nest 单例,却把 source_unit 解析器存成实例字段,每次摄入开头
覆写、在长 async 循环里读。原注释「cold-import 按 host 串行,实例字段安全」在 push
(webhook,按请求并发)接入后失效:

  friday  identity_namespace_field = tenant_id
  jvs-dw  identity_namespace_field = brand

两者并发摄入互相覆写 resolver → 对方的行解析不出命名空间列 → source_unit=''
→ 患者索引 (source_unit, external_id) 未命中 → 建出空命名空间的重复患者主档。

测试服务器实测(2026-07-29 查证):
  - friday 首推 07-23 16:24:26,jvs-dw 首个空品牌患者 16:24:31(5 秒内);
    此前 jvs-dw 自 06-28 起 39 万患者零个空品牌。
  - jvs-dw 空品牌患者 51063 条,其中 51037(99.95%)与真主档同 external_id;
    friday 空品牌 498 条,496 条落在 jvs-dw pull 窗口内,354 条是重复档。

改法照抄 tenantResolver 一贯的纪律:局部 const 构建 + 逐层传参。涉及 4 个入口
(reparse / ingestRawTables / importDirectory / importPatient)与 5 个读取方法
(processPatients / processPatientRelations / processPatientReturnVisits /
processSubject / hydratePushLookupTables)。纯管道改造,无行为变更。

并加源码闸测试 ingest-resolver-no-instance-state.spec.ts:这类 bug 单跑任何一条
路径都正确,只有并发交错才炸,单测抓不到,只能在源码层禁止 `this.*Resolver =`。
(已验证该闸对修复前的代码报 4 处赋值 / 9 处读取。)

注:线上已污染的 5.1 万条空命名空间主档需另行归并,不在本次范围。

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

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

@@ -83,9 +83,14 @@ function patientIndexKey(sourceUnit: string, externalId: string): string {
 @Injectable()
 export class ColdImportService {
   private readonly logger = new Logger(ColdImportService.name);
-  /// 当次 import 的 source_unit 解析器(从 manifest.identity_namespace_field 构建)。
-  /// cold-import 按 host 串行,实例字段安全;未声明命名空间字段时恒返回 ''。
-  private sourceUnitResolver: SourceUnitResolver = { resolve: () => '' };
+  // ⚠️ source_unit 解析器**只能是局部值,逐层传参**,绝不能挂实例字段。
+  //    本服务是 Nest 单例,而 push(webhook,按请求并发)与 cold-import / reparse / pull
+  //    共用它。曾把 resolver 存成实例字段(「cold-import 按 host 串行,实例字段安全」),
+  //    push 接入后该前提失效:friday(命名空间列 tenant_id)与 jvs-dw(brand)并发时互相
+  //    覆写,对方的行解析不出命名空间 → source_unit='' → 患者索引未命中 → 建出空命名空间
+  //    的重复主档。2026-07-23 16:24 friday 首推的 5 秒内 jvs-dw 就出现首个空品牌患者,
+  //    到 07-29 累计 5.1 万条(99.95% 与真主档同 external_id)。同理 tenantResolver 一直
+  //    是局部 const —— 两者必须保持同一纪律。
 
   constructor(
     private readonly prisma: PrismaService,
@@ -130,7 +135,7 @@ export class ColdImportService {
     const host = await this.prisma.host.findFirst({ where: { name: opts.hostName } });
     if (!host) throw new Error(`reparse: host '${opts.hostName}' not found`);
     const tenantResolver = buildTenantResolver(manifest);
-    this.sourceUnitResolver = buildSourceUnitResolver(manifest);
+    const sourceUnitResolver = buildSourceUnitResolver(manifest);
     const normalize = { amountUnit: manifest.amount_unit, timezone: manifest.timezone };
     const subjectCfgs = this.loadAllAssemblers(absDir, manifest).filter((c) => !!c.emits);
     const targetCfgs = opts.subjectTypes?.length
@@ -223,7 +228,7 @@ export class ColdImportService {
       }
       const transformed = transforms.length > 0 ? this.transformEngine.run({ tables, transforms }) : tables;
       for (const cfg of reparseableCfgs) {
-        const stats = await this.processSubject(transformed, cfg, host.id, tenantResolver, seenTenants, normalize, undefined, false);
+        const stats = await this.processSubject(transformed, cfg, host.id, tenantResolver, sourceUnitResolver, seenTenants, normalize, undefined, false);
         const agg = aggByResource.get(cfg.canonical);
         if (!agg) {
           aggByResource.set(cfg.canonical, stats);
@@ -315,7 +320,7 @@ export class ColdImportService {
     const host = await this.prisma.host.findFirst({ where: { name: opts.hostName } });
     if (!host) throw new Error(`ingestRawTables: host '${opts.hostName}' not found`);
     const tenantResolver = buildTenantResolver(manifest);
-    this.sourceUnitResolver = buildSourceUnitResolver(manifest);
+    const sourceUnitResolver = buildSourceUnitResolver(manifest);
     const normalize = { amountUnit: manifest.amount_unit, timezone: manifest.timezone };
     const transforms = manifest.transforms ?? [];
 
@@ -366,7 +371,7 @@ export class ColdImportService {
       const inp = (tf as { input?: string }).input;
       if (inp && !tables[inp]) tables[inp] = [];
     }
-    await this.hydratePushLookupTables(tables, transforms, host.id);
+    await this.hydratePushLookupTables(tables, transforms, host.id, sourceUnitResolver);
     const transformed =
       transforms.length > 0 ? this.transformEngine.run({ tables, transforms }) : tables;
 
@@ -381,19 +386,19 @@ export class ColdImportService {
         if (cfg.canonical === 'patient') {
           perResource.push(
             await this.processPatients(
-              transformed, cfg, host.id, tenantResolver, seenTenants, normalize, !!opts.dryRun, true,
+              transformed, cfg, host.id, tenantResolver, sourceUnitResolver, seenTenants, normalize, !!opts.dryRun, true,
             ),
           );
         } else if (cfg.canonical === 'patient_relation') {
           perResource.push(
             await this.processPatientRelations(
-              transformed, cfg, host.id, tenantResolver, normalize, !!opts.dryRun,
+              transformed, cfg, host.id, tenantResolver, sourceUnitResolver, normalize, !!opts.dryRun,
             ),
           );
         } else if (cfg.canonical === 'patient_return_visit') {
           perResource.push(
             await this.processPatientReturnVisits(
-              transformed, cfg, host.id, tenantResolver, normalize, !!opts.dryRun,
+              transformed, cfg, host.id, tenantResolver, sourceUnitResolver, normalize, !!opts.dryRun,
             ),
           );
         }
@@ -404,6 +409,7 @@ export class ColdImportService {
           cfg,
           host.id,
           tenantResolver,
+          sourceUnitResolver,
           seenTenants,
           normalize,
           syncLog.id,
@@ -595,6 +601,7 @@ export class ColdImportService {
     tables: Record<string, Record<string, unknown>[]>,
     transforms: readonly TransformOp[],
     hostId: string,
+    sourceUnitResolver: SourceUnitResolver,
   ): Promise<void> {
     for (const tf of transforms) {
       if (tf.kind !== 'lookup' || !tf.push_fallback) continue;
@@ -605,7 +612,7 @@ export class ColdImportService {
       const { rows, resolved, ambiguous, absent } = await buildPushLookupFallbackRows(
         tf,
         input,
-        (row) => this.sourceUnitResolver.resolve(row),
+        (row) => sourceUnitResolver.resolve(row),
         async ({ sourceUnits, externalIds }) =>
           (await this.prisma.patient.findMany({
             where: {
@@ -766,7 +773,7 @@ export class ColdImportService {
     }
     // v2.3:per-row tenant resolver(支持单 host 多 tenant)
     const tenantResolver = buildTenantResolver(manifest);
-    this.sourceUnitResolver = buildSourceUnitResolver(manifest);
+    const sourceUnitResolver = buildSourceUnitResolver(manifest);
     const knownTenants = tenantResolver.knownTenants();
     const seenTenants = new Set<string>(); // 跑过程动态收集(pass-through 模式不预知)
     this.logger.log(
@@ -955,6 +962,7 @@ export class ColdImportService {
               manifest,
               host,
               tenantResolver,
+              sourceUnitResolver,
               seenTenants,
               normalize,
               patientCfg,
@@ -1032,6 +1040,7 @@ export class ColdImportService {
           manifest,
           host,
           tenantResolver,
+          sourceUnitResolver,
           seenTenants,
           normalize,
           patientCfg,
@@ -1231,7 +1240,7 @@ export class ColdImportService {
     }
 
     const tenantResolver = buildTenantResolver(manifest);
-    this.sourceUnitResolver = buildSourceUnitResolver(manifest);
+    const sourceUnitResolver = buildSourceUnitResolver(manifest);
     const seenTenants = new Set<string>();
 
     // 3. 单患者 SQL — 改写所有 queries 加 WHERE patient_id+brand
@@ -1285,7 +1294,7 @@ export class ColdImportService {
 
     if (patientCfg) {
       const stats = await this.processPatients(
-        tables, patientCfg, host.id, tenantResolver, seenTenants, normalize, false,
+        tables, patientCfg, host.id, tenantResolver, sourceUnitResolver, seenTenants, normalize, false,
       );
       perResource.push(stats);
       totals.patientsUpserted += stats.patientsUpserted;
@@ -1295,7 +1304,7 @@ export class ColdImportService {
 
     if (patientRelationCfg) {
       const stats = await this.processPatientRelations(
-        tables, patientRelationCfg, host.id, tenantResolver, normalize, false,
+        tables, patientRelationCfg, host.id, tenantResolver, sourceUnitResolver, normalize, false,
       );
       perResource.push(stats);
       totals.failed += stats.failed;
@@ -1303,7 +1312,7 @@ export class ColdImportService {
 
     if (patientReturnVisitCfg) {
       const stats = await this.processPatientReturnVisits(
-        tables, patientReturnVisitCfg, host.id, tenantResolver, normalize, false,
+        tables, patientReturnVisitCfg, host.id, tenantResolver, sourceUnitResolver, normalize, false,
       );
       perResource.push(stats);
       totals.failed += stats.failed;
@@ -1312,7 +1321,7 @@ export class ColdImportService {
     for (const cfg of subjectCfgs) {
       try {
         const stats = await this.processSubject(
-          tables, cfg, host.id, tenantResolver, seenTenants, normalize, syncLog.id, false,
+          tables, cfg, host.id, tenantResolver, sourceUnitResolver, seenTenants, normalize, syncLog.id, false,
         );
         perResource.push(stats);
         totals.transactionsWritten += stats.transactionsWritten;
@@ -1385,6 +1394,8 @@ export class ColdImportService {
     manifest: ColdImportManifest;
     host: { id: string; name: string };
     tenantResolver: TenantResolver;
+    /// 与 tenantResolver 同纪律:由调用方按本次 manifest 构建后传入,不走实例字段(见类头注释)
+    sourceUnitResolver: SourceUnitResolver;
     seenTenants: Set<string>;
     normalize: { amountUnit: 'fen' | 'yuan'; timezone: string };
     patientCfg: AssemblerConfig | undefined;
@@ -1437,6 +1448,7 @@ export class ColdImportService {
         args.patientCfg,
         args.host.id,
         args.tenantResolver,
+        args.sourceUnitResolver,
         args.seenTenants,
         args.normalize,
         args.dryRun,
@@ -1458,6 +1470,7 @@ export class ColdImportService {
         args.patientRelationCfg,
         args.host.id,
         args.tenantResolver,
+        args.sourceUnitResolver,
         args.normalize,
         args.dryRun,
       );
@@ -1475,6 +1488,7 @@ export class ColdImportService {
         args.patientReturnVisitCfg,
         args.host.id,
         args.tenantResolver,
+        args.sourceUnitResolver,
         args.normalize,
         args.dryRun,
       );
@@ -1493,6 +1507,7 @@ export class ColdImportService {
           cfg,
           args.host.id,
           args.tenantResolver,
+          args.sourceUnitResolver,
           args.seenTenants,
           args.normalize,
           args.syncLogId,
@@ -1537,6 +1552,7 @@ export class ColdImportService {
     config: AssemblerConfig,
     hostId: string,
     tenantResolver: TenantResolver,
+    sourceUnitResolver: SourceUnitResolver,
     seenTenants: Set<string>,
     normalize: { amountUnit: 'fen' | 'yuan'; timezone: string },
     dryRun: boolean,
@@ -1558,7 +1574,7 @@ export class ColdImportService {
       // v2.3:per-row tenant 解析(从 host 真原文 rawSource 读 brand 字段)
       const tenantId = tenantResolver.resolve(rawSource);
       // 集团模型:source_unit(发证命名空间)从同一 rawSource 解析,给 external_id 消歧
-      const sourceUnit = this.sourceUnitResolver.resolve(rawSource);
+      const sourceUnit = sourceUnitResolver.resolve(rawSource);
       if (!tenantId) {
         stats.failed++;
         this.logger.warn(
@@ -1623,6 +1639,7 @@ export class ColdImportService {
     config: AssemblerConfig,
     hostId: string,
     tenantResolver: TenantResolver,
+    sourceUnitResolver: SourceUnitResolver,
     normalize: { amountUnit: 'fen' | 'yuan'; timezone: string },
     dryRun: boolean,
   ): Promise<PerResourceStats> {
@@ -1659,7 +1676,7 @@ export class ColdImportService {
         stats.failed++;
         continue;
       }
-      const sourceUnit = this.sourceUnitResolver.resolve(rawSource);
+      const sourceUnit = sourceUnitResolver.resolve(rawSource);
       const idx = await getIndex(tenantId);
       const patientId = idx.get(patientIndexKey(sourceUnit, patientExternalId));
       if (!patientId) continue; // 本人不在 PAC(非 active client)→ 无处挂靠,跳过
@@ -1708,6 +1725,7 @@ export class ColdImportService {
     config: AssemblerConfig,
     hostId: string,
     tenantResolver: TenantResolver,
+    sourceUnitResolver: SourceUnitResolver,
     normalize: { amountUnit: 'fen' | 'yuan'; timezone: string },
     dryRun: boolean,
   ): Promise<PerResourceStats> {
@@ -1742,7 +1760,7 @@ export class ColdImportService {
         stats.failed++;
         continue;
       }
-      const sourceUnit = this.sourceUnitResolver.resolve(rawSource);
+      const sourceUnit = sourceUnitResolver.resolve(rawSource);
       const idx = await getIndex(tenantId);
       const patientId = idx.get(patientIndexKey(sourceUnit, patientExternalId));
       if (!patientId) continue; // 本人不在 PAC → skip
@@ -1790,6 +1808,7 @@ export class ColdImportService {
     config: AssemblerConfig,
     hostId: string,
     tenantResolver: TenantResolver,
+    sourceUnitResolver: SourceUnitResolver,
     seenTenants: Set<string>,
     normalize: { amountUnit: 'fen' | 'yuan'; timezone: string },
     syncLogId: string | undefined,
@@ -1859,7 +1878,7 @@ export class ColdImportService {
           patientIndex = await this.buildPatientIndex(hostId, tenantId);
           patientIndexByTenant.set(tenantId, patientIndex);
         }
-        const sourceUnit = this.sourceUnitResolver.resolve(rawSource);
+        const sourceUnit = sourceUnitResolver.resolve(rawSource);
         // stub auto-create(索引/键带 source_unit,集团内 external_id 撞号消歧)
         const patientExternalId = canonicalRow.patientExternalId as string | undefined;
         if (patientExternalId && !patientIndex.has(patientIndexKey(sourceUnit, patientExternalId))) {

apps/pac-service/tests/ingest-resolver-no-instance-state.spec.ts

+import { readFileSync } from 'node:fs';
+import { join } from 'node:path';
+
+/**
+ * 摄入侧的 per-request 解析器**不许挂单例实例字段**。
+ *
+ * 踩过的坑(2026-07-23 → 07-29,测试服务器):
+ *   ColdImportService 是 Nest 单例,却把 source_unit 解析器存成实例字段
+ *   `private sourceUnitResolver`,每次摄入开头覆写、然后在长 async 循环里读。
+ *   当时的注释写着「cold-import 按 host 串行,实例字段安全」—— 这个前提在 push
+ *   (webhook,按请求并发)接入后就失效了:
+ *
+ *     friday   identity_namespace_field = tenant_id
+ *     jvs-dw   identity_namespace_field = brand
+ *
+ *   两个 host 并发摄入时互相覆写 resolver,对方的行解析不出命名空间列 → source_unit=''
+ *   → 患者索引 `(source_unit, external_id)` 未命中 → 建出空命名空间的**重复主档**。
+ *   friday 首次 push 是 07-23 16:24:26,jvs-dw 第一个空品牌患者出现在 16:24:31(5 秒内),
+ *   此前 39 万条 jvs-dw 患者零个空品牌。到 07-29 累计 5.1 万条,其中 99.95% 与真主档
+ *   同 external_id —— 纯污染。
+ *
+ * 单测抓不到这类 bug(单跑任何一条路径都正确,只有并发交错才炸),所以在源码层设闸:
+ * 解析器一律**局部构建 + 逐层传参**(tenantResolver 从一开始就是这么做的,照抄它)。
+ */
+describe('摄入解析器不得存为单例实例状态', () => {
+  const file = join(__dirname, '../src/modules/sync/cold-import/cold-import.service.ts');
+  const src = readFileSync(file, 'utf8');
+  // 注释行不算(类头就写着这段事故说明,里面必然出现这些词)
+  const code = src
+    .split('\n')
+    .filter((l) => !/^\s*(\/\/|\/\*|\*|\/\/\/)/.test(l))
+    .join('\n');
+
+  test('⭐ 不存在 `this.<x>Resolver = ...` 赋值(per-request 状态写进单例)', () => {
+    const hits = code.match(/this\.\w*[Rr]esolver\s*=[^=]/g) ?? [];
+    expect(hits).toEqual([]);
+  });
+
+  test('⭐ 不存在 `this.sourceUnitResolver` 读取(必须走入参)', () => {
+    const hits = code.match(/this\.sourceUnitResolver/g) ?? [];
+    expect(hits).toEqual([]);
+  });
+
+  test('每个读 source_unit 的 process* 方法都把 resolver 声明为入参', () => {
+    const readers = [
+      'processPatients',
+      'processPatientRelations',
+      'processPatientReturnVisits',
+      'processSubject',
+      'hydratePushLookupTables',
+    ];
+    for (const m of readers) {
+      const sig = code.slice(code.indexOf(`private async ${m}(`));
+      const params = sig.slice(0, sig.indexOf('): Promise<'));
+      expect(`${m}: ${params.includes('sourceUnitResolver')}`).toBe(`${m}: true`);
+    }
+  });
+});