dachui/order_site
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

perf(pipeline): 订单项管道改用水位游标并只扫缺失

Browse Source 
- 新增 watermark-store：统一读写游标/日水位/失败原因
- 渠道判定、商品归一、成本快照从上次游标继续，到表尾后回绕重试
- 成本快照只处理没有有效快照的订单项，避免每 5 分钟全量重扫
This commit is contained in:
yml2213
2026-10-31 20:17:33 +08:00
parent ce775ca102
commit d979ef50d6
6 changed files with 134 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/backend/src/services/cost-center/cost-snapshot-service.ts
+25 -3
View File
@@ -1,6 +1,7 @@
import { query } from '../../db/client.js'
import { nowIso } from '../../utils/time.js'
import { resolveExceptionsForItems, upsertAggregatedException } from './cost-exception-service.js'
import { readWatermarkItemId } from '../watermark-store.js'
/**
* 管道 C1：成本快照（唯一入口）。
@@ -157,6 +158,7 @@ export type SyncCostSnapshotsResult = {
created: number
missing: number
lastId: number
nextCursor: number
}
export async function syncChannelCostSnapshots(
@@ -164,14 +166,24 @@ export async function syncChannelCostSnapshots(
): Promise<SyncCostSnapshotsResult> {
const batchSize = Math.min(5000, Math.max(1, Math.trunc(options.batchSize || 500)))
const maxBatches = Math.min(200, Math.max(1, Math.trunc(options.maxBatches || 20)))
let cursor = Math.max(0, Math.trunc(options.afterId || 0))
let cursor =
options.afterId != null
? Math.max(0, Math.trunc(options.afterId))
: await readWatermarkItemId('cost_snapshot')
let reachedEnd = false
const costItems = await loadCostItems()
const versions = await loadActiveVersions()
const pricingModeByItemId = new Map<number, string>()
for (const item of costItems.values()) pricingModeByItemId.set(item.id, item.pricing_mode)
const result: SyncCostSnapshotsResult = { scanned: 0, created: 0, missing: 0, lastId: cursor }
const result: SyncCostSnapshotsResult = {
scanned: 0,
created: 0,
missing: 0,
lastId: cursor,
nextCursor: cursor,
}
for (let batch = 0; batch < maxBatches; batch += 1) {
const rows = await query<CostCandidateItem>(
@@ -198,6 +210,11 @@ export async function syncChannelCostSnapshots(
oi.canonical_product_id IS NOT NULL
OR (oi.fulfillment_channel = 'work_order' AND wo.product_rule_id IS NOT NULL)
)
-- 只处理还没有有效快照的订单项，避免每轮重复解析整张表。
AND NOT EXISTS (
SELECT 1 FROM cost_snapshots cs
WHERE cs.order_item_id = oi.id AND cs.superseded_at IS NULL
)
AND oi.id > $1
ORDER BY oi.id
LIMIT $2
@@ -205,7 +222,11 @@ export async function syncChannelCostSnapshots(
[cursor, batchSize],
)
if (rows.rows.length === 0) break
if (rows.rows.length === 0) {
reachedEnd = true
break
}
if (rows.rows.length < batchSize) reachedEnd = true
result.scanned += rows.rows.length
cursor = Number(rows.rows[rows.rows.length - 1]!.id)
result.lastId = cursor
@@ -236,6 +257,7 @@ export async function syncChannelCostSnapshots(
await resolveExceptionsForItems(inserts.map((row) => row.item.id))
}
result.nextCursor = reachedEnd ? 0 : cursor
return result
}
apps/backend/src/services/cost-center/product-normalize-service.ts
+20 -3
View File
@@ -1,5 +1,6 @@
import { query } from '../../db/client.js'
import { resolveExceptionsForItems, upsertAggregatedException } from './cost-exception-service.js'
import { readWatermarkItemId } from '../watermark-store.js'
/**
* 管道 B：商品归一。
@@ -171,6 +172,7 @@ export type NormalizeProductsResult = {
matched: number
unmatched: number
lastId: number
nextCursor: number
}
export async function normalizeOrderItemProducts(
@@ -178,10 +180,20 @@ export async function normalizeOrderItemProducts(
): Promise<NormalizeProductsResult> {
const batchSize = Math.min(5000, Math.max(1, Math.trunc(options.batchSize || 500)))
const maxBatches = Math.min(200, Math.max(1, Math.trunc(options.maxBatches || 20)))
let cursor = Math.max(0, Math.trunc(options.afterId || 0))
let cursor =
options.afterId != null
? Math.max(0, Math.trunc(options.afterId))
: await readWatermarkItemId('product_normalize')
let reachedEnd = false
const index = await loadMappingIndex()
const result: NormalizeProductsResult = { scanned: 0, matched: 0, unmatched: 0, lastId: cursor }
const result: NormalizeProductsResult = {
scanned: 0,
matched: 0,
unmatched: 0,
lastId: cursor,
nextCursor: cursor,
}
for (let batch = 0; batch < maxBatches; batch += 1) {
const rows = await query<ItemRow>(
@@ -211,7 +223,11 @@ export async function normalizeOrderItemProducts(
[cursor, batchSize],
)
if (rows.rows.length === 0) break
if (rows.rows.length === 0) {
reachedEnd = true
break
}
if (rows.rows.length < batchSize) reachedEnd = true
result.scanned += rows.rows.length
cursor = Number(rows.rows[rows.rows.length - 1]!.id)
result.lastId = cursor
@@ -249,6 +265,7 @@ export async function normalizeOrderItemProducts(
}
}
result.nextCursor = reachedEnd ? 0 : cursor
return result
}
apps/backend/src/services/fulfillment/channel-assign-service.ts
+15 -2
View File
@@ -2,6 +2,7 @@ import { query } from '../../db/client.js'
import { createHttpError } from '../../utils/http.js'
import { nowIso } from '../../utils/time.js'
import { upsertAggregatedException } from '../cost-center/cost-exception-service.js'
import { readWatermarkItemId } from '../watermark-store.js'
import {
FULFILLMENT_CHANNELS,
MATCH_KEY_TYPES,
@@ -138,6 +139,8 @@ export type AssignChannelsResult = {
conflicted: number
stillPending: number
lastId: number
/** 下一轮起点；到达表尾时回绕到 0，保证漏判的订单还能被重新评估。 */
nextCursor: number
byChannel: Record<string, number>
}
@@ -147,7 +150,11 @@ export async function assignPendingOrderItemChannels(
): Promise<AssignChannelsResult> {
const batchSize = Math.min(5000, Math.max(1, Math.trunc(options.batchSize || 500)))
const maxBatches = Math.min(200, Math.max(1, Math.trunc(options.maxBatches || 20)))
let cursor = Math.max(0, Math.trunc(options.afterId || 0))
let cursor =
options.afterId != null
? Math.max(0, Math.trunc(options.afterId))
: await readWatermarkItemId('channel_assign')
let reachedEnd = false
const rules = await loadEnabledRules()
const result: AssignChannelsResult = {
@@ -156,6 +163,7 @@ export async function assignPendingOrderItemChannels(
conflicted: 0,
stillPending: 0,
lastId: cursor,
nextCursor: cursor,
byChannel: {},
}
@@ -188,7 +196,11 @@ export async function assignPendingOrderItemChannels(
[cursor, batchSize],
)
if (rows.rows.length === 0) break
if (rows.rows.length === 0) {
reachedEnd = true
break
}
if (rows.rows.length < batchSize) reachedEnd = true
result.scanned += rows.rows.length
cursor = Number(rows.rows[rows.rows.length - 1]!.id)
result.lastId = cursor
@@ -245,6 +257,7 @@ export async function assignPendingOrderItemChannels(
}
}
result.nextCursor = reachedEnd ? 0 : cursor
return result
}
apps/backend/src/services/scheduler/finance-pipeline-jobs.ts
+6 -4
View File
@@ -39,10 +39,12 @@ export async function runChannelAssignJob(job: JsonObject = {}) {
const result = await assignPendingOrderItemChannels({
batchSize: clampInt(config.batchSize, 500, 1, 5000),
maxBatches: clampInt(config.maxBatches, 20, 1, 200),
afterId: clampInt(config.afterId, 0, 0, Number.MAX_SAFE_INTEGER),
...(config.afterId != null
? { afterId: clampInt(config.afterId, 0, 0, Number.MAX_SAFE_INTEGER) }
: {}),
})
await touchWatermark('channel_assign', { lastItemId: result.lastId })
await touchWatermark('channel_assign', { lastItemId: result.nextCursor })
return {
status: 'ok',
message:
@@ -188,7 +190,7 @@ export async function runProductNormalizeJob(job: JsonObject = {}) {
maxBatches: clampInt(config.maxBatches, 20, 1, 200),
})
await touchWatermark('product_normalize', { lastItemId: result.lastId })
await touchWatermark('product_normalize', { lastItemId: result.nextCursor })
return {
status: 'ok',
message: `商品归一：扫描 ${result.scanned}，命中 ${result.matched}，未命中 ${result.unmatched}`,
@@ -203,7 +205,7 @@ export async function runCostSnapshotJob(job: JsonObject = {}) {
maxBatches: clampInt(config.maxBatches, 20, 1, 200),
})
await touchWatermark('cost_snapshot', { lastItemId: result.lastId })
await touchWatermark('cost_snapshot', { lastItemId: result.nextCursor })
return {
status: 'ok',
message: `成本快照：扫描 ${result.scanned}，生成 ${result.created}，缺失 ${result.missing}`,
apps/backend/src/services/scheduler/finance-pipeline-status.ts
+4 -40
View File
@@ -1,6 +1,7 @@
import { query } from '../../db/client.js'
import { nowIso } from '../../utils/time.js'
import { getScheduledJobRuntimeStates } from './scheduler-service.js'
import { recordPipelineError, touchWatermark } from '../watermark-store.js'
import type { WatermarkPipeline } from '../watermark-store.js'
/**
* 财务管道可观测性：水位 + 定时任务运行状态。
@@ -22,49 +23,12 @@ export const FINANCE_PIPELINES = [
export type FinancePipelineName = (typeof FINANCE_PIPELINES)[number]['name']
export { recordPipelineError, touchWatermark }
export function isFinancePipelineName(value: unknown): value is FinancePipelineName {
return FINANCE_PIPELINES.some((item) => item.name === value)
}
/** 记录一次成功运行的水位。 */
export async function touchWatermark(
pipeline: FinancePipelineName,
patch: { lastItemId?: number; lastStatDate?: string | null } = {},
): Promise<void> {
const now = nowIso()
await query(
`
INSERT INTO finance_watermarks (pipeline, last_item_id, last_stat_date, last_success_at, updated_at)
VALUES ($1, $2, $3::date, $4, $4)
ON CONFLICT (pipeline) DO UPDATE SET
last_item_id = GREATEST(finance_watermarks.last_item_id, EXCLUDED.last_item_id),
last_stat_date = COALESCE(EXCLUDED.last_stat_date, finance_watermarks.last_stat_date),
last_success_at = EXCLUDED.last_success_at,
last_error = '',
updated_at = EXCLUDED.updated_at
`,
[pipeline, Math.max(0, Math.trunc(patch.lastItemId || 0)), patch.lastStatDate ?? null, now],
)
}
/** 记录一次失败原因，不改水位。 */
export async function recordPipelineError(
pipeline: FinancePipelineName,
error: unknown,
): Promise<void> {
const message = error instanceof Error ? error.message : String(error)
await query(
`
INSERT INTO finance_watermarks (pipeline, last_error, updated_at)
VALUES ($1, $2, $3)
ON CONFLICT (pipeline) DO UPDATE SET
last_error = EXCLUDED.last_error,
updated_at = EXCLUDED.updated_at
`,
[pipeline, message.slice(0, 500), nowIso()],
)
}
export type FinancePipelineStatusRow = {
pipeline: string
label: string
apps/backend/src/services/watermark-store.ts
+64
View File
@@ -0,0 +1,64 @@
import { query } from '../db/client.js'
import { nowIso } from '../utils/time.js'
/**
* 管道水位：订单项级游标 + 按日水位 + 最近失败原因。
*
* 只在服务端使用，不依赖 scheduler，避免循环依赖。
*/
export type WatermarkPipeline =
| 'channel_assign'
| 'carrier_dispatch'
| 'product_normalize'
| 'cost_snapshot'
| 'sales_facts'
| 'ledger_post'
/** 读取订单项级游标；没有记录时返回 0。 */
export async function readWatermarkItemId(pipeline: WatermarkPipeline): Promise<number> {
const result = await query<{ last_item_id: string }>(
`SELECT last_item_id::text FROM finance_watermarks WHERE pipeline = $1`,
[pipeline],
)
return Number(result.rows[0]?.last_item_id || 0)
}
/** 记录一次成功运行的水位。 */
export async function touchWatermark(
pipeline: WatermarkPipeline,
patch: { lastItemId?: number; lastStatDate?: string | null } = {},
): Promise<void> {
const now = nowIso()
await query(
`
INSERT INTO finance_watermarks (pipeline, last_item_id, last_stat_date, last_success_at, updated_at)
VALUES ($1, $2, $3::date, $4, $4)
ON CONFLICT (pipeline) DO UPDATE SET
last_item_id = EXCLUDED.last_item_id,
last_stat_date = COALESCE(EXCLUDED.last_stat_date, finance_watermarks.last_stat_date),
last_success_at = EXCLUDED.last_success_at,
last_error = '',
updated_at = EXCLUDED.updated_at
`,
[pipeline, Math.max(0, Math.trunc(patch.lastItemId || 0)), patch.lastStatDate ?? null, now],
)
}
/** 记录一次失败原因，不改水位。 */
export async function recordPipelineError(
pipeline: WatermarkPipeline,
error: unknown,
): Promise<void> {
const message = error instanceof Error ? error.message : String(error)
await query(
`
INSERT INTO finance_watermarks (pipeline, last_error, updated_at)
VALUES ($1, $2, $3)
ON CONFLICT (pipeline) DO UPDATE SET
last_error = EXCLUDED.last_error,
updated_at = EXCLUDED.updated_at
`,
[pipeline, message.slice(0, 500), nowIso()],
)
}