feat(finance): 完成成本快照运行层、销售事实与人工支出登记

迁移 108-111：快照业务时间与预估转实际、销售事实表、系统成本项、公司人工支出登记。接单平台按工单实际 reward 入账；人工支出支持登记/审批/驳回/报销并写 actual 快照。新增销售毛利页、人工支出页与订单入口。

apps/backend/src/services/cost-center/cost-snapshot-sync-service.ts

@@ -0,0 +1,580 @@
+import type { PoolClient } from 'pg'
+
+import { withTransaction } from '../../db/client.js'
+import { resolveFinanceDateRange } from '../../utils/finance-date-range.js'
+
+export type CostSnapshotSyncInput = {
+  dateFrom?: unknown
+  dateTo?: unknown
+  from?: unknown
+  to?: unknown
+}
+
+export type CostSnapshotSyncResult = {
+  range: { fromDate: string; toDate: string }
+  createdCount: number
+  exceptionCreatedCount: number
+  resolvedExceptionCount: number
+}
+
+export type ActualCostSnapshotSyncResult = {
+  range: { fromDate: string; toDate: string }
+  createdCount: number
+  upgradedCount: number
+  directRewardCount: number
+}
+
+/**
+ * 按订单项批量生成预估成本快照。
+ *
+ * 只处理已有 canonical_product_id 的订单项；未归一订单项进入
+ * product_mapping_missing 异常，不在这里猜测商品。
+ */
+export async function syncEstimatedOrderItemCostSnapshots(
+  input: CostSnapshotSyncInput = {},
+): Promise<CostSnapshotSyncResult> {
+  const range = resolveFinanceDateRange(input, { defaultDays: 7, maxDays: 400 })
+
+  return withTransaction(async (client) => {
+    const inserted = await client.query(
+      `
+        WITH item_scope AS (
+          SELECT
+            oi.id AS order_item_id,
+            oi.order_id,
+            o.provider,
+            o.shop_id,
+            oi.sku_code,
+            oi.sku_name,
+            oi.canonical_product_id,
+            COALESCE(o.paid_at, o.created_at) AS basis_at,
+            GREATEST(COALESCE(oi.normalized_quantity, oi.quantity, 1), 1)::numeric AS quantity,
+            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,
+            CASE
+              WHEN wo.id IS NOT NULL THEN 'work_order'
+              WHEN EXISTS (
+                SELECT 1 FROM fulfillment_tasks ft
+                WHERE ft.order_item_id = oi.id
+                  AND ft.executor_key = 'affiliate_dash'
+              ) THEN 'affiliate_dash'
+              ELSE 'manual'
+            END AS carrier_type
+          FROM order_items oi
+          JOIN orders o ON o.id = oi.order_id
+          LEFT JOIN LATERAL (
+            SELECT id, product_rule_id
+            FROM work_orders
+            WHERE order_item_id = oi.id
+            ORDER BY id DESC
+            LIMIT 1
+          ) wo ON TRUE
+          WHERE o.pay_status IN ('paid', 'refunded')
+            AND (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date
+              BETWEEN $1::date AND $2::date
+        ),
+        resolved AS (
+          SELECT
+            s.*,
+            ci.id AS cost_item_id,
+            ci.pricing_mode,
+            v.id AS cost_version_id,
+            v.amount_fen,
+            v.rate_bps
+          FROM item_scope s
+          JOIN cost_items ci ON ci.status = 'active'
+            AND ci.carrier_type IN (s.carrier_type, 'shared')
+            AND ci.pricing_mode IN ('per_item', 'per_order', 'percent_of_revenue')
+            AND ci.code <> 'sys_work_order_reward'
+          JOIN LATERAL (
+            SELECT v.id, v.amount_fen, v.rate_bps
+            FROM cost_versions v
+            WHERE v.cost_item_id = ci.id
+              AND v.status = 'active'
+              AND v.effective_from <= s.basis_at
+              AND (v.effective_to IS NULL OR v.effective_to > s.basis_at)
+              AND (
+                (v.scope_type = 'rule' AND s.work_product_rule_id IS NOT NULL
+                  AND v.work_product_rule_id = s.work_product_rule_id)
+                OR (v.scope_type = 'product' AND s.canonical_product_id IS NOT NULL
+                  AND v.canonical_product_id = s.canonical_product_id)
+                OR (v.scope_type = 'shop' AND v.provider = s.provider AND v.shop_id = s.shop_id)
+                OR (v.scope_type = 'provider' AND v.provider = s.provider)
+                OR (v.scope_type = 'global')
+              )
+            ORDER BY
+              CASE v.scope_type
+                WHEN 'rule' THEN 1
+                WHEN 'product' THEN 2
+                WHEN 'shop' THEN 3
+                WHEN 'provider' THEN 4
+                ELSE 5
+              END,
+              v.effective_from DESC,
+              v.id DESC
+            LIMIT 1
+          ) v ON TRUE
+          WHERE s.canonical_product_id IS NOT NULL
+        ),
+        calculated AS (
+          SELECT
+            r.*,
+            CASE r.pricing_mode
+              WHEN 'per_item' THEN ROUND(r.amount_fen::numeric * r.quantity)::bigint
+              WHEN 'per_order' THEN r.amount_fen
+              WHEN 'percent_of_revenue'
+                THEN ROUND((r.revenue_fen::numeric * r.rate_bps::numeric) / 10000.0)::bigint
+              ELSE r.amount_fen
+            END AS total_amount_fen
+          FROM resolved r
+        )
+        INSERT INTO cost_snapshots (
+          subject_type, subject_id, order_id, order_item_id, work_order_id,
+          carrier_type, cost_item_id, cost_version_id, canonical_product_id,
+          quantity, unit_amount_fen, total_amount_fen, snapshot_stage, source,
+          metadata_json, cost_basis_at, resolved_at, created_at
+        )
+        SELECT
+          'order_item', c.order_item_id, c.order_id, c.order_item_id, c.work_order_id,
+          c.carrier_type, c.cost_item_id, c.cost_version_id, c.canonical_product_id,
+          CASE WHEN c.pricing_mode = 'per_item' THEN c.quantity ELSE 1 END,
+          CASE WHEN c.pricing_mode = 'percent_of_revenue' THEN c.total_amount_fen ELSE c.amount_fen END,
+          c.total_amount_fen,
+          'estimated', 'resolved',
+          jsonb_build_object(
+            'provider', c.provider,
+            'shopId', c.shop_id,
+            'skuCode', c.sku_code,
+            'skuName', c.sku_name,
+            'pricingMode', c.pricing_mode,
+            'revenueFen', c.revenue_fen,
+            'normalizedQuantity', c.quantity,
+            'basisAt', c.basis_at
+          ),
+          c.basis_at, NOW(), NOW()
+        FROM calculated c
+        ON CONFLICT (subject_type, subject_id, cost_item_id)
+          WHERE superseded_at IS NULL
+        DO NOTHING
+      `,
+      [range.fromDate, range.toDate],
+    )
+
+    const productExceptions = await insertAndResolveProductMappingExceptions(
+      client,
+      range.fromDate,
+      range.toDate,
+    )
+    const costExceptions = await insertAndResolveCostMissingExceptions(
+      client,
+      range.fromDate,
+      range.toDate,
+    )
+
+    return {
+      range,
+      createdCount: inserted.rowCount || 0,
+      exceptionCreatedCount: productExceptions.created + costExceptions.created,
+      resolvedExceptionCount: productExceptions.resolved + costExceptions.resolved,
+    }
+  })
+}
+
+/**
+ * 工单验收后，把订单项当前预估快照升级为实际快照。
+ *
+ * 如果配置了 work_order_reward 成本项，则用工单实际 reward_amount 覆盖金额；
+ * 否则仍冻结当前版本金额，但把阶段标为 actual。旧快照保留 superseded_at。
+ */
+export async function syncActualWorkOrderCostSnapshots(
+  input: CostSnapshotSyncInput = {},
+): Promise<ActualCostSnapshotSyncResult> {
+  const range = resolveFinanceDateRange(input, { defaultDays: 30, maxDays: 400 })
+
+  return withTransaction(async (client) => {
+    const directReward = await syncDirectWorkOrderRewardSnapshots(
+      client,
+      range.fromDate,
+      range.toDate,
+    )
+    const upgraded = await client.query(
+      `
+        WITH accepted AS (
+          SELECT DISTINCT ON (wo.order_item_id)
+            wo.id AS work_order_id,
+            wo.order_item_id,
+            GREATEST(COALESCE(wo.reward_amount, 0), 0)::bigint AS reward_amount,
+            COALESCE(wo.accepted_at, wo.updated_at, o.paid_at, o.created_at, NOW()) AS accepted_at
+          FROM work_orders wo
+          JOIN order_items oi ON oi.id = wo.order_item_id
+          JOIN orders o ON o.id = oi.order_id
+          WHERE wo.status = 'accepted'
+            AND wo.order_item_id IS NOT NULL
+            AND (COALESCE(wo.accepted_at, wo.updated_at, o.paid_at, o.created_at)
+              AT TIME ZONE 'Asia/Shanghai')::date BETWEEN $1::date AND $2::date
+          ORDER BY wo.order_item_id, wo.id DESC
+        ),
+        active AS (
+          SELECT
+            cs.*,
+            a.work_order_id AS actual_work_order_id,
+            a.reward_amount,
+            a.accepted_at,
+            ci.cost_type
+          FROM cost_snapshots cs
+          JOIN accepted a ON cs.subject_type = 'order_item' AND cs.subject_id = a.order_item_id
+          JOIN cost_items ci ON ci.id = cs.cost_item_id
+          WHERE cs.superseded_at IS NULL
+            AND cs.snapshot_stage IN ('estimated', 'actual')
+            AND ci.code <> 'sys_work_order_reward'
+        ),
+        target AS (
+          SELECT *
+          FROM active
+          WHERE snapshot_stage = 'estimated'
+             OR total_amount_fen <> CASE
+                  WHEN cost_type = 'work_order_reward' AND reward_amount > 0 THEN reward_amount
+                  ELSE total_amount_fen
+                END
+        ),
+        superseded AS (
+          UPDATE cost_snapshots cs
+          SET superseded_at = NOW()
+          FROM target t
+          WHERE cs.id = t.id
+            AND cs.superseded_at IS NULL
+          RETURNING
+            cs.id, cs.subject_id, cs.order_id, cs.order_item_id, cs.carrier_type,
+            cs.cost_item_id, cs.cost_version_id, cs.canonical_product_id,
+            cs.quantity, cs.unit_amount_fen, cs.total_amount_fen, cs.metadata_json
+        )
+        INSERT INTO cost_snapshots (
+          subject_type, subject_id, order_id, order_item_id, work_order_id,
+          carrier_type, cost_item_id, cost_version_id, canonical_product_id,
+          quantity, unit_amount_fen, total_amount_fen, snapshot_stage, source,
+          metadata_json, cost_basis_at, resolved_at, created_at
+        )
+        SELECT
+          'order_item', s.subject_id, s.order_id, s.order_item_id, a.work_order_id,
+          s.carrier_type, s.cost_item_id, s.cost_version_id, s.canonical_product_id,
+          CASE WHEN ci.cost_type = 'work_order_reward' AND a.reward_amount > 0 THEN 1
+            ELSE s.quantity END,
+          CASE WHEN ci.cost_type = 'work_order_reward' AND a.reward_amount > 0
+            THEN a.reward_amount ELSE s.unit_amount_fen END,
+          CASE WHEN ci.cost_type = 'work_order_reward' AND a.reward_amount > 0
+            THEN a.reward_amount ELSE s.total_amount_fen END,
+          'actual', 'resolved',
+          s.metadata_json || jsonb_build_object(
+            'actualWorkOrderId', a.work_order_id,
+            'actualRewardFen', a.reward_amount,
+            'basisAt', a.accepted_at
+          ),
+          a.accepted_at, NOW(), NOW()
+        FROM superseded s
+        JOIN accepted a ON a.order_item_id = s.subject_id
+        JOIN cost_items ci ON ci.id = s.cost_item_id
+        ON CONFLICT (subject_type, subject_id, cost_item_id)
+          WHERE superseded_at IS NULL
+        DO NOTHING
+      `,
+      [range.fromDate, range.toDate],
+    )
+
+    return {
+      range,
+      createdCount: directReward.inserted + (upgraded.rowCount || 0),
+      upgradedCount: upgraded.rowCount || 0,
+      directRewardCount: directReward.inserted,
+    }
+  })
+}
+
+/**
+ * 接单平台的实际成本不依赖预估快照：工单验收后直接按 reward_amount 写 actual。
+ * 系统成本项 sys_work_order_reward 的版本金额只是占位，不会进入预估阶段。
+ */
+async function syncDirectWorkOrderRewardSnapshots(
+  client: PoolClient,
+  fromDate: string,
+  toDate: string,
+) {
+  const superseded = await client.query(
+    `
+      WITH system_item AS (
+        SELECT ci.id AS cost_item_id, v.id AS cost_version_id
+        FROM cost_items ci
+        JOIN LATERAL (
+          SELECT id
+          FROM cost_versions
+          WHERE cost_item_id = ci.id
+            AND status = 'active'
+            AND effective_from <= NOW()
+          ORDER BY effective_from DESC, id DESC
+          LIMIT 1
+        ) v ON TRUE
+        WHERE ci.code = 'sys_work_order_reward'
+          AND ci.status = 'active'
+      ),
+      accepted AS (
+        SELECT DISTINCT ON (wo.order_item_id)
+          wo.id AS work_order_id,
+          wo.order_item_id,
+          GREATEST(COALESCE(wo.reward_amount, 0), 0)::bigint AS reward_amount,
+          COALESCE(wo.accepted_at, wo.updated_at, o.paid_at, o.created_at, NOW()) AS accepted_at
+        FROM work_orders wo
+        JOIN order_items oi ON oi.id = wo.order_item_id
+        JOIN orders o ON o.id = oi.order_id
+        WHERE wo.status = 'accepted'
+          AND wo.order_item_id IS NOT NULL
+          AND COALESCE(wo.reward_amount, 0) > 0
+          AND (COALESCE(wo.accepted_at, wo.updated_at, o.paid_at, o.created_at)
+            AT TIME ZONE 'Asia/Shanghai')::date BETWEEN $1::date AND $2::date
+        ORDER BY wo.order_item_id, wo.id DESC
+      )
+      UPDATE cost_snapshots cs
+      SET superseded_at = NOW()
+      FROM accepted a
+      JOIN system_item si ON TRUE
+      JOIN LATERAL (
+        SELECT id, snapshot_stage, total_amount_fen
+        FROM cost_snapshots
+        WHERE subject_type = 'order_item'
+          AND subject_id = a.order_item_id
+          AND cost_item_id = si.cost_item_id
+          AND superseded_at IS NULL
+        ORDER BY id DESC
+        LIMIT 1
+        FOR UPDATE
+      ) current ON TRUE
+      WHERE cs.id = current.id
+        AND cs.superseded_at IS NULL
+        AND (current.snapshot_stage <> 'actual' OR current.total_amount_fen <> a.reward_amount)
+    `,
+    [fromDate, toDate],
+  )
+
+  const inserted = await client.query(
+    `
+      WITH system_item AS (
+        SELECT ci.id AS cost_item_id, v.id AS cost_version_id
+        FROM cost_items ci
+        JOIN LATERAL (
+          SELECT id
+          FROM cost_versions
+          WHERE cost_item_id = ci.id
+            AND status = 'active'
+            AND effective_from <= NOW()
+          ORDER BY effective_from DESC, id DESC
+          LIMIT 1
+        ) v ON TRUE
+        WHERE ci.code = 'sys_work_order_reward'
+          AND ci.status = 'active'
+      ),
+      accepted AS (
+        SELECT DISTINCT ON (wo.order_item_id)
+          wo.id AS work_order_id,
+          wo.order_item_id,
+          GREATEST(COALESCE(wo.reward_amount, 0), 0)::bigint AS reward_amount,
+          COALESCE(wo.accepted_at, wo.updated_at, o.paid_at, o.created_at, NOW()) AS accepted_at
+        FROM work_orders wo
+        JOIN order_items oi ON oi.id = wo.order_item_id
+        JOIN orders o ON o.id = oi.order_id
+        WHERE wo.status = 'accepted'
+          AND wo.order_item_id IS NOT NULL
+          AND COALESCE(wo.reward_amount, 0) > 0
+          AND (COALESCE(wo.accepted_at, wo.updated_at, o.paid_at, o.created_at)
+            AT TIME ZONE 'Asia/Shanghai')::date BETWEEN $1::date AND $2::date
+        ORDER BY wo.order_item_id, wo.id DESC
+      )
+      INSERT INTO cost_snapshots (
+        subject_type, subject_id, order_id, order_item_id, work_order_id,
+        carrier_type, cost_item_id, cost_version_id, canonical_product_id,
+        quantity, unit_amount_fen, total_amount_fen, snapshot_stage, source,
+        metadata_json, cost_basis_at, resolved_at, created_at
+      )
+      SELECT
+        'order_item', a.order_item_id, oi.order_id, a.order_item_id, a.work_order_id,
+        'work_order', si.cost_item_id, si.cost_version_id, oi.canonical_product_id,
+        1, a.reward_amount, a.reward_amount, 'actual', 'resolved',
+        jsonb_build_object(
+          'source', 'system_work_order_reward',
+          'actualWorkOrderId', a.work_order_id,
+          'actualRewardFen', a.reward_amount,
+          'basisAt', a.accepted_at
+        ),
+        a.accepted_at, NOW(), NOW()
+      FROM accepted a
+      CROSS JOIN system_item si
+      JOIN order_items oi ON oi.id = a.order_item_id
+      WHERE NOT EXISTS (
+        SELECT 1 FROM cost_snapshots cs
+        WHERE cs.subject_type = 'order_item'
+          AND cs.subject_id = a.order_item_id
+          AND cs.cost_item_id = si.cost_item_id
+          AND cs.superseded_at IS NULL
+          AND cs.snapshot_stage = 'actual'
+          AND cs.total_amount_fen = a.reward_amount
+      )
+      ON CONFLICT (subject_type, subject_id, cost_item_id)
+        WHERE superseded_at IS NULL
+      DO NOTHING
+    `,
+    [fromDate, toDate],
+  )
+
+  return {
+    inserted: inserted.rowCount || 0,
+    superseded: superseded.rowCount || 0,
+  }
+}
+
+async function insertAndResolveProductMappingExceptions(
+  client: PoolClient,
+  fromDate: string,
+  toDate: string,
+) {
+  const inserted = await client.query(
+    `
+      WITH item_scope AS (
+        SELECT
+          oi.id AS order_item_id,
+          o.provider,
+          o.shop_id,
+          oi.sku_code,
+          oi.sku_name,
+          oi.canonical_product_id,
+          COALESCE(o.paid_at, o.created_at) AS basis_at,
+          wo.id AS work_order_id,
+          CASE
+            WHEN wo.id IS NOT NULL THEN 'work_order'
+            WHEN EXISTS (
+              SELECT 1 FROM fulfillment_tasks ft
+              WHERE ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash'
+            ) THEN 'affiliate_dash'
+            ELSE 'manual'
+          END AS carrier_type
+        FROM order_items oi
+        JOIN orders o ON o.id = oi.order_id
+        LEFT JOIN LATERAL (
+          SELECT id FROM work_orders WHERE order_item_id = oi.id ORDER BY id DESC LIMIT 1
+        ) wo ON TRUE
+        WHERE o.pay_status IN ('paid', 'refunded')
+          AND (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date
+            BETWEEN $1::date AND $2::date
+      )
+      INSERT INTO cost_exceptions (
+        subject_type, subject_id, order_item_id, work_order_id,
+        carrier_type, reason, status, detail_json, created_at
+      )
+      SELECT
+        'order_item', s.order_item_id, s.order_item_id, s.work_order_id,
+        s.carrier_type, 'product_mapping_missing', 'open',
+        jsonb_build_object(
+          'provider', s.provider,
+          'shopId', s.shop_id,
+          'skuCode', s.sku_code,
+          'skuName', s.sku_name,
+          'basisAt', s.basis_at,
+          'message', '订单项没有归一到标准商品，销售事实会进入未归一桶'
+        ),
+        NOW()
+      FROM item_scope s
+      WHERE s.canonical_product_id IS NULL
+      ON CONFLICT (subject_type, subject_id, reason) DO NOTHING
+    `,
+    [fromDate, toDate],
+  )
+
+  const resolved = await client.query(
+    `
+      UPDATE cost_exceptions ce
+      SET status = 'resolved', resolved_at = NOW()
+      WHERE ce.status = 'open'
+        AND ce.subject_type = 'order_item'
+        AND ce.reason = 'product_mapping_missing'
+        AND EXISTS (
+          SELECT 1 FROM order_items oi
+          WHERE oi.id = ce.subject_id AND oi.canonical_product_id IS NOT NULL
+        )
+    `,
+  )
+
+  return { created: inserted.rowCount || 0, resolved: resolved.rowCount || 0 }
+}
+
+async function insertAndResolveCostMissingExceptions(
+  client: PoolClient,
+  fromDate: string,
+  toDate: string,
+) {
+  const inserted = await client.query(
+    `
+      WITH item_scope AS (
+        SELECT
+          oi.id AS order_item_id,
+          oi.canonical_product_id,
+          COALESCE(o.paid_at, o.created_at) AS basis_at,
+          wo.id AS work_order_id,
+          CASE
+            WHEN wo.id IS NOT NULL THEN 'work_order'
+            WHEN EXISTS (
+              SELECT 1 FROM fulfillment_tasks ft
+              WHERE ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash'
+            ) THEN 'affiliate_dash'
+            ELSE 'manual'
+          END AS carrier_type
+        FROM order_items oi
+        JOIN orders o ON o.id = oi.order_id
+        LEFT JOIN LATERAL (
+          SELECT id FROM work_orders WHERE order_item_id = oi.id ORDER BY id DESC LIMIT 1
+        ) wo ON TRUE
+        WHERE o.pay_status IN ('paid', 'refunded')
+          AND (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date
+            BETWEEN $1::date AND $2::date
+      )
+      INSERT INTO cost_exceptions (
+        subject_type, subject_id, order_item_id, work_order_id,
+        carrier_type, reason, status, detail_json, created_at
+      )
+      SELECT
+        'order_item', s.order_item_id, s.order_item_id, s.work_order_id,
+        s.carrier_type, 'cost_missing', 'open',
+        jsonb_build_object(
+          'canonicalProductId', s.canonical_product_id,
+          'carrierType', s.carrier_type,
+          'basisAt', s.basis_at,
+          'message', '订单项没有匹配到任何成本版本'
+        ),
+        NOW()
+      FROM item_scope s
+      WHERE s.canonical_product_id IS NOT NULL
+        AND NOT EXISTS (
+          SELECT 1 FROM cost_snapshots cs
+          WHERE cs.subject_type = 'order_item'
+            AND cs.subject_id = s.order_item_id
+            AND cs.superseded_at IS NULL
+        )
+      ON CONFLICT (subject_type, subject_id, reason) DO NOTHING
+    `,
+    [fromDate, toDate],
+  )
+
+  const resolved = await client.query(
+    `
+      UPDATE cost_exceptions ce
+      SET status = 'resolved', resolved_at = NOW()
+      WHERE ce.status = 'open'
+        AND ce.subject_type = 'order_item'
+        AND ce.reason = 'cost_missing'
+        AND EXISTS (
+          SELECT 1 FROM cost_snapshots cs
+          WHERE cs.subject_type = 'order_item'
+            AND cs.subject_id = ce.subject_id
+            AND cs.superseded_at IS NULL
+        )
+    `,
+  )
+
+  return { created: inserted.rowCount || 0, resolved: resolved.rowCount || 0 }
+}