fix(persona): getMonetaryQuantiles 加 single-flight 防并发惊群卡死

并发重算(--concurrency>1)时发现:缓存空/过期瞬间,N 个 recompute 同时调
getMonetaryQuantiles → 同时发 N 份全租户聚合 $queryRaw(370万 facts 实测单跑 4.1s)
→ 打爆 4 核 DB + 连接池 → 进程卡死(stat=S,0 吞吐)。串行版无事因缓存命中只跑 1 次;
且缓存 30min 过期,长批(~1.6h)半路会再次惊群。

修:single-flight —— 缓存 miss 时把在飞 Promise 存 map,并发调用复用同一个查询,
只跑 1 份。等待者 await JS promise(不占 DB 连接)。根治,也利好增量 cohort 路径。

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

apps/pac-service/src/modules/persona/persona.service.ts

@@ -34,6 +34,8 @@ export class PersonaService {
   /// RFM 的 M 打分要群体分位(图 B.1.1),但 PAC 按患者重算 → 分位阈值缓存复用(分布慢变);
   /// 批量重算每轮首个患者算一次,后续命中缓存;TTL 兜底单刷场景。
   private readonly mQuantileCache = new Map<string, { at: number; q: number[] }>();
+  /** single-flight:缓存空/过期时,并发调用共享同一个在飞查询(防惊群)。 */
+  private readonly mQuantileInflight = new Map<string, Promise<number[]>>();
   private static readonly M_QUANTILE_TTL_MS = 30 * 60 * 1000;
 
   constructor(
@@ -42,31 +44,45 @@ export class PersonaService {
     private readonly gapSelector: PotentialTreatmentSelector,
   ) {}
 
-  /** 租户内"累计净消费/患者"的 [p20,p40,p60,p80](cents)。缓存 30min;失败/无数据 → []。 */
+  /**
+   * 租户内"累计净消费/患者"的 [p20,p40,p60,p80](cents)。缓存 30min;失败/无数据 → []。
+   *
+   * 该查询全租户聚合(大库可达数秒)。**single-flight**:缓存空/过期时,并发调用(批量重算
+   * concurrency>1)共享同一个在飞查询 —— 否则 N 个重查询同时打 DB 会卡死(370万 facts 实测)。
+   */
   private async getMonetaryQuantiles(hostId: string, tenantId: string, nowMs: number): Promise<number[]> {
     const key = `${hostId}:${tenantId}`;
     const hit = this.mQuantileCache.get(key);
     if (hit && nowMs - hit.at < PersonaService.M_QUANTILE_TTL_MS) return hit.q;
-    try {
-      const rows = await this.prisma.$queryRaw<Array<{ q: number[] }>>`
-        WITH spend AS (
-          SELECT patient_id, SUM(
-            CASE type WHEN 'payment_record'  THEN (content->>'amount_cents')::bigint
-                      WHEN 'recharge_record' THEN (content->>'amount_cents')::bigint
-                      WHEN 'refund_record'   THEN -(content->>'amount_cents')::bigint END) AS net
-          FROM patient_facts
-          WHERE host_id = ${hostId}::uuid AND tenant_id = ${tenantId}
-            AND status IN ('active','fulfilled')
-            AND type IN ('payment_record','recharge_record','refund_record')
-          GROUP BY patient_id)
-        SELECT percentile_cont(ARRAY[0.2,0.4,0.6,0.8]) WITHIN GROUP (ORDER BY net)::float8[] AS q FROM spend`;
-      const q = (rows[0]?.q ?? []).map((x) => Number(x));
-      this.mQuantileCache.set(key, { at: nowMs, q });
-      return q;
-    } catch (err) {
-      this.logger.warn(`getMonetaryQuantiles failed (${key}): ${err instanceof Error ? err.message : err}`);
-      return [];
-    }
+    const inflight = this.mQuantileInflight.get(key);
+    if (inflight) return inflight; // 已有查询在跑 → 复用,不再发一份
+
+    const promise = (async () => {
+      try {
+        const rows = await this.prisma.$queryRaw<Array<{ q: number[] }>>`
+          WITH spend AS (
+            SELECT patient_id, SUM(
+              CASE type WHEN 'payment_record'  THEN (content->>'amount_cents')::bigint
+                        WHEN 'recharge_record' THEN (content->>'amount_cents')::bigint
+                        WHEN 'refund_record'   THEN -(content->>'amount_cents')::bigint END) AS net
+            FROM patient_facts
+            WHERE host_id = ${hostId}::uuid AND tenant_id = ${tenantId}
+              AND status IN ('active','fulfilled')
+              AND type IN ('payment_record','recharge_record','refund_record')
+            GROUP BY patient_id)
+          SELECT percentile_cont(ARRAY[0.2,0.4,0.6,0.8]) WITHIN GROUP (ORDER BY net)::float8[] AS q FROM spend`;
+        const q = (rows[0]?.q ?? []).map((x) => Number(x));
+        this.mQuantileCache.set(key, { at: nowMs, q });
+        return q;
+      } catch (err) {
+        this.logger.warn(`getMonetaryQuantiles failed (${key}): ${err instanceof Error ? err.message : err}`);
+        return [];
+      } finally {
+        this.mQuantileInflight.delete(key);
+      }
+    })();
+    this.mQuantileInflight.set(key, promise);
+    return promise;
   }
 
   /**