feat(scheduler): 财务管道自动注册、水位记录与后台状态页

- 老配置读取时自动补齐六个管道任务，升级后无需手工添加
- 每个管道运行后写水位与 last_success_at，失败写 last_error
- 新增 GET /admin/fulfillment/pipeline/status 与单步重跑
- 后台「履约渠道 → 管道状态」展示启停、水位、最近执行与错误
- 定时任务管理接口不再丢弃管道自己的 config 字段

apps/backend/src/services/admin/platform-config/notification-service.ts

@@ -127,7 +127,10 @@ function mapAdminScheduledJobsConfig(config: JsonObject = {}) {
       enabled: item.enabled === true,
       intervalSeconds: Number(item.intervalSeconds || 300),
       cooldownSeconds: Number(item.cooldownSeconds || 1800),
+      // 保留各任务自己的配置键（管道作业的 batchSize / maxBatches / days 等），
+      // 避免管理接口读一次就把没列出来的字段丢掉。
       config: {
+        ...(item.config && typeof item.config === 'object' ? item.config : {}),
         assetThreshold: normalizeNonNegativeNumber(item.config?.assetThreshold, 500),
         accounts: Array.isArray(item.config?.accounts) ? item.config.accounts : [],
         pageSize: Number(item.config?.pageSize || 0) || undefined,

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

@@ -85,6 +85,12 @@ export function normalizeScheduledJobsConfig(rawValue: unknown) {
   const hasDataRetention = jobs.some((item) => item.id === DATA_RETENTION_JOB_ID)
   const hasPddOrderSync = jobs.some((item) => item.id === PDD_ORDER_SYNC_JOB_ID)
   const hasCostFinanceSync = jobs.some((item) => item.id === COST_FINANCE_SYNC_JOB_ID)
+  const hasChannelAssign = jobs.some((item) => item.id === CHANNEL_ASSIGN_JOB_ID)
+  const hasCarrierDispatch = jobs.some((item) => item.id === CARRIER_DISPATCH_JOB_ID)
+  const hasProductNormalize = jobs.some((item) => item.id === PRODUCT_NORMALIZE_JOB_ID)
+  const hasCostSnapshot = jobs.some((item) => item.id === COST_SNAPSHOT_JOB_ID)
+  const hasSalesFacts = jobs.some((item) => item.id === SALES_FACTS_JOB_ID)
+  const hasLedgerPost = jobs.some((item) => item.id === LEDGER_POST_JOB_ID)
 
   if (!hasWorkOrderTimeout) {
     jobs.push(createDefaultWorkOrderTimeoutJob())
@@ -101,6 +107,13 @@ export function normalizeScheduledJobsConfig(rawValue: unknown) {
   if (!hasCostFinanceSync) {
     jobs.push(createDefaultCostFinanceSyncJob())
   }
+  // v2 管道：老配置里没有这些任务时自动补齐，保证升级后无需手工添加。
+  if (!hasChannelAssign) jobs.push(createDefaultChannelAssignJob())
+  if (!hasCarrierDispatch) jobs.push(createDefaultCarrierDispatchJob())
+  if (!hasProductNormalize) jobs.push(createDefaultProductNormalizeJob())
+  if (!hasCostSnapshot) jobs.push(createDefaultCostSnapshotJob())
+  if (!hasSalesFacts) jobs.push(createDefaultSalesFactsJob())
+  if (!hasLedgerPost) jobs.push(createDefaultLedgerPostJob())
 
   return {
     enabled: typeof source.enabled === 'boolean' ? source.enabled : true,

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

@@ -7,6 +7,11 @@ import { upsertAggregatedException } from '../cost-center/cost-exception-service
 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'
+import {
+  recordPipelineError,
+  touchWatermark,
+  type FinancePipelineName,
+} from './finance-pipeline-status.js'
 
 /**
  * 财务与履约管道作业。
@@ -37,6 +42,7 @@ export async function runChannelAssignJob(job: JsonObject = {}) {
     afterId: clampInt(config.afterId, 0, 0, Number.MAX_SAFE_INTEGER),
   })
 
+  await touchWatermark('channel_assign', { lastItemId: result.lastId })
   return {
     status: 'ok',
     message:
@@ -104,6 +110,7 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) {
     })
   }
 
+  await touchWatermark('carrier_dispatch')
   return {
     status: 'ok',
     message: `人工载体挂载 ${attached}，缺配置 ${missingManual.rows.length}`,
@@ -119,6 +126,7 @@ export async function runProductNormalizeJob(job: JsonObject = {}) {
     maxBatches: clampInt(config.maxBatches, 20, 1, 200),
   })
 
+  await touchWatermark('product_normalize', { lastItemId: result.lastId })
   return {
     status: 'ok',
     message: `商品归一：扫描 ${result.scanned}，命中 ${result.matched}，未命中 ${result.unmatched}`,
@@ -133,6 +141,7 @@ export async function runCostSnapshotJob(job: JsonObject = {}) {
     maxBatches: clampInt(config.maxBatches, 20, 1, 200),
   })
 
+  await touchWatermark('cost_snapshot', { lastItemId: result.lastId })
   return {
     status: 'ok',
     message: `成本快照：扫描 ${result.scanned}，生成 ${result.created}，缺失 ${result.missing}`,
@@ -147,6 +156,7 @@ export async function runSalesFactsJob(job: JsonObject = {}) {
   const dateFrom = shiftDate(dateTo, -(days - 1))
   const result = await rebuildChannelSalesDailyFacts({ dateFrom, dateTo })
 
+  await touchWatermark('sales_facts', { lastStatDate: result.dateTo })
   return {
     status: 'ok',
     message: `销售事实：${result.dateFrom}~${result.dateTo} 写入 ${result.inserted} 行`,
@@ -161,6 +171,7 @@ export async function runLedgerPostJob(job: JsonObject = {}) {
   const dateFrom = shiftDate(dateTo, -(days - 1))
   const result = await syncFinanceLedgerFromFacts({ dateFrom, dateTo })
 
+  await touchWatermark('ledger_post', { lastStatDate: dateTo })
   return {
     status: 'ok',
     message: `总账：${dateFrom}~${dateTo} 新增 ${result.created} 条，跳过已关账 ${result.skipped}`,
@@ -179,3 +190,28 @@ export async function runFinancePipelineOnce(input: { days?: number } = {}) {
 
   return { channel, carrier, product, cost, facts, ledger }
 }
+
+const PIPELINE_RUNNERS: Record<
+  FinancePipelineName,
+  (job?: JsonObject) => Promise<Record<string, unknown>>
+> = {
+  channel_assign: runChannelAssignJob,
+  carrier_dispatch: runCarrierDispatchJob,
+  product_normalize: runProductNormalizeJob,
+  cost_snapshot: runCostSnapshotJob,
+  sales_facts: runSalesFactsJob,
+  ledger_post: runLedgerPostJob,
+}
+
+/** 单独执行一个管道（排障 / 补跑），失败时把原因写进水位表。 */
+export async function runSingleFinancePipeline(
+  pipeline: FinancePipelineName,
+  job: JsonObject = {},
+): Promise<Record<string, unknown>> {
+  try {
+    return await PIPELINE_RUNNERS[pipeline](job)
+  } catch (error) {
+    await recordPipelineError(pipeline, error)
+    throw error
+  }
+}

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

@@ -0,0 +1,130 @@
+import { query } from '../../db/client.js'
+import { nowIso } from '../../utils/time.js'
+import { getScheduledJobRuntimeStates } from './scheduler-service.js'
+
+/**
+ * 财务管道可观测性：水位 + 定时任务运行状态。
+ *
+ * 后台「履约渠道 → 管道状态」用的就是这里的数据，用于判断：
+ *   · 管道有没有在跑（last_success_at）
+ *   · 跑到哪了（last_item_id / last_stat_date）
+ *   · 最近为什么失败（last_error）
+ */
+
+export const FINANCE_PIPELINES = [
+  { name: 'channel_assign', label: '渠道判定', jobId: 'channel-assign' },
+  { name: 'carrier_dispatch', label: '履约载体', jobId: 'carrier-dispatch' },
+  { name: 'product_normalize', label: '商品归一', jobId: 'product-normalize' },
+  { name: 'cost_snapshot', label: '成本快照', jobId: 'cost-snapshot' },
+  { name: 'sales_facts', label: '销售事实', jobId: 'sales-facts' },
+  { name: 'ledger_post', label: '总账过账', jobId: 'ledger-post' },
+] as const
+
+export type FinancePipelineName = (typeof FINANCE_PIPELINES)[number]['name']
+
+export function isFinancePipelineName(value: unknown): value is FinancePipelineName {
+  return FINANCE_PIPELINES.some((item) => item.name === value)
+}
+
+/** 记录一次成功运行的水位。 */
+export async function touchWatermark(
+  pipeline: FinancePipelineName,
+  patch: { lastItemId?: number; lastStatDate?: string | null } = {},
+): Promise<void> {
+  const now = nowIso()
+  await query(
+    `
+      INSERT INTO finance_watermarks (pipeline, last_item_id, last_stat_date, last_success_at, updated_at)
+      VALUES ($1, $2, $3::date, $4, $4)
+      ON CONFLICT (pipeline) DO UPDATE SET
+        last_item_id = GREATEST(finance_watermarks.last_item_id, EXCLUDED.last_item_id),
+        last_stat_date = COALESCE(EXCLUDED.last_stat_date, finance_watermarks.last_stat_date),
+        last_success_at = EXCLUDED.last_success_at,
+        last_error = '',
+        updated_at = EXCLUDED.updated_at
+    `,
+    [pipeline, Math.max(0, Math.trunc(patch.lastItemId || 0)), patch.lastStatDate ?? null, now],
+  )
+}
+
+/** 记录一次失败原因，不改水位。 */
+export async function recordPipelineError(
+  pipeline: FinancePipelineName,
+  error: unknown,
+): Promise<void> {
+  const message = error instanceof Error ? error.message : String(error)
+  await query(
+    `
+      INSERT INTO finance_watermarks (pipeline, last_error, updated_at)
+      VALUES ($1, $2, $3)
+      ON CONFLICT (pipeline) DO UPDATE SET
+        last_error = EXCLUDED.last_error,
+        updated_at = EXCLUDED.updated_at
+    `,
+    [pipeline, message.slice(0, 500), nowIso()],
+  )
+}
+
+export type FinancePipelineStatusRow = {
+  pipeline: string
+  label: string
+  lastItemId: number
+  lastStatDate: string | null
+  lastSuccessAt: string | null
+  lastError: string
+  jobId: string
+  scheduler: {
+    running: boolean
+    enabled: boolean
+    lastRunAt: string
+    lastFinishedAt: string
+    nextRunAt: string
+    lastStatus: string
+    lastMessage: string
+  } | null
+}
+
+export async function getFinancePipelineStatus(): Promise<FinancePipelineStatusRow[]> {
+  const marks = await query<{
+    pipeline: string
+    last_item_id: string
+    last_stat_date: string | null
+    last_success_at: string | null
+    last_error: string
+  }>(
+    `
+      SELECT pipeline, last_item_id::text, last_stat_date::text, last_success_at, last_error
+      FROM finance_watermarks
+    `,
+  )
+
+  const byPipeline = new Map(marks.rows.map((row) => [row.pipeline, row]))
+  const runtime = new Map(
+    getScheduledJobRuntimeStates().map((state) => [String(state.id || ''), state]),
+  )
+
+  return FINANCE_PIPELINES.map((item) => {
+    const mark = byPipeline.get(item.name)
+    const state = runtime.get(item.jobId)
+    return {
+      pipeline: item.name,
+      label: item.label,
+      lastItemId: Number(mark?.last_item_id || 0),
+      lastStatDate: mark?.last_stat_date || null,
+      lastSuccessAt: mark?.last_success_at || null,
+      lastError: String(mark?.last_error || ''),
+      jobId: item.jobId,
+      scheduler: state
+        ? {
+            running: state.running === true,
+            enabled: state.enabled === true,
+            lastRunAt: String(state.lastRunAt || ''),
+            lastFinishedAt: String(state.lastFinishedAt || ''),
+            nextRunAt: String(state.nextRunAt || ''),
+            lastStatus: String(state.lastStatus || ''),
+            lastMessage: String(state.lastMessage || ''),
+          }
+        : null,
+    }
+  })
+}