merge: fix/source-unit-resolver-race → test(跨宿主串味根治 + 关系边不再静默丢)

两个 P0(2026-07-29 查验 FRIDAY 测试环境推送时定位):
  1. source_unit 解析器挂在 Nest 单例的实例字段上,push(并发)与 pull 互相覆写
     → 建出空命名空间的重复患者主档(jvs-dw 51063 条、friday 498 条)。
  2. 关系边/回访在本人未入库时静默丢弃,不计 failed 不进回执
     → 宿主先推 customer_referee_circle 时 10169 条边只落 257 条(2.5%)。

部署后需让 FRIDAY 重推 customer_referee_circle 补齐关系边;
线上已有的空命名空间重复主档另行归并,不在本次范围。

apps/pac-service/tests/ingest-patient-stub-consistency.spec.ts

+import { readFileSync } from 'node:fs';
+import { join } from 'node:path';
+
+/**
+ * 「本人不在 PAC 索引」时,所有摄入路径必须给同一个答案:**建空壳主档,记录照落**。
+ *
+ * 踩过的坑(2026-07-27 → 29,测试服务器):
+ *   同一情况三条路给了三个答案 ——
+ *     processSubject               → ensurePatientStub(),照常落库   ✅
+ *     processPatientRelations      → `if (!patientId) continue`      ❌ 静默丢
+ *     processPatientReturnVisits   → `if (!patientId) continue`      ❌ 静默丢
+ *
+ *   那句 continue 的注释写着「非 active client,无处挂靠」,但 cold-import 的 cohort 路径
+ *   每张表都按患者 id 过滤、patients 又先跑,它几乎永不触发 —— 真正踩中的是 push:
+ *   宿主先推 customer_referee_circle、后推 customer_basic_info,10169 条关系边只落 257 条
+ *   (2.5%),而且不计 failed、不进回执,宿主看到的是 `accepted=N, failed=0`,完全无从察觉。
+ *
+ *   契约文档承诺「推送顺序无要求」,兑现它靠的就是空壳兜底(pull 侧的等价物是
+ *   ClickHouseSourceService 的「反向拉主档」:fact 表出现的 patient_id 回头补拉主档)。
+ *
+ * 空壳只在**本人**这一侧建;关系的**对方**(relatedPatientId 可空)不建空壳,
+ * 靠 upsert 的 `update: { relatedPatientId }` 在对方入库后重推时回填。
+ */
+describe('摄入路径:本人缺失一律建空壳,不得静默丢', () => {
+  const file = join(__dirname, '../src/modules/sync/cold-import/cold-import.service.ts');
+  const src = readFileSync(file, 'utf8');
+
+  /**
+   * 取某个 private async 方法的方法体(到下一个同级方法声明为止),**剥掉注释行** ——
+   * 事故说明就写在这些方法头上,不剥的话会扫到注释里引用的旧代码原文。
+   */
+  const bodyOf = (name: string): string => {
+    const start = src.indexOf(`private async ${name}(`);
+    expect(`${name} 存在: ${start >= 0}`).toBe(`${name} 存在: true`);
+    const rest = src.slice(start + 1);
+    // 下一个**同级**成员声明:恰好两空格缩进(方法体一律 ≥4),可带任意多个修饰词
+    const next = rest.search(
+      /\n {2}(?:(?:private|protected|public|static|readonly|async|get|set)\s+)*[a-zA-Z_$]+\s*[(<]/,
+    );
+    expect(`${name} 找到方法体边界: ${next >= 0}`).toBe(`${name} 找到方法体边界: true`);
+    return (next >= 0 ? rest.slice(0, next) : rest)
+      .split('\n')
+      .filter((l) => !/^\s*(\/\/|\/\*|\*)/.test(l))
+      .join('\n');
+  };
+
+  const paths = ['processSubject', 'processPatientRelations', 'processPatientReturnVisits'];
+
+  test.each(paths)('⭐ %s 用 ensurePatientStub 兜住缺席的本人', (name) => {
+    expect(bodyOf(name)).toContain('ensurePatientStub(');
+  });
+
+  test.each(paths)('⭐ %s 不含 `if (!patientId) continue` 式的静默丢弃', (name) => {
+    const hits = bodyOf(name).match(/if\s*\(\s*!\s*patientId\s*\)\s*continue/g) ?? [];
+    expect(hits).toEqual([]);
+  });
+});

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