diff --git a/apps/backend/src/services/cost-center/affiliate-product-service.ts b/apps/backend/src/services/cost-center/affiliate-product-service.ts index 067556d4..3413b0a7 100644 --- a/apps/backend/src/services/cost-center/affiliate-product-service.ts +++ b/apps/backend/src/services/cost-center/affiliate-product-service.ts @@ -431,6 +431,16 @@ export async function setAdminAffiliateProductPrice( * 批量从上游基线导入 38 条自动发货采购成本。 */ export async function importAdminAffiliateDashPricesBaseline(session: AdminSession) { + // 已有成本快照后禁止再导基线:基线是初始化工具,不能用来回写历史订单。 + const snapshotCount = await query<{ total: number }>( + 'SELECT COUNT(*)::int AS total FROM cost_snapshots', + ) + if (Number(snapshotCount.rows[0]?.total || 0) > 0) { + throw createHttpError('已有成本快照,禁止再导入基线价格;请改用改价(只影响后续订单)', { + statusCode: 409, + errorCode: 'cost_baseline_import_blocked', + }) + } await cleanupDirtyAffiliateDashProducts() const rows = AFFILIATE_DASH_PRICES_BASELINE @@ -441,8 +451,7 @@ export async function importAdminAffiliateDashPricesBaseline(session: AdminSessi if (!costItem) throw createHttpError('未找到自动发货采购成本项', { statusCode: 500 }) const now = nowIso() - const today = formatShanghaiDate(new Date()) - const effectiveFrom = shiftDate(today, -30) + 'T00:00:00+08:00' + const effectiveFrom = now let createdProductCount = 0 let createdVersionCount = 0 let updatedVersionCount = 0 @@ -483,16 +492,7 @@ export async function importAdminAffiliateDashPricesBaseline(session: AdminSessi const current = existing.rows[0] if (current && Number(current.amount_fen) === row.costAmountFen) { - if (Date.parse(effectiveFrom) < Date.parse(current.effective_from)) { - await query('UPDATE cost_versions SET effective_from = $1, updated_at = $2 WHERE id = $3', [ - effectiveFrom, - now, - current.id, - ]) - updatedVersionCount += 1 - } else { - skippedCount += 1 - } + skippedCount += 1 continue } diff --git a/apps/backend/src/services/cost-center/cost-definition-service.ts b/apps/backend/src/services/cost-center/cost-definition-service.ts index ffbf857a..84700ef6 100644 --- a/apps/backend/src/services/cost-center/cost-definition-service.ts +++ b/apps/backend/src/services/cost-center/cost-definition-service.ts @@ -426,6 +426,12 @@ export async function createAdminCostVersion(payload: JsonObject, session: Admin } const effectiveFrom = normalizeIsoDate(payload.effectiveFrom, '生效时间') || nowIso() + if (Date.parse(effectiveFrom) < Date.now() - 60_000) { + throw createHttpError('生效时间不能早于当前时间;历史订单成本已固化', { + statusCode: 400, + errorCode: 'cost_effective_from_in_past', + }) + } const effectiveTo = normalizeIsoDate(payload.effectiveTo, '失效时间', true) if (effectiveTo && Date.parse(effectiveTo) <= Date.parse(effectiveFrom)) { throw createHttpError('失效时间必须晚于生效时间', { diff --git a/apps/backend/src/services/cost-center/cost-snapshot-service.test.ts b/apps/backend/src/services/cost-center/cost-snapshot-service.test.ts index b7d3a89c..0dd3f99d 100644 --- a/apps/backend/src/services/cost-center/cost-snapshot-service.test.ts +++ b/apps/backend/src/services/cost-center/cost-snapshot-service.test.ts @@ -356,9 +356,10 @@ test('比例费用按退款后的净收入计算', () => { assert.equal(withRefund[0]?.totalAmountFen, 180) }) -test('历史订单成本固化:只有预估→实际允许替换', () => { +test('历史订单成本固化:只有预估→实际与派生费用允许替换', () => { assert.equal(shouldSupersedeSnapshot(null, 'estimated'), true) assert.equal(shouldSupersedeSnapshot('estimated', 'estimated'), false) + assert.equal(shouldSupersedeSnapshot('estimated', 'estimated', true), true) assert.equal(shouldSupersedeSnapshot('estimated', 'actual'), true) assert.equal(shouldSupersedeSnapshot('actual', 'estimated'), false) assert.equal(shouldSupersedeSnapshot('actual', 'actual'), false) 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 81367d29..eec02cf8 100644 --- a/apps/backend/src/services/cost-center/cost-snapshot-service.ts +++ b/apps/backend/src/services/cost-center/cost-snapshot-service.ts @@ -250,9 +250,14 @@ export function resolveGenericCosts( export function shouldSupersedeSnapshot( currentStage: string | null | undefined, nextStage: string, + mutableEstimated = false, ): boolean { if (!currentStage) return true - return currentStage !== 'actual' && nextStage === 'actual' + if (currentStage === 'estimated' && nextStage === 'actual') return true + // 平台费/支付费是随收入变化的派生费用,退款后允许按净收入重算; + // 采购成本、打手报酬等成交成本仍保持固化。 + if (currentStage === 'estimated' && nextStage === 'estimated' && mutableEstimated) return true + return false } export type SyncCostSnapshotsResult = { @@ -270,20 +275,58 @@ export type SyncCostSnapshotsResult = { affectedDates: string[] } -export async function syncChannelCostSnapshots( - options: { batchSize?: number; maxBatches?: number; afterId?: number } = {}, -): 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))) - const singleItem = options.afterId != null +type CostContext = { + channelCostItems: Map + genericCostItems: CostItemRow[] + versions: CostVersionRow[] + pricingModeByItemId: Map + costModelByItemId: Map + genericConfigured: boolean +} - let coldCursor = singleItem - ? Math.max(0, Math.trunc(options.afterId!)) - : await readWatermarkItemId('cost_snapshot') - let hotCursor = singleItem ? 0 : await readWatermarkItemId('cost_snapshot_hot') - const coldBound = hotCursor // 冷扫描只覆盖本轮开始前已存在的订单项 - let coldReachedEnd = false +const CANDIDATE_SQL = ` + SELECT + oi.id, + oi.order_id, + oi.fulfillment_channel, + oi.canonical_product_id, + GREATEST(COALESCE(oi.normalized_quantity, oi.quantity, 1), 1)::int AS quantity, + COALESCE(o.paid_at, o.created_at) AS basis_at, + (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date::text + AS basis_date, + COALESCE(oi.revenue_amount_fen, 0)::bigint AS revenue_fen, + COALESCE(( + SELECT SUM(r.amount_fen) FROM order_refunds r + WHERE r.order_item_id = oi.id AND r.status = 'confirmed' + ), 0)::bigint AS refund_fen, + wo.id AS work_order_id, + wo.product_rule_id AS work_product_rule_id, + wo.status AS work_order_status, + wo.reward_amount + FROM order_items oi + JOIN orders o ON o.id = oi.order_id + LEFT JOIN work_orders wo ON wo.order_item_id = oi.id + WHERE oi.fulfillment_status = 'assigned' + AND oi.fulfillment_channel IN ( + 'affiliate_dash', 'third_party_voucher', 'work_order', 'manual' + ) +` +function emptyResult(lastId = 0, nextCursor = 0, hotCursor = 0): SyncCostSnapshotsResult { + return { + scanned: 0, + created: 0, + superseded: 0, + missing: 0, + genericMissing: 0, + lastId, + nextCursor, + hotCursor, + affectedDates: [], + } +} + +async function loadCostContext(): Promise { const allCostItems = await loadCostItems() const channelCostItems = new Map() const genericCostItems: CostItemRow[] = [] @@ -296,130 +339,128 @@ export async function syncChannelCostSnapshots( } const versions = await loadActiveVersions() const pricingModeByItemId = new Map() - for (const item of allCostItems) pricingModeByItemId.set(item.id, item.pricing_mode) - - const result: SyncCostSnapshotsResult = { - scanned: 0, - created: 0, - superseded: 0, - missing: 0, - genericMissing: 0, - lastId: coldCursor, - nextCursor: coldCursor, - hotCursor, - affectedDates: [], + const costModelByItemId = new Map() + for (const item of allCostItems) { + pricingModeByItemId.set(item.id, item.pricing_mode) + costModelByItemId.set(Number(item.id), item.cost_model) } - const affectedDates = new Set() - const genericConfigured = hasConfiguredGeneric(genericCostItems, versions) + return { + channelCostItems, + genericCostItems, + versions, + pricingModeByItemId, + costModelByItemId, + genericConfigured: hasConfiguredGeneric(genericCostItems, versions), + } +} - async function processRows(rows: CostCandidateItem[]): Promise { - if (rows.length === 0) return - result.scanned += rows.length - const existingMap = await loadActiveSnapshots(rows.map((row) => row.id)) - const openMissing = await loadOpenMissingItems(rows.map((row) => row.id)) - const inserts: Array<{ item: CostCandidateItem; resolution: NonNullable }> = [] - const supersedeIds: number[] = [] +async function processCostRows( + rows: CostCandidateItem[], + ctx: CostContext, + result: SyncCostSnapshotsResult, + affectedDates: Set, +): Promise { + if (rows.length === 0) return + result.scanned += rows.length + const existingMap = await loadActiveSnapshots(rows.map((row) => row.id)) + const openMissing = await loadOpenMissingItems(rows.map((row) => row.id)) + const inserts: Array<{ item: CostCandidateItem; resolution: NonNullable }> = [] + const supersedeIds: number[] = [] - for (const item of rows) { - const desired: Array> = [] - const code = CHANNEL_COST_CODE[String(item.fulfillment_channel || '')] - const channelResolution = code - ? resolveCost(item, channelCostItems, versions, pricingModeByItemId) - : null - if (channelResolution) { - desired.push(channelResolution) - } else if (code) { - result.missing += 1 - // 已在队列里的缺失项不再重复写三条 SQL,等价格补齐后自然收口。 - if (!openMissing.has(Number(item.id))) { - await upsertAggregatedException({ - reason: 'cost_missing', - scopeKey: `cost_missing|${item.fulfillment_channel}|${item.canonical_product_id}|${item.work_product_rule_id ?? '-'}`, - orderItemId: item.id, - detail: { channel: item.fulfillment_channel }, - }) - } - } - - if (genericCostItems.length > 0) { - const generic = resolveGenericCosts(item, genericCostItems, versions, pricingModeByItemId) - desired.push(...generic) - if (generic.length === 0 && genericConfigured) result.genericMissing += 1 - } - - const existing = existingMap.get(Number(item.id)) || new Map() - for (const resolution of desired) { - const current = existing.get(Number(resolution.costItemId)) - if ( - current && - Number(current.cost_version_id) === Number(resolution.costVersionId) && - Number(current.total_amount_fen) === Number(resolution.totalAmountFen) && - current.snapshot_stage === resolution.stage - ) { - continue - } - // 已经按实际奖励入账的,不允许被后来的预估覆盖(工单回退/取消场景)。 - if (current && current.snapshot_stage === 'actual' && resolution.stage === 'estimated') { - continue - } - if (current) { - // 历史订单成本/成交价已固化,只有「预估→实际」允许升级。 - if (!shouldSupersedeSnapshot(current.snapshot_stage, resolution.stage)) continue - supersedeIds.push(Number(current.id)) - result.superseded += 1 - } - inserts.push({ item, resolution }) - affectedDates.add(item.basis_date) + for (const item of rows) { + const desired: Array> = [] + const code = CHANNEL_COST_CODE[String(item.fulfillment_channel || '')] + const channelResolution = code + ? resolveCost(item, ctx.channelCostItems, ctx.versions, ctx.pricingModeByItemId) + : null + if (channelResolution) { + desired.push(channelResolution) + } else if (code) { + result.missing += 1 + // 已在队列里的缺失项不再重复写三条 SQL,等价格补齐后自然收口。 + if (!openMissing.has(Number(item.id))) { + await upsertAggregatedException({ + reason: 'cost_missing', + scopeKey: `cost_missing|${item.fulfillment_channel}|${item.canonical_product_id}|${item.work_product_rule_id ?? '-'}`, + orderItemId: item.id, + detail: { channel: item.fulfillment_channel }, + }) } } - if (supersedeIds.length === 0 && inserts.length === 0) return - // 替换与写入放在同一事务里,避免事实任务读到「旧行已过期、新行未写入」的空窗。 - const created = await withTransaction(async (client) => { - if (supersedeIds.length > 0) await supersedeSnapshots(supersedeIds, client) - if (inserts.length === 0) return 0 - return insertSnapshots(inserts, client) - }) - result.created += created - if (inserts.length > 0) { - await resolveExceptionsForItems(inserts.map((row) => row.item.id)) - } - } - - const candidateSql = ` - SELECT - oi.id, - oi.order_id, - oi.fulfillment_channel, - oi.canonical_product_id, - GREATEST(COALESCE(oi.normalized_quantity, oi.quantity, 1), 1)::int AS quantity, - COALESCE(o.paid_at, o.created_at) AS basis_at, - (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date::text - AS basis_date, - COALESCE(oi.revenue_amount_fen, 0)::bigint AS revenue_fen, - COALESCE(( - SELECT SUM(r.amount_fen) FROM order_refunds r - WHERE r.order_item_id = oi.id AND r.status = 'confirmed' - ), 0)::bigint AS refund_fen, - wo.id AS work_order_id, - wo.product_rule_id AS work_product_rule_id, - wo.status AS work_order_status, - wo.reward_amount - FROM order_items oi - JOIN orders o ON o.id = oi.order_id - LEFT JOIN work_orders wo ON wo.order_item_id = oi.id - WHERE oi.fulfillment_status = 'assigned' - AND oi.fulfillment_channel IN ( - 'affiliate_dash', 'third_party_voucher', 'work_order', 'manual' + if (ctx.genericCostItems.length > 0) { + const generic = resolveGenericCosts( + item, + ctx.genericCostItems, + ctx.versions, + ctx.pricingModeByItemId, ) - ` + desired.push(...generic) + if (generic.length === 0 && ctx.genericConfigured) result.genericMissing += 1 + } + + const existing = existingMap.get(Number(item.id)) || new Map() + for (const resolution of desired) { + const current = existing.get(Number(resolution.costItemId)) + if ( + current && + Number(current.cost_version_id) === Number(resolution.costVersionId) && + Number(current.total_amount_fen) === Number(resolution.totalAmountFen) && + current.snapshot_stage === resolution.stage + ) { + continue + } + if (current) { + const model = ctx.costModelByItemId.get(Number(resolution.costItemId)) || '' + const mutableEstimated = AUTO_GENERIC_MODELS.has(model) + if (!shouldSupersedeSnapshot(current.snapshot_stage, resolution.stage, mutableEstimated)) { + continue + } + supersedeIds.push(Number(current.id)) + result.superseded += 1 + } + inserts.push({ item, resolution }) + affectedDates.add(item.basis_date) + } + } + + if (supersedeIds.length === 0 && inserts.length === 0) return + // 替换与写入放在同一事务里,避免事实任务读到「旧行已过期、新行未写入」的空窗。 + const created = await withTransaction(async (client) => { + if (supersedeIds.length > 0) await supersedeSnapshots(supersedeIds, client) + if (inserts.length === 0) return 0 + return insertSnapshots(inserts, client) + }) + result.created += created + if (inserts.length > 0) { + await resolveExceptionsForItems(inserts.map((row) => row.item.id)) + } +} + +export async function syncChannelCostSnapshots( + options: { batchSize?: number; maxBatches?: number; afterId?: number } = {}, +): 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))) + const singleItem = options.afterId != null + + let coldCursor = singleItem + ? Math.max(0, Math.trunc(options.afterId!)) + : await readWatermarkItemId('cost_snapshot') + let hotCursor = singleItem ? 0 : await readWatermarkItemId('cost_snapshot_hot') + const coldBound = hotCursor + let coldReachedEnd = false + + const ctx = await loadCostContext() + const result = emptyResult(coldCursor, coldCursor, hotCursor) + const affectedDates = new Set() if (singleItem) { const rows = await query( - `${candidateSql} AND oi.id > $1 ORDER BY oi.id LIMIT $2`, + `${CANDIDATE_SQL} AND oi.id > $1 ORDER BY oi.id LIMIT $2`, [coldCursor, Math.min(batchSize, 1)], ) - await processRows(rows.rows) + await processCostRows(rows.rows, ctx, result, affectedDates) if (rows.rows.length > 0) { const last = Number(rows.rows[rows.rows.length - 1]!.id) result.lastId = last @@ -433,13 +474,13 @@ export async function syncChannelCostSnapshots( let hotBatches = 0 for (let batch = 0; batch < maxBatches; batch += 1) { const rows = await query( - `${candidateSql} AND oi.id > $1 ORDER BY oi.id LIMIT $2`, + `${CANDIDATE_SQL} AND oi.id > $1 ORDER BY oi.id LIMIT $2`, [hotCursor, batchSize], ) if (rows.rows.length === 0) break hotBatches += 1 hotCursor = Number(rows.rows[rows.rows.length - 1]!.id) - await processRows(rows.rows) + await processCostRows(rows.rows, ctx, result, affectedDates) if (rows.rows.length < batchSize) break } result.hotCursor = hotCursor @@ -448,7 +489,7 @@ export async function syncChannelCostSnapshots( const remaining = Math.max(0, maxBatches - hotBatches) for (let batch = 0; batch < remaining; batch += 1) { const rows = await query( - `${candidateSql} AND oi.id > $1 AND oi.id <= $3 ORDER BY oi.id LIMIT $2`, + `${CANDIDATE_SQL} AND oi.id > $1 AND oi.id <= $3 ORDER BY oi.id LIMIT $2`, [coldCursor, batchSize, coldBound], ) if (rows.rows.length === 0) { @@ -457,7 +498,7 @@ export async function syncChannelCostSnapshots( } coldCursor = Number(rows.rows[rows.rows.length - 1]!.id) result.lastId = coldCursor - await processRows(rows.rows) + await processCostRows(rows.rows, ctx, result, affectedDates) if (rows.rows.length < batchSize) { coldReachedEnd = true break @@ -469,6 +510,42 @@ export async function syncChannelCostSnapshots( return result } +/** + * 指定订单项立即算成本(渠道判定完成后调用,避免等冷通道扫到)。 + */ +export async function syncCostSnapshotsForItems(itemIds: number[]): Promise<{ + created: number + superseded: number + missing: number + affectedDates: string[] +}> { + const ids = [ + ...new Set(itemIds.map(Number).filter((id) => Number.isInteger(id) && id > 0)), + ].slice(0, 2000) + if (ids.length === 0) return { created: 0, superseded: 0, missing: 0, affectedDates: [] } + const ctx = await loadCostContext() + const result = emptyResult(0, 0, 0) + const affectedDates = new Set() + const rows = await query(`${CANDIDATE_SQL} AND oi.id = ANY($1::bigint[])`, [ + ids, + ]) + await processCostRows(rows.rows, ctx, result, affectedDates) + return { + created: result.created, + superseded: result.superseded, + missing: result.missing, + affectedDates: [...affectedDates].sort(), + } +} + +type ExistingSnapshot = { + id: number + cost_item_id: number + cost_version_id: number + total_amount_fen: number + snapshot_stage: string +} + /** 批量查已入队的成本缺失项,避免每轮重复 upsert。 */ async function loadOpenMissingItems(itemIds: number[]): Promise> { const set = new Set() @@ -488,14 +565,6 @@ async function loadOpenMissingItems(itemIds: number[]): Promise> { return set } -type ExistingSnapshot = { - id: number - cost_item_id: number - cost_version_id: number - total_amount_fen: number - snapshot_stage: string -} - /** 批量读取候选订单项当前有效快照,按订单项分组。 */ async function loadActiveSnapshots( itemIds: number[],