perf(db): 收入口径回填改为分批脚本

迁移 104 里对 order_items 的全表 UPDATE 实测 34.4 秒（375,447 行），远超默认语句超时 15s，单事务迁移会拖住启动甚至直接失败。 - 104 只保留 DDL 与索引/约束，删除全表 UPDATE； - 新增 scripts/backfill-revenue-basis.ts：按 id 分批、幂等、支持 --dry-run / --batch-size / --max-batches（kebab 与 camel 两种写法）； - package.json 增加 revenue:backfill； - migrate.ts 启动时做一次带索引计数，仍有 unknown 时打印告警并给出命令，不阻塞启动； - 文档补充回填步骤与「不执行的后果」。验证： - 实测 37.5 万行全表更新 34.4s → 脚本按批执行，单批 5000 行； - 功能验证：单条订单项置 unknown → 脚本回填 → 金额与 basis 恢复正确； - 启动告警验证：保留 1 条 unknown 重启 → 日志出现告警；回填后计数归零； - 全新库从 102 迁移到 107 仍成功（2s）。
2026-10-25 12:02:26 +08:00
parent f10fefc303
commit ad1e1bc884
5 changed files with 256 additions and 29 deletions
@@ -11,6 +11,7 @@
    "db:migrate:status": "tsx scripts/migration-status.ts",
    "finance:sync": "tsx scripts/sync-cost-finance.ts",
    "fulfillment:backfill": "tsx scripts/backfill-fulfillment-mode.ts",
+    "revenue:backfill": "tsx scripts/backfill-revenue-basis.ts",
    "storage:migrate": "tsx scripts/migrate-storage.ts",
    "storage:migrate:built": "node dist/storage-migrate.js",
    "check:sql": "tsx scripts/check-sql-guard.ts",
@@ -0,0 +1,207 @@
+/**
+ * 历史订单项收入口径回填（迁移 104 只建列，不再做全表更新）。
+ *
+ * 背景：375,447 行的全表 UPDATE 实测约 34 秒，超过默认语句超时，放在单事务迁移里
+ * 会拖住启动；因此迁到本脚本，按 id 分批执行。
+ *
+ * 口径（与 order-item-repo 写入规则一致）：
+ *   item_count = 1 → revenue = 订单实付，revenue_basis = 'order_total'
+ *   item_count > 1 → 等额拆分，余数给最后一项，revenue_basis = 'equal_split'
+ *
+ * 幂等：只处理 revenue_basis = 'unknown' 的行，重复执行不会二次拆分。
+ *
+ * 用法：
+ *   npm run revenue:backfill                       # 默认每批 5000 行
+ *   npm run revenue:backfill -- --batch-size=2000 --max-batches=100
+ *   npm run revenue:backfill -- --dry-run          # 只统计并抽样预览，不写库
+ */
+import process from 'node:process'
+
+import { closeDb, query } from '../src/db/client.js'
+
+const options = parseOptions(process.argv.slice(2))
+const batchSize = normalizeInteger(options.batchSize, 5000, 1, 50_000)
+const maxBatches = normalizeInteger(options.maxBatches, 1000, 1, 10_000)
+const dryRun = options.dryRun === 'true'
+
+const BATCH_SQL = `
+  WITH target AS (
+    SELECT id, order_id
+    FROM order_items
+    WHERE revenue_basis = 'unknown'
+    ORDER BY id
+    LIMIT $1
+  ),
+  item_counts AS (
+    SELECT oi.order_id, COUNT(*)::int AS item_count
+    FROM order_items oi
+    WHERE oi.order_id IN (SELECT order_id FROM target)
+    GROUP BY oi.order_id
+  ),
+  ranked AS (
+    SELECT
+      oi.id,
+      GREATEST(COALESCE(o.total_amount, 0), 0)::bigint AS order_revenue_fen,
+      ic.item_count,
+      ROW_NUMBER() OVER (PARTITION BY oi.order_id ORDER BY oi.id ASC) AS item_rank
+    FROM order_items oi
+    JOIN orders o ON o.id = oi.order_id
+    JOIN item_counts ic ON ic.order_id = oi.order_id
+    WHERE oi.id IN (SELECT id FROM target)
+  )
+  UPDATE order_items oi
+  SET
+    revenue_amount_fen = CASE
+      WHEN r.item_count = 1 THEN r.order_revenue_fen
+      ELSE (r.order_revenue_fen / r.item_count)
+        + CASE WHEN r.item_rank = r.item_count THEN (r.order_revenue_fen % r.item_count) ELSE 0 END
+    END,
+    revenue_basis = CASE WHEN r.item_count = 1 THEN 'order_total' ELSE 'equal_split' END,
+    updated_at = NOW()
+  FROM ranked r
+  WHERE oi.id = r.id
+    AND oi.revenue_basis = 'unknown'
+  RETURNING oi.id
+`
+
+async function main() {
+  await assertColumnExists()
+
+  const before = await countUnknown()
+  const startedAt = Date.now()
+
+  if (dryRun) {
+    const samples = await previewSamples()
+    console.log(
+      JSON.stringify(
+        {
+          dryRun: true,
+          pendingCount: before,
+          batchSize,
+          estimatedBatches: Math.ceil(before / batchSize),
+          samples,
+          durationMs: Date.now() - startedAt,
+        },
+        null,
+        2,
+      ),
+    )
+    return
+  }
+
+  let batches = 0
+  let updated = 0
+  while (batches < maxBatches) {
+    const result = await query(BATCH_SQL, [batchSize])
+    const affected = result.rowCount ?? 0
+    if (affected === 0) break
+    batches += 1
+    updated += affected
+    if (affected < batchSize) break
+  }
+
+  const remaining = await countUnknown()
+  console.log(
+    JSON.stringify(
+      {
+        dryRun: false,
+        batchSize,
+        maxBatches,
+        batches,
+        updatedCount: updated,
+        pendingBefore: before,
+        remainingCount: remaining,
+        durationMs: Date.now() - startedAt,
+      },
+      null,
+      2,
+    ),
+  )
+
+  if (remaining > 0) {
+    console.error(`仍有 ${remaining} 个订单项未回填；请增大 --max-batches 后重跑（幂等）。`)
+    process.exitCode = 2
+  }
+}
+
+async function assertColumnExists() {
+  const { rows } = await query<{ exists: boolean }>(
+    `
+      SELECT EXISTS (
+        SELECT 1
+        FROM information_schema.columns
+        WHERE table_schema = 'public'
+          AND table_name = 'order_items'
+          AND column_name = 'revenue_basis'
+      ) AS exists
+    `,
+  )
+  if (!rows[0]?.exists) {
+    throw new Error('order_items.revenue_basis 不存在，请先执行 npm run db:migrate（迁移 104）')
+  }
+}
+
+async function countUnknown() {
+  const { rows } = await query<{ count: number }>(
+    `SELECT COUNT(*)::int AS count FROM order_items WHERE revenue_basis = 'unknown'`,
+  )
+  return rows[0]?.count ?? 0
+}
+
+async function previewSamples() {
+  const { rows } = await query<{
+    order_id: number
+    item_count: number
+    order_total: number
+    item_ids: number[]
+  }>(
+    `
+      SELECT
+        oi.order_id,
+        COUNT(*)::int AS item_count,
+        MAX(GREATEST(COALESCE(o.total_amount, 0), 0))::bigint AS order_total,
+        array_agg(oi.id ORDER BY oi.id) AS item_ids
+      FROM order_items oi
+      JOIN orders o ON o.id = oi.order_id
+      WHERE oi.revenue_basis = 'unknown'
+      GROUP BY oi.order_id
+      ORDER BY oi.order_id
+      LIMIT 3
+    `,
+  )
+  return rows
+}
+
+main()
+  .catch((error) => {
+    console.error(error instanceof Error ? error.message : error)
+    process.exitCode = 1
+  })
+  .finally(async () => {
+    await closeDb()
+  })
+
+function parseOptions(args: string[]) {
+  const result: Record<string, string> = {}
+  for (const arg of args) {
+    if (!arg.startsWith('--')) continue
+    const body = arg.slice(2)
+    const index = body.indexOf('=')
+    const key = index === -1 ? body : body.slice(0, index)
+    const value = index === -1 ? 'true' : body.slice(index + 1)
+    // 同时接受 --batch-size 与 --batchSize 两种写法。
+    result[key] = value
+    result[toCamelCase(key)] = value
+  }
+  return result
+}
+
+function toCamelCase(value: string) {
+  return value.replace(/-([a-z])/g, (_, char: string) => char.toUpperCase())
+}
+
+function normalizeInteger(value: unknown, fallback: number, min: number, max: number) {
+  const parsed = Number(value)
+  if (!Number.isInteger(parsed)) return fallback
+  return Math.min(max, Math.max(min, parsed))
+}
@@ -58,6 +58,32 @@ export async function runDatabaseMigrations() {
  } satisfies RunnerOption)

  await syncLegacyMigrationColumns()
+  await warnPendingRevenueBackfill()
+}
+
+/**
+ * 迁移 104 只建列、不再做全表收入回填（37.5 万行实测约 34 秒，会拖住启动）。
+ * 这里只做一次带索引的计数，发现仍有未固化的订单项就提示执行回填脚本，
+ * 不阻塞启动——脚本幂等，可以随时补跑。
+ */
+async function warnPendingRevenueBackfill() {
+  try {
+    const { rows } = await query<{ count: number }>(
+      `SELECT COUNT(*)::int AS count FROM order_items WHERE revenue_basis = 'unknown'`,
+    )
+    const pending = rows[0]?.count ?? 0
+    if (pending > 0) {
+      console.warn(
+        `[db:migrate] 还有 ${pending} 个订单项未固化收入口径，销售事实会偏低；` +
+          '请执行 npm run revenue:backfill（分批、幂等）',
+      )
+    }
+  } catch (error) {
+    // 列还不存在（迁移没跑到 104）或表缺失时跳过，不能影响启动。
+    console.warn(
+      `[db:migrate] 跳过收入口径检查: ${error instanceof Error ? error.message : String(error)}`,
+    )
+  }
 }

 function assertMigrationFilesOrdered() {
@@ -317,35 +317,11 @@ BEGIN
  END IF;
 END $$;

-- 历史订单回填：当前生产快照全部是一单一品，仍按通用规则处理多订单项。
-- 幂等：只处理 revenue_basis 仍为 unknown 的行，重复执行不会二次拆分。
-WITH item_counts AS (
-  SELECT order_id, COUNT(*)::int AS item_count
-  FROM order_items
-  GROUP BY order_id
-),
-ranked AS (
-  SELECT
-    oi.id,
-    GREATEST(COALESCE(o.total_amount, 0), 0)::bigint AS order_revenue_fen,
-    ic.item_count,
-    ROW_NUMBER() OVER (PARTITION BY oi.order_id ORDER BY oi.id ASC) AS item_rank
-  FROM order_items oi
-  JOIN orders o ON o.id = oi.order_id
-  JOIN item_counts ic ON ic.order_id = oi.order_id
-)
-UPDATE order_items oi
-SET
-  revenue_amount_fen = CASE
-    WHEN r.item_count = 1 THEN r.order_revenue_fen
-    ELSE (r.order_revenue_fen / r.item_count)
-      + CASE WHEN r.item_rank = r.item_count THEN (r.order_revenue_fen % r.item_count) ELSE 0 END
-  END,
-  revenue_basis = CASE WHEN r.item_count = 1 THEN 'order_total' ELSE 'equal_split' END,
-  updated_at = NOW()
-FROM ranked r
-WHERE oi.id = r.id
-  AND oi.revenue_basis = 'unknown';
+-- 历史订单的收入回填**不在迁移里做**。
+-- 实测：375,447 行的全表 UPDATE 需要约 34 秒，远超默认语句超时
+-- （DATABASE_STATEMENT_TIMEOUT_MS=15s），放进单事务迁移会拖住启动甚至直接失败。
+-- 改由 `npm run revenue:backfill` 分批执行（幂等、可续跑、支持 --dry-run）；
+-- 应用启动时若仍有 revenue_basis='unknown' 的行，会打印告警提示执行该命令。

 CREATE INDEX IF NOT EXISTS idx_order_items_revenue_basis
  ON order_items (revenue_basis);
@@ -430,6 +430,23 @@ DELETE FROM schema_migrations WHERE name ~ '^(10[3-9]|11[0-2])_';

 旧的逐列改造脚本 `scripts/rewrite-local-db-2026-10-23.sql` 已随迁移合并作废，移入 [archive/2026-10-23/](archive/2026-10-23/)；只在追溯当时 `carrier_type → cost_model` / `fulfillment_mode` 的改造矩阵时查阅，**不要在新环境执行**。

+### 8.1 收入口径回填（迁移 104 不做全表更新）
+
+`order_items.revenue_amount_fen` / `revenue_basis` 在迁移 `104` 里**只建列，不刷历史数据**：
+375,447 行的全表 `UPDATE` 实测约 **34 秒**，超过默认语句超时（`DATABASE_STATEMENT_TIMEOUT_MS=15s`），
+放进单事务迁移会拖住启动甚至直接失败。
+
+历史行由脚本分批回填（幂等、可续跑）：
+
+```bash
+npm run revenue:backfill -- --dry-run                 # 统计待回填量并抽样预览
+npm run revenue:backfill                              # 默认每批 5000 行
+npm run revenue:backfill -- --batch-size=2000 --max-batches=100
+```
+
+服务启动时会做一次带索引的计数：若仍有 `revenue_basis='unknown'` 的订单项，会打印告警并提示执行上面的命令。
+**不执行回填的后果是销售事实收入偏低**（订单项收入按 0 计），不会报错，所以部署后必须确认该计数为 0。
+
 `check-sql-guard.ts` 改为以 `main` 为基线校验迁移字节：**已发布的迁移不可修改**，未合并分支上的新迁移仍可重写；基线可用 `MIGRATION_BASELINE_REF` 覆盖，找不到基线时跳过该项检查。

 ---