merge: fix/push-lookup-fallback → test(push lookup 从 PAC 库回落,兑现推送顺序无要求)

apps/pac-service/data/friday/manifest.yaml

@@ -345,6 +345,15 @@ transforms:
     right_key: id
     select:
       referee_sex: sex
+    # push 是单表请求,customer_basic_info 不在场 → 从 PAC 已入库的 patients 重建对方性别。
+    # 不配这段的话:所有行 referee_sex=null → 码 3 拼出 "3|" → 掉 _default: other,
+    # father/mother 永远为 0(2026-07-28 实测:friday 257 条边 father+mother=0,
+    # 同口径走 cold-import 的 jvs-dw 这两类占 23%),连带「多代之家」画像系统性漏判。
+    # map 把 PAC 归一值反写回宿主原编码,保证 push 与 cold-import 产出同一个 rel_sex_key。
+    push_fallback:
+      entity: patients
+      select:
+        sex: { column: gender, map: { male: '1', female: '2' } }
   - kind: derive
     input: _referee_with_sex
     output: patient_relation_rows

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

@@ -38,6 +38,11 @@ import {
 } from './clickhouse-source.service';
 import { AlertService } from '../../../common/alerting/alert.service';
 import { TransformEngine } from '../transforms/transform-engine';
+import type { TransformOp } from '../transforms/transforms.schema';
+import {
+  buildPushLookupFallbackRows,
+  type PatientFallbackRow,
+} from './push-lookup-fallback';
 import {
   buildTenantResolver,
   buildSourceUnitResolver,
@@ -353,6 +358,7 @@ export class ColdImportService {
       const inp = (tf as { input?: string }).input;
       if (inp && !tables[inp]) tables[inp] = [];
     }
+    await this.hydratePushLookupTables(tables, transforms, host.id);
     const transformed =
       transforms.length > 0 ? this.transformEngine.run({ tables, transforms }) : tables;
 
@@ -568,6 +574,55 @@ export class ColdImportService {
     }
   }
 
+  /**
+   * push 单表:把 lookup 缺席的 `from` 表用 PAC 已入库的 patients 重建。
+   *
+   * 只在 `tables[from]` 缺席或为空时动手 —— cold-import / reparse 的 from 表本来就在,
+   * 走这里会被直接跳过,两条通道的产物因此保持一致(map 把 PAC 归一值反写回宿主原编码)。
+   *
+   * 不配 push_fallback 的 lookup 行为不变(全部未命中填 null),即本改动对存量 host 零影响。
+   * 语义与踩坑背景见 LookupPushFallbackSchema 注释。
+   */
+  private async hydratePushLookupTables(
+    tables: Record<string, Record<string, unknown>[]>,
+    transforms: readonly TransformOp[],
+    hostId: string,
+  ): Promise<void> {
+    for (const tf of transforms) {
+      if (tf.kind !== 'lookup' || !tf.push_fallback) continue;
+      if (tables[tf.from]?.length) continue; // from 表在场(cold-import/reparse)→ 不接管
+      const input = tables[tf.input];
+      if (!input?.length) continue;
+
+      const { rows, resolved, ambiguous, absent } = await buildPushLookupFallbackRows(
+        tf,
+        input,
+        (row) => this.sourceUnitResolver.resolve(row),
+        async ({ sourceUnits, externalIds }) =>
+          (await this.prisma.patient.findMany({
+            where: {
+              hostId,
+              externalId: { in: externalIds },
+              ...(sourceUnits.length > 0 ? { sourceUnit: { in: sourceUnits } } : {}),
+            },
+            select: {
+              externalId: true,
+              sourceUnit: true,
+              gender: true,
+              birthDate: true,
+              name: true,
+              medicalRecordNumber: true,
+            },
+          })) as PatientFallbackRow[],
+      );
+      tables[tf.from] = rows;
+      this.logger.log(
+        `[push/lookup-fallback] "${tf.input}" → 重建 "${tf.from}"(${tf.push_fallback.entity}) ` +
+          `resolved=${resolved} absent=${absent} ambiguous=${ambiguous}`,
+      );
+    }
+  }
+
   /// host → 配置目录(= sync.service.resolveDataDir 同约定:env PAC_INCREMENTAL_DATA_DIR ?? cwd/data,再拼 hostName)
   private resolveHostDir(hostName: string): string {
     const base = process.env.PAC_INCREMENTAL_DATA_DIR ?? path.resolve(process.cwd(), 'data');

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

+import type { LookupOp } from '../transforms/transforms.schema';
+import type { Row } from '../transforms/row';
+
+/**
+ * push 通道 lookup 回落 —— 用 PAC 已入库的 patients 重建缺席的 `from` 表。
+ *
+ * 背景与语义见 LookupPushFallbackSchema 的注释(transforms.schema.ts)。这里只做纯逻辑:
+ * 取数交给调用方传进来的 fetch(便于单测不碰 prisma),本模块负责
+ *   ① 从 input 行收集 (source_unit, external_id) 需求
+ *   ② 按 external_id 归组、跨品牌撞号则跳过
+ *   ③ 按 select 的 column/map 拼出合成行
+ */
+
+/** PAC patients 里回落用得上的列(与 LookupPushFallbackSchema.select.column 白名单对齐) */
+export interface PatientFallbackRow {
+  externalId: string;
+  sourceUnit: string | null;
+  gender: string | null;
+  birthDate: Date | null;
+  name: string | null;
+  medicalRecordNumber: string | null;
+}
+
+export type PatientFallbackFetch = (args: {
+  sourceUnits: string[];
+  externalIds: string[];
+}) => Promise<PatientFallbackRow[]>;
+
+export interface PushLookupFallbackResult {
+  rows: Row[];
+  /** 成功重建的合成行数 */
+  resolved: number;
+  /** 因 external_id 跨 source_unit 撞号而整条跳过的 key 数 */
+  ambiguous: number;
+  /** input 里要查、但 PAC 库还没有的 external_id 数(对方主档尚未推达) */
+  absent: number;
+}
+
+/**
+ * IN 列表分批大小 —— Postgres 单条语句绑定变量上限 32767,留足余量。
+ * (2026-07 存量补摄时踩过 bind-var 溢出,见 clinic-backfill 手册。)
+ */
+const FETCH_CHUNK = 5000;
+
+export async function buildPushLookupFallbackRows(
+  op: LookupOp,
+  inputRows: readonly Row[],
+  resolveSourceUnit: (row: Row) => string,
+  fetch: PatientFallbackFetch,
+): Promise<PushLookupFallbackResult> {
+  const fallback = op.push_fallback;
+  if (!fallback) return { rows: [], resolved: 0, ambiguous: 0, absent: 0 };
+
+  // ① 收集需求:external_id → 该 id 在本批里出现过的 source_unit 集合
+  const wanted = new Map<string, Set<string>>();
+  for (const row of inputRows) {
+    const raw = row[op.left_key];
+    if (raw === null || raw === undefined) continue;
+    const id = String(raw).trim();
+    if (!id) continue;
+    const unit = resolveSourceUnit(row);
+    const units = wanted.get(id);
+    if (units) units.add(unit);
+    else wanted.set(id, new Set([unit]));
+  }
+  if (wanted.size === 0) return { rows: [], resolved: 0, ambiguous: 0, absent: 0 };
+
+  // ② 取数(按 external_id 分批;source_unit 一次性传全,基数很小)
+  const allUnits = [...new Set([...wanted.values()].flatMap((s) => [...s]))];
+  const ids = [...wanted.keys()];
+  const fetched: PatientFallbackRow[] = [];
+  for (let i = 0; i < ids.length; i += FETCH_CHUNK) {
+    fetched.push(
+      ...(await fetch({ sourceUnits: allUnits, externalIds: ids.slice(i, i + FETCH_CHUNK) })),
+    );
+  }
+
+  // ③ 归组 + 撞号裁决
+  const byExternalId = new Map<string, PatientFallbackRow[]>();
+  for (const p of fetched) {
+    const bucket = byExternalId.get(p.externalId);
+    if (bucket) bucket.push(p);
+    else byExternalId.set(p.externalId, [p]);
+  }
+
+  const rows: Row[] = [];
+  let ambiguous = 0;
+  let absent = 0;
+  for (const id of ids) {
+    const candidates = byExternalId.get(id);
+    if (!candidates || candidates.length === 0) {
+      absent++;
+      continue;
+    }
+    // 同一 external_id 落到多个品牌 → 无法判断 input 行指的是哪一个,整条跳过。
+    // 跳过 = 下游填 null → enum_mapping 的 _default 兜底,语义是"不知道",不是"猜一个"。
+    if (candidates.length > 1) {
+      ambiguous++;
+      continue;
+    }
+    rows.push(synthesize(op, candidates[0]!));
+    // resolved 在下面统一按 rows.length 算
+  }
+
+  return { rows, resolved: rows.length, ambiguous, absent };
+}
+
+function synthesize(op: LookupOp, p: PatientFallbackRow): Row {
+  const out: Row = { [op.right_key]: p.externalId };
+  for (const [target, spec] of Object.entries(op.push_fallback!.select)) {
+    const raw = readColumn(p, spec.column);
+    if (raw === null) {
+      out[target] = null;
+      continue;
+    }
+    // map 未命中当作"取不到" —— 宁可填 null 走 _default,也不把 PAC 归一值直接漏给
+    // 只认宿主原编码的下游 enum_mapping(那会静默产生一个谁也没配过的键)。
+    out[target] = spec.map ? (spec.map[raw] ?? null) : raw;
+  }
+  return out;
+}
+
+function readColumn(p: PatientFallbackRow, column: string): string | null {
+  switch (column) {
+    case 'gender':
+      return emptyToNull(p.gender);
+    case 'name':
+      return emptyToNull(p.name);
+    case 'medicalRecordNumber':
+      return emptyToNull(p.medicalRecordNumber);
+    case 'birthDate':
+      return p.birthDate ? p.birthDate.toISOString().slice(0, 10) : null;
+    default:
+      return null;
+  }
+}
+
+/** 宿主"未知性别"归一后落库是空串(见 friday patient.yaml 的 _default: ''),等同取不到 */
+function emptyToNull(v: string | null): string | null {
+  const s = (v ?? '').trim();
+  return s === '' ? null : s;
+}

apps/pac-service/src/modules/sync/transforms/transforms.schema.ts

@@ -322,6 +322,48 @@ const LookupWhereRuleSchema = z.union([
 export type LookupWhereRule = z.infer<typeof LookupWhereRuleSchema>;
 
 /**
+ * push 通道的 `from` 表回落 —— 从 PAC 已入库的实体重建候选行。
+ *
+ * 【为什么需要】push 是**单表**请求:ingestRawTables 只把被推那张表放进 tables,
+ *   lookup 的 `from` 表根本不在场,于是全部未命中、select 字段填 null。
+ *   而接入文档 channel-push.mdx §7.3 明写「推送顺序无要求」—— 不回落就等于偷偷要求
+ *   宿主先推 from 表,违背自己的契约。
+ *   2026-07-28 FRIDAY 实测踩到:customer_referee_circle 拿不到对方性别,关系码 3
+ *   (本人是对方子女)无法拆成 father/mother,257 条关系边里 father/mother **全为 0**
+ *   (同口径的 jvs-dw 走 cold-import,father+mother 占 23%),连带 family_structure
+ *   的「多代之家」系统性漏判。
+ *
+ * 【语义】仅在 `tables[from]` 缺席或为空时生效(cold-import / reparse 表本来就在,不受影响)。
+ *   按 input 行的 left_key 去 PAC 库捞实体,拼成"长得像 from 表"的合成行:
+ *   `{ [right_key]: 实体的 external_id, ...select 映射出的列 }`。
+ *
+ * 【source_unit 消歧】external_id 在集团型宿主里跨品牌会撞号(FRIDAY 22 个 source_unit,
+ *   实测 310 个 id 撞号)。捞取按 input 行解析出的 source_unit 圈定;若同一 external_id
+ *   仍落到多个品牌 → **整条跳过**(填 null 走 _default),不硬猜。
+ *
+ * 【map】把 PAC 归一值反写回宿主原编码(如 male→'1'),保证 push 与 cold-import
+ *   产出**同一个**下游键值 —— 否则两条通道的 enum_mapping 得各配一套,迟早漂移。
+ */
+export const LookupPushFallbackSchema = z.object({
+  entity: z.literal('patients').describe('回落取数的 PAC 实体(目前只支持 patients)'),
+  select: z
+    .record(
+      z.string(),
+      z.object({
+        column: z
+          .enum(['gender', 'birthDate', 'name', 'medicalRecordNumber'])
+          .describe('PAC 实体列(白名单;需要别的列时在此显式加,不做动态列名)'),
+        map: z
+          .record(z.string(), z.string())
+          .optional()
+          .describe('PAC 归一值 → 宿主原编码;未命中的值当作取不到(填 null)'),
+      }),
+    )
+    .describe('合成行列名 ← PAC 实体列(与 lookup.select 的来源侧对齐)'),
+});
+export type LookupPushFallback = z.infer<typeof LookupPushFallbackSchema>;
+
+/**
  * 把 `from` 子表按外键 join 到 `input` 主表,多条候选时按 order_by 挑第一条,
  * 把 select 里的字段附加到主表行(等价 LEFT JOIN + 取一条 + 选列)。
  *
@@ -358,6 +400,9 @@ export const LookupOpSchema = z.object({
   select: z
     .record(z.string(), z.string())
     .describe('附加字段:目标字段(主表列名) ← 来源字段(from 列名)'),
+  push_fallback: LookupPushFallbackSchema.optional().describe(
+    'push 单表时 from 表缺席 → 从 PAC 库重建候选行(不配 = 保持旧行为:全部未命中填 null)',
+  ),
 });
 export type LookupOp = z.infer<typeof LookupOpSchema>;
 

apps/pac-service/tests/push-lookup-fallback.spec.ts

+/**
+ * push 通道 lookup 回落(from 表缺席 → 从 PAC patients 重建)
+ *
+ * 契约背景:channel-push.mdx §7.3「推送顺序无要求」。push 是单表请求,lookup 的 from 表
+ * 不在场,不回落就等于偷偷要求宿主先推 from 表。2026-07-28 FRIDAY 实测踩到:
+ * customer_referee_circle 拿不到对方性别 → 关系码 3 无法拆 father/mother → 全掉 other。
+ *
+ * 这里连着 runLookup 一起测,断言的是**最终喂给 assembler 的那个值**,
+ * 而不是中间产物 —— 中间对了下游错了没有意义。
+ */
+import {
+  buildPushLookupFallbackRows,
+  type PatientFallbackRow,
+} from '../src/modules/sync/cold-import/push-lookup-fallback';
+import { runLookup } from '../src/modules/sync/transforms/operators/lookup.op';
+import type { LookupOp } from '../src/modules/sync/transforms/transforms.schema';
+
+/** friday manifest 里那条真实配置 */
+const OP: LookupOp = {
+  kind: 'lookup',
+  input: 'customer_referee_circle',
+  output: '_referee_with_sex',
+  from: 'customer_basic_info',
+  left_key: 'referee_patient_id',
+  right_key: 'id',
+  select: { referee_sex: 'sex' },
+  push_fallback: {
+    entity: 'patients',
+    select: { sex: { column: 'gender', map: { male: '1', female: '2' } } },
+  },
+};
+
+const patient = (o: Partial<PatientFallbackRow> & { externalId: string }): PatientFallbackRow => ({
+  sourceUnit: 'brandA',
+  gender: null,
+  birthDate: null,
+  name: null,
+  medicalRecordNumber: null,
+  ...o,
+});
+
+/** 源品牌列 = friday 的 identity_namespace_field: tenant_id */
+const resolveUnit = (row: Record<string, unknown>) => String(row.tenant_id ?? '');
+
+const fetchOf = (rows: PatientFallbackRow[]) => async () => rows;
+
+describe('push lookup 回落 | 重建 from 表', () => {
+  test('对方已入库 → 性别按宿主原编码回填,码 3 能拆父/母', async () => {
+    const input = [
+      { customer_id: 'A', referee_patient_id: 'B', referee_relationship: '3', tenant_id: 'brandA' },
+      { customer_id: 'C', referee_patient_id: 'D', referee_relationship: '3', tenant_id: 'brandA' },
+    ];
+    const { rows, resolved, absent, ambiguous } = await buildPushLookupFallbackRows(
+      OP,
+      input,
+      resolveUnit,
+      fetchOf([
+        patient({ externalId: 'B', gender: 'male' }),
+        patient({ externalId: 'D', gender: 'female' }),
+      ]),
+    );
+    expect({ resolved, absent, ambiguous }).toEqual({ resolved: 2, absent: 0, ambiguous: 0 });
+
+    const joined = runLookup(OP, input, rows);
+    expect(joined.rows.map((r) => `${String(r.referee_relationship)}|${String(r.referee_sex)}`))
+      .toEqual(['3|1', '3|2']); // → patient_relation.yaml 里映射成 father / mother
+  });
+
+  test('对方主档还没推达 → 不硬猜,落 null(下游走 _default: other)', async () => {
+    const input = [
+      { customer_id: 'A', referee_patient_id: 'B', referee_relationship: '3', tenant_id: 'brandA' },
+    ];
+    const { rows, resolved, absent } = await buildPushLookupFallbackRows(
+      OP,
+      input,
+      resolveUnit,
+      fetchOf([]),
+    );
+    expect({ resolved, absent }).toEqual({ resolved: 0, absent: 1 });
+    expect(runLookup(OP, input, rows).rows[0]!.referee_sex).toBeNull();
+  });
+
+  test('external_id 跨 source_unit 撞号 → 整条跳过,不猜哪个品牌', async () => {
+    const input = [
+      { customer_id: 'A', referee_patient_id: 'B', referee_relationship: '3', tenant_id: 'brandA' },
+    ];
+    const { rows, ambiguous, resolved } = await buildPushLookupFallbackRows(
+      OP,
+      input,
+      resolveUnit,
+      fetchOf([
+        patient({ externalId: 'B', sourceUnit: 'brandA', gender: 'male' }),
+        patient({ externalId: 'B', sourceUnit: 'brandB', gender: 'female' }),
+      ]),
+    );
+    expect({ ambiguous, resolved }).toEqual({ ambiguous: 1, resolved: 0 });
+    expect(rows).toHaveLength(0);
+  });
+
+  test('宿主未知性别落库是空串 → 等同取不到', async () => {
+    const input = [
+      { customer_id: 'A', referee_patient_id: 'B', referee_relationship: '3', tenant_id: 'brandA' },
+    ];
+    const { rows } = await buildPushLookupFallbackRows(
+      OP,
+      input,
+      resolveUnit,
+      fetchOf([patient({ externalId: 'B', gender: '' })]),
+    );
+    expect(rows[0]!.sex).toBeNull();
+  });
+
+  test('map 未命中 → null,不把 PAC 归一值漏给只认宿主编码的下游', async () => {
+    const input = [
+      { customer_id: 'A', referee_patient_id: 'B', referee_relationship: '3', tenant_id: 'brandA' },
+    ];
+    const { rows } = await buildPushLookupFallbackRows(
+      OP,
+      input,
+      resolveUnit,
+      fetchOf([patient({ externalId: 'B', gender: 'unknown-新值' })]),
+    );
+    expect(rows[0]!.sex).toBeNull();
+  });
+
+  test('没配 push_fallback 的 lookup 保持旧行为(本改动对存量 host 零影响)', async () => {
+    const bare: LookupOp = { ...OP, push_fallback: undefined };
+    const r = await buildPushLookupFallbackRows(
+      bare,
+      [{ referee_patient_id: 'B', tenant_id: 'brandA' }],
+      resolveUnit,
+      fetchOf([patient({ externalId: 'B', gender: 'male' })]),
+    );
+    expect(r).toEqual({ rows: [], resolved: 0, ambiguous: 0, absent: 0 });
+  });
+
+  test('left_key 空/缺失的行不产生查询需求', async () => {
+    const fetch = jest.fn(async () => []);
+    const { resolved, absent } = await buildPushLookupFallbackRows(
+      OP,
+      [
+        { referee_patient_id: null, tenant_id: 'brandA' },
+        { referee_patient_id: '  ', tenant_id: 'brandA' },
+        { tenant_id: 'brandA' },
+      ],
+      resolveUnit,
+      fetch,
+    );
+    expect(fetch).not.toHaveBeenCalled();
+    expect({ resolved, absent }).toEqual({ resolved: 0, absent: 0 });
+  });
+});