rustango 0.61.13

//! Apply rendered DDL idempotently, for the `ensure_*_table` helpers.
//!
//! Five subsystems create their own table on first use — audit, TOTP,
//! passkeys, API keys, permissions. Each swallowed the failure by
//! matching `"already exists"` in the error text, which needed the
//! server to speak English (#1642).
//!
//! **The logged-ERROR symptom is PostgreSQL-only.** A failing
//! `CREATE TABLE` writes two lines to PG's server log and none to
//! MySQL's, so the `IF NOT EXISTS` rewrite runs on PG alone: on MySQL
//! it silences nothing and costs 3.4x, because the no-op now succeeds
//! and is binlogged.

use crate::sql::Pool;

/// MySQL `ER_TABLE_EXISTS_ERROR`, `ER_DUP_KEYNAME`, `ER_FK_DUP_NAME`.
/// Error *numbers*, not `SQLSTATE`s — see [`is_already_exists`].
///
/// Reachable from the tests on every feature set, which is the point:
/// the trap it guards does not need the `mysql` feature to explain.
#[cfg_attr(not(feature = "mysql"), allow(dead_code))]
const MYSQL_DUPLICATES: &[u16] = &[1050, 1061, 1826];

/// MySQL `ER_DUP_FIELDNAME`: "already exists" only for an `ADD COLUMN`.
#[cfg_attr(not(feature = "mysql"), allow(dead_code))]
const MYSQL_DUP_FIELDNAME: u16 = 1060;

/// `true` when `number` is MySQL's way of saying the object `stmt`
/// creates is there.
///
/// Split out so a test can pin the trap without a driver error: these
/// are error numbers, and comparing one against `DatabaseError::code()`
/// — which is the `SQLSTATE` — silently never matches.
#[cfg_attr(not(feature = "mysql"), allow(dead_code))]
fn is_mysql_duplicate(number: u16, msg: &str, stmt: &str) -> bool {
    MYSQL_DUPLICATES.contains(&number)
        || (number == MYSQL_DUP_FIELDNAME && is_add_column(stmt))
        || (number == MARIADB_CANT_CREATE && is_add_constraint(stmt) && msg.contains(": 121 "))
}

/// MariaDB reports a taken FK name as `ER_CANT_CREATE_TABLE` with
/// errno 121, not 1826; other errnos (150: bad FK) are real failures.
#[cfg_attr(not(feature = "mysql"), allow(dead_code))]
const MARIADB_CANT_CREATE: u16 = 1005;

fn is_add_constraint(stmt: &str) -> bool {
    stmt.starts_with("ALTER TABLE ") && stmt.contains(" ADD CONSTRAINT ")
}

fn is_add_column(stmt: &str) -> bool {
    stmt.starts_with("ALTER TABLE ") && stmt.contains(" ADD COLUMN ")
}

/// `true` when the error says the object is already there.
///
/// Dispatched on the dialect rather than tried in sequence, so a
/// backend that reports a code gets decided by that code. Falling
/// through to the message after a code said "no" is what let MySQL
/// `ER_DUP_ENTRY` — a genuinely failed unique index — read as success.
///
/// * **Postgres** delegates to [`crate::sql::is_pg_dup_object_error`],
///   which also covers the `23505` concurrent-create race (#1458).
/// * **MySQL** matches `number()`, because its `SQLSTATE`s are far too
///   coarse: `42000` is also a syntax error and TEXT-in-index (#1646).
///   Deliberately *not* [`crate::sql::is_mysql_dup_index_error`], which
///   matches that catch-all.
/// * **SQLite** reports nothing usable either way, so it keeps the text
///   match. Its messages are not localised.
#[cfg_attr(not(feature = "mysql"), allow(unused_variables))]
pub(crate) fn is_already_exists(e: &crate::sql::ExecError, dialect: &str, stmt: &str) -> bool {
    let crate::sql::ExecError::Driver(err) = e else {
        return false;
    };
    match dialect {
        // 42710 on CREATE TABLE is a type of that name: the table is not there.
        "postgres" => crate::sql::is_pg_dup_object_error(err) && !is_type_clash(err, stmt),
        #[cfg(feature = "mysql")]
        "mysql" => err
            .as_database_error()
            .and_then(|db| db.try_downcast_ref::<sqlx::mysql::MySqlDatabaseError>())
            .is_some_and(|my| is_mysql_duplicate(my.number(), my.message(), stmt)),
        // A duplicate column only means "there" for ADD COLUMN; in CREATE TABLE it is a bad DDL.
        _ => {
            let msg = format!("{e}").to_lowercase();
            msg.contains("already exists")
                || (is_add_column(stmt) && msg.contains("duplicate column"))
        }
    }
}

fn is_create_table(stmt: &str) -> bool {
    stmt.starts_with("CREATE TABLE ")
}

/// PG 42710 on a CREATE TABLE: a type holds the name, a racer's or a user's.
fn is_type_clash(err: &sqlx::Error, stmt: &str) -> bool {
    is_create_table(stmt)
        && err.as_database_error().and_then(|d| d.code()).as_deref() == Some("42710")
}

/// `CREATE TABLE x` -> `CREATE TABLE IF NOT EXISTS x`, on Postgres only.
///
/// The point is PG's server log, not capability — all three backends
/// accept the syntax. On MySQL the rewrite turns a cheap client-side
/// error into a successful statement that gets binlogged and fsynced,
/// 165 -> 568 us, to silence a log line MySQL never wrote.
///
/// `ALTER TABLE … ADD COLUMN` gets `IF NOT EXISTS` the same way.
/// `ADD CONSTRAINT` has no `IF NOT EXISTS`, so PG runs it in a `DO` block
/// that catches the duplicate, which logs nothing. A statement
/// holding the tag runs bare, and the client-side check catches it.
fn idempotent<'a>(stmt: &'a str, dialect: &str) -> std::borrow::Cow<'a, str> {
    if dialect != "postgres" {
        return std::borrow::Cow::Borrowed(stmt);
    }
    if is_add_constraint(stmt) && !stmt.contains("$rustango$") {
        return std::borrow::Cow::Owned(format!(
            "DO $rustango$ BEGIN {stmt}; \
             EXCEPTION WHEN duplicate_object OR duplicate_table THEN NULL; END $rustango$"
        ));
    }
    if let Some(rest) = stmt.strip_prefix("CREATE TABLE ") {
        if !rest.starts_with("IF NOT EXISTS") {
            return std::borrow::Cow::Owned(format!("CREATE TABLE IF NOT EXISTS {rest}"));
        }
    }
    if is_add_column(stmt) && !stmt.contains(" ADD COLUMN IF NOT EXISTS ") {
        if let Some((head, tail)) = stmt.split_once(" ADD COLUMN ") {
            return std::borrow::Cow::Owned(format!("{head} ADD COLUMN IF NOT EXISTS {tail}"));
        }
    }
    std::borrow::Cow::Borrowed(stmt)
}

/// Create every table in `snapshot`, tolerating objects that already
/// exist. The one entry point the `ensure_*` helpers share, so the
/// policy cannot drift between them.
///
/// FK targets are pinned to the schema the tables are created in, so a
/// missing target fails instead of binding to another schema (#1645).
///
/// # Errors
/// Any driver failure that is not "already exists".
pub(crate) async fn apply_idempotent(
    pool: &Pool,
    snapshot: &super::SchemaSnapshot,
) -> Result<(), sqlx::Error> {
    let changes = super::detect_changes(&super::SchemaSnapshot::default(), snapshot);
    apply_changes_idempotent(pool, snapshot, &changes).await
}

/// The statements [`apply_idempotent`] runs for `snapshot`, guards
/// included, FK targets unqualified, for a `--dry-run`.
///
/// # Errors
/// A render failure.
pub(crate) fn idempotent_ddl(
    snapshot: &super::SchemaSnapshot,
    dialect: &dyn crate::sql::Dialect,
) -> Result<Vec<String>, String> {
    let changes = super::detect_changes(&super::SchemaSnapshot::default(), snapshot);
    let batch = super::diff::render_changes_split_in_schema(&changes, snapshot, dialect, None)?;
    Ok(as_run(
        batch.immediate.iter().chain(&batch.deferred_fks),
        dialect.name(),
    )
    .collect())
}

/// `stmts` as [`run_statements`] sends them, for a `--dry-run`.
pub(crate) fn as_run<'a, I>(stmts: I, dialect: &'a str) -> impl Iterator<Item = String> + 'a
where
    I: IntoIterator<Item = &'a String>,
    I::IntoIter: 'a,
{
    stmts
        .into_iter()
        .map(move |s| idempotent(s, dialect).into_owned())
}

/// Add each `(table, column)` of `snapshot` that a table created by an
/// older release lacks. Never creates the table.
///
/// # Errors
/// Any driver failure that is not "already exists", including a
/// missing table.
#[cfg(all(feature = "admin", feature = "totp"))]
pub(crate) async fn add_columns_idempotent(
    pool: &Pool,
    snapshot: &super::SchemaSnapshot,
    columns: &[(&str, &str)],
) -> Result<(), sqlx::Error> {
    let changes: Vec<_> = columns
        .iter()
        .map(|(table, column)| super::SchemaChange::AddColumn {
            table: (*table).to_owned(),
            column: (*column).to_owned(),
        })
        .collect();
    apply_changes_idempotent(pool, snapshot, &changes).await
}

/// Render `changes` against `snapshot` and run them, tolerating objects
/// that already exist.
///
/// # Errors
/// A render failure, or any driver failure that is not "already exists".
pub(crate) async fn apply_changes_idempotent(
    pool: &Pool,
    snapshot: &super::SchemaSnapshot,
    changes: &[super::SchemaChange],
) -> Result<(), sqlx::Error> {
    let schema = creation_schema(pool).await?;
    apply_changes_idempotent_in(pool, snapshot, changes, schema.as_deref()).await
}

/// [`apply_changes_idempotent`] into `schema` rather than the current one.
///
/// # Errors
/// As [`apply_changes_idempotent`].
pub(crate) async fn apply_changes_idempotent_in(
    pool: &Pool,
    snapshot: &super::SchemaSnapshot,
    changes: &[super::SchemaChange],
    schema: Option<&str>,
) -> Result<(), sqlx::Error> {
    let batch =
        super::diff::render_changes_split_in_schema(changes, snapshot, pool.dialect(), schema)
            .map_err(sqlx::Error::Protocol)?;
    apply_batch(pool, &batch).await
}

/// Column names `table` has in the schema new tables land in.
///
/// # Errors
/// Driver failures from the catalog read.
pub(crate) async fn live_columns(
    pool: &Pool,
    table: &str,
) -> Result<std::collections::HashSet<String>, super::MigrateError> {
    Ok(live_column_types(pool, table).await?.into_keys().collect())
}

/// [`live_columns`] with each column's lowercased backend type token
/// (`udt_name` on PG, `data_type` on MySQL, the declaration on SQLite).
///
/// # Errors
/// Driver failures from the catalog read.
pub(crate) async fn live_column_types(
    pool: &Pool,
    table: &str,
) -> Result<std::collections::HashMap<String, String>, super::MigrateError> {
    // MySQL reads `""` as `DATABASE()`; SQLite ignores the schema.
    let schema = creation_schema(pool).await?.unwrap_or_default();
    Ok(super::inspectdb::list_columns(pool, &schema, table)
        .await?
        .into_iter()
        .map(|c| (c.name, c.udt_name.to_ascii_lowercase()))
        .collect())
}

/// Where Postgres creates unqualified tables; `None` on backends
/// without schemas.
pub(crate) async fn creation_schema(pool: &Pool) -> Result<Option<String>, sqlx::Error> {
    match pool {
        #[cfg(feature = "postgres")]
        Pool::Postgres(pg) => sqlx::query_scalar::<_, Option<String>>("SELECT current_schema()")
            .fetch_one(pg)
            .await?
            .map(Some)
            .ok_or_else(|| sqlx::Error::Protocol("search_path names no existing schema".into())),
        #[allow(unreachable_patterns)]
        _ => Ok(None),
    }
}

/// Apply each group of `changes` on its own, FKs last, so one object that
/// fails does not block the others. Returns what could not be applied.
///
/// # Errors
/// Only the catalog read for the creation schema.
pub(crate) async fn converge_groups(
    pool: &Pool,
    snapshot: &super::SchemaSnapshot,
    groups: &[Vec<super::SchemaChange>],
) -> Result<Vec<String>, sqlx::Error> {
    let schema = creation_schema(pool).await?;
    let mut failed = Vec::new();
    let mut fks = Vec::new();
    for Planned {
        label,
        batch,
        retry,
    } in plan_groups(pool, snapshot, groups, schema.as_deref()).await?
    {
        let batch = match batch {
            Ok(b) => b,
            Err(e) => {
                failed.push(format!("{label}: {e}"));
                continue;
            }
        };
        let Err(e) = run_statements(pool, &batch.immediate).await else {
            fks.push((label, batch.deferred_fks));
            continue;
        };
        // SQLite refuses a non-constant DEFAULT on a table with rows.
        match retry.filter(|_| is_non_constant_default(&e)) {
            Some(Ok(r)) if run_statements(pool, &r.batch.immediate).await.is_ok() => {
                tracing::warn!(target: "rustango::migrate", "{label}: {}", r.warning);
            }
            _ => failed.push(format!("{label}: {e}")),
        }
    }
    for (label, stmts) in fks {
        if let Err(e) = run_statements(pool, &stmts).await {
            failed.push(format!("{label}: {e}"));
        }
    }
    Ok(failed)
}

/// One group [`converge_groups`] runs: its SQL, or why `migrate` refuses it.
pub(crate) struct Planned {
    pub(crate) label: String,
    pub(crate) batch: Result<super::diff::RenderedBatch, String>,
    /// What runs instead when SQLite refuses the batch's DEFAULT.
    pub(crate) retry: Option<Result<FilledTableRetry, String>>,
}

/// How [`converge_groups`] renders `groups` into `schema`; reads only, so a
/// `--dry-run` prints the same plan.
///
/// # Errors
/// Only the row probe for a NOT NULL column.
pub(crate) async fn plan_groups(
    pool: &Pool,
    snapshot: &super::SchemaSnapshot,
    groups: &[Vec<super::SchemaChange>],
    schema: Option<&str>,
) -> Result<Vec<Planned>, sqlx::Error> {
    use super::SchemaChange as SC;
    let mut out = Vec::new();
    for group in groups {
        let label = match group.first() {
            Some(SC::AddColumn { table, column }) => format!("column `{table}.{column}`"),
            Some(SC::CreateTable(t) | SC::CreateM2MTable { through: t, .. }) => {
                format!("table `{t}`")
            }
            Some(SC::CreateIndex { name, .. }) => format!("index `{name}`"),
            Some(SC::AlterFkOnDelete { table, column, .. }) => {
                format!("foreign key `{table}.{column}`")
            }
            Some(SC::AddCompositeFk { table, name, .. }) => format!("foreign key `{table}.{name}`"),
            _ => continue,
        };
        let mut no_rows = false;
        if let Some(SC::AddColumn { table, column }) = group.first() {
            let f = snapshot.table(table).and_then(|t| t.field(column));
            if f.is_some_and(|f| !f.nullable && f.default.is_none() && !f.primary_key) {
                // An empty table has no rows to fill (#2066).
                no_rows = is_empty(pool, table).await?;
                if !no_rows {
                    let batch = Err("NOT NULL with no default, so rows have no value".to_owned());
                    out.push(Planned {
                        label,
                        batch,
                        retry: None,
                    });
                    continue;
                }
            }
        }
        let render = if no_rows {
            super::diff::render_changes_split_for_empty
        } else {
            super::diff::render_changes_split_in_schema
        };
        let batch = match render(group, snapshot, pool.dialect(), schema) {
            // Only the runner can rebuild a table; say so rather than skip it.
            Ok(b) if b.rebuild.is_some() => Err("needs a table rebuild".to_owned()),
            r => r,
        };
        let retry = filled_table_retry(pool.dialect(), snapshot, group, |s| {
            super::diff::render_changes_split_in_schema(group, s, pool.dialect(), schema)
        })
        .filter(|_| batch.is_ok() && !no_rows);
        out.push(Planned {
            label,
            batch,
            retry,
        });
    }
    Ok(out)
}

/// Whether `table` has no rows.
async fn is_empty(pool: &Pool, table: &str) -> Result<bool, sqlx::Error> {
    has_no_values(pool, table, None).await
}

/// Whether `table` has no rows, or none with `column` set.
pub(super) async fn has_no_values(
    pool: &Pool,
    table: &str,
    column: Option<&str>,
) -> Result<bool, sqlx::Error> {
    let q = |n: &str| pool.dialect().quote_ident(n);
    let filter = column.map_or(String::new(), |c| format!(" WHERE {} IS NOT NULL", q(c)));
    let sql = format!(
        "SELECT COUNT(*) FROM (SELECT 1 AS one FROM {}{filter} LIMIT 1) AS probe",
        q(table)
    );
    let rows: Vec<(i64,)> = crate::sql::raw_query_pool(&sql, Vec::new(), pool)
        .await
        .map_err(|e| match e {
            crate::sql::ExecError::Driver(d) => d,
            other => sqlx::Error::Protocol(other.to_string()),
        })?;
    Ok(rows.first().is_some_and(|(n,)| *n == 0))
}

/// Whether two rows of `table` share a non-NULL value of every `columns`.
pub(super) async fn has_duplicate_keys(
    pool: &Pool,
    table: &str,
    columns: &[&str],
) -> Result<bool, sqlx::Error> {
    let q = |n: &str| pool.dialect().quote_ident(n);
    let cols: Vec<String> = columns.iter().map(|c| q(c)).collect();
    let not_null: Vec<String> = cols.iter().map(|c| format!("{c} IS NOT NULL")).collect();
    let sql = format!(
        "SELECT COUNT(*) FROM (SELECT 1 AS one FROM {} WHERE {} GROUP BY {} \
         HAVING COUNT(*) > 1 LIMIT 1) AS probe",
        q(table),
        not_null.join(" AND "),
        cols.join(", "),
    );
    let rows: Vec<(i64,)> = crate::sql::raw_query_pool(&sql, Vec::new(), pool)
        .await
        .map_err(|e| match e {
            crate::sql::ExecError::Driver(d) => d,
            other => sqlx::Error::Protocol(other.to_string()),
        })?;
    Ok(rows.first().is_some_and(|(n,)| *n > 0))
}

/// SQLite's error for a non-constant DEFAULT added to a table with rows.
pub(super) fn is_non_constant_default(e: &sqlx::Error) -> bool {
    e.as_database_error()
        .is_some_and(|d| d.message().contains("non-constant default"))
}

/// A rendering of a SQLite AddColumn its non-constant DEFAULT refused.
pub(crate) struct FilledTableRetry {
    pub(super) batch: super::diff::RenderedBatch,
    pub(super) warning: &'static str,
}

impl FilledTableRetry {
    /// Comment lines for a `--dry-run`, which cannot know if the table has rows.
    pub(super) fn preview(&self) -> impl Iterator<Item = String> + '_ {
        std::iter::once(format!(
            "if the table has rows SQLite refuses that DEFAULT, and `migrate` runs this \
             instead ({}):",
            self.warning
        ))
        .chain(self.batch.immediate.iter().map(|s| format!("  {s}")))
    }
}

/// The AddColumn of `group` for a SQLite table with rows (#2017): `now()`
/// frozen to the current time, or a UUID column added without its DEFAULT
/// and backfilled per row. The ORM binds both columns on insert.
pub(super) fn filled_table_retry(
    dialect: &dyn crate::sql::Dialect,
    snapshot: &super::SchemaSnapshot,
    group: &[super::SchemaChange],
    render: impl Fn(&super::SchemaSnapshot) -> Result<super::diff::RenderedBatch, String>,
) -> Option<Result<FilledTableRetry, String>> {
    #[cfg(feature = "sqlite")]
    if let ("sqlite", [super::SchemaChange::AddColumn { table, column }]) = (dialect.name(), group)
    {
        let mut t = snapshot.table(table)?.clone();
        let field = t.fields.iter_mut().find(|f| &f.column == column)?;
        let mut backfill = None;
        let warning = if field
            .default
            .as_deref()
            .is_some_and(crate::sql::is_now_expr)
        {
            let now = crate::sql::encode_datetime(chrono::Utc::now());
            field.default = Some(format!("'{now}'"));
            "added with a fixed DEFAULT: SQLite can't add `now()` to a table with rows"
        } else if super::diff::is_uuid_default(field) {
            backfill = Some(super::diff::fill_nulls_sql(table, field, dialect));
            field.default = None;
            field.nullable = true;
            "added nullable with no DEFAULT: SQLite can't add a UUID DEFAULT to a table with rows"
        } else {
            return None;
        };
        let snap = super::SchemaSnapshot {
            tables: vec![t],
            ..Default::default()
        };
        return Some(render(&snap).map(|mut batch| {
            batch.immediate.extend(backfill);
            FilledTableRetry { batch, warning }
        }));
    }
    let _ = (dialect, snapshot, group, render);
    None
}

/// Run every statement in `batch`, tolerating objects that already exist.
async fn apply_batch(pool: &Pool, batch: &super::RenderedBatch) -> Result<(), sqlx::Error> {
    run_statements(pool, &batch.immediate).await?;
    run_statements(pool, &batch.deferred_fks).await
}

/// Run `stmts` in order, tolerating objects that already exist.
pub(crate) async fn run_statements(pool: &Pool, stmts: &[String]) -> Result<(), sqlx::Error> {
    let dialect = pool.dialect().name();
    for stmt in stmts {
        let stmt = idempotent(stmt, dialect);
        let mut res = crate::sql::raw_execute_pool(pool, &stmt, Vec::new()).await;
        // A racer's committed row type also says 42710; once in, the retry sees its table.
        if dialect == "postgres"
            && matches!(&res, Err(crate::sql::ExecError::Driver(e)) if is_type_clash(e, &stmt))
        {
            res = crate::sql::raw_execute_pool(pool, &stmt, Vec::new()).await;
        }
        if let Err(e) = res {
            if is_already_exists(&e, dialect, &stmt) {
                continue;
            }
            return Err(match e {
                crate::sql::ExecError::Driver(err) => err,
                other => sqlx::Error::Protocol(format!("{other}")),
            });
        }
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::{idempotent, is_mysql_duplicate};

    /// A CREATE TABLE that failed for another reason is not "already there".
    #[cfg(feature = "sqlite")]
    #[tokio::test]
    async fn a_duplicate_column_in_create_table_is_an_error() {
        let pool = crate::sql::Pool::connect("sqlite::memory:").await.unwrap();
        let bad = ["CREATE TABLE t (a INTEGER, a INTEGER)".to_owned()];
        assert!(super::run_statements(&pool, &bad).await.is_err());
        let add = [
            "CREATE TABLE u (a INTEGER)".to_owned(),
            "ALTER TABLE u ADD COLUMN a INTEGER".to_owned(),
        ];
        super::run_statements(&pool, &add)
            .await
            .expect("an existing column is there");
    }

    /// A table named like an existing PG type fails with 42710; it is not there.
    #[cfg(feature = "postgres")]
    #[tokio::test]
    async fn a_table_named_like_a_type_is_an_error_on_postgres() {
        let Ok(url) = std::env::var("DATABASE_URL") else {
            return;
        };
        if !url.starts_with("postgres") {
            return;
        }
        let pool = crate::sql::Pool::connect(&url).await.expect("postgres");
        let run = |sql: &'static str| crate::sql::raw_execute_pool(&pool, sql, Vec::new());
        run("DROP TABLE IF EXISTS ensure_type_clash").await.unwrap();
        run("DROP TYPE IF EXISTS ensure_type_clash").await.unwrap();
        run("CREATE TYPE ensure_type_clash AS ENUM ('a')")
            .await
            .unwrap();
        let create = ["CREATE TABLE ensure_type_clash (id INTEGER)".to_owned()];
        let res = super::run_statements(&pool, &create).await;
        run("DROP TYPE ensure_type_clash").await.unwrap();
        assert!(res.is_err(), "the table was never created");
    }

    /// Only the refused non-constant DEFAULT is retried, not any error.
    #[cfg(feature = "sqlite")]
    #[tokio::test]
    async fn only_a_non_constant_default_is_retried() {
        let pool = sqlx::SqlitePool::connect("sqlite::memory:").await.unwrap();
        for sql in [
            "CREATE TABLE t (id INTEGER PRIMARY KEY)",
            "INSERT INTO t VALUES (1)",
        ] {
            sqlx::query(sql).execute(&pool).await.unwrap();
        }
        let run = |sql| sqlx::query(sql).execute(&pool);
        let refused = run("ALTER TABLE t ADD COLUMN c BLOB DEFAULT (randomblob(16))").await;
        assert!(super::is_non_constant_default(&refused.unwrap_err()));
        let other = run("ALTER TABLE t ADD COLUMN id BLOB DEFAULT (randomblob(16))").await;
        assert!(!super::is_non_constant_default(&other.unwrap_err()));
    }

    #[test]
    fn create_table_gains_if_not_exists_on_postgres() {
        assert_eq!(
            idempotent(r#"CREATE TABLE "t" ("id" BIGSERIAL)"#, "postgres"),
            r#"CREATE TABLE IF NOT EXISTS "t" ("id" BIGSERIAL)"#
        );
    }

    /// MySQL never had the logged ERROR this rewrite exists to silence,
    /// and on MySQL the rewritten statement succeeds and is binlogged —
    /// 3.4x slower for nothing.
    #[test]
    fn other_dialects_are_left_alone() {
        let s = r#"CREATE TABLE "t" ("id" BIGSERIAL)"#;
        assert_eq!(idempotent(s, "mysql"), s);
        assert_eq!(idempotent(s, "sqlite"), s);
    }

    #[test]
    fn already_guarded_create_is_left_alone() {
        let s = r#"CREATE TABLE IF NOT EXISTS "t" ("id" BIGSERIAL)"#;
        assert_eq!(idempotent(s, "postgres"), s, "must not double up the guard");
    }

    #[test]
    fn non_create_table_statements_are_untouched() {
        // PG catches a duplicate constraint in a `DO` block; the others
        // still rely on the code check.
        let s =
            r#"ALTER TABLE "a" ADD CONSTRAINT "a_b_fkey" FOREIGN KEY ("b") REFERENCES "b" ("id")"#;
        let wrapped = idempotent(s, "postgres");
        assert!(
            wrapped.starts_with("DO $rustango$ BEGIN ALTER TABLE"),
            "{wrapped}"
        );
        // Exactly the duplicate codes: a broader catch hides real failures.
        assert!(
            wrapped.ends_with(
                "EXCEPTION WHEN duplicate_object OR duplicate_table THEN NULL; END $rustango$"
            ),
            "{wrapped}"
        );
        assert_eq!(idempotent(s, "mysql"), s);
        let idx = r#"CREATE UNIQUE INDEX "i" ON "t" ("a")"#;
        assert_eq!(idempotent(idx, "postgres"), idx);
    }

    /// A statement holding the tag would end the `DO` body early.
    #[test]
    fn a_statement_holding_the_do_tag_runs_bare() {
        let s = r#"ALTER TABLE "a" ADD CONSTRAINT "ck" CHECK ("b" <> '$rustango$')"#;
        assert_eq!(idempotent(s, "postgres"), s);
    }

    /// The dry-run prints the statements as they run, guards included.
    #[test]
    fn dry_run_ddl_is_the_rewritten_ddl() {
        use crate::core::Model as _;
        let snapshot = crate::migrate::SchemaSnapshot::from_models(&[
            crate::contenttypes::ContentType::SCHEMA,
        ]);
        let ddl = super::idempotent_ddl(&snapshot, &crate::sql::Postgres).unwrap();
        let creates: Vec<_> = ddl.iter().filter(|s| s.contains("CREATE TABLE")).collect();
        assert!(!creates.is_empty());
        for s in creates {
            assert!(s.starts_with("CREATE TABLE IF NOT EXISTS"), "{s}");
        }
        let mysql = super::idempotent_ddl(&snapshot, &crate::sql::MySql).unwrap();
        assert!(
            mysql.iter().all(|s| !s.contains("IF NOT EXISTS")),
            "{mysql:?}"
        );
    }

    #[test]
    fn add_column_gains_if_not_exists_on_postgres_only() {
        let s = r#"ALTER TABLE "t" ADD COLUMN "c" BIGINT"#;
        let guarded = r#"ALTER TABLE "t" ADD COLUMN IF NOT EXISTS "c" BIGINT"#;
        assert_eq!(idempotent(s, "postgres"), guarded);
        assert_eq!(idempotent(guarded, "postgres"), guarded);
        assert_eq!(idempotent(s, "mysql"), s);
    }

    /// The tests above feed in hand-written SQL, so they would pass even
    /// if the renderer stopped emitting the shape `idempotent` looks
    /// for. This one runs a real batch through the real renderer, which
    /// is the coupling that actually has to hold.
    ///
    /// `ContentType` rather than a tenancy model on purpose: `migrate`
    /// is ungated, and naming a `#[cfg]`-gated module here broke every
    /// build without that feature.
    #[test]
    fn the_renderer_output_is_actually_rewritten() {
        use crate::core::Model as _;
        let snapshot = crate::migrate::SchemaSnapshot::from_models(&[
            crate::contenttypes::ContentType::SCHEMA,
        ]);
        let changes =
            crate::migrate::detect_changes(&crate::migrate::SchemaSnapshot::default(), &snapshot);
        let batch = crate::migrate::render_changes_split_with_dialect(
            &changes,
            &snapshot,
            &crate::sql::Postgres,
        )
        .expect("render");

        let creates: Vec<_> = batch
            .immediate
            .iter()
            .filter(|s| s.contains("CREATE TABLE"))
            .collect();
        assert!(!creates.is_empty(), "expected at least one CREATE TABLE");
        for stmt in creates {
            assert!(
                idempotent(stmt, "postgres").starts_with("CREATE TABLE IF NOT EXISTS"),
                "renderer emits a shape `idempotent` does not rewrite, so the \
                 ensure paths are back to erroring per call: {stmt}"
            );
        }
    }

    /// `apply_batch` itself, against a real SQLite database.
    ///
    /// Written because the review found three reverts of it that every
    /// other test survived: dropping the `idempotent` rewrite, dropping
    /// `.chain(deferred_fks)` so FK statements never run, and making
    /// `is_already_exists` return `true` so every failure is silent.
    /// Each assertion below kills one of them.
    #[cfg(feature = "sqlite")]
    #[tokio::test]
    async fn apply_idempotent_runs_both_lists_and_still_propagates() {
        let pool = crate::sql::Pool::connect("sqlite::memory:")
            .await
            .expect("sqlite");

        // `deferred_fks` must run too — this table only exists if it does.
        let batch = super::super::RenderedBatch {
            immediate: vec!["CREATE TABLE a (id INTEGER PRIMARY KEY)".to_owned()],
            deferred_fks: vec!["CREATE TABLE b (id INTEGER PRIMARY KEY)".to_owned()],
            ..Default::default()
        };
        super::apply_batch(&pool, &batch)
            .await
            .expect("first run creates both");
        for t in ["a", "b"] {
            crate::sql::raw_execute_pool(&pool, &format!("SELECT 1 FROM {t}"), Vec::new())
                .await
                .unwrap_or_else(|e| panic!("{t} was never created — deferred_fks skipped? {e}"));
        }

        // Second run is a no-op, not an error.
        super::apply_batch(&pool, &batch)
            .await
            .expect("re-running an ensure must be idempotent");

        // A failure that is NOT "already exists" must still surface.
        let bad = super::super::RenderedBatch {
            immediate: vec!["CREATE TABLE c (id INTEGER PRIMARY KEY".to_owned()],
            ..Default::default()
        };
        assert!(
            super::apply_batch(&pool, &bad).await.is_err(),
            "a syntax error must propagate; swallowing everything would \
             make every ensure silently succeed"
        );
    }

    /// A MySQL error *number* is not a `SQLSTATE`. An earlier draft
    /// compared these against `DatabaseError::code()`, which returns the
    /// `SQLSTATE`, so it could never match — and a text fallback hid it.
    #[test]
    fn mysql_duplicates_are_numbers() {
        let create = "CREATE TABLE t (id INT)";
        assert!(
            is_mysql_duplicate(1050, "", create),
            "ER_TABLE_EXISTS_ERROR"
        );
        assert!(is_mysql_duplicate(1061, "", create), "ER_DUP_KEYNAME");
        assert!(is_mysql_duplicate(1826, "", create), "ER_FK_DUP_NAME");
        // ER_DUP_ENTRY: a unique index that genuinely could not be
        // built. It must propagate, not read as "already exists".
        assert!(
            !is_mysql_duplicate(1062, "", create),
            "ER_DUP_ENTRY must not swallow"
        );
        assert!(
            !is_mysql_duplicate(1064, "", create),
            "syntax error must not swallow"
        );
        assert!(
            !is_mysql_duplicate(1170, "", create),
            "TEXT-in-index must not swallow"
        );
    }

    /// `ER_DUP_FIELDNAME` means "already there" only for `ADD COLUMN`; a
    /// CREATE TABLE that names a column twice is a real error.
    #[test]
    fn mysql_dup_fieldname_only_for_add_column() {
        assert!(is_mysql_duplicate(
            1060,
            "",
            "ALTER TABLE `t` ADD COLUMN `c` BIGINT"
        ));
        assert!(!is_mysql_duplicate(
            1060,
            "",
            "CREATE TABLE `t` (`c` INT, `c` INT)"
        ));
    }

    /// MariaDB 11 answers a re-run FK with 1005 errno 121, so a second
    /// `ensure_*` call failed there; errno 150 (bad FK) must still fail.
    #[test]
    fn mariadb_dup_fk_name_is_already_exists() {
        let fk =
            "ALTER TABLE `c` ADD CONSTRAINT `c_p_fkey` FOREIGN KEY (`p`) REFERENCES `p` (`id`)";
        let dup = "Can't create table `db`.`c` (errno: 121 \"Duplicate key on write or update\")";
        let bad = "Can't create table `db`.`c` (errno: 150 \"Foreign key constraint is incorrectly formed\")";
        assert!(is_mysql_duplicate(1005, dup, fk));
        assert!(!is_mysql_duplicate(1005, bad, fk));
        assert!(!is_mysql_duplicate(
            1005,
            dup,
            "CREATE TABLE `c` (`id` INT)"
        ));
    }
}