318 lines
11 KiB
Plaintext
318 lines
11 KiB
Plaintext
/*
|
||
* Copyright (c) 2025 SimcuTeam. All rights reserved.
|
||
* 迁移执行器：历史表 + 按 migrationId 排序应用未执行的迁移（事务内）。
|
||
*
|
||
* 历史表 simcu_orm_migrations (id VARCHAR(64) PK, name VARCHAR(256), applied_at TIMESTAMP)，
|
||
* 应用顺序 = migrationId 字典序（建议用 "yyyyMMddHHmmss_名称" 命名）。
|
||
*/
|
||
|
||
package simcu::orm.migrations
|
||
|
||
import std.collection.*
|
||
import std.database.sql.*
|
||
import std.time.*
|
||
import simcu::orm.sql.*
|
||
|
||
/**
|
||
* 迁移执行器。
|
||
*/
|
||
public class Migrator {
|
||
private let _datasource: Datasource
|
||
private let _dialect: ISqlDialect
|
||
private let _historyTable = "simcu_orm_migrations"
|
||
|
||
public init(datasource: Datasource, dialect: ISqlDialect) {
|
||
this._datasource = datasource
|
||
this._dialect = dialect
|
||
}
|
||
|
||
public init(datasource: Datasource) {
|
||
this(datasource, OpenGaussDialect())
|
||
}
|
||
|
||
/// 应用所有未执行的迁移，返回本次应用的数量
|
||
public func migrate(migrations: ArrayList<Migration>): Int64 {
|
||
ensureHistoryTable()
|
||
let applied = loadApplied()
|
||
let sorted = sortMigrations(migrations)
|
||
var count: Int64 = 0
|
||
let conn = _datasource.connect()
|
||
let tx = conn.createTransaction()
|
||
tx.begin()
|
||
try {
|
||
for (m in sorted) {
|
||
if (applied.contains(m.migrationId)) {
|
||
continue
|
||
}
|
||
let builder = MigrationBuilder()
|
||
m.up(builder)
|
||
let factory = DdlFactory()
|
||
for (op in builder.operations) {
|
||
let sqls = ArrayList<String>()
|
||
let isColumnOp = match (op.kind) {
|
||
case MigrationOperationKind.AddColumn |
|
||
MigrationOperationKind.AlterColumn => true
|
||
case _ => false
|
||
}
|
||
if (isColumnOp) {
|
||
// 同名列类型变更时先删后加（见 DdlFactory.resolveColumnSqls）
|
||
let col = op.column.getOrThrow()
|
||
let resolved = factory.resolveColumnSqls(op, _dialect,
|
||
columnTypeInDb(conn, op.tableName, col.name))
|
||
for (s in resolved) {
|
||
sqls.add(s)
|
||
}
|
||
} else {
|
||
sqls.add(factory.toSql(op, _dialect))
|
||
}
|
||
for (sql in sqls) {
|
||
let stmt = conn.prepareStatement(sql)
|
||
try {
|
||
stmt.update()
|
||
} finally {
|
||
stmt.close()
|
||
}
|
||
}
|
||
}
|
||
recordApplied(conn, m)
|
||
count += 1
|
||
}
|
||
tx.commit()
|
||
count
|
||
} catch (e: Exception) {
|
||
try {
|
||
tx.rollback()
|
||
} catch (_) {
|
||
()
|
||
}
|
||
throw e
|
||
} finally {
|
||
conn.close()
|
||
}
|
||
}
|
||
|
||
/// 生成"该执行但尚未执行"的迁移集合（不执行，供用户预览/校验）
|
||
public func pending(migrations: ArrayList<Migration>): ArrayList<Migration> {
|
||
let applied = loadApplied()
|
||
let sorted = sortMigrations(migrations)
|
||
let result = ArrayList<Migration>()
|
||
for (m in sorted) {
|
||
if (!applied.contains(m.migrationId)) {
|
||
result.add(m)
|
||
}
|
||
}
|
||
result
|
||
}
|
||
|
||
/// 回退最近一个已应用的迁移（执行 down + 删除历史记录），返回回退的数量
|
||
public func revert(migrations: ArrayList<Migration>): Int64 {
|
||
revert(migrations, None)
|
||
}
|
||
|
||
/// 回退到目标迁移（不含目标）：执行目标之后全部已应用迁移的 down。
|
||
/// target 匹配规则：精确（migrationId / description）或唯一前缀（migrationId）。
|
||
/// 返回回退的数量；无已应用迁移或已处于目标时返回 0。
|
||
public func revert(migrations: ArrayList<Migration>, target: ?String): Int64 {
|
||
ensureHistoryTable()
|
||
let applied = loadApplied()
|
||
if (applied.size == 0) {
|
||
return 0
|
||
}
|
||
let sorted = sortMigrations(migrations)
|
||
// 已应用迁移（按 id 升序）
|
||
let appliedList = ArrayList<Migration>()
|
||
for (m in sorted) {
|
||
if (applied.contains(m.migrationId)) {
|
||
appliedList.add(m)
|
||
}
|
||
}
|
||
if (appliedList.size == 0) {
|
||
return 0
|
||
}
|
||
// 目标下标：默认回退最新一个；指定目标时回退到目标之后（不含目标）
|
||
var targetIdx = appliedList.size - 1
|
||
if (let Some(t) <- target) {
|
||
targetIdx = resolveTarget(appliedList, t) + 1
|
||
if (targetIdx >= appliedList.size) {
|
||
return 0
|
||
}
|
||
}
|
||
var count: Int64 = 0
|
||
let conn = _datasource.connect()
|
||
let tx = conn.createTransaction()
|
||
tx.begin()
|
||
try {
|
||
for (i in targetIdx..appliedList.size) {
|
||
let m = appliedList[appliedList.size - 1 - (i - targetIdx)]
|
||
let builder = MigrationBuilder()
|
||
m.down(builder)
|
||
let factory = DdlFactory()
|
||
for (op in builder.operations) {
|
||
let stmt = conn.prepareStatement(factory.toSql(op, _dialect))
|
||
try {
|
||
stmt.update()
|
||
} finally {
|
||
stmt.close()
|
||
}
|
||
}
|
||
removeApplied(conn, m)
|
||
count += 1
|
||
}
|
||
tx.commit()
|
||
count
|
||
} catch (e: Exception) {
|
||
try {
|
||
tx.rollback()
|
||
} catch (_) {
|
||
()
|
||
}
|
||
throw e
|
||
} finally {
|
||
conn.close()
|
||
}
|
||
}
|
||
|
||
/// 查询库中某列的数据类型（information_schema.data_type）；列不存在返回 None。
|
||
/// 执行 addColumn/alterColumn 前调用，用于识别"同名列修改数据结构"场景。
|
||
private func columnTypeInDb(conn: Connection, table: String, column: String): ?String {
|
||
let stmt = conn.prepareStatement(
|
||
"SELECT data_type FROM information_schema.columns WHERE table_schema = current_schema() " +
|
||
"AND table_name = ? AND column_name = ?")
|
||
try {
|
||
stmt.set<String>(0, table)
|
||
stmt.set<String>(1, column)
|
||
let rs = stmt.query()
|
||
if (rs.next()) {
|
||
rs.getOrNull<String>(0)
|
||
} else {
|
||
None
|
||
}
|
||
} finally {
|
||
stmt.close()
|
||
}
|
||
}
|
||
|
||
private func resolveTarget(appliedList: ArrayList<Migration>, target: String): Int64 {
|
||
var exact: Int64 = -1
|
||
let prefixHits = ArrayList<Int64>()
|
||
for (i in 0..appliedList.size) {
|
||
let m = appliedList[i]
|
||
if (m.migrationId == target || m.description == target) {
|
||
exact = i
|
||
} else if (m.migrationId.startsWith(target)) {
|
||
prefixHits.add(i)
|
||
}
|
||
}
|
||
if (exact >= 0) {
|
||
return exact
|
||
}
|
||
if (prefixHits.size == 1) {
|
||
return prefixHits[0]
|
||
}
|
||
if (prefixHits.size == 0) {
|
||
throw Exception("simorm: 未找到迁移 '${target}'（已应用的迁移中无匹配）")
|
||
}
|
||
throw Exception("simorm: 迁移目标 '${target}' 不唯一，请使用完整 id")
|
||
}
|
||
|
||
private func removeApplied(conn: Connection, m: Migration): Unit {
|
||
let stmt = conn.prepareStatement(
|
||
"DELETE FROM ${_dialect.quoteName(_historyTable)} WHERE ${_dialect.quoteName("id")} = ?")
|
||
try {
|
||
stmt.set<String>(0, m.migrationId)
|
||
stmt.update()
|
||
} finally {
|
||
stmt.close()
|
||
}
|
||
}
|
||
|
||
/// 已应用的迁移 id 列表（字典序，供 CLI list 使用）
|
||
public func appliedMigrationIds(): ArrayList<String> {
|
||
let ids = loadApplied()
|
||
let result = ArrayList<String>()
|
||
for (id in ids) {
|
||
result.add(id)
|
||
}
|
||
// 插入排序（字典序）
|
||
for (i in 1..result.size) {
|
||
var j = i
|
||
while (j > 0 && result[j] < result[j - 1]) {
|
||
let t = result[j]
|
||
result[j] = result[j - 1]
|
||
result[j - 1] = t
|
||
j -= 1
|
||
}
|
||
}
|
||
result
|
||
}
|
||
|
||
private func ensureHistoryTable(): Unit {
|
||
let conn = _datasource.connect()
|
||
try {
|
||
let stmt = conn.prepareStatement(
|
||
"CREATE TABLE IF NOT EXISTS ${_dialect.quoteName(_historyTable)} (\n" +
|
||
" ${_dialect.quoteName("id")} VARCHAR(64) NOT NULL PRIMARY KEY,\n" +
|
||
" ${_dialect.quoteName("name")} VARCHAR(256) NOT NULL,\n" +
|
||
" ${_dialect.quoteName("applied_at")} TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP\n)")
|
||
try {
|
||
stmt.update()
|
||
} finally {
|
||
stmt.close()
|
||
}
|
||
} finally {
|
||
conn.close()
|
||
}
|
||
}
|
||
|
||
private func loadApplied(): HashSet<String> {
|
||
let conn = _datasource.connect()
|
||
try {
|
||
let stmt = conn.prepareStatement(
|
||
"SELECT ${_dialect.quoteName("id")} FROM ${_dialect.quoteName(_historyTable)}")
|
||
try {
|
||
let rs = stmt.query()
|
||
let ids = HashSet<String>()
|
||
while (rs.next()) {
|
||
ids.add(rs.getOrNull<String>(0).getOrThrow())
|
||
}
|
||
ids
|
||
} finally {
|
||
stmt.close()
|
||
}
|
||
} finally {
|
||
conn.close()
|
||
}
|
||
}
|
||
|
||
private func recordApplied(conn: Connection, m: Migration): Unit {
|
||
let stmt = conn.prepareStatement(
|
||
"INSERT INTO ${_dialect.quoteName(_historyTable)} (${_dialect.quoteName("id")}, ${_dialect.quoteName("name")}) VALUES (?, ?)")
|
||
try {
|
||
stmt.set<String>(0, m.migrationId)
|
||
stmt.set<String>(1, if (m.description.isEmpty()) { m.migrationId } else { m.description })
|
||
stmt.update()
|
||
} finally {
|
||
stmt.close()
|
||
}
|
||
}
|
||
|
||
private static func sortMigrations(migrations: ArrayList<Migration>): ArrayList<Migration> {
|
||
let arr = migrations.toArray()
|
||
// 插入排序（migrationId 字典序）
|
||
for (i in 1..arr.size) {
|
||
var j = i
|
||
while (j > 0 && arr[j].migrationId < arr[j - 1].migrationId) {
|
||
let t = arr[j]
|
||
arr[j] = arr[j - 1]
|
||
arr[j - 1] = t
|
||
j -= 1
|
||
}
|
||
}
|
||
let result = ArrayList<Migration>()
|
||
for (m in arr) {
|
||
result.add(m)
|
||
}
|
||
result
|
||
}
|
||
}
|