feat(sync): 定向名单支持命名空间维 —— 患者主键只在命名空间内唯一

测试服实测:7 万个纯 id 的定向名单列出 **140,566** 个 cohort key,整整翻倍 ——
cohort key 是 (patient_key, tenant_key) 复合键,而 PAC_COHORT_ONLY_PATIENT 只承载
patient_key,于是每个 id 在两个命名空间下各命中一次,一半是无关的同号患者:
白摄一倍数据、批次翻倍。结果不错(另一命名空间的患者算出 false,不会误标),但纯属浪费。

⭐ 通用性:这一维**不叫 brand**。整条 cohort 链路(CohortKey / tenant_key_column /
injectCohortFilter)本来就只认 manifest 声明的列名,代码里不出现宿主字样 —— jvs-dw 恰好
填的是 brand,别的宿主可能是区域 / 诊所 / 不设。漏的只有 ONLY_PATIENT 这个运维参数,
它返回 string[] 把第二维丢了。

改:新增 resolveOnlyPatientKeys() → OnlyPatientKey{key, tenant?},名单每项支持
  `1855960`         纯 key(单命名空间宿主 / 该 id 在所有命名空间下都要)—— 行为不变
  `261067|瑞尔`      显式分隔
  `261067<TAB>瑞尔`  TSV(SQL dump 可直接喂)
全部带命名空间且宿主配了 tenant_key_column → 拼复合键 IN;否则退回单键 IN,
混写(只有部分带)不做部分匹配,warn 一次后整体退回 —— 半精确比全模糊更难排查。
分片路径同步支持三种起手形态。resolveOnlyPatientIds() 保留为 key 投影,旧调用点
(cold-import 判"是否定向模式 → 不推进游标")零改动。

测试 799 项(+7)。

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

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

@@ -11,20 +11,55 @@ import { DEFAULT_REVERSE_PULL_FROM, type ClickHouseSource } from './manifest.sch
  * 未设置 / 空 → 返回 [](调用方据此判断"是否定向模式",影响 cohort WHERE 与 cursor 推进)。
  */
 export function resolveOnlyPatientIds(): string[] {
+  return resolveOnlyPatientKeys().map((k) => k.key);
+}
+
+/**
+ * 同上,但保留**命名空间维**(manifest cohort.tenant_key_column 声明的那一维)。
+ *
+ * 【为什么需要第二维】患者主键只在命名空间内唯一:jvs-dw 的 patient_id 261067 在
+ * 「瑞尔」和「瑞泰」是两个不同的人(集团模型,见 manifest identity_namespace_field)。
+ * 只按 key 定向 → 两边同号的人一起被拖进 cohort。2026-08-01 测试服实测:
+ * 7 万个 id 的定向名单列出 **140,566** 个 cohort key —— 整整翻倍,一半是无关的另一命名空间患者。
+ *
+ * 【通用性】这里刻意**不出现 brand 字样**:第二维叫什么由各宿主的 manifest 声明
+ * (jvs-dw 是 brand;别的宿主可能是区域 / 诊所 / 不设)。没配 tenant_key_column 的
+ * 单命名空间宿主写纯 id 即可,行为与本改动前完全一致。
+ *
+ * 格式(每行 / 每个逗号项):
+ *   `1855960`              → { key }              单命名空间,或"该 id 在所有命名空间下都要"
+ *   `1855960|瑞尔`          → { key, tenant }      精确到命名空间
+ *   `1855960<TAB>瑞尔`      → 同上(便于 SQL 直接 dump TSV)
+ */
+export interface OnlyPatientKey {
+  key: string;
+  tenant?: string;
+}
+
+export function resolveOnlyPatientKeys(): OnlyPatientKey[] {
   const raw = process.env.PAC_COHORT_ONLY_PATIENT?.trim();
   if (!raw) return [];
+  let lines: string[];
   if (raw.startsWith('@')) {
     const file = raw.slice(1);
     if (!fs.existsSync(file)) {
       throw new Error(`PAC_COHORT_ONLY_PATIENT 指向的文件不存在:${file}`);
     }
-    return fs
-      .readFileSync(file, 'utf-8')
-      .split('\n')
-      .map((s) => s.trim())
-      .filter(Boolean);
+    lines = fs.readFileSync(file, 'utf-8').split('\n');
+  } else {
+    lines = raw.split(',');
   }
-  return raw.split(',').map((s) => s.trim()).filter(Boolean);
+  return lines
+    .map((s) => s.trim())
+    .filter(Boolean)
+    .map((line) => {
+      // `|` 优先(显式分隔),否则 TAB(SQL dump 的天然形态);都没有 → 只有 key
+      const parts = line.includes('|') ? line.split('|') : line.split('\t');
+      const key = (parts[0] ?? '').trim();
+      const tenant = parts.length > 1 ? (parts[1] ?? '').trim() : undefined;
+      return tenant ? { key, tenant } : { key };
+    })
+    .filter((k) => k.key);
 }
 
 /**
@@ -484,11 +519,34 @@ export class ClickHouseSourceService {
       //   单患者复现 / 定向重摄受影响子集(分类修复后只重摄 reclassify 的患者)用。
       //   ⭐ 大名单(上万个 id)用 `@/path/to/file` 从文件读(每行一个 id):环境变量单值有
       //   128KB(MAX_ARG_STRLEN)上限,3 万个 id 的逗号串约 275KB 会直接 E2BIG。
-      const ids = resolveOnlyPatientIds();
-      if (ids.length > 0) {
-        const quoted = ids.map((id) => `'${id.replace(/'/g, "''")}'`).join(', ');
-        whereParts.push(ids.length === 1 ? `${patient_key_column} = ${quoted}` : `${patient_key_column} IN (${quoted})`);
-        this.logger.log(`[clickhouse·cohort] PAC_COHORT_ONLY_PATIENT 定向:${ids.length} 个患者`);
+      //   ⭐ 名单可带**命名空间维**(`id|<tenant值>` 或 TAB 分隔)—— 患者主键只在命名空间内唯一,
+      //   只按 key 定向会把别的命名空间下的同号患者一起拖进来(实测 7 万名单列出 14 万 key)。
+      //   维名不写死:用 manifest 的 tenant_key_column;没配该列的宿主写纯 id,行为不变。
+      const onlyKeys = resolveOnlyPatientKeys();
+      const ids = onlyKeys.map((k) => k.key);
+      if (onlyKeys.length > 0) {
+        const q = (v: string) => `'${v.replace(/'/g, "''")}'`;
+        const withTenant = onlyKeys.filter((k) => k.tenant);
+        if (tenant_key_column && withTenant.length === onlyKeys.length) {
+          // 全部带命名空间 → 复合键精确定位(与 cohort key 同形态)
+          const tuples = onlyKeys.map((k) => `(${q(k.key)}, ${q(k.tenant!)})`).join(', ');
+          whereParts.push(`(${patient_key_column}, ${tenant_key_column}) IN (${tuples})`);
+        } else {
+          if (withTenant.length > 0) {
+            this.logger.warn(
+              `[clickhouse·cohort] 名单里只有 ${withTenant.length}/${onlyKeys.length} 项带命名空间 —— ` +
+                `混写不做部分匹配,整体退回按 ${patient_key_column} 定向(同号跨命名空间会一并纳入)`,
+            );
+          }
+          const quoted = ids.map(q).join(', ');
+          whereParts.push(
+            ids.length === 1 ? `${patient_key_column} = ${quoted}` : `${patient_key_column} IN (${quoted})`,
+          );
+        }
+        this.logger.log(
+          `[clickhouse·cohort] PAC_COHORT_ONLY_PATIENT 定向:${onlyKeys.length} 个患者` +
+            (tenant_key_column && withTenant.length === onlyKeys.length ? `(含命名空间,精确匹配)` : ''),
+        );
       }
       // --clinics=X,Y:把 cohort 收窄到「在这些诊所看过」的患者。诊所过滤挂在有 org 的事实表
       //   (cohort.clinic_scope),用与 cohort 相同的患者键求交集。⚠️ 只收窄「列哪些患者」,
@@ -559,13 +617,21 @@ export class ClickHouseSourceService {
       //   DISTINCT 在合并后用 Map 去重 —— 与单条 SQL 的结果集完全等价。
       //   注:PAC_COHORT_LIMIT 采样与分片叠加时按"每片各取 LIMIT"处理会超量,故分片时不走
       //   orderTail 的 LIMIT(定向重摄本就是显式点名,不该再被采样截断)。
-      const ID_CHUNK = 10_000; // 10k × ~9B ≈ 90KB,离 256KiB 有足够余量(id 变长也不会顶到)
-      const idChunks: string[][] =
-        ids.length > ID_CHUNK
-          ? Array.from({ length: Math.ceil(ids.length / ID_CHUNK) }, (_, i) =>
-              ids.slice(i * ID_CHUNK, (i + 1) * ID_CHUNK),
+      //   分片切的是 onlyKeys(带不带命名空间维都按同一形态重建 IN 子句)。
+      const ID_CHUNK = 10_000; // 10k × ~9B ≈ 90KB,离 256KiB 有足够余量(带命名空间也顶不到)
+      const idChunks: OnlyPatientKey[][] =
+        onlyKeys.length > ID_CHUNK
+          ? Array.from({ length: Math.ceil(onlyKeys.length / ID_CHUNK) }, (_, i) =>
+              onlyKeys.slice(i * ID_CHUNK, (i + 1) * ID_CHUNK),
             )
           : [];
+      /// 本片的 IN 子句 —— 与上面 whereParts 的形态保持一致(复合键 / 单键)
+      const chunkInClause = (chunk: OnlyPatientKey[]): string => {
+        const q = (v: string) => `'${v.replace(/'/g, "''")}'`;
+        return tenant_key_column && chunk.every((k) => k.tenant)
+          ? `(${patient_key_column}, ${tenant_key_column}) IN (${chunk.map((k) => `(${q(k.key)}, ${q(k.tenant!)})`).join(', ')})`
+          : `${patient_key_column} IN (${chunk.map((k) => q(k.key)).join(', ')})`;
+      };
       const started = Date.now();
       let rows: Array<Record<string, unknown>>;
       if (idChunks.length > 0) {
@@ -574,11 +640,14 @@ export class ClickHouseSourceService {
         );
         const merged = new Map<string, Record<string, unknown>>();
         for (const [i, chunk] of idChunks.entries()) {
-          const quotedChunk = chunk.map((id) => `'${id.replace(/'/g, "''")}'`).join(', ');
-          // 用本片的 IN 替换掉整名单那一条(其余 whereParts 保持不变)
+          // 用本片的 IN 替换掉整名单那一条(其余 whereParts 保持不变)。
+          // 三种起手形态都要认:单键 IN / 单键 = / 复合键 IN。
+          const chunkIn = chunkInClause(chunk);
           const chunkWhere = whereParts.map((w) =>
-            w.startsWith(`${patient_key_column} IN (`) || w.startsWith(`${patient_key_column} = `)
-              ? `${patient_key_column} IN (${quotedChunk})`
+            w.startsWith(`${patient_key_column} IN (`) ||
+            w.startsWith(`${patient_key_column} = `) ||
+            w.startsWith(`(${patient_key_column}, ${tenant_key_column}) IN (`)
+              ? chunkIn
               : w,
           );
           const chunkSql = `SELECT DISTINCT ${selectCols} FROM ${listFrom} WHERE ${chunkWhere.join(' AND ')}`;

apps/pac-service/tests/cohort-only-patient.spec.ts

 import * as fs from 'node:fs';
 import * as os from 'node:os';
 import * as path from 'node:path';
-import { resolveOnlyPatientIds } from '../src/modules/sync/cold-import/clickhouse-source.service';
+import {
+  resolveOnlyPatientIds,
+  resolveOnlyPatientKeys,
+} from '../src/modules/sync/cold-import/clickhouse-source.service';
 
 /**
  * PAC_COHORT_ONLY_PATIENT(定向重摄名单)解析。
@@ -64,4 +67,42 @@ describe('PAC_COHORT_ONLY_PATIENT 解析', () => {
     process.env.PAC_COHORT_ONLY_PATIENT = '@/no/such/pids.txt';
     expect(() => resolveOnlyPatientIds()).toThrow(/不存在/);
   });
+
+  /**
+   * 命名空间维(manifest cohort.tenant_key_column 声明的那一维)。
+   *
+   * 患者主键只在命名空间内唯一 —— jvs-dw 的 patient_id 261067 在两个品牌下是两个人。
+   * 2026-08-01 测试服实测:7 万个纯 id 的定向名单列出 **140,566** 个 cohort key(翻倍),
+   * 一半是另一命名空间下的同号患者,白摄一倍数据。
+   * ⚠️ 维名不写死("brand" 只是 jvs-dw 填进 tenant_key_column 的值),单命名空间宿主写纯 id。
+   */
+  describe('命名空间维', () => {
+    test('纯 id → 无 tenant(单命名空间宿主,行为与改动前一致)', () => {
+      process.env.PAC_COHORT_ONLY_PATIENT = '1855960,1855959';
+      expect(resolveOnlyPatientKeys()).toEqual([{ key: '1855960' }, { key: '1855959' }]);
+    });
+
+    test('`id|命名空间` → 复合键', () => {
+      process.env.PAC_COHORT_ONLY_PATIENT = '261067|瑞尔,261067|瑞泰';
+      expect(resolveOnlyPatientKeys()).toEqual([
+        { key: '261067', tenant: '瑞尔' },
+        { key: '261067', tenant: '瑞泰' },
+      ]);
+    });
+
+    test('TAB 分隔 → 同义(SQL dump TSV 可直接喂)', () => {
+      tmpFile = path.join(os.tmpdir(), `pids-tsv-${Date.now()}.txt`);
+      fs.writeFileSync(tmpFile, '261067\t瑞尔\n261068\t瑞泰\n');
+      process.env.PAC_COHORT_ONLY_PATIENT = `@${tmpFile}`;
+      expect(resolveOnlyPatientKeys()).toEqual([
+        { key: '261067', tenant: '瑞尔' },
+        { key: '261068', tenant: '瑞泰' },
+      ]);
+    });
+
+    test('resolveOnlyPatientIds 仍只返回 key —— 旧调用点(是否定向模式的判定)不受影响', () => {
+      process.env.PAC_COHORT_ONLY_PATIENT = '261067|瑞尔,261068|瑞泰';
+      expect(resolveOnlyPatientIds()).toEqual(['261067', '261068']);
+    });
+  });
 });