/** * 数据库迁移入口。 * * 策略: * - 001_init.sql:空库基线,建立完整结构 * - 002_xxx.sql 起:渐进增量迁移,禁止改已应用历史文件 * - 启动时自动 up;也可用 npm run db:migrate 手动执行 * - 新建迁移:npm run db:migrate:create -- * - 查看状态: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() } 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 }) }