朱子纯 / erp-workflow-plugin

#!/usr/bin/env node
// scripts/promote-to-source.mjs —— 绿后把副本上已验证的**新迁移**晋升到源库。
//
// 模型：新 V<n> 先在副本上被 Flyway apply 并被测试；测试绿后，本脚本把这些"副本有、源库还没有"的
// 迁移应用到源库，使源库停在"上一次通过测试的 schema + 累积数据"。源库数据绝不被清、只做 schema 追加。
//
// 做法（确定性、纯 mysql CLI、无需起后端）：
//   ① 取 delta = 副本 flyway_schema_history 里源库没有的版本行（success=1，按 installed_rank 升序）。
//   ② 逐个：把 sql/migrations/<script> 重放到源库 + 移植该迁移的 flyway 历史行（installed_rank 取源库当前
//      MAX+1，checksum/description/type/script 原样搬用——checksum 由副本上真 Flyway 算好，下次校验必过）。
//   ③ 全程持本仓库根 .tmp/erp-source-promote.lock 文件锁。
//
// 串行化：并行 lane 间的真正串行由**编排层**保证——test-gate（跑 test.mjs、内含本脚本）整段被 withStackLock
// 全局起栈互斥包住，任一时刻只有一个 test-gate 在跑，故对共享源库的晋升天然串行。本脚本的文件锁只兜底
// **同一 worktree 内**的重复运行（lane worktree 各有独立 .tmp/，文件锁跨不了 worktree）。
// 幂等：已在源库的版本自动跳过（可重复运行 / 并发下双跑安全）。源库未被 Flyway 管理（无 history 表）→ 跳过并告警。
// 前提：迁移为 sql/migrations/V<n>__*.sql 纯 SQL（无 Flyway 占位符替换）。由 test.mjs 绿后调用。

import { spawnSync } from 'node:child_process'
import { existsSync, readFileSync, mkdirSync, rmSync, statSync } from 'node:fs'
import { dirname, join } from 'node:path'
import { fileURLToPath } from 'node:url'

const SCRIPT_DIR = dirname(fileURLToPath(import.meta.url))
const PROJECT_ROOT = join(SCRIPT_DIR, '..')
const CONFIG_FILE = join(PROJECT_ROOT, 'config-vars.yaml')
const MIGRATIONS_DIR = join(PROJECT_ROOT, 'sql', 'migrations')
const LOCK_DIR = join(PROJECT_ROOT, '.tmp', 'erp-source-promote.lock')
const LOG = '[promote-to-source]'
const LOCK_STALE_MS = 10 * 60 * 1000 // 持锁 >10min 视为崩溃残留，可窃取
const LOCK_WAIT_MS = 3 * 60 * 1000 // 等锁上限

// 极简 YAML 读取（2 层 map + 标量；与 scripts/setup-test-db.mjs 同规则，内联以免运行时依赖）。
function parseScalar(raw) {
  let s = String(raw).trim()
  if (s === '' || s[0] === '#') return ''
  const q = s[0]
  if (q === '"' || q === "'") {
    const end = s.indexOf(q, 1)
    if (end !== -1) return s.slice(1, end)
  }
  const hash = s.indexOf(' #')
  if (hash !== -1) s = s.slice(0, hash).trim()
  return s
}
function parseYamlConfig(text) {
  const root = {}
  let section = null
  for (const rawLine of text.split('\n')) {
    const line = rawLine.replace(/\r$/, '')
    const trimmed = line.trim()
    if (trimmed === '' || trimmed[0] === '#') continue
    const colon = line.indexOf(':')
    if (colon === -1) continue
    const key = line.slice(0, colon).trim()
    if (key === '') continue
    const indent = line.length - line.replace(/^\s+/, '').length
    const value = parseScalar(line.slice(colon + 1))
    if (indent === 0) {
      if (value === '') { section = {}; root[key] = section }
      else { root[key] = value; section = null }
    } else if (section) { section[key] = value }
    else { root[key] = value }
  }
  return root
}

function quoteSqlString(value) {
  return "'" + String(value).replaceAll('\\', '\\\\').replaceAll("'", "''") + "'"
}
function quoteMySqlIdent(value) {
  return '`' + String(value).replaceAll('`', '``') + '`'
}
const sleep = (ms) => new Promise((r) => setTimeout(r, ms))

if (!existsSync(CONFIG_FILE)) {
  console.error(`${LOG} config-vars.yaml 不存在（${CONFIG_FILE}）`)
  process.exit(1)
}

const db = parseYamlConfig(readFileSync(CONFIG_FILE, 'utf8')).database || {}

function readLaneDbMarker() {
  const markerFile = join(PROJECT_ROOT, '.tmp', 'erp-lane-db')
  if (!existsSync(markerFile)) return ''
  return readFileSync(markerFile, 'utf8').split('\n')[0].trim()
}

const DB_HOST = db.host ?? ''
const DB_PORT = db.port ?? '3306'
const DB_USER = db.user ?? ''
const DB_PASSWORD = db.password ?? ''
const SOURCE_SCHEMA = db.schema ?? ''

function rejectPlaceholder(key, value) {
  if (typeof value === 'string' && value.includes('【人工填写')) {
    console.error(`${LOG} database.${key} 仍是占位，请先在 config-vars.yaml 填真实值`)
    process.exit(1)
  }
}
for (const [key, value] of [['host', DB_HOST], ['port', DB_PORT], ['user', DB_USER], ['password', DB_PASSWORD], ['schema', SOURCE_SCHEMA]]) {
  rejectPlaceholder(key, value)
}
if (!/^\d+$/.test(DB_PORT) || Number(DB_PORT) <= 0 || Number(DB_PORT) > 65535) {
  console.error(`${LOG} database.port 非法: ${DB_PORT}`)
  process.exit(1)
}
if (String(SOURCE_SCHEMA).trim() === '') {
  console.error(`${LOG} database.schema（源库名）未填`)
  process.exit(1)
}

const COPY_SCHEMA = process.env.ERP_TEST_DB_SCHEMA || readLaneDbMarker() || `${SOURCE_SCHEMA}__test`
if (COPY_SCHEMA === SOURCE_SCHEMA) {
  console.error(`${LOG} 拒绝：副本名与源库名相同（${SOURCE_SCHEMA}）——无 delta 可取。`)
  process.exit(1)
}

const CONN_ARGS = [`--host=${DB_HOST}`, `--port=${DB_PORT}`, `--user=${DB_USER}`, `--password=${DB_PASSWORD}`]

// 查询型调用：mysql -N -B（tab 分隔、无表头、NULL 打印为字面 "NULL"）。
function mysqlQuery(sql) {
  return spawnSync('mysql', [...CONN_ARGS, '-N', '-B', '-e', sql], { encoding: 'utf8' })
}
// 应用型调用：SQL 喂 stdin，目标库作位置参数（--comments 防剥注释）。
function mysqlApply(targetSchema, sqlText) {
  return spawnSync('mysql', [...CONN_ARGS, '--comments', targetSchema], { input: sqlText, encoding: 'utf8' })
}
function fatal(res, label) {
  if (res.error) {
    console.error(`${LOG} FATAL: 无法执行 mysql（请确认其在 PATH 中）: ${res.error.message}`)
    releaseLock()
    process.exit(1)
  }
  if (res.status !== 0) {
    if (res.stderr) process.stderr.write(res.stderr)
    console.error(`${LOG} FAIL (${label}): mysql exit=${res.status}`)
    releaseLock()
    process.exit(res.status === null ? 1 : res.status)
  }
}

// ── 文件锁：兜底同一 worktree 内重复运行（跨 lane 串行由编排层 withStackLock 保证，见文件头） ──
let lockHeld = false
function releaseLock() {
  if (lockHeld) {
    try { rmSync(LOCK_DIR, { recursive: true, force: true }) } catch { /* ignore */ }
    lockHeld = false
  }
}
async function acquireLock() {
  mkdirSync(join(PROJECT_ROOT, '.tmp'), { recursive: true })
  const deadline = Date.now() + LOCK_WAIT_MS
  for (;;) {
    try {
      mkdirSync(LOCK_DIR) // 原子：已存在即抛 EEXIST
      lockHeld = true
      return
    } catch (e) {
      if (e?.code !== 'EEXIST') throw e
      // 陈旧锁（持有者疑似崩溃）→ 窃取
      try {
        if (Date.now() - statSync(LOCK_DIR).mtimeMs > LOCK_STALE_MS) {
          rmSync(LOCK_DIR, { recursive: true, force: true })
          continue
        }
      } catch { /* 锁刚被别人释放，重试 */ }
      if (Date.now() > deadline) {
        console.error(`${LOG} 等源库晋升锁超时（${LOCK_WAIT_MS}ms）——疑有其它晋升卡住，请检查 ${LOCK_DIR}`)
        process.exit(1)
      }
      await sleep(300)
    }
  }
}

// ── 主流程 ──
await acquireLock()
try {
  // 源库未被 Flyway 管理 → 无从判定 delta，跳过（不因晋升让已绿的测试闸失败）。
  const srcHasHistory = mysqlQuery(
    `SELECT COUNT(*) FROM information_schema.tables WHERE table_schema = ${quoteSqlString(SOURCE_SCHEMA)} AND table_name = 'flyway_schema_history'`,
  )
  fatal(srcHasHistory, 'check-source-history')
  if (srcHasHistory.stdout.trim() !== '1') {
    console.log(`${LOG} 源库 ${SOURCE_SCHEMA} 未被 Flyway 管理（无 flyway_schema_history），跳过晋升。`)
    releaseLock()
    process.exit(0)
  }

  // 副本无 history（本轮未起后端 / 无迁移）→ 无 delta。
  const copyHasHistory = mysqlQuery(
    `SELECT COUNT(*) FROM information_schema.tables WHERE table_schema = ${quoteSqlString(COPY_SCHEMA)} AND table_name = 'flyway_schema_history'`,
  )
  fatal(copyHasHistory, 'check-copy-history')
  if (copyHasHistory.stdout.trim() !== '1') {
    console.log(`${LOG} 副本 ${COPY_SCHEMA} 无 flyway_schema_history，无迁移可晋升。`)
    releaseLock()
    process.exit(0)
  }

  // delta = 副本有、源库没有的版本行（success=1，按 apply 顺序）。
  const deltaRes = mysqlQuery(
    `SELECT c.version, c.description, c.type, c.script, c.checksum, c.execution_time ` +
      `FROM ${quoteMySqlIdent(COPY_SCHEMA)}.flyway_schema_history c ` +
      `WHERE c.version IS NOT NULL AND c.success = 1 ` +
      `AND NOT EXISTS (SELECT 1 FROM ${quoteMySqlIdent(SOURCE_SCHEMA)}.flyway_schema_history s WHERE s.version = c.version) ` +
      `ORDER BY c.installed_rank`,
  )
  fatal(deltaRes, 'query-delta')

  const rows = deltaRes.stdout.split('\n').map((l) => l.trimEnd()).filter((l) => l !== '')
  if (rows.length === 0) {
    console.log(`${LOG} 源库 ${SOURCE_SCHEMA} 已含副本全部迁移，无新迁移可晋升。`)
    releaseLock()
    process.exit(0)
  }

  console.log(`${LOG} 待晋升到源库 ${SOURCE_SCHEMA} 的新迁移 ${rows.length} 个（副本 ${COPY_SCHEMA} → 源库）`)

  let promoted = 0
  for (const line of rows) {
    const [version, description, type, script, checksum, execTime] = line.split('\t')
    const migFile = join(MIGRATIONS_DIR, script)
    if (!existsSync(migFile)) {
      console.error(`${LOG} 迁移文件不存在，无法晋升: sql/migrations/${script}（version=${version}）`)
      releaseLock()
      process.exit(1)
    }

    // 锁内二次核对（防并发已被别的进程晋升）。
    const already = mysqlQuery(
      `SELECT COUNT(*) FROM ${quoteMySqlIdent(SOURCE_SCHEMA)}.flyway_schema_history WHERE version = ${quoteSqlString(version)}`,
    )
    fatal(already, 'recheck-version')
    if (already.stdout.trim() !== '0') {
      console.log(`${LOG} 跳过（源库已有）: V${version} ${script}`)
      continue
    }

    // checksum / execution_time 可能为 NULL（-B 打印字面 "NULL"）。
    const checksumSql = (checksum === undefined || checksum === 'NULL' || checksum === '') ? 'NULL' : String(Number(checksum))
    const execSql = (execTime === undefined || execTime === 'NULL' || execTime === '') ? '0' : String(Number(execTime))

    // 一次调用内：重放迁移 DDL + 移植历史行（mysql 批处理遇错即停，DDL 失败则历史行不落）。
    const ddl = readFileSync(migFile, 'utf8').trimEnd()
    const sep = /;$/.test(ddl) ? '\n' : ';\n'
    const insert =
      `INSERT INTO flyway_schema_history ` +
      `(installed_rank, version, description, type, script, checksum, installed_by, installed_on, execution_time, success) ` +
      `SELECT COALESCE(MAX(installed_rank), 0) + 1, ${quoteSqlString(version)}, ${quoteSqlString(description ?? '')}, ` +
      `${quoteSqlString(type || 'SQL')}, ${quoteSqlString(script)}, ${checksumSql}, 'promote', NOW(), ${execSql}, 1 ` +
      `FROM flyway_schema_history;`

    console.log(`${LOG} 晋升: V${version} ${script}`)
    fatal(mysqlApply(SOURCE_SCHEMA, `${ddl}${sep}${insert}`), `promote V${version}`)
    promoted += 1
  }

  console.log(`${LOG} done — 晋升 ${promoted} 个迁移到源库 ${SOURCE_SCHEMA}`)
} finally {
  releaseLock()
}