package aceboot::orm.common
import std.collection.*
/**
* 给已有的 QueryRunner(manager.cj,负责事务边界)扩展「迁移用的 DDL/DML + schema 内省/同步」能力。
* 迁移的 up/down 通过 QueryRunner 既能跑裸 SQL(参数化),也能用建表/补列/对比的高层助手——与具体数据库解耦。
* 经 connection()(底层 Driver)与 dialect 委托执行,不引入新可变状态。
*/
extend QueryRunner {
func dialect(): Dialect {
connection().dialect()
}
/** 执行写语句(DDL/DML),返回受影响行数(非 DML/未知返回 -1)。 */
public func exec(sql: String): Int64 {
connection().exec(sql, [])
}
/** 执行带参数的写语句(参数按方言占位符顺序绑定)。 */
public func exec(sql: String, params: Array<DbValue>): Int64 {
connection().exec(sql, params)
}
/** 执行 SELECT,返回行集。 */
public func query(sql: String): Array<Row> {
connection().query(sql, [])
}
/** 执行带参数的 SELECT。 */
public func query(sql: String, params: Array<DbValue>): Array<Row> {
connection().query(sql, params)
}
/** 表是否存在(走方言内省 SQL:SQLite→sqlite_master,PG/MySQL→information_schema)。 */
public func hasTable(table: String): Bool {
let (sql, ps) = dialect().introspectTableExists(table)
return connection().query(sql, ps).size > 0
}
/** 列出某表现有列名(顺序与底层一致)。 */
public func columnNames(table: String): Array<String> {
let (sql, ps) = dialect().introspectColumns(table)
let rows = connection().query(sql, ps)
let out = ArrayList<String>()
for (r in rows) {
out.add(r.getText("name"))
}
return out.toArray()
}
/** 某列是否存在。 */
public func hasColumn(table: String, col: String): Bool {
for (c in columnNames(table)) {
if (c == col) {
return true
}
}
return false
}
/** 列出某表现有列的「名 + 库现状类型」(顺序与底层一致),供 schema diff 比对/删列。 */
public func columnInfos(table: String): Array<(String, String)> {
let (sql, ps) = dialect().introspectColumnTypes(table)
let rows = connection().query(sql, ps)
let out = ArrayList<(String, String)>()
for (r in rows) {
out.add((r.getText("name"), r.getText("type")))
}
return out.toArray()
}
/** 建表(CREATE TABLE IF NOT EXISTS,幂等)。 */
public func createTable(table: String, cols: Array<ColumnSpec>): Unit {
connection().exec(dialect().createTable(table, cols), [])
}
/** 给已有表追加一列(ALTER TABLE ... ADD COLUMN)。 */
public func addColumn(table: String, col: ColumnSpec): Unit {
connection().exec(buildAddColumn(table, col, dialect()), [])
}
/**
* schema sync:对比「现状 → 目标」,把表 cols 同步到位——
* - 缺表则按 cols 建表;
* - 表已存在则逐列检查:
* · 缺列 → ADD COLUMN(已存在列默认保持不动);
* · 类型不一致 → ALTER COLUMN ... TYPE(PG)/MODIFY COLUMN(MySQL);
* SQLite 原地改列类型代价高,本轮 best-effort 跳过(方言 alterColumnType 返回 None)。
* · 主键列不参与改列类型(自增/SERIAL 不应被覆盖)。
* - dropMissing=true(显式开启)时:实体已无、库仍有的列 → DROP COLUMN。
* 默认 false 以防误删数据列;删列前需调用方明确开启。
*/
public func syncTable(table: String, cols: Array<ColumnSpec>, dropMissing!: Bool = false): Unit {
if (!hasTable(table)) {
createTable(table, cols)
return
}
let infos = columnInfos(table)
// 1. 缺列 ADD;2. 类型变更 ALTER/MODIFY(主键列与 SQLite 跳过)。
for (c in cols) {
if (!schemaColumnPresent(infos, c.name)) {
addColumn(table, c)
} else if (!c.isPk && !dialect().columnTypeMatches(schemaColumnType(infos, c.name), c)) {
match (dialect().alterColumnType(table, c)) {
case Some(sql) => connection().exec(sql, [])
case None => () // SQLite:best-effort 跳过原地改列类型
}
}
}
// 3. 删列(显式开启):实体侧已无、库仍有的列 → DROP COLUMN。
if (dropMissing) {
for (info in infos) {
let (name, _) = info
if (!schemaTargetHasColumn(cols, name)) {
connection().exec(dialect().dropColumn(table, name), [])
}
}
}
}
}
// infos 中是否存在名为 name 的列。
func schemaColumnPresent(infos: Array<(String, String)>, name: String): Bool {
for (info in infos) {
let (n, _) = info
if (n == name) {
return true
}
}
return false
}
// 取 infos 中 name 列的库现状类型("" 表示未找到)。
func schemaColumnType(infos: Array<(String, String)>, name: String): String {
for (info in infos) {
let (n, t) = info
if (n == name) {
return t
}
}
return ""
}
// 目标列集合中是否含名为 name 的列(删列 diff 用)。
func schemaTargetHasColumn(cols: Array<ColumnSpec>, name: String): Bool {
for (c in cols) {
if (c.name == name) {
return true
}
}
return false
}
/**
* 一次数据库迁移(对标 TypeORM Migration):name 唯一标识(也是历史表里的去重键),
* up 正向演进 schema/数据,down 回滚。两者均通过 QueryRunner 执行,跨方言可移植。
* 实现为类(class):`class AddUsers <: Migration { ... }`。
*/
public interface Migration {
/** 迁移唯一名(建议带时间戳前缀以保证执行顺序,如 "1700000000_create_user")。 */
func name(): String
/** 正向迁移:建表 / 加列 / 数据回填等。 */
func up(runner: QueryRunner): Unit
/** 回滚迁移:与 up 相反的操作。 */
func down(runner: QueryRunner): Unit
}
/** 迁移历史表名(记录已执行迁移,保证 runMigrations 幂等)。 */
let MIGRATION_HISTORY_TABLE = "ace_migrations"
/**
* 迁移执行器(对标 TypeORM MigrationExecutor):驱动一组 Migration 的执行与回滚,并维护历史表。
* - runMigrations:按数组顺序跑「尚未执行」的迁移的 up,逐条记入历史表(幂等:已执行则跳过)。
* - revertLast:回滚「最近一条已执行」迁移的 down,并从历史表删记录。
* 每条迁移在一个事务内执行(up/down + 历史表写入原子),失败则整体回滚、历史表不留半条。
*/
public class MigrationRunner {
let ds: DataSource
public init(ds: DataSource) {
this.ds = ds
}
/** 用全局数据源构造(配合 @Transactional / OrmComponent 场景)。 */
public static func fromGlobal(): MigrationRunner {
MigrationRunner(globalDataSource())
}
// 确保历史表存在(id 自增主键 / name 唯一 / executedAt 毫秒时间戳)。幂等。
// 兼容旧版历史表(只有 id+name,无 executedAt):CREATE TABLE IF NOT EXISTS 不补列,
// 需在建表后额外检测并 ALTER TABLE ADD COLUMN。
func ensureHistoryTable(): Unit {
let cols = [
ColumnSpec("id", DbType.TInt, true, ""),
ColumnSpec("name", DbType.TText, false, "", nullable: false, unique: true),
ColumnSpec("executedAt", DbType.TInt, false, "", nullable: false)
]
ds.driver.exec(ds.driver.dialect().createTable(MIGRATION_HISTORY_TABLE, cols), [])
let qr = QueryRunner(ds)
if (!qr.hasColumn(MIGRATION_HISTORY_TABLE, "executedAt")) {
qr.addColumn(MIGRATION_HISTORY_TABLE,
ColumnSpec("executedAt", DbType.TInt, false, "", nullable: false))
}
}
// 已执行迁移名集合(用于去重判定)。
func executedNames(): HashSet<String> {
let set = HashSet<String>()
let rows = ds.driver.query("SELECT name FROM " + MIGRATION_HISTORY_TABLE, [])
for (r in rows) {
set.add(r.getText("name"))
}
return set
}
// 记入一条执行历史。
func recordExecuted(name: String): Unit {
let d = ds.driver.dialect()
let sql = "INSERT INTO " + MIGRATION_HISTORY_TABLE + "(name, executedAt) VALUES(" + d.placeholder(0) + ", " +
d.placeholder(1) + ")"
ds.driver.exec(sql, [DbText(name), DbInt(nowMillis())])
}
// 删除一条执行历史。
func deleteRecord(name: String): Unit {
let d = ds.driver.dialect()
ds.driver.exec("DELETE FROM " + MIGRATION_HISTORY_TABLE + " WHERE name = " + d.placeholder(0), [DbText(name)])
}
/**
* 按数组顺序执行所有「尚未执行」的迁移的 up,逐条记入历史表。
* 幂等:历史表里已有同名记录的迁移会被跳过(可安全重复调用)。
* 返回本次实际执行的迁移名(按执行顺序)。
*/
public func runMigrations(migrations: Array<Migration>): Array<String> {
ensureHistoryTable()
let done = executedNames()
let runner = QueryRunner(ds)
let ran = ArrayList<String>()
for (mg in migrations) {
if (done.contains(mg.name())) {
continue
}
let m = mg // 固定绑定,避免闭包捕获循环变量
ds.transaction({
=>
m.up(runner)
recordExecuted(m.name())
})
ran.add(m.name())
}
return ran.toArray()
}
/**
* 回滚最近一条已执行的迁移:跑其 down 并从历史表删记录。
* 在 migrations 里按 name 找到对应迁移对象(才能调 down);找不到则抛清晰错误。
* 历史表为空(无可回滚)时返回 None。返回被回滚的迁移名。
*/
public func revertLast(migrations: Array<Migration>): ?String {
ensureHistoryTable()
let rows = ds.driver.query("SELECT name FROM " + MIGRATION_HISTORY_TABLE + " ORDER BY id DESC LIMIT 1", [])
if (rows.size == 0) {
return None
}
let lastName = rows[0].getText("name")
let runner = QueryRunner(ds)
for (mg in migrations) {
if (mg.name() == lastName) {
let m = mg
ds.transaction({
=>
m.down(runner)
deleteRecord(lastName)
})
return Some(lastName)
}
}
throw OrmException("ACE ORM: 找不到迁移 '" + lastName + "' 的定义,无法回滚(请把对应 Migration 传入 revertLast)")
}
/** 已执行迁移名列表(按执行顺序;对标 migration:show)。 */
public func executedMigrations(): Array<String> {
ensureHistoryTable()
let rows = ds.driver.query("SELECT name FROM " + MIGRATION_HISTORY_TABLE + " ORDER BY id ASC", [])
let out = ArrayList<String>()
for (r in rows) {
out.add(r.getText("name"))
}
return out.toArray()
}
}
