fix(cost): 快照新单优先、费用口径与回填健壮性

- 成本快照拆热/冷双通道：新单走热通道分钟级入账，历史项冷通道轮询重算 - 通用成本只允许平台费/支付费自动全量，售后损失/其他必须按商品配置 - 比例费用按扣除已确认退款后的净收入计算 - 已入账 actual 快照不再被后续预估覆盖 - 受影响日期超过 31 天时按整段重算，不再静默丢弃；关账跳过会提示 - 通用费用解析不到时计数告警；已入队的成本缺失不再重复 upsert - 快照替换与写入放入同一事务，避免事实任务读到空窗
2026-11-01 14:46:08 +08:00
parent 34832139b0
commit cf0d600c7d
4 changed files with 304 additions and 79 deletions
@@ -76,6 +76,7 @@ function item(patch: Record<string, unknown> = {}) {
    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<string, CostItemRow>([
+    [
+      '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)
+})
@@ -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<SyncCostSnapshotsResult> {
  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<string, CostItemRow>()
@@ -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<number, string>()
@@ -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<string>()
+  const genericConfigured = hasConfiguredGeneric(genericCostItems, versions)

-  for (let batch = 0; batch < maxBatches; batch += 1) {
-    const rows = await query<CostCandidateItem>(
-      `
-        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<void> {
+    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<CostResolution> }> = []
    const supersedeIds: number[] = []

-    for (const item of rows.rows) {
+    for (const item of rows) {
      const desired: Array<NonNullable<CostResolution>> = []
      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<number, ExistingSnapshot>()
@@ -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<CostCandidateItem>(
+      `${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<CostCandidateItem>(
+      `${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<CostCandidateItem>(
+      `${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<Set<number>> {
+  const set = new Set<number>()
+  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<void> {
+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<void> {
  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<void> {

 async function insertSnapshots(
  rows: Array<{ item: CostCandidateItem; resolution: NonNullable<CostResolution> }>,
+  client?: PoolClient,
 ): Promise<number> {
  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,
@@ -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<number> {
-  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,
  }
 }

@@ -12,6 +12,7 @@ export type WatermarkPipeline =
  | 'carrier_dispatch'
  | 'product_normalize'
  | 'cost_snapshot'
+  | 'cost_snapshot_hot'
  | 'sales_facts'
  | 'ledger_post'