perf(finance): 管道增量索引与异常批量入队

生产副本 42 万单压测发现： - carrier_missing 查询在 26 万条上 15s 超时 → 改 anti-join，异常去重改批量 - 缺成本异常逐条 3 条 SQL，12 万人工单会跑几十分钟 → 按 scope 批量入队 - 新增 117/118 部分索引，成本快照/人工挂配置/载体检查走 id 顺序扫描 - 人工挂配置查询改为按配置 SKU 数组过滤，避免全量人工单排序
2026-11-01 16:22:52 +08:00
parent 5d7904f6ee
commit 8391912bda
5 changed files with 184 additions and 29 deletions
@@ -0,0 +1,33 @@
+-- 117_fulfillment_pipeline_indexes.sql
+--
+-- 财务管道按订单项 id 增量扫描时的专用部分索引。
+-- 42 万行实测：carrier_dispatch 的「人工配置待挂载」查询原本 10.9s（扫描 12 万人工单），
+-- 加上按 id 的部分索引后可走索引顺序扫描并提前 LIMIT 结束。
+
+-- carrier_dispatch：人工渠道待挂配置
+CREATE INDEX IF NOT EXISTS idx_order_items_manual_pending_config_id
+  ON order_items (id)
+  WHERE fulfillment_status = 'assigned'
+    AND fulfillment_channel = 'manual'
+    AND fulfillment_config_id IS NULL;
+
+-- carrier_dispatch：已配置但未确认履约（人工成本未记录）
+CREATE INDEX IF NOT EXISTS idx_order_items_manual_configured_id
+  ON order_items (id)
+  WHERE fulfillment_status = 'assigned'
+    AND fulfillment_channel = 'manual'
+    AND fulfillment_config_id IS NOT NULL;
+
+-- carrier_dispatch：自动发货/接单平台缺少履约载体
+CREATE INDEX IF NOT EXISTS idx_order_items_carrier_check_id
+  ON order_items (id)
+  WHERE fulfillment_status = 'assigned'
+    AND fulfillment_channel IN ('affiliate_dash', 'work_order');
+
+-- cost_snapshot：按 id 顺序扫描需要核算成本的订单项
+CREATE INDEX IF NOT EXISTS idx_order_items_cost_ready_id
+  ON order_items (id)
+  WHERE fulfillment_status = 'assigned'
+    AND fulfillment_channel IN (
+      'affiliate_dash', 'third_party_voucher', 'work_order', 'manual'
+    );
@@ -0,0 +1,9 @@
+-- 118_order_items_manual_sku_index.sql
+--
+-- carrier_dispatch 按「人工 SKU 配置」反查订单项时，先按 manual_sku_key 命中再按 id 排序；
+-- 42 万行实测原查询 3.3s，补 (manual_sku_key, id) 部分索引后可走位图索引 + 小结果集排序。
+
+CREATE INDEX IF NOT EXISTS idx_order_items_manual_sku_key_id
+  ON order_items (manual_sku_key, id)
+  WHERE fulfillment_status = 'assigned'
+    AND fulfillment_channel = 'manual';
@@ -76,6 +76,95 @@ export async function upsertAggregatedException(input: UpsertExceptionInput): Pr
  )
 }

+export type AggregatedExceptionEntry = {
+  reason: string
+  scopeKey: string
+  orderItemId: number
+  detail?: Record<string, unknown>
+}
+
+/**
+ * 批量入队：同一批内按 scope_key 聚合，避免逐条 3 条 SQL（12 万人工单会跑几十分钟）。
+ */
+export async function upsertAggregatedExceptionsBatch(
+  entries: AggregatedExceptionEntry[],
+): Promise<number> {
+  const unique = new Map<string, AggregatedExceptionEntry>()
+  for (const entry of entries) {
+    if (!Number.isInteger(entry.orderItemId) || entry.orderItemId <= 0) continue
+    unique.set(`${entry.scopeKey}|${entry.orderItemId}`, entry)
+  }
+  const items = [...unique.values()]
+  if (items.length === 0) return 0
+
+  // ON CONFLICT DO UPDATE 不允许同一条语句内出现重复 scope_key，
+  // 聚合异常多个订单项共享一个 scope_key，插入阶段只保留每个 scope 的首条。
+  const scopeFirst = new Map<string, AggregatedExceptionEntry>()
+  for (const entry of items) {
+    if (!scopeFirst.has(entry.scopeKey)) scopeFirst.set(entry.scopeKey, entry)
+  }
+  const heads = [...scopeFirst.values()]
+
+  const now = nowIso()
+  const inserted = await query<{ id: number; scope_key: string }>(
+    `
+      INSERT INTO cost_exceptions (
+        subject_type, subject_id, order_item_id, reason, status, detail_json,
+        scope_key, item_count, first_seen_at, last_seen_at, created_at
+      )
+      SELECT 'order_item', s.order_item_id, s.order_item_id, s.reason, 'open', s.detail::jsonb,
+             s.scope_key, 1, $4, $4, $4
+      FROM unnest($1::bigint[], $2::text[], $3::text[], $5::text[]) AS s(order_item_id, reason, scope_key, detail)
+      ON CONFLICT (scope_key) WHERE status <> 'resolved'
+      DO UPDATE SET
+        last_seen_at = EXCLUDED.last_seen_at,
+        detail_json = EXCLUDED.detail_json
+      RETURNING id, scope_key
+    `,
+    [
+      heads.map((entry) => entry.orderItemId),
+      heads.map((entry) => entry.reason),
+      heads.map((entry) => entry.scopeKey),
+      now,
+      heads.map((entry) => JSON.stringify(entry.detail || {})),
+    ],
+  )
+
+  const idByScope = new Map<string, number>()
+  for (const row of inserted.rows) idByScope.set(row.scope_key, Number(row.id))
+
+  const links = items
+    .map((entry) => ({
+      exceptionId: idByScope.get(entry.scopeKey),
+      orderItemId: entry.orderItemId,
+    }))
+    .filter((row): row is { exceptionId: number; orderItemId: number } => Boolean(row.exceptionId))
+  if (links.length === 0) return 0
+
+  await query(
+    `
+      INSERT INTO cost_exception_items (exception_id, order_item_id, created_at)
+      SELECT input.exception_id, input.order_item_id, $3
+      FROM unnest($1::bigint[], $2::bigint[]) AS input(exception_id, order_item_id)
+      ON CONFLICT (exception_id, order_item_id) DO NOTHING
+    `,
+    [links.map((row) => row.exceptionId), links.map((row) => row.orderItemId), now],
+  )
+
+  const exceptionIds = [...new Set(links.map((row) => row.exceptionId))]
+  await query(
+    `
+      UPDATE cost_exceptions ce
+      SET item_count = (
+        SELECT COUNT(*)::int FROM cost_exception_items i WHERE i.exception_id = ce.id
+      )
+      WHERE ce.id = ANY($1::bigint[])
+    `,
+    [exceptionIds],
+  )
+  return links.length
+}
+
 /** 问题已解决：关闭问题头并清空明细。 */
 export async function resolveAggregatedException(scopeKey: string): Promise<number> {
  const now = nowIso()
@@ -2,7 +2,11 @@ import { query, withTransaction } from '../../db/client.js'
 import type { PoolClient } from 'pg'
 import { nowIso } from '../../utils/time.js'
 import { GENERIC_COST_MODELS } from './cost-center-constants.js'
-import { resolveExceptionsForItems, upsertAggregatedException } from './cost-exception-service.js'
+import {
+  resolveExceptionsForItems,
+  upsertAggregatedExceptionsBatch,
+  type AggregatedExceptionEntry,
+} from './cost-exception-service.js'
 import { readWatermarkItemId } from '../watermark-store.js'

 /**
@@ -366,6 +370,7 @@ async function processCostRows(
  const openMissing = await loadOpenMissingItems(rows.map((row) => row.id))
  const inserts: Array<{ item: CostCandidateItem; resolution: NonNullable<CostResolution> }> = []
  const supersedeIds: number[] = []
+  const missingEntries: AggregatedExceptionEntry[] = []

  for (const item of rows) {
    const desired: Array<NonNullable<CostResolution>> = []
@@ -379,10 +384,10 @@ async function processCostRows(
      result.missing += 1
      // 已在队列里的缺失项不再重复写三条 SQL，等价格补齐后自然收口。
      if (!openMissing.has(Number(item.id))) {
-        await upsertAggregatedException({
+        missingEntries.push({
          reason: 'cost_missing',
          scopeKey: `cost_missing|${item.fulfillment_channel}|${item.canonical_product_id}|${item.work_product_rule_id ?? '-'}`,
-          orderItemId: item.id,
+          orderItemId: Number(item.id),
          detail: { channel: item.fulfillment_channel },
        })
      }
@@ -424,6 +429,9 @@ async function processCostRows(
    }
  }

+  if (missingEntries.length > 0) {
+    await upsertAggregatedExceptionsBatch(missingEntries)
+  }
  if (supersedeIds.length === 0 && inserts.length === 0) return
  // 替换与写入放在同一事务里，避免事实任务读到「旧行已过期、新行未写入」的空窗。
  const created = await withTransaction(async (client) => {
@@ -1,7 +1,11 @@
 import type { JsonObject } from '../../types/json.js'
 import { query } from '../../db/client.js'
 import { assignPendingOrderItemChannels } from '../fulfillment/channel-assign-service.js'
-import { resolveExceptionsForItemsByReason } from '../cost-center/cost-exception-service.js'
+import {
+  resolveExceptionsForItemsByReason,
+  upsertAggregatedExceptionsBatch,
+  type AggregatedExceptionEntry,
+} from '../cost-center/cost-exception-service.js'
 import { readWatermarkItemId } from '../watermark-store.js'
 import { refreshFactsForDates } from '../finance/refresh-facts.js'
 import { rebuildChannelSalesDailyFacts } from '../finance/sales-facts-channel-service.js'
@@ -14,7 +18,6 @@ import {
  syncChannelCostSnapshots,
  syncCostSnapshotsForItems,
 } from '../cost-center/cost-snapshot-service.js'
-import { upsertAggregatedException } from '../cost-center/cost-exception-service.js'
 import { formatShanghaiDate, shiftDate } from '../../utils/finance-date-range.js'
 import {
  recordPipelineError,
@@ -84,6 +87,10 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) {
      ? Math.max(0, Math.trunc(Number(config.afterId)))
      : await readWatermarkItemId('carrier_dispatch')
  let maxId = 0
+  const configKeysResult = await query<{ sku_key: string }>(
+    'SELECT sku_key FROM manual_fulfillment_configs WHERE enabled',
+  )
+  const configKeys = configKeysResult.rows.map((row) => row.sku_key)

  const manual = await query<{ id: number; sku_key: string; config_id: number }>(
    `
@@ -93,11 +100,12 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) {
      WHERE oi.fulfillment_status = 'assigned'
        AND oi.fulfillment_channel = 'manual'
        AND oi.fulfillment_config_id IS NULL
+        AND oi.manual_sku_key = ANY($3::text[])
        AND oi.id > $2
      ORDER BY oi.id
      LIMIT $1
    `,
-    [limit, cursor],
+    [limit, cursor, configKeys],
  )

  let attached = 0
@@ -121,10 +129,7 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) {
      WHERE oi.fulfillment_status = 'assigned'
        AND oi.fulfillment_channel = 'manual'
        AND oi.fulfillment_config_id IS NULL
-        AND NOT EXISTS (
-          SELECT 1 FROM manual_fulfillment_configs cfg
-          WHERE cfg.sku_key = oi.manual_sku_key AND cfg.enabled
-        )
+        AND NOT (oi.manual_sku_key = ANY($3::text[]))
        AND NOT EXISTS (
          SELECT 1
          FROM cost_exception_items i
@@ -137,14 +142,15 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) {
      ORDER BY oi.id
      LIMIT $1
    `,
-    [limit, cursor],
+    [limit, cursor, configKeys],
  )

+  const exceptionEntries: AggregatedExceptionEntry[] = []
  for (const row of missingManual.rows) {
-    await upsertAggregatedException({
+    exceptionEntries.push({
      reason: 'manual_config_missing',
      scopeKey: `manual_config_missing|-|-|${row.sku_key}|manual`,
-      orderItemId: row.id,
+      orderItemId: Number(row.id),
      detail: { skuKey: row.sku_key },
    })
  }
@@ -179,37 +185,28 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) {
  )

  for (const row of pendingManualCost.rows) {
-    await upsertAggregatedException({
+    exceptionEntries.push({
      reason: 'manual_cost_missing',
      scopeKey: `manual_cost_missing|-|-|${row.sku_key}|manual`,
-      orderItemId: row.id,
+      orderItemId: Number(row.id),
      detail: { skuKey: row.sku_key },
    })
  }

  // 自动发货 / 接单平台被运营指定后却没有载体：不能静默卡住，进问题队列。
-  const missingCarrier = await query<{ id: number; channel: string; sku_code: string }>(
+  const missingCarrierRaw = await query<{ id: number; channel: string; sku_code: string }>(
    `
      SELECT oi.id, oi.fulfillment_channel AS channel, oi.sku_code
      FROM order_items oi
+      LEFT JOIN work_orders wo ON wo.order_item_id = oi.id
      WHERE oi.fulfillment_status = 'assigned'
        AND oi.fulfillment_channel IN ('affiliate_dash', 'work_order')
+        AND wo.id IS NULL
        AND NOT EXISTS (
          SELECT 1 FROM fulfillment_tasks ft
          WHERE ft.order_item_id = oi.id
            AND ft.executor_key IN ('affiliate_dash', 'third_party_voucher')
        )
-        AND NOT EXISTS (
-          SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id
-        )
-        AND NOT EXISTS (
-          SELECT 1
-          FROM cost_exception_items i
-          JOIN cost_exceptions ce ON ce.id = i.exception_id
-          WHERE i.order_item_id = oi.id
-            AND ce.reason = 'carrier_missing'
-            AND ce.status <> 'resolved'
-        )
        AND oi.id > $2
      ORDER BY oi.id
      LIMIT $1
@@ -217,14 +214,33 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) {
    [limit, cursor],
  )

+  const queuedCarrier = await query<{ order_item_id: number }>(
+    `
+      SELECT DISTINCT i.order_item_id
+      FROM cost_exception_items i
+      JOIN cost_exceptions ce ON ce.id = i.exception_id
+      WHERE i.order_item_id = ANY($1::bigint[])
+        AND ce.reason = 'carrier_missing'
+        AND ce.status <> 'resolved'
+    `,
+    [missingCarrierRaw.rows.map((row) => row.id)],
+  )
+  const queuedSet = new Set(queuedCarrier.rows.map((row) => Number(row.order_item_id)))
+  const missingCarrier = {
+    rows: missingCarrierRaw.rows.filter((row) => !queuedSet.has(Number(row.id))),
+  }
+
  for (const row of missingCarrier.rows) {
-    await upsertAggregatedException({
+    exceptionEntries.push({
      reason: 'carrier_missing',
      scopeKey: `carrier_missing|${row.channel}|${row.sku_code}`,
-      orderItemId: row.id,
+      orderItemId: Number(row.id),
      detail: { channel: row.channel, skuCode: row.sku_code },
    })
  }
+  if (exceptionEntries.length > 0) {
+    await upsertAggregatedExceptionsBatch(exceptionEntries)
+  }

  for (const row of [
    ...manual.rows,