From 8391912bda579720a77db1371b77f4cdde1c1283 Mon Sep 17 00:00:00 2001 From: yml2213 Date: Sun, 1 Nov 2026 16:22:52 +0800 Subject: [PATCH] =?UTF-8?q?perf(finance):=20=E7=AE=A1=E9=81=93=E5=A2=9E?= =?UTF-8?q?=E9=87=8F=E7=B4=A2=E5=BC=95=E4=B8=8E=E5=BC=82=E5=B8=B8=E6=89=B9?= =?UTF-8?q?=E9=87=8F=E5=85=A5=E9=98=9F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 生产副本 42 万单压测发现： - carrier_missing 查询在 26 万条上 15s 超时 → 改 anti-join，异常去重改批量 - 缺成本异常逐条 3 条 SQL，12 万人工单会跑几十分钟 → 按 scope 批量入队 - 新增 117/118 部分索引，成本快照/人工挂配置/载体检查走 id 顺序扫描 - 人工挂配置查询改为按配置 SKU 数组过滤，避免全量人工单排序 --- .../117_fulfillment_pipeline_indexes.sql | 33 +++++++ .../118_order_items_manual_sku_index.sql | 9 ++ .../cost-center/cost-exception-service.ts | 89 +++++++++++++++++++ .../cost-center/cost-snapshot-service.ts | 14 ++- .../scheduler/finance-pipeline-jobs.ts | 68 ++++++++------ 5 files changed, 184 insertions(+), 29 deletions(-) create mode 100644 apps/backend/src/db/migrations/117_fulfillment_pipeline_indexes.sql create mode 100644 apps/backend/src/db/migrations/118_order_items_manual_sku_index.sql diff --git a/apps/backend/src/db/migrations/117_fulfillment_pipeline_indexes.sql b/apps/backend/src/db/migrations/117_fulfillment_pipeline_indexes.sql new file mode 100644 index 00000000..efc9820f --- /dev/null +++ b/apps/backend/src/db/migrations/117_fulfillment_pipeline_indexes.sql @@ -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' + ); diff --git a/apps/backend/src/db/migrations/118_order_items_manual_sku_index.sql b/apps/backend/src/db/migrations/118_order_items_manual_sku_index.sql new file mode 100644 index 00000000..e63ee322 --- /dev/null +++ b/apps/backend/src/db/migrations/118_order_items_manual_sku_index.sql @@ -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'; diff --git a/apps/backend/src/services/cost-center/cost-exception-service.ts b/apps/backend/src/services/cost-center/cost-exception-service.ts index a21b6dda..0e13d3ee 100644 --- a/apps/backend/src/services/cost-center/cost-exception-service.ts +++ b/apps/backend/src/services/cost-center/cost-exception-service.ts @@ -76,6 +76,95 @@ export async function upsertAggregatedException(input: UpsertExceptionInput): Pr ) } +export type AggregatedExceptionEntry = { + reason: string + scopeKey: string + orderItemId: number + detail?: Record +} + +/** + * 批量入队：同一批内按 scope_key 聚合，避免逐条 3 条 SQL（12 万人工单会跑几十分钟）。 + */ +export async function upsertAggregatedExceptionsBatch( + entries: AggregatedExceptionEntry[], +): Promise { + const unique = new Map() + 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() + 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() + 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 { const now = nowIso() diff --git a/apps/backend/src/services/cost-center/cost-snapshot-service.ts b/apps/backend/src/services/cost-center/cost-snapshot-service.ts index eec02cf8..c7708bba 100644 --- a/apps/backend/src/services/cost-center/cost-snapshot-service.ts +++ b/apps/backend/src/services/cost-center/cost-snapshot-service.ts @@ -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 }> = [] const supersedeIds: number[] = [] + const missingEntries: AggregatedExceptionEntry[] = [] for (const item of rows) { const desired: Array> = [] @@ -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) => { diff --git a/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts b/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts index 70369011..19c1a8b9 100644 --- a/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts +++ b/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts @@ -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,