order_site/apps/backend/src/db/migrate.ts

/**
 * 数据库迁移入口。
 *
 * 策略：
 * - 001_init.sql：空库基线，建立完整结构
 * - 002_xxx.sql 起：渐进增量迁移，禁止改已应用历史文件
 * - 启动时自动 up；也可用 npm run db:migrate 手动执行
 * - 新建迁移：npm run db:migrate:create -- <name>
 * - 查看状态：npm run db:migrate:status
 */
import fs from 'node:fs'
import path from 'node:path'
import process from 'node:process'
import { fileURLToPath } from 'node:url'

import { runner, type RunnerOption } from 'node-pg-migrate'
import type { ClientConfig } from 'pg'

import { runtimeConfig } from '../config/runtime.js'
import { closeDb, query } from './client.js'

const CURRENT_DIR = path.dirname(fileURLToPath(import.meta.url))
const MIGRATIONS_DIR = path.join(CURRENT_DIR, 'migrations')
const MIGRATIONS_TABLE = 'schema_migrations'
const SHORT_NUMERIC_PREFIX_MESSAGE = /^Can't determine timestamp for \d+$/
const MIGRATION_FILE_PATTERN = /^(\d{3})_.+\.sql$/i

export async function runDatabaseMigrations() {
  assertMigrationFilesOrdered()
  await ensureNodePgMigrateMetadataCompatibility()

  console.info(`[db:migrate] 迁移目录: ${MIGRATIONS_DIR}`)

  await runner({
    databaseUrl: createMigrationDatabaseConfig(),
    dir: MIGRATIONS_DIR,
    direction: 'up',
    migrationsTable: MIGRATIONS_TABLE,
    migrationsSchema: 'public',
    schema: 'public',
    singleTransaction: true,
    checkOrder: true,
    logger: {
      debug: () => undefined,
      info: (message) => {
        console.info(`[db:migrate] ${message}`)
      },
      warn: (message) => {
        console.warn(`[db:migrate] ${message}`)
      },
      error: (message) => {
        if (SHORT_NUMERIC_PREFIX_MESSAGE.test(message)) {
          return
        }
        console.error(`[db:migrate] ${message}`)
      },
    },
  } satisfies RunnerOption)

  await syncLegacyMigrationColumns()
  await warnPendingRevenueBackfill()
}

/**
 * 迁移 104 只建列、不再做全表收入回填（37.5 万行实测约 34 秒，会拖住启动）。
 * 这里只做一次带索引的计数，发现仍有未固化的订单项就提示执行回填脚本，
 * 不阻塞启动——脚本幂等，可以随时补跑。
 */
async function warnPendingRevenueBackfill() {
  try {
    const { rows } = await query<{ count: number }>(
      `SELECT COUNT(*)::int AS count FROM order_items WHERE revenue_basis = 'unknown'`,
    )
    const pending = rows[0]?.count ?? 0
    if (pending > 0) {
      console.warn(
        `[db:migrate] 还有 ${pending} 个订单项未固化收入口径，销售事实会偏低；` +
          '请执行 npm run revenue:backfill（分批、幂等）',
      )
    }
  } catch (error) {
    // 列还不存在（迁移没跑到 104）或表缺失时跳过，不能影响启动。
    console.warn(
      `[db:migrate] 跳过收入口径检查: ${error instanceof Error ? error.message : String(error)}`,
    )
  }
}

function assertMigrationFilesOrdered() {
  if (!fs.existsSync(MIGRATIONS_DIR)) {
    throw new Error(`迁移目录不存在: ${MIGRATIONS_DIR}`)
  }

  const files = fs
    .readdirSync(MIGRATIONS_DIR)
    .filter((name) => name.endsWith('.sql'))
    .sort((a, b) => a.localeCompare(b, 'en'))

  if (files.length === 0) {
    throw new Error(`迁移目录为空: ${MIGRATIONS_DIR}`)
  }

  const invalid = files.filter((name) => !MIGRATION_FILE_PATTERN.test(name))
  if (invalid.length > 0) {
    throw new Error(`迁移文件命名必须为 NNN_name.sql（三位序号），非法文件: ${invalid.join(', ')}`)
  }

  const versions = files.map((name) => Number(name.slice(0, 3)))
  for (let index = 1; index < versions.length; index += 1) {
    const current = versions[index]
    const previous = versions[index - 1]
    const currentFile = files[index]
    const previousFile = files[index - 1]
    if (current === undefined || previous === undefined || !currentFile || !previousFile) {
      continue
    }
    if (current < previous) {
      throw new Error(`迁移序号乱序: ${previousFile} 之后出现 ${currentFile}`)
    }
    if (current === previous) {
      throw new Error(`迁移序号重复: ${previousFile} 与 ${currentFile}`)
    }
  }

  if (files[0] !== '001_init.sql') {
    console.warn(
      `[db:migrate] 警告: 首个迁移不是 001_init.sql（当前为 ${files[0]}），请确认是否刻意调整基线`,
    )
  }
}

async function ensureNodePgMigrateMetadataCompatibility() {
  await query(`
    CREATE TABLE IF NOT EXISTS ${MIGRATIONS_TABLE} (
      id BIGSERIAL PRIMARY KEY,
      name VARCHAR(255),
      run_on TIMESTAMPTZ,
      filename TEXT,
      applied_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
    )
  `)

  await query(`
    ALTER TABLE ${MIGRATIONS_TABLE}
      ADD COLUMN IF NOT EXISTS name VARCHAR(255),
      ADD COLUMN IF NOT EXISTS run_on TIMESTAMPTZ,
      ADD COLUMN IF NOT EXISTS filename TEXT,
      ADD COLUMN IF NOT EXISTS applied_at TIMESTAMPTZ
  `)

  await query(`
    ALTER TABLE ${MIGRATIONS_TABLE}
      ALTER COLUMN filename DROP NOT NULL,
      ALTER COLUMN applied_at SET DEFAULT NOW()
  `)

  await query(`
    UPDATE ${MIGRATIONS_TABLE}
    SET
      name = COALESCE(
        NULLIF(name, ''),
        NULLIF(REGEXP_REPLACE(COALESCE(filename, ''), '\\.[^.]+$', ''), ''),
        CONCAT('legacy_', id)
      ),
      run_on = COALESCE(run_on, applied_at, NOW())
    WHERE name IS NULL
      OR name = ''
      OR run_on IS NULL
  `)

  await query(`
    ALTER TABLE ${MIGRATIONS_TABLE}
      ALTER COLUMN name SET NOT NULL,
      ALTER COLUMN run_on SET NOT NULL
  `)

  await query(`
    CREATE UNIQUE INDEX IF NOT EXISTS idx_schema_migrations_name
      ON ${MIGRATIONS_TABLE}(name)
  `)
}

async function syncLegacyMigrationColumns() {
  await query(`
    UPDATE ${MIGRATIONS_TABLE}
    SET
      filename = COALESCE(NULLIF(filename, ''), CONCAT(name, '.sql')),
      applied_at = COALESCE(applied_at, run_on, NOW())
    WHERE filename IS NULL
      OR filename = ''
      OR applied_at IS NULL
  `)
}

function createMigrationDatabaseConfig(): ClientConfig {
  const connectionString = String(runtimeConfig.database?.url || '').trim()
  if (!connectionString) {
    throw new Error('DATABASE_URL 未配置，无法执行数据库迁移')
  }

  return {
    connectionString,
    ssl: runtimeConfig.database?.ssl ? { rejectUnauthorized: false } : false,
  }
}

const currentFilePath = fileURLToPath(import.meta.url)

if (process.argv[1] && path.resolve(process.argv[1]) === currentFilePath) {
  runDatabaseMigrations()
    .then(async () => {
      await closeDb()
    })
    .catch(async (error) => {
      console.error(error)
      await closeDb()
      process.exitCode = 1
    })
}