yml2213
d979ef50d6
- 新增 watermark-store：统一读写游标/日水位/失败原因 - 渠道判定、商品归一、成本快照从上次游标继续，到表尾后回绕重试 - 成本快照只处理没有有效快照的订单项，避免每 5 分钟全量重扫
65 lines
2.1 KiB
TypeScript
65 lines
2.1 KiB
TypeScript
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()],
|
|
)
|
|
}
|