feat(fulfillment): 渠道联动与管道兜底

- 人工配置保存/删除同步渠道规则，人工确认写 channel/status
- 按 SKU 指定渠道默认同时落规则
- 管道补充：载体缺失、人工成本未记录、事实与总账按水位追赶

apps/backend/src/routes/admin/fulfillment-channel.ts

@@ -87,6 +87,7 @@ router.post(
           channel,
           adminUserId,
           note,
+          createRule: body.createRule !== false,
         })
       }
 

apps/backend/src/services/cost-center/cost-exception-service.ts

@@ -18,6 +18,8 @@ export const COST_EXCEPTION_REASONS = [
   'product_mapping_missing',
   'cost_missing',
   'manual_config_missing',
+  'manual_cost_missing',
+  'carrier_missing',
 ] as const
 
 export type CostExceptionReason = (typeof COST_EXCEPTION_REASONS)[number]

apps/backend/src/services/cost-center/manual-fulfillment-config-service.ts

@@ -127,7 +127,7 @@ export async function scanManualSkus(options: ScanOptions) {
           oi.manual_sku_key AS sku_key
         FROM order_items oi
         JOIN orders o ON o.id = oi.order_id
-        WHERE oi.fulfillment_mode = 'manual'
+        WHERE oi.fulfillment_status = 'assigned' AND oi.fulfillment_channel = 'manual'
           AND (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date
               BETWEEN $1::date AND $2::date
       ),
@@ -240,7 +240,7 @@ export async function listManualSkus(input: Record<string, unknown> = {}) {
     conditions.push(`EXISTS (
       SELECT 1 FROM order_items oi
       JOIN orders o ON o.id = oi.order_id
-      WHERE oi.fulfillment_mode = 'manual'
+      WHERE oi.fulfillment_status = 'assigned' AND oi.fulfillment_channel = 'manual'
         AND o.provider = $${params.length}
         AND oi.manual_sku_key = r.sku_key
     )`)
@@ -378,6 +378,9 @@ export async function saveManualFulfillmentConfig(payload: JsonObject) {
 
   const config = requiredRow(result.rows[0], '保存人工履约配置失败')
   await syncRegistryConfigId(config.sku_key, config.id)
+  // 同步渠道规则：人工配置就是「这个内部 SKU 走人工渠道」的唯一声明，
+  // 少了这一步，新配的 SKU 会一直停在待分类。
+  await syncManualChannelRule(config)
   // 配置保存后立刻回填该 SKU 最近 30 天的订单项，避免必须再点一次「重新匹配」。
   // 历史更早的订单项不自动回填（历史人工成本不补录）；需要时再显式点「重新匹配」。
   const today = formatShanghaiDate(new Date())
@@ -403,6 +406,7 @@ export async function deleteManualFulfillmentConfig(skuKey: string) {
     `DELETE FROM manual_fulfillment_configs WHERE sku_key = $1 RETURNING id`,
     [key],
   )
+  await query(`DELETE FROM sku_channel_rules WHERE source = 'manual' AND match_key = $1`, [key])
   await syncRegistryConfigId(key, null)
   return { skuKey: key, deleted: deleted.rowCount || 0 }
 }
@@ -431,7 +435,7 @@ export async function rematchManualFulfillments(input: Record<string, unknown> =
         JOIN orders o ON o.id = oi.order_id
         JOIN manual_fulfillment_configs cfg
           ON cfg.sku_key = oi.manual_sku_key
-        WHERE oi.fulfillment_mode = 'manual'
+        WHERE oi.fulfillment_status = 'assigned' AND oi.fulfillment_channel = 'manual'
           AND cfg.enabled = TRUE
           AND ($1 = '' OR cfg.sku_key = $1)
           AND ($2 = FALSE OR oi.fulfillment_config_id IS NULL)
@@ -699,3 +703,36 @@ function requiredRow<T>(row: T | undefined, message: string): T {
   }
   return row
 }
+
+/**
+ * 人工配置 → 渠道规则（内部 SKU 精确匹配）。
+ *
+ * 幂等：按唯一键 (provider, shop, match_key_type, match_key, channel) upsert。
+ */
+async function syncManualChannelRule(config: {
+  id: number
+  sku_key: string
+  enabled: boolean
+  note: string
+}): Promise<void> {
+  const skuKey = String(config.sku_key || '').trim()
+  if (!skuKey) return
+
+  await query(
+    `
+      INSERT INTO sku_channel_rules (
+        provider, shop_id, match_key_type, match_key, channel, priority,
+        source, source_ref_id, enabled, note, created_at, updated_at
+      )
+      VALUES ('', '', 'internal_sku', $1, 'manual', 70, 'manual', $2, $3, $4, NOW(), NOW())
+      ON CONFLICT (provider, shop_id, match_key_type, match_key, channel)
+      DO UPDATE SET
+        enabled = EXCLUDED.enabled,
+        source = 'manual',
+        source_ref_id = EXCLUDED.source_ref_id,
+        note = EXCLUDED.note,
+        updated_at = NOW()
+    `,
+    [skuKey, config.id, config.enabled !== false, String(config.note || '').slice(0, 200)],
+  )
+}

apps/backend/src/services/finance/manual-fulfillment-service.ts

@@ -96,7 +96,11 @@ export async function listManualFulfillmentPending(
   const keyword = String(input.keyword || '').trim()
   const status = String(input.status || '').trim()
 
-  const conditions = ["oi.fulfillment_mode = 'manual'", 'cfg.id IS NOT NULL']
+  const conditions = [
+    "oi.fulfillment_status = 'assigned'",
+    "oi.fulfillment_channel = 'manual'",
+    'cfg.id IS NOT NULL',
+  ]
   const params: unknown[] = []
 
   if (status === 'pending') {
@@ -219,7 +223,7 @@ export async function confirmManualFulfillment(payload: JsonObject, session: Adm
           COALESCE(oi.revenue_amount_fen, 0)::bigint AS revenue_fen,
           COALESCE(o.paid_at, o.created_at) AS basis_at,
           oi.canonical_product_id,
-          oi.fulfillment_mode,
+          COALESCE(oi.fulfillment_channel, CASE WHEN oi.fulfillment_status = 'archived' THEN 'archived' ELSE 'pending' END) AS fulfillment_mode,
           oi.fulfillment_config_id AS config_id,
           COALESCE(cfg.platform_key, '') AS platform_key,
           COALESCE(cfg.credit_quantity, 0)::bigint AS credit_quantity,
@@ -450,6 +454,21 @@ export async function confirmManualFulfillment(payload: JsonObject, session: Adm
       })
     }
 
+    // 人工确认即渠道成立：写新三列，避免订单一直停在 pending。
+    await client.query(
+      `
+        UPDATE order_items
+        SET fulfillment_channel = 'manual',
+            fulfillment_status = 'assigned',
+            fulfillment_source = 'manual_confirm',
+            fulfillment_decided_at = $2,
+            updated_at = $2
+        WHERE id = $1
+          AND fulfillment_status <> 'archived'
+      `,
+      [orderItemId, now],
+    )
+
     // 人工缺成本异常立即收口，不等下一轮定时任务。
     await client.query(
       `UPDATE cost_exceptions
@@ -461,6 +480,20 @@ export async function confirmManualFulfillment(payload: JsonObject, session: Adm
       [orderItemId, now],
     )
 
+    // v2 聚合异常：删除本条明细，没人再引用的异常头自动关闭。
+    await client.query(`DELETE FROM cost_exception_items WHERE order_item_id = $1`, [orderItemId])
+    await client.query(
+      `
+        UPDATE cost_exceptions ce
+        SET status = 'resolved', resolved_at = $1, item_count = 0, last_seen_at = $1
+        WHERE ce.status <> 'resolved'
+          AND NOT EXISTS (
+            SELECT 1 FROM cost_exception_items i WHERE i.exception_id = ce.id
+          )
+      `,
+      [now],
+    )
+
     return {
       orderItemId,
       orderId: Number(row.order_id_resolved),
@@ -582,7 +615,7 @@ async function matchManualFulfillmentConfigWithClient(
           )
         )
       WHERE oi.id = $1
-        AND oi.fulfillment_mode = 'manual'
+        AND oi.fulfillment_status = 'assigned' AND oi.fulfillment_channel = 'manual'
         AND oi.fulfillment_config_id IS NULL
       ORDER BY
         (cfg.canonical_product_id IS NOT NULL
@@ -820,7 +853,7 @@ export async function getManualFulfillmentCountsByOrderIds(
       LEFT JOIN manual_fulfillment_configs cfg ON cfg.id = oi.fulfillment_config_id
       LEFT JOIN fulfilled ON fulfilled.order_item_id = oi.id
       WHERE oi.order_id = ANY($1::bigint[])
-        AND oi.fulfillment_mode = 'manual'
+        AND oi.fulfillment_status = 'assigned' AND oi.fulfillment_channel = 'manual'
       GROUP BY oi.order_id
     `,
     [ids],

apps/backend/src/services/fulfillment/channel-assign-service.ts

@@ -371,7 +371,8 @@ export async function assignChannelBySku(input: {
   channel: string
   adminUserId: number | null
   note?: string
-}): Promise<{ updated: number }> {
+  createRule?: boolean
+}): Promise<{ updated: number; ruleCreated: boolean }> {
   if (!isFulfillmentChannel(input.channel)) {
     throw createHttpError('渠道无效', { statusCode: 400, errorCode: 'invalid_channel' })
   }
@@ -395,7 +396,26 @@ export async function assignChannelBySku(input: {
     `,
     [input.provider, input.shopId, input.skuCode, input.channel, input.adminUserId],
   )
-  return { updated: result.rowCount || 0 }
+
+  // 默认同时落一条渠道规则：否则同 SKU 的下一单还是会进待分类。
+  let ruleCreated = false
+  if (input.createRule !== false) {
+    const rule = await upsertChannelRule({
+      provider: input.provider,
+      shopId: input.shopId,
+      matchKeyType: 'sku_code',
+      matchKey: input.skuCode,
+      channel: input.channel,
+      priority: 10,
+      source: 'operator',
+      enabled: true,
+      note: `运营在待分类队列指定：${String(input.note || '').slice(0, 120)}`,
+      adminUserId: input.adminUserId,
+    })
+    ruleCreated = rule.id > 0
+  }
+
+  return { updated: result.rowCount || 0, ruleCreated }
 }
 
 export async function listChannelRules(

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

@@ -110,12 +110,74 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) {
     })
   }
 
+  // 已配置人工 SKU、但还没确认履约的单：进入「人工成本未记录」队列。
+  const pendingManualCost = 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 NOT NULL
+        AND NOT EXISTS (
+          SELECT 1 FROM cost_snapshots cs
+          WHERE cs.subject_type = 'manual_case'
+            AND cs.order_item_id = oi.id
+            AND cs.superseded_at IS NULL
+        )
+      ORDER BY oi.id
+      LIMIT $1
+    `,
+    [limit],
+  )
+
+  for (const row of pendingManualCost.rows) {
+    await upsertAggregatedException({
+      reason: 'manual_cost_missing',
+      scopeKey: `manual_cost_missing|-|-|${row.sku_key}|manual`,
+      orderItemId: row.id,
+      detail: { skuKey: row.sku_key },
+    })
+  }
+
+  // 自动发货 / 接单平台被运营指定后却没有载体：不能静默卡住，进问题队列。
+  const missingCarrier = await query<{ id: number; channel: string; sku_code: string }>(
+    `
+      SELECT oi.id, oi.fulfillment_channel AS channel, oi.sku_code
+      FROM order_items oi
+      WHERE oi.fulfillment_status = 'assigned'
+        AND oi.fulfillment_channel IN ('affiliate_dash', 'work_order')
+        AND NOT EXISTS (
+          SELECT 1 FROM fulfillment_tasks ft
+          WHERE ft.order_item_id = oi.id
+            AND ft.executor_key IN ('affiliate_dash', 'third_party_voucher')
+        )
+        AND NOT EXISTS (
+          SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id
+        )
+      ORDER BY oi.id
+      LIMIT $1
+    `,
+    [limit],
+  )
+
+  for (const row of missingCarrier.rows) {
+    await upsertAggregatedException({
+      reason: 'carrier_missing',
+      scopeKey: `carrier_missing|${row.channel}|${row.sku_code}`,
+      orderItemId: row.id,
+      detail: { channel: row.channel, skuCode: row.sku_code },
+    })
+  }
+
   await touchWatermark('carrier_dispatch')
   return {
     status: 'ok',
-    message: `人工载体挂载 ${attached}，缺配置 ${missingManual.rows.length}`,
+    message:
+      `人工载体挂载 ${attached}，缺配置 ${missingManual.rows.length}，` +
+      `缺履约载体 ${missingCarrier.rows.length}`,
     attached,
     missingManual: missingManual.rows.length,
+    missingCarrier: missingCarrier.rows.length,
   }
 }
 
@@ -153,7 +215,7 @@ 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 dateFrom = await resolveCatchUpStart('sales_facts', days, dateTo)
   const result = await rebuildChannelSalesDailyFacts({ dateFrom, dateTo })
 
   await touchWatermark('sales_facts', { lastStatDate: result.dateTo })
@@ -168,7 +230,7 @@ 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 dateFrom = await resolveCatchUpStart('ledger_post', days, dateTo)
   const result = await syncFinanceLedgerFromFacts({ dateFrom, dateTo })
 
   await touchWatermark('ledger_post', { lastStatDate: dateTo })
@@ -215,3 +277,25 @@ export async function runSingleFinancePipeline(
     throw error
   }
 }
+
+/**
+ * 计算本轮起始日期：默认回看 N 天；若水位更早（例如停机数天），
+ * 从水位那天补起，最多补 90 天，避免一次扫描过长。
+ */
+async function resolveCatchUpStart(
+  pipeline: 'sales_facts' | 'ledger_post',
+  days: number,
+  dateTo: string,
+): Promise<string> {
+  const windowStart = shiftDate(dateTo, -(days - 1))
+  const result = await query<{ last_stat_date: string | null }>(
+    `SELECT last_stat_date::text FROM finance_watermarks WHERE pipeline = $1`,
+    [pipeline],
+  )
+  const lastDate = result.rows[0]?.last_stat_date || ''
+  if (!lastDate) return windowStart
+
+  const earliest = shiftDate(dateTo, -89)
+  const candidate = lastDate < windowStart ? lastDate : windowStart
+  return candidate < earliest ? earliest : candidate
+}