Merge bounded return visit and relation ingestion into test

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

@@ -1806,16 +1806,10 @@ export class ColdImportService {
     stats.failed = assembled.stats.failed;
     if (assembled.rows.length > 0)
       stats.sampleCanonical.push(assembled.rows[0]!.canonical);
-    // 本租户 externalId → PAC patient id 索引(本人 + 关系人解析共用)
-    const indexByTenant = new Map<string, Map<string, string>>();
-    const getIndex = async (tenantId: string) => {
-      let idx = indexByTenant.get(tenantId);
-      if (!idx) {
-        idx = await this.buildPatientIndex(hostId, tenantId);
-        indexByTenant.set(tenantId, idx);
-      }
-      return idx;
-    };
+    const getIndex = this.scopedPatientIndexLoader(
+      assembled.rows, hostId, tenantResolver, sourceUnitResolver,
+      ['patientExternalId', 'relatedExternalId'],
+    );
 
     for (const { canonical, rawSource } of assembled.rows) {
       const c = canonical as Record<string, unknown>;
@@ -1838,7 +1832,7 @@ export class ColdImportService {
         continue;
       }
       const sourceUnit = sourceUnitResolver.resolve(rawSource);
-      const idx = await getIndex(tenantId);
+      const idx = await getIndex(tenantId, sourceUnit);
       // 本人不在索引 → 跟 processSubject 同处理:建空壳主档,边照常落。
       //   2026-07-29 前这里是 `if (!patientId) continue`(注释说「非 active client,无处挂靠」),
       //   但 cohort 路径每张表都按患者 id 过滤、patients 又先跑,那句几乎永不触发;
@@ -1908,15 +1902,9 @@ export class ColdImportService {
     stats.failed = assembled.stats.failed;
     if (assembled.rows.length > 0)
       stats.sampleCanonical.push(assembled.rows[0]!.canonical);
-    const indexByTenant = new Map<string, Map<string, string>>();
-    const getIndex = async (tenantId: string) => {
-      let idx = indexByTenant.get(tenantId);
-      if (!idx) {
-        idx = await this.buildPatientIndex(hostId, tenantId);
-        indexByTenant.set(tenantId, idx);
-      }
-      return idx;
-    };
+    const getIndex = this.scopedPatientIndexLoader(
+      assembled.rows, hostId, tenantResolver, sourceUnitResolver, ['patientExternalId'],
+    );
 
     for (const { canonical, rawSource } of assembled.rows) {
       const c = canonical as Record<string, unknown>;
@@ -1938,7 +1926,7 @@ export class ColdImportService {
         continue;
       }
       const sourceUnit = sourceUnitResolver.resolve(rawSource);
-      const idx = await getIndex(tenantId);
+      const idx = await getIndex(tenantId, sourceUnit);
       // 同 processPatientRelations:本人不在索引就建空壳,不再静默丢(见那边注释)
       let patientId = idx.get(patientIndexKey(sourceUnit, patientExternalId));
       if (!patientId) {
@@ -2414,20 +2402,55 @@ export class ColdImportService {
     }
   }
 
+  /** 只加载本批引用的患者；关系人也按同一品牌隔离，dry-run 不触发查询。 */
+  private scopedPatientIndexLoader(
+    rows: Array<{ canonical: unknown; rawSource: Record<string, unknown> }>,
+    hostId: string,
+    tenantResolver: TenantResolver,
+    sourceUnitResolver: SourceUnitResolver,
+    fields: string[],
+  ) {
+    const refs = new Map<string, Set<string>>();
+    const indexes = new Map<string, Map<string, string>>();
+    const scopeKey = (tenant: string, unit: string) => JSON.stringify([tenant, unit]);
+    for (const row of rows) {
+      const tenant = tenantResolver.resolve(row.rawSource);
+      if (!tenant) continue;
+      const key = scopeKey(tenant, sourceUnitResolver.resolve(row.rawSource));
+      const ids = refs.get(key) ?? new Set<string>();
+      const canonical = row.canonical as Record<string, unknown>;
+      for (const field of fields) {
+        const id = String(canonical[field] ?? '').trim();
+        if (id) ids.add(id);
+      }
+      refs.set(key, ids);
+    }
+    return async (tenant: string, unit: string) => {
+      const key = scopeKey(tenant, unit);
+      let index = indexes.get(key);
+      if (!index) {
+        index = await this.buildPatientIndex(hostId, tenant, [...(refs.get(key) ?? [])], unit);
+        indexes.set(key, index);
+      }
+      return index;
+    };
+  }
+
   private async buildPatientIndex(
     hostId: string,
     tenantId: string,
-    externalIds?: string[],
-    sourceUnit?: string,
+    externalIds: string[],
+    sourceUnit: string,
   ): Promise<Map<string, string>> {
     const rows: Array<{ id: string; externalId: string; sourceUnit: string }> = [];
-    const chunks = externalIds === undefined ? [undefined] :
+    const chunks =
       Array.from({ length: Math.ceil(externalIds.length / 1000) }, (_, i) => externalIds.slice(i * 1000, (i + 1) * 1000));
     for (const refs of chunks) {
-      rows.push(...await this.withDbRetry(() => this.prisma.patient.findMany({
-        where: { hostId, tenantId, ...(sourceUnit !== undefined ? { sourceUnit } : {}), ...(refs ? { externalId: { in: refs } } : {}) },
+      const batch = await this.withDbRetry(() => this.prisma.patient.findMany({
+        where: { hostId, tenantId, sourceUnit, externalId: { in: refs } },
         select: { id: true, externalId: true, sourceUnit: true },
-      }), 'buildPatientIndex'));
+      }), 'buildPatientIndex');
+      for (const row of batch) rows.push(row);
     }
     // 集团内 external_id 可跨 source_unit 撞号 → 索引按 (source_unit, external_id) 建键
     return new Map(rows.map((p) => [patientIndexKey(p.sourceUnit, p.externalId), p.id]));

apps/pac-service/tests/push-patient-index.spec.ts

@@ -28,3 +28,51 @@ test('小批事件只读取引用患者，保留跨品牌同号隔离', async ()
   expect(resolutions).toEqual(['pac-brand-a', 'pac-brand-b']);
   expect(patient.upsert).not.toHaveBeenCalled();
 });
+
+describe.each(['processPatientReturnVisits', 'processPatientRelations'])('%s', (method) => {
+  function fixture() {
+    const rows = ['brand-a', 'brand-b'].map(unit => ({
+      canonical: { externalId: `visit-${unit}`, patientExternalId: 'same-id', relatedExternalId: 'relative', relationship: 'friend', taskDirectorId: '5992', taskDirectorName: '客服' },
+      rawSource: { unit },
+    }));
+    const patient = { findMany: jest.fn(async ({ where }: any) => where.externalId.in.map((id: string) => ({
+      id: `${where.sourceUnit}-${id}`, externalId: id, sourceUnit: where.sourceUnit,
+    }))), upsert: jest.fn() };
+    const prisma = { patient, patientReturnVisit: { upsert: jest.fn() }, patientRelation: { upsert: jest.fn() } };
+    const assembler = { assemble: () => ({ rows, stats: { fetched: 2, failed: 0, mappingMisses: [], suspectFields: [] }, rejections: [] }) };
+    const cold = new ColdImportService(prisma as never, {} as never, {} as never, assembler as never, {} as never, {} as never, {} as never);
+    const run = (dry = false) => (cold as any)[method]({}, { canonical: 'test' }, 'host', { resolve: () => 'tenant' }, { resolve: (r: any) => r.unit }, {}, dry);
+    return { run, prisma };
+  }
+  test('仅查询本批引用且品牌隔离，正确挂到患者', async () => {
+    const { run, prisma } = fixture();
+    const stats = await run();
+    expect(stats.failed).toBe(0);
+    expect(stats.patientsUpserted).toBe(2);
+    expect(prisma.patient.findMany).toHaveBeenCalledTimes(2);
+    for (const unit of ['brand-a', 'brand-b']) {
+      expect(prisma.patient.findMany).toHaveBeenCalledWith({ where: {
+        hostId: 'host', tenantId: 'tenant', sourceUnit: unit,
+        externalId: { in: method === 'processPatientRelations' ? ['same-id', 'relative'] : ['same-id'] },
+      }, select: { id: true, externalId: true, sourceUnit: true } });
+    }
+    const writes = method === 'processPatientRelations' ? prisma.patientRelation.upsert : prisma.patientReturnVisit.upsert;
+    expect(writes.mock.calls.map((c: any) => c[0].create.patientId)).toEqual(['brand-a-same-id', 'brand-b-same-id']);
+  });
+  test('预检不读写患者与回访关系表', async () => {
+    const { run, prisma } = fixture();
+    await run(true);
+    expect(prisma.patient.findMany).not.toHaveBeenCalled();
+    expect(prisma.patientReturnVisit.upsert).not.toHaveBeenCalled();
+    expect(prisma.patientRelation.upsert).not.toHaveBeenCalled();
+  });
+});
+
+test('患者索引按1000个ID分块，大查询结果不通过函数参数展开', async () => {
+  const rows = Array.from({ length: 138772 }, (_, i) => ({ id: `p-${i}`, externalId: `${i}`, sourceUnit: 'brand' }));
+  const findMany = jest.fn().mockResolvedValueOnce(rows).mockResolvedValue([]);
+  const cold = new ColdImportService({ patient: { findMany } } as never, {} as never, {} as never, {} as never, {} as never, {} as never, {} as never);
+  const result = await (cold as any).buildPatientIndex('host', 'tenant', Array.from({ length: 1001 }, (_, i) => `${i}`), 'brand');
+  expect(result.size).toBe(138772);
+  expect(findMany.mock.calls.map(c => c[0].where.externalId.in.length)).toEqual([1000, 1]);
+});