drizzle-cli 0.4.0

//! Live-database work for CLI commands: planning and applying migrations,
//! planning and applying `push`, and introspection.
//!
//! Each entry point dispatches on the resolved credentials to whichever
//! driver features were compiled in.

use std::path::Path;

use crate::commands::overrides::ResolvedConnection;
#[cfg(any(feature = "postgres-sync", feature = "tokio-postgres"))]
use crate::config::PostgresCreds;
use crate::config::{Credentials, Dialect, Driver, Extension, IntrospectCasing};
use crate::error::CliError;
use crate::output;
use drizzle_migrations::Migrations;
use drizzle_migrations::schema::Snapshot;

#[cfg(feature = "d1-http")]
mod d1_http;
mod filters;
mod mysql;

pub use filters::apply_snapshot_filters;
#[cfg(test)]
use filters::{compile_patterns, matches_patterns};

/// What [`run_migrations`] did.
#[derive(Debug, Default)]
pub struct MigrationResult {
    /// Number of migrations applied
    pub applied_count: usize,
    /// Tags of applied migrations
    pub applied_migrations: Vec<String>,
    /// Tags of interrupted migrations reconciled by `--repair`.
    pub repaired_migrations: Vec<String>,
}

/// Planned migration execution details.
#[derive(Debug, Clone)]
pub struct MigrationPlan {
    /// Number of already-applied migrations found in the database metadata table.
    pub applied_count: usize,
    /// Number of pending migrations found locally.
    pub pending_count: usize,
    /// Pending migration tags in execution order.
    pub pending_migrations: Vec<String>,
    /// Total number of non-empty SQL statements in pending migrations.
    pub pending_statements: usize,
    /// Integrity findings: hash drift between applied rows and local files,
    /// applied rows with no local migration, duplicate tracking rows, and
    /// interrupted (dirty) migrations. `--plan` reports these as warnings;
    /// `--verify` and `--safe` treat any finding as failure.
    pub findings: Vec<String>,
}

/// A tracking table as `migrate --plan`, `--dry-run` and `--verify` read it:
/// they must not create or upgrade it.
#[cfg(any(
    feature = "rusqlite",
    feature = "libsql",
    feature = "turso",
    feature = "postgres-sync",
    feature = "tokio-postgres",
    feature = "mysql-sync",
    feature = "mysql-async",
))]
pub(crate) enum TrackingState {
    /// No tracking table: no migration has run.
    Missing,
    /// A table from before migrations were recorded by name, as its rows.
    Legacy(Vec<drizzle_migrations::AppliedMigrationMetadata>),
    /// The current layout.
    Current,
}

/// The applied migrations of a legacy tracking table, matched to the local
/// ones by timestamp and hash the way `migrate` upgrades the table.
#[cfg(any(
    feature = "rusqlite",
    feature = "libsql",
    feature = "turso",
    feature = "postgres-sync",
    feature = "tokio-postgres",
    feature = "mysql-sync",
    feature = "mysql-async",
))]
pub(crate) fn legacy_applied_records(
    set: &Migrations,
    rows: &[drizzle_migrations::AppliedMigrationMetadata],
) -> Result<Vec<AppliedMigrationRecord>, CliError> {
    let matched = drizzle_migrations::match_applied_migration_metadata(set.all(), rows)
        .map_err(|e| CliError::MigrationError(e.to_string()))?;
    Ok(matched
        .into_iter()
        .map(|row| AppliedMigrationRecord {
            hash: row.hash,
            name: row.name,
            dirty: false,
        })
        .collect())
}

/// Whether a SQLite database named by `path` exists. Opening a missing file
/// would create it; `:memory:` and `file:` URIs are left to the driver.
#[cfg(any(feature = "rusqlite", feature = "libsql"))]
fn sqlite_database_exists(path: &str) -> bool {
    path == ":memory:" || path.starts_with("file:") || std::path::Path::new(path).exists()
}

#[cfg(any(
    feature = "rusqlite",
    feature = "libsql",
    feature = "turso",
    feature = "postgres-sync",
    feature = "tokio-postgres",
    feature = "mysql-sync",
    feature = "mysql-async",
    feature = "d1-http",
))]
#[derive(Debug, Clone)]
pub(crate) struct AppliedMigrationRecord {
    pub(crate) hash: String,
    pub(crate) name: String,
    /// Row was started but never marked finished (`applied_at IS NULL`).
    pub(crate) dirty: bool,
}

/// SQL that `drizzle push` would run, from [`plan_push`].
#[derive(Debug, Clone)]
pub struct PushPlan {
    /// Statements to execute, in order.
    pub sql_statements: Vec<String>,
    /// Warnings from diffing (for example possible data loss).
    pub warnings: Vec<String>,
    /// `true` when the plan may lose data, so
    /// [`apply_push`] asks for confirmation unless forced.
    pub destructive: bool,
}

/// Optional filters for introspection and push planning.
#[derive(Debug, Clone, Default)]
pub struct SnapshotFilters {
    /// Table name patterns to keep (`tablesFilter`).
    pub tables: Option<Vec<String>>,
    /// PostgreSQL schema patterns to keep (`schemaFilter`).
    pub schemas: Option<Vec<String>>,
    /// PostgreSQL extensions whose objects are left out.
    pub extensions: Option<Vec<Extension>>,
    /// Which `PostgreSQL` roles (and their privileges) to keep, from
    /// `entities.roles`. `None` leaves roles as they are.
    pub roles: Option<crate::config::RolesFilter>,
}

fn connection_driver_mismatch(connection: &ResolvedConnection) -> CliError {
    CliError::Other(format!(
        "driver '{}' cannot use the resolved {} credentials",
        connection.driver, connection.dialect
    ))
}

impl SnapshotFilters {
    const fn is_empty(&self) -> bool {
        self.tables.is_none() && self.schemas.is_none() && self.extensions.is_none()
    }
}

/// Plans a push: introspects the live database and diffs it against
/// `desired`.
///
/// The tracking table and anything `filters` excludes are removed from the
/// live side first; the caller filters `desired`. `breakpoints` is unused.
/// The plan is destructive when a statement drops or truncates something,
/// or (MySQL) when the diff emits any warning.
///
/// # Errors
///
/// Returns [`CliError`] if introspecting the live database fails, if applying
/// the given snapshot filters fails, or if generating the diff SQL fails.
pub fn plan_push(
    connection: &ResolvedConnection,
    desired: &Snapshot,
    breakpoints: bool,
    filters: &SnapshotFilters,
    migrations_table: &str,
) -> Result<PushPlan, CliError> {
    plan_push_with_renames(
        connection,
        desired,
        breakpoints,
        filters,
        migrations_table,
        &mut |_, _, options| Ok(options),
    )
}

/// [`plan_push`] that lets `resolve_renames` settle rename-or-create
/// questions before diffing.
///
/// `resolve_renames` gets the live snapshot (after filtering), the desired
/// snapshot, and the diff options, and returns the options to diff with,
/// typically from
/// [`renames::resolve_for_command`](crate::commands::renames::resolve_for_command).
/// [`plan_push`] passes the options through, so the diff infers renames.
///
/// # Errors
///
/// Returns the errors of [`plan_push`] and of `resolve_renames`.
pub fn plan_push_with_renames(
    connection: &ResolvedConnection,
    desired: &Snapshot,
    breakpoints: bool,
    filters: &SnapshotFilters,
    migrations_table: &str,
    resolve_renames: &mut dyn FnMut(
        &Snapshot,
        &Snapshot,
        drizzle_migrations::DiffOptions,
    ) -> Result<drizzle_migrations::DiffOptions, CliError>,
) -> Result<PushPlan, CliError> {
    let introspected = introspect_database(connection)?;
    let mut current = introspected.snapshot;
    exclude_tracking_table(&mut current, connection.dialect, migrations_table)?;
    apply_snapshot_filters(&mut current, connection.dialect, filters)?;
    // PostgreSQL rewrites stored expressions into its own form; compare them
    // the way drizzle-kit's push does, or every push would recreate them.
    if let (Snapshot::Postgres(live), Snapshot::Postgres(wanted)) = (&mut current, desired) {
        live.align_push_insensitive_fields(wanted);
    }
    let mut options = drizzle_migrations::DiffOptions::new();
    if let Some(defaults) = introspected.mysql_catalog_defaults {
        options = options.mysql_catalog_defaults(defaults);
    }
    // Ask about renames on exactly the snapshots the diff compares.
    let options = resolve_renames(&current, desired, options)?;
    let (sql_statements, warnings) = generate_push_sql(&current, desired, breakpoints, &options)?;
    let destructive = push_requires_confirmation(connection.dialect, &sql_statements, &warnings);

    Ok(PushPlan {
        sql_statements,
        warnings,
        destructive,
    })
}

fn exclude_tracking_table(
    snapshot: &mut Snapshot,
    dialect: Dialect,
    migrations_table: &str,
) -> Result<(), CliError> {
    if migrations_table.trim().is_empty() {
        return Ok(());
    }
    apply_snapshot_filters(
        snapshot,
        dialect,
        &SnapshotFilters {
            tables: Some(vec![format!(
                "!{}",
                glob::Pattern::escape(migrations_table)
            )]),
            schemas: None,
            extensions: None,
            roles: None,
        },
    )
}

/// Reads the live schema for `drizzle seed`: introspects the database,
/// drops the migrations tracking table, and applies `filters`.
///
/// # Errors
///
/// Returns [`CliError`] if introspecting the database or applying the
/// filters fails.
pub fn introspect_for_seed(
    connection: &ResolvedConnection,
    filters: &SnapshotFilters,
    migrations_table: &str,
) -> Result<Snapshot, CliError> {
    let mut snapshot = introspect_database(connection)?.snapshot;
    exclude_tracking_table(&mut snapshot, connection.dialect, migrations_table)?;
    apply_snapshot_filters(&mut snapshot, connection.dialect, filters)?;
    Ok(snapshot)
}

/// Runs the statements of a seed in order, in one transaction where the
/// driver supports it.
///
/// # Errors
///
/// Returns [`CliError`] if no compiled driver matches the credentials,
/// connecting fails, or a statement fails.
pub fn execute_seed(
    connection: &ResolvedConnection,
    statements: &[String],
) -> Result<(), CliError> {
    execute_statements(connection, statements)
}

/// Runs a [`PushPlan`], asking for confirmation first when it is
/// destructive and `force` is `false`.
///
/// # Errors
///
/// Returns [`CliError::Aborted`] if the user declines the confirmation prompt
/// for a destructive plan, and [`CliError`] if the prompt itself fails or if
/// executing the planned SQL statements against the database fails.
pub fn apply_push(
    connection: &ResolvedConnection,
    plan: &PushPlan,
    force: bool,
) -> Result<(), CliError> {
    apply_push_with_confirmation(connection, plan, force, confirm_destructive)
}

/// [`apply_push`] with the destructive-change prompt supplied by the caller.
fn apply_push_with_confirmation(
    connection: &ResolvedConnection,
    plan: &PushPlan,
    force: bool,
    confirm: impl FnOnce() -> Result<bool, CliError>,
) -> Result<(), CliError> {
    if plan.sql_statements.is_empty() {
        return Ok(());
    }

    if plan.destructive && !force && !confirm()? {
        return Err(CliError::Aborted(
            "Push aborted: the destructive changes were not confirmed, so nothing was applied"
                .into(),
        ));
    }

    execute_statements(connection, &plan.sql_statements)
}

/// Compares local migrations with the database's tracking table without
/// changing anything (`migrate --plan` / `--verify`).
///
/// # Errors
///
/// Returns [`CliError`] if no compiled driver matches the credentials, if
/// connecting to the database fails, or if reading the migration tracking
/// table or on-disk migration files fails.
#[allow(unused_variables)] // params consumed inside feature-gated block
pub fn plan_migrations(
    connection: &ResolvedConnection,
    migrations_dir: &Path,
    migrations_table: &str,
    migrations_schema: &str,
) -> Result<MigrationPlan, CliError> {
    let set = load_migration_set(
        connection.dialect,
        migrations_dir,
        migrations_table,
        migrations_schema,
    )?;

    match &connection.credentials {
        #[cfg(feature = "rusqlite")]
        Credentials::Sqlite { path } => inspect_sqlite_migrations(&set, path),

        #[cfg(not(feature = "rusqlite"))]
        Credentials::Sqlite { .. } => Err(CliError::MissingDriver {
            dialect: "SQLite",
            feature: "rusqlite",
        }),

        #[cfg(any(feature = "libsql", feature = "turso"))]
        Credentials::Turso { url, auth_token } => {
            if connection.driver == Driver::Libsql {
                #[cfg(feature = "libsql")]
                {
                    inspect_libsql_local_migrations(&set, url)
                }
                #[cfg(not(feature = "libsql"))]
                {
                    Err(CliError::MissingDriver {
                        dialect: "LibSQL (local)",
                        feature: "libsql",
                    })
                }
            } else if connection.driver == Driver::Turso {
                #[cfg(feature = "turso")]
                {
                    inspect_turso_migrations(&set, url, auth_token.as_deref())
                }
                #[cfg(not(feature = "turso"))]
                {
                    let _ = auth_token;
                    Err(CliError::MissingDriver {
                        dialect: "Turso (remote)",
                        feature: "turso",
                    })
                }
            } else {
                Err(connection_driver_mismatch(connection))
            }
        }

        #[cfg(all(not(feature = "turso"), not(feature = "libsql")))]
        Credentials::Turso { .. } => Err(CliError::MissingDriver {
            dialect: "Turso",
            feature: "turso or libsql",
        }),

        Credentials::Postgres(creds) => {
            let _ = creds;
            match connection.driver {
                Driver::PostgresSync => {
                    #[cfg(feature = "postgres-sync")]
                    {
                        inspect_postgres_sync_migrations(&set, creds)
                    }
                    #[cfg(not(feature = "postgres-sync"))]
                    {
                        Err(CliError::MissingDriver {
                            dialect: "PostgreSQL",
                            feature: "postgres-sync",
                        })
                    }
                }
                Driver::TokioPostgres => {
                    #[cfg(feature = "tokio-postgres")]
                    {
                        inspect_postgres_async_migrations(&set, creds)
                    }
                    #[cfg(not(feature = "tokio-postgres"))]
                    {
                        Err(CliError::MissingDriver {
                            dialect: "PostgreSQL",
                            feature: "tokio-postgres",
                        })
                    }
                }
                _ => Err(connection_driver_mismatch(connection)),
            }
        }

        Credentials::MySQL(creds) => mysql::plan_migrations(connection.driver, creds, &set),

        #[cfg(feature = "d1-http")]
        Credentials::D1 {
            account_id,
            database_id,
            token,
        } => d1_http::inspect_migrations(&set, account_id, database_id, token),

        #[cfg(not(feature = "d1-http"))]
        Credentials::D1 { .. } => Err(CliError::MissingDriver {
            dialect: "Cloudflare D1 (HTTP)",
            feature: "d1-http",
        }),

        Credentials::AwsDataApi { .. } => Err(CliError::UnsupportedForDriver {
            operation: "Migration planning against AWS Data API",
            driver: "aws-data-api",
            hint: "AWS RDS Data API schema ops are not yet wired into this CLI. For now, \
                   run the generated SQL with `aws rds-data execute-statement --sql=...` \
                   (or use tokio-postgres with a temporary direct connection).",
        }),
    }
}

/// Applies the pending migrations in `migrations_dir` and records them in
/// the tracking table, creating or upgrading that table as needed.
///
/// With `repair`, an interrupted migration is first reconciled against the
/// live schema (see [`drizzle_migrations::repair`]); without it, an
/// interrupted migration stops the run.
///
/// # Errors
///
/// Returns [`CliError`] if no compiled driver matches, connecting fails, the
/// migrations cannot be loaded, a migration is interrupted (and `repair` is
/// off or cannot reconcile it), a statement fails, or the tracking table
/// cannot be written.
#[allow(unused_variables)] // params consumed inside feature-gated block
pub fn run_migrations(
    connection: &ResolvedConnection,
    migrations_dir: &Path,
    migrations_table: &str,
    migrations_schema: &str,
    repair: bool,
) -> Result<MigrationResult, CliError> {
    let set = load_migration_set(
        connection.dialect,
        migrations_dir,
        migrations_table,
        migrations_schema,
    )?;

    match &connection.credentials {
        #[cfg(feature = "rusqlite")]
        Credentials::Sqlite { path } => run_sqlite_migrations(&set, path, repair),

        #[cfg(not(feature = "rusqlite"))]
        Credentials::Sqlite { .. } => Err(CliError::MissingDriver {
            dialect: "SQLite",
            feature: "rusqlite",
        }),

        #[cfg(any(feature = "libsql", feature = "turso"))]
        Credentials::Turso { url, auth_token } => {
            if connection.driver == Driver::Libsql {
                #[cfg(feature = "libsql")]
                {
                    run_libsql_local_migrations(&set, url, repair)
                }
                #[cfg(not(feature = "libsql"))]
                {
                    Err(CliError::MissingDriver {
                        dialect: "LibSQL (local)",
                        feature: "libsql",
                    })
                }
            } else if connection.driver == Driver::Turso {
                #[cfg(feature = "turso")]
                {
                    run_turso_migrations(&set, url, auth_token.as_deref(), repair)
                }
                #[cfg(not(feature = "turso"))]
                {
                    let _ = auth_token;
                    Err(CliError::MissingDriver {
                        dialect: "Turso (remote)",
                        feature: "turso",
                    })
                }
            } else {
                Err(connection_driver_mismatch(connection))
            }
        }

        #[cfg(all(not(feature = "turso"), not(feature = "libsql")))]
        Credentials::Turso { .. } => Err(CliError::MissingDriver {
            dialect: "Turso",
            feature: "turso or libsql",
        }),

        Credentials::Postgres(creds) => {
            let _ = creds;
            match connection.driver {
                Driver::PostgresSync => {
                    #[cfg(feature = "postgres-sync")]
                    {
                        run_postgres_sync_migrations(&set, creds, repair)
                    }
                    #[cfg(not(feature = "postgres-sync"))]
                    {
                        Err(CliError::MissingDriver {
                            dialect: "PostgreSQL",
                            feature: "postgres-sync",
                        })
                    }
                }
                Driver::TokioPostgres => {
                    #[cfg(feature = "tokio-postgres")]
                    {
                        run_postgres_async_migrations(&set, creds, repair)
                    }
                    #[cfg(not(feature = "tokio-postgres"))]
                    {
                        Err(CliError::MissingDriver {
                            dialect: "PostgreSQL",
                            feature: "tokio-postgres",
                        })
                    }
                }
                _ => Err(connection_driver_mismatch(connection)),
            }
        }

        Credentials::MySQL(creds) => mysql::run_migrations(connection.driver, creds, &set, repair),

        #[cfg(feature = "d1-http")]
        Credentials::D1 {
            account_id,
            database_id,
            token,
        } => d1_http::run_migrations(&set, account_id, database_id, token, repair),

        #[cfg(not(feature = "d1-http"))]
        Credentials::D1 { .. } => Err(CliError::MissingDriver {
            dialect: "Cloudflare D1 (HTTP)",
            feature: "d1-http",
        }),

        Credentials::AwsDataApi { .. } => Err(CliError::UnsupportedForDriver {
            operation: "Running migrations against AWS Data API",
            driver: "aws-data-api",
            hint: "AWS RDS Data API migrations are not yet wired into this CLI. Run the \
                   generated SQL via `aws rds-data execute-statement` or connect directly \
                   with tokio-postgres.",
        }),
    }
}

fn load_migration_set(
    dialect: Dialect,
    migrations_dir: &Path,
    migrations_table: &str,
    migrations_schema: &str,
) -> Result<Migrations, CliError> {
    let tracking = migration_tracking(dialect, migrations_table, migrations_schema);

    // Load migrations from filesystem
    let migrations = drizzle_migrations::MigrationDir::new(migrations_dir)
        .dialect(dialect.to_base())
        .discover()
        .map_err(|e| CliError::Other(format!("Failed to load migrations: {e}")))?;
    Ok(Migrations::with_tracking(
        migrations,
        dialect.to_base(),
        tracking,
    ))
}

fn migration_tracking(
    dialect: Dialect,
    migrations_table: &str,
    migrations_schema: &str,
) -> drizzle_types::MigrationTracking {
    let mut tracking = match dialect {
        Dialect::Postgresql => drizzle_types::MigrationTracking::POSTGRES,
        Dialect::Mysql => drizzle_types::MigrationTracking::MYSQL,
        Dialect::Sqlite | Dialect::Turso => drizzle_types::MigrationTracking::SQLITE,
    };

    if !migrations_table.trim().is_empty() {
        tracking = tracking.table(migrations_table.to_owned());
    }

    if dialect == Dialect::Postgresql && !migrations_schema.trim().is_empty() {
        tracking = tracking.schema(migrations_schema.to_owned());
    }

    tracking
}

#[cfg(any(
    feature = "rusqlite",
    feature = "libsql",
    feature = "turso",
    feature = "postgres-sync",
    feature = "tokio-postgres",
    feature = "mysql-sync",
    feature = "mysql-async",
    feature = "d1-http",
))]
pub(crate) fn build_migration_plan(
    set: &Migrations,
    applied: &[AppliedMigrationRecord],
) -> Result<MigrationPlan, CliError> {
    let findings = collect_integrity_findings(set, applied)?;

    // Dirty rows are not applied: the migration started but never finished.
    let applied_names = applied
        .iter()
        .filter(|m| !m.dirty)
        .map(|m| m.name.clone())
        .collect::<Vec<_>>();
    let pending = set.pending(&applied_names).collect::<Vec<_>>();

    let pending_statements = pending
        .iter()
        .map(|m| {
            m.statements()
                .iter()
                .filter(|stmt| !stmt.trim().is_empty())
                .count()
        })
        .sum();

    Ok(MigrationPlan {
        applied_count: applied_names.len(),
        pending_count: pending.len(),
        pending_migrations: pending.iter().map(|m| m.tag().to_string()).collect(),
        pending_statements,
        findings,
    })
}

/// Collect integrity findings between the tracking table and local files.
///
/// Findings (drift, missing-local rows, duplicate tracking rows, interrupted
/// migrations) are returned rather than raised: `--plan` shows them as
/// warnings while `--verify`/`--safe` fail on any. A duplicate name in the
/// LOCAL migration set is still a hard error — the local set is broken and
/// no comparison against it is meaningful.
#[cfg(any(
    feature = "rusqlite",
    feature = "libsql",
    feature = "turso",
    feature = "postgres-sync",
    feature = "tokio-postgres",
    feature = "mysql-sync",
    feature = "mysql-async",
    feature = "d1-http",
))]
fn collect_integrity_findings(
    set: &Migrations,
    applied: &[AppliedMigrationRecord],
) -> Result<Vec<String>, CliError> {
    use std::collections::{HashMap, HashSet};

    let mut local_by_name = HashMap::<&str, &str>::new();
    for migration in set.all() {
        if local_by_name
            .insert(migration.name(), migration.hash())
            .is_some()
        {
            return Err(CliError::MigrationError(format!(
                "Local migrations contain duplicate name: {}",
                migration.name()
            )));
        }
    }

    let mut findings = Vec::new();
    let mut seen_db_names = HashSet::<&str>::new();
    for applied_row in applied {
        if !seen_db_names.insert(applied_row.name.as_str()) {
            findings.push(format!(
                "duplicate tracking rows for `{}` in the migrations table",
                applied_row.name
            ));
            continue;
        }

        if applied_row.dirty {
            let recovery = if set.dialect() == drizzle_types::Dialect::MySQL {
                "inspect the partially applied DDL and resolve it manually"
            } else {
                "run `drizzle migrate --repair` or resolve it manually"
            };
            findings.push(format!(
                "`{}` was interrupted mid-apply (started but never finished) — {recovery}",
                applied_row.name,
            ));
        }

        let Some(local_hash) = local_by_name.get(applied_row.name.as_str()) else {
            findings.push(format!(
                "`{}` is recorded in the database but no local migration has that name",
                applied_row.name
            ));
            continue;
        };

        if *local_hash != applied_row.hash {
            findings.push(format!(
                "`{}` has drifted: the local migration file no longer matches what \
                 was applied (database hash {}, local hash {})",
                applied_row.name, applied_row.hash, local_hash
            ));
        }
    }

    Ok(findings)
}

#[cfg(any(feature = "rusqlite", feature = "libsql", feature = "turso"))]
fn escape_sql_literal(value: &str) -> String {
    value.replace('\'', "''")
}

// ============================================================================
// Interrupted ("dirty") migration handling
// ============================================================================
//
// A tracking row with a non-null `name` and a NULL `applied_at` means an
// earlier run recorded a migration as started and never recorded it as
// finished — the database may hold part of that migration's DDL. Every runner
// checks for those rows before applying anything; `--repair` additionally
// reconciles them statement-by-statement against live introspection.

#[cfg(any(
    feature = "rusqlite",
    feature = "libsql",
    feature = "turso",
    feature = "postgres-sync",
    feature = "tokio-postgres",
))]
fn migrator_cli_error(error: drizzle_migrations::MigratorError) -> CliError {
    CliError::MigrationError(error.to_string())
}

/// Turn a dirty-name list into either an error (no `--repair`) or the list of
/// migrations to reconcile.
#[cfg(any(
    feature = "rusqlite",
    feature = "libsql",
    feature = "turso",
    feature = "postgres-sync",
    feature = "tokio-postgres",
))]
fn dirty_migrations_to_repair<'a>(
    set: &'a Migrations,
    dirty: &[String],
    repair: bool,
) -> Result<Vec<&'a drizzle_migrations::Migration>, CliError> {
    if dirty.is_empty() {
        return Ok(Vec::new());
    }
    if !repair {
        return Err(migrator_cli_error(
            set.interrupted_migration_error(dirty)
                .expect("dirty list is non-empty"),
        ));
    }
    set.resolve_dirty_migrations(dirty)
        .map_err(migrator_cli_error)
}

#[cfg(any(
    feature = "rusqlite",
    feature = "libsql",
    feature = "turso",
    feature = "d1-http",
))]
fn sqlite_execution_cli_error(
    error: drizzle_migrations::SqliteMigrationExecutionError,
) -> CliError {
    CliError::MigrationError(error.to_string())
}

#[cfg(any(
    feature = "rusqlite",
    feature = "libsql",
    feature = "turso",
    feature = "d1-http",
))]
#[cfg(feature = "d1-http")]
pub(super) fn reject_sqlite_foreign_key_suspension<'a>(
    migrations: impl IntoIterator<Item = &'a drizzle_migrations::Migration>,
    adapter: &str,
) -> Result<(), CliError> {
    let executions = migrations
        .into_iter()
        .map(drizzle_migrations::Migration::sqlite_execution)
        .collect::<Result<Vec<_>, _>>()
        .map_err(sqlite_execution_cli_error)?;
    if executions
        .iter()
        .any(|execution| execution.suspends_foreign_keys())
    {
        return Err(CliError::MigrationError(format!(
            "{adapter} cannot execute SQLite table rebuilds that suspend foreign keys"
        )));
    }
    Ok(())
}

#[cfg(any(feature = "rusqlite", feature = "libsql", feature = "turso"))]
fn ensure_sqlite_dirty_repairs_are_safe(
    migrations: &[&drizzle_migrations::Migration],
) -> Result<(), CliError> {
    for migration in migrations {
        let execution = migration
            .sqlite_execution()
            .map_err(sqlite_execution_cli_error)?;
        if execution.suspends_foreign_keys() {
            return Err(CliError::MigrationError(format!(
                "Migration '{}' cannot be repaired safely: foreign-key-suspending table rebuilds require the original connection-owned transaction scope",
                migration.tag()
            )));
        }
    }
    Ok(())
}

#[cfg(any(feature = "rusqlite", feature = "libsql", feature = "turso"))]
fn finish_sqlite_foreign_key_scope<T>(
    result: Result<T, CliError>,
    restore: Result<(), CliError>,
) -> Result<T, CliError> {
    match (result, restore) {
        (Err(primary), Err(restore)) => Err(CliError::MigrationError(format!(
            "{primary}; additionally failed to restore SQLite foreign-key enforcement: {restore}"
        ))),
        (Err(primary), Ok(())) => Err(primary),
        (Ok(_), Err(restore)) => Err(restore),
        (Ok(value), Ok(())) => Ok(value),
    }
}

#[cfg(feature = "rusqlite")]
fn introspect_catalog_sqlite(
    conn: &rusqlite::Connection,
) -> Result<drizzle_migrations::repair::Catalog, CliError> {
    let mut stmt = conn
        .prepare(drizzle_migrations::repair::sqlite::OBJECTS_QUERY)
        .map_err(|e| CliError::MigrationError(e.to_string()))?;
    let rows = stmt
        .query_map([], |row| {
            Ok((
                row.get::<_, String>(0)?,
                row.get::<_, String>(1)?,
                row.get::<_, Option<String>>(2)?,
            ))
        })
        .map_err(|e| CliError::MigrationError(e.to_string()))?;

    let rows = rows
        .collect::<Result<Vec<_>, _>>()
        .map_err(|e| CliError::MigrationError(e.to_string()))?;
    Ok(drizzle_migrations::repair::sqlite::catalog(&rows))
}

#[cfg(feature = "rusqlite")]
fn repair_dirty_migrations_sqlite(
    conn: &rusqlite::Connection,
    set: &Migrations,
    repair: bool,
) -> Result<Vec<String>, CliError> {
    let dirty = {
        let mut stmt = match conn.prepare(&set.dirty_names_sql()) {
            Ok(stmt) => stmt,
            Err(error) if is_sqlite_missing_table(&error) => return Ok(Vec::new()),
            Err(error) => return Err(CliError::MigrationError(error.to_string())),
        };
        stmt.query_map([], |row| row.get::<_, String>(0))
            .map_err(|e| CliError::MigrationError(e.to_string()))?
            .collect::<Result<Vec<String>, _>>()
            .map_err(|e| CliError::MigrationError(e.to_string()))?
    };

    let migrations = dirty_migrations_to_repair(set, &dirty, repair)?;
    ensure_sqlite_dirty_repairs_are_safe(&migrations)?;
    let table_ident = set.table_ident_sql();
    let mut repaired = Vec::new();
    for migration in migrations {
        let catalog = introspect_catalog_sqlite(conn)?;
        let plan =
            drizzle_migrations::repair::plan(drizzle_types::Dialect::SQLite, migration, &catalog);
        for statement in plan
            .into_executable(&table_ident)
            .map_err(migrator_cli_error)?
        {
            conn.execute(&statement, [])
                .map_err(|e| CliError::MigrationError(format!("Repair statement failed: {e}")))?;
        }
        conn.execute(&set.record_migration_finished_sql(migration), [])
            .map_err(|e| CliError::MigrationError(e.to_string()))?;
        repaired.push(migration.tag().to_string());
    }
    Ok(repaired)
}

#[cfg(any(feature = "libsql", feature = "turso"))]
async fn introspect_catalog_libsql(
    conn: &libsql::Connection,
) -> Result<drizzle_migrations::repair::Catalog, CliError> {
    let mut rows = conn
        .query(drizzle_migrations::repair::sqlite::OBJECTS_QUERY, ())
        .await
        .map_err(|e| CliError::MigrationError(e.to_string()))?;

    let mut master_rows = Vec::new();
    while let Some(row) = rows
        .next()
        .await
        .map_err(|e| CliError::MigrationError(e.to_string()))?
    {
        master_rows.push((
            row.get::<String>(0)
                .map_err(|e| CliError::MigrationError(e.to_string()))?,
            row.get::<String>(1)
                .map_err(|e| CliError::MigrationError(e.to_string()))?,
            row.get::<Option<String>>(2).ok().flatten(),
        ));
    }
    Ok(drizzle_migrations::repair::sqlite::catalog(&master_rows))
}

#[cfg(any(feature = "libsql", feature = "turso"))]
async fn repair_dirty_migrations_libsql(
    conn: &libsql::Connection,
    set: &Migrations,
    repair: bool,
) -> Result<Vec<String>, CliError> {
    let mut rows = conn
        .query(&set.dirty_names_sql(), ())
        .await
        .map_err(|e| CliError::MigrationError(e.to_string()))?;
    let mut dirty = Vec::new();
    while let Some(row) = rows
        .next()
        .await
        .map_err(|e| CliError::MigrationError(e.to_string()))?
    {
        dirty.push(
            row.get::<String>(0)
                .map_err(|e| CliError::MigrationError(e.to_string()))?,
        );
    }

    let migrations = dirty_migrations_to_repair(set, &dirty, repair)?;
    ensure_sqlite_dirty_repairs_are_safe(&migrations)?;
    let table_ident = set.table_ident_sql();
    let mut repaired = Vec::new();
    for migration in migrations {
        let catalog = introspect_catalog_libsql(conn).await?;
        let plan =
            drizzle_migrations::repair::plan(drizzle_types::Dialect::SQLite, migration, &catalog);
        for statement in plan
            .into_executable(&table_ident)
            .map_err(migrator_cli_error)?
        {
            conn.execute(&statement, ())
                .await
                .map_err(|e| CliError::MigrationError(format!("Repair statement failed: {e}")))?;
        }
        conn.execute(&set.record_migration_finished_sql(migration), ())
            .await
            .map_err(|e| CliError::MigrationError(e.to_string()))?;
        repaired.push(migration.tag().to_string());
    }
    Ok(repaired)
}

#[cfg(feature = "postgres-sync")]
fn introspect_catalog_postgres_sync(
    client: &mut postgres::Client,
) -> Result<drizzle_migrations::repair::Catalog, CliError> {
    use drizzle_migrations::repair::postgres as pg;

    let err = |e: postgres::Error| CliError::MigrationError(e.to_string());
    let two =
        |client: &mut postgres::Client, sql: &str| -> Result<Vec<(String, String)>, CliError> {
            Ok(client
                .query(sql, &[])
                .map_err(err)?
                .iter()
                .map(|row| (row.get::<_, String>(0), row.get::<_, String>(1)))
                .collect())
        };
    let three = |client: &mut postgres::Client,
                 sql: &str|
     -> Result<Vec<(String, String, String)>, CliError> {
        Ok(client
            .query(sql, &[])
            .map_err(err)?
            .iter()
            .map(|row| {
                (
                    row.get::<_, String>(0),
                    row.get::<_, String>(1),
                    row.get::<_, String>(2),
                )
            })
            .collect())
    };

    let indexes = client
        .query(pg::INDEXES_QUERY, &[])
        .map_err(err)?
        .iter()
        .map(|row| {
            (
                row.get::<_, String>(0),
                row.get::<_, String>(1),
                row.get::<_, bool>(2),
            )
        })
        .collect::<Vec<_>>();

    let tables = two(client, pg::TABLES_QUERY)?;
    let columns = three(client, pg::COLUMNS_QUERY)?;
    let views = two(client, pg::VIEWS_QUERY)?;
    let index_columns = three(client, pg::INDEX_COLUMNS_QUERY)?;
    let enums = three(client, pg::ENUMS_QUERY)?;

    Ok(pg::catalog(
        &tables,
        &columns,
        &views,
        &indexes,
        &index_columns,
        &enums,
    ))
}

#[cfg(feature = "postgres-sync")]
fn repair_dirty_migrations_postgres_sync(
    client: &mut postgres::Client,
    set: &Migrations,
    repair: bool,
) -> Result<Vec<String>, CliError> {
    let dirty = client
        .query(&set.dirty_names_sql(), &[])
        .map_err(|e| CliError::MigrationError(e.to_string()))?
        .iter()
        .map(|row| row.try_get::<_, String>(0))
        .collect::<Result<Vec<_>, _>>()
        .map_err(|e| CliError::MigrationError(e.to_string()))?;

    let table_ident = set.table_ident_sql();
    let to_repair = dirty_migrations_to_repair(set, &dirty, repair)?
        .into_iter()
        .cloned()
        .collect::<Vec<_>>();

    let mut repaired = Vec::new();
    for migration in &to_repair {
        let catalog = introspect_catalog_postgres_sync(client)?;
        let plan = drizzle_migrations::repair::plan(
            drizzle_types::Dialect::PostgreSQL,
            migration,
            &catalog,
        );
        for statement in plan
            .into_executable(&table_ident)
            .map_err(migrator_cli_error)?
        {
            client
                .execute(statement.as_str(), &[])
                .map_err(|e| CliError::MigrationError(format!("Repair statement failed: {e}")))?;
        }
        client
            .execute(&set.record_migration_finished_sql(migration), &[])
            .map_err(|e| CliError::MigrationError(e.to_string()))?;
        repaired.push(migration.tag().to_string());
    }
    Ok(repaired)
}

#[cfg(feature = "tokio-postgres")]
async fn introspect_catalog_postgres_async(
    client: &tokio_postgres::Client,
) -> Result<drizzle_migrations::repair::Catalog, CliError> {
    use drizzle_migrations::repair::postgres as pg;

    async fn two(
        client: &tokio_postgres::Client,
        sql: &str,
    ) -> Result<Vec<(String, String)>, CliError> {
        Ok(client
            .query(sql, &[])
            .await
            .map_err(|e| CliError::MigrationError(e.to_string()))?
            .iter()
            .map(|row| (row.get::<_, String>(0), row.get::<_, String>(1)))
            .collect())
    }

    async fn three(
        client: &tokio_postgres::Client,
        sql: &str,
    ) -> Result<Vec<(String, String, String)>, CliError> {
        Ok(client
            .query(sql, &[])
            .await
            .map_err(|e| CliError::MigrationError(e.to_string()))?
            .iter()
            .map(|row| {
                (
                    row.get::<_, String>(0),
                    row.get::<_, String>(1),
                    row.get::<_, String>(2),
                )
            })
            .collect())
    }

    let indexes = client
        .query(pg::INDEXES_QUERY, &[])
        .await
        .map_err(|e| CliError::MigrationError(e.to_string()))?
        .iter()
        .map(|row| {
            (
                row.get::<_, String>(0),
                row.get::<_, String>(1),
                row.get::<_, bool>(2),
            )
        })
        .collect::<Vec<_>>();

    let tables = two(client, pg::TABLES_QUERY).await?;
    let columns = three(client, pg::COLUMNS_QUERY).await?;
    let views = two(client, pg::VIEWS_QUERY).await?;
    let index_columns = three(client, pg::INDEX_COLUMNS_QUERY).await?;
    let enums = three(client, pg::ENUMS_QUERY).await?;

    Ok(pg::catalog(
        &tables,
        &columns,
        &views,
        &indexes,
        &index_columns,
        &enums,
    ))
}

#[cfg(feature = "tokio-postgres")]
async fn repair_dirty_migrations_postgres_async(
    client: &tokio_postgres::Client,
    set: &Migrations,
    repair: bool,
) -> Result<Vec<String>, CliError> {
    let dirty = client
        .query(&set.dirty_names_sql(), &[])
        .await
        .map_err(|e| CliError::MigrationError(e.to_string()))?
        .iter()
        .map(|row| row.try_get::<_, String>(0))
        .collect::<Result<Vec<_>, _>>()
        .map_err(|e| CliError::MigrationError(e.to_string()))?;

    let table_ident = set.table_ident_sql();
    let to_repair = dirty_migrations_to_repair(set, &dirty, repair)?
        .into_iter()
        .cloned()
        .collect::<Vec<_>>();

    let mut repaired = Vec::new();
    for migration in &to_repair {
        let catalog = introspect_catalog_postgres_async(client).await?;
        let plan = drizzle_migrations::repair::plan(
            drizzle_types::Dialect::PostgreSQL,
            migration,
            &catalog,
        );
        for statement in plan
            .into_executable(&table_ident)
            .map_err(migrator_cli_error)?
        {
            client
                .execute(statement.as_str(), &[])
                .await
                .map_err(|e| CliError::MigrationError(format!("Repair statement failed: {e}")))?;
        }
        client
            .execute(&set.record_migration_finished_sql(migration), &[])
            .await
            .map_err(|e| CliError::MigrationError(e.to_string()))?;
        repaired.push(migration.tag().to_string());
    }
    Ok(repaired)
}

/// Reads the tracking table's state without changing it.
#[cfg(feature = "rusqlite")]
fn read_sqlite_tracking_state(
    conn: &rusqlite::Connection,
    set: &Migrations,
) -> Result<TrackingState, CliError> {
    let pragma_sql = format!(
        "SELECT name FROM pragma_table_info('{}')",
        escape_sql_literal(set.table_name())
    );
    let mut stmt = conn
        .prepare(&pragma_sql)
        .map_err(|e| CliError::MigrationError(e.to_string()))?;
    let columns = stmt
        .query_map([], |row| row.get::<_, String>(0))
        .map_err(|e| CliError::MigrationError(e.to_string()))?
        .collect::<Result<Vec<_>, _>>()
        .map_err(|e| CliError::MigrationError(e.to_string()))?;
    if columns.is_empty() {
        return Ok(TrackingState::Missing);
    }
    if columns.iter().any(|column| column == "name") {
        return Ok(TrackingState::Current);
    }

    let mut stmt = conn
        .prepare(&format!(
            "SELECT id, hash, created_at FROM {} ORDER BY id ASC",
            set.table_ident_sql()
        ))
        .map_err(|e| CliError::MigrationError(e.to_string()))?;
    let rows = stmt
        .query_map([], |row| {
            Ok(drizzle_migrations::AppliedMigrationMetadata {
                id: row.get::<_, Option<i64>>(0)?,
                hash: row.get::<_, String>(1)?,
                created_at: row.get::<_, i64>(2)?,
            })
        })
        .map_err(|e| CliError::MigrationError(e.to_string()))?
        .collect::<Result<Vec<_>, _>>()
        .map_err(|e| CliError::MigrationError(e.to_string()))?;
    Ok(TrackingState::Legacy(rows))
}

#[cfg(feature = "rusqlite")]
fn ensure_sqlite_tracking_table(
    conn: &rusqlite::Connection,
    set: &Migrations,
) -> Result<(), CliError> {
    conn.execute(&set.create_table_sql(), [])
        .map_err(|e| CliError::MigrationError(format!("Failed to create migrations table: {e}")))?;

    let TrackingState::Legacy(applied) = read_sqlite_tracking_state(conn, set)? else {
        return Ok(());
    };

    let matched = drizzle_migrations::match_applied_migration_metadata(set.all(), &applied)
        .map_err(|e| CliError::MigrationError(e.to_string()))?;

    conn.execute("BEGIN", [])
        .map_err(|e| CliError::MigrationError(e.to_string()))?;
    let result = (|| -> Result<(), CliError> {
        conn.execute(
            &format!(
                "ALTER TABLE {} ADD COLUMN \"name\" text",
                set.table_ident_sql()
            ),
            [],
        )
        .map_err(|e| CliError::MigrationError(e.to_string()))?;
        conn.execute(
            &format!(
                "ALTER TABLE {} ADD COLUMN \"applied_at\" TEXT",
                set.table_ident_sql()
            ),
            [],
        )
        .map_err(|e| CliError::MigrationError(e.to_string()))?;

        for row in matched {
            conn.execute(&set.backfill_migration_metadata_sql(&row), [])
                .map_err(|e| CliError::MigrationError(e.to_string()))?;
        }

        Ok(())
    })();

    match result {
        Ok(()) => {
            conn.execute("COMMIT", [])
                .map_err(|e| CliError::MigrationError(e.to_string()))?;
            Ok(())
        }
        Err(err) => {
            let _ = conn.execute("ROLLBACK", []);
            Err(err)
        }
    }
}

/// Reads the tracking table's state without changing it.
#[cfg(any(feature = "libsql", feature = "turso"))]
async fn read_sqlite_tracking_state_libsql(
    conn: &libsql::Connection,
    set: &Migrations,
) -> Result<TrackingState, CliError> {
    let pragma_sql = format!(
        "SELECT name FROM pragma_table_info('{}')",
        escape_sql_literal(set.table_name())
    );
    let mut rows = conn
        .query(&pragma_sql, ())
        .await
        .map_err(|e| CliError::MigrationError(e.to_string()))?;
    let mut any_column = false;
    let mut has_name = false;
    while let Some(row) = rows
        .next()
        .await
        .map_err(|e| CliError::MigrationError(e.to_string()))?
    {
        any_column = true;
        let name = row
            .get::<String>(0)
            .map_err(|e| CliError::MigrationError(e.to_string()))?;
        if name == "name" {
            has_name = true;
        }
    }
    if !any_column {
        return Ok(TrackingState::Missing);
    }
    if has_name {
        return Ok(TrackingState::Current);
    }

    let mut rows = conn
        .query(
            &format!(
                "SELECT id, hash, created_at FROM {} ORDER BY id ASC",
                set.table_ident_sql()
            ),
            (),
        )
        .await
        .map_err(|e| CliError::MigrationError(e.to_string()))?;
    let mut applied = Vec::new();
    while let Some(row) = rows
        .next()
        .await
        .map_err(|e| CliError::MigrationError(e.to_string()))?
    {
        applied.push(drizzle_migrations::AppliedMigrationMetadata {
            id: row
                .get::<Option<i64>>(0)
                .map_err(|e| CliError::MigrationError(e.to_string()))?,
            hash: row
                .get::<String>(1)
                .map_err(|e| CliError::MigrationError(e.to_string()))?,
            created_at: row
                .get::<i64>(2)
                .map_err(|e| CliError::MigrationError(e.to_string()))?,
        });
    }
    Ok(TrackingState::Legacy(applied))
}

#[cfg(any(feature = "libsql", feature = "turso"))]
async fn ensure_sqlite_tracking_table_libsql(
    conn: &libsql::Connection,
    set: &Migrations,
) -> Result<(), CliError> {
    conn.execute(&set.create_table_sql(), ())
        .await
        .map_err(|e| CliError::MigrationError(format!("Failed to create migrations table: {e}")))?;

    let TrackingState::Legacy(applied) = read_sqlite_tracking_state_libsql(conn, set).await? else {
        return Ok(());
    };

    let matched = drizzle_migrations::match_applied_migration_metadata(set.all(), &applied)
        .map_err(|e| CliError::MigrationError(e.to_string()))?;

    let tx = conn
        .transaction()
        .await
        .map_err(|e| CliError::MigrationError(e.to_string()))?;
    tx.execute(
        &format!(
            "ALTER TABLE {} ADD COLUMN \"name\" text",
            set.table_ident_sql()
        ),
        (),
    )
    .await
    .map_err(|e| CliError::MigrationError(e.to_string()))?;
    tx.execute(
        &format!(
            "ALTER TABLE {} ADD COLUMN \"applied_at\" TEXT",
            set.table_ident_sql()
        ),
        (),
    )
    .await
    .map_err(|e| CliError::MigrationError(e.to_string()))?;

    for row in matched {
        tx.execute(&set.backfill_migration_metadata_sql(&row), ())
            .await
            .map_err(|e| CliError::MigrationError(e.to_string()))?;
    }

    tx.commit()
        .await
        .map_err(|e| CliError::MigrationError(e.to_string()))?;
    Ok(())
}

/// Reads the tracking table's state without changing it.
#[cfg(feature = "postgres-sync")]
fn read_postgres_tracking_state_sync(
    client: &mut postgres::Client,
    set: &Migrations,
) -> Result<TrackingState, CliError> {
    let schema = set.schema_name().unwrap_or("public");
    let rows = client
        .query(
            "SELECT column_name FROM information_schema.columns WHERE table_schema = $1 AND table_name = $2",
            &[&schema, &set.table_name()],
        )
        .map_err(|e| CliError::MigrationError(e.to_string()))?;
    let columns = rows
        .iter()
        .map(|row| row.try_get::<_, String>(0))
        .collect::<Result<Vec<_>, postgres::Error>>()
        .map_err(|e| CliError::MigrationError(e.to_string()))?;
    if columns.is_empty() {
        return Ok(TrackingState::Missing);
    }
    // Both columns, checked separately: an interrupted pre-transactional
    // upgrade could leave `name` without `applied_at`.
    if has_current_tracking_columns(&columns) {
        return Ok(TrackingState::Current);
    }

    let rows = client
        .query(
            &format!(
                "SELECT id::bigint, hash, created_at FROM {} ORDER BY id ASC",
                set.table_ident_sql()
            ),
            &[],
        )
        .map_err(|e| CliError::MigrationError(e.to_string()))?;
    let applied = rows
        .iter()
        .map(|row| {
            Ok(drizzle_migrations::AppliedMigrationMetadata {
                id: row.try_get::<_, Option<i64>>(0)?,
                hash: row.try_get::<_, String>(1)?,
                created_at: row.try_get::<_, i64>(2)?,
            })
        })
        .collect::<Result<Vec<_>, postgres::Error>>()
        .map_err(|e| CliError::MigrationError(e.to_string()))?;
    Ok(TrackingState::Legacy(applied))
}

#[cfg(feature = "postgres-sync")]
fn ensure_postgres_tracking_table_sync(
    client: &mut postgres::Client,
    set: &Migrations,
) -> Result<(), CliError> {
    client
        .execute(&set.create_table_sql(), &[])
        .map_err(|e| CliError::MigrationError(format!("Failed to create migrations table: {e}")))?;

    let TrackingState::Legacy(applied) = read_postgres_tracking_state_sync(client, set)? else {
        return Ok(());
    };

    let matched = drizzle_migrations::match_applied_migration_metadata(set.all(), &applied)
        .map_err(|e| CliError::MigrationError(e.to_string()))?;
    let (has_name, has_applied_at) = postgres_tracking_columns_sync(client, set)?;

    // One simple-query batch is one implicit transaction: the column
    // additions and backfills land together or not at all.
    client
        .batch_execute(
            &set.tracking_upgrade_sql(has_name, has_applied_at, &matched)
                .join(";\n"),
        )
        .map_err(|e| {
            CliError::MigrationError(format!(
                "Failed to upgrade the migrations table: {}",
                postgres_error_text(&e)
            ))
        })?;

    Ok(())
}

/// Whether the tracking table has the `name` and `applied_at` columns.
#[cfg(feature = "postgres-sync")]
fn postgres_tracking_columns_sync(
    client: &mut postgres::Client,
    set: &Migrations,
) -> Result<(bool, bool), CliError> {
    let schema = set.schema_name().unwrap_or("public");
    let columns = client
        .query(
            "SELECT column_name FROM information_schema.columns WHERE table_schema = $1 AND table_name = $2",
            &[&schema, &set.table_name()],
        )
        .map_err(|e| CliError::MigrationError(e.to_string()))?
        .iter()
        .map(|row| row.try_get::<_, String>(0))
        .collect::<Result<Vec<_>, postgres::Error>>()
        .map_err(|e| CliError::MigrationError(e.to_string()))?;
    Ok((
        columns.iter().any(|column| column == "name"),
        columns.iter().any(|column| column == "applied_at"),
    ))
}

/// `true` when a tracking table already has both v1 columns.
#[cfg(any(feature = "postgres-sync", feature = "tokio-postgres"))]
fn has_current_tracking_columns(columns: &[String]) -> bool {
    columns.iter().any(|column| column == "name")
        && columns.iter().any(|column| column == "applied_at")
}

/// A PostgreSQL error with the server's message: the client error alone
/// only displays `db error`.
#[cfg(any(feature = "postgres-sync", feature = "tokio-postgres"))]
fn postgres_error_text(error: &(dyn std::error::Error + 'static)) -> String {
    match error.source() {
        Some(source) => format!("{error}: {source}"),
        None => error.to_string(),
    }
}

/// Error for a failed migration statement, naming the migration, the
/// statement, and the server's message.
#[cfg(any(feature = "postgres-sync", feature = "tokio-postgres"))]
fn migration_statement_error(
    migration: &drizzle_migrations::Migration,
    statement: &str,
    error: &(dyn std::error::Error + 'static),
) -> CliError {
    CliError::MigrationError(format!(
        "Migration '{}' failed: {}\n  statement: {}",
        migration.tag(),
        postgres_error_text(error),
        statement.trim()
    ))
}

/// Reads the tracking table's state without changing it.
#[cfg(feature = "tokio-postgres")]
async fn read_postgres_tracking_state_async(
    client: &tokio_postgres::Client,
    set: &Migrations,
) -> Result<TrackingState, CliError> {
    let schema = set.schema_name().unwrap_or("public");
    let rows = client
        .query(
            "SELECT column_name FROM information_schema.columns WHERE table_schema = $1 AND table_name = $2",
            &[&schema, &set.table_name()],
        )
        .await
        .map_err(|e| CliError::MigrationError(e.to_string()))?;
    let columns = rows
        .iter()
        .map(|row| row.try_get::<_, String>(0))
        .collect::<Result<Vec<_>, tokio_postgres::Error>>()
        .map_err(|e| CliError::MigrationError(e.to_string()))?;
    if columns.is_empty() {
        return Ok(TrackingState::Missing);
    }
    // Both columns, checked separately: an interrupted pre-transactional
    // upgrade could leave `name` without `applied_at`.
    if has_current_tracking_columns(&columns) {
        return Ok(TrackingState::Current);
    }

    let rows = client
        .query(
            &format!(
                "SELECT id::bigint, hash, created_at FROM {} ORDER BY id ASC",
                set.table_ident_sql()
            ),
            &[],
        )
        .await
        .map_err(|e| CliError::MigrationError(e.to_string()))?;
    let applied = rows
        .iter()
        .map(|row| {
            Ok(drizzle_migrations::AppliedMigrationMetadata {
                id: row.try_get::<_, Option<i64>>(0)?,
                hash: row.try_get::<_, String>(1)?,
                created_at: row.try_get::<_, i64>(2)?,
            })
        })
        .collect::<Result<Vec<_>, tokio_postgres::Error>>()
        .map_err(|e| CliError::MigrationError(e.to_string()))?;
    Ok(TrackingState::Legacy(applied))
}

#[cfg(feature = "tokio-postgres")]
async fn ensure_postgres_tracking_table_async(
    client: &tokio_postgres::Client,
    set: &Migrations,
) -> Result<(), CliError> {
    client
        .execute(&set.create_table_sql(), &[])
        .await
        .map_err(|e| CliError::MigrationError(format!("Failed to create migrations table: {e}")))?;

    let TrackingState::Legacy(applied) = read_postgres_tracking_state_async(client, set).await?
    else {
        return Ok(());
    };

    let matched = drizzle_migrations::match_applied_migration_metadata(set.all(), &applied)
        .map_err(|e| CliError::MigrationError(e.to_string()))?;
    let schema = set.schema_name().unwrap_or("public");
    let columns = client
        .query(
            "SELECT column_name FROM information_schema.columns WHERE table_schema = $1 AND table_name = $2",
            &[&schema, &set.table_name()],
        )
        .await
        .map_err(|e| CliError::MigrationError(e.to_string()))?
        .iter()
        .map(|row| row.try_get::<_, String>(0))
        .collect::<Result<Vec<_>, tokio_postgres::Error>>()
        .map_err(|e| CliError::MigrationError(e.to_string()))?;
    let has_name = columns.iter().any(|column| column == "name");
    let has_applied_at = columns.iter().any(|column| column == "applied_at");

    // One simple-query batch is one implicit transaction: the column
    // additions and backfills land together or not at all.
    client
        .batch_execute(
            &set.tracking_upgrade_sql(has_name, has_applied_at, &matched)
                .join(";\n"),
        )
        .await
        .map_err(|e| {
            CliError::MigrationError(format!(
                "Failed to upgrade the migrations table: {}",
                postgres_error_text(&e)
            ))
        })?;

    Ok(())
}

fn is_destructive_statement(sql: &str) -> bool {
    let s = sql.trim().to_uppercase();
    s.contains("DROP TABLE")
        || s.contains("DROP COLUMN")
        || s.contains("DROP INDEX")
        || s.contains("DROP VIEW")
        || s.contains("DROP MATERIALIZED VIEW")
        || s.contains("DROP TYPE")
        || s.contains("DROP SCHEMA")
        || s.contains("DROP SEQUENCE")
        || s.contains("DROP ROLE")
        || s.contains("DROP POLICY")
        || s.contains("TRUNCATE")
        || (s.contains("ALTER TABLE") && s.contains(" DROP "))
}

fn push_requires_confirmation(
    dialect: Dialect,
    statements: &[String],
    warnings: &[String],
) -> bool {
    statements
        .iter()
        .any(|statement| is_destructive_statement(statement))
        // MySQL's ALTER TABLE ... MODIFY COLUMN syntax does not say DROP even
        // when it can truncate, reject, or recode stored values. Every warning
        // produced by the typed MySQL diff represents one of those risky
        // structural operations, so it belongs behind the same confirmation
        // boundary as explicit DROP/TRUNCATE statements.
        || (dialect == Dialect::Mysql && !warnings.is_empty())
}

#[cfg(any(feature = "postgres-sync", feature = "tokio-postgres"))]
fn is_postgres_concurrent_index_statement(sql: &str) -> bool {
    drizzle_migrations::is_postgres_concurrent_index_statement(sql)
}

#[cfg(any(feature = "postgres-sync", feature = "tokio-postgres"))]
fn has_postgres_concurrent_index(statements: &[String]) -> bool {
    statements
        .iter()
        .any(|stmt| is_postgres_concurrent_index_statement(stmt))
}

fn confirm_destructive() -> Result<bool, CliError> {
    use std::io::{self, IsTerminal, Write};

    if !io::stdin().is_terminal() {
        return Err(CliError::Other(
            "Refusing to apply potentially destructive changes in non-interactive mode. Use --explain or --force."
                .into(),
        ));
    }

    println!(
        "{}",
        output::warning("Potentially destructive changes detected (DROP/TRUNCATE/etc).")
    );
    print!("Apply anyway? [y/N]: ");
    io::stdout()
        .flush()
        .map_err(|e| CliError::IoError(e.to_string()))?;

    let mut line = String::new();
    io::stdin()
        .read_line(&mut line)
        .map_err(|e| CliError::IoError(e.to_string()))?;
    let ans = line.trim().to_ascii_lowercase();
    Ok(ans == "y" || ans == "yes")
}

fn generate_push_sql(
    current: &Snapshot,
    desired: &Snapshot,
    _breakpoints: bool,
    options: &drizzle_migrations::DiffOptions,
) -> Result<(Vec<String>, Vec<String>), CliError> {
    let plan = drizzle_migrations::diff_with(current, desired, options)
        .map_err(|error| CliError::MigrationError(error.to_string()))?;
    Ok((plan.statements, plan.warnings))
}

fn execute_statements(
    connection: &ResolvedConnection,
    statements: &[String],
) -> Result<(), CliError> {
    // In some feature combinations (no drivers), the match arms that would use `statements`
    // are compiled out. Touch it to avoid unused-parameter warnings.
    let _ = statements;

    match &connection.credentials {
        #[cfg(feature = "rusqlite")]
        Credentials::Sqlite { path } if connection.driver == Driver::Rusqlite => {
            execute_sqlite_statements(path, statements)
        }

        #[cfg(feature = "rusqlite")]
        Credentials::Sqlite { .. } => Err(connection_driver_mismatch(connection)),

        #[cfg(not(feature = "rusqlite"))]
        Credentials::Sqlite { .. } => Err(CliError::MissingDriver {
            dialect: "SQLite",
            feature: "rusqlite",
        }),

        #[cfg(any(feature = "libsql", feature = "turso"))]
        Credentials::Turso { url, auth_token } => {
            if connection.driver == Driver::Libsql {
                #[cfg(feature = "libsql")]
                {
                    execute_libsql_local_statements(url, statements)
                }
                #[cfg(not(feature = "libsql"))]
                {
                    Err(CliError::MissingDriver {
                        dialect: "LibSQL (local)",
                        feature: "libsql",
                    })
                }
            } else if connection.driver == Driver::Turso {
                #[cfg(feature = "turso")]
                {
                    execute_turso_statements(url, auth_token.as_deref(), statements)
                }
                #[cfg(not(feature = "turso"))]
                {
                    let _ = auth_token;
                    Err(CliError::MissingDriver {
                        dialect: "Turso (remote)",
                        feature: "turso",
                    })
                }
            } else {
                Err(connection_driver_mismatch(connection))
            }
        }

        #[cfg(all(not(feature = "turso"), not(feature = "libsql")))]
        Credentials::Turso { .. } => Err(CliError::MissingDriver {
            dialect: "Turso",
            feature: "turso or libsql",
        }),

        Credentials::Postgres(creds) => {
            let _ = creds;
            match connection.driver {
                Driver::PostgresSync => {
                    #[cfg(feature = "postgres-sync")]
                    {
                        execute_postgres_sync_statements(creds, statements)
                    }
                    #[cfg(not(feature = "postgres-sync"))]
                    {
                        Err(CliError::MissingDriver {
                            dialect: "PostgreSQL",
                            feature: "postgres-sync",
                        })
                    }
                }
                Driver::TokioPostgres => {
                    #[cfg(feature = "tokio-postgres")]
                    {
                        execute_postgres_async_statements(creds, statements)
                    }
                    #[cfg(not(feature = "tokio-postgres"))]
                    {
                        Err(CliError::MissingDriver {
                            dialect: "PostgreSQL",
                            feature: "tokio-postgres",
                        })
                    }
                }
                _ => Err(connection_driver_mismatch(connection)),
            }
        }

        Credentials::MySQL(creds) => {
            mysql::execute_statements(connection.driver, creds, statements)
        }

        #[cfg(feature = "d1-http")]
        Credentials::D1 {
            account_id,
            database_id,
            token,
        } if connection.driver == Driver::D1Http => {
            d1_http::execute_statements(account_id, database_id, token, statements)
        }

        #[cfg(feature = "d1-http")]
        Credentials::D1 { .. } => Err(connection_driver_mismatch(connection)),

        #[cfg(not(feature = "d1-http"))]
        Credentials::D1 { .. } => Err(CliError::MissingDriver {
            dialect: "Cloudflare D1 (HTTP)",
            feature: "d1-http",
        }),

        Credentials::AwsDataApi { .. } => Err(CliError::UnsupportedForDriver {
            operation: "Direct SQL execution against AWS Data API",
            driver: "aws-data-api",
            hint: "Use `aws rds-data execute-statement --sql=\"...\"` with the matching \
                   --resource-arn / --secret-arn / --database, or connect directly via \
                   tokio-postgres.",
        }),
    }
}

/// Check if a Turso URL is a local libsql database
#[allow(dead_code)]
fn is_local_libsql(url: &str) -> bool {
    url.starts_with("file:")
        || url.starts_with("./")
        || url.starts_with('/')
        || !url.contains("://")
}

// ============================================================================
// SQLite (rusqlite)
// ============================================================================

#[cfg(feature = "rusqlite")]
fn execute_sqlite_statements(path: &str, statements: &[String]) -> Result<(), CliError> {
    let conn = rusqlite::Connection::open(path).map_err(|e| {
        CliError::ConnectionError(format!("Failed to open SQLite database '{path}': {e}"))
    })?;

    execute_sqlite_statements_on_connection(&conn, statements)
}

#[cfg(feature = "rusqlite")]
fn execute_sqlite_statements_on_connection(
    conn: &rusqlite::Connection,
    statements: &[String],
) -> Result<(), CliError> {
    let operation =
        drizzle_migrations::Migration::with_hash("sqlite_push", "", 0, statements.to_vec());
    let execution = operation
        .sqlite_execution()
        .map_err(sqlite_execution_cli_error)?;
    let foreign_keys_were_enabled = sqlite_foreign_keys_enabled_cli(conn)?;
    let suspend_foreign_keys = execution.suspends_foreign_keys() && foreign_keys_were_enabled;
    if suspend_foreign_keys && let Err(error) = set_sqlite_foreign_keys_cli(conn, false) {
        let restore = set_sqlite_foreign_keys_cli(conn, true);
        return finish_sqlite_foreign_key_scope(Err(error), restore);
    }

    let result = (|| -> Result<(), CliError> {
        conn.execute("BEGIN IMMEDIATE", [])
            .map_err(|error| CliError::MigrationError(error.to_string()))?;
        for statement in execution
            .statements()
            .filter(|statement| !statement.trim().is_empty())
        {
            conn.execute(statement, []).map_err(|error| {
                CliError::MigrationError(format!("Statement failed: {error}\n{statement}"))
            })?;
        }
        if execution.suspends_foreign_keys() {
            verify_sqlite_foreign_keys_cli(conn)?;
        }
        conn.execute("COMMIT", [])
            .map_err(|error| CliError::MigrationError(error.to_string()))?;
        Ok(())
    })();
    let result = match result {
        Ok(()) => Ok(()),
        Err(error) => {
            let _ = conn.execute("ROLLBACK", []);
            Err(error)
        }
    };
    let restore = if suspend_foreign_keys {
        set_sqlite_foreign_keys_cli(conn, true)
    } else {
        Ok(())
    };
    finish_sqlite_foreign_key_scope(result, restore)
}

#[cfg(feature = "rusqlite")]
fn run_sqlite_migrations(
    set: &Migrations,
    path: &str,
    repair: bool,
) -> Result<MigrationResult, CliError> {
    let conn = rusqlite::Connection::open(path).map_err(|e| {
        CliError::ConnectionError(format!("Failed to open SQLite database '{path}': {e}"))
    })?;

    run_sqlite_migrations_on_connection(set, &conn, repair)
}

#[cfg(feature = "rusqlite")]
fn run_sqlite_migrations_on_connection(
    set: &Migrations,
    conn: &rusqlite::Connection,
    repair: bool,
) -> Result<MigrationResult, CliError> {
    ensure_sqlite_tracking_table(conn, set)?;
    conn.busy_timeout(std::time::Duration::from_secs(30))
        .map_err(|e| CliError::MigrationError(e.to_string()))?;
    let applied_before_transaction = query_applied_names_sqlite(conn, set)?;
    let suspends_foreign_keys = set
        .pending(&applied_before_transaction)
        .map(|migration| migration.sqlite_execution())
        .collect::<Result<Vec<_>, _>>()
        .map_err(sqlite_execution_cli_error)?
        .iter()
        .any(|execution| execution.suspends_foreign_keys());
    let foreign_keys_were_enabled = sqlite_foreign_keys_enabled_cli(conn)?;
    let suspend_foreign_keys = suspends_foreign_keys && foreign_keys_were_enabled;
    if suspend_foreign_keys && let Err(error) = set_sqlite_foreign_keys_cli(conn, false) {
        let restore = set_sqlite_foreign_keys_cli(conn, true);
        return finish_sqlite_foreign_key_scope(Err(error), restore);
    }

    let result = (|| -> Result<MigrationResult, CliError> {
        conn.execute("BEGIN IMMEDIATE", [])
            .map_err(|e| CliError::MigrationError(e.to_string()))?;
        let repaired = repair_dirty_migrations_sqlite(conn, set, repair)?;
        let applied_names = query_applied_names_sqlite(conn, set)?;
        let pending = set
            .pending(&applied_names)
            .map(|migration| {
                migration
                    .sqlite_execution()
                    .map(|execution| (migration, execution))
            })
            .collect::<Result<Vec<_>, _>>()
            .map_err(sqlite_execution_cli_error)?;

        let mut applied = Vec::with_capacity(pending.len());
        for (migration, execution) in pending {
            for statement in execution
                .statements()
                .filter(|statement| !statement.trim().is_empty())
            {
                conn.execute(statement, []).map_err(|error| {
                    CliError::MigrationError(format!(
                        "Migration '{}' failed: {error}",
                        migration.hash()
                    ))
                })?;
            }
            conn.execute(&set.record_migration_sql(migration), [])
                .map_err(|e| CliError::MigrationError(e.to_string()))?;
            applied.push(migration.hash().to_string());
        }

        Ok(MigrationResult {
            applied_count: applied.len(),
            applied_migrations: applied,
            repaired_migrations: repaired,
        })
    })();

    let result = match result {
        Ok(outcome) => match suspends_foreign_keys
            .then(|| verify_sqlite_foreign_keys_cli(conn))
            .transpose()
        {
            Ok(_) => conn
                .execute("COMMIT", [])
                .map(|_| outcome)
                .map_err(|e| CliError::MigrationError(e.to_string())),
            Err(error) => {
                let _ = conn.execute("ROLLBACK", []);
                Err(error)
            }
        },
        Err(error) => {
            let _ = conn.execute("ROLLBACK", []);
            Err(error)
        }
    };
    let restore = if suspend_foreign_keys {
        set_sqlite_foreign_keys_cli(conn, true)
    } else {
        Ok(())
    };
    finish_sqlite_foreign_key_scope(result, restore)
}

#[cfg(feature = "rusqlite")]
fn sqlite_foreign_keys_enabled_cli(conn: &rusqlite::Connection) -> Result<bool, CliError> {
    conn.query_row("PRAGMA foreign_keys", [], |row| row.get::<_, i64>(0))
        .map(|enabled| enabled != 0)
        .map_err(|error| CliError::MigrationError(error.to_string()))
}

#[cfg(feature = "rusqlite")]
fn set_sqlite_foreign_keys_cli(conn: &rusqlite::Connection, enabled: bool) -> Result<(), CliError> {
    conn.execute_batch(if enabled {
        "PRAGMA foreign_keys=ON"
    } else {
        "PRAGMA foreign_keys=OFF"
    })
    .map_err(|error| CliError::MigrationError(error.to_string()))?;
    if sqlite_foreign_keys_enabled_cli(conn)? != enabled {
        return Err(CliError::MigrationError(format!(
            "SQLite refused to set foreign_keys={}",
            if enabled { "ON" } else { "OFF" }
        )));
    }
    Ok(())
}

#[cfg(feature = "rusqlite")]
fn verify_sqlite_foreign_keys_cli(conn: &rusqlite::Connection) -> Result<(), CliError> {
    let mut statement = conn
        .prepare("PRAGMA foreign_key_check")
        .map_err(|error| CliError::MigrationError(error.to_string()))?;
    let mut rows = statement
        .query([])
        .map_err(|error| CliError::MigrationError(error.to_string()))?;
    if rows
        .next()
        .map_err(|error| CliError::MigrationError(error.to_string()))?
        .is_some()
    {
        return Err(CliError::MigrationError(
            "SQLite foreign_key_check failed after migration rebuild".to_string(),
        ));
    }
    Ok(())
}

#[cfg(feature = "rusqlite")]
fn inspect_sqlite_migrations(set: &Migrations, path: &str) -> Result<MigrationPlan, CliError> {
    if !sqlite_database_exists(path) {
        return build_migration_plan(set, &[]);
    }
    let conn = rusqlite::Connection::open(path).map_err(|e| {
        CliError::ConnectionError(format!("Failed to open SQLite database '{path}': {e}"))
    })?;

    let applied = match read_sqlite_tracking_state(&conn, set)? {
        TrackingState::Missing => Vec::new(),
        TrackingState::Legacy(rows) => legacy_applied_records(set, &rows)?,
        TrackingState::Current => query_applied_records_sqlite(&conn, set)?,
    };
    build_migration_plan(set, &applied)
}

#[cfg(feature = "rusqlite")]
fn is_sqlite_missing_table(error: &rusqlite::Error) -> bool {
    matches!(
        error,
        rusqlite::Error::SqliteFailure(_, Some(message)) if message.contains("no such table")
    )
}

#[cfg(feature = "rusqlite")]
fn query_applied_records_sqlite(
    conn: &rusqlite::Connection,
    set: &Migrations,
) -> Result<Vec<AppliedMigrationRecord>, CliError> {
    let sql = set.applied_records_sql();
    let mut stmt = match conn.prepare(&sql) {
        Ok(stmt) => stmt,
        Err(error) if is_sqlite_missing_table(&error) => return Ok(vec![]),
        Err(error) => return Err(CliError::MigrationError(error.to_string())),
    };

    let mut applied = Vec::new();
    let rows = stmt
        .query_map([], |row| {
            Ok((
                row.get::<_, String>(0)?,
                row.get::<_, String>(1)?,
                row.get::<_, i64>(2)? != 0,
            ))
        })
        .map_err(|e| CliError::MigrationError(e.to_string()))?;

    for row in rows {
        let (hash, name, dirty) = row.map_err(|e| CliError::MigrationError(e.to_string()))?;
        applied.push(AppliedMigrationRecord { hash, name, dirty });
    }

    Ok(applied)
}

#[cfg(feature = "rusqlite")]
fn query_applied_names_sqlite(
    conn: &rusqlite::Connection,
    set: &Migrations,
) -> Result<Vec<String>, CliError> {
    let mut stmt = match conn.prepare(&set.applied_names_sql()) {
        Ok(stmt) => stmt,
        Err(error) if is_sqlite_missing_table(&error) => return Ok(vec![]),
        Err(error) => return Err(CliError::MigrationError(error.to_string())),
    };

    let names = stmt
        .query_map([], |row| row.get::<_, String>(0))
        .map_err(|e| CliError::MigrationError(e.to_string()))?
        .collect::<Result<Vec<_>, _>>()
        .map_err(|error| CliError::MigrationError(error.to_string()))?;

    Ok(names)
}

// ============================================================================
// PostgreSQL (postgres - sync)
// ============================================================================

#[cfg(any(feature = "postgres-sync", feature = "tokio-postgres"))]
fn postgres_tls_connector(
    mode: crate::config::PostgresSslMode,
) -> Result<postgres_native_tls::MakeTlsConnector, CliError> {
    let mut builder = native_tls::TlsConnector::builder();
    if mode == crate::config::PostgresSslMode::VerifyCa {
        builder.danger_accept_invalid_hostnames(true);
    }
    let connector = builder
        .build()
        .map_err(|error| CliError::ConnectionError(error.to_string()))?;
    Ok(postgres_native_tls::MakeTlsConnector::new(connector))
}

#[cfg(feature = "postgres-sync")]
fn connect_postgres_sync(creds: &PostgresCreds) -> Result<postgres::Client, CliError> {
    let connection = creds
        .connection_config()
        .map_err(CliError::ConnectionError)?;
    let config = match creds {
        PostgresCreds::Url(url) => url
            .parse::<postgres::Config>()
            .map_err(|error| CliError::ConnectionError(error.to_string()))?,
        PostgresCreds::Host {
            host,
            port,
            user,
            password,
            database,
            ..
        } => {
            let mut config = postgres::Config::new();
            config
                .host(host.as_ref())
                .port(*port)
                .dbname(database.as_ref());
            if let Some(user) = user {
                config.user(user.as_ref());
            }
            if let Some(password) = password {
                config.password(password.as_bytes());
            }
            config.ssl_mode(connection.config.get_ssl_mode());
            config
        }
    };
    if connection.ssl == crate::config::PostgresSslMode::Disable {
        config
            .connect(postgres::NoTls)
            .map_err(|error| CliError::ConnectionError(error.to_string()))
    } else {
        let connector = postgres_tls_connector(connection.ssl)?;
        config
            .connect(connector)
            .map_err(|error| CliError::ConnectionError(error.to_string()))
    }
}

#[cfg(feature = "tokio-postgres")]
async fn connect_postgres_async(creds: &PostgresCreds) -> Result<tokio_postgres::Client, CliError> {
    let connection = creds
        .connection_config()
        .map_err(CliError::ConnectionError)?;
    if connection.ssl == crate::config::PostgresSslMode::Disable {
        let (client, driver) = connection
            .config
            .connect(tokio_postgres::NoTls)
            .await
            .map_err(|error| CliError::ConnectionError(error.to_string()))?;
        tokio::spawn(async move {
            if let Err(error) = driver.await {
                eprintln!("PostgreSQL connection error: {error}");
            }
        });
        Ok(client)
    } else {
        let connector = postgres_tls_connector(connection.ssl)?;
        let (client, driver) = connection
            .config
            .connect(connector)
            .await
            .map_err(|error| CliError::ConnectionError(error.to_string()))?;
        tokio::spawn(async move {
            if let Err(error) = driver.await {
                eprintln!("PostgreSQL connection error: {error}");
            }
        });
        Ok(client)
    }
}

#[cfg(feature = "postgres-sync")]
fn execute_postgres_sync_statements(
    creds: &PostgresCreds,
    statements: &[String],
) -> Result<(), CliError> {
    let mut client = connect_postgres_sync(creds)?;

    if has_postgres_concurrent_index(statements) {
        // CREATE/DROP INDEX CONCURRENTLY cannot run inside a transaction block.
        for stmt in statements {
            let s = stmt.trim();
            if s.is_empty() {
                continue;
            }
            client
                .execute(s, &[])
                .map_err(|e| CliError::MigrationError(format!("Statement failed: {e}\n{s}")))?;
        }
    } else {
        let mut tx = client
            .transaction()
            .map_err(|e| CliError::MigrationError(e.to_string()))?;

        for stmt in statements {
            let s = stmt.trim();
            if s.is_empty() {
                continue;
            }
            tx.execute(s, &[])
                .map_err(|e| CliError::MigrationError(format!("Statement failed: {e}\n{s}")))?;
        }

        tx.commit()
            .map_err(|e| CliError::MigrationError(e.to_string()))?;
    }

    Ok(())
}

#[cfg(feature = "postgres-sync")]
fn run_postgres_sync_migrations(
    set: &Migrations,
    creds: &PostgresCreds,
    repair: bool,
) -> Result<MigrationResult, CliError> {
    let mut client = connect_postgres_sync(creds)?;

    // Take the lock before creating the schema: concurrent first runs
    // otherwise race on `CREATE SCHEMA/TABLE IF NOT EXISTS`.
    let lock_key = set.postgres_advisory_lock_key();
    client
        .query_one("SELECT pg_advisory_lock($1)", &[&lock_key])
        .map_err(|error| CliError::MigrationError(error.to_string()))?;
    let result = run_postgres_sync_migrations_locked(&mut client, set, repair);
    let unlock = client
        .query_one("SELECT pg_advisory_unlock($1)", &[&lock_key])
        .map_err(|error| CliError::MigrationError(error.to_string()));
    match (result, unlock) {
        (Ok(result), Ok(_)) => Ok(result),
        (Err(error), _) => Err(error),
        (Ok(_), Err(error)) => Err(error),
    }
}

#[cfg(feature = "postgres-sync")]
fn run_postgres_sync_migrations_locked(
    client: &mut postgres::Client,
    set: &Migrations,
    repair: bool,
) -> Result<MigrationResult, CliError> {
    if let Some(schema_sql) = set.create_schema_sql() {
        client
            .execute(&schema_sql, &[])
            .map_err(|e| CliError::MigrationError(postgres_error_text(&e)))?;
    }
    ensure_postgres_tracking_table_sync(client, set)?;
    let repaired = repair_dirty_migrations_postgres_sync(client, set, repair)?;
    let rows = client
        .query(&set.applied_names_sql(), &[])
        .map_err(|error| CliError::MigrationError(error.to_string()))?;
    let applied_names = rows
        .iter()
        .map(|row| row.try_get(0))
        .collect::<Result<Vec<String>, _>>()
        .map_err(|error| CliError::MigrationError(error.to_string()))?;
    let pending: Vec<_> = set.pending(&applied_names).collect();
    if pending.is_empty() {
        return Ok(MigrationResult {
            applied_count: 0,
            applied_migrations: vec![],
            repaired_migrations: repaired,
        });
    }

    let mut applied = Vec::new();
    if set.has_postgres_concurrent_index() {
        // CREATE/DROP INDEX CONCURRENTLY cannot run in a transaction, so this
        // path uses two-phase tracking: mark the migration dirty, run the
        // statements autocommit, then stamp applied_at. A crash in between
        // leaves a recoverable dirty row instead of an untracked partial
        // schema.
        for migration in &pending {
            client
                .execute(&set.record_migration_started_sql(migration), &[])
                .map_err(|error| CliError::MigrationError(error.to_string()))?;

            let mut executed = 0usize;
            for statement in migration.statements() {
                if statement.trim().is_empty() {
                    continue;
                }
                if let Err(error) = client.batch_execute(statement) {
                    // Nothing applied yet - drop the marker instead of
                    // demanding a pointless repair.
                    if executed == 0 {
                        let _ = client.execute(&set.clear_migration_started_sql(migration), &[]);
                    }
                    return Err(migration_statement_error(migration, statement, &error));
                }
                executed += 1;
            }

            client
                .execute(&set.record_migration_finished_sql(migration), &[])
                .map_err(|error| CliError::MigrationError(error.to_string()))?;
            applied.push(migration.hash().to_string());
        }
    } else {
        let mut transaction = client
            .transaction()
            .map_err(|error| CliError::MigrationError(error.to_string()))?;
        for migration in &pending {
            for statement in migration.statements() {
                if !statement.trim().is_empty() {
                    // Simple-query protocol: a breakpoint chunk runs whole.
                    transaction
                        .batch_execute(statement)
                        .map_err(|error| migration_statement_error(migration, statement, &error))?;
                }
            }
            transaction
                .execute(&set.record_migration_sql(migration), &[])
                .map_err(|error| CliError::MigrationError(error.to_string()))?;
            applied.push(migration.hash().to_string());
        }
        transaction
            .commit()
            .map_err(|error| CliError::MigrationError(error.to_string()))?;
    }

    Ok(MigrationResult {
        applied_count: applied.len(),
        applied_migrations: applied,
        repaired_migrations: repaired,
    })
}

#[cfg(feature = "postgres-sync")]
fn inspect_postgres_sync_migrations(
    set: &Migrations,
    creds: &PostgresCreds,
) -> Result<MigrationPlan, CliError> {
    let mut client = connect_postgres_sync(creds)?;

    // Read-only: no CREATE SCHEMA, no tracking table.
    let applied = match read_postgres_tracking_state_sync(&mut client, set)? {
        TrackingState::Missing => Vec::new(),
        TrackingState::Legacy(rows) => legacy_applied_records(set, &rows)?,
        TrackingState::Current => query_applied_records_postgres_sync(&mut client, set)?,
    };
    build_migration_plan(set, &applied)
}

#[cfg(feature = "postgres-sync")]
fn query_applied_records_postgres_sync(
    client: &mut postgres::Client,
    set: &Migrations,
) -> Result<Vec<AppliedMigrationRecord>, CliError> {
    let rows = client
        .query(&set.applied_records_sql(), &[])
        .map_err(|error| CliError::MigrationError(error.to_string()))?;

    let mut applied = Vec::new();
    for row in rows {
        let hash = row
            .try_get::<_, Option<String>>(0)
            .map_err(|error| CliError::MigrationError(error.to_string()))?;
        let name = row
            .try_get::<_, Option<String>>(1)
            .map_err(|error| CliError::MigrationError(error.to_string()))?;
        let dirty = row
            .try_get::<_, bool>(2)
            .map_err(|error| CliError::MigrationError(error.to_string()))?;
        if let (Some(hash), Some(name)) = (hash, name) {
            applied.push(AppliedMigrationRecord { hash, name, dirty });
        }
    }

    Ok(applied)
}

// ============================================================================
// PostgreSQL (tokio-postgres - async)
// ============================================================================

#[cfg(feature = "tokio-postgres")]
fn execute_postgres_async_statements(
    creds: &PostgresCreds,
    statements: &[String],
) -> Result<(), CliError> {
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .map_err(|e| CliError::Other(format!("Failed to create async runtime: {}", e)))?;

    rt.block_on(execute_postgres_async_inner(creds, statements))
}

#[cfg(feature = "tokio-postgres")]
async fn execute_postgres_async_inner(
    creds: &PostgresCreds,
    statements: &[String],
) -> Result<(), CliError> {
    let mut client = connect_postgres_async(creds).await?;

    if has_postgres_concurrent_index(statements) {
        // CREATE/DROP INDEX CONCURRENTLY cannot run inside a transaction block.
        for stmt in statements {
            let s = stmt.trim();
            if s.is_empty() {
                continue;
            }
            client
                .execute(s, &[])
                .await
                .map_err(|e| CliError::MigrationError(format!("Statement failed: {}\n{}", e, s)))?;
        }
    } else {
        let tx = client
            .transaction()
            .await
            .map_err(|e| CliError::MigrationError(e.to_string()))?;

        for stmt in statements {
            let s = stmt.trim();
            if s.is_empty() {
                continue;
            }
            tx.execute(s, &[])
                .await
                .map_err(|e| CliError::MigrationError(format!("Statement failed: {}\n{}", e, s)))?;
        }

        tx.commit()
            .await
            .map_err(|e| CliError::MigrationError(e.to_string()))?;
    }

    Ok(())
}

#[cfg(feature = "tokio-postgres")]
#[allow(dead_code)] // Used when postgres-sync is not enabled
fn run_postgres_async_migrations(
    set: &Migrations,
    creds: &PostgresCreds,
    repair: bool,
) -> Result<MigrationResult, CliError> {
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .map_err(|e| CliError::Other(format!("Failed to create async runtime: {e}")))?;

    rt.block_on(run_postgres_async_inner(set, creds, repair))
}

#[cfg(feature = "tokio-postgres")]
#[allow(dead_code)] // Used when postgres-sync is not enabled
fn inspect_postgres_async_migrations(
    set: &Migrations,
    creds: &PostgresCreds,
) -> Result<MigrationPlan, CliError> {
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .map_err(|e| CliError::Other(format!("Failed to create async runtime: {e}")))?;

    rt.block_on(inspect_postgres_async_inner(set, creds))
}

#[cfg(feature = "tokio-postgres")]
#[allow(dead_code)]
async fn inspect_postgres_async_inner(
    set: &Migrations,
    creds: &PostgresCreds,
) -> Result<MigrationPlan, CliError> {
    let client = connect_postgres_async(creds).await?;

    // Read-only: no CREATE SCHEMA, no tracking table.
    let applied = match read_postgres_tracking_state_async(&client, set).await? {
        TrackingState::Missing => Vec::new(),
        TrackingState::Legacy(rows) => legacy_applied_records(set, &rows)?,
        TrackingState::Current => query_applied_records_postgres_async(&client, set).await?,
    };
    build_migration_plan(set, &applied)
}

#[cfg(feature = "tokio-postgres")]
async fn query_applied_records_postgres_async(
    client: &tokio_postgres::Client,
    set: &Migrations,
) -> Result<Vec<AppliedMigrationRecord>, CliError> {
    let rows = client
        .query(&set.applied_records_sql(), &[])
        .await
        .map_err(|error| CliError::MigrationError(error.to_string()))?;

    let mut applied = Vec::new();
    for row in rows {
        let hash = row
            .try_get::<_, Option<String>>(0)
            .map_err(|error| CliError::MigrationError(error.to_string()))?;
        let name = row
            .try_get::<_, Option<String>>(1)
            .map_err(|error| CliError::MigrationError(error.to_string()))?;
        let dirty = row
            .try_get::<_, bool>(2)
            .map_err(|error| CliError::MigrationError(error.to_string()))?;
        if let (Some(hash), Some(name)) = (hash, name) {
            applied.push(AppliedMigrationRecord { hash, name, dirty });
        }
    }

    Ok(applied)
}

#[cfg(feature = "tokio-postgres")]
#[allow(dead_code)]
async fn run_postgres_async_inner(
    set: &Migrations,
    creds: &PostgresCreds,
    repair: bool,
) -> Result<MigrationResult, CliError> {
    let mut client = connect_postgres_async(creds).await?;

    // Take the lock before creating the schema: concurrent first runs
    // otherwise race on `CREATE SCHEMA/TABLE IF NOT EXISTS`.
    let lock_key = set.postgres_advisory_lock_key();
    client
        .query_one("SELECT pg_advisory_lock($1)", &[&lock_key])
        .await
        .map_err(|error| CliError::MigrationError(error.to_string()))?;
    let result = run_postgres_async_migrations_locked(&mut client, set, repair).await;
    let unlock = client
        .query_one("SELECT pg_advisory_unlock($1)", &[&lock_key])
        .await
        .map_err(|error| CliError::MigrationError(error.to_string()));
    match (result, unlock) {
        (Ok(result), Ok(_)) => Ok(result),
        (Err(error), _) => Err(error),
        (Ok(_), Err(error)) => Err(error),
    }
}

#[cfg(feature = "tokio-postgres")]
async fn run_postgres_async_migrations_locked(
    client: &mut tokio_postgres::Client,
    set: &Migrations,
    repair: bool,
) -> Result<MigrationResult, CliError> {
    if let Some(schema_sql) = set.create_schema_sql() {
        client
            .execute(&schema_sql, &[])
            .await
            .map_err(|e| CliError::MigrationError(postgres_error_text(&e)))?;
    }
    ensure_postgres_tracking_table_async(client, set).await?;
    let repaired = repair_dirty_migrations_postgres_async(client, set, repair).await?;
    let rows = client
        .query(&set.applied_names_sql(), &[])
        .await
        .map_err(|error| CliError::MigrationError(error.to_string()))?;
    let applied_names = rows
        .iter()
        .map(|row| row.try_get(0))
        .collect::<Result<Vec<String>, _>>()
        .map_err(|error| CliError::MigrationError(error.to_string()))?;
    let pending: Vec<_> = set.pending(&applied_names).collect();
    if pending.is_empty() {
        return Ok(MigrationResult {
            applied_count: 0,
            applied_migrations: vec![],
            repaired_migrations: repaired,
        });
    }

    let mut applied = Vec::new();
    if set.has_postgres_concurrent_index() {
        // CREATE/DROP INDEX CONCURRENTLY cannot run in a transaction, so this
        // path uses two-phase tracking: mark the migration dirty, run the
        // statements autocommit, then stamp applied_at. A crash in between
        // leaves a recoverable dirty row instead of an untracked partial
        // schema.
        for migration in &pending {
            client
                .execute(&set.record_migration_started_sql(migration), &[])
                .await
                .map_err(|error| CliError::MigrationError(error.to_string()))?;

            let mut executed = 0usize;
            for statement in migration.statements() {
                if statement.trim().is_empty() {
                    continue;
                }
                if let Err(error) = client.batch_execute(statement).await {
                    // Nothing applied yet - drop the marker instead of
                    // demanding a pointless repair.
                    if executed == 0 {
                        let _ = client
                            .execute(&set.clear_migration_started_sql(migration), &[])
                            .await;
                    }
                    return Err(migration_statement_error(migration, statement, &error));
                }
                executed += 1;
            }

            client
                .execute(&set.record_migration_finished_sql(migration), &[])
                .await
                .map_err(|error| CliError::MigrationError(error.to_string()))?;
            applied.push(migration.hash().to_string());
        }
    } else {
        let transaction = client
            .transaction()
            .await
            .map_err(|error| CliError::MigrationError(error.to_string()))?;
        for migration in &pending {
            for statement in migration.statements() {
                if !statement.trim().is_empty() {
                    // Simple-query protocol: a breakpoint chunk runs whole.
                    transaction
                        .batch_execute(statement)
                        .await
                        .map_err(|error| migration_statement_error(migration, statement, &error))?;
                }
            }
            transaction
                .execute(&set.record_migration_sql(migration), &[])
                .await
                .map_err(|error| CliError::MigrationError(error.to_string()))?;
            applied.push(migration.hash().to_string());
        }
        transaction
            .commit()
            .await
            .map_err(|error| CliError::MigrationError(error.to_string()))?;
    }

    Ok(MigrationResult {
        applied_count: applied.len(),
        applied_migrations: applied,
        repaired_migrations: repaired,
    })
}

// ============================================================================
// LibSQL (local)
// ============================================================================

#[cfg(feature = "libsql")]
fn execute_libsql_local_statements(path: &str, statements: &[String]) -> Result<(), CliError> {
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .map_err(|e| CliError::Other(format!("Failed to create async runtime: {e}")))?;

    rt.block_on(execute_libsql_local_inner(path, statements))
}

#[cfg(feature = "libsql")]
async fn execute_libsql_local_inner(path: &str, statements: &[String]) -> Result<(), CliError> {
    let db = libsql::Builder::new_local(path)
        .build()
        .await
        .map_err(|e| {
            CliError::ConnectionError(format!("Failed to open LibSQL database '{path}': {e}"))
        })?;

    let conn = db
        .connect()
        .map_err(|e| CliError::ConnectionError(e.to_string()))?;

    execute_libsql_sqlite_statements_on_connection(&conn, statements).await
}

#[cfg(any(feature = "libsql", feature = "turso"))]
async fn execute_libsql_sqlite_statements_on_connection(
    conn: &libsql::Connection,
    statements: &[String],
) -> Result<(), CliError> {
    let operation =
        drizzle_migrations::Migration::with_hash("libsql_push", "", 0, statements.to_vec());
    let execution = operation
        .sqlite_execution()
        .map_err(sqlite_execution_cli_error)?;
    let foreign_keys_were_enabled = sqlite_foreign_keys_enabled_libsql_cli(conn).await?;
    let suspend_foreign_keys = execution.suspends_foreign_keys() && foreign_keys_were_enabled;
    if suspend_foreign_keys
        && let Err(error) = set_sqlite_foreign_keys_libsql_cli(conn, false).await
    {
        let restore = set_sqlite_foreign_keys_libsql_cli(conn, true).await;
        return finish_sqlite_foreign_key_scope(Err(error), restore);
    }

    let result = async {
        let tx = conn
            .transaction_with_behavior(libsql::TransactionBehavior::Immediate)
            .await
            .map_err(|error| CliError::MigrationError(error.to_string()))?;
        let transaction_result = async {
            for statement in execution
                .statements()
                .filter(|statement| !statement.trim().is_empty())
            {
                tx.execute(statement, ()).await.map_err(|error| {
                    CliError::MigrationError(format!("Statement failed: {error}\n{statement}"))
                })?;
            }
            if execution.suspends_foreign_keys() {
                verify_sqlite_foreign_keys_libsql_cli(&tx).await?;
            }
            Ok::<_, CliError>(())
        }
        .await;
        match transaction_result {
            Ok(()) => tx
                .commit()
                .await
                .map_err(|error| CliError::MigrationError(error.to_string())),
            Err(error) => {
                tx.rollback().await.ok();
                Err(error)
            }
        }
    }
    .await;
    let restore = if suspend_foreign_keys {
        set_sqlite_foreign_keys_libsql_cli(conn, true).await
    } else {
        Ok(())
    };
    finish_sqlite_foreign_key_scope(result, restore)
}

#[cfg(feature = "libsql")]
fn run_libsql_local_migrations(
    set: &Migrations,
    path: &str,
    repair: bool,
) -> Result<MigrationResult, CliError> {
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .map_err(|e| CliError::Other(format!("Failed to create async runtime: {e}")))?;

    rt.block_on(run_libsql_local_inner(set, path, repair))
}

#[cfg(feature = "libsql")]
fn inspect_libsql_local_migrations(
    set: &Migrations,
    path: &str,
) -> Result<MigrationPlan, CliError> {
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .map_err(|e| CliError::Other(format!("Failed to create async runtime: {e}")))?;

    rt.block_on(inspect_libsql_local_inner(set, path))
}

#[cfg(feature = "libsql")]
async fn inspect_libsql_local_inner(
    set: &Migrations,
    path: &str,
) -> Result<MigrationPlan, CliError> {
    if !sqlite_database_exists(path) {
        return build_migration_plan(set, &[]);
    }
    let db = libsql::Builder::new_local(path)
        .build()
        .await
        .map_err(|e| {
            CliError::ConnectionError(format!("Failed to open LibSQL database '{path}': {e}"))
        })?;

    let conn = db
        .connect()
        .map_err(|e| CliError::ConnectionError(e.to_string()))?;

    let applied = match read_sqlite_tracking_state_libsql(&conn, set).await? {
        TrackingState::Missing => Vec::new(),
        TrackingState::Legacy(rows) => legacy_applied_records(set, &rows)?,
        TrackingState::Current => query_applied_records_libsql(&conn, set).await?,
    };
    build_migration_plan(set, &applied)
}

#[cfg(feature = "libsql")]
async fn run_libsql_local_inner(
    set: &Migrations,
    path: &str,
    repair: bool,
) -> Result<MigrationResult, CliError> {
    let db = libsql::Builder::new_local(path)
        .build()
        .await
        .map_err(|e| {
            CliError::ConnectionError(format!("Failed to open LibSQL database '{path}': {e}"))
        })?;

    let conn = db
        .connect()
        .map_err(|e| CliError::ConnectionError(e.to_string()))?;

    run_libsql_sqlite_migrations_on_connection(set, &conn, repair).await
}

#[cfg(any(feature = "libsql", feature = "turso"))]
async fn query_applied_names_libsql(
    conn: &libsql::Connection,
    set: &Migrations,
) -> Result<Vec<String>, CliError> {
    let mut rows = conn
        .query(&set.applied_names_sql(), ())
        .await
        .map_err(|error| CliError::MigrationError(error.to_string()))?;

    let mut names = Vec::new();
    while let Some(row) = rows
        .next()
        .await
        .map_err(|error| CliError::MigrationError(error.to_string()))?
    {
        names.push(
            row.get::<String>(0)
                .map_err(|error| CliError::MigrationError(error.to_string()))?,
        );
    }

    Ok(names)
}

#[cfg(any(feature = "libsql", feature = "turso"))]
async fn run_libsql_sqlite_migrations_on_connection(
    set: &Migrations,
    conn: &libsql::Connection,
    repair: bool,
) -> Result<MigrationResult, CliError> {
    ensure_sqlite_tracking_table_libsql(conn, set).await?;
    let applied_before_transaction = query_applied_names_libsql(conn, set).await?;
    let suspends_foreign_keys = set
        .pending(&applied_before_transaction)
        .map(|migration| migration.sqlite_execution())
        .collect::<Result<Vec<_>, _>>()
        .map_err(sqlite_execution_cli_error)?
        .iter()
        .any(|execution| execution.suspends_foreign_keys());
    let foreign_keys_were_enabled = sqlite_foreign_keys_enabled_libsql_cli(conn).await?;
    let suspend_foreign_keys = suspends_foreign_keys && foreign_keys_were_enabled;
    if suspend_foreign_keys
        && let Err(error) = set_sqlite_foreign_keys_libsql_cli(conn, false).await
    {
        let restore = set_sqlite_foreign_keys_libsql_cli(conn, true).await;
        return finish_sqlite_foreign_key_scope(Err(error), restore);
    }

    let result = async {
        let repaired = repair_dirty_migrations_libsql(conn, set, repair).await?;
        let tx = conn
            .transaction_with_behavior(libsql::TransactionBehavior::Immediate)
            .await
            .map_err(|error| CliError::MigrationError(error.to_string()))?;
        let transaction_result = async {
            let applied_names = query_applied_names_libsql(&tx, set).await?;
            let pending = set
                .pending(&applied_names)
                .map(|migration| {
                    migration
                        .sqlite_execution()
                        .map(|execution| (migration, execution))
                })
                .collect::<Result<Vec<_>, _>>()
                .map_err(sqlite_execution_cli_error)?;

            let mut applied = Vec::with_capacity(pending.len());
            for (migration, execution) in pending {
                for statement in execution
                    .statements()
                    .filter(|statement| !statement.trim().is_empty())
                {
                    tx.execute(statement, ()).await.map_err(|error| {
                        CliError::MigrationError(format!(
                            "Migration '{}' failed: {error}",
                            migration.hash()
                        ))
                    })?;
                }
                tx.execute(&set.record_migration_sql(migration), ())
                    .await
                    .map_err(|error| CliError::MigrationError(error.to_string()))?;
                applied.push(migration.hash().to_string());
            }

            Ok::<_, CliError>(MigrationResult {
                applied_count: applied.len(),
                applied_migrations: applied,
                repaired_migrations: repaired,
            })
        }
        .await;

        match transaction_result {
            Ok(outcome) => {
                if suspends_foreign_keys
                    && let Err(error) = verify_sqlite_foreign_keys_libsql_cli(&tx).await
                {
                    tx.rollback().await.ok();
                    return Err(error);
                }
                tx.commit()
                    .await
                    .map_err(|error| CliError::MigrationError(error.to_string()))?;
                Ok(outcome)
            }
            Err(error) => {
                tx.rollback().await.ok();
                Err(error)
            }
        }
    }
    .await;
    let restore = if suspend_foreign_keys {
        set_sqlite_foreign_keys_libsql_cli(conn, true).await
    } else {
        Ok(())
    };
    finish_sqlite_foreign_key_scope(result, restore)
}

#[cfg(any(feature = "libsql", feature = "turso"))]
async fn sqlite_foreign_keys_enabled_libsql_cli(
    conn: &libsql::Connection,
) -> Result<bool, CliError> {
    let mut rows = conn
        .query("PRAGMA foreign_keys", ())
        .await
        .map_err(|error| CliError::MigrationError(error.to_string()))?;
    let row = rows
        .next()
        .await
        .map_err(|error| CliError::MigrationError(error.to_string()))?
        .ok_or_else(|| CliError::MigrationError("PRAGMA foreign_keys returned no row".into()))?;
    row.get::<i64>(0)
        .map(|enabled| enabled != 0)
        .map_err(|error| CliError::MigrationError(error.to_string()))
}

#[cfg(any(feature = "libsql", feature = "turso"))]
async fn set_sqlite_foreign_keys_libsql_cli(
    conn: &libsql::Connection,
    enabled: bool,
) -> Result<(), CliError> {
    conn.execute(
        if enabled {
            "PRAGMA foreign_keys=ON"
        } else {
            "PRAGMA foreign_keys=OFF"
        },
        (),
    )
    .await
    .map_err(|error| CliError::MigrationError(error.to_string()))?;
    if sqlite_foreign_keys_enabled_libsql_cli(conn).await? != enabled {
        return Err(CliError::MigrationError(format!(
            "SQLite refused to set foreign_keys={}",
            if enabled { "ON" } else { "OFF" }
        )));
    }
    Ok(())
}

#[cfg(any(feature = "libsql", feature = "turso"))]
async fn verify_sqlite_foreign_keys_libsql_cli(conn: &libsql::Connection) -> Result<(), CliError> {
    let mut rows = conn
        .query("PRAGMA foreign_key_check", ())
        .await
        .map_err(|error| CliError::MigrationError(error.to_string()))?;
    if rows
        .next()
        .await
        .map_err(|error| CliError::MigrationError(error.to_string()))?
        .is_some()
    {
        return Err(CliError::MigrationError(
            "SQLite foreign_key_check failed after migration rebuild".into(),
        ));
    }
    Ok(())
}

#[cfg(feature = "libsql")]
async fn query_applied_records_libsql(
    conn: &libsql::Connection,
    set: &Migrations,
) -> Result<Vec<AppliedMigrationRecord>, CliError> {
    let mut rows = conn
        .query(&set.applied_records_sql(), ())
        .await
        .map_err(|error| CliError::MigrationError(error.to_string()))?;

    let mut applied = Vec::new();
    while let Some(row) = rows
        .next()
        .await
        .map_err(|error| CliError::MigrationError(error.to_string()))?
    {
        let hash = row
            .get::<String>(0)
            .map_err(|error| CliError::MigrationError(error.to_string()))?;
        let name = row
            .get::<String>(1)
            .map_err(|error| CliError::MigrationError(error.to_string()))?;
        let dirty = row
            .get::<i64>(2)
            .map_err(|error| CliError::MigrationError(error.to_string()))?
            != 0;
        applied.push(AppliedMigrationRecord { hash, name, dirty });
    }

    Ok(applied)
}

// ============================================================================
// Turso (remote)
// ============================================================================

#[cfg(feature = "turso")]
fn execute_turso_statements(
    url: &str,
    auth_token: Option<&str>,
    statements: &[String],
) -> Result<(), CliError> {
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .map_err(|e| CliError::Other(format!("Failed to create async runtime: {e}")))?;

    rt.block_on(execute_turso_inner(url, auth_token, statements))
}

#[cfg(feature = "turso")]
async fn execute_turso_inner(
    url: &str,
    auth_token: Option<&str>,
    statements: &[String],
) -> Result<(), CliError> {
    let builder =
        libsql::Builder::new_remote(url.to_string(), auth_token.unwrap_or("").to_string());

    let db = builder.build().await.map_err(|e| {
        CliError::ConnectionError(format!("Failed to connect to Turso '{url}': {e}"))
    })?;

    let conn = db
        .connect()
        .map_err(|e| CliError::ConnectionError(e.to_string()))?;

    execute_libsql_sqlite_statements_on_connection(&conn, statements).await
}

#[cfg(feature = "turso")]
fn run_turso_migrations(
    set: &Migrations,
    url: &str,
    auth_token: Option<&str>,
    repair: bool,
) -> Result<MigrationResult, CliError> {
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .map_err(|e| CliError::Other(format!("Failed to create async runtime: {e}")))?;

    rt.block_on(run_turso_inner(set, url, auth_token, repair))
}

#[cfg(feature = "turso")]
fn inspect_turso_migrations(
    set: &Migrations,
    url: &str,
    auth_token: Option<&str>,
) -> Result<MigrationPlan, CliError> {
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .map_err(|e| CliError::Other(format!("Failed to create async runtime: {e}")))?;

    rt.block_on(inspect_turso_inner(set, url, auth_token))
}

#[cfg(feature = "turso")]
async fn inspect_turso_inner(
    set: &Migrations,
    url: &str,
    auth_token: Option<&str>,
) -> Result<MigrationPlan, CliError> {
    let builder =
        libsql::Builder::new_remote(url.to_string(), auth_token.unwrap_or("").to_string());

    let db = builder.build().await.map_err(|e| {
        CliError::ConnectionError(format!("Failed to connect to Turso '{url}': {e}"))
    })?;

    let conn = db
        .connect()
        .map_err(|e| CliError::ConnectionError(e.to_string()))?;

    let applied = match read_sqlite_tracking_state_libsql(&conn, set).await? {
        TrackingState::Missing => Vec::new(),
        TrackingState::Legacy(rows) => legacy_applied_records(set, &rows)?,
        TrackingState::Current => query_applied_records_turso(&conn, set).await?,
    };
    build_migration_plan(set, &applied)
}

#[cfg(feature = "turso")]
async fn run_turso_inner(
    set: &Migrations,
    url: &str,
    auth_token: Option<&str>,
    repair: bool,
) -> Result<MigrationResult, CliError> {
    let builder =
        libsql::Builder::new_remote(url.to_string(), auth_token.unwrap_or("").to_string());

    let db = builder.build().await.map_err(|e| {
        CliError::ConnectionError(format!("Failed to connect to Turso '{url}': {e}"))
    })?;

    let conn = db
        .connect()
        .map_err(|e| CliError::ConnectionError(e.to_string()))?;

    run_libsql_sqlite_migrations_on_connection(set, &conn, repair).await
}

#[cfg(feature = "turso")]
async fn query_applied_records_turso(
    conn: &libsql::Connection,
    set: &Migrations,
) -> Result<Vec<AppliedMigrationRecord>, CliError> {
    let mut rows = conn
        .query(&set.applied_records_sql(), ())
        .await
        .map_err(|error| CliError::MigrationError(error.to_string()))?;

    let mut applied = Vec::new();
    while let Some(row) = rows
        .next()
        .await
        .map_err(|error| CliError::MigrationError(error.to_string()))?
    {
        let hash = row
            .get::<String>(0)
            .map_err(|error| CliError::MigrationError(error.to_string()))?;
        let name = row
            .get::<String>(1)
            .map_err(|error| CliError::MigrationError(error.to_string()))?;
        let dirty = row
            .get::<i64>(2)
            .map_err(|error| CliError::MigrationError(error.to_string()))?
            != 0;
        applied.push(AppliedMigrationRecord { hash, name, dirty });
    }

    Ok(applied)
}

// ============================================================================
// Database Introspection
// ============================================================================

/// What [`run_introspection`] produced.
#[derive(Debug)]
pub struct IntrospectResult {
    /// Generated Rust schema code
    pub schema_code: String,
    /// Number of tables found
    pub table_count: usize,
    /// Number of indexes found
    pub index_count: usize,
    /// Number of views found
    pub view_count: usize,
    /// Any warnings during introspection
    pub warnings: Vec<String>,
    /// Live MySQL defaults used only while comparing an introspected schema
    /// with a desired schema during push planning.
    pub mysql_catalog_defaults: Option<drizzle_migrations::mysql::MySQLCatalogDefaults>,
    /// The schema snapshot for migration tracking
    pub snapshot: Snapshot,
    /// Path to the generated snapshot file
    pub snapshot_path: std::path::PathBuf,
}

/// Introspects a database and writes `<out_dir>/schema.rs` and a baseline
/// `<out_dir>/<tag>/` folder (`snapshot.json` and a `migration.sql` that
/// creates the whole schema).
///
/// The tracking table is excluded and `filters` applied before anything is
/// written. With `init_metadata`, the baseline is then recorded as applied.
///
/// # Errors
///
/// Returns [`CliError`] if connecting or querying the catalogs fails,
/// applying the filters fails, `out_dir` uses the legacy journal layout, a
/// file cannot be written, or `init_metadata` is refused because migrations
/// are already recorded or the folder holds more than one migration.
#[allow(clippy::too_many_arguments)]
pub fn run_introspection(
    connection: &ResolvedConnection,
    out_dir: &Path,
    init_metadata: bool,
    breakpoints: bool,
    introspect_casing: Option<IntrospectCasing>,
    filters: &SnapshotFilters,
    migrations_table: &str,
    migrations_schema: &str,
) -> Result<IntrospectResult, CliError> {
    use drizzle_migrations::naming::generate_migration_tag;

    // Perform introspection
    let mut result = introspect_database(connection)?;
    exclude_tracking_table(&mut result.snapshot, connection.dialect, migrations_table)?;
    apply_snapshot_filters(&mut result.snapshot, connection.dialect, filters)?;
    if !filters.is_empty() || introspect_casing.is_some() {
        regenerate_schema_from_snapshot(&mut result, connection.dialect, introspect_casing);
    }

    // Write schema file
    let schema_path = out_dir.join("schema.rs");
    if let Some(parent) = schema_path.parent() {
        std::fs::create_dir_all(parent).map_err(|e| {
            CliError::Other(format!(
                "Failed to create output directory '{}': {}",
                parent.display(),
                e
            ))
        })?;
    }
    std::fs::write(&schema_path, &result.schema_code).map_err(|e| {
        CliError::Other(format!(
            "Failed to write schema file '{}': {}",
            schema_path.display(),
            e
        ))
    })?;

    let journal_path = out_dir.join("meta").join("_journal.json");
    if journal_path.exists() {
        return Err(CliError::Other(
            "Detected old drizzle-kit migration folders. Upgrade them before writing new migrations."
                .to_string(),
        ));
    }

    // Generate migration tag (V3 format: timestamp-based)
    let tag = generate_migration_tag(None);

    // Create migration directory: {out}/{tag}/
    let migration_dir = out_dir.join(&tag);
    std::fs::create_dir_all(&migration_dir).map_err(|e| {
        CliError::Other(format!(
            "Failed to create migration directory '{}': {}",
            migration_dir.display(),
            e
        ))
    })?;

    // Save snapshot JSON: {out}/{tag}/snapshot.json
    let snapshot_path = migration_dir.join("snapshot.json");
    result.snapshot.save(&snapshot_path).map_err(|e| {
        CliError::Other(format!(
            "Failed to write snapshot file '{}': {}",
            snapshot_path.display(),
            e
        ))
    })?;

    // Generate initial migration SQL by diffing against empty snapshot
    let base_dialect = connection.dialect.to_base();
    let empty_snapshot = Snapshot::empty(base_dialect);
    let sql_statements =
        generate_introspect_migration(&empty_snapshot, &result.snapshot, breakpoints)?;

    // Write migration.sql: {out}/{tag}/migration.sql
    let migration_sql_path = migration_dir.join("migration.sql");
    let sql_content = format_migration_sql(&sql_statements, breakpoints);
    std::fs::write(&migration_sql_path, &sql_content).map_err(|e| {
        CliError::Other(format!(
            "Failed to write migration file '{}': {}",
            migration_sql_path.display(),
            e
        ))
    })?;

    // Update result with path
    result.snapshot_path = snapshot_path;

    if init_metadata {
        apply_init_metadata(connection, out_dir, migrations_table, migrations_schema)?;
    }

    Ok(result)
}

// =============================================================================
// Init metadata handling
// =============================================================================

#[allow(unused_variables)] // params consumed inside feature-gated block
fn apply_init_metadata(
    connection: &ResolvedConnection,
    out_dir: &Path,
    migrations_table: &str,
    migrations_schema: &str,
) -> Result<(), CliError> {
    let set = load_migration_set(
        connection.dialect,
        out_dir,
        migrations_table,
        migrations_schema,
    )?;

    match &connection.credentials {
        #[cfg(feature = "rusqlite")]
        Credentials::Sqlite { path } if connection.driver == Driver::Rusqlite => {
            init_sqlite_metadata(path, &set)
        }

        #[cfg(feature = "rusqlite")]
        Credentials::Sqlite { .. } => Err(connection_driver_mismatch(connection)),

        #[cfg(not(feature = "rusqlite"))]
        Credentials::Sqlite { .. } => Err(CliError::MissingDriver {
            dialect: "SQLite",
            feature: "rusqlite",
        }),

        #[cfg(any(feature = "libsql", feature = "turso"))]
        Credentials::Turso { url, auth_token } => {
            if connection.driver == Driver::Libsql {
                #[cfg(feature = "libsql")]
                {
                    init_libsql_local_metadata(url, &set)
                }
                #[cfg(not(feature = "libsql"))]
                {
                    Err(CliError::MissingDriver {
                        dialect: "LibSQL (local)",
                        feature: "libsql",
                    })
                }
            } else if connection.driver == Driver::Turso {
                #[cfg(feature = "turso")]
                {
                    init_turso_metadata(url, auth_token.as_deref(), &set)
                }
                #[cfg(not(feature = "turso"))]
                {
                    let _ = auth_token;
                    Err(CliError::MissingDriver {
                        dialect: "Turso (remote)",
                        feature: "turso",
                    })
                }
            } else {
                Err(connection_driver_mismatch(connection))
            }
        }

        #[cfg(all(not(feature = "turso"), not(feature = "libsql")))]
        Credentials::Turso { .. } => Err(CliError::MissingDriver {
            dialect: "Turso",
            feature: "turso or libsql",
        }),

        Credentials::Postgres(creds) => {
            let _ = creds;
            match connection.driver {
                Driver::PostgresSync => {
                    #[cfg(feature = "postgres-sync")]
                    {
                        init_postgres_sync_metadata(creds, &set)
                    }
                    #[cfg(not(feature = "postgres-sync"))]
                    {
                        Err(CliError::MissingDriver {
                            dialect: "PostgreSQL",
                            feature: "postgres-sync",
                        })
                    }
                }
                Driver::TokioPostgres => {
                    #[cfg(feature = "tokio-postgres")]
                    {
                        init_postgres_async_metadata(creds, &set)
                    }
                    #[cfg(not(feature = "tokio-postgres"))]
                    {
                        Err(CliError::MissingDriver {
                            dialect: "PostgreSQL",
                            feature: "tokio-postgres",
                        })
                    }
                }
                _ => Err(connection_driver_mismatch(connection)),
            }
        }

        Credentials::MySQL(creds) => mysql::init_metadata(connection.driver, creds, &set),

        #[cfg(feature = "d1-http")]
        Credentials::D1 {
            account_id,
            database_id,
            token,
        } if connection.driver == Driver::D1Http => {
            d1_http::init_metadata(&set, account_id, database_id, token)
        }

        #[cfg(feature = "d1-http")]
        Credentials::D1 { .. } => Err(connection_driver_mismatch(connection)),

        #[cfg(not(feature = "d1-http"))]
        Credentials::D1 { .. } => Err(CliError::MissingDriver {
            dialect: "Cloudflare D1 (HTTP)",
            feature: "d1-http",
        }),

        Credentials::AwsDataApi { .. } => Err(CliError::UnsupportedForDriver {
            operation: "Migration metadata init against AWS Data API",
            driver: "aws-data-api",
            hint: "AWS RDS Data API schema ops are not yet wired into this CLI. Seed the \
                   migrations table manually via `aws rds-data execute-statement` or \
                   tokio-postgres.",
        }),
    }
}

#[cfg(any(
    feature = "rusqlite",
    feature = "libsql",
    feature = "turso",
    feature = "postgres-sync",
    feature = "tokio-postgres",
    feature = "mysql-sync",
    feature = "mysql-async",
    feature = "d1-http",
))]
pub(crate) fn validate_init_metadata(
    applied_names: &[String],
    set: &Migrations,
) -> Result<(), CliError> {
    if !applied_names.is_empty() {
        return Err(CliError::Other(
            "--init can't be used when database already has migrations set".into(),
        ));
    }

    if set.all().len() > 1 {
        return Err(CliError::Other(
            "--init can't be used with existing migrations".into(),
        ));
    }

    Ok(())
}

// =============================================================================
// Init metadata implementations
// =============================================================================

#[cfg(feature = "rusqlite")]
fn init_sqlite_metadata(path: &str, set: &Migrations) -> Result<(), CliError> {
    let conn = rusqlite::Connection::open(path).map_err(|e| {
        CliError::ConnectionError(format!("Failed to open SQLite database '{path}': {e}"))
    })?;

    ensure_sqlite_tracking_table(&conn, set)?;

    let applied_names = query_applied_names_sqlite(&conn, set)?;
    validate_init_metadata(&applied_names, set)?;

    let Some(first) = set.all().first() else {
        return Ok(());
    };

    conn.execute(&set.record_migration_sql(first), [])
        .map_err(|e| CliError::MigrationError(e.to_string()))?;

    Ok(())
}

#[cfg(feature = "libsql")]
fn init_libsql_local_metadata(path: &str, set: &Migrations) -> Result<(), CliError> {
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .map_err(|e| CliError::Other(format!("Failed to create async runtime: {e}")))?;

    rt.block_on(init_libsql_local_metadata_inner(path, set))
}

#[cfg(feature = "libsql")]
async fn init_libsql_local_metadata_inner(path: &str, set: &Migrations) -> Result<(), CliError> {
    let db = libsql::Builder::new_local(path)
        .build()
        .await
        .map_err(|e| {
            CliError::ConnectionError(format!("Failed to open LibSQL database '{path}': {e}"))
        })?;

    let conn = db
        .connect()
        .map_err(|e| CliError::ConnectionError(e.to_string()))?;

    ensure_sqlite_tracking_table_libsql(&conn, set).await?;

    let applied_names = query_applied_names_libsql(&conn, set).await?;
    validate_init_metadata(&applied_names, set)?;

    let Some(first) = set.all().first() else {
        return Ok(());
    };

    conn.execute(&set.record_migration_sql(first), ())
        .await
        .map_err(|e| CliError::MigrationError(e.to_string()))?;

    Ok(())
}

#[cfg(feature = "turso")]
fn init_turso_metadata(
    url: &str,
    auth_token: Option<&str>,
    set: &Migrations,
) -> Result<(), CliError> {
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .map_err(|e| CliError::Other(format!("Failed to create async runtime: {e}")))?;

    rt.block_on(init_turso_metadata_inner(url, auth_token, set))
}

#[cfg(feature = "turso")]
async fn init_turso_metadata_inner(
    url: &str,
    auth_token: Option<&str>,
    set: &Migrations,
) -> Result<(), CliError> {
    let builder =
        libsql::Builder::new_remote(url.to_string(), auth_token.unwrap_or("").to_string());

    let db = builder.build().await.map_err(|e| {
        CliError::ConnectionError(format!("Failed to connect to Turso '{url}': {e}"))
    })?;

    let conn = db
        .connect()
        .map_err(|e| CliError::ConnectionError(e.to_string()))?;

    ensure_sqlite_tracking_table_libsql(&conn, set).await?;

    let applied_names = query_applied_names_libsql(&conn, set).await?;
    validate_init_metadata(&applied_names, set)?;

    let Some(first) = set.all().first() else {
        return Ok(());
    };

    conn.execute(&set.record_migration_sql(first), ())
        .await
        .map_err(|e| CliError::MigrationError(e.to_string()))?;

    Ok(())
}

#[cfg(feature = "postgres-sync")]
fn init_postgres_sync_metadata(creds: &PostgresCreds, set: &Migrations) -> Result<(), CliError> {
    let mut client = connect_postgres_sync(creds)?;

    if let Some(schema_sql) = set.create_schema_sql() {
        client
            .execute(&schema_sql, &[])
            .map_err(|e| CliError::MigrationError(e.to_string()))?;
    }

    ensure_postgres_tracking_table_sync(&mut client, set)?;

    let rows = client
        .query(&set.applied_names_sql(), &[])
        .map_err(|error| CliError::MigrationError(error.to_string()))?;
    let applied_names = rows
        .iter()
        .map(|row| row.try_get(0))
        .collect::<Result<Vec<String>, _>>()
        .map_err(|error| CliError::MigrationError(error.to_string()))?;

    validate_init_metadata(&applied_names, set)?;

    let Some(first) = set.all().first() else {
        return Ok(());
    };

    client
        .execute(&set.record_migration_sql(first), &[])
        .map_err(|e| CliError::MigrationError(e.to_string()))?;

    Ok(())
}

#[cfg(feature = "tokio-postgres")]
fn init_postgres_async_metadata(creds: &PostgresCreds, set: &Migrations) -> Result<(), CliError> {
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .map_err(|e| CliError::Other(format!("Failed to create async runtime: {}", e)))?;

    rt.block_on(init_postgres_async_inner(creds, set))
}

#[cfg(feature = "tokio-postgres")]
async fn init_postgres_async_inner(
    creds: &PostgresCreds,
    set: &Migrations,
) -> Result<(), CliError> {
    let client = connect_postgres_async(creds).await?;

    if let Some(schema_sql) = set.create_schema_sql() {
        client
            .execute(&schema_sql, &[])
            .await
            .map_err(|e| CliError::MigrationError(e.to_string()))?;
    }

    ensure_postgres_tracking_table_async(&client, set).await?;

    let rows = client
        .query(&set.applied_names_sql(), &[])
        .await
        .map_err(|error| CliError::MigrationError(error.to_string()))?;
    let applied_names = rows
        .iter()
        .map(|row| row.try_get(0))
        .collect::<Result<Vec<String>, _>>()
        .map_err(|error| CliError::MigrationError(error.to_string()))?;

    validate_init_metadata(&applied_names, set)?;

    let Some(first) = set.all().first() else {
        return Ok(());
    };

    client
        .execute(&set.record_migration_sql(first), &[])
        .await
        .map_err(|e| CliError::MigrationError(e.to_string()))?;

    Ok(())
}

/// Generate migration SQL from snapshot diff (for introspection)
fn generate_introspect_migration(
    prev: &Snapshot,
    current: &Snapshot,
    _breakpoints: bool,
) -> Result<Vec<String>, CliError> {
    drizzle_migrations::diff(prev, current)
        .map(|plan| plan.statements)
        .map_err(|error| CliError::MigrationError(error.to_string()))
}

fn format_migration_sql(sql_statements: &[String], breakpoints: bool) -> String {
    if sql_statements.is_empty() {
        "-- No tables to create (empty database)\n".to_string()
    } else if breakpoints {
        sql_statements.join("\n--> statement-breakpoint\n")
    } else {
        sql_statements.join("\n\n")
    }
}

fn regenerate_schema_from_snapshot(
    result: &mut IntrospectResult,
    dialect: Dialect,
    introspect_casing: Option<IntrospectCasing>,
) {
    if result.snapshot.dialect() != dialect.to_base() {
        return;
    }
    let generated = crate::codegen::schema_code(
        &result.snapshot,
        &crate::codegen::SchemaCodeOptions {
            casing: introspect_casing,
            module_doc: "Schema introspected from filtered database objects",
            schema_name: "Schema",
        },
    )
    .expect("initial MySQL introspection already validated lossless code generation");

    result.schema_code = generated.code;
    result.table_count = generated.table_count;
    result.index_count = generated.index_count;
    result.view_count = generated.view_count;
    result.warnings = generated.warnings;
}

/// Introspect a database and generate schema code
fn introspect_database(connection: &ResolvedConnection) -> Result<IntrospectResult, CliError> {
    match connection.dialect {
        Dialect::Sqlite | Dialect::Turso => introspect_sqlite_dialect(connection),
        Dialect::Postgresql => introspect_postgres_dialect(connection),
        Dialect::Mysql => match &connection.credentials {
            Credentials::MySQL(creds) => mysql::introspect(connection.driver, creds),
            _ => Err(connection_driver_mismatch(connection)),
        },
    }
}

/// Introspect SQLite-family databases
fn introspect_sqlite_dialect(
    connection: &ResolvedConnection,
) -> Result<IntrospectResult, CliError> {
    match &connection.credentials {
        #[cfg(feature = "rusqlite")]
        Credentials::Sqlite { path } if connection.driver == Driver::Rusqlite => {
            introspect_rusqlite(path)
        }

        #[cfg(feature = "rusqlite")]
        Credentials::Sqlite { .. } => Err(connection_driver_mismatch(connection)),

        #[cfg(not(feature = "rusqlite"))]
        Credentials::Sqlite { .. } => Err(CliError::MissingDriver {
            dialect: "SQLite",
            feature: "rusqlite",
        }),

        #[cfg(any(feature = "libsql", feature = "turso"))]
        Credentials::Turso { url, auth_token } => {
            if connection.driver == Driver::Libsql {
                #[cfg(feature = "libsql")]
                {
                    introspect_libsql_local(url)
                }
                #[cfg(not(feature = "libsql"))]
                {
                    Err(CliError::MissingDriver {
                        dialect: "LibSQL (local)",
                        feature: "libsql",
                    })
                }
            } else if connection.driver == Driver::Turso {
                #[cfg(feature = "turso")]
                {
                    introspect_turso(url, auth_token.as_deref())
                }
                #[cfg(not(feature = "turso"))]
                {
                    let _ = auth_token;
                    Err(CliError::MissingDriver {
                        dialect: "Turso (remote)",
                        feature: "turso",
                    })
                }
            } else {
                Err(connection_driver_mismatch(connection))
            }
        }

        #[cfg(all(not(feature = "turso"), not(feature = "libsql")))]
        Credentials::Turso { .. } => Err(CliError::MissingDriver {
            dialect: "Turso",
            feature: "turso or libsql",
        }),

        _ => Err(CliError::Other(
            "SQLite introspection requires sqlite path or turso credentials".into(),
        )),
    }
}

/// Introspect `PostgreSQL` databases
fn introspect_postgres_dialect(
    connection: &ResolvedConnection,
) -> Result<IntrospectResult, CliError> {
    match &connection.credentials {
        Credentials::Postgres(creds) => {
            let _ = creds;
            match connection.driver {
                Driver::PostgresSync => {
                    #[cfg(feature = "postgres-sync")]
                    {
                        introspect_postgres_sync(creds)
                    }
                    #[cfg(not(feature = "postgres-sync"))]
                    {
                        Err(CliError::MissingDriver {
                            dialect: "PostgreSQL",
                            feature: "postgres-sync",
                        })
                    }
                }
                Driver::TokioPostgres => {
                    #[cfg(feature = "tokio-postgres")]
                    {
                        introspect_postgres_async(creds)
                    }
                    #[cfg(not(feature = "tokio-postgres"))]
                    {
                        Err(CliError::MissingDriver {
                            dialect: "PostgreSQL",
                            feature: "tokio-postgres",
                        })
                    }
                }
                _ => Err(connection_driver_mismatch(connection)),
            }
        }

        _ => Err(CliError::Other(
            "PostgreSQL introspection requires postgres credentials".into(),
        )),
    }
}

// ============================================================================
// SQLite Introspection (rusqlite)
// ============================================================================

#[cfg(any(feature = "rusqlite", feature = "libsql", feature = "turso"))]
struct SqliteRawData {
    tables: Vec<(String, Option<String>)>,
    raw_columns: Vec<drizzle_migrations::sqlite::introspect::RawColumnInfo>,
    all_indexes: Vec<drizzle_migrations::sqlite::introspect::RawIndexInfo>,
    all_index_columns: Vec<drizzle_migrations::sqlite::introspect::RawIndexColumn>,
    all_fks: Vec<drizzle_migrations::sqlite::introspect::RawForeignKey>,
    all_views: Vec<drizzle_migrations::sqlite::introspect::RawViewInfo>,
    index_sql: Vec<(String, String)>,
}

#[cfg(feature = "rusqlite")]
impl SqliteRawData {
    const fn empty() -> Self {
        Self {
            tables: Vec::new(),
            raw_columns: Vec::new(),
            all_indexes: Vec::new(),
            all_index_columns: Vec::new(),
            all_fks: Vec::new(),
            all_views: Vec::new(),
            index_sql: Vec::new(),
        }
    }
}

#[cfg(feature = "rusqlite")]
fn query_rusqlite_tables_and_columns(
    conn: &rusqlite::Connection,
    raw: &mut SqliteRawData,
) -> Result<(), CliError> {
    use drizzle_migrations::sqlite::introspect::{RawColumnInfo, queries};

    let mut tables_stmt = conn
        .prepare(queries::TABLES_QUERY)
        .map_err(|e| CliError::Other(format!("Failed to prepare tables query: {e}")))?;

    raw.tables = tables_stmt
        .query_map([], |row| Ok((row.get(0)?, row.get(1)?)))
        .map_err(|e| CliError::Other(e.to_string()))?
        .collect::<Result<Vec<_>, _>>()
        .map_err(|e| CliError::Other(e.to_string()))?;

    let mut columns_stmt = conn
        .prepare(queries::COLUMNS_QUERY)
        .map_err(|e| CliError::Other(format!("Failed to prepare columns query: {e}")))?;

    raw.raw_columns = columns_stmt
        .query_map([], |row| {
            Ok(RawColumnInfo {
                table: row.get(0)?,
                cid: row.get(1)?,
                name: row.get(2)?,
                column_type: row.get(3)?,
                not_null: row.get(4)?,
                default_value: row.get(5)?,
                pk: row.get(6)?,
                hidden: row.get(7)?,
                sql: row.get(8)?,
            })
        })
        .map_err(|e| CliError::Other(e.to_string()))?
        .collect::<Result<Vec<_>, _>>()
        .map_err(|e| CliError::Other(e.to_string()))?;

    Ok(())
}

#[cfg(feature = "rusqlite")]
fn query_rusqlite_indexes_and_foreign_keys(
    conn: &rusqlite::Connection,
    raw: &mut SqliteRawData,
) -> Result<(), CliError> {
    use drizzle_migrations::sqlite::introspect::{
        RawForeignKey, RawIndexColumn, RawIndexInfo, queries,
    };

    let mut indexes_stmt = conn
        .prepare(queries::INDEXES_QUERY)
        .map_err(|e| CliError::Other(e.to_string()))?;
    raw.all_indexes = indexes_stmt
        .query_map([], |row| {
            Ok(RawIndexInfo {
                table: row.get(0)?,
                name: row.get(1)?,
                unique: row.get::<_, i32>(2)? != 0,
                origin: row.get(3)?,
                partial: row.get::<_, i32>(4)? != 0,
            })
        })
        .map_err(|e| CliError::Other(e.to_string()))?
        .collect::<Result<Vec<_>, _>>()
        .map_err(|e| CliError::Other(e.to_string()))?;

    let mut columns_stmt = conn
        .prepare(queries::INDEX_COLUMNS_QUERY)
        .map_err(|e| CliError::Other(e.to_string()))?;
    raw.all_index_columns = columns_stmt
        .query_map([], |row| {
            Ok(RawIndexColumn {
                index_name: row.get(0)?,
                seqno: row.get(1)?,
                cid: row.get(2)?,
                name: row.get(3)?,
                desc: row.get::<_, i32>(4)? != 0,
                coll: row.get(5)?,
                key: row.get::<_, i32>(6)? != 0,
            })
        })
        .map_err(|e| CliError::Other(e.to_string()))?
        .collect::<Result<Vec<_>, _>>()
        .map_err(|e| CliError::Other(e.to_string()))?;

    let mut foreign_keys_stmt = conn
        .prepare(queries::FOREIGN_KEYS_QUERY)
        .map_err(|e| CliError::Other(e.to_string()))?;
    raw.all_fks = foreign_keys_stmt
        .query_map([], |row| {
            Ok(RawForeignKey {
                table: row.get(0)?,
                id: row.get(1)?,
                seq: row.get(2)?,
                to_table: row.get(3)?,
                from_column: row.get(4)?,
                to_column: row.get(5)?,
                on_update: row.get(6)?,
                on_delete: row.get(7)?,
                r#match: row.get(8)?,
            })
        })
        .map_err(|e| CliError::Other(e.to_string()))?
        .collect::<Result<Vec<_>, _>>()
        .map_err(|e| CliError::Other(e.to_string()))?;

    Ok(())
}

#[cfg(feature = "rusqlite")]
fn query_rusqlite_views_and_view_columns(
    conn: &rusqlite::Connection,
    raw: &mut SqliteRawData,
) -> Result<(), CliError> {
    use drizzle_migrations::sqlite::introspect::{RawColumnInfo, RawViewInfo, queries};

    let mut views_stmt = conn
        .prepare(queries::VIEWS_QUERY)
        .map_err(|e| CliError::Other(e.to_string()))?;
    let view_iter = views_stmt
        .query_map([], |row| {
            Ok(RawViewInfo {
                name: row.get(0)?,
                sql: row.get(1)?,
            })
        })
        .map_err(|e| CliError::Other(e.to_string()))?;
    raw.all_views.extend(
        view_iter
            .collect::<Result<Vec<_>, _>>()
            .map_err(|e| CliError::Other(e.to_string()))?,
    );

    let mut view_cols_stmt = conn
        .prepare(queries::VIEW_COLUMNS_QUERY)
        .map_err(|e| CliError::Other(e.to_string()))?;
    let col_iter = view_cols_stmt
        .query_map([], |row| {
            Ok(RawColumnInfo {
                table: row.get(0)?,
                cid: row.get(1)?,
                name: row.get(2)?,
                column_type: row.get(3)?,
                not_null: row.get::<_, i32>(4)? != 0,
                default_value: row.get(5)?,
                pk: row.get(6)?,
                hidden: row.get(7)?,
                sql: row.get(8)?,
            })
        })
        .map_err(|e| CliError::Other(e.to_string()))?;
    raw.raw_columns.extend(
        col_iter
            .collect::<Result<Vec<_>, _>>()
            .map_err(|e| CliError::Other(e.to_string()))?,
    );
    Ok(())
}

#[cfg(feature = "rusqlite")]
fn query_rusqlite_raw(conn: &rusqlite::Connection) -> Result<SqliteRawData, CliError> {
    use drizzle_migrations::sqlite::introspect::queries;

    let mut raw = SqliteRawData::empty();
    query_rusqlite_tables_and_columns(conn, &mut raw)?;
    query_rusqlite_indexes_and_foreign_keys(conn, &mut raw)?;
    query_rusqlite_views_and_view_columns(conn, &mut raw)?;

    let mut index_sql_stmt = conn
        .prepare(queries::INDEX_SQL_QUERY)
        .map_err(|e| CliError::Other(format!("Failed to prepare index sql query: {e}")))?;
    raw.index_sql = index_sql_stmt
        .query_map([], |row| Ok((row.get(0)?, row.get(1)?)))
        .map_err(|e| CliError::Other(e.to_string()))?
        .collect::<Result<Vec<_>, _>>()
        .map_err(|e| CliError::Other(e.to_string()))?;

    Ok(raw)
}

#[cfg(any(feature = "rusqlite", feature = "libsql", feature = "turso"))]
fn build_sqlite_ddl(raw: SqliteRawData) -> drizzle_migrations::sqlite::SQLiteDDL {
    drizzle_migrations::sqlite::introspect::assemble_ddl(
        drizzle_migrations::sqlite::introspect::RawIntrospection {
            tables: raw.tables,
            columns: raw.raw_columns,
            indexes: raw.all_indexes,
            index_columns: raw.all_index_columns,
            foreign_keys: raw.all_fks,
            views: raw.all_views,
            index_sql: raw.index_sql,
        },
    )
}

#[cfg(feature = "rusqlite")]
fn introspect_rusqlite(path: &str) -> Result<IntrospectResult, CliError> {
    use drizzle_migrations::sqlite::codegen::{CodegenOptions, FieldCasing, generate_rust_schema};

    let conn = rusqlite::Connection::open(path).map_err(|e| {
        CliError::ConnectionError(format!("Failed to open SQLite database '{path}': {e}"))
    })?;

    let raw = query_rusqlite_raw(&conn)?;
    let ddl = build_sqlite_ddl(raw);

    let options = CodegenOptions {
        module_doc: Some(format!("Schema introspected from {path}")),
        include_schema: true,
        schema_name: "Schema".to_string(),
        use_pub: true,
        field_casing: FieldCasing::default(),
    };

    let generated = generate_rust_schema(&ddl, &options);

    let mut sqlite_snapshot = drizzle_migrations::sqlite::SQLiteSnapshot::new();
    for entity in ddl.to_entities() {
        sqlite_snapshot.add_entity(entity);
    }
    let snapshot = Snapshot::Sqlite(sqlite_snapshot);

    Ok(IntrospectResult {
        schema_code: generated.code,
        table_count: generated.tables.len(),
        index_count: generated.indexes.len(),
        view_count: ddl.views.len(),
        warnings: generated.warnings,
        mysql_catalog_defaults: None,
        snapshot,
        snapshot_path: std::path::PathBuf::new(),
    })
}

// ============================================================================
// LibSQL Introspection (local)
// ============================================================================

#[cfg(feature = "libsql")]
fn introspect_libsql_local(path: &str) -> Result<IntrospectResult, CliError> {
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .map_err(|e| CliError::Other(format!("Failed to create async runtime: {e}")))?;

    rt.block_on(introspect_libsql_inner(path, None))
}

#[cfg(any(feature = "libsql", feature = "turso"))]
async fn query_libsql_tables_and_columns(
    conn: &libsql::Connection,
) -> Result<
    (
        Vec<(String, Option<String>)>,
        Vec<drizzle_migrations::sqlite::introspect::RawColumnInfo>,
    ),
    CliError,
> {
    use drizzle_migrations::sqlite::introspect::{RawColumnInfo, queries};

    let mut tables_rows = conn
        .query(queries::TABLES_QUERY, ())
        .await
        .map_err(|e| CliError::Other(format!("Failed to query tables: {e}")))?;

    let mut tables: Vec<(String, Option<String>)> = Vec::new();
    while let Some(row) = tables_rows
        .next()
        .await
        .map_err(|e| CliError::Other(e.to_string()))?
    {
        let name: String = row.get(0).map_err(|e| CliError::Other(e.to_string()))?;
        let sql: Option<String> = row.get(1).ok();
        tables.push((name, sql));
    }

    let mut columns_rows = conn
        .query(queries::COLUMNS_QUERY, ())
        .await
        .map_err(|e| CliError::Other(format!("Failed to query columns: {e}")))?;

    let mut raw_columns: Vec<RawColumnInfo> = Vec::new();
    while let Some(row) = columns_rows
        .next()
        .await
        .map_err(|e| CliError::Other(e.to_string()))?
    {
        raw_columns.push(RawColumnInfo {
            table: row.get(0).map_err(|e| CliError::Other(e.to_string()))?,
            cid: row.get(1).map_err(|e| CliError::Other(e.to_string()))?,
            name: row.get(2).map_err(|e| CliError::Other(e.to_string()))?,
            column_type: row.get(3).map_err(|e| CliError::Other(e.to_string()))?,
            not_null: row
                .get::<i32>(4)
                .map_err(|e| CliError::Other(e.to_string()))?
                != 0,
            default_value: row.get(5).ok(),
            pk: row.get(6).map_err(|e| CliError::Other(e.to_string()))?,
            hidden: row.get(7).map_err(|e| CliError::Other(e.to_string()))?,
            sql: row.get(8).ok(),
        });
    }

    Ok((tables, raw_columns))
}

#[cfg(any(feature = "libsql", feature = "turso"))]
async fn query_libsql_indexes_and_foreign_keys(
    conn: &libsql::Connection,
) -> Result<
    (
        Vec<drizzle_migrations::sqlite::introspect::RawIndexInfo>,
        Vec<drizzle_migrations::sqlite::introspect::RawIndexColumn>,
        Vec<drizzle_migrations::sqlite::introspect::RawForeignKey>,
    ),
    CliError,
> {
    use drizzle_migrations::sqlite::introspect::{
        RawForeignKey, RawIndexColumn, RawIndexInfo, queries,
    };

    let mut all_indexes = Vec::<RawIndexInfo>::new();
    let mut index_rows = conn
        .query(queries::INDEXES_QUERY, ())
        .await
        .map_err(|e| CliError::Other(e.to_string()))?;
    while let Some(row) = index_rows
        .next()
        .await
        .map_err(|e| CliError::Other(e.to_string()))?
    {
        all_indexes.push(RawIndexInfo {
            table: row.get(0).map_err(|e| CliError::Other(e.to_string()))?,
            name: row.get(1).map_err(|e| CliError::Other(e.to_string()))?,
            unique: row
                .get::<i32>(2)
                .map_err(|e| CliError::Other(e.to_string()))?
                != 0,
            origin: row.get(3).map_err(|e| CliError::Other(e.to_string()))?,
            partial: row
                .get::<i32>(4)
                .map_err(|e| CliError::Other(e.to_string()))?
                != 0,
        });
    }

    let mut all_index_columns = Vec::<RawIndexColumn>::new();
    let mut column_rows = conn
        .query(queries::INDEX_COLUMNS_QUERY, ())
        .await
        .map_err(|e| CliError::Other(e.to_string()))?;
    while let Some(row) = column_rows
        .next()
        .await
        .map_err(|e| CliError::Other(e.to_string()))?
    {
        all_index_columns.push(RawIndexColumn {
            index_name: row.get(0).map_err(|e| CliError::Other(e.to_string()))?,
            seqno: row.get(1).map_err(|e| CliError::Other(e.to_string()))?,
            cid: row.get(2).map_err(|e| CliError::Other(e.to_string()))?,
            name: row.get(3).ok(),
            desc: row
                .get::<i32>(4)
                .map_err(|e| CliError::Other(e.to_string()))?
                != 0,
            coll: row.get(5).map_err(|e| CliError::Other(e.to_string()))?,
            key: row
                .get::<i32>(6)
                .map_err(|e| CliError::Other(e.to_string()))?
                != 0,
        });
    }

    let mut all_fks = Vec::<RawForeignKey>::new();
    let mut foreign_key_rows = conn
        .query(queries::FOREIGN_KEYS_QUERY, ())
        .await
        .map_err(|e| CliError::Other(e.to_string()))?;
    while let Some(row) = foreign_key_rows
        .next()
        .await
        .map_err(|e| CliError::Other(e.to_string()))?
    {
        all_fks.push(RawForeignKey {
            table: row.get(0).map_err(|e| CliError::Other(e.to_string()))?,
            id: row.get(1).map_err(|e| CliError::Other(e.to_string()))?,
            seq: row.get(2).map_err(|e| CliError::Other(e.to_string()))?,
            to_table: row.get(3).map_err(|e| CliError::Other(e.to_string()))?,
            from_column: row.get(4).map_err(|e| CliError::Other(e.to_string()))?,
            to_column: row.get(5).map_err(|e| CliError::Other(e.to_string()))?,
            on_update: row.get(6).map_err(|e| CliError::Other(e.to_string()))?,
            on_delete: row.get(7).map_err(|e| CliError::Other(e.to_string()))?,
            r#match: row.get(8).map_err(|e| CliError::Other(e.to_string()))?,
        });
    }

    Ok((all_indexes, all_index_columns, all_fks))
}

#[cfg(any(feature = "libsql", feature = "turso"))]
async fn query_libsql_views_and_view_columns(
    conn: &libsql::Connection,
    raw_columns: &mut Vec<drizzle_migrations::sqlite::introspect::RawColumnInfo>,
) -> Result<Vec<drizzle_migrations::sqlite::introspect::RawViewInfo>, CliError> {
    use drizzle_migrations::sqlite::introspect::{RawColumnInfo, RawViewInfo, queries};

    let mut all_views: Vec<RawViewInfo> = Vec::new();
    let mut views_rows = conn
        .query(queries::VIEWS_QUERY, ())
        .await
        .map_err(|e| CliError::Other(e.to_string()))?;
    while let Some(row) = views_rows
        .next()
        .await
        .map_err(|e| CliError::Other(e.to_string()))?
    {
        let name: String = row.get(0).map_err(|e| CliError::Other(e.to_string()))?;
        let sql: String = row.get(1).map_err(|e| CliError::Other(e.to_string()))?;
        all_views.push(RawViewInfo { name, sql });
    }

    let mut view_cols_rows = conn
        .query(queries::VIEW_COLUMNS_QUERY, ())
        .await
        .map_err(|e| CliError::Other(e.to_string()))?;
    while let Some(row) = view_cols_rows
        .next()
        .await
        .map_err(|e| CliError::Other(e.to_string()))?
    {
        raw_columns.push(RawColumnInfo {
            table: row.get(0).map_err(|e| CliError::Other(e.to_string()))?,
            cid: row.get(1).map_err(|e| CliError::Other(e.to_string()))?,
            name: row.get(2).map_err(|e| CliError::Other(e.to_string()))?,
            column_type: row.get(3).map_err(|e| CliError::Other(e.to_string()))?,
            not_null: row
                .get::<i32>(4)
                .map_err(|e| CliError::Other(e.to_string()))?
                != 0,
            default_value: row.get(5).ok(),
            pk: row.get(6).map_err(|e| CliError::Other(e.to_string()))?,
            hidden: row.get(7).map_err(|e| CliError::Other(e.to_string()))?,
            sql: row.get(8).ok(),
        });
    }
    Ok(all_views)
}

#[cfg(any(feature = "libsql", feature = "turso"))]
async fn query_libsql_raw(conn: &libsql::Connection) -> Result<SqliteRawData, CliError> {
    use drizzle_migrations::sqlite::introspect::queries;

    let (tables, mut raw_columns) = query_libsql_tables_and_columns(conn).await?;
    let (all_indexes, all_index_columns, all_fks) =
        query_libsql_indexes_and_foreign_keys(conn).await?;
    let all_views = query_libsql_views_and_view_columns(conn, &mut raw_columns).await?;

    let mut index_sql = Vec::new();
    let mut rows = conn
        .query(queries::INDEX_SQL_QUERY, ())
        .await
        .map_err(|e| CliError::Other(e.to_string()))?;
    while let Some(row) = rows
        .next()
        .await
        .map_err(|e| CliError::Other(e.to_string()))?
    {
        let name: String = row.get(0).map_err(|e| CliError::Other(e.to_string()))?;
        let sql: String = row.get(1).map_err(|e| CliError::Other(e.to_string()))?;
        index_sql.push((name, sql));
    }

    Ok(SqliteRawData {
        tables,
        raw_columns,
        all_indexes,
        all_index_columns,
        all_fks,
        all_views,
        index_sql,
    })
}

#[cfg(feature = "libsql")]
async fn introspect_libsql_inner(
    path: &str,
    _auth_token: Option<&str>,
) -> Result<IntrospectResult, CliError> {
    use drizzle_migrations::sqlite::codegen::{CodegenOptions, FieldCasing, generate_rust_schema};

    let db = libsql::Builder::new_local(path)
        .build()
        .await
        .map_err(|e| {
            CliError::ConnectionError(format!("Failed to open LibSQL database '{path}': {e}"))
        })?;

    let conn = db
        .connect()
        .map_err(|e| CliError::ConnectionError(e.to_string()))?;

    let raw = query_libsql_raw(&conn).await?;
    let ddl = build_sqlite_ddl(raw);

    let options = CodegenOptions {
        module_doc: Some(format!("Schema introspected from {path}")),
        include_schema: true,
        schema_name: "Schema".to_string(),
        use_pub: true,
        field_casing: FieldCasing::default(),
    };

    let generated = generate_rust_schema(&ddl, &options);

    let mut sqlite_snapshot = drizzle_migrations::sqlite::SQLiteSnapshot::new();
    for entity in ddl.to_entities() {
        sqlite_snapshot.add_entity(entity);
    }
    let snapshot = Snapshot::Sqlite(sqlite_snapshot);

    Ok(IntrospectResult {
        schema_code: generated.code,
        table_count: generated.tables.len(),
        index_count: generated.indexes.len(),
        view_count: ddl.views.len(),
        warnings: generated.warnings,
        mysql_catalog_defaults: None,
        snapshot,
        snapshot_path: std::path::PathBuf::new(),
    })
}

// ============================================================================
// Turso Introspection (remote)
// ============================================================================

#[cfg(feature = "turso")]
fn introspect_turso(url: &str, auth_token: Option<&str>) -> Result<IntrospectResult, CliError> {
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .map_err(|e| CliError::Other(format!("Failed to create async runtime: {e}")))?;

    rt.block_on(introspect_turso_inner(url, auth_token))
}

#[cfg(feature = "turso")]
async fn introspect_turso_inner(
    url: &str,
    auth_token: Option<&str>,
) -> Result<IntrospectResult, CliError> {
    use drizzle_migrations::sqlite::codegen::{CodegenOptions, FieldCasing, generate_rust_schema};

    let builder =
        libsql::Builder::new_remote(url.to_string(), auth_token.unwrap_or("").to_string());

    let db = builder.build().await.map_err(|e| {
        CliError::ConnectionError(format!("Failed to connect to Turso '{url}': {e}"))
    })?;

    let conn = db
        .connect()
        .map_err(|e| CliError::ConnectionError(e.to_string()))?;

    let raw = query_libsql_raw(&conn).await?;
    let ddl = build_sqlite_ddl(raw);

    let options = CodegenOptions {
        module_doc: Some(format!("Schema introspected from Turso: {url}")),
        include_schema: true,
        schema_name: "Schema".to_string(),
        use_pub: true,
        field_casing: FieldCasing::default(),
    };

    let generated = generate_rust_schema(&ddl, &options);

    let mut sqlite_snapshot = drizzle_migrations::sqlite::SQLiteSnapshot::new();
    for entity in ddl.to_entities() {
        sqlite_snapshot.add_entity(entity);
    }
    let snapshot = Snapshot::Sqlite(sqlite_snapshot);

    Ok(IntrospectResult {
        schema_code: generated.code,
        table_count: generated.tables.len(),
        index_count: generated.indexes.len(),
        view_count: ddl.views.len(),
        warnings: generated.warnings,
        mysql_catalog_defaults: None,
        snapshot,
        snapshot_path: std::path::PathBuf::new(),
    })
}

// ============================================================================
// PostgreSQL Introspection
// ============================================================================

#[cfg(feature = "postgres-sync")]
fn introspect_postgres_sync(creds: &PostgresCreds) -> Result<IntrospectResult, CliError> {
    let mut client = connect_postgres_sync(creds)?;

    let raw = query_postgres_sync_raw(&mut client)?;
    let ddl = build_postgres_ddl(raw);

    Ok(finalize_postgres_introspection(
        &ddl,
        "configured PostgreSQL database",
    ))
}

#[cfg(any(feature = "postgres-sync", feature = "tokio-postgres"))]
impl PostgresRawData {
    const fn empty() -> Self {
        Self {
            schemas: Vec::new(),
            tables: Vec::new(),
            columns: Vec::new(),
            enums: Vec::new(),
            sequences: Vec::new(),
            views: Vec::new(),
            indexes: Vec::new(),
            foreign_keys: Vec::new(),
            primary_keys: Vec::new(),
            uniques: Vec::new(),
            checks: Vec::new(),
            roles: Vec::new(),
            policies: Vec::new(),
        }
    }
}

#[cfg(feature = "postgres-sync")]
fn query_postgres_sync_raw(client: &mut postgres::Client) -> Result<PostgresRawData, CliError> {
    let mut raw = PostgresRawData::empty();
    query_pg_sync_core(client, &mut raw)?;
    query_pg_sync_codegen_meta(client, &mut raw)?;
    query_pg_sync_constraints(client, &mut raw)?;
    query_pg_sync_security(client, &mut raw)?;
    Ok(raw)
}

#[cfg(feature = "postgres-sync")]
fn query_pg_sync_core(
    client: &mut postgres::Client,
    raw: &mut PostgresRawData,
) -> Result<(), CliError> {
    use drizzle_migrations::postgres::introspect::{RawColumnInfo, RawTableInfo, queries};

    raw.schemas = client
        .query(queries::SCHEMAS_QUERY, &[])
        .map_err(|e| CliError::Other(format!("Failed to query schemas: {e}")))?
        .into_iter()
        .map(|row| RawSchemaInfo {
            name: row.get::<_, String>(0),
        })
        .collect();

    raw.tables = client
        .query(queries::TABLES_QUERY, &[])
        .map_err(|e| CliError::Other(format!("Failed to query tables: {e}")))?
        .into_iter()
        .map(|row| RawTableInfo {
            schema: row.get::<_, String>(0),
            name: row.get::<_, String>(1),
            is_rls_enabled: row.get::<_, bool>(2),
            is_unlogged: row.get::<_, bool>(3),
            is_temporary: row.get::<_, bool>(4),
            tablespace: row.get::<_, Option<String>>(5),
            comment: row.get::<_, Option<String>>(6),
        })
        .collect();

    raw.columns = client
        .query(queries::COLUMNS_QUERY, &[])
        .map_err(|e| CliError::Other(format!("Failed to query columns: {e}")))?
        .into_iter()
        .map(|row| RawColumnInfo {
            schema: row.get::<_, String>(0),
            table: row.get::<_, String>(1),
            name: row.get::<_, String>(2),
            column_type: row.get::<_, String>(3),
            type_schema: row.get::<_, Option<String>>(4),
            not_null: row.get::<_, bool>(5),
            default_value: row.get::<_, Option<String>>(6),
            is_identity: row.get::<_, bool>(7),
            identity_type: row.get::<_, Option<String>>(8),
            is_generated: row.get::<_, bool>(9),
            generated_expression: row.get::<_, Option<String>>(10),
            generated_stored: row.get::<_, bool>(11),
            dimensions: row.get::<_, Option<i32>>(12),
            comment: row.get::<_, Option<String>>(13),
            ordinal_position: row.get::<_, i32>(14),
        })
        .collect();

    Ok(())
}

#[cfg(feature = "postgres-sync")]
fn query_pg_sync_codegen_meta(
    client: &mut postgres::Client,
    raw: &mut PostgresRawData,
) -> Result<(), CliError> {
    use drizzle_migrations::postgres::introspect::{
        RawEnumInfo, RawSequenceInfo, RawViewInfo, queries,
    };

    raw.enums = client
        .query(queries::ENUMS_QUERY, &[])
        .map_err(|e| CliError::Other(format!("Failed to query enums: {e}")))?
        .into_iter()
        .map(|row| RawEnumInfo {
            schema: row.get::<_, String>(0),
            name: row.get::<_, String>(1),
            values: row.get::<_, Vec<String>>(2),
        })
        .collect();

    raw.sequences = client
        .query(queries::SEQUENCES_QUERY, &[])
        .map_err(|e| CliError::Other(format!("Failed to query sequences: {e}")))?
        .into_iter()
        .map(|row| RawSequenceInfo {
            schema: row.get::<_, String>(0),
            name: row.get::<_, String>(1),
            data_type: row.get::<_, Option<String>>(2),
            start_value: row.get::<_, Option<String>>(3),
            min_value: row.get::<_, Option<String>>(4),
            max_value: row.get::<_, Option<String>>(5),
            increment: row.get::<_, Option<String>>(6),
            cycle: row.get::<_, Option<bool>>(7),
            cache_value: row.get::<_, Option<String>>(8),
            owned_by: row.get::<_, Option<String>>(9),
        })
        .collect();

    let view_schema_filters: Option<Vec<String>> = None;
    raw.views = client
        .query(queries::VIEWS_QUERY, &[&view_schema_filters])
        .map_err(|e| CliError::Other(format!("Failed to query views: {e}")))?
        .into_iter()
        .map(|row| RawViewInfo {
            schema: row.get::<_, String>(0),
            name: row.get::<_, String>(1),
            definition: row.get::<_, String>(2),
            is_materialized: row.get::<_, bool>(3),
        })
        .collect();

    Ok(())
}

#[cfg(feature = "postgres-sync")]
fn query_pg_sync_constraints(
    client: &mut postgres::Client,
    raw: &mut PostgresRawData,
) -> Result<(), CliError> {
    use drizzle_migrations::postgres::introspect::{
        RawCheckInfo, RawForeignKeyInfo, RawIndexInfo, RawPrimaryKeyInfo, RawUniqueInfo,
        action_code_to_string, parse_index_columns, queries,
    };

    raw.indexes = client
        .query(queries::INDEXES_QUERY, &[])
        .map_err(|e| CliError::Other(format!("Failed to query indexes: {e}")))?
        .into_iter()
        .map(|row| {
            let cols: Vec<String> = row.get(6);
            RawIndexInfo {
                schema: row.get::<_, String>(0),
                table: row.get::<_, String>(1),
                name: row.get::<_, String>(2),
                is_unique: row.get::<_, bool>(3),
                is_primary: row.get::<_, bool>(4),
                method: row.get::<_, String>(5),
                columns: parse_index_columns(cols),
                where_clause: row.get::<_, Option<String>>(7),
                concurrent: false,
            }
        })
        .collect();

    raw.foreign_keys = client
        .query(queries::FOREIGN_KEYS_QUERY, &[])
        .map_err(|e| CliError::Other(format!("Failed to query foreign keys: {e}")))?
        .into_iter()
        .map(|row| RawForeignKeyInfo {
            schema: row.get::<_, String>(0),
            table: row.get::<_, String>(1),
            name: row.get::<_, String>(2),
            columns: row.get::<_, Vec<String>>(3),
            schema_to: row.get::<_, String>(4),
            table_to: row.get::<_, String>(5),
            columns_to: row.get::<_, Vec<String>>(6),
            on_update: action_code_to_string(&row.get::<_, String>(7)),
            on_delete: action_code_to_string(&row.get::<_, String>(8)),
            deferrable: row.get::<_, bool>(9),
            initially_deferred: row.get::<_, bool>(10),
        })
        .collect();

    raw.primary_keys = client
        .query(queries::PRIMARY_KEYS_QUERY, &[])
        .map_err(|e| CliError::Other(format!("Failed to query primary keys: {e}")))?
        .into_iter()
        .map(|row| RawPrimaryKeyInfo {
            schema: row.get::<_, String>(0),
            table: row.get::<_, String>(1),
            name: row.get::<_, String>(2),
            columns: row.get::<_, Vec<String>>(3),
        })
        .collect();

    raw.uniques = client
        .query(queries::UNIQUES_QUERY, &[])
        .map_err(|e| CliError::Other(format!("Failed to query unique constraints: {e}")))?
        .into_iter()
        .map(|row| RawUniqueInfo {
            schema: row.get::<_, String>(0),
            table: row.get::<_, String>(1),
            name: row.get::<_, String>(2),
            columns: row.get::<_, Vec<String>>(3),
            nulls_not_distinct: row.get::<_, bool>(4),
            deferrable: row.get::<_, bool>(5),
            initially_deferred: row.get::<_, bool>(6),
        })
        .collect();

    raw.checks = client
        .query(queries::CHECKS_QUERY, &[])
        .map_err(|e| CliError::Other(format!("Failed to query check constraints: {e}")))?
        .into_iter()
        .map(|row| RawCheckInfo {
            schema: row.get::<_, String>(0),
            table: row.get::<_, String>(1),
            name: row.get::<_, String>(2),
            expression: row.get::<_, String>(3),
        })
        .collect();

    Ok(())
}

#[cfg(feature = "postgres-sync")]
fn query_pg_sync_security(
    client: &mut postgres::Client,
    raw: &mut PostgresRawData,
) -> Result<(), CliError> {
    use drizzle_migrations::postgres::introspect::{RawPolicyInfo, RawRoleInfo, queries};

    raw.roles = client
        .query(queries::ROLES_QUERY, &[])
        .map_err(|e| CliError::Other(format!("Failed to query roles: {e}")))?
        .into_iter()
        .map(|row| RawRoleInfo {
            name: row.get::<_, String>(0),
            create_db: row.get::<_, bool>(1),
            create_role: row.get::<_, bool>(2),
            inherit: row.get::<_, bool>(3),
        })
        .collect();

    raw.policies = client
        .query(queries::POLICIES_QUERY, &[])
        .map_err(|e| CliError::Other(format!("Failed to query policies: {e}")))?
        .into_iter()
        .map(|row| RawPolicyInfo {
            schema: row.get::<_, String>(0),
            table: row.get::<_, String>(1),
            name: row.get::<_, String>(2),
            as_clause: row.get::<_, String>(3),
            for_clause: row.get::<_, String>(4),
            to: row.get::<_, Vec<String>>(5),
            using: row.get::<_, Option<String>>(6),
            with_check: row.get::<_, Option<String>>(7),
        })
        .collect();

    Ok(())
}

#[cfg(feature = "tokio-postgres")]
fn introspect_postgres_async(creds: &PostgresCreds) -> Result<IntrospectResult, CliError> {
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .map_err(|e| CliError::Other(format!("Failed to create async runtime: {e}")))?;

    rt.block_on(introspect_postgres_async_inner(creds))
}

#[cfg(feature = "tokio-postgres")]
async fn introspect_postgres_async_inner(
    creds: &PostgresCreds,
) -> Result<IntrospectResult, CliError> {
    let client = connect_postgres_async(creds).await?;

    let raw = query_postgres_async_raw(&client).await?;
    let ddl = build_postgres_ddl(raw);

    Ok(finalize_postgres_introspection(
        &ddl,
        "configured PostgreSQL database",
    ))
}

#[cfg(feature = "tokio-postgres")]
async fn query_postgres_async_raw(
    client: &tokio_postgres::Client,
) -> Result<PostgresRawData, CliError> {
    let mut raw = PostgresRawData::empty();
    query_pg_async_core(client, &mut raw).await?;
    query_pg_async_codegen_meta(client, &mut raw).await?;
    query_pg_async_constraints(client, &mut raw).await?;
    query_pg_async_security(client, &mut raw).await?;
    Ok(raw)
}

#[cfg(feature = "tokio-postgres")]
async fn query_pg_async_core(
    client: &tokio_postgres::Client,
    raw: &mut PostgresRawData,
) -> Result<(), CliError> {
    use drizzle_migrations::postgres::introspect::{RawColumnInfo, RawTableInfo, queries};

    raw.schemas = client
        .query(queries::SCHEMAS_QUERY, &[])
        .await
        .map_err(|e| CliError::Other(format!("Failed to query schemas: {e}")))?
        .into_iter()
        .map(|row| RawSchemaInfo {
            name: row.get::<_, String>(0),
        })
        .collect();

    raw.tables = client
        .query(queries::TABLES_QUERY, &[])
        .await
        .map_err(|e| CliError::Other(format!("Failed to query tables: {e}")))?
        .into_iter()
        .map(|row| RawTableInfo {
            schema: row.get::<_, String>(0),
            name: row.get::<_, String>(1),
            is_rls_enabled: row.get::<_, bool>(2),
            is_unlogged: row.get::<_, bool>(3),
            is_temporary: row.get::<_, bool>(4),
            tablespace: row.get::<_, Option<String>>(5),
            comment: row.get::<_, Option<String>>(6),
        })
        .collect();

    raw.columns = client
        .query(queries::COLUMNS_QUERY, &[])
        .await
        .map_err(|e| CliError::Other(format!("Failed to query columns: {e}")))?
        .into_iter()
        .map(|row| RawColumnInfo {
            schema: row.get::<_, String>(0),
            table: row.get::<_, String>(1),
            name: row.get::<_, String>(2),
            column_type: row.get::<_, String>(3),
            type_schema: row.get::<_, Option<String>>(4),
            not_null: row.get::<_, bool>(5),
            default_value: row.get::<_, Option<String>>(6),
            is_identity: row.get::<_, bool>(7),
            identity_type: row.get::<_, Option<String>>(8),
            is_generated: row.get::<_, bool>(9),
            generated_expression: row.get::<_, Option<String>>(10),
            generated_stored: row.get::<_, bool>(11),
            dimensions: row.get::<_, Option<i32>>(12),
            comment: row.get::<_, Option<String>>(13),
            ordinal_position: row.get::<_, i32>(14),
        })
        .collect();

    Ok(())
}

#[cfg(feature = "tokio-postgres")]
async fn query_pg_async_codegen_meta(
    client: &tokio_postgres::Client,
    raw: &mut PostgresRawData,
) -> Result<(), CliError> {
    use drizzle_migrations::postgres::introspect::{
        RawEnumInfo, RawSequenceInfo, RawViewInfo, queries,
    };

    raw.enums = client
        .query(queries::ENUMS_QUERY, &[])
        .await
        .map_err(|e| CliError::Other(format!("Failed to query enums: {e}")))?
        .into_iter()
        .map(|row| RawEnumInfo {
            schema: row.get::<_, String>(0),
            name: row.get::<_, String>(1),
            values: row.get::<_, Vec<String>>(2),
        })
        .collect();

    raw.sequences = client
        .query(queries::SEQUENCES_QUERY, &[])
        .await
        .map_err(|e| CliError::Other(format!("Failed to query sequences: {e}")))?
        .into_iter()
        .map(|row| RawSequenceInfo {
            schema: row.get::<_, String>(0),
            name: row.get::<_, String>(1),
            data_type: row.get::<_, Option<String>>(2),
            start_value: row.get::<_, Option<String>>(3),
            min_value: row.get::<_, Option<String>>(4),
            max_value: row.get::<_, Option<String>>(5),
            increment: row.get::<_, Option<String>>(6),
            cycle: row.get::<_, Option<bool>>(7),
            cache_value: row.get::<_, Option<String>>(8),
            owned_by: row.get::<_, Option<String>>(9),
        })
        .collect();

    let view_schema_filters: Option<Vec<String>> = None;
    raw.views = client
        .query(queries::VIEWS_QUERY, &[&view_schema_filters])
        .await
        .map_err(|e| CliError::Other(format!("Failed to query views: {e}")))?
        .into_iter()
        .map(|row| RawViewInfo {
            schema: row.get::<_, String>(0),
            name: row.get::<_, String>(1),
            definition: row.get::<_, String>(2),
            is_materialized: row.get::<_, bool>(3),
        })
        .collect();

    Ok(())
}

#[cfg(feature = "tokio-postgres")]
async fn query_pg_async_constraints(
    client: &tokio_postgres::Client,
    raw: &mut PostgresRawData,
) -> Result<(), CliError> {
    use drizzle_migrations::postgres::introspect::{
        RawCheckInfo, RawForeignKeyInfo, RawIndexInfo, RawPrimaryKeyInfo, RawUniqueInfo,
        action_code_to_string, parse_index_columns, queries,
    };

    raw.indexes = client
        .query(queries::INDEXES_QUERY, &[])
        .await
        .map_err(|e| CliError::Other(format!("Failed to query indexes: {e}")))?
        .into_iter()
        .map(|row| {
            let cols: Vec<String> = row.get(6);
            RawIndexInfo {
                schema: row.get::<_, String>(0),
                table: row.get::<_, String>(1),
                name: row.get::<_, String>(2),
                is_unique: row.get::<_, bool>(3),
                is_primary: row.get::<_, bool>(4),
                method: row.get::<_, String>(5),
                columns: parse_index_columns(cols),
                where_clause: row.get::<_, Option<String>>(7),
                concurrent: false,
            }
        })
        .collect();

    raw.foreign_keys = client
        .query(queries::FOREIGN_KEYS_QUERY, &[])
        .await
        .map_err(|e| CliError::Other(format!("Failed to query foreign keys: {e}")))?
        .into_iter()
        .map(|row| RawForeignKeyInfo {
            schema: row.get::<_, String>(0),
            table: row.get::<_, String>(1),
            name: row.get::<_, String>(2),
            columns: row.get::<_, Vec<String>>(3),
            schema_to: row.get::<_, String>(4),
            table_to: row.get::<_, String>(5),
            columns_to: row.get::<_, Vec<String>>(6),
            on_update: action_code_to_string(&row.get::<_, String>(7)),
            on_delete: action_code_to_string(&row.get::<_, String>(8)),
            deferrable: row.get::<_, bool>(9),
            initially_deferred: row.get::<_, bool>(10),
        })
        .collect();

    raw.primary_keys = client
        .query(queries::PRIMARY_KEYS_QUERY, &[])
        .await
        .map_err(|e| CliError::Other(format!("Failed to query primary keys: {e}")))?
        .into_iter()
        .map(|row| RawPrimaryKeyInfo {
            schema: row.get::<_, String>(0),
            table: row.get::<_, String>(1),
            name: row.get::<_, String>(2),
            columns: row.get::<_, Vec<String>>(3),
        })
        .collect();

    raw.uniques = client
        .query(queries::UNIQUES_QUERY, &[])
        .await
        .map_err(|e| CliError::Other(format!("Failed to query unique constraints: {e}")))?
        .into_iter()
        .map(|row| RawUniqueInfo {
            schema: row.get::<_, String>(0),
            table: row.get::<_, String>(1),
            name: row.get::<_, String>(2),
            columns: row.get::<_, Vec<String>>(3),
            nulls_not_distinct: row.get::<_, bool>(4),
            deferrable: row.get::<_, bool>(5),
            initially_deferred: row.get::<_, bool>(6),
        })
        .collect();

    raw.checks = client
        .query(queries::CHECKS_QUERY, &[])
        .await
        .map_err(|e| CliError::Other(format!("Failed to query check constraints: {e}")))?
        .into_iter()
        .map(|row| RawCheckInfo {
            schema: row.get::<_, String>(0),
            table: row.get::<_, String>(1),
            name: row.get::<_, String>(2),
            expression: row.get::<_, String>(3),
        })
        .collect();

    Ok(())
}

#[cfg(feature = "tokio-postgres")]
async fn query_pg_async_security(
    client: &tokio_postgres::Client,
    raw: &mut PostgresRawData,
) -> Result<(), CliError> {
    use drizzle_migrations::postgres::introspect::{RawPolicyInfo, RawRoleInfo, queries};

    raw.roles = client
        .query(queries::ROLES_QUERY, &[])
        .await
        .map_err(|e| CliError::Other(format!("Failed to query roles: {e}")))?
        .into_iter()
        .map(|row| RawRoleInfo {
            name: row.get::<_, String>(0),
            create_db: row.get::<_, bool>(1),
            create_role: row.get::<_, bool>(2),
            inherit: row.get::<_, bool>(3),
        })
        .collect();

    raw.policies = client
        .query(queries::POLICIES_QUERY, &[])
        .await
        .map_err(|e| CliError::Other(format!("Failed to query policies: {e}")))?
        .into_iter()
        .map(|row| RawPolicyInfo {
            schema: row.get::<_, String>(0),
            table: row.get::<_, String>(1),
            name: row.get::<_, String>(2),
            as_clause: row.get::<_, String>(3),
            for_clause: row.get::<_, String>(4),
            to: row.get::<_, Vec<String>>(5),
            using: row.get::<_, Option<String>>(6),
            with_check: row.get::<_, Option<String>>(7),
        })
        .collect();

    Ok(())
}

/// Minimal schema list for snapshot
#[cfg(any(feature = "postgres-sync", feature = "tokio-postgres"))]
#[derive(Debug, Clone)]
struct RawSchemaInfo {
    name: String,
}

/// Aggregated raw introspection data collected from a `PostgreSQL` database.
///
/// Shared between `postgres-sync` and `tokio-postgres` code paths — the
/// collection phase differs (blocking vs async), but the downstream
/// processing is identical.
#[cfg(any(feature = "postgres-sync", feature = "tokio-postgres"))]
struct PostgresRawData {
    schemas: Vec<RawSchemaInfo>,
    tables: Vec<drizzle_migrations::postgres::introspect::RawTableInfo>,
    columns: Vec<drizzle_migrations::postgres::introspect::RawColumnInfo>,
    enums: Vec<drizzle_migrations::postgres::introspect::RawEnumInfo>,
    sequences: Vec<drizzle_migrations::postgres::introspect::RawSequenceInfo>,
    views: Vec<drizzle_migrations::postgres::introspect::RawViewInfo>,
    indexes: Vec<drizzle_migrations::postgres::introspect::RawIndexInfo>,
    foreign_keys: Vec<drizzle_migrations::postgres::introspect::RawForeignKeyInfo>,
    primary_keys: Vec<drizzle_migrations::postgres::introspect::RawPrimaryKeyInfo>,
    uniques: Vec<drizzle_migrations::postgres::introspect::RawUniqueInfo>,
    checks: Vec<drizzle_migrations::postgres::introspect::RawCheckInfo>,
    roles: Vec<drizzle_migrations::postgres::introspect::RawRoleInfo>,
    policies: Vec<drizzle_migrations::postgres::introspect::RawPolicyInfo>,
}

/// Build a [`PostgresDDL`] from raw introspection data.
///
/// Identical across the sync and async paths — only the query phase differs.
#[cfg(any(feature = "postgres-sync", feature = "tokio-postgres"))]
fn build_postgres_ddl(
    raw: PostgresRawData,
) -> drizzle_migrations::postgres::collection::PostgresDDL {
    use drizzle_migrations::postgres::ddl::Schema;
    use drizzle_migrations::postgres::introspect::{RawIntrospection, assemble_ddl};

    assemble_ddl(RawIntrospection {
        schemas: raw
            .schemas
            .into_iter()
            .map(|schema| Schema::new(schema.name))
            .collect(),
        tables: raw.tables,
        columns: raw.columns,
        enums: raw.enums,
        sequences: raw.sequences,
        views: raw.views,
        indexes: raw.indexes,
        foreign_keys: raw.foreign_keys,
        primary_keys: raw.primary_keys,
        unique_constraints: raw.uniques,
        check_constraints: raw.checks,
        roles: raw.roles,
        policies: raw.policies,
    })
}

/// Package a generated DDL + generated code into an [`IntrospectResult`].
#[cfg(any(feature = "postgres-sync", feature = "tokio-postgres"))]
fn finalize_postgres_introspection(
    ddl: &drizzle_migrations::postgres::collection::PostgresDDL,
    url: &str,
) -> IntrospectResult {
    use drizzle_migrations::postgres::codegen::{
        CodegenOptions, FieldCasing, generate_rust_schema,
    };

    let options = CodegenOptions {
        module_doc: Some(format!(
            "Schema introspected from {}",
            crate::output::mask_url(url)
        )),
        include_schema: true,
        schema_name: "Schema".to_string(),
        use_pub: true,
        field_casing: FieldCasing::default(),
    };
    let generated = generate_rust_schema(ddl, &options);

    let mut snap = drizzle_migrations::postgres::PostgresSnapshot::new();
    for entity in ddl.to_entities() {
        snap.add_entity(entity);
    }

    IntrospectResult {
        schema_code: generated.code,
        table_count: ddl.tables.list().len(),
        index_count: ddl.indexes.list().len(),
        view_count: ddl.views.list().len(),
        warnings: generated.warnings,
        mysql_catalog_defaults: None,
        snapshot: Snapshot::Postgres(snap),
        snapshot_path: std::path::PathBuf::new(),
    }
}

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

    #[cfg(any(feature = "postgres-sync", feature = "tokio-postgres"))]
    fn test_postgres_url() -> String {
        std::env::var("DATABASE_URL")
            .unwrap_or_else(|_| "postgres://postgres:postgres@localhost:5432/drizzle_test".into())
    }

    #[cfg(any(feature = "postgres-sync", feature = "tokio-postgres"))]
    fn test_postgres_creds() -> crate::config::PostgresCreds {
        crate::config::PostgresCreds::Url(test_postgres_url().into_boxed_str())
    }

    #[cfg(any(feature = "postgres-sync", feature = "tokio-postgres"))]
    fn unique_pg_name(prefix: &str) -> String {
        use std::time::{SystemTime, UNIX_EPOCH};

        let nanos = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .expect("time went backwards")
            .as_nanos();
        format!("{}_{}_{}", prefix, std::process::id(), nanos)
    }

    #[test]
    fn destructive_statement_detection_covers_drop_variants() {
        assert!(is_destructive_statement("DROP TABLE users;"));
        assert!(is_destructive_statement("DROP VIEW active_users;"));
        assert!(is_destructive_statement("DROP TYPE status;"));
        assert!(is_destructive_statement("DROP SCHEMA auth;"));
        assert!(is_destructive_statement("DROP ROLE app_user;"));
        assert!(is_destructive_statement(
            "DROP POLICY users_rls_policy ON users;"
        ));
        assert!(is_destructive_statement("TRUNCATE users;"));
        assert!(is_destructive_statement(
            "ALTER TABLE users DROP CONSTRAINT users_email_key;"
        ));

        assert!(!is_destructive_statement("CREATE TABLE users(id INTEGER);"));
        assert!(!is_destructive_statement(
            "ALTER TABLE users ADD COLUMN email text;"
        ));
    }

    #[test]
    fn mysql_structural_warnings_require_push_confirmation() {
        let statements =
            vec!["ALTER TABLE `users` MODIFY COLUMN `email` varchar(32) NOT NULL;".to_string()];
        let warnings = vec![
            "changing the type of users.email can truncate or reject existing values".to_string(),
        ];

        assert!(push_requires_confirmation(
            Dialect::Mysql,
            &statements,
            &warnings,
        ));
        assert!(!push_requires_confirmation(
            Dialect::Mysql,
            &statements,
            &[],
        ));
    }

    #[cfg(feature = "rusqlite")]
    #[test]
    fn sqlite_migrations_run_both_when_created_at_collides() {
        // Regression for drizzle-orm beta.19: migration identity is the folder
        // name, not the `created_at` timestamp. Two migrations that share a
        // wall-second must both apply.
        use drizzle_migrations::{Migration, Migrations};

        let dir = tempfile::tempdir().expect("tempdir");
        let db_path = dir.path().join("migrate.sqlite");
        let db_path_str = db_path.to_string_lossy().to_string();

        let first_set = Migrations::new(
            vec![Migration::with_hash(
                "20230331141203_first",
                "hash_one",
                1_680_271_923_000,
                vec!["CREATE TABLE created_at_dedupe_a (id INTEGER PRIMARY KEY)".to_string()],
            )],
            drizzle_types::Dialect::SQLite,
        );

        let first =
            run_sqlite_migrations(&first_set, &db_path_str, false).expect("first migrate succeeds");
        assert_eq!(first.applied_count, 1);

        let second_set = Migrations::new(
            vec![
                Migration::with_hash(
                    "20230331141203_first",
                    "hash_one",
                    1_680_271_923_000,
                    vec!["CREATE TABLE created_at_dedupe_a (id INTEGER PRIMARY KEY)".to_string()],
                ),
                Migration::with_hash(
                    "20230331141203_second",
                    "hash_two",
                    1_680_271_923_000,
                    vec!["CREATE TABLE created_at_dedupe_b (id INTEGER PRIMARY KEY)".to_string()],
                ),
            ],
            drizzle_types::Dialect::SQLite,
        );

        let second = run_sqlite_migrations(&second_set, &db_path_str, false)
            .expect("second migrate succeeds");
        assert_eq!(
            second.applied_count, 1,
            "only the second (newly introduced by name) migration should apply"
        );

        let conn = rusqlite::Connection::open(&db_path).expect("open sqlite");
        let rows: i64 = conn
            .query_row("SELECT COUNT(*) FROM __drizzle_migrations", [], |row| {
                row.get(0)
            })
            .expect("count migrations rows");
        assert_eq!(rows, 2, "both migration records should be stored");

        let table_b_exists: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='created_at_dedupe_b'",
                [],
                |row| row.get(0),
            )
            .expect("query sqlite_master");
        assert_eq!(
            table_b_exists, 1,
            "the new-name migration must execute even though created_at collides"
        );
    }

    #[cfg(feature = "rusqlite")]
    #[test]
    fn sqlite_cli_rebuild_preserves_cascade_children() {
        use drizzle_migrations::{Migration, Migrations};

        let dir = tempfile::tempdir().expect("tempdir");
        let db_path = dir.path().join("rebuild.sqlite");
        let conn = rusqlite::Connection::open(&db_path).expect("open sqlite");
        conn.execute_batch(
            "PRAGMA foreign_keys=ON;
             CREATE TABLE rebuild_parent (id INTEGER PRIMARY KEY);
             CREATE TABLE rebuild_child (
                 id INTEGER PRIMARY KEY,
                 parent_id INTEGER NOT NULL REFERENCES rebuild_parent(id) ON DELETE CASCADE
             );
             INSERT INTO rebuild_parent VALUES (1);
             INSERT INTO rebuild_child VALUES (1, 1);",
        )
        .expect("seed related tables");

        let migration = Migration::new(
            "20260825000000_rebuild",
            "PRAGMA foreign_keys=OFF;
             --> statement-breakpoint
             CREATE TABLE rebuild_parent_new (id INTEGER PRIMARY KEY) STRICT;
             --> statement-breakpoint
             INSERT INTO rebuild_parent_new SELECT * FROM rebuild_parent;
             --> statement-breakpoint
             DROP TABLE rebuild_parent;
             --> statement-breakpoint
             ALTER TABLE rebuild_parent_new RENAME TO rebuild_parent;
             --> statement-breakpoint
             PRAGMA foreign_keys=ON;",
        );
        let set = Migrations::new(vec![migration], drizzle_types::Dialect::SQLite);

        run_sqlite_migrations_on_connection(&set, &conn, false).expect("run safe rebuild");

        let children: i64 = conn
            .query_row("SELECT count(*) FROM rebuild_child", [], |row| row.get(0))
            .expect("count cascade children");
        assert_eq!(children, 1, "rebuild must not trigger ON DELETE CASCADE");
        let violations: i64 = conn
            .query_row("SELECT count(*) FROM pragma_foreign_key_check", [], |row| {
                row.get(0)
            })
            .expect("check rebuilt foreign keys");
        assert_eq!(violations, 0);
        let foreign_keys: i64 = conn
            .query_row("PRAGMA foreign_keys", [], |row| row.get(0))
            .expect("read restored foreign-key mode");
        assert_eq!(foreign_keys, 1);
    }

    #[cfg(feature = "rusqlite")]
    #[test]
    fn sqlite_push_rebuild_preserves_cascade_children() {
        let conn = rusqlite::Connection::open_in_memory().expect("open sqlite");
        conn.execute_batch(
            "PRAGMA foreign_keys=ON;
             CREATE TABLE rebuild_parent (id INTEGER PRIMARY KEY);
             CREATE TABLE rebuild_child (
                 id INTEGER PRIMARY KEY,
                 parent_id INTEGER NOT NULL REFERENCES rebuild_parent(id) ON DELETE CASCADE
             );
             INSERT INTO rebuild_parent VALUES (1);
             INSERT INTO rebuild_child VALUES (1, 1);",
        )
        .expect("seed related tables");
        let migration = drizzle_migrations::Migration::new(
            "push_rebuild",
            "PRAGMA foreign_keys=OFF;
             --> statement-breakpoint
             CREATE TABLE rebuild_parent_new (id INTEGER PRIMARY KEY) STRICT;
             --> statement-breakpoint
             INSERT INTO rebuild_parent_new SELECT * FROM rebuild_parent;
             --> statement-breakpoint
             DROP TABLE rebuild_parent;
             --> statement-breakpoint
             ALTER TABLE rebuild_parent_new RENAME TO rebuild_parent;
             --> statement-breakpoint
             PRAGMA foreign_keys=ON;",
        );

        execute_sqlite_statements_on_connection(&conn, migration.statements())
            .expect("apply push statements");

        let children: i64 = conn
            .query_row("SELECT count(*) FROM rebuild_child", [], |row| row.get(0))
            .expect("count cascade children");
        assert_eq!(children, 1, "push rebuild must not trigger cascade");
        let foreign_keys: i64 = conn
            .query_row("PRAGMA foreign_keys", [], |row| row.get(0))
            .expect("read restored foreign-key mode");
        assert_eq!(foreign_keys, 1);
    }

    #[cfg(feature = "rusqlite")]
    #[test]
    fn declined_destructive_push_is_an_abort_and_applies_nothing() {
        let dir = tempfile::tempdir().expect("tempdir");
        let db_path = dir.path().join("push.sqlite");
        rusqlite::Connection::open(&db_path)
            .expect("open sqlite")
            .execute_batch("CREATE TABLE keep_me (id INTEGER PRIMARY KEY);")
            .expect("seed table");

        let connection = ResolvedConnection {
            dialect: Dialect::Sqlite,
            driver: Driver::Rusqlite,
            credentials: Credentials::Sqlite {
                path: db_path.to_string_lossy().into_owned().into_boxed_str(),
            },
        };
        let plan = PushPlan {
            sql_statements: vec!["DROP TABLE `keep_me`;".to_string()],
            warnings: Vec::new(),
            destructive: true,
        };

        let error = apply_push_with_confirmation(&connection, &plan, false, || Ok(false))
            .expect_err("declining the prompt must not report success");
        assert!(matches!(error, CliError::Aborted(_)), "{error:?}");
        assert!(error.to_string().contains("aborted"), "{error}");

        let kept: i64 = rusqlite::Connection::open(&db_path)
            .expect("reopen sqlite")
            .query_row(
                "SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='keep_me'",
                [],
                |row| row.get(0),
            )
            .expect("query sqlite_master");
        assert_eq!(kept, 1, "a declined push must not execute any statement");
    }

    #[cfg(feature = "rusqlite")]
    #[test]
    fn run_migrations_creates_metadata_table_with_no_local_migrations() {
        let dir = tempfile::tempdir().expect("tempdir");
        let db_path = dir.path().join("empty.sqlite");
        let migrations_dir = dir.path().join("migrations");
        std::fs::create_dir_all(&migrations_dir).expect("create migrations dir");

        let creds = crate::config::Credentials::Sqlite {
            path: db_path.to_string_lossy().to_string().into_boxed_str(),
        };
        let connection = ResolvedConnection {
            dialect: crate::config::Dialect::Sqlite,
            driver: Driver::Rusqlite,
            credentials: creds,
        };

        let result = run_migrations(
            &connection,
            &migrations_dir,
            "__drizzle_migrations",
            "drizzle",
            false,
        )
        .expect("run migrations");
        assert_eq!(result.applied_count, 0);

        let conn = rusqlite::Connection::open(&db_path).expect("open sqlite");
        let exists: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='__drizzle_migrations'",
                [],
                |row| row.get(0),
            )
            .expect("check metadata table");
        assert_eq!(exists, 1, "migrations metadata table should be created");
    }

    #[cfg(any(
        feature = "rusqlite",
        feature = "libsql",
        feature = "turso",
        feature = "postgres-sync",
        feature = "tokio-postgres",
        feature = "mysql-sync",
        feature = "mysql-async"
    ))]
    #[test]
    fn validate_init_metadata_matches_drizzle_orm_semantics() {
        use drizzle_migrations::{Migration, Migrations};

        let empty_set = Migrations::empty(drizzle_types::Dialect::SQLite);
        validate_init_metadata(&[], &empty_set).expect("empty local migrations should be allowed");

        let single = Migrations::new(
            vec![Migration::with_hash(
                "20230331141203_init",
                "hash_single",
                1_680_271_923_000,
                vec!["CREATE TABLE t(id INTEGER PRIMARY KEY)".to_string()],
            )],
            drizzle_types::Dialect::SQLite,
        );
        validate_init_metadata(&[], &single).expect("single local migration should be allowed");

        let multiple = Migrations::new(
            vec![
                Migration::with_hash(
                    "20230331141203_first",
                    "hash_a",
                    1_680_271_923_000,
                    vec!["CREATE TABLE a(id INTEGER PRIMARY KEY)".to_string()],
                ),
                Migration::with_hash(
                    "20230331150000_second",
                    "hash_b",
                    1_680_275_400_000,
                    vec!["CREATE TABLE b(id INTEGER PRIMARY KEY)".to_string()],
                ),
            ],
            drizzle_types::Dialect::SQLite,
        );

        let err =
            validate_init_metadata(&[], &multiple).expect_err("multiple local migrations rejected");
        assert_eq!(
            err.to_string(),
            "--init can't be used with existing migrations"
        );

        let err = validate_init_metadata(&["20230331141203_init".to_string()], &single)
            .expect_err("existing db metadata should be rejected");
        assert_eq!(
            err.to_string(),
            "--init can't be used when database already has migrations set"
        );
    }

    #[cfg(any(
        feature = "rusqlite",
        feature = "libsql",
        feature = "turso",
        feature = "postgres-sync",
        feature = "tokio-postgres",
        feature = "mysql-sync",
        feature = "mysql-async",
        feature = "d1-http",
    ))]
    #[test]
    fn integrity_findings_detect_hash_drift() {
        use drizzle_migrations::{Migration, Migrations};

        let set = Migrations::new(
            vec![Migration::with_hash(
                "20230331141203_verify",
                "local_hash",
                1_680_271_923_000,
                vec!["CREATE TABLE t(id INTEGER PRIMARY KEY)".to_string()],
            )],
            drizzle_types::Dialect::SQLite,
        );

        let applied = vec![AppliedMigrationRecord {
            hash: "db_hash".to_string(),
            name: "20230331141203_verify".to_string(),
            dirty: false,
        }];

        let findings =
            collect_integrity_findings(&set, &applied).expect("local set is well-formed");
        assert_eq!(findings.len(), 1, "expected one finding: {findings:?}");
        assert!(
            findings[0].contains("20230331141203_verify") && findings[0].contains("drifted"),
            "unexpected finding: {}",
            findings[0]
        );
    }

    #[cfg(any(
        feature = "rusqlite",
        feature = "libsql",
        feature = "turso",
        feature = "postgres-sync",
        feature = "tokio-postgres",
        feature = "mysql-sync",
        feature = "mysql-async",
        feature = "d1-http",
    ))]
    #[test]
    fn integrity_findings_report_dirty_and_missing_local_rows() {
        use drizzle_migrations::{Migration, Migrations};

        let set = Migrations::new(
            vec![Migration::with_hash(
                "20230331141203_first",
                "hash_a",
                1_680_271_923_000,
                vec!["CREATE TABLE a(id INTEGER PRIMARY KEY)".to_string()],
            )],
            drizzle_types::Dialect::SQLite,
        );

        let applied = vec![
            AppliedMigrationRecord {
                hash: "hash_a".to_string(),
                name: "20230331141203_first".to_string(),
                dirty: true,
            },
            AppliedMigrationRecord {
                hash: "hash_gone".to_string(),
                name: "20230401000000_deleted_locally".to_string(),
                dirty: false,
            },
        ];

        let findings =
            collect_integrity_findings(&set, &applied).expect("local set is well-formed");
        assert_eq!(findings.len(), 2, "expected two findings: {findings:?}");
        assert!(
            findings[0].contains("interrupted mid-apply"),
            "unexpected finding: {}",
            findings[0]
        );
        assert!(
            findings[1].contains("no local migration"),
            "unexpected finding: {}",
            findings[1]
        );
    }

    #[cfg(any(
        feature = "rusqlite",
        feature = "libsql",
        feature = "turso",
        feature = "postgres-sync",
        feature = "tokio-postgres",
        feature = "mysql-sync",
        feature = "mysql-async",
        feature = "d1-http",
    ))]
    #[test]
    fn mysql_integrity_findings_do_not_recommend_automatic_repair() {
        use drizzle_migrations::{Migration, Migrations};

        let set = Migrations::new(
            vec![Migration::with_hash(
                "20230331141203_first",
                "hash_a",
                1_680_271_923_000,
                vec!["CREATE TABLE a(id INTEGER PRIMARY KEY)".to_string()],
            )],
            drizzle_types::Dialect::MySQL,
        );
        let findings = collect_integrity_findings(
            &set,
            &[AppliedMigrationRecord {
                hash: "hash_a".to_string(),
                name: "20230331141203_first".to_string(),
                dirty: true,
            }],
        )
        .expect("local set is well-formed");

        assert!(findings[0].contains("resolve it manually"), "{findings:?}");
        assert!(!findings[0].contains("--repair"), "{findings:?}");
    }

    #[cfg(any(
        feature = "rusqlite",
        feature = "libsql",
        feature = "turso",
        feature = "postgres-sync",
        feature = "tokio-postgres",
        feature = "mysql-sync",
        feature = "mysql-async",
        feature = "d1-http",
    ))]
    #[test]
    fn dirty_rows_are_not_counted_as_applied_in_plan() {
        use drizzle_migrations::{Migration, Migrations};

        let set = Migrations::new(
            vec![Migration::with_hash(
                "20230331141203_first",
                "hash_a",
                1_680_271_923_000,
                vec!["CREATE TABLE a(id INTEGER PRIMARY KEY)".to_string()],
            )],
            drizzle_types::Dialect::SQLite,
        );

        let applied = vec![AppliedMigrationRecord {
            hash: "hash_a".to_string(),
            name: "20230331141203_first".to_string(),
            dirty: true,
        }];

        let plan = build_migration_plan(&set, &applied).expect("build migration plan");
        assert_eq!(plan.applied_count, 0);
        assert_eq!(plan.pending_count, 1, "interrupted migration is unfinished");
        assert_eq!(plan.findings.len(), 1);
    }

    /// `migrate --plan`, `--dry-run` and `--verify` must not write: no
    /// database file, no tracking table.
    #[cfg(feature = "rusqlite")]
    #[test]
    fn sqlite_plan_creates_neither_the_database_nor_the_tracking_table() {
        use drizzle_migrations::{Migration, Migrations};

        let set = Migrations::new(
            vec![Migration::with_hash(
                "20230331141203_first",
                "hash_a",
                1_680_271_923_000,
                vec!["CREATE TABLE a(id INTEGER PRIMARY KEY)".to_string()],
            )],
            drizzle_types::Dialect::SQLite,
        );
        let dir = tempfile::tempdir().expect("tempdir");

        let missing = dir.path().join("missing.db");
        let plan = inspect_sqlite_migrations(&set, missing.to_str().expect("utf-8 path"))
            .expect("plan a missing database");
        assert_eq!(plan.pending_count, 1);
        assert!(!missing.exists(), "planning created the database file");

        let existing = dir.path().join("existing.db");
        rusqlite::Connection::open(&existing)
            .and_then(|conn| conn.execute("CREATE TABLE users(id INTEGER PRIMARY KEY)", []))
            .expect("seed database");
        let plan = inspect_sqlite_migrations(&set, existing.to_str().expect("utf-8 path"))
            .expect("plan an existing database");
        assert_eq!(plan.pending_count, 1);
        let tracking_tables: i64 = rusqlite::Connection::open(&existing)
            .and_then(|conn| {
                conn.query_row(
                    "SELECT count(*) FROM sqlite_master WHERE name = ?1",
                    [set.table_name()],
                    |row| row.get(0),
                )
            })
            .expect("inspect database");
        assert_eq!(tracking_tables, 0, "planning created the tracking table");
    }

    #[cfg(any(
        feature = "rusqlite",
        feature = "libsql",
        feature = "turso",
        feature = "postgres-sync",
        feature = "tokio-postgres",
        feature = "mysql-sync",
        feature = "mysql-async",
        feature = "d1-http",
    ))]
    #[test]
    fn build_migration_plan_counts_pending_statements() {
        use drizzle_migrations::{Migration, Migrations};

        let set = Migrations::new(
            vec![
                Migration::with_hash(
                    "20230331141203_first",
                    "hash_a",
                    1_680_271_923_000,
                    vec![
                        "CREATE TABLE a(id INTEGER PRIMARY KEY)".to_string(),
                        "CREATE INDEX a_id_idx ON a(id)".to_string(),
                    ],
                ),
                Migration::with_hash(
                    "20230331150000_second",
                    "hash_b",
                    1_680_275_400_000,
                    vec!["CREATE TABLE b(id INTEGER PRIMARY KEY)".to_string()],
                ),
            ],
            drizzle_types::Dialect::SQLite,
        );

        let applied = vec![AppliedMigrationRecord {
            hash: "hash_a".to_string(),
            name: "20230331141203_first".to_string(),
            dirty: false,
        }];

        let plan = build_migration_plan(&set, &applied).expect("build migration plan");
        assert_eq!(plan.applied_count, 1);
        assert_eq!(plan.pending_count, 1);
        assert_eq!(plan.pending_statements, 1);
        assert_eq!(plan.pending_migrations, vec!["20230331150000_second"]);
        assert!(plan.findings.is_empty());
    }

    #[cfg(any(feature = "postgres-sync", feature = "tokio-postgres"))]
    #[test]
    fn postgres_concurrent_index_detection_is_case_insensitive() {
        assert!(is_postgres_concurrent_index_statement(
            "CREATE INDEX CONCURRENTLY users_email_idx ON users (email);"
        ));
        assert!(is_postgres_concurrent_index_statement(
            "CREATE UNIQUE INDEX CONCURRENTLY users_email_idx ON users (email);"
        ));
        assert!(is_postgres_concurrent_index_statement(
            "drop index concurrently users_email_idx;"
        ));
        assert!(!is_postgres_concurrent_index_statement(
            "CREATE INDEX users_email_idx ON users (email);"
        ));
    }

    #[cfg(any(feature = "postgres-sync", feature = "tokio-postgres"))]
    #[test]
    fn postgres_concurrent_index_detection_over_statement_list() {
        let with_concurrent = vec![
            "CREATE TABLE users(id integer);".to_string(),
            "CREATE INDEX CONCURRENTLY users_email_idx ON users (id);".to_string(),
        ];
        let without_concurrent = vec![
            "CREATE TABLE users(id integer);".to_string(),
            "CREATE INDEX users_email_idx ON users (id);".to_string(),
        ];

        assert!(has_postgres_concurrent_index(&with_concurrent));
        assert!(!has_postgres_concurrent_index(&without_concurrent));
    }

    #[test]
    fn generate_push_sql_includes_concurrent_postgres_index() {
        use drizzle_migrations::parser::SchemaParser;
        use drizzle_types::Dialect;

        let previous = r#"
#[PostgresTable]
pub struct Users {
    #[column(primary)]
    pub id: i32,
    pub email: String,
}
"#;

        let current = r#"
#[PostgresTable]
pub struct Users {
    #[column(primary)]
    pub id: i32,
    pub email: String,
}

#[PostgresIndex(concurrent)]
pub struct UsersEmailIdx(Users::email);
"#;

        let prev =
            Snapshot::from_parse_result(&SchemaParser::parse(previous), Dialect::PostgreSQL, None);
        let curr =
            Snapshot::from_parse_result(&SchemaParser::parse(current), Dialect::PostgreSQL, None);

        let (sql, warnings) =
            generate_push_sql(&prev, &curr, false, &drizzle_migrations::DiffOptions::new())
                .expect("push sql generation");
        assert!(warnings.is_empty());
        assert_eq!(sql.len(), 1);
        assert_eq!(
            sql[0],
            "CREATE INDEX CONCURRENTLY \"users_email_idx\" ON \"users\"(\"email\");"
        );
    }

    #[test]
    fn generate_push_sql_uses_live_mysql_catalog_defaults() {
        use drizzle_migrations::mysql::MySQLCatalogDefaults;
        use drizzle_migrations::parser::SchemaParser;
        use drizzle_types::Dialect;

        let introspected = r#"
#[MySQLTable(NAME = "users")]
pub struct Users {
    #[column(PRIMARY)]
    pub id: i32,
}
"#;
        let desired = r#"
#[MySQLTable(NAME = "users", ENGINE = "InnoDB")]
pub struct Users {
    #[column(PRIMARY)]
    pub id: i32,
}
"#;
        let current =
            Snapshot::from_parse_result(&SchemaParser::parse(introspected), Dialect::MySQL, None);
        let desired =
            Snapshot::from_parse_result(&SchemaParser::parse(desired), Dialect::MySQL, None);
        let options = drizzle_migrations::DiffOptions::new()
            .mysql_catalog_defaults(MySQLCatalogDefaults::new().engine("InnoDB"));

        let (sql, warnings) = generate_push_sql(&current, &desired, false, &options)
            .expect("push SQL with live MySQL defaults");
        assert!(sql.is_empty(), "inherited defaults must not churn: {sql:?}");
        assert!(warnings.is_empty());
    }

    #[test]
    fn filter_patterns_support_negation_globs() {
        let raw = vec![
            "users_*".to_string(),
            "!users_4".to_string(),
            "!ad*".to_string(),
        ];
        let patterns = compile_patterns(Some(&raw)).expect("compile patterns");

        assert!(matches_patterns("users_1", patterns.as_deref()));
        assert!(!matches_patterns("users_4", patterns.as_deref()));
        assert!(!matches_patterns("admin", patterns.as_deref()));
        assert!(!matches_patterns("audit", patterns.as_deref()));
    }

    #[test]
    fn postgres_table_filter_matches_table_name_only() {
        use drizzle_migrations::parser::SchemaParser;
        use drizzle_migrations::schema::Snapshot;
        use drizzle_types::Dialect as BaseDialect;

        let code = r#"
#[PostgresTable(schema = "admin")]
pub struct AuditLog {
    #[column(primary)]
    pub id: i32,
}

#[PostgresTable(schema = "public")]
pub struct Users {
    #[column(primary)]
    pub id: i32,
}
"#;

        let parsed = SchemaParser::parse(code);
        let mut snapshot = Snapshot::from_parse_result(&parsed, BaseDialect::PostgreSQL, None);
        let filters = SnapshotFilters {
            tables: Some(vec!["admin.*".to_string()]),
            schemas: None,
            extensions: None,
            roles: None,
        };

        apply_snapshot_filters(&mut snapshot, crate::config::Dialect::Postgresql, &filters)
            .expect("apply filters");

        let remaining_tables = match snapshot {
            Snapshot::Postgres(s) => s
                .ddl
                .iter()
                .filter_map(|e| {
                    if let drizzle_types::postgres::ddl::PostgresEntity::Table(t) = e {
                        Some((t.schema.to_string(), t.name.to_string()))
                    } else {
                        None
                    }
                })
                .collect::<Vec<_>>(),
            _ => panic!("expected postgres snapshot"),
        };

        assert!(remaining_tables.is_empty());
    }

    #[test]
    fn postgres_schema_and_table_filters_intersect() {
        use drizzle_migrations::parser::SchemaParser;
        use drizzle_migrations::schema::Snapshot;
        use drizzle_types::Dialect as BaseDialect;

        let code = r#"
#[PostgresTable(schema = "dev")]
pub struct UsersDev {
    #[column(primary)]
    pub id: i32,
}

#[PostgresTable(schema = "public")]
pub struct UsersPublic {
    #[column(primary)]
    pub id: i32,
}
"#;

        let parsed = SchemaParser::parse(code);
        let mut snapshot = Snapshot::from_parse_result(&parsed, BaseDialect::PostgreSQL, None);
        let filters = SnapshotFilters {
            tables: Some(vec!["users_*".to_string()]),
            schemas: Some(vec!["!dev".to_string()]),
            extensions: None,
            roles: None,
        };

        apply_snapshot_filters(&mut snapshot, crate::config::Dialect::Postgresql, &filters)
            .expect("apply filters");

        let remaining_tables = match snapshot {
            Snapshot::Postgres(s) => s
                .ddl
                .iter()
                .filter_map(|e| {
                    if let drizzle_types::postgres::ddl::PostgresEntity::Table(t) = e {
                        Some((t.schema.to_string(), t.name.to_string()))
                    } else {
                        None
                    }
                })
                .collect::<Vec<_>>(),
            _ => panic!("expected postgres snapshot"),
        };

        assert_eq!(
            remaining_tables,
            vec![("public".to_string(), "users_public".to_string())]
        );
    }

    #[test]
    fn postgres_extensions_filter_excludes_postgis_internal_objects() {
        use drizzle_migrations::parser::SchemaParser;
        use drizzle_migrations::schema::Snapshot;
        use drizzle_types::Dialect as BaseDialect;

        let code = r#"
#[PostgresTable(schema = "topology")]
pub struct TopologyLayer {
    #[column(primary)]
    pub id: i32,
}

#[PostgresTable]
pub struct SpatialRefSys {
    #[column(primary)]
    pub id: i32,
}

#[PostgresTable]
pub struct Users {
    #[column(primary)]
    pub id: i32,
}
"#;

        let parsed = SchemaParser::parse(code);
        let mut snapshot = Snapshot::from_parse_result(&parsed, BaseDialect::PostgreSQL, None);
        let filters = SnapshotFilters {
            tables: None,
            schemas: None,
            extensions: Some(vec![Extension::Postgis]),
            roles: None,
        };

        apply_snapshot_filters(&mut snapshot, crate::config::Dialect::Postgresql, &filters)
            .expect("apply filters");

        let remaining_tables = match snapshot {
            Snapshot::Postgres(s) => s
                .ddl
                .iter()
                .filter_map(|e| {
                    if let drizzle_types::postgres::ddl::PostgresEntity::Table(t) = e {
                        Some((t.schema.to_string(), t.name.to_string()))
                    } else {
                        None
                    }
                })
                .collect::<Vec<_>>(),
            _ => panic!("expected postgres snapshot"),
        };

        assert_eq!(
            remaining_tables,
            vec![("public".to_string(), "users".to_string())]
        );
    }

    #[test]
    fn format_migration_sql_respects_breakpoints_flag() {
        let sql = vec![
            "CREATE TABLE users(id integer);".to_string(),
            "CREATE INDEX users_id_idx ON users(id);".to_string(),
        ];

        let with_breakpoints = format_migration_sql(&sql, true);
        assert_eq!(
            with_breakpoints,
            "CREATE TABLE users(id integer);\n--> statement-breakpoint\nCREATE INDEX users_id_idx ON users(id);"
        );

        let without_breakpoints = format_migration_sql(&sql, false);
        assert_eq!(
            without_breakpoints,
            "CREATE TABLE users(id integer);\n\nCREATE INDEX users_id_idx ON users(id);"
        );

        let empty = format_migration_sql(&[], false);
        assert_eq!(empty, "-- No tables to create (empty database)\n");
    }

    #[test]
    fn regenerate_sqlite_schema_applies_introspect_casing() {
        use crate::config::{Casing, IntrospectCasing};
        use drizzle_migrations::parser::SchemaParser;
        use drizzle_types::Dialect as BaseDialect;

        let code = r#"
#[SQLiteTable]
pub struct AuditLogs {
    #[column(primary)]
    pub id: i64,
    pub user_name: String,
}
"#;

        let parsed = SchemaParser::parse(code);
        let snapshot =
            Snapshot::from_parse_result(&parsed, BaseDialect::SQLite, Some(Casing::SnakeCase));

        let mut camel = IntrospectResult {
            schema_code: String::new(),
            table_count: 0,
            index_count: 0,
            view_count: 0,
            warnings: Vec::new(),
            mysql_catalog_defaults: None,
            snapshot: snapshot.clone(),
            snapshot_path: std::path::PathBuf::new(),
        };
        regenerate_schema_from_snapshot(
            &mut camel,
            crate::config::Dialect::Sqlite,
            Some(IntrospectCasing::Camel),
        );

        assert_eq!(
            camel.schema_code,
            "\
//! Auto-generated SQLite schema from introspection
//!
//! Schema introspected from filtered database objects

use drizzle::sqlite::prelude::*;

#[SQLiteTable]
pub struct AuditLogs {
    #[column(primary)]
    pub id: i64,
    pub userName: String,
}

#[derive(SQLiteSchema)]
pub struct Schema {
    pub auditLogs: AuditLogs,
}
"
        );

        let mut preserve = IntrospectResult {
            schema_code: String::new(),
            table_count: 0,
            index_count: 0,
            view_count: 0,
            warnings: Vec::new(),
            mysql_catalog_defaults: None,
            snapshot,
            snapshot_path: std::path::PathBuf::new(),
        };
        regenerate_schema_from_snapshot(
            &mut preserve,
            crate::config::Dialect::Sqlite,
            Some(IntrospectCasing::Preserve),
        );

        assert_eq!(
            preserve.schema_code,
            "\
//! Auto-generated SQLite schema from introspection
//!
//! Schema introspected from filtered database objects

use drizzle::sqlite::prelude::*;

#[SQLiteTable]
pub struct AuditLogs {
    #[column(primary)]
    pub id: i64,
    pub user_name: String,
}

#[derive(SQLiteSchema)]
pub struct Schema {
    pub audit_logs: AuditLogs,
}
"
        );
    }

    #[test]
    fn regenerate_postgres_schema_applies_introspect_casing() {
        use crate::config::{Casing, IntrospectCasing};
        use drizzle_migrations::parser::SchemaParser;
        use drizzle_types::Dialect as BaseDialect;

        let code = r#"
#[PostgresTable]
pub struct AuditLogs {
    #[column(primary)]
    pub id: i32,
    pub user_name: String,
}
"#;

        let parsed = SchemaParser::parse(code);
        let snapshot =
            Snapshot::from_parse_result(&parsed, BaseDialect::PostgreSQL, Some(Casing::SnakeCase));

        let mut camel = IntrospectResult {
            schema_code: String::new(),
            table_count: 0,
            index_count: 0,
            view_count: 0,
            warnings: Vec::new(),
            mysql_catalog_defaults: None,
            snapshot: snapshot.clone(),
            snapshot_path: std::path::PathBuf::new(),
        };
        regenerate_schema_from_snapshot(
            &mut camel,
            crate::config::Dialect::Postgresql,
            Some(IntrospectCasing::Camel),
        );

        assert_eq!(
            camel.schema_code,
            "\
//! Auto-generated PostgreSQL schema from introspection
//!
//! Schema introspected from filtered database objects

use drizzle::postgres::prelude::*;

#[PostgresTable]
pub struct AuditLogs {
    #[column(primary)]
    pub id: i32,
    pub userName: String,
}

#[derive(PostgresSchema)]
pub struct Schema {
    pub auditLogs: AuditLogs,
}
"
        );

        let mut preserve = IntrospectResult {
            schema_code: String::new(),
            table_count: 0,
            index_count: 0,
            view_count: 0,
            warnings: Vec::new(),
            mysql_catalog_defaults: None,
            snapshot,
            snapshot_path: std::path::PathBuf::new(),
        };
        regenerate_schema_from_snapshot(
            &mut preserve,
            crate::config::Dialect::Postgresql,
            Some(IntrospectCasing::Preserve),
        );

        assert_eq!(
            preserve.schema_code,
            "\
//! Auto-generated PostgreSQL schema from introspection
//!
//! Schema introspected from filtered database objects

use drizzle::postgres::prelude::*;

#[PostgresTable]
pub struct AuditLogs {
    #[column(primary)]
    pub id: i32,
    pub user_name: String,
}

#[derive(PostgresSchema)]
pub struct Schema {
    pub audit_logs: AuditLogs,
}
"
        );
    }

    #[cfg(feature = "postgres-sync")]
    #[test]
    fn postgres_sync_migrate_applies_concurrent_index_without_transaction() {
        use drizzle_migrations::{Migration, Migrations};
        use drizzle_types::Dialect;

        let creds = test_postgres_creds();
        let mut setup_client = match connect_postgres_sync(&creds) {
            Ok(c) => c,
            Err(e) => {
                eprintln!(
                    "Skipping postgres_sync_migrate_applies_concurrent_index_without_transaction: {}",
                    e
                );
                return;
            }
        };

        let table = unique_pg_name("cli_sync_users");
        let index = format!("{}_email_idx", table);
        let migration_schema = unique_pg_name("cli_sync_mig");

        setup_client
            .batch_execute(&format!(
                "DROP TABLE IF EXISTS \"{table}\" CASCADE; \
                 CREATE TABLE \"{table}\" (id integer, email text NOT NULL); \
                 INSERT INTO \"{table}\" (id, email) VALUES (1, 'a@example.com');"
            ))
            .expect("setup table for concurrent index test");
        drop(setup_client);

        let migration_tag = format!("20260212000000_{table}");
        let migration_sql =
            format!("CREATE INDEX CONCURRENTLY \"{index}\" ON \"{table}\" (\"email\");");
        let set = Migrations::with_tracking(
            vec![Migration::new(&migration_tag, &migration_sql)],
            Dialect::PostgreSQL,
            drizzle_migrations::Tracking::POSTGRES.schema(migration_schema.clone()),
        );

        let result = run_postgres_sync_migrations(&set, &creds, false)
            .expect("sync migration with concurrent index should succeed");
        assert_eq!(result.applied_count, 1);

        let mut verify_client = connect_postgres_sync(&creds).expect("reconnect for verification");
        let exists: i64 = verify_client
            .query_one(
                "SELECT COUNT(*)::bigint FROM pg_indexes \
                 WHERE schemaname = 'public' AND tablename = $1 AND indexname = $2",
                &[&table, &index],
            )
            .expect("query pg_indexes")
            .get(0);
        assert_eq!(exists, 1, "concurrent index was not created");

        let _ = verify_client.batch_execute(&format!(
            "DROP TABLE IF EXISTS \"{table}\" CASCADE; \
             DROP SCHEMA IF EXISTS \"{migration_schema}\" CASCADE;"
        ));
    }

    #[cfg(feature = "postgres-sync")]
    #[test]
    fn postgres_sync_migrate_upgrades_legacy_tracking_table_and_applies_pending() {
        use drizzle_migrations::{Migration, Migrations};
        use drizzle_types::Dialect;

        let creds = test_postgres_creds();
        let mut client = connect_postgres_sync(&creds)
            .expect("connect postgres for legacy tracking upgrade test");

        let applied_table = unique_pg_name("cli_sync_applied");
        let pending_table = unique_pg_name("cli_sync_pending");
        let migration_schema = unique_pg_name("cli_sync_tracking");

        client
            .batch_execute(&format!(
                "DROP TABLE IF EXISTS \"{applied_table}\" CASCADE; \
                 DROP TABLE IF EXISTS \"{pending_table}\" CASCADE; \
                 DROP SCHEMA IF EXISTS \"{migration_schema}\" CASCADE; \
                 CREATE SCHEMA \"{migration_schema}\"; \
                 CREATE TABLE \"{migration_schema}\".\"__drizzle_migrations\" (id SERIAL PRIMARY KEY, hash TEXT NOT NULL, created_at BIGINT); \
                 CREATE TABLE \"{applied_table}\" (id integer primary key);"
            ))
            .expect("setup legacy postgres tracking metadata");

        let first = Migration::new(
            &format!("20230331141203_{applied_table}"),
            &format!("CREATE TABLE \"{applied_table}\" (id integer primary key);"),
        );
        let second = Migration::new(
            &format!("20230331141204_{pending_table}"),
            &format!("CREATE TABLE \"{pending_table}\" (id integer primary key);"),
        );
        let set = Migrations::with_tracking(
            vec![first.clone(), second.clone()],
            Dialect::PostgreSQL,
            drizzle_migrations::Tracking::POSTGRES.schema(migration_schema.clone()),
        );

        client
            .execute(
                &format!(
                    "INSERT INTO \"{migration_schema}\".\"__drizzle_migrations\" (hash, created_at) VALUES ($1, $2)"
                ),
                &[&first.hash(), &first.created_at()],
            )
            .expect("insert legacy applied migration row");
        drop(client);

        let result = run_postgres_sync_migrations(&set, &creds, false)
            .expect("sync migration upgrade should succeed");
        assert_eq!(result.applied_count, 1);
        assert_eq!(result.applied_migrations, vec![second.hash().to_string()]);

        let mut verify_client = connect_postgres_sync(&creds).expect("reconnect for verification");
        let columns: Vec<String> = verify_client
            .query(
                "SELECT column_name FROM information_schema.columns WHERE table_schema = $1 AND table_name = '__drizzle_migrations' ORDER BY ordinal_position",
                &[&migration_schema],
            )
            .expect("query upgraded tracking columns")
            .into_iter()
            .map(|row| row.get(0))
            .collect();
        assert_eq!(
            columns,
            vec!["id", "hash", "created_at", "name", "applied_at"]
        );

        let rows: Vec<(String, i64, String, Option<String>)> = verify_client
            .query(
                &format!(
                    "SELECT hash, created_at, name, applied_at::text FROM \"{migration_schema}\".\"__drizzle_migrations\" ORDER BY id ASC"
                ),
                &[],
            )
            .expect("query upgraded metadata rows")
            .into_iter()
            .map(|row| (row.get(0), row.get(1), row.get(2), row.get(3)))
            .collect();
        assert_eq!(rows.len(), 2);
        assert_eq!(rows[0].0, first.hash());
        assert_eq!(rows[0].1, first.created_at());
        assert_eq!(rows[0].2, first.name());
        assert!(
            rows[0].3.is_some(),
            "legacy rows get applied_at backfilled so they cannot read as interrupted"
        );
        assert_eq!(rows[1].0, second.hash());
        assert_eq!(rows[1].1, second.created_at());
        assert_eq!(rows[1].2, second.name());
        assert!(rows[1].3.is_some());

        let pending_exists: i64 = verify_client
            .query_one(
                "SELECT COUNT(*)::bigint FROM information_schema.tables WHERE table_schema = 'public' AND table_name = $1",
                &[&pending_table],
            )
            .expect("query pending table")
            .get(0);
        assert_eq!(pending_exists, 1);

        let _ = verify_client.batch_execute(&format!(
            "DROP TABLE IF EXISTS \"{applied_table}\" CASCADE; \
             DROP TABLE IF EXISTS \"{pending_table}\" CASCADE; \
             DROP SCHEMA IF EXISTS \"{migration_schema}\" CASCADE;"
        ));
    }

    /// An interrupted pre-transactional upgrade left `name` without
    /// `applied_at`; the next run finishes it instead of failing on the
    /// missing column. Statement errors also carry the server message.
    #[cfg(feature = "postgres-sync")]
    #[test]
    fn postgres_sync_migrate_finishes_half_upgraded_tracking_table() {
        use drizzle_migrations::{Migration, Migrations};
        use drizzle_types::Dialect;

        let creds = test_postgres_creds();
        let mut client = connect_postgres_sync(&creds).expect("connect postgres");
        let migration_schema = unique_pg_name("cli_sync_half_upgraded");
        client
            .batch_execute(&format!(
                "DROP SCHEMA IF EXISTS \"{migration_schema}\" CASCADE; \
                 CREATE SCHEMA \"{migration_schema}\"; \
                 CREATE TABLE \"{migration_schema}\".\"__drizzle_migrations\" \
                     (id SERIAL PRIMARY KEY, hash TEXT NOT NULL, created_at BIGINT, name TEXT); \
                 INSERT INTO \"{migration_schema}\".\"__drizzle_migrations\" (hash, created_at) \
                     VALUES ('half_hash', 1680271923456);"
            ))
            .expect("reproduce the half-upgraded table");
        drop(client);

        let applied = Migration::with_hash(
            "20230331141203_cli_half",
            "half_hash",
            1_680_271_923_000,
            vec![],
        );
        let broken = Migration::new(
            "20230401000000_cli_broken",
            &format!("INSERT INTO \"{migration_schema}\".missing_table VALUES (1);"),
        );
        let tracking = drizzle_migrations::Tracking::POSTGRES.schema(migration_schema.clone());

        let set =
            Migrations::with_tracking(vec![applied.clone()], Dialect::PostgreSQL, tracking.clone());
        let result = run_postgres_sync_migrations(&set, &creds, false)
            .expect("a half-upgraded tracking table is completed");
        assert_eq!(result.applied_count, 0);

        let set = Migrations::with_tracking(vec![applied, broken], Dialect::PostgreSQL, tracking);
        let error = run_postgres_sync_migrations(&set, &creds, false)
            .expect_err("the broken migration fails")
            .to_string();
        assert!(error.contains("20230401000000_cli_broken"), "{error}");
        assert!(
            error.contains("does not exist"),
            "server message is kept: {error}"
        );

        let mut verify_client = connect_postgres_sync(&creds).expect("reconnect");
        let row = verify_client
            .query_one(
                &format!(
                    "SELECT name, applied_at IS NOT NULL FROM \"{migration_schema}\".\"__drizzle_migrations\""
                ),
                &[],
            )
            .expect("upgraded row");
        assert_eq!(row.get::<_, String>(0), "20230331141203_cli_half");
        assert!(row.get::<_, bool>(1));
        let _ = verify_client.batch_execute(&format!(
            "DROP SCHEMA IF EXISTS \"{migration_schema}\" CASCADE;"
        ));
    }

    /// Concurrent first runs take the advisory lock before creating the
    /// tracking schema, so none of them races on `CREATE SCHEMA`.
    #[cfg(feature = "postgres-sync")]
    #[test]
    fn postgres_sync_concurrent_first_migrate_creates_tracking_schema_once() {
        use drizzle_migrations::{Migration, Migrations};
        use drizzle_types::Dialect;

        let creds = test_postgres_creds();
        let migration_schema = unique_pg_name("cli_sync_concurrent_first");
        let table = unique_pg_name("cli_sync_concurrent_table");
        let set = Migrations::with_tracking(
            vec![Migration::new(
                "20240101000000_cli_concurrent",
                &format!("CREATE TABLE \"{table}\" (id integer);"),
            )],
            Dialect::PostgreSQL,
            drizzle_migrations::Tracking::POSTGRES.schema(migration_schema.clone()),
        );

        let barrier = std::sync::Barrier::new(6);
        let applied: usize = std::thread::scope(|scope| {
            let handles = (0..6)
                .map(|_| {
                    scope.spawn(|| {
                        barrier.wait();
                        run_postgres_sync_migrations(&set, &creds, false)
                            .map(|result| result.applied_count)
                            .map_err(|error| error.to_string())
                    })
                })
                .collect::<Vec<_>>();
            handles
                .into_iter()
                .map(|handle| {
                    handle
                        .join()
                        .expect("thread")
                        .unwrap_or_else(|error| panic!("{error}"))
                })
                .sum()
        });
        assert_eq!(applied, 1);

        let mut verify_client = connect_postgres_sync(&creds).expect("reconnect");
        let _ = verify_client.batch_execute(&format!(
            "DROP TABLE IF EXISTS \"{table}\" CASCADE; \
             DROP SCHEMA IF EXISTS \"{migration_schema}\" CASCADE;"
        ));
    }

    #[cfg(feature = "postgres-sync")]
    #[test]
    fn postgres_sync_migrate_upgrade_rejects_unmatched_legacy_rows() {
        use drizzle_migrations::{Migration, Migrations};
        use drizzle_types::Dialect;

        let creds = test_postgres_creds();
        let mut client =
            connect_postgres_sync(&creds).expect("connect postgres for unmatched legacy row test");

        let migration_schema = unique_pg_name("cli_sync_tracking_unmatched");
        client
            .batch_execute(&format!(
                "DROP SCHEMA IF EXISTS \"{migration_schema}\" CASCADE; \
                 CREATE SCHEMA \"{migration_schema}\"; \
                 CREATE TABLE \"{migration_schema}\".\"__drizzle_migrations\" (id SERIAL PRIMARY KEY, hash TEXT NOT NULL, created_at BIGINT);"
            ))
            .expect("setup unmatched legacy metadata");

        let migration = Migration::new(
            "20230331141203_cli_sync_first",
            "CREATE TABLE \"cli_sync_unmatched_target\" (id integer primary key);",
        );
        let set = Migrations::with_tracking(
            vec![migration],
            Dialect::PostgreSQL,
            drizzle_migrations::Tracking::POSTGRES.schema(migration_schema.clone()),
        );

        client
            .execute(
                &format!(
                    "INSERT INTO \"{migration_schema}\".\"__drizzle_migrations\" (hash, created_at) VALUES ($1, $2)"
                ),
                &[&"unknown_hash", &1_680_271_924_000_i64],
            )
            .expect("insert unmatched legacy row");
        drop(client);

        let err = run_postgres_sync_migrations(&set, &creds, false)
            .expect_err("unmatched legacy metadata should fail");
        assert!(err.to_string().contains("do not match local migrations"));

        let mut verify_client = connect_postgres_sync(&creds).expect("reconnect for verification");
        let columns: Vec<String> = verify_client
            .query(
                "SELECT column_name FROM information_schema.columns WHERE table_schema = $1 AND table_name = '__drizzle_migrations' ORDER BY ordinal_position",
                &[&migration_schema],
            )
            .expect("query legacy tracking columns")
            .into_iter()
            .map(|row| row.get(0))
            .collect();
        assert_eq!(columns, vec!["id", "hash", "created_at"]);

        let _ = verify_client.batch_execute(&format!(
            "DROP TABLE IF EXISTS \"cli_sync_unmatched_target\" CASCADE; \
             DROP SCHEMA IF EXISTS \"{migration_schema}\" CASCADE;"
        ));
    }

    #[cfg(feature = "tokio-postgres")]
    #[test]
    fn tokio_postgres_migrate_applies_concurrent_index_without_transaction() {
        use drizzle_migrations::{Migration, Migrations};
        use drizzle_types::Dialect;

        let creds = test_postgres_creds();
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .expect("create tokio runtime");

        rt.block_on(async {
            let client = match connect_postgres_async(&creds).await {
                Ok(client) => client,
                Err(e) => {
                    eprintln!(
                        "Skipping tokio_postgres_migrate_applies_concurrent_index_without_transaction: {}",
                        e
                    );
                    return;
                }
            };

            let table = unique_pg_name("cli_async_users");
            let index = format!("{}_email_idx", table);
            let migration_schema = unique_pg_name("cli_async_mig");

            client
                .batch_execute(&format!(
                    "DROP TABLE IF EXISTS \"{table}\" CASCADE; \
                     CREATE TABLE \"{table}\" (id integer, email text NOT NULL); \
                     INSERT INTO \"{table}\" (id, email) VALUES (1, 'a@example.com');"
                ))
                .await
                .expect("setup table for async concurrent index test");

            let migration_tag = format!("20260212000000_{table}");
            let migration_sql = format!(
                "CREATE INDEX CONCURRENTLY \"{index}\" ON \"{table}\" (\"email\");"
            );
            let set = Migrations::with_tracking(
                vec![Migration::new(&migration_tag, &migration_sql)],
                Dialect::PostgreSQL,
                drizzle_migrations::Tracking::POSTGRES.schema(migration_schema.clone()),
            );

            let result = run_postgres_async_inner(&set, &creds, false)
                .await
                .expect("async migration with concurrent index should succeed");
            assert_eq!(result.applied_count, 1);

            let exists: i64 = client
                .query_one(
                    "SELECT COUNT(*)::bigint FROM pg_indexes \
                     WHERE schemaname = 'public' AND tablename = $1 AND indexname = $2",
                    &[&table, &index],
                )
                .await
                .expect("query pg_indexes")
                .get(0);
            assert_eq!(exists, 1, "async concurrent index was not created");

            let _ = client
                .batch_execute(&format!(
                    "DROP TABLE IF EXISTS \"{table}\" CASCADE; \
                     DROP SCHEMA IF EXISTS \"{migration_schema}\" CASCADE;"
                ))
                .await;
        });
    }

    #[cfg(feature = "tokio-postgres")]
    #[test]
    fn tokio_postgres_migrate_upgrades_legacy_tracking_table_and_applies_pending() {
        use drizzle_migrations::{Migration, Migrations};
        use drizzle_types::Dialect;

        let creds = test_postgres_creds();
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .expect("create tokio runtime");

        rt.block_on(async {
            let client = connect_postgres_async(&creds)
                .await
                .expect("connect tokio-postgres for legacy tracking upgrade test");

            let applied_table = unique_pg_name("cli_async_applied");
            let pending_table = unique_pg_name("cli_async_pending");
            let migration_schema = unique_pg_name("cli_async_tracking");

            client
                .batch_execute(&format!(
                    "DROP TABLE IF EXISTS \"{applied_table}\" CASCADE; \
                     DROP TABLE IF EXISTS \"{pending_table}\" CASCADE; \
                     DROP SCHEMA IF EXISTS \"{migration_schema}\" CASCADE; \
                     CREATE SCHEMA \"{migration_schema}\"; \
                     CREATE TABLE \"{migration_schema}\".\"__drizzle_migrations\" (id SERIAL PRIMARY KEY, hash TEXT NOT NULL, created_at BIGINT); \
                     CREATE TABLE \"{applied_table}\" (id integer primary key);"
                ))
                .await
                .expect("setup legacy postgres tracking metadata");

            let first = Migration::new(
                &format!("20230331141203_{applied_table}"),
                &format!("CREATE TABLE \"{applied_table}\" (id integer primary key);"),
            );
            let second = Migration::new(
                &format!("20230331141204_{pending_table}"),
                &format!("CREATE TABLE \"{pending_table}\" (id integer primary key);"),
            );
            let set = Migrations::with_tracking(
                vec![first.clone(), second.clone()],
                Dialect::PostgreSQL,
                drizzle_migrations::Tracking::POSTGRES.schema(migration_schema.clone()),
            );

            client
                .execute(
                    &format!(
                        "INSERT INTO \"{migration_schema}\".\"__drizzle_migrations\" (hash, created_at) VALUES ($1, $2)"
                    ),
                    &[&first.hash(), &first.created_at()],
                )
                .await
                .expect("insert legacy applied migration row");

            let result = run_postgres_async_inner(&set, &creds, false)
                .await
                .expect("async migration upgrade should succeed");
            assert_eq!(result.applied_count, 1);
            assert_eq!(result.applied_migrations, vec![second.hash().to_string()]);

            let columns: Vec<String> = client
                .query(
                    "SELECT column_name FROM information_schema.columns WHERE table_schema = $1 AND table_name = '__drizzle_migrations' ORDER BY ordinal_position",
                    &[&migration_schema],
                )
                .await
                .expect("query upgraded tracking columns")
                .into_iter()
                .map(|row| row.get(0))
                .collect();
            assert_eq!(
                columns,
                vec!["id", "hash", "created_at", "name", "applied_at"]
            );

            let rows: Vec<(String, i64, String, Option<String>)> = client
                .query(
                    &format!(
                        "SELECT hash, created_at, name, applied_at::text FROM \"{migration_schema}\".\"__drizzle_migrations\" ORDER BY id ASC"
                    ),
                    &[],
                )
                .await
                .expect("query upgraded metadata rows")
                .into_iter()
                .map(|row| (row.get(0), row.get(1), row.get(2), row.get(3)))
                .collect();
            assert_eq!(rows.len(), 2);
            assert_eq!(rows[0].0, first.hash());
            assert_eq!(rows[0].1, first.created_at());
            assert_eq!(rows[0].2, first.name());
            assert!(
                rows[0].3.is_some(),
                "legacy rows get applied_at backfilled so they cannot read as interrupted"
            );
            assert_eq!(rows[1].0, second.hash());
            assert_eq!(rows[1].1, second.created_at());
            assert_eq!(rows[1].2, second.name());
            assert!(rows[1].3.is_some());

            let pending_exists: i64 = client
                .query_one(
                    "SELECT COUNT(*)::bigint FROM information_schema.tables WHERE table_schema = 'public' AND table_name = $1",
                    &[&pending_table],
                )
                .await
                .expect("query pending table")
                .get(0);
            assert_eq!(pending_exists, 1);

            let _ = client
                .batch_execute(&format!(
                    "DROP TABLE IF EXISTS \"{applied_table}\" CASCADE; \
                     DROP TABLE IF EXISTS \"{pending_table}\" CASCADE; \
                     DROP SCHEMA IF EXISTS \"{migration_schema}\" CASCADE;"
                ))
                .await;
        });
    }
}