fix(cost): 派生费用可随退款重算，成本版本与基线禁止追溯

- 平台费/支付费允许 estimated→estimated 按净收入重算；采购与打手报酬保持固化
- 创建成本版本时拒绝早于当前时间的 effectiveFrom
- 已有快照后禁止导入价格基线；基线生效时间改为当前
- 删除基线导入中前移现有版本生效时间的逻辑

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
     }
 

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('失效时间必须晚于生效时间', {

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)

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<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)))
-  const singleItem = options.afterId != null
+type CostContext = {
+  channelCostItems: Map<string, CostItemRow>
+  genericCostItems: CostItemRow[]
+  versions: CostVersionRow[]
+  pricingModeByItemId: Map<number, string>
+  costModelByItemId: Map<number, string>
+  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<CostContext> {
   const allCostItems = await loadCostItems()
   const channelCostItems = new Map<string, CostItemRow>()
   const genericCostItems: CostItemRow[] = []
@@ -296,130 +339,128 @@ export async function syncChannelCostSnapshots(
   }
   const versions = await loadActiveVersions()
   const pricingModeByItemId = new Map<number, string>()
-  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<number, string>()
+  for (const item of allCostItems) {
+    pricingModeByItemId.set(item.id, item.pricing_mode)
+    costModelByItemId.set(Number(item.id), item.cost_model)
   }
-  const affectedDates = new Set<string>()
-  const genericConfigured = hasConfiguredGeneric(genericCostItems, versions)
+  return {
+    channelCostItems,
+    genericCostItems,
+    versions,
+    pricingModeByItemId,
+    costModelByItemId,
+    genericConfigured: hasConfiguredGeneric(genericCostItems, versions),
+  }
+}
 
-  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[] = []
+async function processCostRows(
+  rows: CostCandidateItem[],
+  ctx: CostContext,
+  result: SyncCostSnapshotsResult,
+  affectedDates: Set<string>,
+): 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) {
-      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
-        // 已在队列里的缺失项不再重复写三条 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<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 && 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<NonNullable<CostResolution>> = []
+    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<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) {
+        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<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)))
+  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<string>()
 
   if (singleItem) {
     const rows = await query<CostCandidateItem>(
-      `${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<CostCandidateItem>(
-      `${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<CostCandidateItem>(
-      `${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<string>()
+  const rows = await query<CostCandidateItem>(`${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<Set<number>> {
   const set = new Set<number>()
@@ -488,14 +565,6 @@ async function loadOpenMissingItems(itemIds: number[]): Promise<Set<number>> {
   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[],