order_site/apps/backend/scripts/reclassify-post-cutover.ts

/**
 * 断代后订单按新的商品渠道能力重判。
 *
 * 仅处理：assigned、未锁定、无成功履约证据、有商品能力的订单项。
 * 默认 dry-run；--apply 才写库。
 */
import process from 'node:process'

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

const CHANNEL_COST_CODE: Record<string, string> = {
  affiliate_dash: 'sys_affiliate_purchase',
  third_party_voucher: 'sys_third_party_voucher_cost',
  work_order: 'sys_work_order_reward',
}

type Candidate = {
  id: number
  current_channel: string
  desired_channel: string
  basis_date: string
}

async function loadCandidates(limit: number): Promise<Candidate[]> {
  const result = await query<Candidate>(
    `
      SELECT oi.id, oi.fulfillment_channel AS current_channel,
             cap.channel AS desired_channel,
             (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
      JOIN LATERAL (
        SELECT c.channel
        FROM internal_sku_channels c
        WHERE c.enabled
          AND c.channel IN ('affiliate_dash', 'manual', 'third_party_voucher')
          AND btrim(c.internal_sku) = COALESCE(NULLIF(btrim(oi.manual_sku_key), ''), btrim(oi.sku_name))
        ORDER BY
          c.priority ASC,
          CASE c.channel
            WHEN 'affiliate_dash' THEN 1
            WHEN 'work_order' THEN 2
            WHEN 'third_party_voucher' THEN 3
            WHEN 'manual' THEN 4
            ELSE 9
          END,
          c.channel ASC
        LIMIT 1
      ) cap ON TRUE
      WHERE oi.fulfillment_status = 'assigned'
        AND oi.fulfillment_locked = FALSE
        AND COALESCE(o.paid_at, o.created_at) >= '2026-10-31 00:00:00+08'
        AND oi.fulfillment_source NOT IN ('operator_assigned', 'manual_confirm')
        AND (
          cap.channel <> 'affiliate_dash'
          OR EXISTS (
            SELECT 1 FROM fulfillment_tasks ft
            WHERE ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash'
          )
        )
        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 IN ('redeemed', 'paid')
        )
        AND NOT EXISTS (
          SELECT 1 FROM work_orders wo
          WHERE wo.order_item_id = oi.id AND wo.status = 'accepted'
        )
        AND NOT EXISTS (
          SELECT 1 FROM cost_snapshots cs
          JOIN cost_items ci ON ci.id = cs.cost_item_id
          WHERE cs.order_item_id = oi.id
            AND ci.code = 'sys_manual_service'
            AND cs.superseded_at IS NULL
            AND cs.reversed_at IS NULL
        )
      ORDER BY oi.id
      LIMIT $1
    `,
    [limit],
  )
  return result.rows
}

async function applyOne(row: Candidate): Promise<boolean> {
  const previous = row.current_channel
  if (!previous || previous === row.desired_channel) return false

  const changed = await withTransaction(async (client) => {
    const result = await client.query(
      `
        UPDATE order_items
        SET original_channel = COALESCE(original_channel, $2),
            fulfillment_channel = $3,
            fulfillment_source = 'auto_rule',
            channel_switch_reason = 'reclassify',
            channel_switched_at = NOW(),
            updated_at = NOW()
        WHERE id = $1
          AND fulfillment_status = 'assigned'
          AND fulfillment_locked = FALSE
          AND fulfillment_channel = $2
        RETURNING id
      `,
      [row.id, previous, row.desired_channel],
    )
    if ((result.rowCount || 0) === 0) return false

    const code = CHANNEL_COST_CODE[previous]
    if (code) {
      await client.query(
        `
          UPDATE cost_snapshots
          SET reversed_at = NOW(), reversal_reason = 'reclassify'
          WHERE order_item_id = $1
            AND cost_item_id = (SELECT id FROM cost_items WHERE code = $2)
            AND superseded_at IS NULL
            AND reversed_at IS NULL
        `,
        [row.id, code],
      )
    }
    return true
  })
  if (!changed) return false

  await syncCostSnapshotsForItems([row.id]).catch(() => undefined)
  return true
}

async function main() {
  const apply = process.argv.includes('--apply')
  const limit = 5000
  const rows = await loadCandidates(limit)
  const changes = rows.filter((row) => row.current_channel !== row.desired_channel)
  console.log(
    JSON.stringify(
      {
        mode: apply ? 'apply' : 'dry-run',
        scanned: rows.length,
        toChange: changes.length,
        sample: changes.slice(0, 20).map((row) => ({
          id: row.id,
          from: row.current_channel,
          to: row.desired_channel,
          date: row.basis_date,
        })),
      },
      null,
      2,
    ),
  )
  if (!apply || changes.length === 0) return

  let switched = 0
  const dates = new Set<string>()
  for (const row of changes) {
    if (await applyOne(row)) {
      switched += 1
      dates.add(row.basis_date)
    }
  }
  await refreshFactsForDates([...dates])
  console.log(JSON.stringify({ switched, dates: [...dates].sort() }, null, 2))
}

main()
  .catch((error) => {
    console.error(error)
    process.exitCode = 1
  })
  .finally(() => closeDb())