feat(scheduler): 六个财务管道作业，模板改价即同步成本版本

- channel_assign / carrier_dispatch / product_normalize / cost_snapshot / sales_facts / ledger_post
- 每个作业只写自己那一层，失败互不影响，可单独启停
- 接单模板保存后自动同步规则级成本版本，失败不阻塞模板保存

apps/backend/src/services/scheduler/config-service.test.ts

@@ -43,3 +43,72 @@ test('normalizeScheduledJobsConfig ensures default jobs exist', () => {
   assert.ok(ids.includes('pdd_order_sync'))
   assert.ok(ids.includes('cost-finance-sync'))
 })
+
+test('normalizeScheduledJobsConfig 收录 v2 六个管道作业并保持各自配置', () => {
+  const config = normalizeScheduledJobsConfig({
+    enabled: true,
+    jobs: [
+      {
+        id: 'channel-assign',
+        type: 'channel_assign',
+        enabled: true,
+        intervalSeconds: 60,
+        config: { batchSize: 800, maxBatches: 5 },
+      },
+      {
+        id: 'carrier-dispatch',
+        type: 'carrier_dispatch',
+        enabled: false,
+        intervalSeconds: 60,
+        config: { limit: 50 },
+      },
+      {
+        id: 'product-normalize',
+        type: 'product_normalize',
+        enabled: true,
+        intervalSeconds: 300,
+        config: {},
+      },
+      {
+        id: 'cost-snapshot',
+        type: 'cost_snapshot',
+        enabled: true,
+        intervalSeconds: 300,
+        config: {},
+      },
+      {
+        id: 'sales-facts',
+        type: 'sales_facts',
+        enabled: true,
+        intervalSeconds: 600,
+        config: { days: 7 },
+      },
+      {
+        id: 'ledger-post',
+        type: 'ledger_post',
+        enabled: true,
+        intervalSeconds: 600,
+        config: { days: 7 },
+      },
+    ],
+  })
+
+  const byType = new Map(config.jobs.map((item) => [item.type, item]))
+  assert.equal(byType.get('channel_assign')?.config.batchSize, 800)
+  assert.equal(byType.get('carrier_dispatch')?.enabled, false)
+  assert.equal(byType.get('sales_facts')?.config.days, 7)
+  assert.equal(byType.get('ledger_post')?.config.days, 7)
+  assert.ok(byType.has('product_normalize'))
+  assert.ok(byType.has('cost_snapshot'))
+})
+
+test('normalizeScheduledJobsConfig 丢弃未知任务类型', () => {
+  const config = normalizeScheduledJobsConfig({
+    enabled: true,
+    jobs: [{ id: 'nope', type: 'unknown_type', enabled: true, intervalSeconds: 60, config: {} }],
+  })
+  assert.equal(
+    config.jobs.some((item) => item.type === 'unknown_type'),
+    false,
+  )
+})

apps/backend/src/services/scheduler/config-service.ts

@@ -12,6 +12,14 @@ const DATA_RETENTION_JOB_ID = 'data-retention'
 const PDD_ORDER_SYNC_JOB_ID = 'pdd_order_sync'
 const COST_FINANCE_SYNC_JOB_ID = 'cost-finance-sync'
 
+/** v2 管道作业：各自只负责一层，可单独启停。 */
+const CHANNEL_ASSIGN_JOB_ID = 'channel-assign'
+const CARRIER_DISPATCH_JOB_ID = 'carrier-dispatch'
+const PRODUCT_NORMALIZE_JOB_ID = 'product-normalize'
+const COST_SNAPSHOT_JOB_ID = 'cost-snapshot'
+const SALES_FACTS_JOB_ID = 'sales-facts'
+const LEDGER_POST_JOB_ID = 'ledger-post'
+
 export function getScheduledJobsFilePath() {
   return SCHEDULED_JOBS_FILE_PATH
 }
@@ -125,6 +133,24 @@ function normalizeScheduledJob(rawValue: unknown) {
   if (type === 'cost_finance_sync') {
     return normalizeCostFinanceSyncJob(rawValue)
   }
+  if (type === 'channel_assign') {
+    return normalizeChannelAssignJob(rawValue)
+  }
+  if (type === 'carrier_dispatch') {
+    return normalizeCarrierDispatchJob(rawValue)
+  }
+  if (type === 'product_normalize') {
+    return normalizeProductNormalizeJob(rawValue)
+  }
+  if (type === 'cost_snapshot') {
+    return normalizeCostSnapshotJob(rawValue)
+  }
+  if (type === 'sales_facts') {
+    return normalizeSalesFactsJob(rawValue)
+  }
+  if (type === 'ledger_post') {
+    return normalizeLedgerPostJob(rawValue)
+  }
 
   return null
 }
@@ -204,10 +230,157 @@ function createDefaultScheduledJobsConfig() {
       createDefaultDataRetentionJob(),
       createDefaultPddOrderSyncJob(),
       createDefaultCostFinanceSyncJob(),
+      createDefaultChannelAssignJob(),
+      createDefaultCarrierDispatchJob(),
+      createDefaultProductNormalizeJob(),
+      createDefaultCostSnapshotJob(),
+      createDefaultSalesFactsJob(),
+      createDefaultLedgerPostJob(),
     ],
   }
 }
 
+function normalizeChannelAssignJob(rawValue: JsonObject) {
+  const config = isPlainObject(rawValue.config) ? rawValue.config : {}
+  return {
+    id: CHANNEL_ASSIGN_JOB_ID,
+    type: 'channel_assign',
+    enabled: rawValue.enabled !== false,
+    intervalSeconds: normalizeRangeInteger(rawValue.intervalSeconds, 60, 30, 3600),
+    config: {
+      batchSize: normalizeRangeInteger(config.batchSize, 500, 50, 5000),
+      maxBatches: normalizeRangeInteger(config.maxBatches, 20, 1, 200),
+    },
+  }
+}
+
+function normalizeCarrierDispatchJob(rawValue: JsonObject) {
+  const config = isPlainObject(rawValue.config) ? rawValue.config : {}
+  return {
+    id: CARRIER_DISPATCH_JOB_ID,
+    type: 'carrier_dispatch',
+    enabled: rawValue.enabled !== false,
+    intervalSeconds: normalizeRangeInteger(rawValue.intervalSeconds, 60, 30, 3600),
+    config: {
+      limit: normalizeRangeInteger(config.limit, 200, 10, 2000),
+    },
+  }
+}
+
+function normalizeProductNormalizeJob(rawValue: JsonObject) {
+  const config = isPlainObject(rawValue.config) ? rawValue.config : {}
+  return {
+    id: PRODUCT_NORMALIZE_JOB_ID,
+    type: 'product_normalize',
+    enabled: rawValue.enabled !== false,
+    intervalSeconds: normalizeRangeInteger(rawValue.intervalSeconds, 300, 60, 3600),
+    config: {
+      batchSize: normalizeRangeInteger(config.batchSize, 500, 50, 5000),
+      maxBatches: normalizeRangeInteger(config.maxBatches, 20, 1, 200),
+    },
+  }
+}
+
+function normalizeCostSnapshotJob(rawValue: JsonObject) {
+  const config = isPlainObject(rawValue.config) ? rawValue.config : {}
+  return {
+    id: COST_SNAPSHOT_JOB_ID,
+    type: 'cost_snapshot',
+    enabled: rawValue.enabled !== false,
+    intervalSeconds: normalizeRangeInteger(rawValue.intervalSeconds, 300, 60, 3600),
+    config: {
+      batchSize: normalizeRangeInteger(config.batchSize, 500, 50, 5000),
+      maxBatches: normalizeRangeInteger(config.maxBatches, 20, 1, 200),
+    },
+  }
+}
+
+function normalizeSalesFactsJob(rawValue: JsonObject) {
+  const config = isPlainObject(rawValue.config) ? rawValue.config : {}
+  return {
+    id: SALES_FACTS_JOB_ID,
+    type: 'sales_facts',
+    enabled: rawValue.enabled !== false,
+    intervalSeconds: normalizeRangeInteger(rawValue.intervalSeconds, 600, 60, 3600),
+    config: {
+      days: normalizeRangeInteger(config.days, 3, 1, 60),
+    },
+  }
+}
+
+function normalizeLedgerPostJob(rawValue: JsonObject) {
+  const config = isPlainObject(rawValue.config) ? rawValue.config : {}
+  return {
+    id: LEDGER_POST_JOB_ID,
+    type: 'ledger_post',
+    enabled: rawValue.enabled !== false,
+    intervalSeconds: normalizeRangeInteger(rawValue.intervalSeconds, 600, 60, 3600),
+    config: {
+      days: normalizeRangeInteger(config.days, 3, 1, 60),
+    },
+  }
+}
+
+function createDefaultChannelAssignJob() {
+  return {
+    id: CHANNEL_ASSIGN_JOB_ID,
+    type: 'channel_assign',
+    enabled: true,
+    intervalSeconds: 60,
+    config: { batchSize: 500, maxBatches: 20 },
+  }
+}
+
+function createDefaultCarrierDispatchJob() {
+  return {
+    id: CARRIER_DISPATCH_JOB_ID,
+    type: 'carrier_dispatch',
+    enabled: true,
+    intervalSeconds: 60,
+    config: { limit: 200 },
+  }
+}
+
+function createDefaultProductNormalizeJob() {
+  return {
+    id: PRODUCT_NORMALIZE_JOB_ID,
+    type: 'product_normalize',
+    enabled: true,
+    intervalSeconds: 300,
+    config: { batchSize: 500, maxBatches: 20 },
+  }
+}
+
+function createDefaultCostSnapshotJob() {
+  return {
+    id: COST_SNAPSHOT_JOB_ID,
+    type: 'cost_snapshot',
+    enabled: true,
+    intervalSeconds: 300,
+    config: { batchSize: 500, maxBatches: 20 },
+  }
+}
+
+function createDefaultSalesFactsJob() {
+  return {
+    id: SALES_FACTS_JOB_ID,
+    type: 'sales_facts',
+    enabled: true,
+    intervalSeconds: 600,
+    config: { days: 3 },
+  }
+}
+
+function createDefaultLedgerPostJob() {
+  return {
+    id: LEDGER_POST_JOB_ID,
+    type: 'ledger_post',
+    enabled: true,
+    intervalSeconds: 600,
+    config: { days: 3 },
+  }
+}
+
 function createDefaultPddOrderSyncJob() {
   return {
     id: PDD_ORDER_SYNC_JOB_ID,

apps/backend/src/services/scheduler/finance-pipeline-jobs.ts

@@ -0,0 +1,181 @@
+import type { JsonObject } from '../../types/json.js'
+import { query } from '../../db/client.js'
+import { assignPendingOrderItemChannels } from '../fulfillment/channel-assign-service.js'
+import { normalizeOrderItemProducts } from '../cost-center/product-normalize-service.js'
+import { syncChannelCostSnapshots } from '../cost-center/cost-snapshot-service.js'
+import { upsertAggregatedException } from '../cost-center/cost-exception-service.js'
+import { rebuildChannelSalesDailyFacts } from '../finance/sales-facts-channel-service.js'
+import { syncFinanceLedgerFromFacts } from '../finance/finance-ledger-service.js'
+import { formatShanghaiDate, shiftDate } from '../../utils/finance-date-range.js'
+
+/**
+ * 财务与履约管道作业。
+ *
+ * 每个作业只做一件事、可单独重跑，彼此不共享事务：
+ *   channel_assign → carrier_dispatch → product_normalize
+ *   → cost_snapshot → sales_facts → ledger_post
+ *
+ * 与旧 `cost_finance_sync` 的区别：旧作业把六件事串在一个函数里，
+ * 任一步失败就半途，而且会顺手写履约字段；这里每个作业只碰自己那一列。
+ */
+
+function jobConfig(job: JsonObject): JsonObject {
+  return job.config && typeof job.config === 'object' ? (job.config as JsonObject) : {}
+}
+
+function clampInt(value: unknown, fallback: number, min: number, max: number): number {
+  const parsed = Number(value)
+  if (!Number.isInteger(parsed)) return fallback
+  return Math.min(max, Math.max(min, parsed))
+}
+
+export async function runChannelAssignJob(job: JsonObject = {}) {
+  const config = jobConfig(job)
+  const result = await assignPendingOrderItemChannels({
+    batchSize: clampInt(config.batchSize, 500, 1, 5000),
+    maxBatches: clampInt(config.maxBatches, 20, 1, 200),
+    afterId: clampInt(config.afterId, 0, 0, Number.MAX_SAFE_INTEGER),
+  })
+
+  return {
+    status: 'ok',
+    message:
+      `渠道判定：扫描 ${result.scanned}，已定 ${result.assigned}，` +
+      `冲突 ${result.conflicted}，仍待定 ${result.stillPending}`,
+    ...result,
+  }
+}
+
+export async function runCarrierDispatchJob(job: JsonObject = {}) {
+  const config = jobConfig(job)
+  const limit = clampInt(config.limit, 200, 1, 2000)
+
+  const manual = await query<{ id: number; sku_key: string; config_id: number }>(
+    `
+      SELECT oi.id, oi.manual_sku_key AS sku_key, cfg.id AS config_id
+      FROM order_items oi
+      JOIN manual_fulfillment_configs cfg ON cfg.sku_key = oi.manual_sku_key AND cfg.enabled
+      WHERE oi.fulfillment_status = 'assigned'
+        AND oi.fulfillment_channel = 'manual'
+        AND oi.fulfillment_config_id IS NULL
+      ORDER BY oi.id
+      LIMIT $1
+    `,
+    [limit],
+  )
+
+  let attached = 0
+  for (const row of manual.rows) {
+    const updated = await query(
+      `
+        UPDATE order_items
+        SET fulfillment_config_id = $2, updated_at = NOW()
+        WHERE id = $1 AND fulfillment_config_id IS NULL
+      `,
+      [row.id, row.config_id],
+    )
+    attached += updated.rowCount || 0
+  }
+
+  // 人工渠道但没有配置：进「人工 SKU 未配置」队列。
+  const missingManual = await query<{ id: number; sku_key: string }>(
+    `
+      SELECT oi.id, oi.manual_sku_key AS sku_key
+      FROM order_items oi
+      WHERE oi.fulfillment_status = 'assigned'
+        AND oi.fulfillment_channel = 'manual'
+        AND oi.fulfillment_config_id IS NULL
+        AND NOT EXISTS (
+          SELECT 1 FROM manual_fulfillment_configs cfg
+          WHERE cfg.sku_key = oi.manual_sku_key AND cfg.enabled
+        )
+      ORDER BY oi.id
+      LIMIT $1
+    `,
+    [limit],
+  )
+
+  for (const row of missingManual.rows) {
+    await upsertAggregatedException({
+      reason: 'manual_config_missing',
+      scopeKey: `manual_config_missing|-|-|${row.sku_key}|manual`,
+      orderItemId: row.id,
+      detail: { skuKey: row.sku_key },
+    })
+  }
+
+  return {
+    status: 'ok',
+    message: `人工载体挂载 ${attached}，缺配置 ${missingManual.rows.length}`,
+    attached,
+    missingManual: missingManual.rows.length,
+  }
+}
+
+export async function runProductNormalizeJob(job: JsonObject = {}) {
+  const config = jobConfig(job)
+  const result = await normalizeOrderItemProducts({
+    batchSize: clampInt(config.batchSize, 500, 1, 5000),
+    maxBatches: clampInt(config.maxBatches, 20, 1, 200),
+  })
+
+  return {
+    status: 'ok',
+    message: `商品归一：扫描 ${result.scanned}，命中 ${result.matched}，未命中 ${result.unmatched}`,
+    ...result,
+  }
+}
+
+export async function runCostSnapshotJob(job: JsonObject = {}) {
+  const config = jobConfig(job)
+  const result = await syncChannelCostSnapshots({
+    batchSize: clampInt(config.batchSize, 500, 1, 5000),
+    maxBatches: clampInt(config.maxBatches, 20, 1, 200),
+  })
+
+  return {
+    status: 'ok',
+    message: `成本快照：扫描 ${result.scanned}，生成 ${result.created}，缺失 ${result.missing}`,
+    ...result,
+  }
+}
+
+export async function runSalesFactsJob(job: JsonObject = {}) {
+  const config = jobConfig(job)
+  const days = clampInt(config.days, 3, 1, 60)
+  const dateTo = formatShanghaiDate(new Date())
+  const dateFrom = shiftDate(dateTo, -(days - 1))
+  const result = await rebuildChannelSalesDailyFacts({ dateFrom, dateTo })
+
+  return {
+    status: 'ok',
+    message: `销售事实：${result.dateFrom}~${result.dateTo} 写入 ${result.inserted} 行`,
+    ...result,
+  }
+}
+
+export async function runLedgerPostJob(job: JsonObject = {}) {
+  const config = jobConfig(job)
+  const days = clampInt(config.days, 3, 1, 60)
+  const dateTo = formatShanghaiDate(new Date())
+  const dateFrom = shiftDate(dateTo, -(days - 1))
+  const result = await syncFinanceLedgerFromFacts({ dateFrom, dateTo })
+
+  return {
+    status: 'ok',
+    message: `总账：${dateFrom}~${dateTo} 新增 ${result.created} 条，跳过已关账 ${result.skipped}`,
+    ...result,
+  }
+}
+
+/** 顺序执行全部管道，供管理端「一键重算」使用。 */
+export async function runFinancePipelineOnce(input: { days?: number } = {}) {
+  const channel = await runChannelAssignJob({ config: { batchSize: 1000, maxBatches: 50 } })
+  const carrier = await runCarrierDispatchJob({ config: { limit: 500 } })
+  const product = await runProductNormalizeJob({ config: { batchSize: 1000, maxBatches: 50 } })
+  const cost = await runCostSnapshotJob({ config: { batchSize: 1000, maxBatches: 50 } })
+  const facts = await runSalesFactsJob({ config: { days: input.days ?? 3 } })
+  const ledger = await runLedgerPostJob({ config: { days: input.days ?? 3 } })
+
+  return { channel, carrier, product, cost, facts, ledger }
+}

apps/backend/src/services/scheduler/scheduler-service.ts

@@ -7,6 +7,14 @@ import { runWorkOrderTimeoutJob } from './work-order-timeout-job.js'
 import { runDataRetentionJob } from './data-retention-job.js'
 import { runPddOrderSyncJob } from './pdd-order-sync-job.js'
 import { runCostFinanceSyncJob } from './cost-finance-sync-job.js'
+import {
+  runCarrierDispatchJob,
+  runChannelAssignJob,
+  runCostSnapshotJob,
+  runLedgerPostJob,
+  runProductNormalizeJob,
+  runSalesFactsJob,
+} from './finance-pipeline-jobs.js'
 
 const timers = new Map<string, NodeJS.Timeout>()
 const jobStates = new Map<string, JsonObject>()
@@ -208,6 +216,24 @@ function dispatchJob(job: JsonObject) {
   if (job.type === 'cost_finance_sync') {
     return runCostFinanceSyncJob(job)
   }
+  if (job.type === 'channel_assign') {
+    return runChannelAssignJob(job)
+  }
+  if (job.type === 'carrier_dispatch') {
+    return runCarrierDispatchJob(job)
+  }
+  if (job.type === 'product_normalize') {
+    return runProductNormalizeJob(job)
+  }
+  if (job.type === 'cost_snapshot') {
+    return runCostSnapshotJob(job)
+  }
+  if (job.type === 'sales_facts') {
+    return runSalesFactsJob(job)
+  }
+  if (job.type === 'ledger_post') {
+    return runLedgerPostJob(job)
+  }
 
   throw createHttpError(`不支持的定时任务类型：${job.type}`, {
     statusCode: 400,

apps/backend/src/services/worker-platform/admin-worker-catalog-service.ts

@@ -27,6 +27,7 @@ import {
   upsertWorkProductRule,
 } from '../../repositories/worker-platform/work-product-rule-repo.js'
 import type { WorkOrderRow, WorkProductRuleRow } from '../../repositories/worker-platform/types.js'
+import { syncAdminWorkOrderTemplateCostVersions } from '../cost-center/work-order-product-service.js'
 import {
   normalizeWorkOrderAcceptanceMode,
   normalizeWorkOrderGiftCooldownHours,
@@ -531,6 +532,17 @@ export async function saveAdminWorkProductRule(payload: JsonObject = {}) {
     sortOrder: normalizeInteger(payload.sortOrder, 100),
     now: nowIso(),
   })
+
+  // 模板价格变更即同步成本版本（事件驱动）；失败不阻塞模板保存，
+  // 夜间巡检与后台「同步发单金额」按钮仍可修复。
+  if (rule) {
+    try {
+      await syncAdminWorkOrderTemplateCostVersions(rule.id)
+    } catch {
+      // 成本版本同步失败不允许影响模板本身；由巡检补齐。
+    }
+  }
+
   return { rule: mapWorkProductRule(rule) }
 }