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 12322944..8059733d 100644 --- a/apps/backend/src/services/cost-center/cost-snapshot-service.ts +++ b/apps/backend/src/services/cost-center/cost-snapshot-service.ts @@ -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 { 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() 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( @@ -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 } diff --git a/apps/backend/src/services/cost-center/product-normalize-service.ts b/apps/backend/src/services/cost-center/product-normalize-service.ts index 05159aa6..3800e6fa 100644 --- a/apps/backend/src/services/cost-center/product-normalize-service.ts +++ b/apps/backend/src/services/cost-center/product-normalize-service.ts @@ -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 { 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( @@ -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 } diff --git a/apps/backend/src/services/fulfillment/channel-assign-service.ts b/apps/backend/src/services/fulfillment/channel-assign-service.ts index 16d844c5..9e1cb18e 100644 --- a/apps/backend/src/services/fulfillment/channel-assign-service.ts +++ b/apps/backend/src/services/fulfillment/channel-assign-service.ts @@ -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 } @@ -147,7 +150,11 @@ export async function assignPendingOrderItemChannels( ): Promise { 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 } diff --git a/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts b/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts index a03d4631..9eef7e5b 100644 --- a/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts +++ b/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts @@ -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}`, diff --git a/apps/backend/src/services/scheduler/finance-pipeline-status.ts b/apps/backend/src/services/scheduler/finance-pipeline-status.ts index 6b2ef15b..4308ded5 100644 --- a/apps/backend/src/services/scheduler/finance-pipeline-status.ts +++ b/apps/backend/src/services/scheduler/finance-pipeline-status.ts @@ -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 { - 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 { - 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 diff --git a/apps/backend/src/services/watermark-store.ts b/apps/backend/src/services/watermark-store.ts new file mode 100644 index 00000000..2da03fa0 --- /dev/null +++ b/apps/backend/src/services/watermark-store.ts @@ -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 { + 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 { + 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 { + 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()], + ) +}