architect_sdk/

migration.rs

//! Apply config to the database: DDL for schemas, enums, tables, indexes, and foreign keys.
//! Order follows PostgreSQL dependencies (see docs/postgres-config-schema.md § 3.5).

use crate::config::types::*;
use crate::config::{validate, FullConfig};
use crate::db::parse_canonical;
use crate::db::pool::Pool;
use crate::db::{ColumnFacts, DbSnapshot, Dialect};
use crate::error::AppError;
use serde::{Deserialize, Serialize};
use std::collections::{HashMap, HashSet};

fn quote(s: &str) -> String {
    format!("\"{}\"", s.replace('\\', "\\\\").replace('"', "\\\""))
}

/// Name of the column added to app tables when RLS is enabled. Used by migration and CRUD.
pub const RLS_TENANT_COLUMN: &str = "tenant_id";

/// Ensure every table in `config` has the RLS tenant column, row-level security enabled, and the
/// four tenant-isolation policies. Tables flagged `global` instead get asymmetric policies — every
/// tenant may read (`SELECT USING (true)`) but only the Platform Admin tenant may write — so they
/// hold data shared across all RLS tenants. Fully idempotent (`ADD COLUMN IF NOT EXISTS`,
/// `DROP POLICY IF EXISTS` then `CREATE POLICY`), so it is safe to run on both fresh installs and
/// upgrades. On upgrades this is the path that backfills the tenant column + policies for newly
/// created tables, which the diff-based migration plan does not touch.
pub async fn apply_rls_to_tables(
    pool: &Pool,
    config: &FullConfig,
    schema_override: Option<&str>,
    rls_tenant_column: &str,
    dialect: &dyn Dialect,
) -> Result<(), AppError> {
    if !dialect.supports_rls() {
        tracing::warn!(dialect = %dialect.name(), "RLS requested but not supported by this dialect; skipping");
        return Ok(());
    }
    let default_sid = config.schemas.first().map(|s| s.id.as_str()).unwrap_or("");
    let platform_id = crate::tenant::platform_tenant_id();
    let schemas_by_id: HashMap<_, _> = config.schemas.iter().map(|s| (s.id.as_str(), s)).collect();
    let columns_by_table: HashMap<_, Vec<&ColumnConfig>> =
        config.columns.iter().fold(HashMap::new(), |mut m, c| {
            m.entry(c.table_id.as_str()).or_default().push(c);
            m
        });
    let col = rls_tenant_column;

    for t in &config.tables {
        let sid = t.schema_id.as_deref().unwrap_or(default_sid);
        let schema = match schemas_by_id.get(sid) {
            Some(s) => s,
            None => continue,
        };
        let schema_name = quote(schema_override.unwrap_or(&schema.name));
        let full_name = format!("{}.{}", schema_name, quote(&t.name));
        let config_col_names: HashSet<&str> = columns_by_table
            .get(t.id.as_str())
            .map(|v| v.iter().map(|c| c.name.as_str()).collect())
            .unwrap_or_default();

        if !config_col_names.contains(col) {
            let add_col = format!(
                "ALTER TABLE {} ADD COLUMN IF NOT EXISTS {} TEXT",
                full_name,
                quote(col)
            );
            sqlx::query(&add_col).execute(pool).await?;
        }
        let enable_rls = format!("ALTER TABLE {} ENABLE ROW LEVEL SECURITY", full_name);
        sqlx::query(&enable_rls).execute(pool).await?;
        let q_col = quote(col);
        let setting = "current_setting('app.tenant_id', true)";
        // Tenant-isolated condition: a row belongs to the caller's tenant.
        let tenant_cond = format!("{} = {}", q_col, setting);
        // Platform-admin condition: the caller's session is the Platform Admin tenant. Keys off the
        // session setting, not a row column, so it gates writes regardless of the row's tenant_id.
        let admin_cond = format!("{} = '{}'", setting, platform_id.replace('\'', "''"));
        let read_all = "true".to_string();
        let policy_prefix = format!("rls_tenant_{}", t.name);
        // Global tables: everyone reads (USING true), only the Platform Admin writes.
        // Tenant tables: every operation is scoped to the caller's tenant_id.
        let policies: Vec<(&str, &str, Option<&str>, Option<&str>)> = if t.global {
            vec![
                ("select", "SELECT", Some(read_all.as_str()), None),
                ("insert", "INSERT", None, Some(admin_cond.as_str())),
                (
                    "update",
                    "UPDATE",
                    Some(admin_cond.as_str()),
                    Some(admin_cond.as_str()),
                ),
                ("delete", "DELETE", Some(admin_cond.as_str()), None),
            ]
        } else {
            vec![
                ("select", "SELECT", Some(tenant_cond.as_str()), None),
                ("insert", "INSERT", None, Some(tenant_cond.as_str())),
                (
                    "update",
                    "UPDATE",
                    Some(tenant_cond.as_str()),
                    Some(tenant_cond.as_str()),
                ),
                ("delete", "DELETE", Some(tenant_cond.as_str()), None),
            ]
        };
        for (suffix, cmd, using_cond, with_check) in policies.iter() {
            let policy_name = format!("{}_{}", policy_prefix, suffix);
            let drop_sql = format!(
                "DROP POLICY IF EXISTS {} ON {}",
                quote(&policy_name),
                full_name
            );
            let _ = sqlx::query(&drop_sql).execute(pool).await;
            let create_sql = match (using_cond, with_check) {
                (Some(u), Some(w)) => format!(
                    "CREATE POLICY {} ON {} FOR {} USING ( {} ) WITH CHECK ( {} )",
                    quote(&policy_name),
                    full_name,
                    cmd,
                    u,
                    w
                ),
                (Some(u), None) => format!(
                    "CREATE POLICY {} ON {} FOR {} USING ( {} )",
                    quote(&policy_name),
                    full_name,
                    cmd,
                    u
                ),
                (None, Some(w)) => format!(
                    "CREATE POLICY {} ON {} FOR {} WITH CHECK ( {} )",
                    quote(&policy_name),
                    full_name,
                    cmd,
                    w
                ),
                (None, None) => continue,
            };
            sqlx::query(&create_sql).execute(pool).await?;
        }
    }
    Ok(())
}

/// Add any of `desired` columns that the table does not have yet.
///
/// `CREATE TABLE IF NOT EXISTS` is a no-op on a table that already exists, so a table created by
/// an earlier version of a package never picks up columns added since. This closes that gap and
/// makes install/bootstrap idempotent in the same way an upgrade plan is.
///
/// Best-effort by design: a column that cannot be added (most often `NOT NULL` with no default on
/// a table that already has rows) is retried as nullable and then logged, never returned as an
/// error — refusing to install because of pre-existing drift helps nobody.
async fn add_missing_columns(
    pool: &Pool,
    snapshot: &DbSnapshot,
    dialect: &dyn Dialect,
    schema: &str,
    table: &str,
    desired: &[(String, String)],
) {
    if !snapshot.introspected {
        return;
    }
    let full = format!("{}.{}", quote(schema), quote(table));
    for (name, def) in desired {
        if snapshot.has_column(schema, table, name) {
            continue;
        }
        let col_def = format!("{} {}", quote(name), def);
        let sql = add_column_ddl(dialect, &full, &col_def);
        match sqlx::query(&sql).execute(pool).await {
            Ok(_) => {
                tracing::info!(schema, table, column = %name, "added missing column to pre-existing table");
            }
            Err(e) => {
                // NOT NULL without a default cannot be added to a populated table; fall back to a
                // nullable column so reads and writes of that field at least work.
                let upper = def.to_uppercase();
                if upper.contains("NOT NULL") && !upper.contains("DEFAULT") {
                    let relaxed_def = def.replace("NOT NULL", "").trim().to_string();
                    let retry =
                        add_column_ddl(dialect, &full, &format!("{} {}", quote(name), relaxed_def));
                    if sqlx::query(&retry).execute(pool).await.is_ok() {
                        tracing::warn!(schema, table, column = %name, "added missing column as NULLABLE — NOT NULL could not be applied to the existing table");
                        continue;
                    }
                }
                tracing::warn!(schema, table, column = %name, error = %e, "could not add missing column to pre-existing table");
            }
        }
    }
}

/// Source columns as a companion (`_audit` / `_history`) table replicates them: same type,
/// always nullable, no constraints. Mirrors `audit_table_ddl` / `history_table_ddl`.
fn companion_source_columns(
    source_cols: &[&ColumnConfig],
    dialect: &dyn Dialect,
) -> Vec<(String, String)> {
    let mut out: Vec<(String, String)> = source_cols
        .iter()
        .map(|c| (c.name.clone(), dialect.ddl_type(&parse_canonical(&c.type_))))
        .collect();
    let config_col_names: HashSet<&str> = source_cols.iter().map(|c| c.name.as_str()).collect();
    let audit_ts = dialect.audit_timestamp_type();
    for (name, typ) in [
        ("created_at", audit_ts),
        ("updated_at", audit_ts),
        ("archived_at", audit_ts),
        ("created_by", "TEXT"),
        ("updated_by", "TEXT"),
    ] {
        if !config_col_names.contains(name) {
            out.push((name.to_string(), typ.to_string()));
        }
    }
    out
}

/// Apply full config to the database: CREATE SCHEMA, CREATE TYPE, CREATE TABLE, CREATE INDEX, ADD FK.
/// Validates config first. Idempotent for schemas and types (IF NOT EXISTS); tables are CREATE TABLE only (fails if exists).
/// When `schema_override` is `Some(s)`, app tables/indexes/FKs are created in schema `s` instead of config schema names (e.g. for schema-strategy tenants).
/// When `rls_tenant_column` is `Some(col)`, each table gets that column (if missing), RLS enabled, and policies using `current_setting('app.tenant_id', true)`.
pub async fn apply_migrations(
    pool: &Pool,
    config: &FullConfig,
    schema_override: Option<&str>,
    rls_tenant_column: Option<&str>,
    dialect: &dyn Dialect,
    cross_package_configs: &HashMap<String, FullConfig>,
) -> Result<(), AppError> {
    validate(config)?;
    let default_sid = config
        .schemas
        .first()
        .map(|s| s.id.as_str())
        .ok_or_else(|| {
            AppError::Config(crate::error::ConfigError::Validation(
                "at least one schema required".into(),
            ))
        })?;

    if dialect.supports_schemas() {
        if let Some(s) = schema_override {
            let name = quote(s);
            sqlx::query(&format!("CREATE SCHEMA IF NOT EXISTS {}", name))
                .execute(pool)
                .await?;
        }
    }

    let schemas_by_id: HashMap<_, _> = config.schemas.iter().map(|s| (s.id.as_str(), s)).collect();
    let tables_by_id: HashMap<_, _> = config.tables.iter().map(|t| (t.id.as_str(), t)).collect();
    let columns_by_table: HashMap<_, Vec<&ColumnConfig>> =
        config.columns.iter().fold(HashMap::new(), |mut m, c| {
            m.entry(c.table_id.as_str()).or_default().push(c);
            m
        });

    // When schema_override is set, we only create the override schema; otherwise create config schemas.
    if schema_override.is_none() && dialect.supports_schemas() {
        for s in &config.schemas {
            let name = quote(&s.name);
            let comment = s
                .comment
                .as_ref()
                .map(|c| format!("COMMENT ON SCHEMA {} IS '{}'", name, c.replace('\'', "''")));
            sqlx::query(&format!("CREATE SCHEMA IF NOT EXISTS {}", name))
                .execute(pool)
                .await?;
            if let Some(sql) = comment {
                let _ = sqlx::query(&sql).execute(pool).await;
            }
        }
    }

    for e in &config.enums {
        let sid = e.schema_id.as_deref().unwrap_or(default_sid);
        let schema = schemas_by_id.get(sid).ok_or_else(|| {
            AppError::Config(crate::error::ConfigError::MissingReference {
                kind: "schema",
                id: sid.to_string(),
            })
        })?;
        let schema_name = quote(schema_override.unwrap_or(&schema.name));
        let type_name = quote(&e.name);
        if dialect.supports_named_enum_types() {
            let values: Vec<String> = e
                .values
                .iter()
                .map(|v| format!("'{}'", v.replace('\'', "''")))
                .collect();
            let sql = format!(
                "CREATE TYPE {}.{} AS ENUM ({})",
                schema_name,
                type_name,
                values.join(", ")
            );
            let _ = sqlx::query(&sql).execute(pool).await;
        }
    }

    // Which tables already exist, and with which columns? `CREATE TABLE IF NOT EXISTS` silently
    // leaves a pre-existing table alone, so without this a table created by an older version of
    // the package would never gain the columns added since. Taken before any DDL runs, so
    // `has_table` means "existed before this call".
    let target_schemas: Vec<String> = match schema_override {
        Some(o) => vec![o.to_string()],
        None => config.schemas.iter().map(|s| s.name.clone()).collect(),
    };
    let pre_snapshot = crate::db::introspect(pool, dialect, &target_schemas).await;

    for t in &config.tables {
        let sid = t.schema_id.as_deref().unwrap_or(default_sid);
        let schema = schemas_by_id.get(sid).ok_or_else(|| {
            AppError::Config(crate::error::ConfigError::MissingReference {
                kind: "schema",
                id: sid.to_string(),
            })
        })?;
        let schema_raw = schema_override.unwrap_or(&schema.name);
        let schema_name = quote(schema_raw);
        let table_name = quote(&t.name);
        let full_name = format!("{}.{}", schema_name, table_name);
        let table_pre_existed = pre_snapshot.has_table(schema_raw, &t.name);

        let cols = columns_by_table
            .get(t.id.as_str())
            .map(|v| v.as_slice())
            .unwrap_or(&[]);
        // (column name, DDL suffix) pairs — reused below to add columns a pre-existing table
        // is missing, so CREATE TABLE and ALTER TABLE can never disagree about a definition.
        let mut columns: Vec<(String, String)> = Vec::new();
        for c in cols {
            let typ = dialect.ddl_type(&parse_canonical(&c.type_));
            let mut def = typ;
            if !c.nullable {
                def.push_str(" NOT NULL");
            }
            if let Some(ref d) = c.default {
                def.push_str(" DEFAULT ");
                match d {
                    ColumnDefaultConfig::Literal(s) => def.push_str(s),
                    ColumnDefaultConfig::Expression { expression } => def.push_str(expression),
                }
            }
            columns.push((c.name.clone(), def));
        }

        let config_col_names: HashSet<&str> = cols.iter().map(|c| c.name.as_str()).collect();
        let ts_default = format!(
            "{} NOT NULL DEFAULT {}",
            dialect.sys_timestamp_type(),
            dialect.now_fn()
        );
        let ts_nullable = dialect.sys_timestamp_type().to_string();
        for (name, def_suffix) in [
            ("created_at", ts_default.as_str()),
            ("updated_at", ts_default.as_str()),
            ("archived_at", ts_nullable.as_str()),
            ("created_by", "TEXT"),
            ("updated_by", "TEXT"),
        ] {
            if !config_col_names.contains(name) {
                columns.push((name.to_string(), def_suffix.to_string()));
            }
        }

        let mut col_defs: Vec<String> = columns
            .iter()
            .map(|(name, def)| format!("{} {}", quote(name), def))
            .collect();

        let pk_cols = match &t.primary_key {
            PrimaryKeyConfig::Single(s) => vec![quote(s)],
            PrimaryKeyConfig::Composite(v) => v.iter().map(|s| quote(s)).collect::<Vec<_>>(),
        };
        let pk_def = format!("PRIMARY KEY ({})", pk_cols.join(", "));
        col_defs.push(pk_def);

        for u in &t.unique {
            let cols: Vec<String> = u.iter().map(|s| quote(s)).collect();
            col_defs.push(format!("UNIQUE ({})", cols.join(", ")));
        }
        for ch in &t.check {
            col_defs.push(format!(
                "CONSTRAINT {} CHECK ({})",
                quote(&ch.name),
                ch.expression
            ));
        }

        let sql = format!(
            "CREATE TABLE IF NOT EXISTS {} (\n  {}\n)",
            full_name,
            col_defs.join(",\n  ")
        );
        sqlx::query(&sql).execute(pool).await?;

        if table_pre_existed {
            add_missing_columns(pool, &pre_snapshot, dialect, schema_raw, &t.name, &columns).await;
        }

        if t.audit_log {
            let audit_sql = audit_table_ddl(schema_raw, &t.name, cols, dialect);
            sqlx::query(&audit_sql).execute(pool).await?;
            let pk_col = match &t.primary_key {
                PrimaryKeyConfig::Single(s) => s.clone(),
                PrimaryKeyConfig::Composite(v) => v[0].clone(),
            };
            let audit_full = format!(
                "{}.{}",
                quote(schema_raw),
                quote(&format!("{}_audit", t.name))
            );
            let idx_sql = format!(
                "CREATE INDEX IF NOT EXISTS {} ON {} ({}, {})",
                quote(&format!("{}_audit_record_idx", t.name)),
                audit_full,
                quote(&pk_col),
                quote("audit_at")
            );
            let _ = sqlx::query(&idx_sql).execute(pool).await;

            let audit_table = format!("{}_audit", t.name);
            if pre_snapshot.has_table(schema_raw, &audit_table) {
                add_missing_columns(
                    pool,
                    &pre_snapshot,
                    dialect,
                    schema_raw,
                    &audit_table,
                    &companion_source_columns(cols, dialect),
                )
                .await;
            }
        }

        if t.versioning.as_ref().is_some_and(|v| v.enabled) {
            let pk_col = match &t.primary_key {
                PrimaryKeyConfig::Single(s) => s.clone(),
                PrimaryKeyConfig::Composite(v) => v[0].clone(),
            };
            let history_ddl = history_table_ddl(schema_raw, &t.name, &pk_col, cols, dialect);
            // history_table_ddl embeds a comment line for the index; execute CREATE TABLE only
            let create_only = history_ddl
                .lines()
                .take_while(|l| !l.trim_start().starts_with("-- index:"))
                .collect::<Vec<_>>()
                .join("\n");
            sqlx::query(create_only.trim()).execute(pool).await?;
            let idx_sql = history_index_ddl(schema_raw, &t.name, &pk_col);
            let _ = sqlx::query(&idx_sql).execute(pool).await;

            let history_table = format!("{}_history", t.name);
            if pre_snapshot.has_table(schema_raw, &history_table) {
                add_missing_columns(
                    pool,
                    &pre_snapshot,
                    dialect,
                    schema_raw,
                    &history_table,
                    &companion_source_columns(cols, dialect),
                )
                .await;
            }
        }
    }

    if let Some(col) = rls_tenant_column {
        apply_rls_to_tables(pool, config, schema_override, col, dialect).await?;
    }

    for idx in &config.indexes {
        let sid = idx.schema_id.as_deref().unwrap_or(default_sid);
        let schema = schemas_by_id.get(sid).ok_or_else(|| {
            AppError::Config(crate::error::ConfigError::MissingReference {
                kind: "schema",
                id: sid.to_string(),
            })
        })?;
        let table = tables_by_id.get(idx.table_id.as_str()).ok_or_else(|| {
            AppError::Config(crate::error::ConfigError::MissingReference {
                kind: "table",
                id: idx.table_id.clone(),
            })
        })?;
        let schema_name = quote(schema_override.unwrap_or(&schema.name));
        let table_name = quote(&table.name);
        let full_table = format!("{}.{}", schema_name, table_name);
        let index_name = quote(&idx.name);

        let mut col_parts: Vec<String> = Vec::new();
        for col in &idx.columns {
            match col {
                IndexColumnEntry::Name(n) => col_parts.push(quote(n)),
                IndexColumnEntry::Spec {
                    name, direction, ..
                } => {
                    let dir = direction
                        .as_deref()
                        .map(|d| format!(" {}", d.to_uppercase()))
                        .unwrap_or_default();
                    col_parts.push(format!("{}{}", quote(name), dir));
                }
                IndexColumnEntry::Expression { expression } => col_parts.push(expression.clone()),
            }
        }
        let method = idx.method.as_deref().unwrap_or("btree");
        let unique = if idx.unique { "UNIQUE " } else { "" };
        let include: String = if idx.include.is_empty() {
            String::new()
        } else {
            let inc: Vec<String> = idx.include.iter().map(|s| quote(s)).collect();
            format!(" INCLUDE ({})", inc.join(", "))
        };
        let where_clause: String = idx
            .where_
            .as_ref()
            .map(|w| format!(" WHERE {}", w))
            .unwrap_or_default();

        let sql = format!(
            "CREATE {}INDEX IF NOT EXISTS {} ON {} USING {} ({}){}{}",
            unique,
            index_name,
            full_table,
            method,
            col_parts.join(", "),
            include,
            where_clause
        );
        let _ = sqlx::query(&sql).execute(pool).await;
    }

    for rel in &config.relationships {
        let from_sid = rel.from_schema_id.as_deref().unwrap_or(default_sid);
        let from_schema = schemas_by_id.get(from_sid).ok_or_else(|| {
            AppError::Config(crate::error::ConfigError::MissingReference {
                kind: "schema",
                id: from_sid.to_string(),
            })
        })?;
        let from_table = tables_by_id
            .get(rel.from_table_id.as_str())
            .ok_or_else(|| {
                AppError::Config(crate::error::ConfigError::MissingReference {
                    kind: "table",
                    id: rel.from_table_id.clone(),
                })
            })?;

        // Resolve the target schema and table — either from a cross-package config or this config.
        let (to_schema_name_owned, to_table_name, to_col_name) = if let Some(pkg_id) =
            rel.to_package_id.as_deref()
        {
            let foreign = cross_package_configs.get(pkg_id).ok_or_else(|| {
                AppError::Config(crate::error::ConfigError::MissingReference {
                    kind: "cross_package",
                    id: pkg_id.to_string(),
                })
            })?;
            let foreign_tables: HashMap<_, _> =
                foreign.tables.iter().map(|t| (t.id.as_str(), t)).collect();
            let foreign_schemas: HashMap<_, _> =
                foreign.schemas.iter().map(|s| (s.id.as_str(), s)).collect();
            let to_tbl = foreign_tables
                .get(rel.to_table_id.as_str())
                .ok_or_else(|| {
                    AppError::Config(crate::error::ConfigError::MissingReference {
                        kind: "table",
                        id: rel.to_table_id.clone(),
                    })
                })?;
            let foreign_default_sid = foreign.schemas.first().map(|s| s.id.as_str()).unwrap_or("");
            let to_sid = rel.to_schema_id.as_deref().unwrap_or(foreign_default_sid);
            let to_schema = foreign_schemas.get(to_sid).ok_or_else(|| {
                AppError::Config(crate::error::ConfigError::MissingReference {
                    kind: "schema",
                    id: to_sid.to_string(),
                })
            })?;
            let col_name = foreign
                .columns
                .iter()
                .find(|c| c.id == rel.to_column_id)
                .map(|c| c.name.clone())
                .ok_or_else(|| {
                    AppError::Config(crate::error::ConfigError::MissingReference {
                        kind: "column",
                        id: rel.to_column_id.clone(),
                    })
                })?;
            // Cross-package FKs always use the real schema name (no schema_override — the
            // target package lives in its own schema, not the tenant override).
            (to_schema.name.clone(), to_tbl.name.clone(), col_name)
        } else {
            let to_sid = rel.to_schema_id.as_deref().unwrap_or(default_sid);
            let to_schema = schemas_by_id.get(to_sid).ok_or_else(|| {
                AppError::Config(crate::error::ConfigError::MissingReference {
                    kind: "schema",
                    id: to_sid.to_string(),
                })
            })?;
            let to_table = tables_by_id.get(rel.to_table_id.as_str()).ok_or_else(|| {
                AppError::Config(crate::error::ConfigError::MissingReference {
                    kind: "table",
                    id: rel.to_table_id.clone(),
                })
            })?;
            let col_name = config
                .columns
                .iter()
                .find(|c| c.id == rel.to_column_id)
                .map(|c| c.name.clone())
                .ok_or_else(|| {
                    AppError::Config(crate::error::ConfigError::MissingReference {
                        kind: "column",
                        id: rel.to_column_id.clone(),
                    })
                })?;
            (
                schema_override.unwrap_or(&to_schema.name).to_string(),
                to_table.name.clone(),
                col_name,
            )
        };

        let from_schema_name = schema_override.unwrap_or(&from_schema.name);
        let from_col = config
            .columns
            .iter()
            .find(|c| c.id == rel.from_column_id)
            .map(|c| c.name.as_str())
            .ok_or_else(|| {
                AppError::Config(crate::error::ConfigError::MissingReference {
                    kind: "column",
                    id: rel.from_column_id.clone(),
                })
            })?;

        let from_full = format!("{}.{}", quote(from_schema_name), quote(&from_table.name));
        let to_full = format!("{}.{}", quote(&to_schema_name_owned), quote(&to_table_name));
        let constraint_name = rel.name.as_deref().unwrap_or(&rel.id);
        let on_update = rel.on_update.as_deref().unwrap_or("NO ACTION");
        let on_delete = rel.on_delete.as_deref().unwrap_or("NO ACTION");

        let sql = format!(
            "ALTER TABLE {} ADD CONSTRAINT {} FOREIGN KEY ({}) REFERENCES {} ({}) ON UPDATE {} ON DELETE {}",
            from_full,
            quote(constraint_name),
            quote(from_col),
            to_full,
            quote(&to_col_name),
            on_update,
            on_delete
        );
        let _ = sqlx::query(&sql).execute(pool).await;
    }

    Ok(())
}

/// Revert migrations for a package: drop tables, enum types, and schema (if not public) in reverse order of apply.
/// Uses the same schema_override as apply_migrations (tables/enums live in that schema).
pub async fn revert_migrations(
    pool: &Pool,
    config: &FullConfig,
    schema_override: Option<&str>,
) -> Result<(), AppError> {
    let default_sid = config
        .schemas
        .first()
        .map(|s| s.id.as_str())
        .ok_or_else(|| {
            AppError::Config(crate::error::ConfigError::Validation(
                "at least one schema required".into(),
            ))
        })?;

    let schemas_by_id: HashMap<_, _> = config.schemas.iter().map(|s| (s.id.as_str(), s)).collect();

    // 1. Drop tables (CASCADE drops FKs and dependent objects)
    for t in &config.tables {
        let sid = t.schema_id.as_deref().unwrap_or(default_sid);
        let schema = schemas_by_id.get(sid).ok_or_else(|| {
            AppError::Config(crate::error::ConfigError::MissingReference {
                kind: "schema",
                id: sid.to_string(),
            })
        })?;
        let schema_raw = schema_override.unwrap_or(&schema.name);
        let schema_name = quote(schema_raw);
        let table_name = quote(&t.name);
        let full_name = format!("{}.{}", schema_name, table_name);
        if t.audit_log {
            let audit_full = format!("{}.{}", schema_name, quote(&format!("{}_audit", t.name)));
            let _ = sqlx::query(&format!("DROP TABLE IF EXISTS {} CASCADE", audit_full))
                .execute(pool)
                .await;
        }
        if t.versioning.as_ref().is_some_and(|v| v.enabled) {
            let history_full = format!("{}.{}", schema_name, quote(&format!("{}_history", t.name)));
            let _ = sqlx::query(&format!("DROP TABLE IF EXISTS {} CASCADE", history_full))
                .execute(pool)
                .await;
        }
        let drop_sql = format!("DROP TABLE IF EXISTS {} CASCADE", full_name);
        let _ = sqlx::query(&drop_sql).execute(pool).await;
    }

    // 2. Drop enum types
    for e in &config.enums {
        let sid = e.schema_id.as_deref().unwrap_or(default_sid);
        let schema = schemas_by_id.get(sid).ok_or_else(|| {
            AppError::Config(crate::error::ConfigError::MissingReference {
                kind: "schema",
                id: sid.to_string(),
            })
        })?;
        let schema_name = quote(schema_override.unwrap_or(&schema.name));
        let type_name = quote(&e.name);
        let drop_sql = format!("DROP TYPE IF EXISTS {}.{} CASCADE", schema_name, type_name);
        let _ = sqlx::query(&drop_sql).execute(pool).await;
    }

    // 3. Drop schema only if not public (shared schema)
    if schema_override.is_none() {
        for s in &config.schemas {
            if s.name.eq_ignore_ascii_case("public") {
                continue;
            }
            let schema_name = quote(&s.name);
            let drop_sql = format!("DROP SCHEMA IF EXISTS {} CASCADE", schema_name);
            let _ = sqlx::query(&drop_sql).execute(pool).await;
        }
    }

    Ok(())
}

/// `ALTER TABLE … ADD COLUMN` with an `IF NOT EXISTS` guard where the dialect supports it.
///
/// The guard makes the step a no-op when the column is already present — the common case when a
/// plan is replayed after a partial failure, or broadcast to a tenant database that already has
/// the column. Dialects without the syntax (MySQL, SQLite) rely on the executor's pre-flight
/// introspection instead (see `db::introspect`).
fn add_column_ddl(dialect: &dyn Dialect, full_table: &str, col_def: &str) -> String {
    if dialect.supports_add_column_if_not_exists() {
        format!(
            "ALTER TABLE {} ADD COLUMN IF NOT EXISTS {}",
            full_table, col_def
        )
    } else {
        format!("ALTER TABLE {} ADD COLUMN {}", full_table, col_def)
    }
}

// ─── Migration plan types ────────────────────────────────────────────────────

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum MigrationOperation {
    CreateSchema,
    CreateEnum,
    DropEnum,
    AddEnumValue,
    RemoveEnumValue,
    CreateTable,
    DropTable,
    AddColumn,
    DropColumn,
    RenameColumn,
    AlterColumnType,
    BackfillNulls,
    SetNotNull,
    DropNotNull,
    SetDefault,
    DropDefault,
    CreateIndex,
    DropIndex,
    AddForeignKey,
    DropForeignKey,
}

impl std::fmt::Display for MigrationOperation {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let s = serde_json::to_value(self)
            .ok()
            .and_then(|v| v.as_str().map(String::from))
            .unwrap_or_else(|| format!("{:?}", self));
        write!(f, "{}", s)
    }
}

/// How safely a migration step can be executed.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum MigrationSafety {
    /// Guaranteed to succeed, no data impact.
    Safe,
    /// Attempted; execution failure is captured as a warning instead of aborting.
    BestEffort,
    /// No DDL generated — config change noted as a warning only (e.g. removed tables/columns).
    WarnOnly,
}

/// Risk category associated with a migration step.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum MigrationRisk {
    None,
    /// Cast may fail for incompatible values (e.g. TEXT → INTEGER).
    MayFail,
    /// SET NOT NULL will fail if any existing row has NULL in this column.
    ExistingNullsMustBeAbsent,
    /// Existing NULL rows will be overwritten with the column default.
    DataWillBeModified,
    /// Cannot be automated — requires a manual database action.
    ManualActionRequired,
}

/// One step in a migration plan: a DDL statement with metadata.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MigrationStep {
    pub step: usize,
    pub operation: MigrationOperation,
    pub schema: String,
    pub table: Option<String>,
    /// Column name, index name, FK constraint name, enum name, etc.
    pub object: String,
    /// Previous name of `object` for rename operations. `None` for every other operation.
    /// Optional in the serialized form so migration plans saved before this field existed
    /// still deserialize.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub from_object: Option<String>,
    /// "column" | "table" | "index" | "foreign_key" | "enum" | "enum_value" | "schema"
    pub object_type: String,
    pub description: String,
    /// The SQL to execute. None for WarnOnly steps.
    pub ddl: Option<String>,
    pub safety: MigrationSafety,
    pub risk: MigrationRisk,
    pub risk_detail: Option<String>,
}

/// Computed diff between two package versions expressed as ordered migration steps.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MigrationPlan {
    pub steps: Vec<MigrationStep>,
}

#[derive(Debug, Clone, Serialize)]
pub struct MigrationSummary {
    pub total: usize,
    pub safe: usize,
    pub best_effort: usize,
    pub warn_only: usize,
}

impl MigrationPlan {
    pub fn summary(&self) -> MigrationSummary {
        let (mut safe, mut best_effort, mut warn_only) = (0, 0, 0);
        for s in &self.steps {
            match s.safety {
                MigrationSafety::Safe => safe += 1,
                MigrationSafety::BestEffort => best_effort += 1,
                MigrationSafety::WarnOnly => warn_only += 1,
            }
        }
        MigrationSummary {
            total: self.steps.len(),
            safe,
            best_effort,
            warn_only,
        }
    }
}

/// Result returned by `execute_migration_plan`.
pub struct MigrationExecutionResult {
    pub applied: usize,
    pub warned: usize,
    /// Steps whose effect was already present in the database and were not re-run.
    pub skipped: usize,
    pub warnings: Vec<String>,
    /// One human-readable line per skipped step.
    pub skips: Vec<String>,
}

fn default_str(d: &ColumnDefaultConfig) -> String {
    match d {
        ColumnDefaultConfig::Literal(s) => s.clone(),
        ColumnDefaultConfig::Expression { expression } => expression.clone(),
    }
}

/// A column whose type is (an array of) a given enum — i.e. one that must be recast when the
/// enum type is rebuilt.
struct EnumColumnRef {
    schema: String,
    table: String,
    column: String,
    default: Option<String>,
    is_array: bool,
}

/// If `t` is a custom enum reference (`schema.name`, bare `name`, or an array of one), return
/// `(enum_name, is_array)` where `enum_name` is the unqualified type name. None for built-in types.
fn enum_type_name(t: &crate::db::CanonicalType) -> Option<(String, bool)> {
    use crate::db::CanonicalType;
    let unqualified = |s: &str| s.rsplit('.').next().unwrap_or(s).to_string();
    match t {
        CanonicalType::Custom(s) => Some((unqualified(s), false)),
        CanonicalType::Array(inner) => match inner.as_ref() {
            CanonicalType::Custom(s) => Some((unqualified(s), true)),
            _ => None,
        },
        _ => None,
    }
}

/// Find every column in `new` whose type is `new_enum` (or an array of it), resolved to its live
/// table/schema name. Used to drive the recast steps of an enum rebuild.
fn enum_dependent_columns(
    new_enum: &EnumConfig,
    new: &FullConfig,
    new_tables: &HashMap<&str, &TableConfig>,
    new_schemas: &HashMap<&str, &SchemaConfig>,
    schema_override: Option<&str>,
) -> Vec<EnumColumnRef> {
    let default_sid = new.schemas.first().map(|s| s.id.as_str()).unwrap_or("");
    let mut out = Vec::new();
    for c in &new.columns {
        let Some((tyname, is_array)) = enum_type_name(&parse_canonical(&c.type_)) else {
            continue;
        };
        if tyname != new_enum.name {
            continue;
        }
        let Some(table) = new_tables.get(c.table_id.as_str()) else {
            continue;
        };
        let tsid = table.schema_id.as_deref().unwrap_or(default_sid);
        let schema = schema_override.map(String::from).unwrap_or_else(|| {
            new_schemas
                .get(tsid)
                .map(|s| s.name.clone())
                .unwrap_or_else(|| tsid.to_string())
        });
        out.push(EnumColumnRef {
            schema,
            table: table.name.clone(),
            column: c.name.clone(),
            default: c.default.as_ref().map(default_str),
            is_array,
        });
    }
    out
}

/// PostgreSQL cannot drop a value from an enum in place, so when one is removed the type must be
/// rebuilt: rename the live type aside, create it afresh with the reduced value set, recast every
/// dependent column through a text cast, then drop the old type. Steps are emitted as `BestEffort`
/// — a recast that fails because a row still holds a removed value is surfaced as a warning rather
/// than aborting the whole upgrade. Each statement is a separate step because `execute_migration_plan`
/// runs them individually (the extended protocol forbids multiple statements per query).
fn recreate_enum_steps(
    steps: &mut Vec<MigrationStep>,
    schema: &str,
    new_enum: &EnumConfig,
    removed: &[&str],
    dependents: &[EnumColumnRef],
) {
    let type_q = format!("{}.{}", quote(schema), quote(&new_enum.name));
    let tmp_name = format!("{}__arch_old", new_enum.name);
    let values: Vec<String> = new_enum
        .values
        .iter()
        .map(|v| format!("'{}'", v.replace('\'', "''")))
        .collect();

    // 0. Informational summary of the destructive rebuild (no DDL).
    steps.push(MigrationStep {
        step: 0,
        operation: MigrationOperation::RemoveEnumValue,
        schema: schema.to_string(),
        table: None,
        object: format!("{}:{}", new_enum.name, removed.join(",")),
        object_type: "enum".into(),
        from_object: None,
        description: format!(
            "Rebuild enum \"{}\".\"{}\" to remove value(s): {}",
            schema,
            new_enum.name,
            removed.join(", ")
        ),
        ddl: None,
        safety: MigrationSafety::WarnOnly,
        risk: MigrationRisk::ManualActionRequired,
        risk_detail: Some(format!(
            "PostgreSQL cannot drop enum values in place. The type is rebuilt and {} dependent \
             column(s) are recast via a text cast. Any existing row holding a removed value ({}) \
             will make its recast fail — reassign those rows first.",
            dependents.len(),
            removed.join(", ")
        )),
    });

    // 1. Rename the live type aside.
    steps.push(MigrationStep {
        step: 0,
        operation: MigrationOperation::DropEnum,
        schema: schema.to_string(),
        table: None,
        object: new_enum.name.clone(),
        object_type: "enum".into(),
        from_object: None,
        description: format!(
            "Rename enum \"{}\".\"{}\" to \"{}\" before rebuild",
            schema, new_enum.name, tmp_name
        ),
        ddl: Some(format!(
            "ALTER TYPE {} RENAME TO {}",
            type_q,
            quote(&tmp_name)
        )),
        safety: MigrationSafety::BestEffort,
        risk: MigrationRisk::None,
        risk_detail: None,
    });

    // 2. Create the type afresh with the reduced value set (folds in any added values too).
    steps.push(MigrationStep {
        step: 0,
        operation: MigrationOperation::CreateEnum,
        schema: schema.to_string(),
        table: None,
        object: new_enum.name.clone(),
        object_type: "enum".into(),
        from_object: None,
        description: format!(
            "Recreate enum \"{}\".\"{}\" with {} value(s)",
            schema,
            new_enum.name,
            new_enum.values.len()
        ),
        ddl: Some(format!(
            "CREATE TYPE {} AS ENUM ({})",
            type_q,
            values.join(", ")
        )),
        safety: MigrationSafety::BestEffort,
        risk: MigrationRisk::None,
        risk_detail: None,
    });

    // 3. Recast every dependent column from the renamed type onto the rebuilt one. A column with a
    //    default must drop it first (the default still references the renamed type) and restore it after.
    for dep in dependents {
        let table_q = format!("{}.{}", quote(&dep.schema), quote(&dep.table));
        let col_q = quote(&dep.column);

        if dep.default.is_some() {
            steps.push(MigrationStep {
                step: 0,
                operation: MigrationOperation::DropDefault,
                schema: dep.schema.clone(),
                table: Some(dep.table.clone()),
                object: dep.column.clone(),
                object_type: "column".into(),
                from_object: None,
                description: format!(
                    "Drop default on {}.{} before enum recast",
                    dep.table, dep.column
                ),
                ddl: Some(format!(
                    "ALTER TABLE {} ALTER COLUMN {} DROP DEFAULT",
                    table_q, col_q
                )),
                safety: MigrationSafety::BestEffort,
                risk: MigrationRisk::None,
                risk_detail: None,
            });
        }

        let (col_type, using) = if dep.is_array {
            (
                format!("{}[]", type_q),
                format!("{}::text[]::{}[]", col_q, type_q),
            )
        } else {
            (type_q.clone(), format!("{}::text::{}", col_q, type_q))
        };
        steps.push(MigrationStep {
            step: 0,
            operation: MigrationOperation::AlterColumnType,
            schema: dep.schema.clone(),
            table: Some(dep.table.clone()),
            object: dep.column.clone(),
            object_type: "column".into(),
            from_object: None,
            description: format!(
                "Recast {}.{} onto rebuilt enum \"{}\"",
                dep.table, dep.column, new_enum.name
            ),
            ddl: Some(format!(
                "ALTER TABLE {} ALTER COLUMN {} TYPE {} USING {}",
                table_q, col_q, col_type, using
            )),
            safety: MigrationSafety::BestEffort,
            risk: MigrationRisk::MayFail,
            risk_detail: Some(format!(
                "Cast fails if any row holds a removed value ({}).",
                removed.join(", ")
            )),
        });

        if let Some(def) = &dep.default {
            steps.push(MigrationStep {
                step: 0,
                operation: MigrationOperation::SetDefault,
                schema: dep.schema.clone(),
                table: Some(dep.table.clone()),
                object: dep.column.clone(),
                object_type: "column".into(),
                from_object: None,
                description: format!(
                    "Restore default on {}.{} after enum recast",
                    dep.table, dep.column
                ),
                ddl: Some(format!(
                    "ALTER TABLE {} ALTER COLUMN {} SET DEFAULT {}",
                    table_q, col_q, def
                )),
                safety: MigrationSafety::BestEffort,
                risk: MigrationRisk::None,
                risk_detail: None,
            });
        }
    }

    // 4. Drop the renamed original type.
    steps.push(MigrationStep {
        step: 0,
        operation: MigrationOperation::DropEnum,
        schema: schema.to_string(),
        table: None,
        object: tmp_name.clone(),
        object_type: "enum".into(),
        from_object: None,
        description: format!("Drop superseded enum \"{}\".\"{}\"", schema, tmp_name),
        ddl: Some(format!(
            "DROP TYPE IF EXISTS {}.{}",
            quote(schema),
            quote(&tmp_name)
        )),
        safety: MigrationSafety::BestEffort,
        risk: MigrationRisk::None,
        risk_detail: None,
    });
}

// ─── compute_migration_plan ──────────────────────────────────────────────────

/// Diff two package configs and produce an ordered list of migration steps.
/// This is a pure function — it does not touch the database.
/// Pass the result to `execute_migration_plan` after user confirmation.
pub fn compute_migration_plan(
    old: &FullConfig,
    new: &FullConfig,
    schema_override: Option<&str>,
    _rls_tenant_column: Option<&str>,
    dialect: &dyn Dialect,
    cross_package_configs: &HashMap<String, FullConfig>,
) -> Result<MigrationPlan, AppError> {
    validate(new)?;

    let default_old_sid = old.schemas.first().map(|s| s.id.as_str()).unwrap_or("");
    let default_new_sid = new.schemas.first().map(|s| s.id.as_str()).unwrap_or("");

    let old_schemas: HashMap<&str, &SchemaConfig> =
        old.schemas.iter().map(|s| (s.id.as_str(), s)).collect();
    let new_schemas: HashMap<&str, &SchemaConfig> =
        new.schemas.iter().map(|s| (s.id.as_str(), s)).collect();
    let old_tables: HashMap<&str, &TableConfig> =
        old.tables.iter().map(|t| (t.id.as_str(), t)).collect();
    let new_tables: HashMap<&str, &TableConfig> =
        new.tables.iter().map(|t| (t.id.as_str(), t)).collect();
    let old_columns: HashMap<&str, &ColumnConfig> =
        old.columns.iter().map(|c| (c.id.as_str(), c)).collect();
    let old_enums: HashMap<&str, &EnumConfig> =
        old.enums.iter().map(|e| (e.id.as_str(), e)).collect();
    let new_enums: HashMap<&str, &EnumConfig> =
        new.enums.iter().map(|e| (e.id.as_str(), e)).collect();
    let old_indexes: HashMap<&str, &IndexConfig> =
        old.indexes.iter().map(|i| (i.id.as_str(), i)).collect();
    let new_indexes: HashMap<&str, &IndexConfig> =
        new.indexes.iter().map(|i| (i.id.as_str(), i)).collect();
    let old_rels: HashMap<&str, &RelationshipConfig> = old
        .relationships
        .iter()
        .map(|r| (r.id.as_str(), r))
        .collect();
    let new_rels: HashMap<&str, &RelationshipConfig> = new
        .relationships
        .iter()
        .map(|r| (r.id.as_str(), r))
        .collect();

    let mut steps: Vec<MigrationStep> = Vec::new();

    let schema_name_for = |sid: &str, schemas: &HashMap<&str, &SchemaConfig>| -> String {
        schema_override.map(String::from).unwrap_or_else(|| {
            schemas
                .get(sid)
                .map(|s| s.name.clone())
                .unwrap_or_else(|| sid.to_string())
        })
    };

    // ── 1. New schemas ───────────────────────────────────────────────────────
    if schema_override.is_none() {
        for s in &new.schemas {
            if !old_schemas.contains_key(s.id.as_str()) {
                steps.push(MigrationStep {
                    step: 0,
                    operation: MigrationOperation::CreateSchema,
                    schema: s.name.clone(),
                    table: None,
                    object: s.name.clone(),
                    object_type: "schema".into(),
                    from_object: None,
                    description: format!("Create schema \"{}\"", s.name),
                    ddl: Some(format!("CREATE SCHEMA IF NOT EXISTS {}", quote(&s.name))),
                    safety: MigrationSafety::Safe,
                    risk: MigrationRisk::None,
                    risk_detail: None,
                });
            }
        }
    }

    // ── 2. Enums ─────────────────────────────────────────────────────────────
    for new_enum in &new.enums {
        let sid = new_enum.schema_id.as_deref().unwrap_or(default_new_sid);
        let schema = schema_name_for(sid, &new_schemas);

        if let Some(old_enum) = old_enums.get(new_enum.id.as_str()) {
            let old_vals: HashSet<&str> = old_enum.values.iter().map(String::as_str).collect();
            let new_vals: HashSet<&str> = new_enum.values.iter().map(String::as_str).collect();
            let removed: Vec<&str> = old_enum
                .values
                .iter()
                .map(String::as_str)
                .filter(|v| !new_vals.contains(v))
                .collect();

            if removed.is_empty() {
                // Purely additive — append each new value in place (cheap, non-destructive).
                for val in new_enum
                    .values
                    .iter()
                    .map(String::as_str)
                    .filter(|v| !old_vals.contains(v))
                {
                    steps.push(MigrationStep {
                        step: 0,
                        operation: MigrationOperation::AddEnumValue,
                        schema: schema.clone(),
                        table: None,
                        object: format!("{}:{}", new_enum.name, val),
                        object_type: "enum_value".into(),
                        from_object: None,
                        description: format!(
                            "Add value '{}' to enum \"{}\".\"{}\"",
                            val, schema, new_enum.name
                        ),
                        ddl: Some(format!(
                            "ALTER TYPE {}.{} ADD VALUE IF NOT EXISTS '{}'",
                            quote(&schema),
                            quote(&new_enum.name),
                            val.replace('\'', "''")
                        )),
                        safety: MigrationSafety::Safe,
                        risk: MigrationRisk::None,
                        risk_detail: None,
                    });
                }
            } else {
                // One or more values removed. PostgreSQL has no DROP VALUE, so rebuild the type and
                // recast every dependent column. Added values (if any) are folded into the rebuilt
                // value list, so no separate ADD VALUE step is needed.
                let dependents = enum_dependent_columns(
                    new_enum,
                    new,
                    &new_tables,
                    &new_schemas,
                    schema_override,
                );
                recreate_enum_steps(&mut steps, &schema, new_enum, &removed, &dependents);
            }
        } else {
            let values: Vec<String> = new_enum
                .values
                .iter()
                .map(|v| format!("'{}'", v.replace('\'', "''")))
                .collect();
            steps.push(MigrationStep {
                step: 0,
                operation: MigrationOperation::CreateEnum,
                schema: schema.clone(),
                table: None,
                object: new_enum.name.clone(),
                object_type: "enum".into(),
                from_object: None,
                description: format!("Create enum type \"{}\".\"{}\"", schema, new_enum.name),
                ddl: Some(format!("CREATE TYPE {}.{} AS ENUM ({})", quote(&schema), quote(&new_enum.name), values.join(", "))),
                safety: MigrationSafety::BestEffort,
                risk: MigrationRisk::None,
                risk_detail: Some("PostgreSQL has no CREATE TYPE IF NOT EXISTS; ignored if the type already exists.".into()),
            });
        }
    }
    for old_enum in &old.enums {
        if !new_enums.contains_key(old_enum.id.as_str()) {
            let sid = old_enum.schema_id.as_deref().unwrap_or(default_old_sid);
            let schema = schema_name_for(sid, &old_schemas);
            steps.push(MigrationStep {
                step: 0,
                operation: MigrationOperation::DropEnum,
                schema: schema.clone(),
                table: None,
                object: old_enum.name.clone(),
                object_type: "enum".into(),
                from_object: None,
                description: format!("Enum \"{}\".\"{}\" removed from config", schema, old_enum.name),
                ddl: None,
                safety: MigrationSafety::WarnOnly,
                risk: MigrationRisk::ManualActionRequired,
                risk_detail: Some("Enum type NOT dropped from database (data safety). Run DROP TYPE manually if intended.".into()),
            });
        }
    }

    // ── 3. New and removed tables ────────────────────────────────────────────
    let added_table_ids: HashSet<&str> = new
        .tables
        .iter()
        .filter(|t| !old_tables.contains_key(t.id.as_str()))
        .map(|t| t.id.as_str())
        .collect();

    let cols_by_table: HashMap<&str, Vec<&ColumnConfig>> =
        new.columns.iter().fold(HashMap::new(), |mut m, c| {
            m.entry(c.table_id.as_str()).or_default().push(c);
            m
        });

    for new_table in &new.tables {
        if !added_table_ids.contains(new_table.id.as_str()) {
            continue;
        }
        let sid = new_table.schema_id.as_deref().unwrap_or(default_new_sid);
        let schema = schema_name_for(sid, &new_schemas);
        let full = format!("{}.{}", quote(&schema), quote(&new_table.name));

        let cols = cols_by_table
            .get(new_table.id.as_str())
            .map(|v| v.as_slice())
            .unwrap_or(&[]);
        let mut col_defs: Vec<String> = Vec::new();
        for c in cols {
            let typ = dialect.ddl_type(&parse_canonical(&c.type_));
            let mut def = format!("{} {}", quote(&c.name), typ);
            if !c.nullable {
                def.push_str(" NOT NULL");
            }
            if let Some(ref d) = c.default {
                def.push_str(" DEFAULT ");
                match d {
                    ColumnDefaultConfig::Literal(s) => def.push_str(s),
                    ColumnDefaultConfig::Expression { expression } => def.push_str(expression),
                }
            }
            col_defs.push(def);
        }
        let cfg_col_names: HashSet<&str> = cols.iter().map(|c| c.name.as_str()).collect();
        // Note: compute_migration_plan is a pure DDL-generation function; timestamp strings are
        // embedded in the DDL output for display/execution. We use postgres-compatible strings
        // here since the plan is always applied to a real DB via execute_migration_plan which
        // uses the dialect there. If dialect-awareness is needed here in future, pass dialect in.
        for (name, suf) in [
            ("created_at", "TIMESTAMPTZ NOT NULL DEFAULT NOW()"),
            ("updated_at", "TIMESTAMPTZ NOT NULL DEFAULT NOW()"),
            ("archived_at", "TIMESTAMPTZ"),
            ("created_by", "TEXT"),
            ("updated_by", "TEXT"),
        ] {
            if !cfg_col_names.contains(name) {
                col_defs.push(format!("{} {}", quote(name), suf));
            }
        }
        let pk_cols = match &new_table.primary_key {
            PrimaryKeyConfig::Single(s) => vec![quote(s)],
            PrimaryKeyConfig::Composite(v) => v.iter().map(|s| quote(s)).collect(),
        };
        col_defs.push(format!("PRIMARY KEY ({})", pk_cols.join(", ")));
        for u in &new_table.unique {
            col_defs.push(format!(
                "UNIQUE ({})",
                u.iter().map(|s| quote(s)).collect::<Vec<_>>().join(", ")
            ));
        }
        for ch in &new_table.check {
            col_defs.push(format!(
                "CONSTRAINT {} CHECK ({})",
                quote(&ch.name),
                ch.expression
            ));
        }

        steps.push(MigrationStep {
            step: 0,
            operation: MigrationOperation::CreateTable,
            schema: schema.clone(),
            table: Some(new_table.name.clone()),
            object: new_table.name.clone(),
            object_type: "table".into(),
            from_object: None,
            description: format!("Create table \"{}\".\"{}\"", schema, new_table.name),
            ddl: Some(format!(
                "CREATE TABLE IF NOT EXISTS {} (\n  {}\n)",
                full,
                col_defs.join(",\n  ")
            )),
            safety: MigrationSafety::Safe,
            risk: MigrationRisk::None,
            risk_detail: None,
        });
        if new_table.audit_log {
            let audit_ddl = audit_table_ddl(&schema, &new_table.name, cols, dialect);
            steps.push(MigrationStep {
                step: 0,
                operation: MigrationOperation::CreateTable,
                schema: schema.clone(),
                table: Some(format!("{}_audit", new_table.name)),
                object: format!("{}_audit", new_table.name),
                object_type: "table".into(),
                from_object: None,
                description: format!(
                    "Create audit table \"{}\".\"{}_audit\"",
                    schema, new_table.name
                ),
                ddl: Some(audit_ddl),
                safety: MigrationSafety::Safe,
                risk: MigrationRisk::None,
                risk_detail: None,
            });
        }
        if new_table.versioning.as_ref().is_some_and(|v| v.enabled) {
            let pk_col = match &new_table.primary_key {
                PrimaryKeyConfig::Single(s) => s.clone(),
                PrimaryKeyConfig::Composite(v) => v[0].clone(),
            };
            // Split history DDL into CREATE TABLE and index
            let history_create = format!(
                "CREATE TABLE IF NOT EXISTS {}.{} (\n  {}\n)",
                quote(&schema),
                quote(&format!("{}_history", new_table.name)),
                {
                    let full_ddl =
                        history_table_ddl(&schema, &new_table.name, &pk_col, cols, dialect);
                    full_ddl
                        .lines()
                        .skip(1) // skip CREATE TABLE line
                        .take_while(|l| !l.trim_start().starts_with("-- index:"))
                        .collect::<Vec<_>>()
                        .join("\n")
                        .trim_end_matches(['\n', ',', ')'])
                        .to_string()
                        + "\n)"
                }
            );
            steps.push(MigrationStep {
                step: 0,
                operation: MigrationOperation::CreateTable,
                schema: schema.clone(),
                table: Some(format!("{}_history", new_table.name)),
                object: format!("{}_history", new_table.name),
                object_type: "table".into(),
                from_object: None,
                description: format!(
                    "Create history table \"{}\".\"{}_history\" (versioning)",
                    schema, new_table.name
                ),
                ddl: Some(history_create),
                safety: MigrationSafety::Safe,
                risk: MigrationRisk::None,
                risk_detail: None,
            });
            steps.push(MigrationStep {
                step: 0,
                operation: MigrationOperation::CreateIndex,
                schema: schema.clone(),
                table: Some(format!("{}_history", new_table.name)),
                object: format!("{}_history_{}_idx", new_table.name, pk_col),
                object_type: "index".into(),
                from_object: None,
                description: format!(
                    "Create index on history table \"{}\".\"{}\" ({pk_col}, _version DESC)",
                    schema, new_table.name
                ),
                ddl: Some(history_index_ddl(&schema, &new_table.name, &pk_col)),
                safety: MigrationSafety::Safe,
                risk: MigrationRisk::None,
                risk_detail: None,
            });
        }
    }

    // Existing tables that gained audit_log or versioning
    for new_table in &new.tables {
        if added_table_ids.contains(new_table.id.as_str()) {
            continue;
        }
        if let Some(old_table) = old_tables.get(new_table.id.as_str()) {
            let sid = new_table.schema_id.as_deref().unwrap_or(default_new_sid);
            let schema = schema_name_for(sid, &new_schemas);
            let cols = cols_by_table
                .get(new_table.id.as_str())
                .map(|v| v.as_slice())
                .unwrap_or(&[]);

            if !old_table.audit_log && new_table.audit_log {
                let audit_ddl = audit_table_ddl(&schema, &new_table.name, cols, dialect);
                steps.push(MigrationStep {
                    step: 0,
                    operation: MigrationOperation::CreateTable,
                    schema: schema.clone(),
                    table: Some(format!("{}_audit", new_table.name)),
                    object: format!("{}_audit", new_table.name),
                    object_type: "table".into(),
                    from_object: None,
                    description: format!(
                        "Enable audit log: create \"{}\".\"{}_audit\"",
                        schema, new_table.name
                    ),
                    ddl: Some(audit_ddl),
                    safety: MigrationSafety::Safe,
                    risk: MigrationRisk::None,
                    risk_detail: None,
                });
            }

            let old_versioning_enabled = old_table.versioning.as_ref().is_some_and(|v| v.enabled);
            let new_versioning_enabled = new_table.versioning.as_ref().is_some_and(|v| v.enabled);
            if !old_versioning_enabled && new_versioning_enabled {
                let pk_col = match &new_table.primary_key {
                    PrimaryKeyConfig::Single(s) => s.clone(),
                    PrimaryKeyConfig::Composite(v) => v[0].clone(),
                };
                let history_ddl =
                    history_table_ddl(&schema, &new_table.name, &pk_col, cols, dialect);
                let create_only = history_ddl
                    .lines()
                    .take_while(|l| !l.trim_start().starts_with("-- index:"))
                    .collect::<Vec<_>>()
                    .join("\n");
                steps.push(MigrationStep {
                    step: 0,
                    operation: MigrationOperation::CreateTable,
                    schema: schema.clone(),
                    table: Some(format!("{}_history", new_table.name)),
                    object: format!("{}_history", new_table.name),
                    object_type: "table".into(),
                    from_object: None,
                    description: format!(
                        "Enable versioning: create \"{}\".\"{}_history\"",
                        schema, new_table.name
                    ),
                    ddl: Some(create_only.trim().to_string()),
                    safety: MigrationSafety::Safe,
                    risk: MigrationRisk::None,
                    risk_detail: None,
                });
                steps.push(MigrationStep {
                    step: 0,
                    operation: MigrationOperation::CreateIndex,
                    schema: schema.clone(),
                    table: Some(format!("{}_history", new_table.name)),
                    object: format!("{}_history_{}_idx", new_table.name, pk_col),
                    object_type: "index".into(),
                    from_object: None,
                    description: format!(
                        "Create history index on \"{}\".\"{}\"",
                        schema, new_table.name
                    ),
                    ddl: Some(history_index_ddl(&schema, &new_table.name, &pk_col)),
                    safety: MigrationSafety::Safe,
                    risk: MigrationRisk::None,
                    risk_detail: None,
                });
            }
        }
    }

    for old_table in &old.tables {
        if !new_tables.contains_key(old_table.id.as_str()) {
            let sid = old_table.schema_id.as_deref().unwrap_or(default_old_sid);
            let schema = schema_name_for(sid, &old_schemas);
            steps.push(MigrationStep {
                step: 0,
                operation: MigrationOperation::DropTable,
                schema: schema.clone(),
                table: Some(old_table.name.clone()),
                object: old_table.name.clone(),
                object_type: "table".into(),
                from_object: None,
                description: format!("Table \"{}\".\"{}\" removed from config", schema, old_table.name),
                ddl: None,
                safety: MigrationSafety::WarnOnly,
                risk: MigrationRisk::ManualActionRequired,
                risk_detail: Some("Table NOT dropped from database (data safety). Run DROP TABLE manually if intended.".into()),
            });
        }
    }

    // ── 4. Column changes for existing tables ────────────────────────────────
    for new_col in &new.columns {
        if added_table_ids.contains(new_col.table_id.as_str()) {
            continue;
        }
        let table = match new_tables.get(new_col.table_id.as_str()) {
            Some(t) => t,
            None => continue,
        };
        let sid = table.schema_id.as_deref().unwrap_or(default_new_sid);
        let schema = schema_name_for(sid, &new_schemas);
        let full = format!("{}.{}", quote(&schema), quote(&table.name));

        if let Some(old_col) = old_columns.get(new_col.id.as_str()) {
            if old_col.table_id != new_col.table_id {
                steps.push(MigrationStep {
                    step: 0,
                    operation: MigrationOperation::AddColumn,
                    schema: schema.clone(),
                    table: Some(table.name.clone()),
                    object: new_col.name.clone(),
                    object_type: "column".into(),
                    from_object: None,
                    description: format!("Column \"{}\" (id: {}) appears to have moved tables — manual migration required", new_col.name, new_col.id),
                    ddl: None,
                    safety: MigrationSafety::WarnOnly,
                    risk: MigrationRisk::ManualActionRequired,
                    risk_detail: Some(format!("Cannot automate column move from table {} to {}.", old_col.table_id, new_col.table_id)),
                });
                continue;
            }

            // Rename
            if old_col.name != new_col.name {
                steps.push(MigrationStep {
                    step: 0,
                    operation: MigrationOperation::RenameColumn,
                    schema: schema.clone(),
                    table: Some(table.name.clone()),
                    object: new_col.name.clone(),
                    object_type: "column".into(),
                    from_object: Some(old_col.name.clone()),
                    description: format!(
                        "Rename column \"{}\" → \"{}\" on \"{}\".\"{}\"",
                        old_col.name, new_col.name, schema, table.name
                    ),
                    ddl: Some(format!(
                        "ALTER TABLE {} RENAME COLUMN {} TO {}",
                        full,
                        quote(&old_col.name),
                        quote(&new_col.name)
                    )),
                    safety: MigrationSafety::Safe,
                    risk: MigrationRisk::None,
                    risk_detail: None,
                });
                steps.extend(companion_column_steps(
                    &schema,
                    table,
                    &CompanionColumnOp::Rename {
                        old: &old_col.name,
                        new: &new_col.name,
                    },
                    dialect,
                ));
            }

            // Type change
            let old_type = dialect.ddl_type(&parse_canonical(&old_col.type_));
            let new_type = dialect.ddl_type(&parse_canonical(&new_col.type_));
            if old_type.to_uppercase() != new_type.to_uppercase() {
                let col_name = &new_col.name;
                steps.push(MigrationStep {
                    step: 0,
                    operation: MigrationOperation::AlterColumnType,
                    schema: schema.clone(),
                    table: Some(table.name.clone()),
                    object: col_name.clone(),
                    object_type: "column".into(),
                    from_object: None,
                    description: format!("Change type of \"{}\".\"{}\".\"{}\": {} → {}", schema, table.name, col_name, old_type, new_type),
                    ddl: Some(format!("ALTER TABLE {} ALTER COLUMN {} TYPE {} USING {}::{}", full, quote(col_name), new_type, quote(col_name), new_type)),
                    safety: MigrationSafety::BestEffort,
                    risk: MigrationRisk::MayFail,
                    risk_detail: Some(format!("USING {}::{} cast may fail for incompatible values. Provide a custom USING expression if needed.", col_name, new_type)),
                });
                steps.extend(companion_column_steps(
                    &schema,
                    table,
                    &CompanionColumnOp::AlterType {
                        name: col_name.as_str(),
                        ty: &new_type,
                    },
                    dialect,
                ));
            }

            // Nullability: nullable → NOT NULL
            if old_col.nullable && !new_col.nullable {
                if let Some(ref d) = new_col.default {
                    let default_val = default_str(d);
                    // Backfill NULLs first using the configured default
                    steps.push(MigrationStep {
                        step: 0,
                        operation: MigrationOperation::BackfillNulls,
                        schema: schema.clone(),
                        table: Some(table.name.clone()),
                        object: new_col.name.clone(),
                        object_type: "column".into(),
                        from_object: None,
                        description: format!("Backfill NULLs in \"{}\".\"{}\".\"{}\": SET {} = {} WHERE {} IS NULL", schema, table.name, new_col.name, new_col.name, default_val, new_col.name),
                        ddl: Some(format!("UPDATE {} SET {} = {} WHERE {} IS NULL", full, quote(&new_col.name), default_val, quote(&new_col.name))),
                        safety: MigrationSafety::Safe,
                        risk: MigrationRisk::DataWillBeModified,
                        risk_detail: Some(format!("Existing NULLs in column \"{}\" will be set to {} before NOT NULL is enforced.", new_col.name, default_val)),
                    });
                    // Then set NOT NULL — safe because NULLs are gone
                    steps.push(MigrationStep {
                        step: 0,
                        operation: MigrationOperation::SetNotNull,
                        schema: schema.clone(),
                        table: Some(table.name.clone()),
                        object: new_col.name.clone(),
                        object_type: "column".into(),
                        from_object: None,
                        description: format!("Set NOT NULL on \"{}\".\"{}\".\"{}\": NULLs pre-filled with default ({})", schema, table.name, new_col.name, default_val),
                        ddl: Some(format!("ALTER TABLE {} ALTER COLUMN {} SET NOT NULL", full, quote(&new_col.name))),
                        safety: MigrationSafety::Safe,
                        risk: MigrationRisk::None,
                        risk_detail: None,
                    });
                } else {
                    // No default — best effort; will fail if NULLs exist
                    steps.push(MigrationStep {
                        step: 0,
                        operation: MigrationOperation::SetNotNull,
                        schema: schema.clone(),
                        table: Some(table.name.clone()),
                        object: new_col.name.clone(),
                        object_type: "column".into(),
                        from_object: None,
                        description: format!("Set NOT NULL on \"{}\".\"{}\".\"{}\": no default configured — will fail if NULLs exist", schema, table.name, new_col.name),
                        ddl: Some(format!("ALTER TABLE {} ALTER COLUMN {} SET NOT NULL", full, quote(&new_col.name))),
                        safety: MigrationSafety::BestEffort,
                        risk: MigrationRisk::ExistingNullsMustBeAbsent,
                        risk_detail: Some(format!(
                            "No default value configured for column \"{}\". Add a default to the config to enable automatic NULL backfill before enforcing NOT NULL.",
                            new_col.name
                        )),
                    });
                }
            }

            // Nullability: NOT NULL → nullable
            if !old_col.nullable && new_col.nullable {
                steps.push(MigrationStep {
                    step: 0,
                    operation: MigrationOperation::DropNotNull,
                    schema: schema.clone(),
                    table: Some(table.name.clone()),
                    object: new_col.name.clone(),
                    object_type: "column".into(),
                    from_object: None,
                    description: format!(
                        "Drop NOT NULL on \"{}\".\"{}\".\"{}\": column becomes nullable",
                        schema, table.name, new_col.name
                    ),
                    ddl: Some(format!(
                        "ALTER TABLE {} ALTER COLUMN {} DROP NOT NULL",
                        full,
                        quote(&new_col.name)
                    )),
                    safety: MigrationSafety::Safe,
                    risk: MigrationRisk::None,
                    risk_detail: None,
                });
            }

            // Default change
            let old_def = old_col.default.as_ref().map(default_str);
            let new_def = new_col.default.as_ref().map(default_str);
            if old_def != new_def {
                match &new_col.default {
                    Some(d) => {
                        let val = default_str(d);
                        steps.push(MigrationStep {
                            step: 0,
                            operation: MigrationOperation::SetDefault,
                            schema: schema.clone(),
                            table: Some(table.name.clone()),
                            object: new_col.name.clone(),
                            object_type: "column".into(),
                            from_object: None,
                            description: format!(
                                "Set DEFAULT {} on \"{}\".\"{}\".\"{}\": was {}",
                                val,
                                schema,
                                table.name,
                                new_col.name,
                                old_def.as_deref().unwrap_or("none")
                            ),
                            ddl: Some(format!(
                                "ALTER TABLE {} ALTER COLUMN {} SET DEFAULT {}",
                                full,
                                quote(&new_col.name),
                                val
                            )),
                            safety: MigrationSafety::Safe,
                            risk: MigrationRisk::None,
                            risk_detail: None,
                        });
                    }
                    None => {
                        steps.push(MigrationStep {
                            step: 0,
                            operation: MigrationOperation::DropDefault,
                            schema: schema.clone(),
                            table: Some(table.name.clone()),
                            object: new_col.name.clone(),
                            object_type: "column".into(),
                            from_object: None,
                            description: format!(
                                "Drop DEFAULT on \"{}\".\"{}\".\"{}\": was {}",
                                schema,
                                table.name,
                                new_col.name,
                                old_def.as_deref().unwrap_or("none")
                            ),
                            ddl: Some(format!(
                                "ALTER TABLE {} ALTER COLUMN {} DROP DEFAULT",
                                full,
                                quote(&new_col.name)
                            )),
                            safety: MigrationSafety::Safe,
                            risk: MigrationRisk::None,
                            risk_detail: None,
                        });
                    }
                }
            }
        } else {
            // New column: ADD COLUMN
            let new_type = dialect.ddl_type(&parse_canonical(&new_col.type_));
            let mut col_def = format!("{} {}", quote(&new_col.name), new_type);
            if !new_col.nullable {
                col_def.push_str(" NOT NULL");
            }
            if let Some(ref d) = new_col.default {
                col_def.push_str(" DEFAULT ");
                match d {
                    ColumnDefaultConfig::Literal(s) => col_def.push_str(s),
                    ColumnDefaultConfig::Expression { expression } => col_def.push_str(expression),
                }
            }
            steps.push(MigrationStep {
                step: 0,
                operation: MigrationOperation::AddColumn,
                schema: schema.clone(),
                table: Some(table.name.clone()),
                object: new_col.name.clone(),
                object_type: "column".into(),
                from_object: None,
                description: format!(
                    "Add column \"{}\" {} to \"{}\".\"{}\"",
                    new_col.name, new_type, schema, table.name
                ),
                ddl: Some(add_column_ddl(dialect, &full, &col_def)),
                safety: MigrationSafety::Safe,
                risk: MigrationRisk::None,
                risk_detail: None,
            });
            steps.extend(companion_column_steps(
                &schema,
                table,
                &CompanionColumnOp::Add {
                    name: &new_col.name,
                    ty: &new_type,
                },
                dialect,
            ));
        }
    }

    // Removed columns (warn only)
    for old_col in &old.columns {
        if new.columns.iter().any(|c| c.id == old_col.id) {
            continue;
        }
        if !new_tables.contains_key(old_col.table_id.as_str()) {
            continue;
        }
        let table_name = old_tables
            .get(old_col.table_id.as_str())
            .map(|t| t.name.as_str())
            .unwrap_or(&old_col.table_id);
        let sid = old_tables
            .get(old_col.table_id.as_str())
            .and_then(|t| t.schema_id.as_deref())
            .unwrap_or(default_old_sid);
        let schema = schema_name_for(sid, &old_schemas);
        steps.push(MigrationStep {
            step: 0,
            operation: MigrationOperation::DropColumn,
            schema: schema.clone(),
            table: Some(table_name.to_string()),
            object: old_col.name.clone(),
            object_type: "column".into(),
            from_object: None,
            description: format!("Column \"{}\" removed from config on \"{}\".\"{}\"", old_col.name, schema, table_name),
            ddl: None,
            safety: MigrationSafety::WarnOnly,
            risk: MigrationRisk::ManualActionRequired,
            risk_detail: Some("Column NOT dropped from database (data safety). Run ALTER TABLE DROP COLUMN manually if intended.".into()),
        });
    }

    // ── 5. Indexes ───────────────────────────────────────────────────────────
    for old_idx in &old.indexes {
        if !new_indexes.contains_key(old_idx.id.as_str()) {
            let sid = old_idx.schema_id.as_deref().unwrap_or(default_old_sid);
            let schema = schema_name_for(sid, &old_schemas);
            steps.push(MigrationStep {
                step: 0,
                operation: MigrationOperation::DropIndex,
                schema: schema.clone(),
                table: old_tables
                    .get(old_idx.table_id.as_str())
                    .map(|t| t.name.clone()),
                object: old_idx.name.clone(),
                object_type: "index".into(),
                from_object: None,
                description: format!("Drop index \"{}\" in schema \"{}\"", old_idx.name, schema),
                ddl: Some(format!(
                    "DROP INDEX IF EXISTS {}.{}",
                    quote(&schema),
                    quote(&old_idx.name)
                )),
                safety: MigrationSafety::Safe,
                risk: MigrationRisk::None,
                risk_detail: None,
            });
        }
    }
    for new_idx in &new.indexes {
        if old_indexes.contains_key(new_idx.id.as_str())
            || added_table_ids.contains(new_idx.table_id.as_str())
        {
            continue;
        }
        let sid = new_idx.schema_id.as_deref().unwrap_or(default_new_sid);
        let schema = match new_schemas.get(sid) {
            Some(s) => schema_override.unwrap_or(&s.name).to_string(),
            None => continue,
        };
        let table = match new_tables.get(new_idx.table_id.as_str()) {
            Some(t) => t,
            None => continue,
        };
        let full_table = format!("{}.{}", quote(&schema), quote(&table.name));
        let mut col_parts: Vec<String> = Vec::new();
        for col in &new_idx.columns {
            match col {
                IndexColumnEntry::Name(n) => col_parts.push(quote(n)),
                IndexColumnEntry::Spec {
                    name, direction, ..
                } => {
                    let dir = direction
                        .as_deref()
                        .map(|d| format!(" {}", d.to_uppercase()))
                        .unwrap_or_default();
                    col_parts.push(format!("{}{}", quote(name), dir));
                }
                IndexColumnEntry::Expression { expression } => col_parts.push(expression.clone()),
            }
        }
        let method = new_idx.method.as_deref().unwrap_or("btree");
        let unique_kw = if new_idx.unique { "UNIQUE " } else { "" };
        let include = if new_idx.include.is_empty() {
            String::new()
        } else {
            format!(
                " INCLUDE ({})",
                new_idx
                    .include
                    .iter()
                    .map(|s| quote(s))
                    .collect::<Vec<_>>()
                    .join(", ")
            )
        };
        let where_clause = new_idx
            .where_
            .as_ref()
            .map(|w| format!(" WHERE {}", w))
            .unwrap_or_default();
        steps.push(MigrationStep {
            step: 0,
            operation: MigrationOperation::CreateIndex,
            schema: schema.clone(),
            table: Some(table.name.clone()),
            object: new_idx.name.clone(),
            object_type: "index".into(),
            from_object: None,
            description: format!(
                "Create {}index \"{}\" on \"{}\".\"{}\"",
                if new_idx.unique { "unique " } else { "" },
                new_idx.name,
                schema,
                table.name
            ),
            ddl: Some(format!(
                "CREATE {}INDEX IF NOT EXISTS {} ON {} USING {} ({}){}{}",
                unique_kw,
                quote(&new_idx.name),
                full_table,
                method,
                col_parts.join(", "),
                include,
                where_clause
            )),
            safety: MigrationSafety::Safe,
            risk: MigrationRisk::None,
            risk_detail: None,
        });
    }

    // ── 6. Foreign keys ──────────────────────────────────────────────────────
    for old_rel in &old.relationships {
        if !new_rels.contains_key(old_rel.id.as_str()) {
            let from_sid_fallback = old_rel.from_schema_id.as_deref().unwrap_or(default_old_sid);
            let from_schema = old_schemas
                .get(from_sid_fallback)
                .map(|s| s.name.as_str())
                .unwrap_or(from_sid_fallback);
            let from_table = old_tables
                .get(old_rel.from_table_id.as_str())
                .map(|t| t.name.as_str())
                .unwrap_or(&old_rel.from_table_id);
            let constraint = old_rel.name.as_deref().unwrap_or(&old_rel.id);
            let schema_q = quote(schema_override.unwrap_or(from_schema));
            steps.push(MigrationStep {
                step: 0,
                operation: MigrationOperation::DropForeignKey,
                schema: schema_override.unwrap_or(from_schema).to_string(),
                table: Some(from_table.to_string()),
                object: constraint.to_string(),
                object_type: "foreign_key".into(),
                from_object: None,
                description: format!(
                    "Drop FK \"{}\" from \"{}\".\"{}\"",
                    constraint,
                    schema_override.unwrap_or(from_schema),
                    from_table
                ),
                ddl: Some(format!(
                    "ALTER TABLE {}.{} DROP CONSTRAINT IF EXISTS {}",
                    schema_q,
                    quote(from_table),
                    quote(constraint)
                )),
                safety: MigrationSafety::Safe,
                risk: MigrationRisk::None,
                risk_detail: None,
            });
        }
    }
    for new_rel in &new.relationships {
        if old_rels.contains_key(new_rel.id.as_str())
            || added_table_ids.contains(new_rel.from_table_id.as_str())
            || added_table_ids.contains(new_rel.to_table_id.as_str())
        {
            continue;
        }
        let from_sid = new_rel.from_schema_id.as_deref().unwrap_or(default_new_sid);
        let from_schema = match new_schemas.get(from_sid) {
            Some(s) => s,
            None => continue,
        };
        let from_table = match new_tables.get(new_rel.from_table_id.as_str()) {
            Some(t) => t,
            None => continue,
        };
        let from_col = new
            .columns
            .iter()
            .find(|c| c.id == new_rel.from_column_id)
            .map(|c| c.name.clone())
            .unwrap_or_else(|| new_rel.from_column_id.clone());

        // Resolve the target side — cross-package or same-package.
        let (to_schema_name, to_table_name, to_col) =
            if let Some(pkg_id) = new_rel.to_package_id.as_deref() {
                match cross_package_configs.get(pkg_id) {
                    Some(foreign) => {
                        let foreign_tables: HashMap<_, _> =
                            foreign.tables.iter().map(|t| (t.id.as_str(), t)).collect();
                        let foreign_schemas: HashMap<_, _> =
                            foreign.schemas.iter().map(|s| (s.id.as_str(), s)).collect();
                        let foreign_default_sid =
                            foreign.schemas.first().map(|s| s.id.as_str()).unwrap_or("");
                        let to_sid = new_rel
                            .to_schema_id
                            .as_deref()
                            .unwrap_or(foreign_default_sid);
                        let tbl = match foreign_tables.get(new_rel.to_table_id.as_str()) {
                            Some(t) => t,
                            None => continue,
                        };
                        let schema = match foreign_schemas.get(to_sid) {
                            Some(s) => s,
                            None => continue,
                        };
                        let col = foreign
                            .columns
                            .iter()
                            .find(|c| c.id == new_rel.to_column_id)
                            .map(|c| c.name.clone())
                            .unwrap_or_else(|| new_rel.to_column_id.clone());
                        (schema.name.clone(), tbl.name.clone(), col)
                    }
                    None => continue,
                }
            } else {
                let to_sid = new_rel.to_schema_id.as_deref().unwrap_or(default_new_sid);
                let to_schema = match new_schemas.get(to_sid) {
                    Some(s) => s,
                    None => continue,
                };
                let to_table = match new_tables.get(new_rel.to_table_id.as_str()) {
                    Some(t) => t,
                    None => continue,
                };
                let col = new
                    .columns
                    .iter()
                    .find(|c| c.id == new_rel.to_column_id)
                    .map(|c| c.name.clone())
                    .unwrap_or_else(|| new_rel.to_column_id.clone());
                (
                    schema_override.unwrap_or(&to_schema.name).to_string(),
                    to_table.name.clone(),
                    col,
                )
            };

        let from_schema_str = schema_override.unwrap_or(&from_schema.name);
        let from_q = format!("{}.{}", quote(from_schema_str), quote(&from_table.name));
        let to_q = format!("{}.{}", quote(&to_schema_name), quote(&to_table_name));
        let constraint = new_rel.name.as_deref().unwrap_or(&new_rel.id);
        let on_update = new_rel.on_update.as_deref().unwrap_or("NO ACTION");
        let on_delete = new_rel.on_delete.as_deref().unwrap_or("NO ACTION");
        steps.push(MigrationStep {
            step: 0,
            operation: MigrationOperation::AddForeignKey,
            schema: from_schema_str.to_string(),
            table: Some(from_table.name.clone()),
            object: constraint.to_string(),
            object_type: "foreign_key".into(),
            from_object: None,
            description: format!(
                "Add FK \"{}\" on \"{}\".\"{}\" → \"{}\".\"{}\"",
                constraint, from_schema_str, from_table.name, to_schema_name, to_table_name
            ),
            ddl: Some(format!(
                "ALTER TABLE {} ADD CONSTRAINT {} FOREIGN KEY ({}) REFERENCES {} ({}) ON UPDATE {} ON DELETE {}",
                from_q, quote(constraint), quote(&from_col), to_q, quote(&to_col), on_update, on_delete
            )),
            safety: MigrationSafety::BestEffort,
            risk: MigrationRisk::None,
            risk_detail: Some("PostgreSQL has no ADD CONSTRAINT IF NOT EXISTS; ignored if constraint already exists.".into()),
        });
    }

    // Assign sequential step numbers
    for (i, s) in steps.iter_mut().enumerate() {
        s.step = i + 1;
    }

    Ok(MigrationPlan { steps })
}

// ─── Plan reconciliation against the physical database ───────────────────────

/// Whether a plan step still has work to do against a particular database.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum StepDecision {
    /// Run the step's DDL.
    Execute,
    /// The step's effect is already present; the payload explains why it was skipped.
    Skip(String),
}

/// Decide whether `step` needs to run against the database described by `snap`.
///
/// Migration plans are a pure config-vs-config diff, so they assume the database matches the
/// *old* config exactly. It often does not: an upgrade may have failed halfway and been retried,
/// one plan is broadcast to many tenant databases at different states, and RLS/companion DDL adds
/// columns outside the diff. Rather than aborting on `column "x" already exists`, skip the steps
/// whose outcome is already true.
///
/// Negative rules ("the column is missing, so this step cannot run") only fire when the table is
/// actually known to the snapshot — an empty or failed introspection therefore skips nothing.
pub fn reconcile_step(step: &MigrationStep, snap: &DbSnapshot) -> StepDecision {
    if step.ddl.is_none() || !snap.introspected {
        return StepDecision::Execute;
    }
    let schema = step.schema.as_str();
    let object = step.object.as_str();
    let Some(table) = step.table.as_deref() else {
        return StepDecision::Execute;
    };
    let table_known = snap.has_table(schema, table);
    let column = snap.column(schema, table, object);
    // Only meaningful for a table we could actually read: an unknown table means the snapshot
    // has nothing to say, not that the column is missing.
    let column_gone = table_known && column.is_none();
    let missing_column = || {
        format!(
            "column \"{}\" does not exist on \"{}\".\"{}\"",
            object, schema, table
        )
    };

    use MigrationOperation as Op;
    let reason: Option<String> = match step.operation {
        Op::CreateTable if table_known => {
            Some(format!("table \"{}\".\"{}\" already exists", schema, table))
        }
        Op::AddColumn if column.is_some() => Some(format!(
            "column \"{}\" already exists on \"{}\".\"{}\"",
            object, schema, table
        )),
        Op::RenameColumn => rename_skip_reason(step, snap, schema, table, table_known),

        // A column that is gone cannot be retyped, defaulted or backfilled.
        Op::AlterColumnType | Op::BackfillNulls | Op::SetDefault if column_gone => {
            Some(missing_column())
        }

        Op::SetNotNull if column.is_some_and(|f| !f.nullable) => Some(format!(
            "column \"{}\".\"{}\".\"{}\" is already NOT NULL",
            schema, table, object
        )),
        Op::DropNotNull if column.is_some_and(|f| f.nullable) => Some(format!(
            "column \"{}\".\"{}\".\"{}\" is already nullable",
            schema, table, object
        )),
        Op::DropDefault if column.is_some_and(|f| !f.has_default) => Some(format!(
            "column \"{}\".\"{}\".\"{}\" has no DEFAULT",
            schema, table, object
        )),
        Op::SetNotNull | Op::DropNotNull | Op::DropDefault if column_gone => Some(missing_column()),

        Op::CreateIndex if snap.indexes_known && snap.has_index(schema, object) => Some(format!(
            "index \"{}\".\"{}\" already exists",
            schema, object
        )),
        Op::AddForeignKey
            if snap.constraints_known && snap.has_constraint(schema, table, object) =>
        {
            Some(format!(
                "constraint \"{}\" already exists on \"{}\".\"{}\"",
                object, schema, table
            ))
        }

        // Drops already carry IF EXISTS, and enum steps are either IF NOT EXISTS or part of a
        // rename/recreate sequence whose intermediate state a pre-flight snapshot cannot describe.
        _ => None,
    };

    match reason {
        Some(r) => StepDecision::Skip(r),
        None => StepDecision::Execute,
    }
}

/// Skip reason for a `RenameColumn` step, or `None` when the rename still has work to do.
fn rename_skip_reason(
    step: &MigrationStep,
    snap: &DbSnapshot,
    schema: &str,
    table: &str,
    table_known: bool,
) -> Option<String> {
    // Plans saved before `from_object` existed carry no previous name; nothing to check, so let
    // the statement run.
    let from = step.from_object.as_deref()?;
    if !table_known {
        return None;
    }
    let to = step.object.as_str();
    match (
        snap.has_column(schema, table, from),
        snap.has_column(schema, table, to),
    ) {
        (false, true) => Some(format!(
            "column \"{}\".\"{}\".\"{}\" is already named \"{}\"",
            schema, table, from, to
        )),
        (false, false) => Some(format!(
            "neither \"{}\" nor \"{}\" exists on \"{}\".\"{}\" — nothing to rename",
            from, to, schema, table
        )),
        _ => None,
    }
}

/// Fold a successfully executed step into `snap` so later steps in the same plan see the state
/// the earlier ones produced (e.g. a column added and then made NOT NULL).
///
/// `CreateTable` is deliberately not recorded: registering a table with no known columns would
/// make later column steps look impossible and get them wrongly skipped.
fn apply_step_to_snapshot(step: &MigrationStep, snap: &mut DbSnapshot) {
    let schema = step.schema.as_str();
    let object = step.object.as_str();
    let table = step.table.as_deref();
    // `from_object` rides in the tuple so the rename arm needs no nested conditional.
    match (step.operation.clone(), table, step.from_object.as_deref()) {
        (MigrationOperation::AddColumn, Some(t), _) => {
            let ddl = step.ddl.as_deref().unwrap_or("").to_uppercase();
            snap.add_column(
                schema,
                t,
                object,
                ColumnFacts {
                    data_type: String::new(),
                    nullable: !ddl.contains(" NOT NULL"),
                    has_default: ddl.contains(" DEFAULT "),
                },
            );
        }
        (MigrationOperation::RenameColumn, Some(t), Some(from)) => {
            snap.rename_column(schema, t, from, object)
        }
        (MigrationOperation::SetNotNull, Some(t), _) => snap.set_nullable(schema, t, object, false),
        (MigrationOperation::DropNotNull, Some(t), _) => snap.set_nullable(schema, t, object, true),
        (MigrationOperation::SetDefault, Some(t), _) => {
            snap.set_has_default(schema, t, object, true)
        }
        (MigrationOperation::DropDefault, Some(t), _) => {
            snap.set_has_default(schema, t, object, false)
        }
        (MigrationOperation::CreateIndex, _, _) => snap.add_index(schema, object),
        (MigrationOperation::DropIndex, _, _) => snap.remove_index(schema, object),
        (MigrationOperation::AddForeignKey, Some(t), _) => snap.add_constraint(schema, t, object),
        (MigrationOperation::DropForeignKey, Some(t), _) => {
            snap.remove_constraint(schema, t, object)
        }
        _ => {}
    }
}

/// Recognise "the object I tried to create already exists" errors.
///
/// Backstop for whatever pre-flight introspection could not see: a schema it had no privileges
/// on, an object created by a concurrent migration, or a dialect that reports no index and
/// constraint catalogs.
fn duplicate_object_reason(dialect: &dyn Dialect, e: &sqlx::Error) -> Option<String> {
    let dbe = e.as_database_error()?;
    let duplicate_code = dbe
        .code()
        .is_some_and(|code| dialect.is_duplicate_object_code(code.as_ref()));
    if duplicate_code {
        return Some(format!(
            "object already exists (SQLSTATE {}): {}",
            dbe.code().unwrap_or_default(),
            dbe.message()
        ));
    }
    let msg = dbe.message().to_ascii_lowercase();
    if msg.contains("already exists")
        || msg.contains("duplicate column name")
        || msg.contains("duplicate key name")
    {
        return Some(format!("object already exists: {}", dbe.message()));
    }
    None
}

// ─── execute_migration_plan ──────────────────────────────────────────────────

/// Identifiers of the migration being executed, threaded into every audit row.
struct AuditContext<'a> {
    migration_plan_id: &'a str,
    package_id: &'a str,
    tenant_id: &'a str,
    from_version: Option<&'a str>,
    to_version: &'a str,
}

async fn audit_step(
    config_pool: &Pool,
    ctx: &AuditContext<'_>,
    step: &MigrationStep,
    status: &str,
    error_message: Option<&str>,
) {
    let _ = crate::store::insert_migration_audit(
        config_pool,
        ctx.migration_plan_id,
        ctx.package_id,
        ctx.tenant_id,
        ctx.from_version,
        ctx.to_version,
        step.step as i32,
        &step.operation.to_string(),
        &step.schema,
        step.table.as_deref(),
        &step.object,
        &step.object_type,
        &step.description,
        step.ddl.as_deref(),
        &format!("{:?}", step.safety),
        &format!("{:?}", step.risk),
        status,
        error_message,
    )
    .await;
}

/// Execute a pre-computed `MigrationPlan` against the tenant database.
///
/// The plan is reconciled against the database's actual state first (see [`reconcile_step`]), so
/// steps whose effect is already present are skipped instead of failing the whole migration.
/// Writes per-step audit records to the config (architect) database.
/// Returns counts and any warning messages collected from best-effort failures.
#[allow(clippy::too_many_arguments)]
pub async fn execute_migration_plan(
    migration_pool: &Pool,
    config_pool: &Pool,
    plan: &MigrationPlan,
    migration_plan_id: &str,
    package_id: &str,
    tenant_id: &str,
    from_version: Option<&str>,
    to_version: &str,
    dialect: &dyn Dialect,
) -> Result<MigrationExecutionResult, AppError> {
    let mut applied = 0usize;
    let mut warned = 0usize;
    let mut skipped = 0usize;
    let mut warnings: Vec<String> = Vec::new();
    let mut skips: Vec<String> = Vec::new();

    let ctx = AuditContext {
        migration_plan_id,
        package_id,
        tenant_id,
        from_version,
        to_version,
    };

    // Pre-flight: what does this database actually look like? Best-effort — on failure the
    // snapshot stays empty, nothing is skipped, and execution behaves exactly as before.
    let schemas: Vec<String> = {
        let mut seen: Vec<String> = Vec::new();
        for step in &plan.steps {
            if step.ddl.is_some() && !seen.iter().any(|s| s == &step.schema) {
                seen.push(step.schema.clone());
            }
        }
        seen
    };
    let mut snapshot = crate::db::introspect(migration_pool, dialect, &schemas).await;

    for step in &plan.steps {
        let op = step.operation.to_string();

        match step.safety {
            MigrationSafety::WarnOnly => {
                let msg = step
                    .risk_detail
                    .clone()
                    .unwrap_or_else(|| step.description.clone());
                tracing::warn!(step = step.step, %op, "migration plan warning (no DDL)");
                warnings.push(format!("[Step {}] {}", step.step, msg));
                audit_step(config_pool, &ctx, step, "skipped", None).await;
                warned += 1;
            }
            MigrationSafety::Safe | MigrationSafety::BestEffort => {
                let Some(ref sql) = step.ddl else {
                    continue;
                };

                if let StepDecision::Skip(reason) = reconcile_step(step, &snapshot) {
                    tracing::info!(step = step.step, %op, %reason, "migration step already satisfied — skipping");
                    skips.push(format!(
                        "[Step {}] {} — {}",
                        step.step, step.description, reason
                    ));
                    audit_step(config_pool, &ctx, step, "skipped_exists", Some(&reason)).await;
                    skipped += 1;
                    continue;
                }

                tracing::info!(step = step.step, %op, %sql, "executing migration step");
                match sqlx::query(sql).execute(migration_pool).await {
                    Ok(_) => {
                        apply_step_to_snapshot(step, &mut snapshot);
                        audit_step(config_pool, &ctx, step, "applied", None).await;
                        applied += 1;
                    }
                    Err(e) => {
                        // The object already exists: the step's effect is in place, whatever the
                        // config diff believed. Record it and carry on.
                        if let Some(reason) = duplicate_object_reason(dialect, &e) {
                            tracing::info!(step = step.step, %op, %reason, "migration step already satisfied — skipping");
                            skips.push(format!(
                                "[Step {}] {} — {}",
                                step.step, step.description, reason
                            ));
                            audit_step(config_pool, &ctx, step, "skipped_exists", Some(&reason))
                                .await;
                            apply_step_to_snapshot(step, &mut snapshot);
                            skipped += 1;
                            continue;
                        }

                        let err_str = e.to_string();
                        if matches!(step.safety, MigrationSafety::BestEffort) {
                            tracing::warn!(step = step.step, %op, error = %e, "migration step failed (best-effort, continuing)");
                            warnings.push(format!(
                                "[Step {}] {} — Error: {}",
                                step.step, step.description, err_str
                            ));
                            audit_step(config_pool, &ctx, step, "warned", Some(&err_str)).await;
                            warned += 1;
                        } else {
                            audit_step(config_pool, &ctx, step, "failed", Some(&err_str)).await;
                            return Err(AppError::Db(e));
                        }
                    }
                }
            }
        }
    }

    Ok(MigrationExecutionResult {
        applied,
        warned,
        skipped,
        warnings,
        skips,
    })
}

/// Build CREATE TABLE DDL for the `{table}_history` companion table used by row versioning.
/// All source columns are replicated with their types but as nullable and without any constraints
/// (no NOT NULL, no UNIQUE, no FK, no CHECK). Five versioning metadata columns are prepended.
pub fn history_table_ddl(
    schema_name: &str,
    table_name: &str,
    pk_col: &str,
    source_cols: &[&ColumnConfig],
    dialect: &dyn Dialect,
) -> String {
    let history_name = format!("{}_history", table_name);
    let history_full = format!("{}.{}", quote(schema_name), quote(&history_name));

    let mut col_defs: Vec<String> = Vec::new();
    col_defs.push(format!(
        "{} {} NOT NULL DEFAULT {}",
        quote("_history_id"),
        "UUID",
        dialect.uuid_default_expr()
    ));
    col_defs.push(format!("{} BIGINT NOT NULL", quote("_version")));
    col_defs.push(format!("{} TEXT NOT NULL", quote("_operation")));
    col_defs.push(format!(
        "{} {} NOT NULL DEFAULT {}",
        quote("_recorded_at"),
        dialect.audit_timestamp_type(),
        dialect.now_fn()
    ));
    col_defs.push(format!(
        "{} {}",
        quote("_valid_from"),
        dialect.audit_timestamp_type()
    ));
    col_defs.push(format!(
        "{} {}",
        quote("_valid_to"),
        dialect.audit_timestamp_type()
    ));

    let config_col_names: HashSet<&str> = source_cols.iter().map(|c| c.name.as_str()).collect();
    for c in source_cols {
        let typ = dialect.ddl_type(&parse_canonical(&c.type_));
        col_defs.push(format!("{} {}", quote(&c.name), typ));
    }
    let audit_ts = dialect.audit_timestamp_type();
    for (name, typ) in [
        ("created_at", audit_ts),
        ("updated_at", audit_ts),
        ("archived_at", audit_ts),
        ("created_by", "TEXT"),
        ("updated_by", "TEXT"),
    ] {
        if !config_col_names.contains(name) {
            col_defs.push(format!("{} {}", quote(name), typ));
        }
    }
    col_defs.push(format!("PRIMARY KEY ({})", quote("_history_id")));

    let history_full_quoted = format!("{}.{}", quote(schema_name), quote(&history_name));
    let idx_sql = format!(
        "-- index: CREATE INDEX IF NOT EXISTS {} ON {} ({}, {})",
        quote(&format!("{}_history_{}_idx", table_name, pk_col)),
        history_full_quoted,
        quote(pk_col),
        quote("_version")
    );

    format!(
        "CREATE TABLE IF NOT EXISTS {} (\n  {}\n)\n{}",
        history_full,
        col_defs.join(",\n  "),
        idx_sql
    )
}

/// Build just the index DDL for the `{table}_history` table (separate from CREATE TABLE).
fn history_index_ddl(schema_name: &str, table_name: &str, pk_col: &str) -> String {
    format!(
        "CREATE INDEX IF NOT EXISTS {} ON {}.{} ({}, {} DESC)",
        quote(&format!("{}_history_{}_idx", table_name, pk_col)),
        quote(schema_name),
        quote(&format!("{}_history", table_name)),
        quote(pk_col),
        quote("_version")
    )
}

/// A structural column change that must be mirrored onto companion (`_audit`/`_history`) tables.
enum CompanionColumnOp<'a> {
    /// A new column was added to the source table.
    Add { name: &'a str, ty: &'a str },
    /// A source column was renamed.
    Rename { old: &'a str, new: &'a str },
    /// A source column's type changed.
    AlterType { name: &'a str, ty: &'a str },
}

/// Companion-table suffixes that are enabled for a table (`audit`, `history`).
fn enabled_companion_suffixes(table: &TableConfig) -> Vec<&'static str> {
    let mut suffixes = Vec::new();
    if table.audit_log {
        suffixes.push("audit");
    }
    if table.versioning.as_ref().is_some_and(|v| v.enabled) {
        suffixes.push("history");
    }
    suffixes
}

/// Generate ALTER steps that keep the `{table}_audit` / `{table}_history` companion tables in
/// schema-sync with their source table when a column is added, renamed, or retyped.
///
/// Companion tables replicate source columns as **nullable with no constraints**, so only
/// structural changes propagate here. Nullability and default changes on the source column are
/// intentionally NOT mirrored — companion rows are historical snapshots that must stay nullable.
fn companion_column_steps(
    schema: &str,
    table: &TableConfig,
    op: &CompanionColumnOp<'_>,
    dialect: &dyn Dialect,
) -> Vec<MigrationStep> {
    let mut steps = Vec::new();
    for suffix in enabled_companion_suffixes(table) {
        let companion = format!("{}_{}", table.name, suffix);
        let full = format!("{}.{}", quote(schema), quote(&companion));
        let (operation, object, from_object, ddl, description, safety, risk, risk_detail) = match op
        {
            CompanionColumnOp::Add { name, ty } => (
                MigrationOperation::AddColumn,
                name.to_string(),
                None,
                // The IF NOT EXISTS guard (where supported) covers collisions with synthetic
                // columns the companion table may already carry (e.g. created_at/updated_by).
                add_column_ddl(dialect, &full, &format!("{} {}", quote(name), ty)),
                format!(
                    "Sync {} table: add column \"{}\" to \"{}\".\"{}\"",
                    suffix, name, schema, companion
                ),
                MigrationSafety::Safe,
                MigrationRisk::None,
                None,
            ),
            CompanionColumnOp::Rename { old, new } => (
                MigrationOperation::RenameColumn,
                new.to_string(),
                Some(old.to_string()),
                format!(
                    "ALTER TABLE {} RENAME COLUMN {} TO {}",
                    full,
                    quote(old),
                    quote(new)
                ),
                format!(
                    "Sync {} table: rename column \"{}\" → \"{}\" on \"{}\".\"{}\"",
                    suffix, old, new, schema, companion
                ),
                MigrationSafety::Safe,
                MigrationRisk::None,
                None,
            ),
            CompanionColumnOp::AlterType { name, ty } => (
                MigrationOperation::AlterColumnType,
                name.to_string(),
                None,
                format!(
                    "ALTER TABLE {} ALTER COLUMN {} TYPE {} USING {}::{}",
                    full,
                    quote(name),
                    ty,
                    quote(name),
                    ty
                ),
                format!(
                    "Sync {} table: change type of \"{}\".\"{}\".\"{}\" → {}",
                    suffix, schema, companion, name, ty
                ),
                MigrationSafety::BestEffort,
                MigrationRisk::MayFail,
                Some(format!(
                    "USING {}::{} cast may fail for incompatible values in the {} table.",
                    name, ty, suffix
                )),
            ),
        };
        steps.push(MigrationStep {
            step: 0,
            operation,
            schema: schema.to_string(),
            table: Some(companion.clone()),
            object,
            object_type: "column".into(),
            from_object,
            description,
            ddl: Some(ddl),
            safety,
            risk,
            risk_detail,
        });
    }
    steps
}

/// Build CREATE TABLE DDL for the `{table}_audit` companion table.
/// All source columns are replicated as nullable with no constraints, plus five audit metadata
/// columns prepended: audit_id (PK), audit_action, audit_at, audit_by, changed_fields.
fn audit_table_ddl(
    schema_name: &str,
    table_name: &str,
    source_cols: &[&ColumnConfig],
    dialect: &dyn Dialect,
) -> String {
    let audit_name = format!("{}_audit", table_name);
    let audit_full = format!("{}.{}", quote(schema_name), quote(&audit_name));

    let mut col_defs: Vec<String> = Vec::new();
    col_defs.push(format!(
        "{} {} NOT NULL DEFAULT {}",
        quote("audit_id"),
        "UUID",
        dialect.uuid_default_expr()
    ));
    col_defs.push(format!("{} TEXT NOT NULL", quote("audit_action")));
    col_defs.push(format!(
        "{} {} NOT NULL DEFAULT {}",
        quote("audit_at"),
        dialect.audit_timestamp_type(),
        dialect.now_fn()
    ));
    col_defs.push(format!("{} TEXT", quote("audit_by")));
    col_defs.push(format!(
        "{} {}",
        quote("changed_fields"),
        dialect.sys_json_type()
    ));

    let config_col_names: HashSet<&str> = source_cols.iter().map(|c| c.name.as_str()).collect();
    for c in source_cols {
        let typ = dialect.ddl_type(&parse_canonical(&c.type_));
        col_defs.push(format!("{} {}", quote(&c.name), typ));
    }
    let audit_ts = dialect.audit_timestamp_type();
    for (name, typ) in [
        ("created_at", audit_ts),
        ("updated_at", audit_ts),
        ("archived_at", audit_ts),
        ("created_by", "TEXT"),
        ("updated_by", "TEXT"),
    ] {
        if !config_col_names.contains(name) {
            col_defs.push(format!("{} {}", quote(name), typ));
        }
    }
    col_defs.push(format!("PRIMARY KEY ({})", quote("audit_id")));

    format!(
        "CREATE TABLE IF NOT EXISTS {} (\n  {}\n)",
        audit_full,
        col_defs.join(",\n  ")
    )
}

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

    fn schema(id: &str, name: &str) -> SchemaConfig {
        SchemaConfig {
            id: id.into(),
            name: name.into(),
            comment: None,
        }
    }

    fn table(id: &str, name: &str, schema_id: &str) -> TableConfig {
        TableConfig {
            id: id.into(),
            schema_id: Some(schema_id.into()),
            name: name.into(),
            comment: None,
            primary_key: PrimaryKeyConfig::Single("id".into()),
            unique: vec![],
            check: vec![],
            audit_log: false,
            versioning: None,
            global: false,
        }
    }

    fn col(id: &str, table_id: &str, name: &str, ty: &str, default: Option<&str>) -> ColumnConfig {
        ColumnConfig {
            id: id.into(),
            table_id: table_id.into(),
            name: name.into(),
            type_: ColumnTypeConfig::Simple(ty.into()),
            nullable: true,
            default: default.map(|d| ColumnDefaultConfig::Literal(d.into())),
            comment: None,
            asset: None,
            extensible: false,
        }
    }

    fn enum_cfg(id: &str, name: &str, schema_id: &str, values: &[&str]) -> EnumConfig {
        EnumConfig {
            id: id.into(),
            schema_id: Some(schema_id.into()),
            name: name.into(),
            values: values.iter().map(|s| s.to_string()).collect(),
            comment: None,
        }
    }

    fn ddls(steps: &[MigrationStep]) -> Vec<String> {
        steps.iter().filter_map(|s| s.ddl.clone()).collect()
    }

    #[test]
    fn finds_scalar_and_array_dependent_columns() {
        let mut cfg = FullConfig::default();
        cfg.schemas = vec![schema("s1", "app")];
        cfg.tables = vec![table("t_orders", "orders", "s1")];
        cfg.columns = vec![
            col(
                "c1",
                "t_orders",
                "status",
                "order_status",
                Some("'pending'"),
            ),
            col("c2", "t_orders", "tags", "order_status[]", None),
            col("c3", "t_orders", "name", "text", None), // not an enum
        ];
        let e = enum_cfg("e1", "order_status", "s1", &["pending", "shipped"]);
        let new_tables: HashMap<&str, &TableConfig> =
            cfg.tables.iter().map(|t| (t.id.as_str(), t)).collect();
        let new_schemas: HashMap<&str, &SchemaConfig> =
            cfg.schemas.iter().map(|s| (s.id.as_str(), s)).collect();

        let deps = enum_dependent_columns(&e, &cfg, &new_tables, &new_schemas, None);
        assert_eq!(deps.len(), 2);
        let scalar = deps.iter().find(|d| d.column == "status").unwrap();
        assert_eq!(scalar.schema, "app");
        assert_eq!(scalar.table, "orders");
        assert!(!scalar.is_array);
        assert_eq!(scalar.default.as_deref(), Some("'pending'"));
        let arr = deps.iter().find(|d| d.column == "tags").unwrap();
        assert!(arr.is_array);
        assert!(arr.default.is_none());
    }

    #[test]
    fn recreate_sequence_emits_rename_create_recast_drop() {
        let e = enum_cfg("e1", "order_status", "s1", &["pending", "shipped"]);
        let deps = vec![
            EnumColumnRef {
                schema: "app".into(),
                table: "orders".into(),
                column: "status".into(),
                default: Some("'pending'".into()),
                is_array: false,
            },
            EnumColumnRef {
                schema: "app".into(),
                table: "orders".into(),
                column: "tags".into(),
                default: None,
                is_array: true,
            },
        ];
        let mut steps = Vec::new();
        recreate_enum_steps(&mut steps, "app", &e, &["cancelled"], &deps);
        let sql = ddls(&steps);

        // Leading informational step carries no DDL.
        assert!(matches!(steps[0].safety, MigrationSafety::WarnOnly));
        assert!(steps[0].ddl.is_none());

        // Rename → create → (drop default, recast, set default) → recast array → drop old.
        assert_eq!(
            sql[0],
            r#"ALTER TYPE "app"."order_status" RENAME TO "order_status__arch_old""#
        );
        assert_eq!(
            sql[1],
            r#"CREATE TYPE "app"."order_status" AS ENUM ('pending', 'shipped')"#
        );
        assert_eq!(
            sql[2],
            r#"ALTER TABLE "app"."orders" ALTER COLUMN "status" DROP DEFAULT"#
        );
        assert_eq!(
            sql[3],
            r#"ALTER TABLE "app"."orders" ALTER COLUMN "status" TYPE "app"."order_status" USING "status"::text::"app"."order_status""#
        );
        assert_eq!(
            sql[4],
            r#"ALTER TABLE "app"."orders" ALTER COLUMN "status" SET DEFAULT 'pending'"#
        );
        // Array column: no default, single recast with array casts.
        assert_eq!(
            sql[5],
            r#"ALTER TABLE "app"."orders" ALTER COLUMN "tags" TYPE "app"."order_status"[] USING "tags"::text[]::"app"."order_status"[]"#
        );
        assert_eq!(
            *sql.last().unwrap(),
            r#"DROP TYPE IF EXISTS "app"."order_status__arch_old""#
        );
    }
}

#[cfg(all(test, feature = "sqlite"))]
mod companion_sync_tests {
    use super::*;
    use crate::db::sqlite::SqliteDialect;

    fn schema(id: &str, name: &str) -> SchemaConfig {
        SchemaConfig {
            id: id.into(),
            name: name.into(),
            comment: None,
        }
    }

    fn table(id: &str, name: &str, audit: bool, versioning: bool) -> TableConfig {
        TableConfig {
            id: id.into(),
            schema_id: Some("s1".into()),
            name: name.into(),
            comment: None,
            primary_key: PrimaryKeyConfig::Single("id".into()),
            unique: vec![],
            check: vec![],
            audit_log: audit,
            versioning: versioning.then(|| VersioningConfig {
                enabled: true,
                keep_versions: None,
            }),
            global: false,
        }
    }

    fn col(id: &str, name: &str, ty: &str) -> ColumnConfig {
        ColumnConfig {
            id: id.into(),
            table_id: "t1".into(),
            name: name.into(),
            type_: ColumnTypeConfig::Simple(ty.into()),
            nullable: true,
            default: None,
            comment: None,
            asset: None,
            extensible: false,
        }
    }

    fn base(audit: bool, versioning: bool) -> FullConfig {
        let mut cfg = FullConfig::default();
        cfg.schemas = vec![schema("s1", "app")];
        cfg.tables = vec![table("t1", "orders", audit, versioning)];
        cfg.columns = vec![col("c0", "id", "uuid"), col("c1", "status", "text")];
        cfg
    }

    fn plan(old: &FullConfig, new: &FullConfig) -> Vec<String> {
        let dialect = SqliteDialect;
        compute_migration_plan(old, new, None, None, &dialect, &HashMap::new())
            .unwrap()
            .steps
            .into_iter()
            .filter_map(|s| s.ddl)
            .collect()
    }

    #[test]
    fn add_column_syncs_audit_and_history() {
        let old = base(true, true);
        let mut new = base(true, true);
        new.columns.push(col("c2", "note", "text"));

        // SQLite has no ADD COLUMN IF NOT EXISTS, so the guard must not appear here.
        let sql = plan(&old, &new);
        assert!(sql
            .iter()
            .any(|s| s == r#"ALTER TABLE "app"."orders" ADD COLUMN "note" TEXT"#));
        assert!(sql
            .iter()
            .any(|s| s == r#"ALTER TABLE "app"."orders_audit" ADD COLUMN "note" TEXT"#));
        assert!(sql
            .iter()
            .any(|s| s == r#"ALTER TABLE "app"."orders_history" ADD COLUMN "note" TEXT"#));
    }

    #[test]
    fn rename_column_syncs_companions() {
        let old = base(true, true);
        let mut new = base(true, true);
        new.columns[1].name = "state".into();

        let sql = plan(&old, &new);
        assert!(sql
            .iter()
            .any(|s| s == r#"ALTER TABLE "app"."orders_audit" RENAME COLUMN "status" TO "state""#));
        assert!(sql.iter().any(
            |s| s == r#"ALTER TABLE "app"."orders_history" RENAME COLUMN "status" TO "state""#
        ));
    }

    #[test]
    fn alter_type_syncs_companions() {
        let old = base(true, false);
        let mut new = base(true, false);
        new.columns[1].type_ = ColumnTypeConfig::Simple("integer".into());

        let sql = plan(&old, &new);
        assert!(sql.iter().any(|s| s
            == r#"ALTER TABLE "app"."orders_audit" ALTER COLUMN "status" TYPE INTEGER USING "status"::INTEGER"#));
        // versioning disabled → no history sync
        assert!(!sql.iter().any(|s| s.contains("orders_history")));
    }

    #[test]
    fn no_companion_steps_when_features_disabled() {
        let old = base(false, false);
        let mut new = base(false, false);
        new.columns.push(col("c2", "note", "text"));

        let sql = plan(&old, &new);
        assert!(sql
            .iter()
            .any(|s| s == r#"ALTER TABLE "app"."orders" ADD COLUMN "note" TEXT"#));
        assert!(!sql.iter().any(|s| s.contains("orders_audit")));
        assert!(!sql.iter().any(|s| s.contains("orders_history")));
    }

    #[test]
    fn nullability_change_does_not_touch_companions() {
        let old = base(true, true);
        let mut new = base(true, true);
        new.columns[1].nullable = false;

        let sql = plan(&old, &new);
        // Main table gets SET NOT NULL, companions are untouched (snapshots stay nullable).
        assert!(sql.iter().any(|s| s.contains("SET NOT NULL")));
        assert!(!sql.iter().any(|s| s.contains("orders_audit")));
        assert!(!sql.iter().any(|s| s.contains("orders_history")));
    }
}

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

    fn snapshot() -> DbSnapshot {
        let mut snap = DbSnapshot::default();
        snap.introspected = true;
        snap.indexes_known = true;
        snap.constraints_known = true;
        snap
    }

    fn facts(nullable: bool, has_default: bool) -> ColumnFacts {
        ColumnFacts {
            data_type: "text".into(),
            nullable,
            has_default,
        }
    }

    fn step(op: MigrationOperation, table: Option<&str>, object: &str) -> MigrationStep {
        MigrationStep {
            step: 1,
            operation: op,
            schema: "app".into(),
            table: table.map(String::from),
            object: object.into(),
            object_type: "column".into(),
            from_object: None,
            description: "test step".into(),
            ddl: Some("SELECT 1".into()),
            safety: MigrationSafety::Safe,
            risk: MigrationRisk::None,
            risk_detail: None,
        }
    }

    fn skipped(d: StepDecision) -> bool {
        matches!(d, StepDecision::Skip(_))
    }

    #[test]
    fn add_column_is_skipped_when_the_column_already_exists() {
        // The regression this whole path exists for: an upgrade plan that adds a column the
        // physical table already has must not fail the migration.
        let mut snap = snapshot();
        snap.add_column("app", "transport_units", "project_id", facts(true, false));

        let s = step(
            MigrationOperation::AddColumn,
            Some("transport_units"),
            "project_id",
        );
        assert!(skipped(reconcile_step(&s, &snap)));
    }

    #[test]
    fn add_column_runs_when_the_column_is_absent() {
        let mut snap = snapshot();
        snap.add_column("app", "transport_units", "id", facts(false, false));

        let s = step(
            MigrationOperation::AddColumn,
            Some("transport_units"),
            "project_id",
        );
        assert_eq!(reconcile_step(&s, &snap), StepDecision::Execute);
    }

    #[test]
    fn nothing_is_skipped_when_introspection_produced_nothing() {
        // A failed or empty introspection must never be read as "the object does not exist".
        let snap = DbSnapshot::default();
        for op in [
            MigrationOperation::AddColumn,
            MigrationOperation::SetNotNull,
            MigrationOperation::CreateIndex,
        ] {
            let s = step(op, Some("orders"), "note");
            assert_eq!(reconcile_step(&s, &snap), StepDecision::Execute);
        }
    }

    #[test]
    fn create_table_is_skipped_when_the_table_exists() {
        let mut snap = snapshot();
        snap.add_column("app", "orders", "id", facts(false, false));

        let s = step(MigrationOperation::CreateTable, Some("orders"), "orders");
        assert!(skipped(reconcile_step(&s, &snap)));
    }

    #[test]
    fn rename_is_skipped_once_it_has_been_applied() {
        let mut snap = snapshot();
        snap.add_column("app", "orders", "state", facts(true, false));

        let mut s = step(MigrationOperation::RenameColumn, Some("orders"), "state");
        s.from_object = Some("status".into());
        assert!(skipped(reconcile_step(&s, &snap)));
    }

    #[test]
    fn rename_runs_while_the_old_column_is_still_there() {
        let mut snap = snapshot();
        snap.add_column("app", "orders", "status", facts(true, false));

        let mut s = step(MigrationOperation::RenameColumn, Some("orders"), "state");
        s.from_object = Some("status".into());
        assert_eq!(reconcile_step(&s, &snap), StepDecision::Execute);
    }

    #[test]
    fn rename_is_skipped_when_neither_name_exists() {
        let mut snap = snapshot();
        snap.add_column("app", "orders", "id", facts(false, false));

        let mut s = step(MigrationOperation::RenameColumn, Some("orders"), "state");
        s.from_object = Some("status".into());
        assert!(skipped(reconcile_step(&s, &snap)));
    }

    #[test]
    fn nullability_and_default_steps_respect_the_current_column_state() {
        let mut snap = snapshot();
        snap.add_column("app", "orders", "not_null_col", facts(false, false));
        snap.add_column("app", "orders", "nullable_col", facts(true, true));

        assert!(skipped(reconcile_step(
            &step(
                MigrationOperation::SetNotNull,
                Some("orders"),
                "not_null_col"
            ),
            &snap
        )));
        assert_eq!(
            reconcile_step(
                &step(
                    MigrationOperation::SetNotNull,
                    Some("orders"),
                    "nullable_col"
                ),
                &snap
            ),
            StepDecision::Execute
        );
        assert!(skipped(reconcile_step(
            &step(
                MigrationOperation::DropNotNull,
                Some("orders"),
                "nullable_col"
            ),
            &snap
        )));
        assert!(skipped(reconcile_step(
            &step(
                MigrationOperation::DropDefault,
                Some("orders"),
                "not_null_col"
            ),
            &snap
        )));
        assert_eq!(
            reconcile_step(
                &step(
                    MigrationOperation::DropDefault,
                    Some("orders"),
                    "nullable_col"
                ),
                &snap
            ),
            StepDecision::Execute
        );
    }

    #[test]
    fn column_steps_are_skipped_when_the_column_is_gone_from_a_known_table() {
        let mut snap = snapshot();
        snap.add_column("app", "orders", "id", facts(false, false));

        for op in [
            MigrationOperation::AlterColumnType,
            MigrationOperation::SetDefault,
            MigrationOperation::SetNotNull,
            MigrationOperation::BackfillNulls,
        ] {
            assert!(
                skipped(reconcile_step(&step(op, Some("orders"), "dropped"), &snap)),
                "expected a skip for a column that no longer exists"
            );
        }
    }

    #[test]
    fn index_and_foreign_key_steps_are_skipped_when_the_object_exists() {
        let mut snap = snapshot();
        snap.add_column("app", "orders", "user_id", facts(true, false));
        snap.add_index("app", "orders_user_idx");
        snap.add_constraint("app", "orders", "fk_orders_user");

        let mut idx = step(
            MigrationOperation::CreateIndex,
            Some("orders"),
            "orders_user_idx",
        );
        idx.object_type = "index".into();
        assert!(skipped(reconcile_step(&idx, &snap)));

        let mut fk = step(
            MigrationOperation::AddForeignKey,
            Some("orders"),
            "fk_orders_user",
        );
        fk.object_type = "foreign_key".into();
        assert!(skipped(reconcile_step(&fk, &snap)));
    }

    #[test]
    fn index_steps_run_when_the_dialect_cannot_report_indexes() {
        let mut snap = snapshot();
        snap.indexes_known = false;
        snap.add_column("app", "orders", "user_id", facts(true, false));

        let s = step(
            MigrationOperation::CreateIndex,
            Some("orders"),
            "orders_user_idx",
        );
        assert_eq!(reconcile_step(&s, &snap), StepDecision::Execute);
    }

    #[test]
    fn executed_steps_are_folded_into_the_snapshot() {
        // Later steps must see what earlier ones did, or a plan that adds a column and then
        // constrains it would skip the second half.
        let mut snap = snapshot();
        snap.add_table("app", "orders");

        let mut add = step(MigrationOperation::AddColumn, Some("orders"), "note");
        add.ddl = Some(r#"ALTER TABLE "app"."orders" ADD COLUMN "note" TEXT NOT NULL"#.into());
        apply_step_to_snapshot(&add, &mut snap);
        assert!(snap.has_column("app", "orders", "note"));

        // Re-running the same step is now a no-op.
        assert!(skipped(reconcile_step(&add, &snap)));

        // …and the recorded nullability matches the DDL that was executed.
        assert!(skipped(reconcile_step(
            &step(MigrationOperation::SetNotNull, Some("orders"), "note"),
            &snap
        )));
    }

    #[test]
    fn drops_and_enum_steps_are_never_skipped() {
        let mut snap = snapshot();
        snap.add_column("app", "orders", "note", facts(true, false));

        for op in [
            MigrationOperation::DropIndex,
            MigrationOperation::DropForeignKey,
            MigrationOperation::CreateEnum,
            MigrationOperation::AddEnumValue,
        ] {
            assert_eq!(
                reconcile_step(&step(op, Some("orders"), "note"), &snap),
                StepDecision::Execute
            );
        }
    }
}

#[cfg(all(test, feature = "postgres"))]
mod postgres_idempotent_ddl_tests {
    use super::*;
    use crate::db::postgres::PostgresDialect;

    #[test]
    fn add_column_carries_an_if_not_exists_guard() {
        let ddl = add_column_ddl(&PostgresDialect, r#""app"."orders""#, r#""note" TEXT"#);
        assert_eq!(
            ddl,
            r#"ALTER TABLE "app"."orders" ADD COLUMN IF NOT EXISTS "note" TEXT"#
        );
    }

    #[test]
    fn duplicate_object_sqlstates_are_recognised() {
        let d = PostgresDialect;
        // duplicate_column — what "column already exists" reports.
        assert!(d.is_duplicate_object_code("42701"));
        // duplicate_table / duplicate_object.
        assert!(d.is_duplicate_object_code("42P07"));
        assert!(d.is_duplicate_object_code("42710"));
        // undefined_column must still be a real failure.
        assert!(!d.is_duplicate_object_code("42703"));
    }
}