merge: 落库改 unnest,甩掉 PG bind 上限 → test

单批上限 500 → 20000。事务边界一字未动(仍是一个事务、三个并发守卫、RETURNING 照旧),
超时改成随条数放宽(30 秒打底 + 每条 3ms,上限 180 秒)。

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

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

@@ -196,7 +196,12 @@ export class PlanAssignmentService {
     if (existing) return existing;
 
     if (dto.items.length > ASSIGNMENT_ITEMS_HARD_LIMIT) {
-      throw new BadRequestException(`单次分配最多 ${ASSIGNMENT_ITEMS_HARD_LIMIT} 条`);
+      // ⚠️ 与 schema 那条同一句口径 —— 说清他能动哪个旋钮,别让他找"本批人数"那个不存在的输入框
+      throw new BadRequestException(
+        `一次最多分 ${ASSIGNMENT_ITEMS_HARD_LIMIT.toLocaleString('zh-CN')} 条,` +
+          `这一批 ${dto.items.length.toLocaleString('zh-CN')} 条超了。` +
+          `把时限改短、或者把每天几通调小,本批人数会跟着降下来。`,
+      );
     }
 
     // ── 闸 3:scope 绑定 ─────────────────────────────────────
@@ -325,34 +330,63 @@ export class PlanAssignmentService {
             select: { id: true, expiresAt: true },
           });
 
-          // ⭐ **一条 SQL 打完全部 500 行**,而不是按客服分桶发多条 updateMany。
-          //   换法的原因是快照列:priority_score_at_assign / dedicated_cs_at_assign 等
-          //   **逐行不同**,updateMany 的 data 是全桶共用的,表达不了。
-          //   逐行 update 则是 500 次往返。VALUES join 两个问题一起解决,
-          //   而且 RETURNING 直接给出**哪些行真的落上了**,不必再回查一次去猜差额。
-          //   bind 变量:每行 9 个 × 500 = 4,500,远低于 PG 上限 32,767。
-          //
-          // ⚠️ 走原生 SQL 的两个代价,都必须手工补上:
-          //   ① Prisma 的 `@updatedAt` **不会触发** → 显式 SET updated_at
-          //   ② where 条件要自己写全 —— 这里的 `status='active' AND assignee IS NULL
-          //      AND superseded_at IS NULL` 就是并发安全的全部依据:
-          //      从上面 findMany 到这一刻若有人抢先认领,该行不满足条件、不被更新,
-          //      RETURNING 里也就没有它。
-          const values = applicable.map(({ row, item }) => {
-            const perItemExpires =
+          /**
+           * ⭐ **一条 SQL 打完全部行**,而不是按客服分桶发多条 updateMany。
+           *   换法的原因是快照列:priority_score_at_assign / dedicated_cs_at_assign 等
+           *   **逐行不同**,updateMany 的 data 是全桶共用的,表达不了。
+           *   逐行 update 则是 N 次往返。而 RETURNING 直接给出**哪些行真的落上了**,
+           *   不必再回查一次去猜差额。
+           *
+           * ═══ 为什么是 `unnest(数组)` 而不是 `VALUES (…),(…),…`(2026-08-20 改)═════
+           * 🔴 `VALUES` 那种写法**每行吃 9 个 bind 变量**,而 PG 单条语句上限 32,767 ——
+           *   `32767 / 9 ≈ 3,640 行`就是一堵硬墙。测试机实测 N=4000 直接报
+           *   `too many bind variables in prepared statement, expected maximum of 32767,
+           *    received 36000`,与业务无关,纯粹是写法撞的。
+           * ✅ 换成**每列一个数组**:不管多少行,**永远 9 个 bind**。墙消失。
+           * ⭐ 顺带更快 —— PG 不用再解析几千个占位符。测试机实测(事务内跑完回滚):
+           *      N=500    VALUES 444ms → unnest  74ms
+           *      N=2000   VALUES 458ms → unnest 220ms
+           *      N=4000   VALUES ❌炸  → unnest 433ms
+           *   忠实形状(13 列全写 + 同事务账本写入)线性到 5 万条:
+           *      500→0.25s  2千→0.6s  5千→1.4s  1万→2.7s  2万→5.6s  5万→13.6s
+           * ⚠️ 数组里**混 null 是安全的**(测试机验过:null 原样落到对应列),
+           *   空数组也不炸 —— 但上面 `applicable.length === 0` 已经先挡掉了。
+           *
+           * ⚠️ 走原生 SQL 的两个代价,都必须手工补上:
+           *   ① Prisma 的 `@updatedAt` **不会触发** → 显式 SET updated_at
+           *   ② where 条件要自己写全 —— 这里的 `status='active' AND assignee IS NULL
+           *      AND superseded_at IS NULL` 就是并发安全的全部依据:
+           *      从上面 findMany 到这一刻若有人抢先认领,该行不满足条件、不被更新,
+           *      RETURNING 里也就没有它。
+           * ⛔ **改这段时别把九个数组拆成不等长的** —— unnest 会按最长的那个补 null,
+           *   短的那几列就整段错位,而且**不报错**。它们都从同一个 `applicable` 循环里推,
+           *   保持这一点就不会错位。
+           */
+          const aPlanId: string[] = [];
+          const aAssignee: string[] = [];
+          const aExpires: Date[] = [];
+          const aStrategy: string[] = [];
+          const aDedicatedCs: (string | null)[] = [];
+          const aDedicatedCsLastVisit: (Date | null)[] = [];
+          const aPriorityScore: number[] = [];
+          const aSourceConfidence: (number | null)[] = [];
+          const aSelectionMode: string[] = [];
+          for (const { row, item } of applicable) {
+            const dcs = dedicatedByPatient.get(row.patientId) ?? null;
+            aPlanId.push(row.id);
+            aAssignee.push(item.assigneeUserId);
+            aExpires.push(
               item.expiresInDays != null
                 ? endOfDayInHostTimezone(now, item.expiresInDays, tz)
-                : head.expiresAt;
-            const dcs = dedicatedByPatient.get(row.patientId) ?? null;
-            return Prisma.sql`(
-              ${row.id}::uuid, ${item.assigneeUserId}::text, ${perItemExpires}::timestamptz,
-              ${item.assignStrategy}::text, ${dcs}::text,
-              ${dcs ? (lastVisitByCs.get(dcs) ?? null) : null}::timestamptz,
-              ${row.priorityScore}::double precision,
-              ${confidenceByPlan.get(row.id) ?? null}::double precision,
-              ${item.selectionMode ?? 'rank'}::text
-            )`;
-          });
+                : head.expiresAt,
+            );
+            aStrategy.push(item.assignStrategy);
+            aDedicatedCs.push(dcs);
+            aDedicatedCsLastVisit.push(dcs ? (lastVisitByCs.get(dcs) ?? null) : null);
+            aPriorityScore.push(row.priorityScore);
+            aSourceConfidence.push(confidenceByPlan.get(row.id) ?? null);
+            aSelectionMode.push(item.selectionMode ?? 'rank');
+          }
 
           const appliedRows = await tx.$queryRaw<Array<{ id: string }>>(Prisma.sql`
             UPDATE followup_plans AS fp SET
@@ -371,11 +405,23 @@ export class PlanAssignmentService {
               source_confidence_at_assign = v.source_confidence,
               selection_mode              = v.selection_mode,
               updated_at                  = ${now}::timestamptz
-            FROM (VALUES ${Prisma.join(values)}) AS v(
-              plan_id, assignee, expires_at, strategy,
-              dedicated_cs, dedicated_cs_last_visit,
-              priority_score, source_confidence, selection_mode
-            )
+            FROM (
+              SELECT * FROM unnest(
+                ${aPlanId}::uuid[],
+                ${aAssignee}::text[],
+                ${aExpires}::timestamptz[],
+                ${aStrategy}::text[],
+                ${aDedicatedCs}::text[],
+                ${aDedicatedCsLastVisit}::timestamptz[],
+                ${aPriorityScore}::double precision[],
+                ${aSourceConfidence}::double precision[],
+                ${aSelectionMode}::text[]
+              ) AS t(
+                plan_id, assignee, expires_at, strategy,
+                dedicated_cs, dedicated_cs_last_visit,
+                priority_score, source_confidence, selection_mode
+              )
+            ) AS v
             WHERE fp.id = v.plan_id
               AND fp.status = 'active'
               AND fp.assignee_user_id IS NULL
@@ -426,8 +472,20 @@ export class PlanAssignmentService {
 
           return { head, applied: applied.length, lost };
         },
-        // 500 条 × (updateMany + 事件) 的余量;maxWait 是等连接池的时间,不是执行时间
-        { maxWait: 10_000, timeout: 30_000 },
+        /**
+         * ⚠️ `maxWait` 是**等连接池**的时间,不是执行时间;`timeout` 才是这一整个事务的墙。
+         *
+         * 🔴 **超时随条数放宽,⛔ 不是一刀切 30 秒**(2026-08-20):
+         *   实测约 0.27ms/条(UPDATE + 同事务账本写入),2 万条 5.6 秒。
+         *   一刀切 30 秒的话,小批次白留 100 倍余量、大批次又不够。
+         *   这里给「30 秒打底 + 每条 3ms」,2 万条 = 90 秒,对实测留 16 倍余量
+         *   (生产是托管 RDS,比测试机多一跳,余量要留够);上限 180 秒兜住。
+         * ⛔ 别把它调成无限:事务开着就一直**持有这些行的锁**,卡住的时候要能自己断。
+         */
+        {
+          maxWait: 10_000,
+          timeout: Math.min(180_000, 30_000 + applicable.length * 3),
+        },
       );
 
       for (const id of result.lost) skipped.push({ planId: id, reason: 'claimed_by_other' });

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

@@ -199,7 +199,9 @@ export class PlanRearrangeService {
     const out = new Map<string, { label: string | null; temperature: string | null }>();
     if (planIds.length === 0) return out;
 
-    const filter = Prisma.sql`fp.id IN (${Prisma.join(planIds.map((i) => Prisma.sql`${i}::uuid`))})`;
+    // ⚠️ `= ANY(数组)` 而不是 `IN (…)`:后者每个 id 占一个 bind,32,767 就是墙;
+    //   数组写法永远 1 个 bind,而且更快(实测 3 万个 id:IN 521ms → ANY 219ms)。
+    const filter = Prisma.sql`fp.id = ANY(${planIds}::uuid[])`;
     const bucket = temperatureBucketCaseSql(
       Prisma.sql`max(la.hot_until)`,
       Prisma.sql`max(la.warm_until)`,
@@ -386,15 +388,20 @@ export class PlanRearrangeService {
       //   到这一刻若有人抢先改派/退回,该行不满足条件、不被更新,RETURNING 里就没有它。
       let movedIds: string[] = [];
       if (moves.length > 0) {
-        const values = moves.map(
-          ({ row, to }) =>
-            Prisma.sql`(${row.id}::uuid, ${row.assigneeUserId!}::text, ${to}::text)`,
-        );
+        // ⚠️ `unnest(数组)` 而不是 `VALUES (…),(…)` —— 后者每行吃 3 个 bind,行一多就撞
+        //   PG 的 32,767 上限(分配那边实测 N=4000 直接炸,沿革见 plan-assignment 的红字)。
+        //   数组写法不管多少行永远 3 个 bind,而且更快。⛔ 三个数组必须等长,见那边说明。
+        const mPlanId = moves.map(({ row }) => row.id);
+        const mFrom = moves.map(({ row }) => row.assigneeUserId!);
+        const mTo = moves.map(({ to }) => to);
         const applied = await tx.$queryRaw<Array<{ id: string }>>(Prisma.sql`
           UPDATE followup_plans AS fp SET
             assignee_user_id = v.to_user,
             updated_at       = ${now}::timestamptz
-          FROM (VALUES ${Prisma.join(values)}) AS v(plan_id, from_user, to_user)
+          FROM (
+            SELECT * FROM unnest(${mPlanId}::uuid[], ${mFrom}::text[], ${mTo}::text[])
+              AS t(plan_id, from_user, to_user)
+          ) AS v
           WHERE fp.id = v.plan_id
             AND fp.status = 'assigned'
             AND fp.superseded_at IS NULL
@@ -424,15 +431,16 @@ export class PlanRearrangeService {
       // ── ② 改时限 ──────────────────────────────────────────
       let expiredIds: string[] = [];
       if (expiries.length > 0) {
-        const values = expiries.map(
-          ({ row, days }) =>
-            Prisma.sql`(${row.id}::uuid, ${endOfDayInHostTimezone(now, days, tz)}::timestamptz)`,
-        );
+        const ePlanId = expiries.map(({ row }) => row.id);
+        const eExpires = expiries.map(({ days }) => endOfDayInHostTimezone(now, days, tz));
         const applied = await tx.$queryRaw<Array<{ id: string }>>(Prisma.sql`
           UPDATE followup_plans AS fp SET
             assignment_expires_at = v.expires_at,
             updated_at            = ${now}::timestamptz
-          FROM (VALUES ${Prisma.join(values)}) AS v(plan_id, expires_at)
+          FROM (
+            SELECT * FROM unnest(${ePlanId}::uuid[], ${eExpires}::timestamptz[])
+              AS t(plan_id, expires_at)
+          ) AS v
           WHERE fp.id = v.plan_id
             AND fp.status = 'assigned'
             AND fp.superseded_at IS NULL
@@ -463,9 +471,8 @@ export class PlanRearrangeService {
       // ── ③ 移出(收回池子)──────────────────────────────────
       let removedIds: string[] = [];
       if (removes.length > 0) {
-        const values = removes.map(
-          (row) => Prisma.sql`(${row.id}::uuid, ${row.assigneeUserId!}::text)`,
-        );
+        const rPlanId = removes.map((row) => row.id);
+        const rFrom = removes.map((row) => row.assigneeUserId!);
         /**
          * 清空哪些列与客服退回(`PlanService.recycle`)**逐列对齐**:
          *   ⛔ 不清 assignment_id / assigned_by / assign_strategy —— 那是"这单来自哪批"
@@ -482,7 +489,10 @@ export class PlanRearrangeService {
             recycle_at            = NULL,
             assignment_expires_at = NULL,
             updated_at            = ${now}::timestamptz
-          FROM (VALUES ${Prisma.join(values)}) AS v(plan_id, from_user)
+          FROM (
+            SELECT * FROM unnest(${rPlanId}::uuid[], ${rFrom}::text[])
+              AS t(plan_id, from_user)
+          ) AS v
           WHERE fp.id = v.plan_id
             AND fp.status = 'assigned'
             AND fp.superseded_at IS NULL

apps/pac-service/tests/batch-write-bind-limit.spec.ts

+import { readFileSync } from 'node:fs';
+import { join } from 'node:path';
+import { ASSIGNMENT_ITEMS_HARD_LIMIT } from '@pac/types';
+
+/**
+ * 批量落库的**写法**回归 —— 这一类失败全是"到了某个条数才炸",本地小数据永远测不出来。
+ *
+ * ═══ 事故(2026-08-20 生产实遇)═══════════════════════════════════════
+ * 杭州大厦 23 位客服,按默认值 `23 × 15 通 × 3 天 = 1035`,提案出了 537 条的确认单,
+ * 主管点确认收到一句「请求字段校验失败」,在卡片上改什么都没用。
+ * 根因不是业务,是**写法**:落库那条 UPDATE 用 `VALUES (…),(…),…`,每行 9 个 bind,
+ * 而 PG 单条语句上限 32,767 → `32767/9 ≈ 3,640 行`是一堵硬墙。
+ * 测试机实测 N=4000 直接报
+ *   `too many bind variables in prepared statement, expected maximum of 32767, received 36000`
+ *
+ * ⇒ 改成 `unnest(数组)`:每列一个数组,**不管多少行永远 9 个 bind**,墙消失,而且更快
+ *   (N=500 444ms → 74ms;N=2000 458ms → 220ms)。
+ *
+ * 这几条断言守的就是"别改回去"。
+ */
+const BIND_MAX = 32_767;
+/** 旧写法每行 9 个 bind,能撑到的最大行数 —— 新上限必须**明显超过**它,否则等于没改 */
+const OLD_WALL = Math.floor(BIND_MAX / 9);
+
+const read = (f: string) => readFileSync(join(__dirname, `../src/modules/plan/${f}`), 'utf8');
+const ASSIGN = read('plan-assignment.service.ts');
+const REARRANGE = read('plan-rearrange.service.ts');
+
+/** 去注释 —— 上面那几段红字里当然会出现 `VALUES`,不剥掉断言会被自己的说明打败 */
+const strip = (x: string) =>
+  x.replace(/\/\*[\s\S]*?\*\//g, '').replace(/(^|[^:])\/\/[^\n]*/g, '$1');
+
+describe('批量落库必须用 unnest(数组),⛔ 不能回到逐行 VALUES', () => {
+  test('🔴 分配落库用 unnest,且⛔ 没有逐行 VALUES', () => {
+    const code = strip(ASSIGN);
+    expect(code).toMatch(/FROM \(\s*SELECT \* FROM unnest\(/);
+    // 逐行 VALUES 的特征:`VALUES ${Prisma.join(...)}`
+    expect(code).not.toMatch(/VALUES \$\{Prisma\.join/);
+  });
+
+  test('🔴 调整在手的三条写路径也都用 unnest', () => {
+    const code = strip(REARRANGE);
+    // 改派 / 改时限 / 移出
+    expect((code.match(/FROM unnest\(/g) ?? []).length).toBeGreaterThanOrEqual(3);
+    expect(code).not.toMatch(/VALUES \$\{Prisma\.join/);
+  });
+
+  test('⛔ 大 id 列表不许用 `IN (Prisma.join(...))` —— 每个 id 占一个 bind', () => {
+    // `= ANY($1::uuid[])` 永远 1 个 bind,而且更快(3 万个 id:IN 521ms → ANY 219ms)
+    expect(strip(REARRANGE)).not.toMatch(/IN \(\$\{Prisma\.join/);
+  });
+
+  test('⭐ 上限必须**越过**旧写法那堵墙,否则等于没解决', () => {
+    expect(OLD_WALL).toBe(3640); // 32767 / 9,钉住这个推导本身
+    expect(ASSIGNMENT_ITEMS_HARD_LIMIT).toBeGreaterThan(OLD_WALL);
+  });
+
+  test('⭐ 事务超时随条数放宽 —— ⛔ 不是一刀切', () => {
+    // 一刀切的话:小批次白留 100 倍余量,大批次又不够
+    expect(strip(ASSIGN)).toMatch(/timeout: Math\.min\(180_000, 30_000 \+ applicable\.length \* 3\)/);
+  });
+
+  test('⭐ 报错文案要说清他能动哪个旋钮(他手上没有"本批人数"这个输入框)', () => {
+    expect(ASSIGN).toMatch(/把时限改短、或者把每天几通调小/);
+  });
+});

apps/pac-service/tests/plan-rearrange.spec.ts

@@ -58,8 +58,14 @@ function makeService(opts: { rows: Row[]; alreadyDone?: boolean; applied?: 'all'
     const text = (sql.strings ?? []).join(' ');
     statements.push(text);
     if (!applyAll) return [];
-    // 从 VALUES 里把 plan_id 挑出来(每组的第一个值就是 plan_id)
-    const ids = (sql.values ?? []).filter(
+    /**
+     * 把 plan_id 从 bind 参数里挑出来。
+     * ⚠️ 2026-08-20 起写法是 `unnest(数组)`(为了甩掉 PG 的 32,767 bind 上限),
+     *   所以 `sql.values` 里是**数组**不是散值 —— 这里要先摊平一层,
+     *   ⛔ 不摊平的话这个假 DB 永远返回空,而真实现是对的(测试假绿的反面:假红)。
+     */
+    const flat = (sql.values ?? []).flatMap((v) => (Array.isArray(v) ? v : [v]));
+    const ids = flat.filter(
       (v): v is string => typeof v === 'string' && opts.rows.some((r) => r.id === v),
     );
     return [...new Set(ids)].map((id) => ({ id }));

packages/types/src/schemas/plan-assignment.ts

@@ -41,11 +41,30 @@ export const AssignmentItemSchema = z.object({
 export type AssignmentItem = z.infer<typeof AssignmentItemSchema>;
 
 /**
- * ⚠️ 500 是**技术护栏**(单事务 + PG bind 变量上限),不是批次规模的业务上限。
- * 业务上限该由产品定(教条七·待确认),定了之后加在**助手圈人那一层**,别写死在这里 ——
+ * ⚠️ 这是**技术护栏**,不是批次规模的业务上限。
+ * 业务上限该由产品定,定了之后加在**助手圈人那一层**,别写死在这里 ——
  * 写在这里会让"技术能不能扛"和"运营该不该分这么多"两件事永远分不开。
+ *
+ * ═══ 2026-08-20:500 → 20,000,依据是实测不是拍脑袋 ═══════════════════
+ * 旧的 500 卡的是**写法**不是能力:落库那条 UPDATE 原来用 `VALUES (…),(…),…`,
+ * 每行吃 9 个 bind,而 PG 单条语句上限 32,767 → `32767/9 ≈ 3,640 行`是硬墙。
+ * 改成 `unnest(数组)` 之后**不管多少行永远 9 个 bind**,墙没了(见 create() 的红字)。
+ *
+ * 测试机实测(事务内跑完回滚,13 列全写 + 同事务账本写入):
+ *   500 → 0.25s   2,000 → 0.6s   5,000 → 1.4s
+ *   10,000 → 2.7s  20,000 → 5.6s  50,000 → 13.6s   —— 线性,约 0.27ms/条
+ * 取 20,000:5.6s 对事务超时留了 5 倍余量,而生产是托管 RDS(多一跳),留够。
+ *
+ * 🔴 **⛔ 别再往上调之前先量这三样**,它们是下一批墙、且都还没量过:
+ *   ① 事务外那几个 `id: { in: [...] }` 读查询 —— Prisma 逐个元素占 bind,
+ *      ~32,000 个就炸(实测 30,000 个 IN 要 521ms,还能跑但已接近)。
+ *   ② 确认单**推给浏览器**的载荷 —— 实测一条 ProposedItem = 329 字节(含姓名/病历号/医生),
+ *      20,000 条 ≈ 6.3MB。它走侧信道(SSE)、不受请求体上限管,但浏览器要一口吃下。
+ *      ⚠️ 提交回来的 `items` 是精简形状(只有 planId/承接人/策略/入选方式)≈130 字节,
+ *        20,000 条 ≈ 2.6MB,在 10MB 请求体上限内 —— **这两个数别搞混**。
+ *   ③ 卡片渲染 20,000 行(默认折叠,展开单个客服才铺开,但没实测过)。
  */
-export const ASSIGNMENT_ITEMS_HARD_LIMIT = 500;
+export const ASSIGNMENT_ITEMS_HARD_LIMIT = 20_000;
 
 export const CreateAssignmentRequestSchema = z.object({
   /**
@@ -73,8 +92,14 @@ export const CreateAssignmentRequestSchema = z.object({
     .min(1)
     /// ⚠️ 消息要带数字:zod 的默认文案会被 all-exceptions.filter 包成一句笼统的
     ///   「请求字段校验失败」,主管看不出是哪一项超了、上限是多少(2026-08-20 生产实遇)。
+    /**
+     * ⚠️ 文案要说清**他能动哪个旋钮**。他手上只有两个(时限、每天几通),
+     *   人数是这两个乘出来的 —— 只说「请减少本批人数」等于让他找一个不存在的输入框。
+     */
     .max(ASSIGNMENT_ITEMS_HARD_LIMIT, {
-      message: `单次分配最多 ${ASSIGNMENT_ITEMS_HARD_LIMIT} 条,请减少本批人数后重试`,
+      message:
+        `一次最多分 ${ASSIGNMENT_ITEMS_HARD_LIMIT.toLocaleString('zh-CN')} 条。` +
+        `把时限改短、或者把每天几通调小,本批人数会跟着降下来。`,
     }),
 });
 export type CreateAssignmentRequest = z.infer<typeof CreateAssignmentRequestSchema>;