fix(pipeline): 变更按业务日回填，异常自动收口，载体管道加游标

- 渠道判定/成本快照返回受影响业务日，任务按日重算事实与总账 - 人工配置或载体补齐后，manual_config_missing / carrier_missing 自动关闭 - carrier_dispatch 使用水位游标，避免每轮从 id=0 重扫
2026-10-31 20:41:43 +08:00
parent 42d5874411
commit a62f6e1ec8
5 changed files with 244 additions and 12 deletions
@@ -175,6 +175,37 @@ export async function listExceptionQueue(input: {
 *
 * 批量版本：一次处理一批订单项，避免逐条 UPDATE。
 */
+export async function resolveExceptionsForItemsByReason(
+  orderItemIds: number[],
+  reason: string,
+): Promise<number> {
+  const ids = orderItemIds.filter((id) => Number.isInteger(id) && id > 0)
+  if (ids.length === 0) return 0
+  const now = nowIso()
+  const result = await query<{ id: number }>(
+    `
+      WITH removed AS (
+        DELETE FROM cost_exception_items i
+        USING cost_exceptions ce
+        WHERE i.exception_id = ce.id
+          AND i.order_item_id = ANY($1::bigint[])
+          AND ce.reason = $3
+        RETURNING i.exception_id
+      )
+      UPDATE cost_exceptions ce
+      SET status = 'resolved', resolved_at = $2, item_count = 0, last_seen_at = $2
+      WHERE ce.id IN (SELECT DISTINCT exception_id FROM removed)
+        AND ce.status <> 'resolved'
+        AND NOT EXISTS (
+          SELECT 1 FROM cost_exception_items i WHERE i.exception_id = ce.id
+        )
+      RETURNING ce.id
+    `,
+    [ids, now, reason],
+  )
+  return result.rowCount || 0
+}
+
 export async function resolveExceptionsForItems(orderItemIds: number[]): Promise<number> {
  const ids = orderItemIds.filter((id) => Number.isInteger(id) && id > 0)
  if (ids.length === 0) return 0
@@ -1,4 +1,5 @@
 import { query, withTransaction } from '../../db/client.js'
+import { resolveExceptionsForItemsByReason } from './cost-exception-service.js'
 import type { JsonObject } from '../../types/json.js'
 import { createHttpError } from '../../utils/http.js'
 import { formatShanghaiDate, shiftDate } from '../../utils/finance-date-range.js'
@@ -424,7 +425,7 @@ export async function rematchManualFulfillments(input: Record<string, unknown> =
  const dateTo = normalizeOptionalDate(input.dateTo ?? input.date_to, '结束日期')

  const now = nowIso()
-  const result = await query<{ updated_count: number }>(
+  const result = await query<{ id: number }>(
    `
      WITH target AS (
        SELECT
@@ -454,11 +455,17 @@ export async function rematchManualFulfillments(input: Record<string, unknown> =
          updated_at = $5
      FROM target
      WHERE oi.id = target.order_item_id
-      RETURNING 1
+      RETURNING oi.id
    `,
    [skuKey, onlyUnmatched, dateFrom, dateTo, now],
  )

+  // 配置补上以后，「人工 SKU 未配置」异常要立即收口，不能等人工去点。
+  const updatedIds = result.rows.map((row) => Number(row.id))
+  if (updatedIds.length > 0) {
+    await resolveExceptionsForItemsByReason(updatedIds, 'manual_config_missing')
+  }
+
  // 配置生效后，把 SKU 对应的成本版本挂上（若已配置商品）。
  await attachCostVersionsForManualConfigs()

@@ -18,6 +18,7 @@ function buildItem(patch: Partial<ChannelCandidateItem> = {}): ChannelCandidateI
    normalized_sku_name: '套装-测试',
    has_work_order: false,
    has_affiliate_task: false,
+    basis_date: '2026-10-31',
    ...patch,
  }
 }
@@ -50,6 +50,7 @@ export type ChannelCandidateItem = {
  normalized_sku_name: string | null
  has_work_order: boolean
  has_affiliate_task: boolean
+  basis_date: string
 }

 export type ChannelDecision = {
@@ -142,6 +143,8 @@ export type AssignChannelsResult = {
  /** 下一轮起点；到达表尾时回绕到 0，保证漏判的订单还能被重新评估。 */
  nextCursor: number
  byChannel: Record<string, number>
+  /** 本轮被判定的订单所属业务日，供事实/总账按日回填。 */
+  affectedDates: string[]
 }

 /** 批量判定：默认只扫 pending 且未锁定的订单项。 */
@@ -165,7 +168,9 @@ export async function assignPendingOrderItemChannels(
    lastId: cursor,
    nextCursor: cursor,
    byChannel: {},
+    affectedDates: [],
  }
+  const affectedDates = new Set<string>()

  for (let batch = 0; batch < maxBatches; batch += 1) {
    const rows = await query<ChannelCandidateItem>(
@@ -178,6 +183,8 @@ export async function assignPendingOrderItemChannels(
          oi.sku_name,
          oi.manual_sku_key,
          oi.normalized_sku_name,
+          (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date::text
+            AS basis_date,
          EXISTS (
            SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id
          ) AS has_work_order,
@@ -213,6 +220,7 @@ export async function assignPendingOrderItemChannels(

    for (const item of rows.rows) {
      const decision = resolveChannelDecision(item, rules)
+      if (decision.channel) affectedDates.add(item.basis_date)
      if (decision.conflict) {
        result.conflicted += 1
        await upsertAggregatedException({
@@ -258,6 +266,7 @@ export async function assignPendingOrderItemChannels(
  }

  result.nextCursor = reachedEnd ? 0 : cursor
+  result.affectedDates = [...affectedDates].sort()
  return result
 }

@@ -543,3 +552,45 @@ async function loadEnabledRules(): Promise<ChannelRule[]> {
 export const CHANNEL_RULE_KEY_TYPES = MATCH_KEY_TYPES
 export const CHANNEL_RULE_SOURCES = RULE_SOURCES
 export const CHANNEL_VALUES = FULFILLMENT_CHANNELS
+
+/**
+ * 为卡券商品（tpv: 前缀）批量补渠道规则。
+ *
+ * 卡券没有履约载体证据，只能靠规则判定；商品目录建好后一次补齐，
+ * 避免运营按 SKU 逐条点「指定渠道」。
+ */
+export async function ensureVoucherChannelRules(): Promise<{
+  created: number
+  productCount: number
+}> {
+  const created = await query<{ id: number }>(
+    `
+      INSERT INTO sku_channel_rules (
+        provider, shop_id, match_key_type, match_key, channel, priority,
+        source, enabled, note, created_at, updated_at
+      )
+      SELECT
+        '', '', 'sku_code', substring(p.canonical_sku from 5), 'third_party_voucher', 70,
+        'voucher', TRUE, '卡券商品自动生成', NOW(), NOW()
+      FROM product_catalog p
+      WHERE p.status = 'active'
+        AND p.canonical_sku LIKE 'tpv:%'
+        AND btrim(substring(p.canonical_sku from 5)) <> ''
+        AND NOT EXISTS (
+          SELECT 1 FROM sku_channel_rules r
+          WHERE r.provider = ''
+            AND r.shop_id = ''
+            AND r.match_key_type = 'sku_code'
+            AND r.match_key = substring(p.canonical_sku from 5)
+            AND r.channel = 'third_party_voucher'
+        )
+      ON CONFLICT DO NOTHING
+      RETURNING id
+    `,
+  )
+  const total = await query<{ count: number }>(
+    `SELECT COUNT(*)::int AS count FROM product_catalog
+     WHERE status = 'active' AND canonical_sku LIKE 'tpv:%'`,
+  )
+  return { created: created.rowCount || 0, productCount: Number(total.rows[0]?.count || 0) }
+}
@@ -1,11 +1,13 @@
 import type { JsonObject } from '../../types/json.js'
 import { query } from '../../db/client.js'
 import { assignPendingOrderItemChannels } from '../fulfillment/channel-assign-service.js'
+import { resolveExceptionsForItemsByReason } from '../cost-center/cost-exception-service.js'
+import { readWatermarkItemId } from '../watermark-store.js'
+import { rebuildChannelSalesDailyFacts } from '../finance/sales-facts-channel-service.js'
+import { syncFinanceLedgerFromFacts } from '../finance/finance-ledger-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'
 import {
  recordPipelineError,
@@ -45,18 +47,46 @@ export async function runChannelAssignJob(job: JsonObject = {}) {
  })

  await touchWatermark('channel_assign', { lastItemId: result.nextCursor })
+  const refreshed = await refreshFactsForDates(result.affectedDates)
  return {
    status: 'ok',
    message:
      `渠道判定：扫描 ${result.scanned}，已定 ${result.assigned}，` +
-      `冲突 ${result.conflicted}，仍待定 ${result.stillPending}`,
+      `冲突 ${result.conflicted}，仍待定 ${result.stillPending}，回填 ${refreshed} 天`,
    ...result,
+    refreshedDays: refreshed,
  }
 }

+/**
+ * 渠道/成本变更后，按业务日回填销售事实与总账。
+ *
+ * 事实任务只回看最近几天；补配价格、迟到分类这类变更必须按受影响日期重算，
+ * 否则旧日期的毛利会一直停在缺失成本/未分类。
+ */
+async function refreshFactsForDates(dates: string[]): Promise<number> {
+  const unique = [...new Set(dates.filter(Boolean))].sort().slice(0, 31)
+  let done = 0
+  for (const date of unique) {
+    try {
+      await rebuildChannelSalesDailyFacts({ dateFrom: date, dateTo: date })
+      await syncFinanceLedgerFromFacts({ dateFrom: date, dateTo: date })
+      done += 1
+    } catch {
+      // 单日失败不影响管道本身，下一轮或人工重跑会再处理。
+    }
+  }
+  return done
+}
+
 export async function runCarrierDispatchJob(job: JsonObject = {}) {
  const config = jobConfig(job)
  const limit = clampInt(config.limit, 200, 1, 2000)
+  const cursor =
+    config.afterId != null
+      ? Math.max(0, Math.trunc(Number(config.afterId)))
+      : await readWatermarkItemId('carrier_dispatch')
+  let maxId = 0

  const manual = await query<{ id: number; sku_key: string; config_id: number }>(
    `
@@ -66,10 +96,11 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) {
      WHERE oi.fulfillment_status = 'assigned'
        AND oi.fulfillment_channel = 'manual'
        AND oi.fulfillment_config_id IS NULL
+        AND oi.id > $2
      ORDER BY oi.id
      LIMIT $1
    `,
-    [limit],
+    [limit, cursor],
  )

  let attached = 0
@@ -105,10 +136,11 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) {
            AND ce.reason = 'manual_config_missing'
            AND ce.status <> 'resolved'
        )
+        AND oi.id > $2
      ORDER BY oi.id
      LIMIT $1
    `,
-    [limit],
+    [limit, cursor],
  )

  for (const row of missingManual.rows) {
@@ -142,10 +174,11 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) {
            AND ce.reason = 'manual_cost_missing'
            AND ce.status <> 'resolved'
        )
+        AND oi.id > $2
      ORDER BY oi.id
      LIMIT $1
    `,
-    [limit],
+    [limit, cursor],
  )

  for (const row of pendingManualCost.rows) {
@@ -180,10 +213,11 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) {
            AND ce.reason = 'carrier_missing'
            AND ce.status <> 'resolved'
        )
+        AND oi.id > $2
      ORDER BY oi.id
      LIMIT $1
    `,
-    [limit],
+    [limit, cursor],
  )

  for (const row of missingCarrier.rows) {
@@ -195,18 +229,122 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) {
    })
  }

-  await touchWatermark('carrier_dispatch')
+  for (const row of [
+    ...manual.rows,
+    ...missingManual.rows,
+    ...pendingManualCost.rows,
+    ...missingCarrier.rows,
+  ]) {
+    maxId = Math.max(maxId, Number(row.id))
+  }
+
+  const resolved = await resolveCarrierDispatchExceptions()
+  const nextCursor = maxId === 0 ? 0 : maxId
+  await touchWatermark('carrier_dispatch', { lastItemId: nextCursor })
  return {
    status: 'ok',
    message:
      `人工载体挂载 ${attached}，缺配置 ${missingManual.rows.length}，` +
-      `缺履约载体 ${missingCarrier.rows.length}`,
+      `缺履约载体 ${missingCarrier.rows.length}，异常收口 ${resolved}`,
    attached,
    missingManual: missingManual.rows.length,
    missingCarrier: missingCarrier.rows.length,
+    resolved,
  }
 }

+/**
+ * 载体/配置补上后，把对应异常自动关掉（每轮最多处理 500 条）。
+ */
+async function resolveCarrierDispatchExceptions(): Promise<number> {
+  // 游标会跳过已扫过的订单项；配置是后来补的，这里兜底把配置挂上。
+  await query(
+    `
+      UPDATE order_items oi
+      SET fulfillment_config_id = cfg.id, updated_at = NOW()
+      FROM manual_fulfillment_configs cfg
+      WHERE cfg.sku_key = oi.manual_sku_key
+        AND cfg.enabled
+        AND oi.fulfillment_status = 'assigned'
+        AND oi.fulfillment_channel = 'manual'
+        AND oi.fulfillment_config_id IS NULL
+        AND EXISTS (
+          SELECT 1
+          FROM cost_exception_items i
+          JOIN cost_exceptions ce ON ce.id = i.exception_id
+          WHERE i.order_item_id = oi.id
+            AND ce.reason = 'manual_config_missing'
+            AND ce.status <> 'resolved'
+        )
+    `,
+  )
+
+  const [carrier, config, manualCost] = await Promise.all([
+    query<{ order_item_id: number }>(
+      `
+        SELECT DISTINCT i.order_item_id
+        FROM cost_exception_items i
+        JOIN cost_exceptions ce ON ce.id = i.exception_id
+        JOIN order_items oi ON oi.id = i.order_item_id
+        WHERE ce.reason = 'carrier_missing'
+          AND ce.status <> 'resolved'
+          AND (
+            EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id)
+            OR EXISTS (
+              SELECT 1 FROM fulfillment_tasks ft
+              WHERE ft.order_item_id = oi.id
+                AND ft.executor_key IN ('affiliate_dash', 'third_party_voucher')
+            )
+          )
+        LIMIT 500
+      `,
+    ),
+    query<{ order_item_id: number }>(
+      `
+        SELECT DISTINCT i.order_item_id
+        FROM cost_exception_items i
+        JOIN cost_exceptions ce ON ce.id = i.exception_id
+        JOIN order_items oi ON oi.id = i.order_item_id
+        WHERE ce.reason = 'manual_config_missing'
+          AND ce.status <> 'resolved'
+          AND oi.fulfillment_config_id IS NOT NULL
+        LIMIT 500
+      `,
+    ),
+    query<{ order_item_id: number }>(
+      `
+        SELECT DISTINCT i.order_item_id
+        FROM cost_exception_items i
+        JOIN cost_exceptions ce ON ce.id = i.exception_id
+        WHERE ce.reason = 'manual_cost_missing'
+          AND ce.status <> 'resolved'
+          AND EXISTS (
+            SELECT 1 FROM cost_snapshots cs
+            WHERE cs.subject_type = 'manual_case'
+              AND cs.order_item_id = i.order_item_id
+              AND cs.superseded_at IS NULL
+          )
+        LIMIT 500
+      `,
+    ),
+  ])
+
+  let resolved = 0
+  resolved += await resolveExceptionsForItemsByReason(
+    carrier.rows.map((row) => Number(row.order_item_id)),
+    'carrier_missing',
+  )
+  resolved += await resolveExceptionsForItemsByReason(
+    config.rows.map((row) => Number(row.order_item_id)),
+    'manual_config_missing',
+  )
+  resolved += await resolveExceptionsForItemsByReason(
+    manualCost.rows.map((row) => Number(row.order_item_id)),
+    'manual_cost_missing',
+  )
+  return resolved
+}
+
 export async function runProductNormalizeJob(job: JsonObject = {}) {
  const config = jobConfig(job)
  const result = await normalizeOrderItemProducts({
@@ -230,10 +368,14 @@ export async function runCostSnapshotJob(job: JsonObject = {}) {
  })

  await touchWatermark('cost_snapshot', { lastItemId: result.nextCursor })
+  const refreshed = await refreshFactsForDates(result.affectedDates)
  return {
    status: 'ok',
-    message: `成本快照：扫描 ${result.scanned}，生成 ${result.created}，缺失 ${result.missing}`,
+    message:
+      `成本快照：扫描 ${result.scanned}，生成 ${result.created}，` +
+      `替换 ${result.superseded}，缺失 ${result.missing}，回填 ${refreshed} 天`,
    ...result,
+    refreshedDays: refreshed,
  }
 }