fix(sync): 关系边/回访的本人缺席改建空壳 —— 兑现「推送顺序无要求」

同一情况三条路给了三个答案:
  processSubject              → ensurePatientStub(),照常落库
  processPatientRelations     → `if (!patientId) continue`,静默丢
  processPatientReturnVisits  → `if (!patientId) continue`,静默丢

那句 continue 的注释写着「非 active client,无处挂靠」,但 cold-import 的 cohort 路径
经 injectPatientFilter 对每张表都按患者 id 过滤、patients 又先跑,它几乎永不触发;
真正踩中的是 push —— 宿主先推 customer_referee_circle(07-27)、后推
customer_basic_info(07-28~29),10169 条关系边只落 257 条(2.5%),
且不计 failed、不进回执,宿主看到的是 accepted=N / failed=0,完全无从察觉。

契约文档承诺「推送顺序无要求」,兑现它靠的就是空壳兜底(pull 侧的等价物是
ClickHouseSourceService 的「反向拉主档」)。空壳只建**本人**这一侧;关系的对方
(relatedPatientId 可空)不建,靠 upsert 的 `update: { relatedPatientId }`
在对方入库后重推时回填 —— 那段回填代码本来就在。

加源码闸 ingest-patient-stub-consistency.spec.ts 锁住「三条路对齐」,
将来加第四个 process* 方法不至于再漏。已验证该闸对修复前的代码报
processPatientRelations / processPatientReturnVisits 各 1 处静默丢弃、
且两者都缺 ensurePatientStub。

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

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

@@ -1678,8 +1678,17 @@ export class ColdImportService {
       }
       const sourceUnit = sourceUnitResolver.resolve(rawSource);
       const idx = await getIndex(tenantId);
-      const patientId = idx.get(patientIndexKey(sourceUnit, patientExternalId));
-      if (!patientId) continue; // 本人不在 PAC(非 active client)→ 无处挂靠,跳过
+      // 本人不在索引 → 跟 processSubject 同处理:建空壳主档,边照常落。
+      //   2026-07-29 前这里是 `if (!patientId) continue`(注释说「非 active client,无处挂靠」),
+      //   但 cohort 路径每张表都按患者 id 过滤、patients 又先跑,那句几乎永不触发;
+      //   真正踩中的是 push —— 宿主先推 customer_referee_circle、后推 customer_basic_info,
+      //   1 万条边只落 257 条(2.5%),且不计 failed 不进回执,宿主完全看不出来。
+      //   文档承诺「推送顺序无要求」,靠的就是空壳兜底(pull 侧的等价物是反向拉主档)。
+      let patientId = idx.get(patientIndexKey(sourceUnit, patientExternalId));
+      if (!patientId) {
+        patientId = await this.ensurePatientStub(hostId, tenantId, sourceUnit, patientExternalId);
+        idx.set(patientIndexKey(sourceUnit, patientExternalId), patientId);
+      }
       const relatedPatientId = idx.get(patientIndexKey(sourceUnit, relatedExternalId)) ?? null;
 
       try {
@@ -1762,8 +1771,12 @@ export class ColdImportService {
       }
       const sourceUnit = sourceUnitResolver.resolve(rawSource);
       const idx = await getIndex(tenantId);
-      const patientId = idx.get(patientIndexKey(sourceUnit, patientExternalId));
-      if (!patientId) continue; // 本人不在 PAC → skip
+      // 同 processPatientRelations:本人不在索引就建空壳,不再静默丢(见那边注释)
+      let patientId = idx.get(patientIndexKey(sourceUnit, patientExternalId));
+      if (!patientId) {
+        patientId = await this.ensurePatientStub(hostId, tenantId, sourceUnit, patientExternalId);
+        idx.set(patientIndexKey(sourceUnit, patientExternalId), patientId);
+      }
 
       const data = {
         clinicId: (c.clinicId as string | undefined) ?? null,

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([]);
+  });
+});