朱子纯 / erp-workflow-plugin

// lib/bootstrap-source-flyway.mjs —— 把**源库**建立成 Flyway 托管态（Plan 期 db-init 调用）。
//
// 复制测试模型需要一个已存在、且被 Flyway 托管（有 flyway_schema_history）的源库才能复制。但 Plan 阶段
// 没有后端 app 可起、跑不了真 Flyway。本脚本用纯 mysql2 完成等效初始化：
//   ① CREATE DATABASE IF NOT EXISTS <source>（源库）；
//   ② 若已有 flyway_schema_history → 幂等跳过（已托管，绝不重灌 / 绝不 DROP）；
//   ③ 否则：建 flyway_schema_history（Flyway 10 的表结构）→ 按版本序 apply sql/migrations/V*.sql →
//      为每个迁移写一条历史行，**checksum 用与 Flyway 完全一致的 CRC32 复算**，使首次起后端时 Flyway
//      校验通过（不会报 checksum mismatch）、且只 apply 之后新增的迁移。
//
// apply V1 本身就是 db-init 的「V1 DDL 烟测」：V1 有 DDL 错会在此 apply 失败暴露。
// 凭据从 config-vars.yaml 的 database: 段读取。mysql2 从目标项目解析（见 apply-ddl.resolveMysql2Path）。

import { parseYamlConfig } from './yaml-config.mjs'
import { resolveMysql2Path, resolveDbConfig, MysqlUnavailableError } from './apply-ddl.mjs'
import { readFileSync, readdirSync, existsSync } from 'node:fs'
import { crc32 } from 'node:zlib'
import { pathToFileURL } from 'node:url'
import { dirname, resolve as resolvePath, join } from 'node:path'

/**
 * Flyway SQL-migration checksum：与 Flyway（org.flywaydb）对 SQL 迁移的算法一致——
 * 按行读取（BufferedReader.readLine 语义：\n / \r / \r\n 均为行终止符且被剥离；首行剥 UTF-8 BOM），
 * 对每行的 UTF-8 字节累进 CRC32，最终以**有符号 int32** 返回（Flyway 的 checksum 列是 signed int）。
 * 因按行读取，CRLF 与 LF 版本的同一内容 checksum 相同（行终止符归一化）。
 * @param {string} content 迁移文件文本
 * @returns {number} 有符号 32 位校验和
 */
export function flywayChecksum(content) {
  let text = content
  if (text.charCodeAt(0) === 0xFEFF) text = text.slice(1) // 剥首行 UTF-8 BOM
  const lines = text.split(/\r\n|\r|\n/)
  // 文件以行终止符结尾时 split 会多出一个尾部空串——BufferedReader 不会返回这个"最后的空行"，去掉。
  if (lines.length > 0 && lines[lines.length - 1] === '') lines.pop()
  let crc = 0
  for (const line of lines) crc = crc32(Buffer.from(line, 'utf8'), crc)
  return crc | 0 // 转有符号 int32
}

/** 解析 `V<version>__<description>.sql` → { version, description }（下划线在版本里当分隔符、在描述里当空格）。 */
export function parseMigrationName(filename) {
  const m = filename.match(/^V(\d+(?:[._]\d+)*)__(.+)\.sql$/i)
  if (!m) return null
  return { version: m[1].replace(/_/g, '.'), description: m[2].replace(/_/g, ' ') }
}

/** 版本号数值序比较（逐段比较，与 Flyway 一致；段数不同时短的在前）。 */
function compareVersions(a, b) {
  const pa = a.split('.').map(Number)
  const pb = b.split('.').map(Number)
  for (let i = 0; i < Math.max(pa.length, pb.length); i++) {
    const d = (pa[i] ?? 0) - (pb[i] ?? 0)
    if (d !== 0) return d
  }
  return 0
}

// Flyway 10 (MySQL) 的 flyway_schema_history 建表 DDL —— 结构与 Flyway 自建一致，使其在首次起后端时识别复用。
const FLYWAY_HISTORY_DDL =
  'CREATE TABLE `flyway_schema_history` (' +
  '`installed_rank` INT NOT NULL,' +
  '`version` VARCHAR(50),' +
  '`description` VARCHAR(200) NOT NULL,' +
  '`type` VARCHAR(20) NOT NULL,' +
  '`script` VARCHAR(1000) NOT NULL,' +
  '`checksum` INT,' +
  '`installed_by` VARCHAR(100) NOT NULL,' +
  '`installed_on` TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,' +
  '`execution_time` INT NOT NULL,' +
  '`success` BOOL NOT NULL,' +
  'CONSTRAINT `flyway_schema_history_pk` PRIMARY KEY (`installed_rank`)' +
  ') ENGINE=InnoDB'

function quoteIdent(value) {
  return '`' + String(value).replaceAll('`', '``') + '`'
}

/**
 * 把源库建立成 Flyway 托管态。幂等：已有 flyway_schema_history 即跳过。
 * @param {{configPath: string, migrationsDir: string}} opts
 * @returns {Promise<{bootstrapped: boolean, reason?: string, applied?: string[]}>}
 */
export async function bootstrapSourceFlyway({ configPath, migrationsDir }) {
  const config = parseYamlConfig(readFileSync(configPath, 'utf8'))
  const { host, port, user, password, database } = resolveDbConfig(config, configPath)

  let mysql
  try {
    const resolved = resolveMysql2Path(dirname(resolvePath(configPath)))
    ;({ default: mysql } = await import(pathToFileURL(resolved).href))
  } catch {
    throw new MysqlUnavailableError()
  }

  // ① 无 database 连接，建源库（存在则不动）。
  const admin = await mysql.createConnection({ host, port, user, password })
  try {
    await admin.query(`CREATE DATABASE IF NOT EXISTS ${quoteIdent(database)} CHARACTER SET utf8 COLLATE utf8_general_ci`)
  } finally {
    await admin.end()
  }

  const conn = await mysql.createConnection({ host, port, user, password, database, multipleStatements: true })
  try {
    // ② 已被 Flyway 管理 → 幂等跳过（绝不重灌、绝不 DROP 源库数据）。
    const [rows] = await conn.query(
      "SELECT COUNT(*) AS n FROM information_schema.tables WHERE table_schema = ? AND table_name = 'flyway_schema_history'",
      [database],
    )
    if (Number(rows[0].n) > 0) {
      return { bootstrapped: false, reason: 'already-managed' }
    }

    // 源库有业务表但无 flyway 历史 = 非 Flyway 托管的既有库——不安全 bootstrap（apply V1 会撞已存在的表）。
    const [tblRows] = await conn.query(
      'SELECT COUNT(*) AS n FROM information_schema.tables WHERE table_schema = ?',
      [database],
    )
    if (Number(tblRows[0].n) > 0) {
      throw new Error(
        `源库 ${database} 已有表但无 flyway_schema_history —— 疑为非 Flyway 托管的既有库。` +
          `拒绝自动 bootstrap（apply V1 会撞已存在的表）。请人工确认后清空该库、或改用已托管的库。`,
      )
    }

    // ③ 建历史表 + 按版本序 apply 迁移 + 移植正确 checksum 的历史行。
    await conn.query(FLYWAY_HISTORY_DDL)

    const files = existsSync(migrationsDir)
      ? readdirSync(migrationsDir).filter((f) => /^V.*\.sql$/i.test(f))
      : []
    const parsed = files
      .map((f) => ({ file: f, meta: parseMigrationName(f) }))
      .filter((x) => x.meta)
      .sort((a, b) => compareVersions(a.meta.version, b.meta.version))

    const applied = []
    let rank = 0
    for (const { file, meta } of parsed) {
      const sql = readFileSync(join(migrationsDir, file), 'utf8')
      await conn.query(sql) // apply DDL（同时是 V<n> 的 DDL 烟测）
      rank += 1
      await conn.query(
        'INSERT INTO flyway_schema_history ' +
          '(installed_rank, version, description, type, script, checksum, installed_by, installed_on, execution_time, success) ' +
          'VALUES (?, ?, ?, ?, ?, ?, ?, NOW(), ?, 1)',
        [rank, meta.version, meta.description, 'SQL', file, flywayChecksum(sql), 'bootstrap', 0],
      )
      applied.push(meta.version)
    }
    return { bootstrapped: true, applied }
  } finally {
    await conn.end()
  }
}

// CLI entry guard（与 apply-ddl 同款 pathToFileURL 规范化）。
if (process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href) {
  const [configPath, migrationsDir] = process.argv.slice(2)
  if (!configPath || !migrationsDir) {
    console.error('usage: node lib/bootstrap-source-flyway.mjs <configPath> <migrationsDir>')
    process.exit(2)
  }
  if (!existsSync(configPath)) { console.error(`bootstrap-source-flyway: 配置文件不存在: ${configPath}`); process.exit(2) }
  try {
    const r = await bootstrapSourceFlyway({ configPath, migrationsDir })
    if (r.bootstrapped) console.log(`bootstrap-source-flyway: 源库已 Flyway 托管化，apply 迁移 ${r.applied.length} 个: ${r.applied.map((v) => 'V' + v).join(', ') || '(无)'}`)
    else console.log(`bootstrap-source-flyway: 跳过（${r.reason === 'already-managed' ? '源库已有 flyway_schema_history，已托管' : r.reason}）`)
  } catch (e) {
    if (e instanceof MysqlUnavailableError) {
      console.error('bootstrap-source-flyway: mysql2 not found. Please run `npm i mysql2` in the target project.')
      process.exit(1)
    }
    console.error(`bootstrap-source-flyway: failed — ${e?.message || e}`)
    process.exit(1)
  }
}