// Copyright 2026 Leo Cheng
// SPDX-License-Identifier: Apache-2.0
// Versioned schema migrations — the Alembic/`alembic upgrade`, Diesel-migrations
// counterpart. A `Migration` bundles an integer version with the forward (`up`) and
// reverse (`down`) statement lists; a `Migrator` tracks which versions are applied
// in a `schema_migrations` table and drives the schema forward or back. Everything
// runs through the `@moondb.Driver` seam, so the same migrations apply against the
// MockDriver, real SQLite, or any other backend.
///|
/// One schema change: a monotonically increasing `version`, a human `name`, and the
/// ordered SQL statements that apply it (`up`) and undo it (`down`). Each statement
/// runs through `Session::execute`, so a backend that prepares one statement per
/// call (SQLite) still applies a multi-statement migration.
pub(all) struct Migration {
version : Int
name : String
up : Array[String]
down : Array[String]
}
///|
/// Tracks and applies migrations against a `schema_migrations`-style bookkeeping
/// table (its name is configurable for coexistence with other tools). Build one
/// with `Migrator::new`.
pub struct Migrator {
table : String
}
///|
/// A migrator recording applied versions in `table` (default `schema_migrations`).
/// The name must be a bare SQL identifier — it is interpolated as an identifier,
/// never bound — so a non-identifier is refused up front.
pub fn Migrator::new(
table? : String = "schema_migrations",
) -> Migrator raise @moondb.DbError {
if !is_ident(table) {
raise @moondb.QueryError("invalid migrations table name: " + table)
}
{ table, }
}
///|
/// Create the bookkeeping table if it is absent. Idempotent, so it is safe to call
/// before every `up`/`down`.
pub fn Migrator::ensure_table(
self : Migrator,
sess : Session,
) -> Unit raise @moondb.DbError {
sess.execute(
"CREATE TABLE IF NOT EXISTS " +
self.table +
" (version INTEGER PRIMARY KEY, name TEXT NOT NULL)",
[],
)
|> ignore
}
///|
/// Every applied version, ascending.
pub fn Migrator::applied_versions(
self : Migrator,
sess : Session,
) -> Array[Int] raise @moondb.DbError {
self.ensure_table(sess)
let rows = sess.query(
"SELECT version FROM " + self.table + " ORDER BY version ASC",
[],
)
let out : Array[Int] = []
for r in rows {
out.push(r.int(0))
}
out
}
///|
/// The highest applied version, or `0` when nothing has been applied yet.
pub fn Migrator::current_version(
self : Migrator,
sess : Session,
) -> Int raise @moondb.DbError {
let vs = self.applied_versions(sess)
if vs.length() == 0 {
0
} else {
vs[vs.length() - 1]
}
}
///|
/// Whether `version` is recorded as applied.
fn Migrator::is_applied(
self : Migrator,
sess : Session,
version : Int,
) -> Bool raise @moondb.DbError {
let rows = sess.query(
"SELECT version FROM " + self.table + " WHERE version = ?",
[Int(version)],
)
rows.length() > 0
}
///|
/// Apply every pending migration (those whose version is not yet recorded), in
/// ascending version order, running each one's `up` statements and recording it.
/// Returns how many were applied. Already-applied versions are skipped, so this is
/// safe to run repeatedly (it converges the schema to the latest version).
pub fn Migrator::up(
self : Migrator,
sess : Session,
migrations : Array[Migration],
) -> Int raise @moondb.DbError {
self.ensure_table(sess)
let ordered = sort_by_version(migrations, ascending=true)
let mut applied = 0
for m in ordered {
if self.is_applied(sess, m.version) {
continue
}
for stmt in m.up {
sess.execute(stmt, []) |> ignore
}
sess.execute(
"INSERT INTO " + self.table + " (version, name) VALUES (?, ?)",
[Int(m.version), Text(m.name)],
)
|> ignore
applied = applied + 1
}
applied
}
///|
/// Roll back every applied migration whose version is greater than `target`, in
/// descending version order, running each one's `down` statements and removing its
/// bookkeeping row. Returns how many were rolled back. `down_to(0)` unwinds
/// everything.
pub fn Migrator::down_to(
self : Migrator,
sess : Session,
migrations : Array[Migration],
target : Int,
) -> Int raise @moondb.DbError {
self.ensure_table(sess)
let ordered = sort_by_version(migrations, ascending=false)
let mut rolled = 0
for m in ordered {
if m.version <= target {
continue
}
if !self.is_applied(sess, m.version) {
continue
}
for stmt in m.down {
sess.execute(stmt, []) |> ignore
}
sess.execute("DELETE FROM " + self.table + " WHERE version = ?", [
Int(m.version),
])
|> ignore
rolled = rolled + 1
}
rolled
}
///|
/// Return a copy of `migrations` sorted by version. Migrations are typically a
/// handful, so an insertion sort keeps the code obvious and stable (equal versions
/// keep their input order, though versions are expected unique).
fn sort_by_version(
migrations : Array[Migration],
ascending~ : Bool,
) -> Array[Migration] {
let out : Array[Migration] = []
for m in migrations {
out.push(m)
}
for i in 1..= 0 {
let greater = out[j].version > key.version
let move_it = if ascending { greater } else { !greater }
if move_it {
out[j + 1] = out[j]
j = j - 1
} else {
break
}
}
out[j + 1] = key
}
out
}
///|
/// Map a SQLite declared type keyword back to a `ColumnType` (the inverse of
/// `ColumnType::sql`). Unknown affinities fall back to `TextType`.
fn coltype_of(kw : String) -> ColumnType {
match kw {
"INTEGER" => IntType
"TEXT" => TextType
"REAL" => RealType
"BLOB" => BlobType
"NUMERIC" => NumericType
"DATETIME" => DateTimeType
"DATE" => DateType
"TIME" => TimeType
_ => TextType
}
}
///|
/// Parse `PRAGMA table_info` rows into `Column`s — the reflection core, testable
/// without a live database. The pragma projects, in order, `cid, name, type,
/// notnull, dflt_value, pk`; a non-zero `pk` marks the primary key and `notnull`
/// its NOT NULL. This is the read half of Alembic-style autogenerate: compare the
/// reflected columns against a `Model`'s declared columns to diff a schema.
pub fn reflect_columns(
rows : Array[@moondb.Row],
) -> Array[Column] raise @moondb.DbError {
let cols = []
for r in rows {
cols.push(
column(
r.text(1),
coltype_of(r.text(2)),
primary_key=r.int(5) != 0,
nullable=r.int(3) == 0,
),
)
}
cols
}
///|
/// Reflect a table's columns off a live connection: run `PRAGMA table_info` and
/// parse it. The thin driver wrapper over `reflect_columns`.
pub fn reflect_table(
driver : &@moondb.Driver,
table : String,
) -> Array[Column] raise @moondb.DbError {
// PRAGMA takes an identifier, not a bindable parameter, so the name is validated
// rather than concatenated blind — the same is_ident discipline the rest uses.
guard is_ident(table) else {
raise @moondb.QueryError(
"reflect_table: invalid table identifier '\{table}'",
)
}
reflect_columns(driver.query("PRAGMA table_info(" + table + ")", []))
}
///|
/// One difference between a declared schema and a reflected one (Alembic-autogenerate
/// style): a column `Added` in the model but absent in the database, one `Removed`
/// from the model but still present, or one whose `TypeChanged`. Compared by column
/// name; ordering follows the declared columns, then the leftover reflected ones.
pub(all) enum ColumnDiff {
Added(name~ : String, col_type~ : ColumnType)
Removed(String)
TypeChanged(name~ : String, from~ : ColumnType, to~ : ColumnType)
} derive(Eq)
///|
/// The first column named `name`, or `None`.
fn find_col(cols : Array[Column], name : String) -> Column? {
cols.iter().find_first(fn(c) { c.name == name })
}
///|
/// Diff a model's `declared` columns against the `reflected` ones from a live table:
/// declared-only columns are `Added`, reflected-only are `Removed`, and a name in
/// both with a different `col_type` is `TypeChanged`. An empty result means the table
/// matches the model. This is the write half of autogenerate — feed the diff to DDL
/// to synthesise the migration.
///
/// Scope: reflection recovers only a column's storage type, so declare a model at the
/// same granularity to avoid spurious diffs — a `Bool` stored as `INTEGER` reflects as
/// `Int`, `Uuid`/`Json` stored as `TEXT` reflect as `Text`. And the diff reports
/// add/remove/type only, not nullable / default / unique / PK / FK changes.
pub fn diff_schema(
declared : Array[Column],
reflected : Array[Column],
) -> Array[ColumnDiff] {
let out : Array[ColumnDiff] = []
for d in declared {
match find_col(reflected, d.name) {
Some(r) =>
if d.col_type != r.col_type {
out.push(TypeChanged(name=d.name, from=r.col_type, to=d.col_type))
}
None => out.push(Added(name=d.name, col_type=d.col_type))
}
}
for r in reflected {
if find_col(declared, r.name) is None {
out.push(Removed(r.name))
}
}
out
}
///|
/// Render a `CREATE INDEX` statement (SQLAlchemy's `Index` / `index=True`). With
/// `unique=true` it is a `CREATE UNIQUE INDEX`; multiple `columns` make a composite
/// index. Names are trusted identifiers, never bound. Pair it with a `Migration`'s
/// `up`, and a `DROP INDEX` in its `down`.
pub fn index_ddl(
name : String,
table : String,
columns : Array[String],
unique? : Bool = false,
) -> String {
let kw = if unique { "CREATE UNIQUE INDEX " } else { "CREATE INDEX " }
kw + name + " ON " + table + " (" + join(columns, ", ") + ")"
}
///|
/// Map a PostgreSQL `information_schema.columns.data_type` back to a `ColumnType`.
fn coltype_of_pg(ty : String) -> ColumnType {
match ty {
"integer" | "bigint" | "smallint" => IntType
"boolean" => BoolType
"double precision" | "real" => RealType
"bytea" => BlobType
"numeric" => NumericType
"timestamp without time zone" | "timestamp with time zone" | "timestamp" =>
DateTimeType
"date" => DateType
"time without time zone" | "time" => TimeType
"uuid" => UuidType
"jsonb" | "json" => JsonType
_ => TextType
}
}
///|
/// Map a MySQL column `Type` (e.g. `int`, `varchar(255)`, `tinyint(1)`) back to a
/// `ColumnType`. Matched on the base keyword, most specific first.
fn coltype_of_mysql(ty : String) -> ColumnType {
if ty.has_prefix("tinyint(1)") {
BoolType
} else if ty.has_prefix("int") ||
ty.has_prefix("bigint") ||
ty.has_prefix("smallint") ||
ty.has_prefix("tinyint") {
IntType
} else if ty.has_prefix("varchar") || ty.has_prefix("char") {
TextType
} else if ty.has_prefix("text") {
TextType
} else if ty.has_prefix("double") || ty.has_prefix("float") {
RealType
} else if ty.has_prefix("blob") {
BlobType
} else if ty.has_prefix("decimal") || ty.has_prefix("numeric") {
NumericType
} else if ty.has_prefix("datetime") || ty.has_prefix("timestamp") {
DateTimeType
} else if ty.has_prefix("date") {
DateType
} else if ty.has_prefix("time") {
TimeType
} else if ty.has_prefix("json") {
JsonType
} else {
TextType
}
}
///|
/// Parse PostgreSQL `information_schema.columns` rows (`column_name`, `data_type`,
/// `is_nullable`) into `Column`s — the PG counterpart of `reflect_columns`.
pub fn reflect_columns_pg(
rows : Array[@moondb.Row],
) -> Array[Column] raise @moondb.DbError {
let cols = []
for r in rows {
cols.push(
column(r.text(0), coltype_of_pg(r.text(1)), nullable=r.text(2) == "YES"),
)
}
cols
}
///|
/// Parse MySQL `SHOW COLUMNS` rows (`Field`, `Type`, `Null`, `Key`, `Default`,
/// `Extra`) into `Column`s, recovering the primary key (`Key == "PRI"`) and
/// autoincrement (`Extra` contains `auto_increment`).
pub fn reflect_columns_mysql(
rows : Array[@moondb.Row],
) -> Array[Column] raise @moondb.DbError {
let cols = []
for r in rows {
cols.push(
column(
r.text(0),
coltype_of_mysql(r.text(1)),
nullable=r.text(2) == "YES",
primary_key=r.text(3) == "PRI",
autoincrement=r.text(5).contains("auto_increment"),
),
)
}
cols
}
///|
/// Reflect a table's columns off a live connection for a specific `Dialect`: SQLite
/// via `PRAGMA table_info`, PostgreSQL via `information_schema.columns`, MySQL via
/// `SHOW COLUMNS`. The multi-dialect counterpart of `reflect_table`.
pub fn reflect_table_for(
driver : &@moondb.Driver,
table : String,
dialect : Dialect,
) -> Array[Column] raise @moondb.DbError {
match dialect {
Sqlite => reflect_table(driver, table)
Postgres =>
reflect_columns_pg(
driver.query(
"SELECT column_name, data_type, is_nullable FROM information_schema.columns WHERE table_name = ? ORDER BY ordinal_position",
[Text(table)],
),
)
Mysql => {
// SHOW COLUMNS names an identifier, not a bindable parameter — validate it.
guard is_ident(table) else {
raise @moondb.QueryError("reflect: invalid table identifier '\{table}'")
}
reflect_columns_mysql(driver.query("SHOW COLUMNS FROM " + table, []))
}
}
}