Files
baiyue-firstleg/deploy/sql/scripts/db-migrations.mjs
T
2026-09-19 17:30:20 +08:00

323 lines
14 KiB
JavaScript
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/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<integer>__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 <manifest>]\n node deploy/sql/scripts/db-migrations.mjs plan --services auth,iam --regions CN,OVERSEAS --out <file>\n node deploy/sql/scripts/db-migrations.mjs run --manifest <file> --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
})
}