renox-core 1.1.0

//! Turns a list of schema changes into SQL: the writing half of `db:diff`.

use std::collections::HashMap;

use super::quote;
use super::schema_check::{ForeignKey, TableColumn, kind_fits};
use super::schema_diff::{Change, TableState};
use super::{ColumnKind, Dialect, ModelColumn};

/// The SQL type of a column kind; `None` for [`ColumnKind::Unknown`].
pub(crate) fn sql_type(kind: ColumnKind, dialect: Dialect) -> Option<&'static str> {
    use ColumnKind as K;
    let pg = dialect == Dialect::Postgres;
    let (lite, postgres) = match kind {
        K::BigInt => ("INTEGER", "BIGINT"),
        K::Int => ("INTEGER", "INTEGER"),
        K::SmallInt => ("INTEGER", "SMALLINT"),
        K::Double => ("REAL", "DOUBLE PRECISION"),
        K::Real => ("REAL", "REAL"),
        K::Text => ("TEXT", "TEXT"),
        K::Blob => ("BLOB", "BYTEA"),
        K::Bool => ("INTEGER", "BOOLEAN"),
        K::DateTime => ("TEXT", "TIMESTAMPTZ"),
        K::NaiveDateTime => ("TEXT", "TIMESTAMP"),
        K::Date => ("TEXT", "DATE"),
        K::Time => ("TEXT", "TIME"),
        K::Json => ("TEXT", "JSONB"),
        K::Uuid => ("BLOB", "UUID"),
        K::Unknown => return None,
    };
    Some(if pg { postgres } else { lite })
}

/// `"name" TYPE [NOT NULL] [DEFAULT x] [REFERENCES "t" ("id")]`; the `id`
/// column is the primary key.
pub(crate) fn column_def(c: &ModelColumn, dialect: Dialect) -> String {
    let name = quote(c.name);
    let ty = sql_type(c.kind, dialect).unwrap_or("TEXT");
    if c.name == "id" {
        return match (c.kind, dialect) {
            (ColumnKind::BigInt | ColumnKind::Int, Dialect::Sqlite) => {
                format!("{name} INTEGER PRIMARY KEY AUTOINCREMENT")
            }
            (ColumnKind::BigInt | ColumnKind::Int, Dialect::Postgres) => {
                format!("{name} BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY")
            }
            _ => format!("{name} {ty} PRIMARY KEY"),
        };
    }
    let mut out = format!("{name} {ty}");
    if !c.nullable {
        out.push_str(" NOT NULL");
    }
    if let Some(default) = c.default {
        out.push_str(&format!(" DEFAULT {default}"));
    }
    if let Some(table) = c.references {
        out.push_str(&format!(" REFERENCES {} (\"id\")", quote(table)));
    }
    out
}

fn columns_list(columns: &[String]) -> String {
    columns
        .iter()
        .map(|c| quote(c))
        .collect::<Vec<_>>()
        .join(", ")
}

/// The PostgreSQL `(up, down)` SQL for `changes`: up in order, down in
/// reverse order, every statement ending in `;\n`.
pub(crate) fn postgres(changes: &[Change]) -> (String, String) {
    let d = Dialect::Postgres;
    let mut up = String::new();
    let mut downs: Vec<String> = Vec::new();
    for change in changes {
        let mut down = String::new();
        match change {
            Change::CreateTable { model } => {
                let defs: Vec<String> = model.columns.iter().map(|c| column_def(c, d)).collect();
                up.push_str(&format!(
                    "CREATE TABLE {} (\n    {}\n);\n",
                    quote(model.table),
                    defs.join(",\n    ")
                ));
                down.push_str(&format!("DROP TABLE {};\n", quote(model.table)));
            }
            Change::AddColumn { table, column } => {
                up.push_str(&format!(
                    "ALTER TABLE {} ADD COLUMN {};\n",
                    quote(table),
                    column_def(column, d)
                ));
                down.push_str(&format!(
                    "ALTER TABLE {} DROP COLUMN {};\n",
                    quote(table),
                    quote(column.name)
                ));
            }
            Change::DropColumn { table, old } => {
                up.push_str(&format!(
                    "ALTER TABLE {} DROP COLUMN {};\n",
                    quote(table),
                    quote(&old.name)
                ));
                let mut def = format!("{} {}", quote(&old.name), old.sql_type);
                if old.not_null {
                    def.push_str(" NOT NULL");
                }
                if let Some(default) = &old.default {
                    def.push_str(&format!(" DEFAULT {default}"));
                }
                down.push_str(&format!("ALTER TABLE {} ADD COLUMN {def};\n", quote(table)));
            }
            Change::AlterColumn { table, old, new } => {
                let t = quote(table);
                let c = quote(new.name);
                let new_ty = sql_type(new.kind, d).unwrap_or("TEXT");
                if !kind_fits(d, new.kind, &old.sql_type) {
                    up.push_str(&format!(
                        "ALTER TABLE {t} ALTER COLUMN {c} TYPE {new_ty} USING {c}::{new_ty};\n"
                    ));
                    down.push_str(&format!(
                        "ALTER TABLE {t} ALTER COLUMN {c} TYPE {0} USING {c}::{0};\n",
                        old.sql_type
                    ));
                }
                if new.nullable == old.not_null {
                    let (to, back) = if new.nullable {
                        ("DROP", "SET")
                    } else {
                        ("SET", "DROP")
                    };
                    up.push_str(&format!(
                        "ALTER TABLE {t} ALTER COLUMN {c} {to} NOT NULL;\n"
                    ));
                    down.insert_str(
                        0,
                        &format!("ALTER TABLE {t} ALTER COLUMN {c} {back} NOT NULL;\n"),
                    );
                }
            }
            Change::RenameColumn { table, from, to } => {
                up.push_str(&format!(
                    "ALTER TABLE {} RENAME COLUMN {} TO {};\n",
                    quote(table),
                    quote(from),
                    quote(to)
                ));
                down.push_str(&format!(
                    "ALTER TABLE {} RENAME COLUMN {} TO {};\n",
                    quote(table),
                    quote(to),
                    quote(from)
                ));
            }
            Change::CreateIndex {
                table,
                name,
                columns,
                unique,
            } => {
                up.push_str(&format!(
                    "CREATE {}INDEX {} ON {} ({});\n",
                    if *unique { "UNIQUE " } else { "" },
                    quote(name),
                    quote(table),
                    columns_list(columns)
                ));
                down.push_str(&format!("DROP INDEX {};\n", quote(name)));
            }
            Change::DropIndex { old, .. } => {
                up.push_str(&format!("DROP INDEX {};\n", quote(&old.name)));
                if let Some(sql) = &old.sql {
                    down.push_str(sql.trim_end_matches(';'));
                    down.push_str(";\n");
                }
            }
        }
        downs.push(down);
    }
    downs.reverse();
    (up, downs.concat())
}

/// What the SQLite writer needs to know about the current database; the
/// caller fills it with the readers of `schema_check`.
#[derive(Debug, Clone, Default)]
pub(crate) struct SqliteContext {
    /// The tables as they are now.
    pub tables: HashMap<String, TableState>,
    /// Each table's own `CREATE TABLE` text.
    pub table_sql: HashMap<String, String>,
    /// The foreign keys of other tables that point at each table.
    pub referenced_by: HashMap<String, Vec<ForeignKey>>,
}

fn change_table(change: &Change) -> &str {
    match change {
        Change::CreateTable { model } => model.table,
        Change::AddColumn { table, .. }
        | Change::DropColumn { table, .. }
        | Change::AlterColumn { table, .. }
        | Change::RenameColumn { table, .. }
        | Change::CreateIndex { table, .. }
        | Change::DropIndex { table, .. } => table,
    }
}

/// Whether SQLite can do the group's changes with `ALTER TABLE` alone.
fn is_simple(group: &[&Change]) -> bool {
    group
        .iter()
        .any(|c| matches!(c, Change::CreateTable { .. }))
        || group.iter().all(|c| match c {
            Change::AddColumn { column, .. } => {
                column.nullable || (column.default.is_some() && column.references.is_none())
            }
            Change::RenameColumn { .. } | Change::CreateIndex { .. } | Change::DropIndex { .. } => {
                true
            }
            _ => false,
        })
}

fn create_index_sql(table: &str, name: &str, columns: &[String], unique: bool) -> String {
    format!(
        "CREATE {}INDEX {} ON {} ({});\n",
        if unique { "UNIQUE " } else { "" },
        quote(name),
        quote(table),
        columns_list(columns)
    )
}

fn with_semicolon(sql: &str) -> String {
    format!("{};\n", sql.trim().trim_end_matches(';'))
}

/// The SQL for a group of changes that need only `ALTER TABLE` and indexes.
fn simple(group: &[&Change]) -> (String, String) {
    let d = Dialect::Sqlite;
    let mut up = String::new();
    let mut downs: Vec<String> = Vec::new();
    for change in group {
        let mut down = String::new();
        match change {
            Change::CreateTable { model } => {
                let defs: Vec<String> = model.columns.iter().map(|c| column_def(c, d)).collect();
                up.push_str(&format!(
                    "CREATE TABLE {} (\n    {}\n);\n",
                    quote(model.table),
                    defs.join(",\n    ")
                ));
                down.push_str(&format!("DROP TABLE {};\n", quote(model.table)));
            }
            Change::AddColumn { table, column } => {
                up.push_str(&format!(
                    "ALTER TABLE {} ADD COLUMN {};\n",
                    quote(table),
                    column_def(column, d)
                ));
                down.push_str(&format!(
                    "ALTER TABLE {} DROP COLUMN {};\n",
                    quote(table),
                    quote(column.name)
                ));
            }
            Change::RenameColumn { table, from, to } => {
                up.push_str(&format!(
                    "ALTER TABLE {} RENAME COLUMN {} TO {};\n",
                    quote(table),
                    quote(from),
                    quote(to)
                ));
                down.push_str(&format!(
                    "ALTER TABLE {} RENAME COLUMN {} TO {};\n",
                    quote(table),
                    quote(to),
                    quote(from)
                ));
            }
            Change::CreateIndex {
                table,
                name,
                columns,
                unique,
            } => {
                up.push_str(&create_index_sql(table, name, columns, *unique));
                down.push_str(&format!("DROP INDEX {};\n", quote(name)));
            }
            Change::DropIndex { old, .. } => {
                up.push_str(&format!("DROP INDEX {};\n", quote(&old.name)));
                if let Some(sql) = &old.sql {
                    down.push_str(&with_semicolon(sql));
                }
            }
            Change::DropColumn { .. } | Change::AlterColumn { .. } => {}
        }
        downs.push(down);
    }
    downs.reverse();
    (up, downs.concat())
}

/// Whether `word` appears in `sql` as a keyword (not inside a longer name);
/// with `paren`, only when an opening parenthesis follows it.
fn has_keyword(sql: &str, word: &str, paren: bool) -> bool {
    let lower = sql.to_lowercase();
    let bytes = lower.as_bytes();
    let ident = |b: u8| b.is_ascii_alphanumeric() || b == b'_';
    let mut from = 0;
    while let Some(at) = lower[from..].find(word) {
        let start = from + at;
        let end = start + word.len();
        from = start + 1;
        if start > 0 && ident(bytes[start - 1]) {
            continue;
        }
        if end < bytes.len() && ident(bytes[end]) {
            continue;
        }
        if paren && !lower[end..].trim_start().starts_with('(') {
            continue;
        }
        return true;
    }
    false
}

/// The first reason the table can't be rebuilt safely, if any.
fn rebuild_problem(table: &str, cx: &SqliteContext) -> Option<String> {
    let Some(sql) = cx.table_sql.get(table) else {
        return Some("its CREATE TABLE text is unknown".into());
    };
    if has_keyword(sql, "check", false) {
        return Some("it has a CHECK constraint".into());
    }
    if has_keyword(sql, "constraint", false) {
        return Some("it has a named constraint".into());
    }
    if has_keyword(sql, "unique", true) {
        return Some("it has a table-level UNIQUE".into());
    }
    if has_keyword(sql, "primary key", true) {
        return Some("it has a table-level PRIMARY KEY".into());
    }
    if let Some(key) = cx.referenced_by.get(table).and_then(|keys| {
        keys.iter().find(|k| {
            let a = k.on_delete.to_uppercase();
            a != "NO ACTION" && a != "RESTRICT"
        })
    }) {
        return Some(format!(
            "`{}.{}` references it with ON DELETE {}",
            key.table, key.column, key.on_delete
        ));
    }
    None
}

fn needs_hand(table: &str, why: &str) -> crate::Error {
    crate::Error::Internal(anyhow::anyhow!(
        "`{table}` needs a table rebuild that Renox can't write safely ({why}); \
         write this migration by hand"
    ))
}

/// The pieces of a `CREATE TABLE`: its column definitions (by lowercase name)
/// and the other clauses, in order.
struct Body {
    columns: Vec<(String, String)>,
    others: Vec<String>,
}

fn split_body(sql: &str) -> Option<Body> {
    let open = sql.find('(')?;
    let close = sql.rfind(')')?;
    if close <= open {
        return None;
    }
    let body = &sql[open + 1..close];
    let mut parts: Vec<String> = Vec::new();
    let (mut depth, mut quote_ch, mut cur) = (0usize, None::<char>, String::new());
    for c in body.chars() {
        match quote_ch {
            Some(q) => {
                if c == q || (q == '[' && c == ']') {
                    quote_ch = None;
                }
                cur.push(c);
            }
            None => match c {
                '\'' | '"' | '`' | '[' => {
                    quote_ch = Some(c);
                    cur.push(c);
                }
                '(' => {
                    depth += 1;
                    cur.push(c);
                }
                ')' => {
                    depth = depth.saturating_sub(1);
                    cur.push(c);
                }
                ',' if depth == 0 => parts.push(std::mem::take(&mut cur)),
                _ => cur.push(c),
            },
        }
    }
    parts.push(cur);
    let mut out = Body {
        columns: Vec::new(),
        others: Vec::new(),
    };
    for part in parts {
        let part = part.trim().to_string();
        if part.is_empty() {
            continue;
        }
        let (name, rest) = leading_identifier(&part);
        let lower = name.to_lowercase();
        if matches!(
            lower.as_str(),
            "constraint" | "primary" | "unique" | "check" | "foreign"
        ) && !part.starts_with(['"', '`', '['])
        {
            out.others.push(part);
        } else {
            let _ = rest;
            out.columns.push((lower, part));
        }
    }
    Some(out)
}

/// The identifier a definition starts with, and the text after it.
fn leading_identifier(def: &str) -> (String, &str) {
    let mut chars = def.char_indices();
    match chars.next() {
        Some((_, q @ ('"' | '`' | '['))) => {
            let end = if q == '[' { ']' } else { q };
            match def[1..].find(end) {
                Some(i) => (def[1..1 + i].to_string(), &def[i + 2..]),
                None => (def.to_string(), ""),
            }
        }
        _ => match def.find(char::is_whitespace) {
            Some(i) => (def[..i].to_string(), &def[i..]),
            None => (def.to_string(), ""),
        },
    }
}

/// A definition from the table's own SQL, under a new name when asked.
fn kept_def(body: Option<&Body>, col: &TableColumn, new_name: &str) -> String {
    if let Some(body) = body
        && let Some((_, def)) = body
            .columns
            .iter()
            .find(|(n, _)| *n == col.name.to_lowercase())
    {
        let (_, rest) = leading_identifier(def);
        return format!("{}{}", quote(new_name), rest);
    }
    let mut out = format!("{} {}", quote(new_name), col.sql_type);
    if col.primary_key {
        out.push_str(" PRIMARY KEY");
    } else {
        if col.not_null {
            out.push_str(" NOT NULL");
        }
        if let Some(default) = &col.default {
            out.push_str(&format!(" DEFAULT {default}"));
        }
    }
    out
}

/// The statements that replace `table` with one of the given shape, copying
/// `into` from `from` (a select list over the old table).
///
/// The rows go through a staging table `__new_t`, and the table is created
/// again under its own name rather than renamed: after `DROP TABLE`, SQLite
/// only forgets the foreign keys of other tables that pointed at it when rows
/// are inserted into a table of that name (renaming leaves `COMMIT` failing).
fn rebuild_statements(table: &str, defs: &[String], into: &[String], from: &str) -> String {
    let temp = quote(&format!("__new_{table}"));
    let t = quote(table);
    let body = defs.join(",\n    ");
    let mut out = String::from("PRAGMA defer_foreign_keys = ON;\n");
    out.push_str(&format!("CREATE TABLE {temp} (\n    {body}\n);\n"));
    if !into.is_empty() {
        out.push_str(&format!(
            "INSERT INTO {temp} ({}) SELECT {from} FROM {t};\n",
            columns_list(into)
        ));
    }
    out.push_str(&format!("DROP TABLE {t};\n"));
    out.push_str(&format!("CREATE TABLE {t} (\n    {body}\n);\n"));
    if !into.is_empty() {
        let list = columns_list(into);
        out.push_str(&format!(
            "INSERT INTO {t} ({list}) SELECT {list} FROM {temp};\n"
        ));
    }
    out.push_str(&format!("DROP TABLE {temp};\n"));
    out
}

/// Rebuilds `table` (decision 9): `(up, down)`.
fn rebuild(table: &str, group: &[&Change], cx: &SqliteContext) -> crate::Result<(String, String)> {
    if let Some(why) = rebuild_problem(table, cx) {
        return Err(needs_hand(table, &why));
    }
    let state = cx.tables.get(table).ok_or_else(|| {
        crate::Error::Internal(anyhow::anyhow!(
            "the table `{table}` is not in the schema read"
        ))
    })?;
    let body = cx.table_sql.get(table).and_then(|s| split_body(s));
    let mut renames: HashMap<&str, &str> = HashMap::new();
    let mut dropped: Vec<&str> = Vec::new();
    let mut altered: HashMap<&str, &ModelColumn> = HashMap::new();
    let mut added: Vec<&ModelColumn> = Vec::new();
    for change in group {
        match change {
            Change::RenameColumn { from, to, .. } => {
                renames.insert(from, to);
            }
            Change::DropColumn { old, .. } => dropped.push(&old.name),
            Change::AlterColumn { old, new, .. } => {
                altered.insert(&old.name, new);
            }
            Change::AddColumn { column, .. } => added.push(column),
            _ => {}
        }
    }
    let mut new_defs: Vec<String> = Vec::new();
    let mut shared: Vec<(&str, String)> = Vec::new(); // (old name, new name)
    let mut coalesce: HashMap<&str, &str> = HashMap::new();
    for col in &state.columns {
        if dropped.contains(&col.name.as_str()) {
            continue;
        }
        let new_name = renames
            .get(col.name.as_str())
            .map_or(col.name.clone(), |n| n.to_string());
        match altered.get(col.name.as_str()) {
            Some(new) => {
                new_defs.push(column_def(new, Dialect::Sqlite));
                // NULLs become the default when the column turns NOT NULL.
                if let (false, false, Some(default)) = (new.nullable, col.not_null, new.default) {
                    coalesce.insert(&col.name, default);
                }
            }
            None => new_defs.push(kept_def(body.as_ref(), col, &new_name)),
        }
        shared.push((&col.name, new_name));
    }
    for column in &added {
        new_defs.push(column_def(column, Dialect::Sqlite));
    }
    let touched = |name: &str| dropped.contains(&name) || renames.contains_key(name);
    for other in body.iter().flat_map(|b| &b.others) {
        for col in &state.columns {
            if touched(&col.name) && has_keyword(other, &col.name.to_lowercase(), false) {
                return Err(needs_hand(
                    table,
                    "a table-level clause names a changed column",
                ));
            }
        }
        new_defs.push(other.clone());
    }
    let into: Vec<String> = shared.iter().map(|(_, n)| n.clone()).collect();
    let from: Vec<String> = shared
        .iter()
        .map(|(o, _)| match coalesce.get(o) {
            Some(default) => format!("COALESCE({}, {default})", quote(o)),
            None => quote(o),
        })
        .collect();
    let mut up = rebuild_statements(table, &new_defs, &into, &from.join(", "));
    // Indexes: the old ones that survive, then the new ones.
    let dropped_indexes: Vec<&str> = group
        .iter()
        .filter_map(|c| match c {
            Change::DropIndex { old, .. } => Some(old.name.as_str()),
            _ => None,
        })
        .collect();
    for index in &state.indexes {
        let Some(sql) = &index.sql else { continue };
        if dropped_indexes.contains(&index.name.as_str())
            || index.columns.iter().any(|c| dropped.contains(&c.as_str()))
        {
            continue;
        }
        if index
            .columns
            .iter()
            .any(|c| renames.contains_key(c.as_str()))
        {
            if has_keyword(sql, "where", false) {
                return Err(needs_hand(table, "a partial index uses a renamed column"));
            }
            let columns: Vec<String> = index
                .columns
                .iter()
                .map(|c| renames.get(c.as_str()).map_or(c.clone(), |n| n.to_string()))
                .collect();
            up.push_str(&create_index_sql(
                table,
                &index.name,
                &columns,
                index.unique,
            ));
        } else {
            up.push_str(&with_semicolon(sql));
        }
    }
    for change in group {
        if let Change::CreateIndex {
            name,
            columns,
            unique,
            ..
        } = change
        {
            up.push_str(&create_index_sql(table, name, columns, *unique));
        }
    }
    // Down: the old table again.
    let old_defs: Vec<String> = match &body {
        Some(b) => b
            .columns
            .iter()
            .map(|(_, d)| d.clone())
            .chain(b.others.iter().cloned())
            .collect(),
        None => state
            .columns
            .iter()
            .map(|c| kept_def(None, c, &c.name))
            .collect(),
    };
    let back_into: Vec<String> = shared.iter().map(|(o, _)| o.to_string()).collect();
    let back_from = columns_list(&into);
    let mut down = rebuild_statements(table, &old_defs, &back_into, &back_from);
    for index in &state.indexes {
        if let Some(sql) = &index.sql {
            down.push_str(&with_semicolon(sql));
        }
    }
    Ok((up, down))
}

/// The SQLite `(up, down)` SQL for `changes`: up in order, down in reverse
/// order. A table that `ALTER TABLE` can't change is rebuilt (decision 9).
pub(crate) fn sqlite(changes: &[Change], cx: &SqliteContext) -> crate::Result<(String, String)> {
    let mut groups: Vec<(String, Vec<&Change>)> = Vec::new();
    for change in changes {
        let table = change_table(change);
        match groups.iter_mut().find(|(t, _)| t == table) {
            Some((_, g)) => g.push(change),
            None => groups.push((table.to_string(), vec![change])),
        }
    }
    let mut up = String::new();
    let mut downs: Vec<String> = Vec::new();
    for (table, group) in &groups {
        let (u, d) = if is_simple(group) {
            simple(group)
        } else {
            rebuild(table, group, cx)?
        };
        up.push_str(&u);
        downs.push(d);
    }
    downs.reverse();
    Ok((up, downs.concat()))
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::db::ModelIndex;
    use crate::db::schema_check::{TableColumn, TableIndex};
    use crate::db::schema_diff::ModelState;

    fn col(name: &'static str, kind: ColumnKind, nullable: bool) -> ModelColumn {
        ModelColumn::new(name, "T", kind, nullable)
    }

    fn old(name: &str, ty: &str, not_null: bool) -> TableColumn {
        TableColumn {
            name: name.into(),
            sql_type: ty.into(),
            not_null,
            default: None,
            primary_key: false,
            identity: false,
        }
    }

    #[test]
    fn types_follow_the_table() {
        assert_eq!(sql_type(ColumnKind::Json, Dialect::Postgres), Some("JSONB"));
        assert_eq!(sql_type(ColumnKind::Json, Dialect::Sqlite), Some("TEXT"));
        assert_eq!(
            sql_type(ColumnKind::SmallInt, Dialect::Postgres),
            Some("SMALLINT")
        );
        assert_eq!(
            sql_type(ColumnKind::Int, Dialect::Postgres),
            Some("INTEGER")
        );
        assert_eq!(sql_type(ColumnKind::Real, Dialect::Postgres), Some("REAL"));
        assert_eq!(sql_type(ColumnKind::Uuid, Dialect::Sqlite), Some("BLOB"));
        assert_eq!(sql_type(ColumnKind::Unknown, Dialect::Postgres), None);
    }

    #[test]
    fn create_table_and_index() {
        let model = ModelState {
            table: "posts",
            columns: vec![
                col("id", ColumnKind::BigInt, false),
                col("user_id", ColumnKind::BigInt, false).references("users"),
                col("views", ColumnKind::BigInt, false).default_sql("0"),
                col("note", ColumnKind::Text, true),
            ],
            indexes: vec![ModelIndex::new(&["user_id"], false)],
        };
        let changes = [
            Change::CreateTable { model },
            Change::CreateIndex {
                table: "posts".into(),
                name: "posts_user_id_index".into(),
                columns: vec!["user_id".into()],
                unique: false,
            },
        ];
        let (up, down) = postgres(&changes);
        assert_eq!(
            up,
            "CREATE TABLE \"posts\" (\n    \"id\" BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,\n    \"user_id\" BIGINT NOT NULL REFERENCES \"users\" (\"id\"),\n    \"views\" BIGINT NOT NULL DEFAULT 0,\n    \"note\" TEXT\n);\nCREATE INDEX \"posts_user_id_index\" ON \"posts\" (\"user_id\");\n"
        );
        assert_eq!(
            down,
            "DROP INDEX \"posts_user_id_index\";\nDROP TABLE \"posts\";\n"
        );
    }

    #[test]
    fn columns_alter_rename_and_drop() {
        let changes = [
            Change::AddColumn {
                table: "t".into(),
                column: col("a", ColumnKind::Bool, false).default_sql("false"),
            },
            Change::DropColumn {
                table: "t".into(),
                old: TableColumn {
                    default: Some("'x'::text".into()),
                    ..old("b", "text", true)
                },
            },
            Change::AlterColumn {
                table: "t".into(),
                old: old("c", "integer", true),
                new: col("c", ColumnKind::Text, true),
            },
            Change::AlterColumn {
                table: "t".into(),
                old: old("d", "text", false),
                new: col("d", ColumnKind::Text, false),
            },
            Change::RenameColumn {
                table: "t".into(),
                from: "e".into(),
                to: "f".into(),
            },
        ];
        let (up, down) = postgres(&changes);
        assert_eq!(
            up,
            "ALTER TABLE \"t\" ADD COLUMN \"a\" BOOLEAN NOT NULL DEFAULT false;\n\
             ALTER TABLE \"t\" DROP COLUMN \"b\";\n\
             ALTER TABLE \"t\" ALTER COLUMN \"c\" TYPE TEXT USING \"c\"::TEXT;\n\
             ALTER TABLE \"t\" ALTER COLUMN \"c\" DROP NOT NULL;\n\
             ALTER TABLE \"t\" ALTER COLUMN \"d\" SET NOT NULL;\n\
             ALTER TABLE \"t\" RENAME COLUMN \"e\" TO \"f\";\n"
        );
        assert_eq!(
            down,
            "ALTER TABLE \"t\" RENAME COLUMN \"f\" TO \"e\";\n\
             ALTER TABLE \"t\" ALTER COLUMN \"d\" DROP NOT NULL;\n\
             ALTER TABLE \"t\" ALTER COLUMN \"c\" SET NOT NULL;\n\
             ALTER TABLE \"t\" ALTER COLUMN \"c\" TYPE integer USING \"c\"::integer;\n\
             ALTER TABLE \"t\" ADD COLUMN \"b\" text NOT NULL DEFAULT 'x'::text;\n\
             ALTER TABLE \"t\" DROP COLUMN \"a\";\n"
        );
    }

    #[test]
    fn indexes_drop_and_unique() {
        let idx = TableIndex {
            name: "t_a_index".into(),
            columns: vec!["a".into()],
            unique: false,
            sql: Some("CREATE INDEX t_a_index ON public.t USING btree (a)".into()),
        };
        let changes = [
            Change::DropIndex {
                table: "t".into(),
                old: idx,
            },
            Change::CreateIndex {
                table: "t".into(),
                name: "t_a_b_unique".into(),
                columns: vec!["a".into(), "b".into()],
                unique: true,
            },
        ];
        let (up, down) = postgres(&changes);
        assert_eq!(
            up,
            "DROP INDEX \"t_a_index\";\nCREATE UNIQUE INDEX \"t_a_b_unique\" ON \"t\" (\"a\", \"b\");\n"
        );
        assert_eq!(
            down,
            "DROP INDEX \"t_a_b_unique\";\nCREATE INDEX t_a_index ON public.t USING btree (a);\n"
        );
    }

    fn cx_for(table: &str, create: &str, cols: Vec<TableColumn>) -> SqliteContext {
        let mut cx = SqliteContext::default();
        cx.tables.insert(
            table.into(),
            TableState {
                table: table.into(),
                columns: cols,
                indexes: vec![TableIndex {
                    name: "t_a_index".into(),
                    columns: vec!["a".into()],
                    unique: false,
                    sql: Some("CREATE INDEX t_a_index ON t (a)".into()),
                }],
            },
        );
        cx.table_sql.insert(table.into(), create.into());
        cx
    }

    const CREATE: &str =
        "CREATE TABLE t (id INTEGER PRIMARY KEY AUTOINCREMENT, a INTEGER NOT NULL, b TEXT)";

    fn t_cols() -> Vec<TableColumn> {
        let mut id = old("id", "INTEGER", false);
        id.primary_key = true;
        vec![id, old("a", "INTEGER", true), old("b", "TEXT", false)]
    }

    #[test]
    fn sqlite_adds_stay_alter_table() {
        let changes = [
            Change::AddColumn {
                table: "t".into(),
                column: col("c", ColumnKind::Text, true),
            },
            Change::AddColumn {
                table: "t".into(),
                column: col("d", ColumnKind::Bool, false).default_sql("0"),
            },
            Change::RenameColumn {
                table: "t".into(),
                from: "b".into(),
                to: "e".into(),
            },
        ];
        let (up, down) = sqlite(&changes, &SqliteContext::default()).unwrap();
        assert!(up.contains("ALTER TABLE \"t\" ADD COLUMN \"c\" TEXT;"));
        assert!(up.contains("ADD COLUMN \"d\" INTEGER NOT NULL DEFAULT 0;"));
        assert!(!up.contains("__new_"));
        assert!(down.starts_with("ALTER TABLE \"t\" RENAME COLUMN \"e\" TO \"b\";"));
        assert!(down.contains("DROP COLUMN \"c\""));
    }

    #[test]
    fn sqlite_type_change_is_a_rebuild() {
        let cx = cx_for("t", CREATE, t_cols());
        let changes = [Change::AlterColumn {
            table: "t".into(),
            old: old("b", "TEXT", false),
            new: col("b", ColumnKind::Int, false),
        }];
        let (up, down) = sqlite(&changes, &cx).unwrap();
        assert!(up.starts_with("PRAGMA defer_foreign_keys = ON;\nCREATE TABLE \"__new_t\""));
        assert!(up.contains("\"b\" INTEGER NOT NULL"));
        assert!(up.contains(
            "INSERT INTO \"__new_t\" (\"id\", \"a\", \"b\") SELECT \"id\", \"a\", \"b\" FROM \"t\";"
        ));
        assert!(up.contains("DROP TABLE \"t\";\nCREATE TABLE \"t\" ("));
        assert!(up.contains("DROP TABLE \"__new_t\";"));
        assert!(up.contains("CREATE INDEX t_a_index ON t (a);"));
        assert!(up.contains("AUTOINCREMENT"));
        assert!(down.contains("b TEXT"));
    }

    #[test]
    fn sqlite_rebuild_guards() {
        let change = [Change::DropColumn {
            table: "t".into(),
            old: old("b", "TEXT", false),
        }];
        for create in [
            "CREATE TABLE t (id INTEGER PRIMARY KEY, a INTEGER CHECK (a > 0), b TEXT)",
            "CREATE TABLE t (id INTEGER PRIMARY KEY, a INTEGER, b TEXT, CONSTRAINT x UNIQUE (a))",
            "CREATE TABLE t (id INTEGER, a INTEGER, b TEXT, UNIQUE (a))",
            "CREATE TABLE t (id INTEGER, a INTEGER, b TEXT, PRIMARY KEY (id))",
        ] {
            let cx = cx_for("t", create, t_cols());
            let err = sqlite(&change, &cx).unwrap_err().to_string();
            assert!(
                err.contains("write this migration by hand"),
                "{create}: {err}"
            );
        }
        let mut cx = cx_for("t", CREATE, t_cols());
        for (action, fails) in [
            ("CASCADE", true),
            ("SET NULL", true),
            ("NO ACTION", false),
            ("RESTRICT", false),
        ] {
            cx.referenced_by.insert(
                "t".into(),
                vec![ForeignKey {
                    table: "child".into(),
                    column: "t_id".into(),
                    references: "t".into(),
                    on_delete: action.into(),
                }],
            );
            assert_eq!(sqlite(&change, &cx).is_err(), fails, "{action}");
        }
        // A column called "checked" is not a CHECK constraint.
        let cx = cx_for(
            "t",
            "CREATE TABLE t (id INTEGER PRIMARY KEY, a INTEGER, checked INTEGER, b TEXT)",
            t_cols(),
        );
        assert!(sqlite(&change, &cx).is_ok());
    }

    #[test]
    fn sqlite_drop_removes_its_indexes() {
        let cx = cx_for("t", CREATE, t_cols());
        let changes = [Change::DropColumn {
            table: "t".into(),
            old: old("a", "INTEGER", true),
        }];
        let (up, down) = sqlite(&changes, &cx).unwrap();
        assert!(!up.contains("t_a_index"));
        assert!(down.contains("CREATE INDEX t_a_index ON t (a);"));
    }

    /// Up then down returns to the starting columns (PostgreSQL only).
    #[tokio::test]
    async fn up_then_down_restores_the_columns_on_postgres() {
        use crate::db::schema_check::table_columns;
        let db = crate::db::connect(&crate::Config::default()).await.unwrap();
        if db.dialect() != Dialect::Postgres {
            return;
        }
        crate::db::script(
            &db,
            "CREATE TABLE rt_posts (id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, \
             a INTEGER NOT NULL, b TEXT, c TEXT NOT NULL DEFAULT 'x', d TEXT);",
        )
        .await
        .unwrap();
        let before = table_columns(&db, "rt_posts").await.unwrap();
        let changes = [
            Change::AddColumn {
                table: "rt_posts".into(),
                column: col("e", ColumnKind::Bool, false).default_sql("false"),
            },
            Change::DropColumn {
                table: "rt_posts".into(),
                old: before.iter().find(|c| c.name == "c").unwrap().clone(),
            },
            Change::AlterColumn {
                table: "rt_posts".into(),
                old: before.iter().find(|c| c.name == "a").unwrap().clone(),
                new: col("a", ColumnKind::Text, true),
            },
            Change::AlterColumn {
                table: "rt_posts".into(),
                old: before.iter().find(|c| c.name == "b").unwrap().clone(),
                new: col("b", ColumnKind::Text, false),
            },
            Change::RenameColumn {
                table: "rt_posts".into(),
                from: "d".into(),
                to: "f".into(),
            },
            Change::CreateIndex {
                table: "rt_posts".into(),
                name: "rt_posts_a_index".into(),
                columns: vec!["a".into()],
                unique: false,
            },
        ];
        let (up, down) = postgres(&changes);
        for text in [&up, &down] {
            for stmt in text.split(";\n").filter(|s| !s.trim().is_empty()) {
                crate::db::sql(stmt).execute(&db).await.unwrap();
            }
            if text == &up {
                let mid = table_columns(&db, "rt_posts").await.unwrap();
                assert_ne!(mid.len(), 0);
                assert!(mid.iter().any(|c| c.name == "f" && c.name != "d"));
            }
        }
        let after = table_columns(&db, "rt_posts").await.unwrap();
        let key = |v: &[crate::db::schema_check::TableColumn]| {
            v.iter()
                .map(|c| (c.name.clone(), c.sql_type.to_lowercase(), c.not_null))
                .collect::<Vec<_>>()
        };
        let mut b = key(&before);
        let mut a = key(&after);
        b.sort();
        a.sort();
        assert_eq!(b, a);
    }

    /// A rebuild keeps rows and indexes, and the down restores the columns
    /// (SQLite only).
    #[tokio::test]
    async fn rebuild_keeps_rows_and_indexes_on_sqlite() {
        use crate::db::schema_check::{
            foreign_keys_to, sqlite_table_sql, table_columns, table_indexes,
        };
        let db = crate::db::connect(&crate::Config::default()).await.unwrap();
        if db.dialect() != Dialect::Sqlite {
            return;
        }
        crate::db::script(
            &db,
            "CREATE TABLE rb_posts (id INTEGER PRIMARY KEY AUTOINCREMENT, a TEXT NOT NULL, \
             b TEXT, c TEXT NOT NULL DEFAULT 'x');\n\
             CREATE INDEX rb_posts_a_index ON rb_posts (a);\n\
             CREATE TABLE rb_comments (id INTEGER PRIMARY KEY, post_id INTEGER NOT NULL \
             REFERENCES rb_posts (id));\n\
             INSERT INTO rb_posts (a, b) VALUES ('1', 'x'), ('2', NULL);\n\
             INSERT INTO rb_comments (post_id) VALUES (1), (2);",
        )
        .await
        .unwrap();
        let before = table_columns(&db, "rb_posts").await.unwrap();
        let mut cx = SqliteContext::default();
        cx.tables.insert(
            "rb_posts".into(),
            TableState {
                table: "rb_posts".into(),
                columns: before.clone(),
                indexes: table_indexes(&db, "rb_posts").await.unwrap(),
            },
        );
        cx.table_sql.insert(
            "rb_posts".into(),
            sqlite_table_sql(&db, "rb_posts").await.unwrap().unwrap(),
        );
        cx.referenced_by.insert(
            "rb_posts".into(),
            foreign_keys_to(&db, "rb_posts").await.unwrap(),
        );
        let changes = [
            Change::AlterColumn {
                table: "rb_posts".into(),
                old: before.iter().find(|c| c.name == "a").unwrap().clone(),
                new: col("a", ColumnKind::BigInt, false),
            },
            Change::AlterColumn {
                table: "rb_posts".into(),
                old: before.iter().find(|c| c.name == "b").unwrap().clone(),
                new: col("b", ColumnKind::Text, false).default_sql("''"),
            },
        ];
        let (up, down) = sqlite(&changes, &cx).unwrap();
        // Migrations run in a transaction, which `defer_foreign_keys` needs.
        crate::db::script(&db, &format!("BEGIN;\n{up}COMMIT;"))
            .await
            .unwrap();
        let mid = table_columns(&db, "rb_posts").await.unwrap();
        assert!(mid.iter().any(|c| c.name == "b" && c.not_null));
        assert_eq!(
            crate::db::sql("SELECT COUNT(*) FROM rb_posts")
                .scalar::<i64>(&db)
                .await
                .unwrap(),
            2
        );
        assert_eq!(table_indexes(&db, "rb_posts").await.unwrap().len(), 1);
        let violations = crate::db::sql("SELECT COUNT(*) FROM pragma_foreign_key_check")
            .scalar::<i64>(&db)
            .await
            .unwrap();
        assert_eq!(violations, 0);
        crate::db::script(&db, &format!("BEGIN;\n{down}COMMIT;"))
            .await
            .unwrap();
        let after = table_columns(&db, "rb_posts").await.unwrap();
        let key = |v: &[TableColumn]| {
            v.iter()
                .map(|c| (c.name.clone(), c.sql_type.to_lowercase(), c.not_null))
                .collect::<Vec<_>>()
        };
        assert_eq!(key(&before), key(&after));
        assert_eq!(table_indexes(&db, "rb_posts").await.unwrap().len(), 1);
    }
}