优化

deploy/sql/scripts/db-migrations.mjs

@@ -0,0 +1,322 @@
+#!/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
+  })
+}

deploy/sql/scripts/db-migrations.test.mjs

@@ -0,0 +1,77 @@
+import assert from 'node:assert/strict'
+import { mkdtemp, mkdir, readFile, rm, writeFile } from 'node:fs/promises'
+import os from 'node:os'
+import path from 'node:path'
+import test from 'node:test'
+import {
+  buildPlan,
+  check,
+  createMigration,
+  discoverMigrations,
+  loadScopes,
+  runManifest,
+} from './db-migrations.mjs'
+
+async function fixture() {
+  const root = await mkdtemp(path.join(os.tmpdir(), 'db-migrations-'))
+  const sqlRoot = path.join(root, 'sql')
+  await mkdir(path.join(sqlRoot, 'migrations', 'cn', 'baiyue_auth'), { recursive: true })
+  await writeFile(path.join(sqlRoot, 'scopes.json'), JSON.stringify({
+    schemaVersion: 1,
+    regions: {
+      CN: {
+        baseline: 'baseline/cn.sql',
+        schemas: [
+          { name: 'baiyue_auth', services: ['auth'], order: 10 },
+          { name: 'baiyue_order', services: ['order'], order: 20 },
+        ],
+      },
+    },
+  }), 'utf8')
+  return { root, sqlRoot, scopesPath: path.join(sqlRoot, 'scopes.json') }
+}
+
+test('new 自动分配 schema 内下一个版本，并且可重复执行检查', async (t) => {
+  const context = await fixture()
+  t.after(() => rm(context.root, { recursive: true, force: true }))
+  const scopes = await loadScopes(context.scopesPath)
+  const first = await createMigration({ sqlRoot: context.sqlRoot, scopes, region: 'CN', schemaName: 'baiyue_auth', description: 'Add login audit' })
+  assert.equal(first.fileName, 'V1__add_login_audit.sql')
+  const second = await createMigration({ sqlRoot: context.sqlRoot, scopes, region: 'CN', schemaName: 'baiyue_auth', description: 'add_session_index' })
+  assert.equal(second.fileName, 'V2__add_session_index.sql')
+  assert.match(await readFile(path.join(context.sqlRoot, first.path), 'utf8'), /Schema: baiyue_auth/)
+  assert.equal((await discoverMigrations(context.sqlRoot, 'CN', 'baiyue_auth')).length, 2)
+})
+
+test('plan 只包含选中服务，不把同提交其他 schema 的 SQL 带入制品', async (t) => {
+  const context = await fixture()
+  t.after(() => rm(context.root, { recursive: true, force: true }))
+  const scopes = await loadScopes(context.scopesPath)
+  const created = await createMigration({ sqlRoot: context.sqlRoot, scopes, region: 'CN', schemaName: 'baiyue_auth', description: 'add_login_audit' })
+  const plan = await buildPlan({ sqlRoot: context.sqlRoot, scopes, regions: ['CN'], services: ['auth'], commit: 'abc1234' })
+  assert.equal(plan.sourceCommit, 'abc1234')
+  assert.deepEqual(plan.regions[0].schemas.map((schema) => schema.schema), ['baiyue_auth'])
+  assert.equal(plan.regions[0].schemas[0].target, '1')
+  assert.equal(plan.regions[0].schemas[0].migrations[0].path, created.path)
+})
+
+test('check 拒绝跨 schema 引用和跨库控制语句', async (t) => {
+  const context = await fixture()
+  t.after(() => rm(context.root, { recursive: true, force: true }))
+  const scopes = await loadScopes(context.scopesPath)
+  await writeFile(path.join(context.sqlRoot, 'migrations', 'cn', 'baiyue_auth', 'V1__bad.sql'), 'ALTER TABLE baiyue_order.t ADD c INT;\nUSE baiyue_order;\n', 'utf8')
+  await assert.rejects(() => check({ sqlRoot: context.sqlRoot, scopes, regions: ['CN'] }), /其他 schema|跨库控制语句/)
+})
+
+test('run 默认只能 dry-run，不能无意连接数据库', async (t) => {
+  const context = await fixture()
+  t.after(() => rm(context.root, { recursive: true, force: true }))
+  const scopes = await loadScopes(context.scopesPath)
+  await createMigration({ sqlRoot: context.sqlRoot, scopes, region: 'CN', schemaName: 'baiyue_auth', description: 'add_login_audit' })
+  const plan = await buildPlan({ sqlRoot: context.sqlRoot, scopes, regions: ['CN'], services: ['auth'] })
+  const manifestPath = path.join(context.root, 'release-manifest.json')
+  await writeFile(manifestPath, JSON.stringify(plan), 'utf8')
+  const dryRun = await runManifest({ sqlRoot: context.sqlRoot, manifestPath, dryRun: true })
+  assert.equal(dryRun.mode, 'dry-run')
+  await assert.rejects(() => runManifest({ sqlRoot: context.sqlRoot, manifestPath }), /DB_MIGRATION_ALLOW/)
+})