slave-framework 0.13.1

//! Schema migrations: SurrealQL files applied in order.
//!
//! Each migration is a `.surql` file in [`DIR`], named
//! `<YYYYMMDD_HHMMSS>_<name>.surql` so that names sort in the order they
//! were written. A migration runs in one transaction together with its
//! record in the [`TABLE`] table, so it is applied completely or not at
//! all, and never twice. Migrations must not contain `BEGIN`/`COMMIT`.
//!
//! Migrations only move forward: to undo one, write another.

use std::path::Path;

use surrealdb::types::SurrealValue;

use super::Db;
use crate::{Error, Result};

/// Where migrations live, relative to the project root.
pub const DIR: &str = "database/migrations";

/// The table recording applied migrations.
pub const TABLE: &str = "_migration";

/// A migration file.
#[derive(Debug, Clone)]
pub struct Migration {
    /// The file stem, e.g. `20260101_120000_create_posts`.
    pub name: String,
    pub source: String,
}

/// Every `.surql` file in `dir`, in name order.
pub fn load(dir: &Path) -> Result<Vec<Migration>> {
    let entries = std::fs::read_dir(dir)
        .map_err(|err| Error::Migration(format!("cannot read {}: {err}", dir.display())))?;
    let mut migrations = Vec::new();
    for path in entries.flatten().map(|entry| entry.path()) {
        if path.extension().is_none_or(|ext| ext != "surql") {
            continue;
        }
        let name = path
            .file_stem()
            .unwrap_or_default()
            .to_string_lossy()
            .into_owned();
        let source = std::fs::read_to_string(&path)
            .map_err(|err| Error::Migration(format!("cannot read {}: {err}", path.display())))?;
        if mentions_transaction(&source) {
            return Err(Error::Migration(format!(
                "{name}: remove BEGIN/COMMIT; every migration already runs in a transaction"
            )));
        }
        migrations.push(Migration { name, source });
    }
    migrations.sort_by(|a, b| a.name.cmp(&b.name));
    Ok(migrations)
}

/// The names of the applied migrations, in order.
pub async fn applied(db: &Db) -> Result<Vec<String>> {
    ensure_table(db).await?;
    let names: Vec<String> = db
        .query(format!("SELECT VALUE name FROM {TABLE} ORDER BY name"))
        .await?
        .check()?
        .take(0)?;
    Ok(names)
}

/// Each migration in `dir` and whether it has been applied.
pub async fn status(db: &Db, dir: &Path) -> Result<Vec<(Migration, bool)>> {
    let applied = applied(db).await?;
    Ok(load(dir)?
        .into_iter()
        .map(|migration| {
            let done = applied.contains(&migration.name);
            (migration, done)
        })
        .collect())
}

/// Applies the pending migrations in `dir`, in order, and returns their names.
pub async fn apply(db: &Db, dir: &Path) -> Result<Vec<String>> {
    let mut done = Vec::new();
    for (migration, applied) in status(db, dir).await? {
        if applied {
            continue;
        }
        let statement = format!(
            "BEGIN TRANSACTION;\n{}\n;\nCREATE {TABLE} SET name = $name, applied_at = time::now();\nCOMMIT TRANSACTION;",
            migration.source
        );
        db.query(statement)
            .bind(("name", migration.name.clone()))
            .await?
            .check()
            .map_err(|err| Error::Migration(format!("{}: {err}", migration.name)))?;
        done.push(migration.name);
    }
    Ok(done)
}

#[derive(Debug, Default, SurrealValue)]
struct DatabaseInfo {
    #[surreal(default)]
    tables: std::collections::BTreeMap<String, String>,
    #[surreal(default)]
    sequences: std::collections::BTreeMap<String, String>,
    #[surreal(default)]
    functions: std::collections::BTreeMap<String, String>,
    #[surreal(default)]
    params: std::collections::BTreeMap<String, String>,
    #[surreal(default)]
    analyzers: std::collections::BTreeMap<String, String>,
}

/// Removes every table (with its data), sequence, function, param and
/// analyzer from the database, including the record of applied migrations.
pub async fn reset(db: &Db) -> Result<()> {
    let info: Option<DatabaseInfo> = db.query("INFO FOR DB").await?.check()?.take(0)?;
    let info = info.unwrap_or_default();
    let mut statements = Vec::new();
    for (prefix, names) in [
        ("TABLE ", &info.tables),
        ("SEQUENCE ", &info.sequences),
        ("FUNCTION fn::", &info.functions),
        ("PARAM $", &info.params),
        ("ANALYZER ", &info.analyzers),
    ] {
        for name in names.keys() {
            statements.push(format!("REMOVE {prefix}{name};"));
        }
    }
    if !statements.is_empty() {
        db.query(statements.join("\n")).await?.check()?;
    }
    Ok(())
}

/// The command line of an app's `migrate` binary:
///
/// - `migrate` or `migrate apply`: apply pending migrations;
/// - `migrate status`: list migrations and whether each is applied;
/// - `migrate fresh [--force]`: drop everything, then apply all migrations
///   (asks to confirm unless `--force`).
///
/// Run from the project root, where [`DIR`] is.
pub async fn run_cli(db: &Db, args: &[String]) -> Result<()> {
    let dir = Path::new(DIR);
    let command = args.first().map(String::as_str).unwrap_or("apply");
    match command {
        "apply" => report_applied(apply(db, dir).await?),
        "status" => {
            let migrations = status(db, dir).await?;
            if migrations.is_empty() {
                println!("No migrations in {DIR}.");
            }
            for (migration, applied) in migrations {
                let state = if applied { "applied" } else { "pending" };
                println!("{state:<8} {}", migration.name);
            }
        }
        "fresh" => {
            let force = args.iter().any(|arg| arg == "--force");
            if !force && !confirm("Drop every table in this database and migrate again?")? {
                println!("Cancelled.");
                return Ok(());
            }
            reset(db).await?;
            println!("Dropped every table.");
            report_applied(apply(db, dir).await?);
        }
        other => {
            return Err(Error::Migration(format!(
                "unknown command `{other}`; use apply, status or fresh [--force]"
            )));
        }
    }
    Ok(())
}

fn report_applied(names: Vec<String>) {
    if names.is_empty() {
        println!("Nothing to migrate.");
    }
    for name in names {
        println!("Applied {name}");
    }
}

fn confirm(question: &str) -> Result<bool> {
    use std::io::Write;
    print!("{question} [y/N] ");
    std::io::stdout().flush().map_err(Error::Server)?;
    let mut answer = String::new();
    std::io::stdin()
        .read_line(&mut answer)
        .map_err(Error::Server)?;
    Ok(matches!(answer.trim(), "y" | "Y" | "yes"))
}

async fn ensure_table(db: &Db) -> Result<()> {
    db.query(format!(
        "DEFINE TABLE IF NOT EXISTS {TABLE} SCHEMAFULL;
         DEFINE FIELD IF NOT EXISTS name ON {TABLE} TYPE string;
         DEFINE FIELD IF NOT EXISTS applied_at ON {TABLE} TYPE datetime;
         DEFINE INDEX IF NOT EXISTS {TABLE}_name ON {TABLE} FIELDS name UNIQUE;"
    ))
    .await?
    .check()?;
    Ok(())
}

/// Whether `source` has a `BEGIN` or `COMMIT` statement (outside comments).
fn mentions_transaction(source: &str) -> bool {
    source
        .lines()
        .map(|line| line.trim_start().to_ascii_uppercase())
        .filter(|line| !line.starts_with("--") && !line.starts_with("//") && !line.starts_with('#'))
        .flat_map(|line| {
            line.split(';')
                .map(|statement| statement.trim().to_owned())
                .collect::<Vec<_>>()
        })
        .any(|statement| statement.starts_with("BEGIN") || statement.starts_with("COMMIT"))
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn spots_transactions_outside_comments() {
        assert!(mentions_transaction("BEGIN TRANSACTION;\nDEFINE TABLE x;"));
        assert!(mentions_transaction("DEFINE TABLE x; COMMIT;"));
        assert!(!mentions_transaction("-- BEGIN here\nDEFINE TABLE x;"));
        assert!(!mentions_transaction("DEFINE TABLE begin_at;"));
    }
}