order_site/apps/backend/scripts/finance-cutover-v2.ts

import process from 'node:process'

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

/**
 * 财务断代 v2：清理断代前的派生数据，为 2026-10-31 起点重建让路。
 *
 * 只删「派生数据」，不删业务原始票据：
 *   删除：cost_snapshots / cost_exceptions(+items) / sales_daily_facts /
 *         finance_journal_lines(+entries) / finance_reconciliations(+items) /
 *         finance_refund_intents / order_refunds / manual_expenses /
 *         fund_entries / platform_credit_entries / manual_sku_registry
 *   保留：orders / order_items / work_orders / fulfillment_tasks（审计只读）
 *         product_catalog / channel_product_mappings / cost_items /
 *         cost_versions / work_product_rules / manual_fulfillment_configs /
 *         sku_channel_rules（配置）
 *
 * 前置：
 *   1. 迁移 112/113 已执行；
 *   2. 已做数据库备份。
 *
 * 断代本身包含两步：先把断代前的订单项批量置为 archived，再清理派生数据。
 *
 * 默认 dry-run，只打印将删除的行数；加 `--execute --confirm=2026-10-31` 才真正执行。
 * 不可重复执行：成功后会写入 finance_cutovers，第二次会被拦下。
 */

const CUTOVER_DATE = '2026-10-31'

const options = parseOptions(process.argv.slice(2))
const execute = options.execute === 'true'
const confirm = String(options.confirm || '').trim()

/** 派生表：按外键顺序删除。 */
const DERIVED_TABLES: Array<{ table: string; order: number; where?: string }> = [
  { table: 'cost_exception_items', order: 10 },
  { table: 'cost_exceptions', order: 20 },
  { table: 'cost_snapshots', order: 30 },
  { table: 'sales_daily_facts', order: 40 },
  { table: 'finance_journal_lines', order: 50 },
  { table: 'finance_journal_entries', order: 60 },
  { table: 'finance_reconciliation_items', order: 70 },
  { table: 'finance_reconciliations', order: 80 },
  { table: 'finance_refund_intents', order: 90 },
  { table: 'order_refunds', order: 100 },
  { table: 'manual_expenses', order: 110 },
  { table: 'fund_entries', order: 120 },
  { table: 'platform_credit_entries', order: 130 },
  { table: 'manual_sku_registry', order: 140 },
]

async function main() {
  await assertPreconditions()

  const startedAt = Date.now()
  const plan = await buildPlan()

  if (!execute) {
    console.log(
      JSON.stringify(
        {
          mode: 'dry-run',
          cutoverDate: CUTOVER_DATE,
          deletePlan: plan,
          hint: '确认无误后追加 --execute --confirm=2026-10-31',
          durationMs: Date.now() - startedAt,
        },
        null,
        2,
      ),
    )
    return
  }

  if (confirm !== CUTOVER_DATE) {
    throw new Error(`--confirm 必须是 ${CUTOVER_DATE}，当前为 "${confirm}"`)
  }

  await assertNotExecuted()

  const archived = await archiveHistoricalItems()

  const deleted: Record<string, number> = {}
  for (const item of [...DERIVED_TABLES].sort((a, b) => a.order - b.order)) {
    const result = await query(
      `DELETE FROM ${item.table}${item.where ? ` WHERE ${item.where}` : ''}`,
    )
    deleted[item.table] = result.rowCount || 0
    process.stderr.write(`删除 ${item.table}: ${result.rowCount || 0}\n`)
  }

  // 旧归档表不再保留：生产库不留可查询的历史异常副本。
  await query(`DROP TABLE IF EXISTS cost_exceptions_archive`)

  // 水位归零：管道从断代日起重新扫描。
  await query(
    `
      UPDATE finance_watermarks
      SET last_item_id = 0, last_stat_date = $1::date, last_success_at = NULL,
          last_error = '', updated_at = NOW()
    `,
    [CUTOVER_DATE],
  )

  await query(
    `
      INSERT INTO finance_cutovers (cutover_date, executed_at, detail_json)
      VALUES ($1::date, NOW(), $2::jsonb)
      ON CONFLICT (cutover_date) DO NOTHING
    `,
    [CUTOVER_DATE, JSON.stringify({ deleted, tables: DERIVED_TABLES.map((t) => t.table) })],
  )

  console.log(
    JSON.stringify(
      {
        mode: 'execute',
        cutoverDate: CUTOVER_DATE,
        archivedItems: archived.updated,
        deleted,
        durationMs: Date.now() - startedAt,
      },
      null,
      2,
    ),
  )
}

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

async function buildPlan() {
  const plan: Array<{ table: string; rows: number }> = []
  for (const item of [...DERIVED_TABLES].sort((a, b) => a.order - b.order)) {
    const result = await query<{ count: number }>(
      `SELECT COUNT(*)::int AS count FROM ${item.table}${item.where ? ` WHERE ${item.where}` : ''}`,
    )
    plan.push({ table: item.table, rows: result.rows[0]?.count || 0 })
  }
  return plan
}

async function assertPreconditions() {
  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_channel', 'fulfillment_status', 'fulfillment_source')
    `,
  )
  if (columns.rows.length < 3) {
    throw new Error('缺少 112 的三列；请先执行迁移 112')
  }

  const watermark = await query<{ exists: boolean }>(
    `SELECT to_regclass('public.finance_watermarks') IS NOT NULL AS exists`,
  )
  if (!watermark.rows[0]?.exists) {
    throw new Error('finance_watermarks 不存在；请先执行迁移 112')
  }

  await query(
    `
      CREATE TABLE IF NOT EXISTS finance_cutovers (
        cutover_date DATE PRIMARY KEY,
        executed_at  TIMESTAMPTZ NOT NULL DEFAULT NOW(),
        detail_json  JSONB NOT NULL DEFAULT '{}'::jsonb
      )
    `,
  )

  const inconsistent = await query<{ count: number }>(
    `
      SELECT COUNT(*)::int AS count
      FROM order_items
      WHERE fulfillment_status = 'assigned' AND fulfillment_channel IS NULL
    `,
  )
  const bad = inconsistent.rows[0]?.count || 0
  if (bad > 0) {
    throw new Error(
      `有 ${bad} 行 status=assigned 但 channel 为空；请先执行 fulfillment:backfill-channel`,
    )
  }
}

/**
 * 断代前的订单项统一置为 archived（分批，避免 42 万行整表更新撞超时）。
 */
async function archiveHistoricalItems(): Promise<{ updated: number }> {
  let cursor = 0
  let updated = 0
  for (let batch = 0; batch < 500; batch += 1) {
    const rows = await query<{ id: string }>(
      `
        SELECT oi.id::text
        FROM order_items oi
        JOIN orders o ON o.id = oi.order_id
        WHERE (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date < $1::date
          AND oi.fulfillment_status <> 'archived'
          AND oi.id > $2
        ORDER BY oi.id
        LIMIT 20000
      `,
      [CUTOVER_DATE, cursor],
    )
    if (rows.rows.length === 0) break
    cursor = Number(rows.rows[rows.rows.length - 1]!.id)

    const result = await query(
      `
        UPDATE order_items
        SET fulfillment_status = 'archived',
            fulfillment_channel = NULL,
            fulfillment_source = 'cutover_archive',
            fulfillment_decided_at = NOW(),
            updated_at = NOW()
        WHERE id = ANY($1::bigint[])
          AND fulfillment_status <> 'archived'
      `,
      [rows.rows.map((row) => Number(row.id))],
    )
    updated += result.rowCount || 0
    process.stderr.write(`归档 ${result.rowCount || 0} 行（游标 ${cursor}）\n`)
  }
  return { updated }
}

async function assertNotExecuted() {
  const done = await query<{ count: number }>(
    `SELECT COUNT(*)::int AS count FROM finance_cutovers WHERE cutover_date = $1::date`,
    [CUTOVER_DATE],
  )
  if ((done.rows[0]?.count || 0) > 0) {
    throw new Error(`${CUTOVER_DATE} 的断代已执行过，禁止重复执行`)
  }
}

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
    result[rawKey.replace(/-([a-z])/g, (_, char: string) => char.toUpperCase())] = value
  }
  return result
}