#!/usr/bin/env node import { createHash } from 'node:crypto' import { mkdir, open, readFile, readdir, stat, writeFile } from 'node:fs/promises' import { spawnSync } from 'node:child_process' import path from 'node:path' import process from 'node:process' import { fileURLToPath } from 'node:url' const SCRIPT_DIR = path.dirname(fileURLToPath(import.meta.url)) const SQL_ROOT = path.resolve(SCRIPT_DIR, '..') const SCOPES_PATH = path.join(SQL_ROOT, 'migration-scopes.json') const MIGRATION_NAME = /^V(\d+)__([a-z0-9][a-z0-9_-]*)\.sql$/ function fail(message) { throw new Error(message) } function parseArgs(argv) { const [command = 'help', ...rest] = argv const args = { command, _: [] } for (let index = 0; index < rest.length; index += 1) { const item = rest[index] if (!item.startsWith('--')) { args._.push(item) continue } const equals = item.indexOf('=') if (equals >= 0) { args[item.slice(2, equals)] = item.slice(equals + 1) continue } const key = item.slice(2) const next = rest[index + 1] if (next && !next.startsWith('--')) { args[key] = next index += 1 } else { args[key] = true } } return args } function absoluteFrom(root, value) { return path.isAbsolute(value) ? value : path.resolve(root, value) } async function loadScopes(scopesPath = SCOPES_PATH) { const scopes = JSON.parse(await readFile(scopesPath, 'utf8')) if (scopes.schemaVersion !== 1 || !scopes.regions || typeof scopes.regions !== 'object') { fail('migration-scopes.json 必须使用 schemaVersion=1 并提供 regions') } const seenSchemas = new Set() for (const [region, config] of Object.entries(scopes.regions)) { if (!/^[A-Z][A-Z0-9_]*$/.test(region)) fail(`区域名非法:${region}`) if (!Array.isArray(config.schemas) || config.schemas.length === 0) fail(`${region} 没有 schema 清单`) const orders = new Set() for (const schema of config.schemas) { if (!/^[a-z][a-z0-9_]*$/.test(schema.name)) fail(`schema 名称非法:${schema.name}`) if (seenSchemas.has(schema.name)) fail(`schema 重复登记:${schema.name}`) if (!Number.isInteger(schema.order) || schema.order < 1 || orders.has(schema.order)) fail(`${region} 的 schema order 非法或重复:${schema.name}`) if (!Array.isArray(schema.services) || schema.services.length === 0) fail(`${schema.name} 没有所属服务`) seenSchemas.add(schema.name) orders.add(schema.order) } } return scopes } function regionConfig(scopes, region) { const value = scopes.regions[region] if (!value) fail(`未知区域:${region};允许值:${Object.keys(scopes.regions).join(', ')}`) return value } function schemaConfig(scopes, region, schemaName) { const config = regionConfig(scopes, region) const schema = config.schemas.find((item) => item.name === schemaName) if (!schema) fail(`${region} 未登记 schema:${schemaName}`) return schema } async function listMigrations(sqlRoot, region, schemaName) { const directory = path.join(sqlRoot, 'migrations', region.toLowerCase(), schemaName) let entries = [] try { entries = await readdir(directory, { withFileTypes: true }) } catch (error) { if (error.code === 'ENOENT') return [] throw error } const migrations = [] for (const entry of entries) { if (!entry.isFile() || entry.name.startsWith('.')) continue const match = entry.name.match(MIGRATION_NAME) if (!match) fail(`${path.relative(sqlRoot, path.join(directory, entry.name))} 文件名不符合 V__description.sql`) const version = Number(match[1]) if (!Number.isSafeInteger(version) || version < 1) fail(`迁移版本非法:${entry.name}`) migrations.push({ version, description: match[2], fileName: entry.name, relativePath: path.relative(sqlRoot, path.join(directory, entry.name)).replaceAll(path.sep, '/'), absolutePath: path.join(directory, entry.name), }) } migrations.sort((left, right) => left.version - right.version || left.fileName.localeCompare(right.fileName)) for (let index = 1; index < migrations.length; index += 1) { if (migrations[index - 1].version === migrations[index].version) fail(`${region}/${schemaName} 版本重复:${migrations[index].version}`) } return migrations } async function sha256(filePath) { return createHash('sha256').update(await readFile(filePath)).digest('hex') } function selectedServices(value) { if (!value || value === 'ALL') return null const services = value.split(',').map((item) => item.trim()).filter(Boolean) if (services.length === 0) fail('--services 不能为空') return [...new Set(services)] } function selectSchemas(scopes, region, services) { const config = regionConfig(scopes, region) const selected = services === null ? config.schemas : config.schemas.filter((schema) => schema.services.some((service) => services.includes(service))) return [...selected].sort((left, right) => left.order - right.order) } async function buildPlan({ sqlRoot = SQL_ROOT, scopes, regions, services, commit = process.env.GIT_COMMIT ?? null }) { const knownServices = new Set(scopes.services ?? Object.values(scopes.regions).flatMap((config) => config.schemas.flatMap((schema) => schema.services))) const unknownServices = (services ?? []).filter((service) => !knownServices.has(service)) if (unknownServices.length) fail(`范围映射中未登记服务:${unknownServices.join(', ')}`) const plan = { schemaVersion: 1, sourceCommit: commit, services: services ?? ['ALL'], regions: [], } for (const region of regions) { const config = regionConfig(scopes, region) const schemas = [] for (const schema of selectSchemas(scopes, region, services)) { const migrations = await listMigrations(sqlRoot, region, schema.name) schemas.push({ schema: schema.name, services: schema.services, order: schema.order, migrationDirectory: `migrations/${region.toLowerCase()}/${schema.name}`, baseline: config.baseline, target: migrations.length ? String(migrations.at(-1).version) : '0.0.0', migrations: await Promise.all(migrations.map(async (migration) => ({ version: String(migration.version), description: migration.description, path: migration.relativePath, sha256: await sha256(migration.absolutePath), }))), }) } plan.regions.push({ region, schemas }) } return plan } async function createMigration({ sqlRoot = SQL_ROOT, scopes, region, schemaName, description }) { const schema = schemaConfig(scopes, region, schemaName) const directory = path.join(sqlRoot, 'migrations', region.toLowerCase(), schema.name) await mkdir(directory, { recursive: true }) const normalizedDescription = description.trim() if (!normalizedDescription) fail('必须填写变更说明') const safeDescription = normalizedDescription.toLowerCase().replaceAll(/[^a-z0-9]+/g, '_').replace(/^_+|_+$/g, '') || 'change' for (let attempt = 0; attempt < 20; attempt += 1) { const migrations = await listMigrations(sqlRoot, region, schema.name) const version = (migrations.at(-1)?.version ?? 0) + 1 const fileName = `V${version}__${safeDescription}.sql` const filePath = path.join(directory, fileName) try { const handle = await open(filePath, 'wx') await handle.writeFile([ `-- Region: ${region}`, `-- Schema: ${schema.name}`, `-- Purpose: ${description.trim()}`, '-- Preconditions:', '-- Postconditions:', '', ].join('\n'), 'utf8') await handle.close() return { region, schema: schema.name, version, fileName, path: path.relative(sqlRoot, filePath).replaceAll(path.sep, '/') } } catch (error) { if (error.code !== 'EEXIST') throw error } } fail(`${region}/${schema.name} 并发生成迁移失败:重试次数已耗尽`) } async function checkProtected(sqlRoot, protectedPath) { if (!protectedPath) return [] const document = JSON.parse(await readFile(protectedPath, 'utf8')) const entries = document.files ?? document.migrations ?? [] if (!Array.isArray(entries)) fail('保护清单的 files/migrations 必须是数组') const errors = [] for (const entry of entries) { if (!['success', 'frozen'].includes(entry.status)) continue if (!entry.path || !/^[a-f0-9]{64}$/.test(entry.sha256 ?? '')) fail('保护清单条目必须包含 path 和 64 位 sha256') const current = await sha256(absoluteFrom(sqlRoot, entry.path)) if (current !== entry.sha256) errors.push(`受保护迁移已修改:${entry.path}`) } return errors } async function checkSqlFiles(sqlRoot, scopes, regions) { const errors = [] const schemasByName = new Map() for (const region of regions) { for (const schema of regionConfig(scopes, region).schemas) schemasByName.set(schema.name, { region, schema }) } for (const [schemaName, { region }] of schemasByName) { const migrations = await listMigrations(sqlRoot, region, schemaName) for (const migration of migrations) { const sql = await readFile(migration.absolutePath, 'utf8') for (const candidate of schemasByName.keys()) { if (candidate !== schemaName && new RegExp(`(?:\\b|\`)${candidate}(?:\\b|\`)`, 'i').test(sql)) { errors.push(`${migration.relativePath} 引用了其他 schema:${candidate}`) } } if (/\b(?:DROP\s+DATABASE|CREATE\s+DATABASE|USE\s+)/i.test(sql)) errors.push(`${migration.relativePath} 不得包含跨库控制语句`) } } return errors } async function check({ sqlRoot = SQL_ROOT, scopes, regions = Object.keys(scopes.regions), protectedPath }) { const errors = await checkSqlFiles(sqlRoot, scopes, regions) errors.push(...await checkProtected(sqlRoot, protectedPath)) if (errors.length) fail(errors.join('\n')) return { ok: true, regions, checkedAt: new Date().toISOString() } } async function runManifest({ sqlRoot = SQL_ROOT, manifestPath, dryRun = false }) { const manifest = JSON.parse(await readFile(manifestPath, 'utf8')) if (manifest.schemaVersion !== 1 || !Array.isArray(manifest.regions)) fail('发布清单格式不受支持') const commands = [] for (const region of manifest.regions) { for (const schema of region.schemas ?? []) { if (schema.target === '0.0.0' && (schema.migrations ?? []).length === 0) continue const location = path.join(sqlRoot, schema.migrationDirectory) await stat(location).catch(() => fail(`迁移目录不存在:${schema.migrationDirectory}`)) commands.push({ region: region.region, schema: schema.schema, target: schema.target, location }) } } if (dryRun) return { mode: 'dry-run', commands } if (process.env.DB_MIGRATION_ALLOW !== '1') fail('实际数据库执行被保护:先设置 DB_MIGRATION_ALLOW=1,并显式提供测试环境凭据') if (!process.env.FLYWAY_BIN) fail('实际数据库执行需要 FLYWAY_BIN,阶段 B 只支持 --dry-run') const results = [] for (const command of commands) { const result = spawnSync(process.env.FLYWAY_BIN, [ '-locations=filesystem:' + command.location, '-schemas=' + command.schema, '-defaultSchema=' + command.schema, '-table=flyway_schema_history', '-baselineOnMigrate=false', '-cleanDisabled=true', '-outOfOrder=false', '-validateOnMigrate=true', '-target=' + command.target, 'migrate', ], { stdio: 'inherit', env: process.env }) if (result.status !== 0) fail(`${command.region}/${command.schema} Flyway 执行失败,退出码 ${result.status}`) results.push(command) } return { mode: 'execute', commands: results } } function usage() { console.log(`用法:\n node deploy/sql/scripts/db-migrations.mjs new --region CN --schema baiyue_auth --description add_login_audit\n node deploy/sql/scripts/db-migrations.mjs check [--protected ]\n node deploy/sql/scripts/db-migrations.mjs plan --services auth,iam --regions CN,OVERSEAS --out \n node deploy/sql/scripts/db-migrations.mjs run --manifest --dry-run`) } async function main(argv = process.argv.slice(2)) { const args = parseArgs(argv) const scopesPath = args.scopes ? absoluteFrom(process.cwd(), args.scopes) : SCOPES_PATH const scopes = await loadScopes(scopesPath) const sqlRoot = path.dirname(scopesPath) if (args.command === 'help') return usage() if (args.command === 'new') { const result = await createMigration({ sqlRoot, scopes, region: String(args.region ?? ''), schemaName: String(args.schema ?? ''), description: String(args.description ?? '') }) console.log(JSON.stringify(result, null, 2)) return } if (args.command === 'check') { const result = await check({ sqlRoot, scopes, protectedPath: args.protected ? absoluteFrom(process.cwd(), args.protected) : undefined }) console.log(JSON.stringify(result, null, 2)) return } if (args.command === 'plan') { const regions = String(args.regions ?? Object.keys(scopes.regions).join(',')).split(',').map((item) => item.trim()).filter(Boolean) const plan = await buildPlan({ sqlRoot, scopes, regions, services: selectedServices(args.services) }) const output = JSON.stringify(plan, null, 2) + '\n' if (args.out) await writeFile(absoluteFrom(process.cwd(), args.out), output, 'utf8') else process.stdout.write(output) return } if (args.command === 'run') { const result = await runManifest({ sqlRoot, manifestPath: absoluteFrom(process.cwd(), String(args.manifest ?? '')), dryRun: Boolean(args['dry-run']) }) console.log(JSON.stringify(result, null, 2)) return } usage() fail(`未知命令:${args.command}`) } export { buildPlan, check, createMigration, listMigrations as discoverMigrations, loadScopes, parseArgs, runManifest } if (process.argv[1] && path.resolve(process.argv[1]) === fileURLToPath(import.meta.url)) { main().catch((error) => { console.error(`ERROR: ${error.message}`) process.exitCode = 1 }) }