feat(tenant): P0-3 patient_facts/return_visits 加 source_unit + fact-writer 版本链隔离

- schema+migration:patient_facts (host,tenant,source_unit,subject_id,version)、
  patient_return_visits (host,tenant,source_unit,external_id)
- fact-writer:writeDraft/bulkWrite 的版本链 findFirst + 史表查 + liveLatest map
  全按 (source_unit, subject_id) 复合键隔离;create 带 source_unit
  (集团内 subject_id 品牌级撞 631,不隔离会取错品牌版本=改坏 fact 历史)
- parser-pipeline:按 patientId 查患者 source_unit(批内一次取),注入 PrismaService
- cold-import:return_visit upsert 带 source_unit
- 删旧原地迁移脚本 — 改用"清空重摄"(本地/服务器均可),无原地改 fact 版本链风险

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

apps/pac-service/prisma/data-migrations/2026-06-27_tenant_to_group_jvs-dw.sql

--- 一次性数据迁移(jvs-dw 专属):品牌级 tenant → 集团级 tenant。
---
--- 背景:租户模型改造 P0。把两个品牌 tenant(瑞尔 ba67e6cf… / 瑞泰 77057aed…)
---       合并为一个集团 tenant slug 'ruier-grp',品牌降为 patients.source_unit。
---
--- 安全性(见 docs/tenant-group-org-model-refactor.md §2/§9):
---   "换槽位不加信息" — 把 brand 从 tenant_id 槽搬到 source_unit 槽。
---   472 撞号(同 external_id 不同品牌 = 不同人)由 source_unit 分开,不发生任何合并。
---   顺序:先回填 source_unit(读旧 tenant_id),再 remap tenant_id。
---
--- 可逆:source_unit ↔ 旧 tenant 映射保留;回退即按 source_unit 反推旧 UUID。
--- 幂等:WHERE 限定旧 UUID,重复执行不会二次改(旧 UUID 已不存在)。
-
-BEGIN;
-
--- ① 回填 patients.source_unit(从旧品牌 tenant_id)
-UPDATE patients SET source_unit = '瑞尔' WHERE tenant_id = 'ba67e6cf30dc4f9c9c46adef188bbd04';
-UPDATE patients SET source_unit = '瑞泰' WHERE tenant_id = '77057aed269f4a14957ae0ad0eff359a';
-
--- ② 14 张带 tenant_id 的表:品牌 UUID → 集团 slug 'ruier-grp'
-DO $$
-DECLARE t text;
-BEGIN
-  FOREACH t IN ARRAY ARRAY[
-    'agent_invocations','followup_plans','patient_facts','patient_relations',
-    'patient_return_visits','patient_transactions','patients','persona_recompute_logs',
-    'personas','plan_executions','plan_generation_logs','plan_scripts',
-    'plan_summaries','sync_logs'
-  ] LOOP
-    EXECUTE format(
-      'UPDATE %I SET tenant_id = ''ruier-grp'' WHERE tenant_id IN (%L, %L)',
-      t, 'ba67e6cf30dc4f9c9c46adef188bbd04', '77057aed269f4a14957ae0ad0eff359a'
-    );
-  END LOOP;
-END $$;
-
-COMMIT;

apps/pac-service/prisma/migrations/20260627121924_add_fact_returnvisit_source_unit/migration.sql

+-- patient_facts / patient_return_visits 加 source_unit(发证命名空间)。
+-- 集团模型:subject_id / external_id 品牌级,合 tenant 会撞键;source_unit 进键消歧。
+-- 存量 source_unit 默认 '',新键唯一性与旧键等价,迁移安全。
+
+DROP INDEX "patient_facts_host_id_tenant_id_subject_id_version_key";
+DROP INDEX "patient_return_visits_host_id_tenant_id_external_id_key";
+
+ALTER TABLE "patient_facts" ADD COLUMN "source_unit" TEXT NOT NULL DEFAULT '';
+ALTER TABLE "patient_return_visits" ADD COLUMN "source_unit" TEXT NOT NULL DEFAULT '';
+
+CREATE UNIQUE INDEX "patient_facts_host_id_tenant_id_source_unit_subject_id_vers_key" ON "patient_facts"("host_id", "tenant_id", "source_unit", "subject_id", "version");
+CREATE UNIQUE INDEX "patient_return_visits_host_id_tenant_id_source_unit_externa_key" ON "patient_return_visits"("host_id", "tenant_id", "source_unit", "external_id");

apps/pac-service/prisma/schema.prisma

@@ -331,6 +331,8 @@ model PatientReturnVisit {
   hostId         String    @map("host_id") @db.Uuid
   tenantId       String    @map("tenant_id")
   patientId      String    @map("patient_id") @db.Uuid
+  /// 源组织命名空间(= 该患者的 source_unit);集团内 external_id 品牌级会撞,进键消歧。默认 ''。
+  sourceUnit     String    @default("") @map("source_unit")
   /// host 回访记录 id(幂等键)
   externalId     String    @map("external_id")
   clinicId       String?   @map("clinic_id")
@@ -354,7 +356,7 @@ model PatientReturnVisit {
 
   patient Patient @relation(fields: [patientId], references: [id], onDelete: Cascade)
 
-  @@unique([hostId, tenantId, externalId])
+  @@unique([hostId, tenantId, sourceUnit, externalId])
   @@index([patientId, taskDate(sort: Desc)])
   @@map("patient_return_visits")
 }
@@ -518,6 +520,9 @@ model PatientFact {
   hostId    String @map("host_id") @db.Uuid
   tenantId  String @map("tenant_id")
   patientId String @map("patient_id") @db.Uuid
+  /// 源组织命名空间(= 该患者的 source_unit)。集团内 subject_id 品牌级会撞
+  /// (encounter_record:1011032 在瑞尔/瑞泰是两条不同就诊),进键消歧。默认 ''。
+  sourceUnit String @default("") @map("source_unit")
 
   /// **业务身份,跨版本稳定**(对应宿主侧实体 ID,如 apt_001 / enc_001 的语义身份)。
   /// 同一 subject 被多次变更时新版本沿用同一 subject_id,version 递增
@@ -607,7 +612,7 @@ model PatientFact {
   patient Patient @relation(fields: [patientId], references: [id], onDelete: Restrict)
 
   /// 版本去重 — 同 subject 同 version 不能有两条
-  @@unique([hostId, tenantId, subjectId, version])
+  @@unique([hostId, tenantId, sourceUnit, subjectId, version])
   /// 隔离基础过滤
   @@index([hostId, tenantId])
   /// 拉某患者当前 active 视图(persona / plan 重算主路径)

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

@@ -1513,9 +1513,9 @@ export class ColdImportService {
           () =>
             this.prisma.patientReturnVisit.upsert({
               where: {
-                hostId_tenantId_externalId: { hostId, tenantId, externalId },
+                hostId_tenantId_sourceUnit_externalId: { hostId, tenantId, sourceUnit, externalId },
               },
-              create: { hostId, tenantId, externalId, patientId, ...data },
+              create: { hostId, tenantId, sourceUnit, externalId, patientId, ...data },
               update: { patientId, ...data },
             }),
           'patientReturnVisit upsert',

apps/pac-service/src/modules/sync/pipeline/fact-writer.service.ts

@@ -9,6 +9,11 @@ import {
   validateFactContent,
 } from './fact-content-schemas';
 
+/** fact 版本链 key:集团内 subject_id 品牌级会撞,按 (source_unit, subject_id) 建键。 */
+function factVersionKey(sourceUnit: string, subjectId: string): string {
+  return `${sourceUnit}#${subjectId}`;
+}
+
 /**
  * FactWriter — patient_facts 版本流写入器
  *
@@ -37,10 +42,12 @@ export class FactWriter {
     draft: FactDraft;
     hostId: string;
     tenantId: string;
+    /// 源组织命名空间(= 该患者 source_unit)。集团内 subject_id 品牌级会撞,进键/版本链消歧。
+    sourceUnit: string;
     patientId: string;
     transactionId: string;
   }): Promise<FactWriteResult> {
-    const { draft, hostId, tenantId, patientId, transactionId } = input;
+    const { draft, hostId, tenantId, sourceUnit, patientId, transactionId } = input;
 
     // ─── 闸 1:写库前 zod 强校验(canonical-fact-layer.md §二)──
     // 失败抛 FactContentSchemaError,由 ParserPipeline 计入 factsFailed 并 log issues。
@@ -77,6 +84,7 @@ export class FactWriter {
         where: {
           hostId,
           tenantId,
+          sourceUnit, // ⭐ 集团内 subject_id 跨品牌会撞,版本链必须按 source_unit 隔离,否则取错品牌版本
           subjectId: draft.subjectId,
         },
         orderBy: { version: 'desc' },
@@ -129,6 +137,7 @@ export class FactWriter {
         data: {
           hostId,
           tenantId,
+          sourceUnit,
           patientId,
           subjectId: validatedDraft.subjectId,
           kind: validatedDraft.kind,
@@ -200,15 +209,18 @@ export class FactWriter {
     });
 
     // 2. 一次 SELECT 把所有相关 subject 的 latest version 拿回
+    //    ⭐ 集团内 subject_id 跨品牌会撞 → 史表查带 source_unit 过滤,版本 map 按 (source_unit, subject_id) 建键
     const subjectIds = [...new Set(validatedEntries.map((e) => e.draft.subjectId))];
+    const sourceUnits = [...new Set(validatedEntries.map((e) => e.sourceUnit))];
     const allHist = await this.prisma.patientFact.findMany({
-      where: { hostId, tenantId, subjectId: { in: subjectIds } },
+      where: { hostId, tenantId, sourceUnit: { in: sourceUnits }, subjectId: { in: subjectIds } },
       orderBy: [{ subjectId: 'asc' }, { version: 'desc' }],
     });
-    // 取每 subject 的 highest version 行
+    // 取每 (source_unit, subject) 的 highest version 行
     const latestBySubject = new Map<string, (typeof allHist)[number]>();
     for (const f of allHist) {
-      if (!latestBySubject.has(f.subjectId)) latestBySubject.set(f.subjectId, f);
+      const k = factVersionKey(f.sourceUnit, f.subjectId);
+      if (!latestBySubject.has(k)) latestBySubject.set(k, f);
     }
 
     // 3. 决策每个 entry,累积批操作
@@ -224,8 +236,8 @@ export class FactWriter {
       hash: string;
       transactionIds: string[];
     }>();
-    for (const [sid, latest] of latestBySubject.entries()) {
-      liveLatest.set(sid, {
+    for (const [k, latest] of latestBySubject.entries()) {
+      liveLatest.set(k, {
         id: latest.id,
         version: latest.version,
         status: latest.status,
@@ -236,7 +248,8 @@ export class FactWriter {
 
     for (const entry of validatedEntries) {
       const sid = entry.draft.subjectId;
-      const live = liveLatest.get(sid);
+      const vkey = factVersionKey(entry.sourceUnit, sid); // (source_unit, subject_id) 复合键
+      const live = liveLatest.get(vkey);
       const draftStatus = entry.draft.status ?? FactStatus.ACTIVE;
 
       if (live && live.hash === entry.hash && live.status === draftStatus) {
@@ -271,6 +284,7 @@ export class FactWriter {
       toCreate.push({
         hostId,
         tenantId,
+        sourceUnit: entry.sourceUnit,
         patientId: entry.patientId,
         subjectId: sid,
         kind: entry.draft.kind,
@@ -290,7 +304,7 @@ export class FactWriter {
       // batch 内链式:把"虚拟 latest"刷新成本 draft 内容,后续同 subject 的 draft 用它当 prev
       // (注意 id=undefined 表示是本 batch 内 create,真正写入后才有 id;但 batch 内不会再 supersede 它,
       //  因为 batch 内同 subject 第二次出现也是 create,不 update)
-      liveLatest.set(sid, {
+      liveLatest.set(vkey, {
         id: undefined,
         version: nextVersion,
         status: draftStatus,
@@ -381,6 +395,8 @@ export interface BulkEntry {
   draft: FactDraft;
   hostId: string;
   tenantId: string;
+  /// 源组织命名空间(= 该患者 source_unit);集团内 subject_id 品牌级会撞,版本链按它隔离。
+  sourceUnit: string;
   patientId: string;
   transactionId: string;
 }

apps/pac-service/src/modules/sync/pipeline/parser-pipeline.service.ts

 import { Injectable, Logger } from '@nestjs/common';
 import type { Action } from '@pac/types';
+import { PrismaService } from '../../../prisma/prisma.service';
 import { ParserRegistry } from './parsers/parser.registry';
 import {
   type BulkEntry,
@@ -30,6 +31,7 @@ export class ParserPipeline {
   constructor(
     private readonly registry: ParserRegistry,
     private readonly writer: FactWriter,
+    private readonly prisma: PrismaService,
   ) {}
 
   /**
@@ -78,12 +80,22 @@ export class ParserPipeline {
 
     const drafts = parser.parse({ transaction, canonicalRow });
 
+    // fact 的 source_unit = 该患者的 source_unit(集团内 subject_id 撞号消歧)
+    const patientSourceUnit =
+      (
+        await this.prisma.patient.findUnique({
+          where: { id: transaction.patientId },
+          select: { sourceUnit: true },
+        })
+      )?.sourceUnit ?? '';
+
     for (const draft of drafts) {
       try {
         const result = await this.writer.writeDraft({
           draft,
           hostId: transaction.hostId,
           tenantId: transaction.tenantId,
+          sourceUnit: patientSourceUnit,
           patientId: transaction.patientId,
           transactionId: transaction.id,
         });
@@ -134,6 +146,17 @@ export class ParserPipeline {
       writes: [],
     };
 
+    // fact 的 source_unit = 该患者的 source_unit(集团内 subject_id 撞号消歧)。
+    // patient 先于 fact 落库,按 patientId 批量取一次。
+    const patientIds = [
+      ...new Set(items.map((i) => i.transaction.patientId).filter((p): p is string => !!p)),
+    ];
+    const suRows = await this.prisma.patient.findMany({
+      where: { id: { in: patientIds } },
+      select: { id: true, sourceUnit: true },
+    });
+    const suMap = new Map(suRows.map((p) => [p.id, p.sourceUnit]));
+
     // 1. 全部 tx 走 parser,收集 drafts(in-memory)
     const bulkEntries: BulkEntry[] = [];
     for (const item of items) {
@@ -153,6 +176,7 @@ export class ParserPipeline {
             draft,
             hostId: item.transaction.hostId,
             tenantId: item.transaction.tenantId,
+            sourceUnit: suMap.get(item.transaction.patientId) ?? '',
             patientId: item.transaction.patientId,
             transactionId: item.transaction.id,
           });
@@ -199,6 +223,7 @@ export class ParserPipeline {
             draft: e.draft,
             hostId: e.hostId,
             tenantId: e.tenantId,
+            sourceUnit: e.sourceUnit,
             patientId: e.patientId,
             transactionId: e.transactionId,
           });