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 c72f57b3..f99c6aa7 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 @@ -76,6 +76,7 @@ function item(patch: Record = {}) { work_order_status: null as string | null, reward_amount: null as number | null, revenue_fen: 10000, + refund_fen: 0, basis_date: '2026-10-31', ...patch, } @@ -274,3 +275,82 @@ test('通用成本模型按商品/全局版本解析且能与渠道成本共存' assert.equal(generic.length, 1) assert.equal(generic[0]?.totalAmountFen, 600) }) + +test('售后损失/其他成本只有商品级版本才自动入账', () => { + const items: CostItemRow[] = [ + { + id: 30, + code: 'sys_after_sales_loss', + pricing_mode: 'per_order', + cost_model: 'after_sales_loss', + }, + ] + const globalVersion: CostVersionRow[] = [ + { + id: 903, + cost_item_id: 30, + scope_type: 'global', + canonical_product_id: null, + work_product_rule_id: null, + amount_fen: 500, + rate_bps: 0, + effective_from: '2026-09-01T00:00:00+08:00', + }, + ] + assert.equal( + resolveGenericCosts(item(), items, globalVersion, new Map([[30, 'per_order']])).length, + 0, + '全局售后损失不能套到所有订单', + ) + + const productVersion: CostVersionRow[] = globalVersion.map((row) => ({ + ...row, + scope_type: 'product', + canonical_product_id: 11, + })) + assert.equal( + resolveGenericCosts(item(), items, productVersion, new Map([[30, 'per_order']])).length, + 1, + '商品级售后损失可以入账', + ) +}) + +test('比例费用按退款后的净收入计算', () => { + const feeItems = new Map([ + [ + 'sys_platform_fee', + { + id: 20, + code: 'sys_platform_fee', + pricing_mode: 'percent_of_revenue', + cost_model: 'platform_fee', + }, + ], + ]) + const feeVersions: CostVersionRow[] = [ + { + id: 904, + cost_item_id: 20, + scope_type: 'global', + canonical_product_id: null, + work_product_rule_id: null, + amount_fen: 0, + rate_bps: 300, + effective_from: '2026-09-01T00:00:00+08:00', + }, + ] + const noRefund = resolveGenericCosts( + item({ revenue_fen: 10000, refund_fen: 0 }), + [...feeItems.values()], + feeVersions, + new Map([[20, 'percent_of_revenue']]), + ) + const withRefund = resolveGenericCosts( + item({ revenue_fen: 10000, refund_fen: 4000 }), + [...feeItems.values()], + feeVersions, + new Map([[20, 'percent_of_revenue']]), + ) + assert.equal(noRefund[0]?.totalAmountFen, 300) + assert.equal(withRefund[0]?.totalAmountFen, 180) +}) 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 369477f3..38df8dd3 100644 --- a/apps/backend/src/services/cost-center/cost-snapshot-service.ts +++ b/apps/backend/src/services/cost-center/cost-snapshot-service.ts @@ -1,4 +1,5 @@ -import { query } from '../../db/client.js' +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' @@ -59,6 +60,7 @@ export type CostCandidateItem = { work_order_status: string | null reward_amount: number | null revenue_fen: number + refund_fen: number basis_date: string } @@ -158,7 +160,8 @@ export function resolveCost( const rate = Number(version.rate_bps) || 0 // 比例成本只看费率:费率为 0 说明还没配价,不能当 0 元成本入账。 if (rate <= 0) return null - const total = Math.round((Number(item.revenue_fen) || 0) * (rate / 10000)) + const netRevenue = Math.max(0, (Number(item.revenue_fen) || 0) - (Number(item.refund_fen) || 0)) + const total = Math.round(netRevenue * (rate / 10000)) return { costItemId: costItem.id, costVersionId: version.id, @@ -184,6 +187,8 @@ export function resolveCost( * 通用成本模型:平台费/支付费/售后损失等,对所有渠道生效。 * 只按商品或全局版本解析,不挂接单规则。 */ +const AUTO_GENERIC_MODELS = new Set(['platform_fee', 'payment_fee']) + export function resolveGenericCosts( item: CostCandidateItem, genericItems: CostItemRow[], @@ -199,13 +204,22 @@ export function resolveGenericCosts( basisAt: item.basis_at, }) if (!version) continue + // 平台费/支付费是「每单都会发生」的通用费用,可按 global/product 自动计; + // 售后损失与其他成本必须显式按商品配置,避免一条 global 版本套到所有订单。 + if (!AUTO_GENERIC_MODELS.has(costItem.cost_model) && version.scope_type !== 'product') { + continue + } const pricingMode = pricingModeByItemId.get(costItem.id) || costItem.pricing_mode const amount = Number(version.amount_fen) || 0 if (pricingMode === 'percent_of_revenue') { const rate = Number(version.rate_bps) || 0 if (rate <= 0) continue - const total = Math.round((Number(item.revenue_fen) || 0) * (rate / 10000)) + const netRevenue = Math.max( + 0, + (Number(item.revenue_fen) || 0) - (Number(item.refund_fen) || 0), + ) + const total = Math.round(netRevenue * (rate / 10000)) result.push({ costItemId: costItem.id, costVersionId: version.id, @@ -234,8 +248,12 @@ export type SyncCostSnapshotsResult = { created: number superseded: number missing: number + /** 通用费用已配置、但该订单项解析不到通用版本的条数(仅提示,不建异常)。 */ + genericMissing: number lastId: number nextCursor: number + /** 热通道游标:本轮之后新增订单的处理起点。 */ + hotCursor: number /** 发生过快照新增/替换的业务日,供事实与总账按日回填。 */ affectedDates: string[] } @@ -245,11 +263,14 @@ 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 = - options.afterId != null - ? Math.max(0, Math.trunc(options.afterId)) - : await readWatermarkItemId('cost_snapshot') - let reachedEnd = false + 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 allCostItems = await loadCostItems() const channelCostItems = new Map() @@ -257,7 +278,9 @@ export async function syncChannelCostSnapshots( for (const item of allCostItems) { if ((GENERIC_COST_MODELS as readonly string[]).includes(item.cost_model)) { genericCostItems.push(item) - } else if (item.code) channelCostItems.set(item.code, item) + } else if (item.code) { + channelCostItems.set(item.code, item) + } } const versions = await loadActiveVersions() const pricingModeByItemId = new Map() @@ -268,57 +291,24 @@ export async function syncChannelCostSnapshots( created: 0, superseded: 0, missing: 0, - lastId: cursor, - nextCursor: cursor, + genericMissing: 0, + lastId: coldCursor, + nextCursor: coldCursor, + hotCursor, affectedDates: [], } const affectedDates = new Set() + const genericConfigured = hasConfiguredGeneric(genericCostItems, versions) - for (let batch = 0; batch < maxBatches; batch += 1) { - const rows = await query( - ` - 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, - 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' - ) - AND oi.id > $1 - ORDER BY oi.id - LIMIT $2 - `, - [cursor, batchSize], - ) - - 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 - - const existingMap = await loadActiveSnapshots(rows.rows.map((row) => row.id)) + 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[] = [] - for (const item of rows.rows) { + for (const item of rows) { const desired: Array> = [] const code = CHANNEL_COST_CODE[String(item.fulfillment_channel || '')] const channelResolution = code @@ -327,18 +317,22 @@ export async function syncChannelCostSnapshots( if (channelResolution) { desired.push(channelResolution) } else if (code) { - // 只有渠道成本项缺失才算异常;人工渠道本来就不写渠道成本。 result.missing += 1 - 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 }, - }) + // 已在队列里的缺失项不再重复写三条 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) { - desired.push(...resolveGenericCosts(item, genericCostItems, versions, pricingModeByItemId)) + 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() @@ -352,6 +346,10 @@ export async function syncChannelCostSnapshots( ) { continue } + // 已经按实际奖励入账的,不允许被后来的预估覆盖(工单回退/取消场景)。 + if (current && current.snapshot_stage === 'actual' && resolution.stage === 'estimated') { + continue + } if (current) { supersedeIds.push(Number(current.id)) result.superseded += 1 @@ -361,20 +359,121 @@ export async function syncChannelCostSnapshots( } } - if (supersedeIds.length > 0) { - await supersedeSnapshots(supersedeIds) - } - if (inserts.length === 0) continue - const created = await insertSnapshots(inserts) + 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 - await resolveExceptionsForItems(inserts.map((row) => row.item.id)) + if (inserts.length > 0) { + await resolveExceptionsForItems(inserts.map((row) => row.item.id)) + } } - result.nextCursor = reachedEnd ? 0 : cursor + 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 (singleItem) { + const rows = await query( + `${candidateSql} AND oi.id > $1 ORDER BY oi.id LIMIT $2`, + [coldCursor, Math.min(batchSize, 1)], + ) + await processRows(rows.rows) + if (rows.rows.length > 0) { + const last = Number(rows.rows[rows.rows.length - 1]!.id) + result.lastId = last + result.nextCursor = last + } + result.affectedDates = [...affectedDates].sort() + return result + } + + // 热通道:只处理上次运行之后新增的订单项,保证新单分钟级入账。 + 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`, + [hotCursor, batchSize], + ) + if (rows.rows.length === 0) break + hotBatches += 1 + hotCursor = Number(rows.rows[rows.rows.length - 1]!.id) + await processRows(rows.rows) + if (rows.rows.length < batchSize) break + } + result.hotCursor = hotCursor + + // 冷通道:从旧游标开始回扫历史订单项,处理补配价格等后置变更。 + 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`, + [coldCursor, batchSize, coldBound], + ) + if (rows.rows.length === 0) { + coldReachedEnd = true + break + } + coldCursor = Number(rows.rows[rows.rows.length - 1]!.id) + result.lastId = coldCursor + await processRows(rows.rows) + if (rows.rows.length < batchSize) { + coldReachedEnd = true + break + } + } + + result.nextCursor = coldReachedEnd ? 0 : coldCursor result.affectedDates = [...affectedDates].sort() return result } +/** 批量查已入队的成本缺失项,避免每轮重复 upsert。 */ +async function loadOpenMissingItems(itemIds: number[]): Promise> { + const set = new Set() + if (itemIds.length === 0) return set + const rows = 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 = 'cost_missing' + AND ce.status <> 'resolved' + `, + [itemIds], + ) + for (const row of rows.rows) set.add(Number(row.order_item_id)) + return set +} + type ExistingSnapshot = { id: number cost_item_id: number @@ -409,9 +508,16 @@ async function loadActiveSnapshots( return map } -async function supersedeSnapshots(ids: number[]): Promise { +function hasConfiguredGeneric(genericItems: CostItemRow[], versions: CostVersionRow[]): boolean { + return genericItems.some((item) => + versions.some((version) => Number(version.cost_item_id) === Number(item.id)), + ) +} + +async function supersedeSnapshots(ids: number[], client?: PoolClient): Promise { if (ids.length === 0) return - await query('UPDATE cost_snapshots SET superseded_at = $1 WHERE id = ANY($2::bigint[])', [ + const run = client ? client.query.bind(client) : query + await run('UPDATE cost_snapshots SET superseded_at = $1 WHERE id = ANY($2::bigint[])', [ nowIso(), ids, ]) @@ -419,9 +525,11 @@ async function supersedeSnapshots(ids: number[]): Promise { async function insertSnapshots( rows: Array<{ item: CostCandidateItem; resolution: NonNullable }>, + client?: PoolClient, ): Promise { const now = nowIso() - const result = await query( + const run = client ? client.query.bind(client) : query + const result = await run( ` INSERT INTO cost_snapshots ( subject_type, subject_id, order_id, order_item_id, work_order_id, diff --git a/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts b/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts index a332ef82..853a2df1 100644 --- a/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts +++ b/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts @@ -52,9 +52,11 @@ export async function runChannelAssignJob(job: JsonObject = {}) { status: 'ok', message: `渠道判定:扫描 ${result.scanned},已定 ${result.assigned},` + - `冲突 ${result.conflicted},仍待定 ${result.stillPending},回填 ${refreshed} 天`, + `冲突 ${result.conflicted},仍待定 ${result.stillPending},回填 ${refreshed.days} 天` + + (refreshed.skippedClosed > 0 ? `(${refreshed.skippedClosed} 天已关账跳过)` : ''), ...result, - refreshedDays: refreshed, + refreshedDays: refreshed.days, + skippedClosedDays: refreshed.skippedClosed, } } @@ -64,19 +66,49 @@ export async function runChannelAssignJob(job: JsonObject = {}) { * 事实任务只回看最近几天;补配价格、迟到分类这类变更必须按受影响日期重算, * 否则旧日期的毛利会一直停在缺失成本/未分类。 */ -async function refreshFactsForDates(dates: string[]): Promise { - const unique = [...new Set(dates.filter(Boolean))].sort().slice(0, 31) +async function refreshFactsForDates( + dates: string[], +): Promise<{ days: number; skippedClosed: number }> { + const unique = [...new Set(dates.filter(Boolean))].sort() + if (unique.length === 0) return { days: 0, skippedClosed: 0 } + + // 日期过多时按整段重算,避免截断导致部分日期永远不更新。 + if (unique.length > 31) { + const dateFrom = unique[0]! + const dateTo = unique[unique.length - 1]! + try { + const facts = await rebuildChannelSalesDailyFacts({ dateFrom, dateTo }) + await syncFinanceLedgerFromFacts({ dateFrom, dateTo }) + if (facts.skippedClosedPeriods.length > 0) { + console.warn('[finance-pipeline] 部分日期已关账,事实未回填', { + dateFrom, + dateTo, + skipped: facts.skippedClosedPeriods, + }) + } + return { days: unique.length, skippedClosed: facts.skippedClosedPeriods.length } + } catch { + return { days: 0, skippedClosed: 0 } + } + } + let done = 0 + let skippedClosed = 0 for (const date of unique) { try { - await rebuildChannelSalesDailyFacts({ dateFrom: date, dateTo: date }) + const facts = await rebuildChannelSalesDailyFacts({ dateFrom: date, dateTo: date }) + if (facts.skippedClosedPeriods.length > 0) { + skippedClosed += facts.skippedClosedPeriods.length + console.warn('[finance-pipeline] 日期已关账,事实未回填', { date }) + continue + } await syncFinanceLedgerFromFacts({ dateFrom: date, dateTo: date }) done += 1 } catch { // 单日失败不影响管道本身,下一轮或人工重跑会再处理。 } } - return done + return { days: done, skippedClosed } } export async function runCarrierDispatchJob(job: JsonObject = {}) { @@ -368,14 +400,18 @@ export async function runCostSnapshotJob(job: JsonObject = {}) { }) await touchWatermark('cost_snapshot', { lastItemId: result.nextCursor }) + await touchWatermark('cost_snapshot_hot', { lastItemId: result.hotCursor }) const refreshed = await refreshFactsForDates(result.affectedDates) return { status: 'ok', message: `成本快照:扫描 ${result.scanned},生成 ${result.created},` + - `替换 ${result.superseded},缺失 ${result.missing},回填 ${refreshed} 天`, + `替换 ${result.superseded},缺失 ${result.missing},回填 ${refreshed.days} 天` + + (result.genericMissing > 0 ? `,通用费用未解析 ${result.genericMissing} 项` : '') + + (refreshed.skippedClosed > 0 ? `(${refreshed.skippedClosed} 天已关账跳过)` : ''), ...result, - refreshedDays: refreshed, + refreshedDays: refreshed.days, + skippedClosedDays: refreshed.skippedClosed, } } diff --git a/apps/backend/src/services/watermark-store.ts b/apps/backend/src/services/watermark-store.ts index 2da03fa0..4ce7b2c6 100644 --- a/apps/backend/src/services/watermark-store.ts +++ b/apps/backend/src/services/watermark-store.ts @@ -12,6 +12,7 @@ export type WatermarkPipeline = | 'carrier_dispatch' | 'product_normalize' | 'cost_snapshot' + | 'cost_snapshot_hot' | 'sales_facts' | 'ledger_post'