fix(cost): 快照变更即重算,0 元占位视为缺失,通用成本入账

- 预估→实际、补配价格时 supersede 旧快照再写新行,不再一次写死
- 全局 0 元版本不产生快照,转入 cost_missing
- 比例成本按 rate_bps 计算,费率为 0 视为未配置
- 平台费/支付费/售后等通用成本模型按渠道+通用各写一条快照
This commit is contained in:
yml2213
2026-10-31 20:41:38 +08:00
parent fef1d0b77e
commit 42d5874411
2 changed files with 286 additions and 30 deletions
@@ -3,6 +3,7 @@ import test from 'node:test'
import {
resolveCost,
resolveGenericCosts,
selectCostVersion,
type CostItemRow,
type CostVersionRow,
@@ -74,6 +75,8 @@ function item(patch: Record<string, unknown> = {}) {
work_product_rule_id: null as number | null,
work_order_status: null as string | null,
reward_amount: null as number | null,
revenue_fen: 10000,
basis_date: '2026-10-31',
...patch,
}
}
@@ -172,3 +175,102 @@ test('成本解析:人工渠道不在这里算成本', () => {
)
assert.equal(resolution, null)
})
test('全局 0 元占位版本不能当有效成本', () => {
const zeroVersions: CostVersionRow[] = [
{
id: 900,
cost_item_id: 1,
scope_type: 'global',
canonical_product_id: null,
work_product_rule_id: null,
amount_fen: 0,
rate_bps: 0,
effective_from: '2026-09-01T00:00:00+08:00',
},
]
assert.equal(
resolveCost(
item({ fulfillment_channel: 'work_order', canonical_product_id: null }),
costItems,
zeroVersions,
pricingModes,
),
null,
)
})
test('比例成本按费率计算,费率为 0 视为未配置', () => {
const percentItems = new Map<string, CostItemRow>([
[
'sys_work_order_reward',
{
id: 1,
code: 'sys_work_order_reward',
pricing_mode: 'percent_of_revenue',
cost_model: 'work_order_reward',
},
],
])
const percentVersions: CostVersionRow[] = [
{
id: 901,
cost_item_id: 1,
scope_type: 'rule',
canonical_product_id: null,
work_product_rule_id: 9,
amount_fen: 0,
rate_bps: 500,
effective_from: '2026-09-01T00:00:00+08:00',
},
]
const resolution = resolveCost(
item({ fulfillment_channel: 'work_order', work_product_rule_id: 9, revenue_fen: 10000 }),
percentItems,
percentVersions,
new Map([[1, 'percent_of_revenue']]),
)
assert.equal(resolution?.totalAmountFen, 500)
const zeroRate = percentVersions.map((row) => ({ ...row, rate_bps: 0 }))
assert.equal(
resolveCost(
item({ fulfillment_channel: 'work_order', work_product_rule_id: 9 }),
percentItems,
zeroRate,
new Map([[1, 'percent_of_revenue']]),
),
null,
)
})
test('通用成本模型按商品/全局版本解析且能与渠道成本共存', () => {
const genericItems: CostItemRow[] = [
{
id: 20,
code: 'sys_platform_fee',
pricing_mode: 'percent_of_revenue',
cost_model: 'platform_fee',
},
]
const versionsWithGeneric: CostVersionRow[] = [
{
id: 902,
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 generic = resolveGenericCosts(
item({ revenue_fen: 20000 }),
genericItems,
versionsWithGeneric,
new Map([[20, 'percent_of_revenue']]),
)
assert.equal(generic.length, 1)
assert.equal(generic[0]?.totalAmountFen, 600)
})
@@ -1,5 +1,6 @@
import { query } from '../../db/client.js'
import { nowIso } from '../../utils/time.js'
import { GENERIC_COST_MODELS } from './cost-center-constants.js'
import { resolveExceptionsForItems, upsertAggregatedException } from './cost-exception-service.js'
import { readWatermarkItemId } from '../watermark-store.js'
@@ -12,8 +13,14 @@ import { readWatermarkItemId } from '../watermark-store.js'
* 渠道 → 成本项:
* affiliate_dash → sys_affiliate_purchase(按商品)
* third_party_voucher → sys_third_party_voucher_cost(按商品)
* work_order → sys_work_order_reward(优先按接单规则,其次按商品)
* manual → 跳过(确认履约时按实际录入)
* work_order → sys_work_order_reward(优先按接单规则,其次按商品;已接单按实际奖励)
* manual → 不写渠道成本(确认履约时按实际录入)
*
* 通用成本模型(platform_fee / payment_fee / after_sales_loss)对所有渠道生效,
* 与渠道成本项各写一条快照;解析不到版本时静默跳过,不进异常队列。
*
* 快照不是「一次性」:每轮对比当前解析结果与已有有效快照,金额/版本/阶段变化时
* 先 supersede 旧行再写新行,保证「预估 → 实际」「补配价格」都能被纠正。
*/
const CHANNEL_COST_CODE: Record<string, string> = {
@@ -26,6 +33,7 @@ export type CostItemRow = {
id: number
code: string
pricing_mode: string
cost_model: string
}
export type CostVersionRow = {
@@ -35,6 +43,7 @@ export type CostVersionRow = {
canonical_product_id: number | null
work_product_rule_id: number | null
amount_fen: number
rate_bps: number
effective_from: string
}
@@ -49,6 +58,8 @@ export type CostCandidateItem = {
work_product_rule_id: number | null
work_order_status: string | null
reward_amount: number | null
revenue_fen: number
basis_date: string
}
export type CostResolution = {
@@ -142,8 +153,24 @@ export function resolveCost(
const pricingMode = pricingModeByItemId.get(costItem.id) || costItem.pricing_mode
const amount = Number(version.amount_fen) || 0
const quantity = Math.max(1, Number(item.quantity) || 1)
const total = pricingMode === 'per_item' ? amount * quantity : amount
if (pricingMode === 'percent_of_revenue') {
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))
return {
costItemId: costItem.id,
costVersionId: version.id,
unitAmountFen: total,
totalAmountFen: total,
stage: 'estimated',
}
}
// 全局 0 元是占位版本,不是真实成本;否则毛利会被静默抬高。
if (amount <= 0) return null
const total = pricingMode === 'per_item' ? amount * quantity : amount
return {
costItemId: costItem.id,
costVersionId: version.id,
@@ -153,12 +180,64 @@ export function resolveCost(
}
}
/**
* 通用成本模型:平台费/支付费/售后损失等,对所有渠道生效。
* 只按商品或全局版本解析,不挂接单规则。
*/
export function resolveGenericCosts(
item: CostCandidateItem,
genericItems: CostItemRow[],
versions: CostVersionRow[],
pricingModeByItemId: Map<number, string>,
): Array<NonNullable<CostResolution>> {
const result: Array<NonNullable<CostResolution>> = []
for (const costItem of genericItems) {
const version = selectCostVersion(versions, {
costItemId: costItem.id,
productId: item.canonical_product_id,
ruleId: null,
basisAt: item.basis_at,
})
if (!version) 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))
result.push({
costItemId: costItem.id,
costVersionId: version.id,
unitAmountFen: total,
totalAmountFen: total,
stage: 'estimated',
})
continue
}
if (amount <= 0) continue
const quantity = Math.max(1, Number(item.quantity) || 1)
const total = pricingMode === 'per_item' ? amount * quantity : amount
result.push({
costItemId: costItem.id,
costVersionId: version.id,
unitAmountFen: amount,
totalAmountFen: total,
stage: 'estimated',
})
}
return result
}
export type SyncCostSnapshotsResult = {
scanned: number
created: number
superseded: number
missing: number
lastId: number
nextCursor: number
/** 发生过快照新增/替换的业务日,供事实与总账按日回填。 */
affectedDates: string[]
}
export async function syncChannelCostSnapshots(
@@ -172,18 +251,28 @@ export async function syncChannelCostSnapshots(
: await readWatermarkItemId('cost_snapshot')
let reachedEnd = false
const costItems = await loadCostItems()
const allCostItems = await loadCostItems()
const channelCostItems = new Map<string, CostItemRow>()
const genericCostItems: CostItemRow[] = []
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)
}
const versions = await loadActiveVersions()
const pricingModeByItemId = new Map<number, string>()
for (const item of costItems.values()) pricingModeByItemId.set(item.id, item.pricing_mode)
for (const item of allCostItems) pricingModeByItemId.set(item.id, item.pricing_mode)
const result: SyncCostSnapshotsResult = {
scanned: 0,
created: 0,
superseded: 0,
missing: 0,
lastId: cursor,
nextCursor: cursor,
affectedDates: [],
}
const affectedDates = new Set<string>()
for (let batch = 0; batch < maxBatches; batch += 1) {
const rows = await query<CostCandidateItem>(
@@ -195,6 +284,9 @@ export async function syncChannelCostSnapshots(
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,
@@ -203,17 +295,8 @@ export async function syncChannelCostSnapshots(
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')
-- 接单平台的成本可以只靠规则版本,不要求商品已归一;
-- 其余渠道必须有商品。
AND (
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.fulfillment_channel IN (
'affiliate_dash', 'third_party_voucher', 'work_order', 'manual'
)
AND oi.id > $1
ORDER BY oi.id
@@ -231,14 +314,20 @@ export async function syncChannelCostSnapshots(
cursor = Number(rows.rows[rows.rows.length - 1]!.id)
result.lastId = cursor
const inserts: Array<{
item: CostCandidateItem
resolution: NonNullable<CostResolution>
}> = []
const existingMap = await loadActiveSnapshots(rows.rows.map((row) => row.id))
const inserts: Array<{ item: CostCandidateItem; resolution: NonNullable<CostResolution> }> = []
const supersedeIds: number[] = []
for (const item of rows.rows) {
const resolution = resolveCost(item, costItems, versions, pricingModeByItemId)
if (!resolution) {
const desired: Array<NonNullable<CostResolution>> = []
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
await upsertAggregatedException({
reason: 'cost_missing',
@@ -246,11 +335,35 @@ export async function syncChannelCostSnapshots(
orderItemId: item.id,
detail: { channel: item.fulfillment_channel },
})
continue
}
inserts.push({ item, resolution })
if (genericCostItems.length > 0) {
desired.push(...resolveGenericCosts(item, genericCostItems, versions, pricingModeByItemId))
}
const existing = existingMap.get(Number(item.id)) || new Map<number, ExistingSnapshot>()
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) {
supersedeIds.push(Number(current.id))
result.superseded += 1
}
inserts.push({ item, resolution })
affectedDates.add(item.basis_date)
}
}
if (supersedeIds.length > 0) {
await supersedeSnapshots(supersedeIds)
}
if (inserts.length === 0) continue
const created = await insertSnapshots(inserts)
result.created += created
@@ -258,9 +371,52 @@ export async function syncChannelCostSnapshots(
}
result.nextCursor = reachedEnd ? 0 : cursor
result.affectedDates = [...affectedDates].sort()
return result
}
type ExistingSnapshot = {
id: number
cost_item_id: number
cost_version_id: number
total_amount_fen: number
snapshot_stage: string
}
/** 批量读取候选订单项当前有效快照,按订单项分组。 */
async function loadActiveSnapshots(
itemIds: number[],
): Promise<Map<number, Map<number, ExistingSnapshot>>> {
const map = new Map<number, Map<number, ExistingSnapshot>>()
if (itemIds.length === 0) return map
const rows = await query<ExistingSnapshot & { order_item_id: number }>(
`
SELECT id, order_item_id, cost_item_id, cost_version_id, total_amount_fen, snapshot_stage
FROM cost_snapshots
WHERE superseded_at IS NULL
AND order_item_id = ANY($1::bigint[])
`,
[itemIds],
)
for (const row of rows.rows) {
let perItem = map.get(Number(row.order_item_id))
if (!perItem) {
perItem = new Map<number, ExistingSnapshot>()
map.set(Number(row.order_item_id), perItem)
}
perItem.set(Number(row.cost_item_id), row)
}
return map
}
async function supersedeSnapshots(ids: number[]): Promise<void> {
if (ids.length === 0) return
await query('UPDATE cost_snapshots SET superseded_at = $1 WHERE id = ANY($2::bigint[])', [
nowIso(),
ids,
])
}
async function insertSnapshots(
rows: Array<{ item: CostCandidateItem; resolution: NonNullable<CostResolution> }>,
): Promise<number> {
@@ -322,20 +478,18 @@ async function insertSnapshots(
return result.rowCount || 0
}
async function loadCostItems(): Promise<Map<string, CostItemRow>> {
async function loadCostItems(): Promise<CostItemRow[]> {
const rows = await query<CostItemRow>(
`SELECT id, code, pricing_mode FROM cost_items WHERE status = 'active'`,
`SELECT id, code, pricing_mode, cost_model FROM cost_items WHERE status = 'active'`,
)
const map = new Map<string, CostItemRow>()
for (const row of rows.rows) map.set(row.code, row)
return map
return rows.rows
}
async function loadActiveVersions(): Promise<CostVersionRow[]> {
const rows = await query<CostVersionRow>(
`
SELECT id, cost_item_id, scope_type, canonical_product_id,
work_product_rule_id, amount_fen, effective_from
work_product_rule_id, amount_fen, rate_bps, effective_from
FROM cost_versions
WHERE status = 'active'
`,