order_site/apps/backend/scripts/backfill-fulfillment-channel.ts

import process from 'node:process'

import { closeDb, query } from '../src/db/client.js'

/**
 * 112 之后的渠道回填：把旧 `fulfillment_mode` 转成
 * `fulfillment_channel` + `fulfillment_status` + `fulfillment_source`。
 *
 * 规则（2026-10-31 断代）：
 *   1. 业务日期 < 2026-10-31  → status=archived，source=cutover_archive；
 *      channel 按旧 mode 映射（legacy_platform 没有新渠道，置空）。
 *   2. 业务日期 >= 2026-10-31：
 *      · work_order / affiliate_dash → status=assigned，source=carrier_evidence；
 *      · manual → status=pending（旧 manual 是兜底值，必须由运营重新确认）；
 *      · 其余 / NULL → status=pending。
 *
 * 幂等：每批按 id 重算，只有值变化才写；运营锁定行（fulfillment_locked）跳过。
 * 分批：keyset 扫描，单批默认 5000，避免 438k 行整表 UPDATE 撞语句超时。
 *
 * 用法：
 *   npm run fulfillment:backfill-channel -- --dry-run
 *   npm run fulfillment:backfill-channel -- --batch-size=5000
 *   npm run fulfillment:backfill-channel -- --max-batches=100 --after-id=123456
 */

/** 财务断代日：这一天之前的数据只保留审计价值。 */
const CUTOVER_DATE = '2026-10-31'

const options = parseOptions(process.argv.slice(2))
const batchSize = normalizeInteger(options.batchSize, 5000, 1, 50_000)
const maxBatches = normalizeInteger(options.maxBatches, 1000, 1, 100_000)
const afterId = normalizeInteger(options.afterId, 0, 0, Number.MAX_SAFE_INTEGER)
const dryRun = options.dryRun === 'true' || options['dry-run'] === 'true'

const CHANNEL_SQL = `
  CASE oi.fulfillment_mode
    WHEN 'work_order' THEN 'work_order'
    WHEN 'affiliate_dash' THEN 'affiliate_dash'
    WHEN 'manual' THEN 'manual'
    ELSE NULL
  END
`

const DECISION_SQL = `
  CASE
    WHEN (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date
      < DATE '${CUTOVER_DATE}' THEN 'archived'
    WHEN oi.fulfillment_mode = 'manual' THEN 'pending'
    WHEN oi.fulfillment_mode IN ('work_order', 'affiliate_dash') THEN 'assigned'
    ELSE 'pending'
  END
`

const SOURCE_SQL = `
  CASE
    WHEN (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date
      < DATE '${CUTOVER_DATE}' THEN 'cutover_archive'
    WHEN oi.fulfillment_mode IN ('work_order', 'affiliate_dash') THEN 'carrier_evidence'
    ELSE ''
  END
`

type BatchRow = {
  id: number
  channel: string | null
  status: string
  source: string
  changed: boolean
}

async function main() {
  await assertSchema()

  const startedAt = Date.now()
  let cursor = afterId
  let scanned = 0
  let updated = 0
  const summary = new Map<string, number>()

  for (let batch = 0; batch < maxBatches; batch += 1) {
    const rows = await query<BatchRow>(
      `
        SELECT
          oi.id,
          ${CHANNEL_SQL} AS channel,
          ${DECISION_SQL} AS status,
          ${SOURCE_SQL} AS source,
          (
            oi.fulfillment_channel IS DISTINCT FROM (${CHANNEL_SQL})
            OR oi.fulfillment_status IS DISTINCT FROM (${DECISION_SQL})
            OR oi.fulfillment_source IS DISTINCT FROM (${SOURCE_SQL})
          ) AS changed
        FROM order_items oi
        JOIN orders o ON o.id = oi.order_id
        WHERE oi.id > $1
        ORDER BY oi.id
        LIMIT $2
      `,
      [cursor, batchSize],
    )

    if (rows.rows.length === 0) break

    scanned += rows.rows.length
    cursor = rows.rows[rows.rows.length - 1]!.id

    for (const row of rows.rows) {
      const key = `${row.status}/${row.channel ?? '-'}`
      summary.set(key, (summary.get(key) || 0) + 1)
    }

    const changedRows = rows.rows.filter((row) => row.changed)
    if (changedRows.length === 0) {
      process.stderr.write(`批 ${batch + 1}: 扫描 ${rows.rows.length}，无变化\n`)
      continue
    }

    if (dryRun) {
      updated += changedRows.length
      process.stderr.write(
        `批 ${batch + 1}: 扫描 ${rows.rows.length}，将更新 ${changedRows.length}\n`,
      )
      continue
    }

    const ids = changedRows.map((row) => row.id)
    const channels = changedRows.map((row) => row.channel)
    const statuses = changedRows.map((row) => row.status)
    const sources = changedRows.map((row) => row.source)

    const result = await query(
      `
        UPDATE order_items oi
        SET
          fulfillment_channel = input.channel,
          fulfillment_status = input.status,
          fulfillment_source = input.source,
          fulfillment_decided_at = NOW(),
          updated_at = NOW()
        FROM unnest($1::bigint[], $2::text[], $3::text[], $4::text[])
          AS input(id, channel, status, source)
        WHERE oi.id = input.id
          AND oi.fulfillment_locked = FALSE
      `,
      [ids, channels, statuses, sources],
    )

    updated += result.rowCount || 0
    process.stderr.write(
      `批 ${batch + 1}: 扫描 ${rows.rows.length}，更新 ${result.rowCount || 0}\n`,
    )
  }

  const remaining = await query<{ count: number }>(
    `SELECT COUNT(*)::int AS count FROM order_items WHERE id > $1`,
    [cursor],
  )

  console.log(
    JSON.stringify(
      {
        cutoverDate: CUTOVER_DATE,
        dryRun,
        batchSize,
        afterId,
        scanned,
        updated,
        lastId: cursor,
        remaining: remaining.rows[0]?.count || 0,
        summary: Object.fromEntries([...summary.entries()].sort()),
        durationMs: Date.now() - startedAt,
      },
      null,
      2,
    ),
  )

  if ((remaining.rows[0]?.count || 0) > 0) {
    console.error(
      `仍有 ${remaining.rows[0]?.count} 个订单项未扫描；请增大 --max-batches 或用 --after-id=${cursor} 续跑。`,
    )
    process.exitCode = 2
  }
}

main()
  .catch((error) => {
    console.error(error instanceof Error ? error.message : error)
    process.exitCode = 1
  })
  .finally(async () => {
    await closeDb()
  })

async function assertSchema() {
  const columns = await query<{ column_name: string }>(
    `
      SELECT column_name
      FROM information_schema.columns
      WHERE table_schema = 'public'
        AND table_name = 'order_items'
        AND column_name IN (
          'fulfillment_mode', 'fulfillment_channel',
          'fulfillment_status', 'fulfillment_source', 'fulfillment_locked'
        )
    `,
  )
  const found = new Set(columns.rows.map((row) => row.column_name))
  const missing = [
    'fulfillment_mode',
    'fulfillment_channel',
    'fulfillment_status',
    'fulfillment_source',
    'fulfillment_locked',
  ].filter((name) => !found.has(name))

  if (missing.length > 0) {
    throw new Error(`order_items 缺少列：${missing.join(', ')}；请先执行迁移 112`)
  }
}

function parseOptions(args: string[]) {
  const result: Record<string, string> = {}
  for (const arg of args) {
    if (!arg.startsWith('--')) continue
    const body = arg.slice(2)
    const index = body.indexOf('=')
    const rawKey = index === -1 ? body : body.slice(0, index)
    const value = index === -1 ? 'true' : body.slice(index + 1)
    result[rawKey] = value
    // 同时支持 --batch-size 与 --batchSize 两种写法。
    result[rawKey.replace(/-([a-z])/g, (_, char: string) => char.toUpperCase())] = value
  }
  return result
}

function normalizeInteger(value: unknown, fallback: number, min: number, max: number) {
  const parsed = Number(value)
  if (!Number.isInteger(parsed)) return fallback
  return Math.min(max, Math.max(min, parsed))
}