merge: feat/recompute-plans-clinics → test(plans 支持按诊所子集重算)

apps/pac-service/src/cli/recompute-plans.cli.ts

@@ -5,6 +5,11 @@
  *   pnpm recompute-plans                   # 默认 host=demo,全量
  *   pnpm recompute-plans -- --host=friday
  *   pnpm recompute-plans -- --host=jvs-dw --pids=998421,xxx   # 只重算指定 externalId(定向,O(子集))
+ *   pnpm recompute-plans -- --host=jvs-dw --clinics=<orgId>,<orgId>  # 按诊所补摄后定向重算
+ *
+ * ⚠️ **--clinics 只用于「按诊所补摄后立刻补计划」**,不能拿它替代日常全量。
+ *    召回是**时间驱动**的:数据一个字没变,沉默时长跨过阈值也该出计划 —— 收窄到几家诊所
+ *    就等于其余诊所当天不评估。全量那轮(定时任务)照跑,这个参数只是省掉补摄后的那次全扫。
  */
 import { NestFactory } from '@nestjs/core';
 import { Logger } from '@nestjs/common';
@@ -15,6 +20,7 @@ import { PrismaService } from '../prisma/prisma.service';
 interface Args {
   host: string;
   pids?: string[]; // 指定 externalId(逗号分隔)→ 只重算这些患者(定向,recomputeForPatient)
+  clinics?: string[]; // --clinics=X,Y:只重算"在这些诊所有过操作"的患者(批量路径收窄,非逐患者)
 }
 
 function parseArgs(argv: string[]): Args {
@@ -23,6 +29,8 @@ function parseArgs(argv: string[]): Args {
     if (a.startsWith('--host=')) args.host = a.slice('--host='.length);
     else if (a.startsWith('--pids=')) {
       args.pids = a.slice('--pids='.length).split(',').map((s) => s.trim()).filter(Boolean);
+    } else if (a.startsWith('--clinics=')) {
+      args.clinics = a.slice('--clinics='.length).split(',').map((s) => s.trim()).filter(Boolean);
     }
   }
   return args;
@@ -72,6 +80,22 @@ async function bootstrap() {
       return;
     }
 
+    // ── --clinics:把本轮收窄到"在这些诊所有过操作(patient_transactions.clinic_id)"的患者 ──
+    //   与 cold-import / recompute-persona 的 --clinics **同一口径**,用于按诊所补摄后只补这批人的
+    //   计划,免掉全 host 全扫(生产实测全量单轮 ≈1h50m,其中 ~1h43m 花在 selectHits 全表扫)。
+    let clinicPatientIds: string[] | undefined;
+    if (args.clinics?.length) {
+      const rows = await prisma.patientTransaction.findMany({
+        where: { hostId: host.id, clinicId: { in: args.clinics } },
+        select: { patientId: true },
+        distinct: ['patientId'],
+      });
+      const ids = rows.map((r) => r.patientId).filter((x): x is string => !!x);
+      logger.log(`--clinics=${args.clinics.join(',')} → 命中 ${ids.length} 位患者(按诊所收窄)`);
+      if (ids.length === 0) throw new Error('--clinics 未命中任何患者(诊所 id 是否正确 / 是否已摄入?)');
+      clinicPatientIds = ids;
+    }
+
     // 取该 host 第一个 tenant(demo 场景固定一个)
     const tenants = await prisma.patient.findMany({
       where: { hostId: host.id },
@@ -81,10 +105,14 @@ async function bootstrap() {
     if (tenants.length === 0) throw new Error('No tenants found for host');
 
     for (const t of tenants) {
-      logger.log(`▶ Running engine for host=${args.host} tenant=${t.tenantId} ...`);
+      logger.log(
+        `▶ Running engine for host=${args.host} tenant=${t.tenantId}` +
+          `${clinicPatientIds ? ` (子集:${clinicPatientIds.length} 位患者)` : ' (全量)'} ...`,
+      );
       const r = await engine.runAllForHost({
         hostId: host.id,
         tenantId: t.tenantId,
+        ...(clinicPatientIds ? { patientIds: clinicPatientIds } : {}),
       });
       logger.log(`──────────────────────────────────────────────────────`);
       logger.log(`Result(${t.tenantId}):`);

apps/pac-service/src/modules/plan/engine/plan-engine.service.ts

@@ -221,14 +221,25 @@ export class PlanEngineService {
     hostId: string;
     tenantId: string;
     now?: Date;
+    /// 可选:把本轮**收窄到给定患者子集**(按诊所补摄后的定向重算,见 recompute-plans --clinics)。
+    /// selectHits 与第 3 步关闭**同时**收窄 —— 两者必须同进同退,原因见第 3 步注释。
+    patientIds?: string[];
   }): Promise<EngineRunResult> {
     const startedAt = new Date();
     const now = input.now ?? new Date();
+    const scopedPatientIds = input.patientIds?.length ? new Set(input.patientIds) : null;
     const scope: ScenarioScope = {
       hostId: input.hostId,
       tenantId: input.tenantId,
       now,
+      ...(scopedPatientIds ? { patientIds: input.patientIds } : {}),
     };
+    if (scopedPatientIds) {
+      this.logger.log(
+        `▶ 子集模式:本轮只评估 ${scopedPatientIds.size} 位患者;` +
+          `关闭步骤同步收窄到同一子集(范围外的 plan 一律不动)。`,
+      );
+    }
 
     // 1. 各 scenario 跑 selector,汇总 hits
     const hitsByPatient = new Map<string, ScenarioHitWithKey[]>();
@@ -336,7 +347,12 @@ export class PlanEngineService {
       // 多取两列供记账用(assignedAt 清不清都要先算持有时长)
       select: { id: true, patientId: true, assigneeUserId: true, assignedAt: true },
     });
-    const staleRows = activePlans.filter((pl) => !hitsByPatient.has(pl.patientId));
+    //   ⭐ 子集模式(--clinics)必须**同步收窄**:本轮只评估了子集,子集外的患者天然 0 命中,
+    //     不收窄就会把整个 host 的召回池当成"信号全消失"清空,认领中的单还各记一条 auto_release。
+    //     用内存 Set 过滤而非 SQL `patientId: { in: ids }` —— 子集动辄数万,会撞 PG 32767 bind 上限。
+    const staleRows = activePlans.filter(
+      (pl) => !hitsByPatient.has(pl.patientId) && (!scopedPatientIds || scopedPatientIds.has(pl.patientId)),
+    );
     if (staleRows.length > 0) {
       // ⚠️ 必须分片:原实现是一条 `id: { in: staleIds }`,PG bind 变量上限 32767 —— 池子上了
       //   三万条(生产 44 万患者完全可能)就会直接报错。顺带让每片自成事务,

apps/pac-service/src/modules/plan/engine/scenario.interface.ts

@@ -21,6 +21,12 @@ export interface ScenarioScope {
   /// 可选:只评估单个 patient(详情页"刷新"单刷场景)。
   /// 设了 → selectHits SQL 加 `AND p.id = patientId`,从全租户扫降为单患者扫(O(1))。
   patientId?: string;
+  /// 可选:只评估给定患者集合(按诊所补摄后的定向重算,见 recompute-plans --clinics)。
+  /// 设了 → selectHits SQL 加 `AND p.id = ANY(ids)`,把全租户扫降为子集扫。
+  /// ⚠️ 与 patientId 互斥语义上不冲突(两个都设 = 交集),但调用方应只设其一。
+  /// ⚠️ **收窄了 selectHits 就必须同步收窄 runAllForHost 的关闭步骤** —— 否则范围外
+  ///    患者会因"本轮 0 命中"被误判信号消失而清空召回池。见 plan-engine.service.ts 第 3 步。
+  patientIds?: string[];
 }
 
 export interface ScenarioHit {

apps/pac-service/src/modules/plan/engine/scenarios/treatment-initiation-recall.scenario.ts

@@ -277,10 +277,24 @@ export class TreatmentInitiationRecallScenario implements PlanScenarioPlugin {
     const expectedCats = rule.categories as readonly string[];
     const resolverCats = resolverCategoriesFor(cfg.primaryCode) as readonly string[];
 
-    // 单 patient 收窄(详情页"刷新"):设了 scope.patientId → 只扫该患者,O(全租户)→O(1)
+    // 收窄(可空,两种粒度):
+    //   - scope.patientId  单患者(详情页"刷新"):O(全租户) → O(1)
+    //   - scope.patientIds 患者子集(按诊所补摄后定向重算,recompute-plans --clinics):
+    //       整个 id 数组是**一个** bind 参数(同 allCodes 的写法),不受 PG 32767 bind 上限影响。
+    //
+    // ⚠️ 子集这行**必须写成 `IN (SELECT … unnest(array))`,不能写 `= ANY(array)`** ——
+    //    两者结果等价,代价差一个数量级。`= ANY(数组常量)` 让规划器把它当成廉价的行过滤,
+    //    转去走嵌套循环 + 索引探查,而本 SQL 带 gap lateral join,每行代价很高;
+    //    `IN (SELECT …)` 是**半连接**,规划器会先把 id 集哈希掉再 join,和大表扫描的代价模型对得上。
+    //    2026-08-21 本地实测(30000 患者库,子集 5825 人,11 个子场景合计):
+    //      = ANY(array)          53.6s  ← 比全量 42.3s 还慢,其中 perio_no_srp 一个就 35.5s
+    //      IN (SELECT unnest())  10.2s  ← 命中数逐个相同,快 4.2 倍
+    //    改这行前先按上面的口径量一遍,别只看"看起来更简洁"。
     const patientFilter = scope.patientId
       ? Prisma.sql`AND p.id = ${scope.patientId}::uuid`
-      : Prisma.empty;
+      : scope.patientIds?.length
+        ? Prisma.sql`AND p.id IN (SELECT u FROM unnest(${scope.patientIds}::uuid[]) u)`
+        : Prisma.empty;
 
     // ⭐ gap 核心(sig 牙位 / resolved / remaining + ⑤a 判定 + 废用牙/先天剔除)抽到共享模块
     //   potential-treatment-gap.sql —— 召回与潜在治疗画像【单一真理源】,SQL 逻辑零改动只搬家。
@@ -379,7 +393,7 @@ export class TreatmentInitiationRecallScenario implements PlanScenarioPlugin {
       ${gap.lateralJoin}
       WHERE p.host_id = ${scope.hostId}::uuid              -- ① 隔离闸
         AND p.tenant_id = ${scope.tenantId}                -- ① 隔离闸
-        ${patientFilter}                                   -- 单刷收窄(可空)
+        ${patientFilter}                                   -- 单刷 / 子集收窄(可空)
         AND p.active = true                                -- ② 合规闸
         AND pp.do_not_contact = false                      -- ② 合规闸
         AND pp.deceased = false                            -- ② 合规闸