order_site/apps/backend/src/services/fulfillment/affiliate-fallback-service.ts

import { query, withTransaction } from '../../db/client.js'
import { syncCostSnapshotsForItems } from '../cost-center/cost-snapshot-service.js'
import { refreshFactsForDates } from '../finance/refresh-facts.js'

/**
 * 自动发货失败 → 人工兜底。
 *
 * 触发：该订单项最新一条 affiliate_dash 任务处于失败状态，且没有任何成功任务。
 * 切换后：原渠道成本（sys_affiliate_purchase）冲销，人工成本等确认时再记；
 * 自动任务本身不改状态，只记录订单项的切换原因，避免重复触发。
 */
export const AFFILIATE_FAILURE_STATUSES = [
  'failed',
  'closed',
  'expired',
  'manual_review',
  'retry_pending',
  'cancelled',
] as const

const AFFILIATE_SUCCESS_STATUSES = ['redeemed', 'paid'] as const

export async function runAffiliateFallback(
  options: { limit?: number } = {},
): Promise<{ scanned: number; switched: number }> {
  const limit = Math.min(1000, Math.max(1, Math.trunc(options.limit || 200)))
  const candidates = await query<{
    id: number
    order_item_id: number
    provider: string
    shop_id: string
    internal_sku: string
    sku_name: string
    config_id: number | null
    config_product_id: number | null
    basis_date: string
  }>(
    `
      SELECT
        oi.id,
        oi.id AS order_item_id,
        o.provider,
        o.shop_id,
        resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code) AS internal_sku,
        oi.sku_name,
        cfg.id AS config_id,
        cfg.canonical_product_id AS config_product_id,
        (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date::text AS basis_date
      FROM order_items oi
      JOIN orders o ON o.id = oi.order_id
      LEFT JOIN manual_fulfillment_configs cfg
        ON canonicalize_internal_sku(cfg.sku_key) = resolve_internal_sku(
          oi.manual_sku_key, oi.sku_name, oi.sku_code
        )
       AND cfg.enabled
      WHERE oi.fulfillment_status = 'assigned'
        AND oi.fulfillment_channel = 'affiliate_dash'
        AND oi.fulfillment_locked = FALSE
        AND cfg.id IS NOT NULL
        AND EXISTS (
          SELECT 1 FROM internal_sku_channels channel
          WHERE channel.enabled = TRUE
            AND channel.channel = 'manual'
            AND channel.is_fallback = TRUE
            AND canonicalize_internal_sku(channel.internal_sku) = resolve_internal_sku(
              oi.manual_sku_key, oi.sku_name, oi.sku_code
            )
        )
        AND (
          SELECT task.task_status
          FROM fulfillment_tasks task
          WHERE task.order_item_id = oi.id AND task.executor_key = 'affiliate_dash'
          ORDER BY task.unit_index DESC, task.updated_at DESC, task.id DESC
          LIMIT 1
        ) = ANY($1::text[])
        AND NOT EXISTS (
          SELECT 1 FROM fulfillment_tasks ft
          WHERE ft.order_item_id = oi.id
            AND ft.executor_key = 'affiliate_dash'
            AND ft.task_status = ANY($2::text[])
        )
      ORDER BY oi.id
      LIMIT $3
    `,
    [AFFILIATE_FAILURE_STATUSES, AFFILIATE_SUCCESS_STATUSES, limit],
  )

  const switchedIds: number[] = []
  const affectedDates = new Set<string>()

  for (const row of candidates.rows) {
    const updated = await withTransaction(async (client) => {
      const result = await client.query(
        `
          UPDATE order_items
          SET
            original_channel = COALESCE(original_channel, 'affiliate_dash'),
            fulfillment_channel = 'manual',
            fulfillment_status = 'assigned',
            fulfillment_source = 'affiliate_fallback',
            channel_switch_reason = 'affiliate_failed',
            channel_switched_at = NOW(),
            fulfillment_config_id = COALESCE(fulfillment_config_id, $2),
            canonical_product_id = COALESCE($3, canonical_product_id),
            fulfillment_pending_reason = NULL,
            fulfillment_pending_detail = NULL,
            fulfillment_decided_at = NOW(),
            item_snapshot_json = COALESCE(item_snapshot_json, '{}'::jsonb) || jsonb_build_object(
              'internalSkuChannel', 'manual',
              'internalSkuChannelSource', 'affiliate_fallback',
              'internalSkuChannelReason', '自动发货失败，转人工兜底',
              'isConfigured', TRUE,
              'affiliateDash', NULL,
              'fulfillmentRoute', jsonb_build_object(
                'selectedExecutorKey', 'manual_dispatch',
                'selectedRuleId', '',
                'reason', '自动发货失败，转人工兜底',
                'matchedRule', NULL,
                'candidates', '[]'::jsonb,
                'skipped', '[]'::jsonb
              ),
              'classification', jsonb_build_object(
                'channel', 'manual',
                'canonicalSku', resolve_internal_sku(manual_sku_key, sku_name, sku_code)
              )
            ),
            updated_at = NOW()
          WHERE id = $1
            AND fulfillment_status = 'assigned'
            AND fulfillment_channel = 'affiliate_dash'
            AND fulfillment_locked = FALSE
          RETURNING id
        `,
        [row.id, row.config_id, row.config_product_id],
      )
      if ((result.rowCount || 0) === 0) return false

      await client.query(
        `
          UPDATE cost_snapshots
          SET reversed_at = NOW(), reversal_reason = 'affiliate_failed'
          WHERE order_item_id = $1
            AND cost_item_id = (SELECT id FROM cost_items WHERE code = 'sys_affiliate_purchase')
            AND superseded_at IS NULL AND reversed_at IS NULL
            AND reversed_at IS NULL
        `,
        [row.id],
      )
      return true
    })
    if (!updated) continue
    switchedIds.push(Number(row.id))
    affectedDates.add(row.basis_date)
  }

  for (const id of switchedIds) {
    await syncCostSnapshotsForItems([id]).catch(() => undefined)
  }
  await refreshFactsForDates([...affectedDates]).catch(() => undefined)

  return { scanned: candidates.rows.length, switched: switchedIds.length }
}