perf(plan): runAllForHost 批量化 — bulk 预取 + 分块并发 + 日志 createMany

夜间全 host plan 重算的 upsert 段原为逐患者串行(每人 2 读 latest/snoozed + persona + 2 日志写),13万患者 ~49min。改为:循环前一次性 bulk 预取 latest(含 reasons)/snooze 抑制集/active persona → 逐患复用同一 upsertPlan 判定(结果等价)→ 写操作分块并发(PAC_PLAN_BATCH_CONCURRENCY 默认 8)→ PlanGenerationLog 收集后 createMany(每患仍一行,口径不变)。selectHits 全表扫不在本次范围(时间规则可改,每轮须全量重选)。单刷 recomputeForPatient 不传 prefetched,行为不变。

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

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

@@ -183,46 +183,73 @@ export class PlanEngineService {
     let plansSuppressed = 0;
     let plansClosed = 0;
 
-    // 2. 逐 patient 写 plan(每个 patient 一条 PlanGenerationLog,跟 schema 设计一致)
-    for (const [patientId, hits] of hitsByPatient.entries()) {
-      const log = await this.prisma.planGenerationLog.create({
-        data: {
-          hostId: scope.hostId,
-          tenantId: scope.tenantId,
-          patientId,
-          triggeredBy: 'cli:batch',
-          status: 'running',
-          scenarioRunCount: this.scenarios.length,
-        },
-      });
-      try {
-        const result = await this.upsertPlan({ scope, patientId, hits });
-        if (result === 'created') plansCreated++;
-        else if (result === 'superseded') plansSuperseded++;
-        else if (result === 'unchanged') plansUnchanged++;
-        else if (result === 'suppressed') plansSuppressed++;
-
-        await this.prisma.planGenerationLog.update({
-          where: { id: log.id },
-          data: {
-            status: 'success',
-            plansCreated: result === 'created' || result === 'superseded' ? 1 : 0,
-            endedAt: new Date(),
-          },
-        });
-      } catch (err) {
-        await this.prisma.planGenerationLog.update({
-          where: { id: log.id },
-          data: {
-            status: 'failed',
-            errorMessage: err instanceof Error ? err.message : String(err),
-            endedAt: new Date(),
-          },
-        });
-        this.logger.error(
-          `plan upsert failed patient=${patientId}: ${err instanceof Error ? err.message : err}`,
-        );
-      }
+    // 2. 批量写 plan。
+    //   优化(2026-06):原"逐 patient 串行 + 每人 2 读(latest/snoozed)+ persona + 2 日志写",
+    //     全 host(~13万)上 upsert 段 ~49min。改为:循环前一次性 **bulk 预取** latest/snoozed/persona
+    //     → 逐患复用**同一** upsert 判定(结果等价)→ 写操作**分块并发** → PlanGenerationLog 收集后
+    //     一次性 createMany(每患仍一行,审计/监控口径不变,只是写法从 12.5万次 → 几次)。
+    //   注:selectHits 的全表扫不在本优化内(时间规则随时可改,每轮必须全量重选才正确)。
+    const patientIds = [...hitsByPatient.keys()];
+    const { latestByPatient, snoozedByPatient, personaByPatient } = await this.prefetchForBatch(
+      scope,
+      patientIds,
+      now,
+    );
+    const EMPTY_SNOOZE = new Set<string>();
+    const logRows: Prisma.PlanGenerationLogCreateManyInput[] = [];
+    const concurrency = Math.max(1, Number(process.env.PAC_PLAN_BATCH_CONCURRENCY) || 8);
+    const entries = [...hitsByPatient.entries()];
+    for (let i = 0; i < entries.length; i += concurrency) {
+      await Promise.all(
+        entries.slice(i, i + concurrency).map(async ([patientId, hits]) => {
+          try {
+            const result = await this.upsertPlan({
+              scope,
+              patientId,
+              hits,
+              prefetched: {
+                latest: latestByPatient.get(patientId) ?? null,
+                snoozedKeys: snoozedByPatient.get(patientId) ?? EMPTY_SNOOZE,
+                personaId: personaByPatient.get(patientId) ?? null,
+              },
+            });
+            // 计数:JS 单线程,await 恢复后同步自增,chunk 内并发无交错
+            if (result === 'created') plansCreated++;
+            else if (result === 'superseded') plansSuperseded++;
+            else if (result === 'unchanged') plansUnchanged++;
+            else if (result === 'suppressed') plansSuppressed++;
+            logRows.push({
+              hostId: scope.hostId,
+              tenantId: scope.tenantId,
+              patientId,
+              triggeredBy: 'cli:batch',
+              status: 'success',
+              scenarioRunCount: this.scenarios.length,
+              plansCreated: result === 'created' || result === 'superseded' ? 1 : 0,
+              startedAt,
+              endedAt: new Date(),
+            });
+          } catch (err) {
+            logRows.push({
+              hostId: scope.hostId,
+              tenantId: scope.tenantId,
+              patientId,
+              triggeredBy: 'cli:batch',
+              status: 'failed',
+              scenarioRunCount: this.scenarios.length,
+              errorMessage: err instanceof Error ? err.message : String(err),
+              startedAt,
+              endedAt: new Date(),
+            });
+            this.logger.error(
+              `plan upsert failed patient=${patientId}: ${err instanceof Error ? err.message : err}`,
+            );
+          }
+        }),
+      );
+    }
+    for (let i = 0; i < logRows.length; i += 5000) {
+      await this.prisma.planGenerationLog.createMany({ data: logRows.slice(i, i + 5000) });
     }
 
     // 3. ⭐ 缺口1 修复:关闭"有 active plan 但本轮 0 命中"的患者的遗留 plan。
@@ -265,18 +292,28 @@ export class PlanEngineService {
   }
 
   // ── plan upsert(单 patient,事务内 supersede + insert)──
+  // prefetched:批量路径(runAllForHost)循环前一次性 bulk 取好 latest/snoozed/persona,
+  //   逐患者复用**同一**判定逻辑但不再每人发 3 条读查询(结果等价,仅省往返)。
+  //   单刷路径(recomputeForPatient)不传 → 照旧逐查,行为不变。
   private async upsertPlan(input: {
     scope: ScenarioScope;
     patientId: string;
     hits: ScenarioHitWithKey[];
+    prefetched?: {
+      latest: PlanWithReasons | null;
+      snoozedKeys: Set<string>;
+      personaId: string | null;
+    };
   }): Promise<'created' | 'superseded' | 'unchanged' | 'suppressed'> {
-    const { scope, patientId, hits } = input;
+    const { scope, patientId, hits, prefetched } = input;
 
-    const latest = await this.prisma.followupPlan.findFirst({
-      where: { hostId: scope.hostId, tenantId: scope.tenantId, patientId },
-      orderBy: { version: 'desc' },
-      include: { reasons: true },
-    });
+    const latest = prefetched
+      ? prefetched.latest
+      : await this.prisma.followupPlan.findFirst({
+          where: { hostId: scope.hostId, tenantId: scope.tenantId, patientId },
+          orderBy: { version: 'desc' },
+          include: { reasons: true },
+        });
 
     // ⭐ 设计决策(2026-06):assigned 不再整体 skip。
     //   理由:客服"正在执行"恰好撞上"正在重算"概率极低,为这点概率冻结内容、让客服拿过时理由白打,
@@ -290,7 +327,9 @@ export class PlanEngineService {
     //         并集作抑制集,逐 hit 过滤掉命中抑制集的信号;剩下的(全新诊断)继续走下面生成逻辑。
     //   修复的 bug:旧版"整患者抑制" → 成功转化 K08 后,snooze 期内新长的 K02 也被压住不召(误伤新缺口)。
     //   多次结案也正确:查的是全部未到期终态 plan 的并集(不止 latest 一条)。
-    const snoozedKeys = await this.fetchSnoozedSignalKeys(scope, patientId, scope.now);
+    const snoozedKeys = prefetched
+      ? prefetched.snoozedKeys
+      : await this.fetchSnoozedSignalKeys(scope, patientId, scope.now);
     const usableHits =
       snoozedKeys.size === 0
         ? hits
@@ -335,12 +374,16 @@ export class PlanEngineService {
     // head = 优先级最高的 hit(用于 plan 顶层字段:goal / recommendedRole / recommendedAt / recommendedChannel)
     const head = [...usableHits].sort((a, b) => b.priorityScore - a.priorityScore)[0]!;
 
-    // 拉对应 patient 的最新 active persona(可空)
-    const persona = await this.prisma.persona.findFirst({
-      where: { patientId, supersededAt: null },
-      orderBy: { version: 'desc' },
-      select: { id: true },
-    });
+    // 拉对应 patient 的最新 active persona(可空);批量路径用预取值,单刷照旧查
+    const personaId = prefetched
+      ? prefetched.personaId
+      : (
+          await this.prisma.persona.findFirst({
+            where: { patientId, supersededAt: null },
+            orderBy: { version: 'desc' },
+            select: { id: true },
+          })
+        )?.id ?? null;
 
     await this.prisma.$transaction(async (tx) => {
       if (latest && (latest.status === 'active' || latest.status === 'assigned')) {
@@ -354,7 +397,7 @@ export class PlanEngineService {
           hostId: scope.hostId,
           tenantId: scope.tenantId,
           patientId,
-          personaId: persona?.id ?? null,
+          personaId: personaId ?? null,
           targetClinicId: head.targetClinicId ?? null,
           version: nextVersion,
           priorityScore: newPriorityScore,
@@ -437,6 +480,68 @@ export class PlanEngineService {
     return keys;
   }
 
+  /**
+   * 批量预取 —— runAllForHost 循环前一次性把每患者的 latest plan(含 reasons)/ snooze 抑制集 /
+   * active persona id 取好,避免循环内每人发 3 条读查询(等价,仅省往返)。
+   * 取数口径与 upsertPlan(findFirst orderBy version desc)/ fetchSnoozedSignalKeys(终态+未到期)
+   * 严格对齐;IN 列表分块防参数超限。
+   */
+  private async prefetchForBatch(
+    scope: ScenarioScope,
+    patientIds: string[],
+    now: Date,
+  ): Promise<{
+    latestByPatient: Map<string, PlanWithReasons>;
+    snoozedByPatient: Map<string, Set<string>>;
+    personaByPatient: Map<string, string>;
+  }> {
+    const latestByPatient = new Map<string, PlanWithReasons>();
+    const snoozedByPatient = new Map<string, Set<string>>();
+    const personaByPatient = new Map<string, string>();
+    const CHUNK = 2000;
+    for (let i = 0; i < patientIds.length; i += CHUNK) {
+      const ids = patientIds.slice(i, i + CHUNK);
+      // latest plan(含 reasons)— 每患者最高 version 那条(对齐 upsertPlan 的 orderBy version desc)
+      const plans = await this.prisma.followupPlan.findMany({
+        where: { hostId: scope.hostId, tenantId: scope.tenantId, patientId: { in: ids } },
+        include: { reasons: true },
+        orderBy: [{ patientId: 'asc' }, { version: 'desc' }],
+      });
+      for (const p of plans) {
+        if (p.patientId && !latestByPatient.has(p.patientId)) latestByPatient.set(p.patientId, p);
+      }
+      // snooze 抑制集(对齐 fetchSnoozedSignalKeys:终态 + 冷静期未到期 plan 的 reason 并集)
+      const terminal = await this.prisma.followupPlan.findMany({
+        where: {
+          hostId: scope.hostId,
+          tenantId: scope.tenantId,
+          patientId: { in: ids },
+          status: { in: ['completed', 'abandoned'] },
+          snoozedUntil: { gt: now },
+        },
+        select: { patientId: true, reasons: { select: { scenario: true, subKey: true } } },
+      });
+      for (const t of terminal) {
+        if (!t.patientId) continue;
+        const set = snoozedByPatient.get(t.patientId) ?? new Set<string>();
+        for (const r of t.reasons) set.add(`${r.scenario}|${r.subKey ?? ''}`);
+        snoozedByPatient.set(t.patientId, set);
+      }
+      // active persona id(每患最新 active 版本)
+      const personas = await this.prisma.persona.findMany({
+        where: { patientId: { in: ids }, supersededAt: null },
+        orderBy: [{ patientId: 'asc' }, { version: 'desc' }],
+        select: { id: true, patientId: true },
+      });
+      for (const pe of personas) {
+        if (pe.patientId && !personaByPatient.has(pe.patientId)) {
+          personaByPatient.set(pe.patientId, pe.id);
+        }
+      }
+    }
+    return { latestByPatient, snoozedByPatient, personaByPatient };
+  }
+
   /// 子场景 → 生命周期类型(one_shot 短效 / recurring 长效)
   private lifecycleFor(scenario: string, subKey?: string): string {
     // 复查类长效(种植年度 / 牙周维护),其余短效
@@ -446,6 +551,8 @@ export class PlanEngineService {
 }
 
 type ScenarioHitWithKey = ScenarioHit & { scenarioKey: string };
+/// 批量预取/upsert 用:plan 含 reasons(对齐 upsertPlan 的 include:{reasons:true})
+type PlanWithReasons = Prisma.FollowupPlanGetPayload<{ include: { reasons: true } }>;
 
 export interface EngineRunResult {
   scenariosRun: number;