autumn-cli 0.9.0

//! Declarative-schema tooling (tracking issue #1975).
//!
//! [`parse`] reads the `#[model]` structs into the [`autumn_schema_core`] IR:
//! the desired state. [`snapshot`] holds the checked-in baseline. [`diff`]
//! compares the two and writes the migration. [`run`] backs the
//! `autumn schema parse|snapshot|diff|pull|migrate|doctor` commands.
//!
//! [`rename`] applies `#[renamed_from]` hints in the diff. [`shadow`] replays
//! the migrations on an empty dev database for `schema diff --dev-url`.
//! [`schema_rs`] writes the `diesel::table!` blocks of the managed models to
//! `src/schema.rs` on `schema diff --write-migration`.

pub mod diff;
pub mod doctor;
pub mod introspect;
pub mod migrate;
pub mod parse;
pub mod rename;
pub mod schema_rs;
pub mod shadow;
pub mod snapshot;

use std::collections::BTreeMap;
use std::path::{Path, PathBuf};

use autumn_schema_core::{Backend, Column, Table};

use diff::{MigrationPlan, SchemaChange};
use parse::parse_models_path;
use snapshot::{SNAPSHOT_DEFAULT_PATH, SchemaSnapshot};

/// A `--backend pg|sqlite` override for the `snapshot` action, mapping onto the
/// schema-core [`Backend`] dialect tag.
#[derive(clap::ValueEnum, Debug, Clone, Copy)]
pub enum BackendArg {
    /// `PostgreSQL` (accepted as `pg` or `postgres`).
    #[value(alias = "postgres")]
    Pg,
    /// `SQLite`.
    Sqlite,
}

impl From<BackendArg> for Backend {
    fn from(arg: BackendArg) -> Self {
        match arg {
            BackendArg::Pg => Self::Postgres,
            BackendArg::Sqlite => Self::Sqlite,
        }
    }
}

/// The environment variable that `schema diff --dev-url` also reads.
const DEV_URL_ENV: &str = "AUTUMN_DEV_URL";

/// The `autumn schema` subcommand actions (experimental).
#[derive(clap::Subcommand, Debug)]
pub enum SchemaAction {
    /// Parse `#[model]` structs at PATH (a `.rs` file or a directory of them)
    /// and print the resulting schema IR as JSON. Experimental / read-only.
    Parse {
        /// A `.rs` file or a directory containing `*.rs` model files.
        path: std::path::PathBuf,
    },
    /// Write a canonical, versioned, dialect-tagged snapshot of the declared
    /// `#[model]` structs — the checked-in diff baseline a later slice's diff
    /// engine compares the desired state against. Read-only w.r.t. the models.
    Snapshot {
        /// A `.rs` model file or a directory of them to snapshot. Defaults to
        /// the app's `src/models` directory, else `src/models.rs`.
        #[arg(long, value_name = "PATH")]
        from: Option<PathBuf>,
        /// Where to write the snapshot. Defaults to `.autumn/schema-snapshot.json`.
        /// Mutually exclusive with `--stdout` (writing a file and printing to
        /// stdout are two different output modes). The default is applied in
        /// code when omitted, so `conflicts_with` never misfires on the default.
        #[arg(long, value_name = "PATH", conflicts_with = "stdout")]
        out: Option<PathBuf>,
        /// The dialect to tag the snapshot with. Defaults to the project's
        /// configured database backend (`detect_backend`).
        #[arg(long, value_enum)]
        backend: Option<BackendArg>,
        /// Print the canonical JSON to stdout instead of writing a file (useful
        /// for diffing / tests).
        #[arg(long)]
        stdout: bool,
    },
    /// Diff the declared `#[model]` structs against the checked-in snapshot and
    /// either print the pending migration (default) or write it as a diesel
    /// `up.sql`/`down.sql` pair (`--write-migration`). `--write-migration` also
    /// advances the snapshot and writes `src/schema.rs`. Destructive drops are
    /// refused unless `--allow-destructive` is passed.
    Diff {
        /// Models source (a `.rs` file or a directory). Defaults to `src/models`
        /// then `src/models.rs`.
        #[arg(long, value_name = "PATH")]
        from: Option<PathBuf>,
        /// Baseline snapshot. Defaults to `.autumn/schema-snapshot.json`.
        #[arg(long, value_name = "PATH")]
        snapshot: Option<PathBuf>,
        /// The dialect. Defaults to the project's configured backend.
        #[arg(long, value_enum)]
        backend: Option<BackendArg>,
        /// Write `migrations/<timestamp>_<name>/{up,down}.sql` instead of
        /// printing. Also advance the snapshot and update `src/schema.rs`.
        #[arg(long)]
        write_migration: bool,
        /// Migration directory suffix when writing. Defaults to `schema_update`.
        #[arg(long, value_name = "NAME")]
        name: Option<String>,
        /// Permit destructive drops / an independent drop+add (tier-2 guard).
        #[arg(long)]
        allow_destructive: bool,
        /// Use the migrations, replayed on a scratch database, as the baseline.
        /// On Postgres the command creates the scratch database on the server
        /// of URL (the role needs CREATEDB) and drops it after; the database in
        /// URL does not change. The snapshot becomes optional. Also read from
        /// `AUTUMN_DEV_URL`, which keeps the password out of the process list.
        #[arg(
            long,
            value_name = "URL",
            env = DEV_URL_ENV,
            hide_env_values = true
        )]
        dev_url: Option<String>,
    },
    /// Introspect the configured Postgres database and write a canonical,
    /// dialect-tagged snapshot of its live shape — the DB-derived diff baseline.
    ///
    /// Read-only w.r.t. the database (only catalog reads). Unlike `snapshot`
    /// (which derives the baseline from the declared `#[model]` structs), `pull`
    /// derives it from the live database, so a brownfield schema — or a schema
    /// that drifted from the models — is captured exactly as it exists. A
    /// `SQLite` URL needs a CLI built with `--features sqlite`; the default
    /// build refuses it with a clear message.
    Pull {
        /// Config profile whose database URL to introspect (defaults to the
        /// ambient profile resolution the other CLI commands use).
        #[arg(long, value_name = "PROFILE")]
        profile: Option<String>,
        /// Where to write the snapshot. Defaults to `.autumn/schema-snapshot.json`.
        #[arg(long, value_name = "PATH")]
        out: Option<PathBuf>,
        /// Override the dialect tag / apply-path selection. Defaults to the
        /// backend implied by the resolved database URL.
        #[arg(long, value_enum)]
        backend: Option<BackendArg>,
        /// Print what would change without writing the snapshot file.
        #[arg(long)]
        dry_run: bool,
    },
    /// Apply pending migration files against the configured database.
    ///
    /// Does NOT modify the checked-in schema snapshot: the baseline advances at
    /// generation time via `schema diff --write-migration`, so this command only
    /// applies pending migrations. Provider-locked against the snapshot's dialect;
    /// the destructive-change guards ran at diff time, so migration files apply
    /// verbatim here.
    Migrate {
        /// Config profile whose database URL to apply against (defaults to the
        /// ambient profile resolution the other CLI commands use).
        #[arg(long, value_name = "PROFILE")]
        profile: Option<String>,
    },
    /// Read-only diagnosis of the declarative-schema state: filesystem, snapshot,
    /// model drift, backend provider-lock, and pending migrations. Exits
    /// non-zero when any check is an actionable error.
    Doctor {
        /// Config profile whose database URL to probe (defaults to the ambient
        /// profile resolution the other CLI commands use).
        #[arg(long, value_name = "PROFILE")]
        profile: Option<String>,
        /// Emit the checks as JSON instead of the aligned text report.
        #[arg(long)]
        json: bool,
    },
}

/// Run an `autumn schema` action. Prints to stdout on success; on error, writes
/// a message to stderr and exits non-zero (matching the other CLI handlers,
/// which own their own error reporting).
pub fn run(action: SchemaAction) {
    let result = match action {
        SchemaAction::Parse { path } => run_parse(&path).map(|json| println!("{json}")),
        SchemaAction::Snapshot {
            from,
            out,
            backend,
            stdout,
        } => run_snapshot(from.as_deref(), out.as_deref(), backend, stdout),
        SchemaAction::Diff {
            from,
            snapshot,
            backend,
            write_migration,
            name,
            allow_destructive,
            dev_url,
        } => run_diff(&DiffArgs {
            from: from.as_deref(),
            snapshot: snapshot.as_deref(),
            backend,
            write_migration,
            name: name.as_deref(),
            allow_destructive,
            dev_url: dev_url.as_deref(),
        }),
        SchemaAction::Pull {
            profile,
            out,
            backend,
            dry_run,
        } => run_pull(profile.as_deref(), out.as_deref(), backend, dry_run),
        SchemaAction::Migrate { profile } => migrate::run_migrate(profile.as_deref()),
        SchemaAction::Doctor { profile, json } => doctor::run_doctor(profile.as_deref(), json),
    };
    if let Err(message) = result {
        eprintln!("error: {message}");
        std::process::exit(1);
    }
}

/// Map the CLI's runtime [`autumn_web::config::DatabaseBackend`] onto the
/// schema-core [`Backend`] dialect tag.
const fn map_detected_backend(detected: autumn_web::config::DatabaseBackend) -> Backend {
    match detected {
        autumn_web::config::DatabaseBackend::Postgres => Backend::Postgres,
        autumn_web::config::DatabaseBackend::Sqlite => Backend::Sqlite,
    }
}

/// Resolve the schema-command backend for an explicit `--profile`, from the
/// already profile-resolved primary database `url`.
///
/// When a URL is configured its scheme is authoritative
/// ([`DatabaseBackend::detect`](autumn_web::config::DatabaseBackend::detect)),
/// so `--profile <name>` selects the apply path / provider-lock of the database
/// it actually acts against; only when no URL is configured does it fall back to
/// the profile-aware project default
/// ([`crate::generate::detect_backend_for_profile`]). Shared by `schema migrate`
/// and `schema doctor` so the two commands can never derive the backend
/// inconsistently for the same profile. With `profile == None` the resolution is
/// unchanged from the ambient behavior.
fn backend_for_url(project_root: &Path, profile: Option<&str>, url: Option<&str>) -> Backend {
    url.and_then(autumn_web::config::DatabaseBackend::detect)
        .map_or_else(
            || {
                map_detected_backend(crate::generate::detect_backend_for_profile(
                    project_root,
                    profile,
                ))
            },
            map_detected_backend,
        )
}

/// Resolve the default models source when `--from` is omitted: the app's
/// `src/models` directory if it exists, else the single-file `src/models.rs`
/// layout, else a clear error telling the user to pass `--from`. Autumn supports
/// both layouts, so the snapshot default must accept either.
///
/// Takes an explicit `project_root` (rather than reading the process CWD) so it
/// is testable without mutating the current directory.
fn resolve_default_models_path(project_root: &Path) -> Result<PathBuf, String> {
    let dir = project_root.join("src").join("models");
    if dir.is_dir() {
        return Ok(dir);
    }
    let file = project_root.join("src").join("models.rs");
    if file.is_file() {
        return Ok(file);
    }
    Err(format!(
        "no models found at {} or {} — pass --from to point at your models file or directory",
        dir.display(),
        file.display()
    ))
}

/// Resolve the declarative models source only when it actually exists.
///
/// Mirrors [`resolve_default_models_path`]'s `src/models` dir → `src/models.rs`
/// file precedence, but returns `None` (not an error) when neither is present —
/// so callers that must degrade gracefully on absence (the `migrate` snapshot
/// refresh, `doctor`'s drift check) share one resolver instead of duplicating it.
pub fn existing_models_path(project_root: &Path) -> Option<PathBuf> {
    let dir = project_root.join("src").join("models");
    if dir.is_dir() {
        return Some(dir);
    }
    let file = project_root.join("src").join("models.rs");
    if file.is_file() {
        return Some(file);
    }
    None
}

/// Whether the `snapshot` command requires the current directory to be the
/// project root. It does whenever any input falls back to a project-relative
/// default: the source (`--from` omitted → `src/models`), the default output
/// file (`--out` omitted while not writing to stdout → `.autumn/…`), or the
/// auto-detected backend (`--backend` omitted — detection reads the root's
/// `autumn.toml` / `.env`, so anywhere else it would silently mis-detect
/// Postgres).
///
/// A fully explicit invocation — `--from` plus `--out`/`--stdout` plus
/// `--backend` — needs no project root and can run from anywhere. Kept pure so
/// the decision matrix is unit-testable without touching the process CWD.
const fn needs_project_root(
    from: Option<&Path>,
    out: Option<&Path>,
    stdout: bool,
    backend: Option<BackendArg>,
) -> bool {
    from.is_none() || (out.is_none() && !stdout) || backend.is_none()
}

/// The error surfaced when the `snapshot` command needs the project root but the
/// current directory is not one. When auto-detecting the backend is one of the
/// reasons, the message names `--backend` so the user can either run from the
/// root or opt out of detection; otherwise it is the generic project-root
/// message (a default path, not the backend, forced the requirement).
fn project_root_required_error(backend_is_a_reason: bool) -> String {
    if backend_is_a_reason {
        "cannot auto-detect the database backend outside the project root — \
         run from the project root or pass --backend pg|sqlite"
            .to_string()
    } else {
        crate::generate::GenerateError::NotInProject.to_string()
    }
}

/// Assemble a snapshot from the declared models and either write it to `out`
/// (default [`SNAPSHOT_DEFAULT_PATH`]) or print it to stdout.
///
/// The backend is `--backend` when given, else the project's configured backend
/// (`generate::detect_backend`, the same pg-vs-sqlite resolution the generator
/// uses). The source is `--from` (a `.rs` file or a directory), else the
/// project's default models path (`src/models` directory, else `src/models.rs`;
/// see [`resolve_default_models_path`]). Parser diagnostics are surfaced on
/// stderr (non-fatal), mirroring `parse`.
///
/// Building the baseline from the declared models (not a live DB) is deliberate:
/// at adoption time the models and the database agree, so this establishes the
/// initial baseline with no database connection. A `--from-db` oracle that reads
/// a live Postgres schema is deferred to a later slice (see the module docs).
fn run_snapshot(
    from: Option<&Path>,
    out: Option<&Path>,
    backend: Option<BackendArg>,
    stdout: bool,
) -> Result<(), String> {
    let project_root = std::env::current_dir()
        .map_err(|e| format!("failed to resolve the current directory: {e}"))?;

    // When any input falls back to a project-relative default the command needs
    // the current directory to be the project root (see `needs_project_root`):
    // the source (`--from` omitted → `src/models`), the default output file
    // (`--out` omitted while writing a file), or — critically — the auto-detected
    // backend (`--backend` omitted → detection reads the root's `autumn.toml` /
    // `.env`). Run from a subdirectory without `--backend`, `detect_backend`
    // would find no config and silently tag the snapshot Postgres, producing a
    // wrong baseline for a SQLite app. Fail fast with a clear message instead of
    // a confusing IO error or a mis-tagged snapshot. A fully explicit invocation
    // (`--from` plus `--out`/`--stdout` plus `--backend`) needs no project root.
    let backend_given = backend.is_some();
    if needs_project_root(from, out, stdout, backend)
        && crate::generate::ensure_project_root(&project_root).is_err()
    {
        return Err(project_root_required_error(!backend_given));
    }

    let models_path = match from {
        Some(path) => path.to_path_buf(),
        None => resolve_default_models_path(&project_root)?,
    };

    let backend = backend.map_or_else(
        || map_detected_backend(crate::generate::detect_backend(&project_root)),
        Backend::from,
    );

    let parsed = parse_models_path(&models_path, backend).map_err(|e| e.to_string())?;
    for diag in &parsed.diagnostics {
        eprintln!("warning: {}", diag.message);
    }

    let snapshot = SchemaSnapshot::new(backend, parsed.tables);
    if stdout {
        print!("{}", snapshot::to_canonical_json(&snapshot));
        return Ok(());
    }

    let out_path = out.map_or_else(|| PathBuf::from(SNAPSHOT_DEFAULT_PATH), Path::to_path_buf);
    snapshot::write_snapshot(&out_path, &snapshot).map_err(|e| e.to_string())?;
    println!(
        "wrote schema snapshot for {} table(s) to {}",
        snapshot.tables.len(),
        out_path.display()
    );
    Ok(())
}

/// Introspect the configured Postgres database and write (or, with `--dry-run`,
/// describe) a canonical, dialect-tagged snapshot of its live shape.
///
/// This is the DB-introspection counterpart to `snapshot` (which derives the
/// baseline from the declared models). The resolution order — project root, then
/// the profile-resolved database URL, then the URL-implied backend (honoring a
/// `--backend` override) — mirrors `schema migrate` / `schema doctor`, so the
/// three commands can never derive the backend inconsistently for one profile.
///
/// Both backends are supported: a resolved `SQLite` backend introspects via
/// `introspect::introspect_sqlite` on the `sqlite` build (and is refused loudly on
/// a default Postgres-only build). Before writing, a **provider-lock** guard refuses
/// to clobber an existing snapshot tagged for another backend.
fn run_pull(
    profile: Option<&str>,
    out: Option<&Path>,
    backend: Option<BackendArg>,
    dry_run: bool,
) -> Result<(), String> {
    let project_root = std::env::current_dir()
        .map_err(|e| format!("failed to resolve the current directory: {e}"))?;
    pull_at(&project_root, profile, out, backend, dry_run)
}

/// The body of [`run_pull`], taking an explicit `project_root` so the wiring is
/// testable without mutating the process CWD (mirrors `diff_at` / `migrate_at`).
fn pull_at(
    project_root: &Path,
    profile: Option<&str>,
    out: Option<&Path>,
    backend: Option<BackendArg>,
    dry_run: bool,
) -> Result<(), String> {
    crate::generate::ensure_project_root(project_root)
        .map_err(|_| crate::generate::GenerateError::NotInProject.to_string())?;

    // Resolve the write/primary database URL exactly as the other schema commands
    // do, honoring an explicit `--profile`.
    let url = crate::migrate::resolve_primary_url(profile).ok_or_else(|| {
        "no database URL configured — set AUTUMN_DATABASE__URL or DATABASE_URL, or add a \
         [database] primary_url to autumn.toml"
            .to_string()
    })?;

    // Backend from the SAME profile-resolved context as the URL (its scheme is
    // authoritative), honoring an explicit `--backend` override.
    let backend: Backend = backend.map_or_else(
        || backend_for_url(project_root, profile, Some(url.as_str())),
        Backend::from,
    );

    let out_path = out.map_or_else(
        || project_root.join(SNAPSHOT_DEFAULT_PATH),
        Path::to_path_buf,
    );

    // PROVIDER-LOCK: refuse to overwrite an existing snapshot tagged for another
    // backend (mirrors `ensure_backend_matches` / the doctor provider-lock check).
    // A missing or unreadable-as-absent snapshot is fine — this is the first
    // baseline. The existing tables also serve as the `--dry-run` diff baseline.
    let existing = load_existing_snapshot(&out_path);
    if let Some(existing) = &existing
        && existing.backend != backend
    {
        return Err(format!(
            "refusing to overwrite the existing snapshot at {} (backend {:?}) with a {:?} pull \
             — the snapshot is provider-locked to {:?}. Point --out elsewhere or remove the \
             mismatched snapshot first.",
            out_path.display(),
            existing.backend,
            backend,
            existing.backend
        ));
    }

    // Introspect the live database into the IR, dispatching on the resolved
    // backend. Each arm references only its own connection type — the SQLite arm is
    // compiled only under the `sqlite` backend-flip (see `introspect_sqlite`); a
    // default (Postgres) build that somehow resolves a SQLite backend fails loudly
    // rather than silently mis-introspecting.
    let tables = match backend {
        Backend::Postgres => introspect::introspect_postgres(&url).map_err(|e| e.to_string())?,
        #[cfg(feature = "sqlite")]
        Backend::Sqlite => introspect::introspect_sqlite(&url).map_err(|e| e.to_string())?,
        #[cfg(not(feature = "sqlite"))]
        Backend::Sqlite => {
            return Err(
                "`autumn schema pull` detected a SQLite backend, but this CLI build targets \
                 Postgres only. Rebuild with `--features sqlite` to introspect a SQLite \
                 database (the sqlite backend-flip must never be co-built with the default \
                 Postgres backend). No snapshot was written."
                    .to_string(),
            );
        }
    };
    let snapshot = SchemaSnapshot::new(backend, tables);

    if dry_run {
        report_pull_dry_run(&snapshot, existing.as_ref(), &out_path);
        return Ok(());
    }

    // A file that is present on disk but did not load (`existing` is `None` while
    // the path exists) is corrupt/unreadable-as-absent — a non-dry-run pull is
    // about to replace it. Tell the user rather than silently clobber it. An
    // absent snapshot stays silent (it exists nowhere to overwrite); a readable
    // but backend-mismatched snapshot never reaches here (provider-lock refused
    // above).
    if existing.is_none() && out_path.exists() {
        eprintln!(
            "note: existing snapshot at {} was unreadable; replacing it.",
            out_path.display()
        );
    }

    snapshot::write_snapshot(&out_path, &snapshot).map_err(|e| e.to_string())?;
    println!(
        "pulled schema snapshot for {} table(s) from the database to {}",
        snapshot.tables.len(),
        out_path.display()
    );
    Ok(())
}

/// Load the snapshot at `path` for the pull's provider-lock / dry-run baseline.
/// A missing OR unreadable file is treated as **absent** (`Ok(None)`) — a pull
/// establishes a fresh baseline, so a corrupt prior file is not fatal — but a
/// backend-mismatched *readable* snapshot still surfaces to the provider-lock
/// caller (it loaded fine; only its backend disagrees).
fn load_existing_snapshot(path: &Path) -> Option<SchemaSnapshot> {
    snapshot::load_snapshot(path).ok()
}

/// Print what a `--dry-run` pull would change between the existing snapshot
/// (baseline) and the freshly-pulled tables, without writing anything.
///
/// Drift is computed BIDIRECTIONALLY (via [`doctor::compute_db_schema_drift`]) so
/// the dry-run never under-reports: the forward plan is what pulling would change
/// in the snapshot, but a manually-dropped default / FK / CHECK in the live DB is
/// invisible to the forward pass (a desired `None` is "retained") and only surfaces
/// in the reverse plan. The report is built by the pure [`format_pull_dry_run_report`]
/// so it can be unit-tested without any I/O.
fn report_pull_dry_run(
    pulled: &SchemaSnapshot,
    existing: Option<&SchemaSnapshot>,
    out_path: &Path,
) {
    match existing {
        None => {
            println!(
                "--dry-run: no existing snapshot at {} — would create a new baseline with {} \
                 table(s) from the database.",
                out_path.display(),
                pulled.tables.len()
            );
        }
        Some(existing) => {
            let drift = doctor::compute_db_schema_drift(&existing.tables, &pulled.tables);
            print!(
                "{}",
                format_pull_dry_run_report(&drift.forward, &drift.reverse, out_path)
            );
        }
    }
}

/// Format the `--dry-run` report body for a snapshot that already exists, from the
/// pre-computed bidirectional plans. Pure (no I/O) so it is unit-testable.
///
/// The report ALWAYS surfaces both directions independently:
///
/// * The **forward** plan — the structural changes a real pull would apply to the
///   snapshot (added/dropped columns, type changes, index adds/drops, …) — is
///   printed via [`diff::describe_plan`] whenever it is non-empty.
/// * The **reverse-only** removals — a default / foreign key / CHECK the snapshot
///   still carries but the live DB has dropped, which the forward pass structurally
///   cannot express (a desired `None` is "retained", never a drop) — are ALWAYS
///   named, phrased as removals, INDEPENDENT of whether the forward plan had
///   changes. Without this a pull that both adds a column AND drops a default would
///   report only the added column and silently omit that pulling also removes the
///   default from the snapshot.
///
/// Structural adds/drops already in the forward plan are never re-listed here — only
/// the asymmetric [`reverse_only_removals`] facets are surfaced from the reverse
/// plan, so nothing is double-counted. When BOTH directions are empty the snapshot
/// is up to date.
fn format_pull_dry_run_report(
    forward: &MigrationPlan,
    reverse: &MigrationPlan,
    out_path: &Path,
) -> String {
    use std::fmt::Write as _;

    let removals = reverse_only_removals(reverse);
    if forward.is_empty() && removals.is_empty() {
        return format!(
            "--dry-run: the snapshot at {} is up to date — the database matches it.\n",
            out_path.display()
        );
    }

    let mut out = String::new();
    if !forward.is_empty() {
        let _ = writeln!(
            out,
            "--dry-run: the database differs from the snapshot at {} — {} change(s) would be \
             captured (nothing written):",
            out_path.display(),
            forward.changes.len()
        );
        out.push_str(&diff::describe_plan(forward));
    }
    if !removals.is_empty() {
        let _ = writeln!(
            out,
            "--dry-run: pulling would also remove from the snapshot at {}: {} (nothing written).",
            out_path.display(),
            removals.join(", ")
        );
    }
    out
}

/// The reverse-plan changes that represent a facet present in the snapshot but
/// **dropped from the live DB** — the asymmetric removals the forward pass cannot
/// emit (`diff_column` treats a desired `None` as "unknown, retained", never a
/// drop). Rendered as human-readable removal phrases for the dry-run report.
///
/// Only genuine removals are surfaced, so a facet that merely *changed* value (and
/// is therefore already captured by the forward plan) is never double-counted:
///
/// * [`SchemaChange::SetDefault`] with `from: None` — the reverse baseline (the DB)
///   had no default at all, so the snapshot's default was dropped. A reverse
///   `SetDefault` whose `from` is `Some` is a value *change* the forward plan
///   already reports and is skipped here.
/// * [`SchemaChange::AddForeignKey`] / [`SchemaChange::AddForeignKeyToExistingColumn`]
///   — a foreign key the snapshot carries that the DB dropped (the reverse pass
///   only ever emits these when the DB-side column had no FK; a *retarget* surfaces
///   as `ForeignKeyChange` in BOTH directions and is not listed here).
/// * [`SchemaChange::AddCheck`] — a CHECK constraint the snapshot carries that the
///   DB dropped (checks are name-keyed add-only, so a reverse `AddCheck` is always a
///   removal).
fn reverse_only_removals(reverse: &MigrationPlan) -> Vec<String> {
    reverse
        .changes
        .iter()
        .filter_map(|change| match change {
            SchemaChange::SetDefault {
                table,
                column,
                from: None,
                ..
            } => Some(format!("default on `{table}.{column}`")),
            SchemaChange::AddForeignKey { table, column, .. }
            | SchemaChange::AddForeignKeyToExistingColumn { table, column } => {
                Some(format!("foreign key on `{table}.{column}`"))
            }
            SchemaChange::AddCheck { table, check } => Some(format!(
                "CHECK `{}` on `{table}`",
                check.name.as_deref().unwrap_or("(unnamed)")
            )),
            _ => None,
        })
        .collect()
}

/// Diff the declared models against the checked-in snapshot baseline and either
/// print the pending migration or write it as a diesel `up.sql`/`down.sql` pair.
///
/// **Provider-lock ordering (caller-owned):** the snapshot is loaded and its
/// backend tag is validated against the desired backend *before* the models are
/// parsed or diffed, so a dialect-mismatched snapshot fails cleanly with
/// [`snapshot::SnapshotError::BackendMismatch`] and no diff work runs. This is
/// the first caller of [`SchemaSnapshot::ensure_backend_matches`].
/// The inputs of `autumn schema diff`.
#[derive(Debug, Default, Clone, Copy)]
struct DiffArgs<'a> {
    /// `--from`: the models source.
    from: Option<&'a Path>,
    /// `--snapshot`: the baseline snapshot path.
    snapshot: Option<&'a Path>,
    /// `--backend`: the dialect.
    backend: Option<BackendArg>,
    /// `--write-migration`.
    write_migration: bool,
    /// `--name`: the migration directory suffix.
    name: Option<&'a str>,
    /// `--allow-destructive`.
    allow_destructive: bool,
    /// `--dev-url`: replay the migrations on this database and use the result
    /// as the baseline (see [`shadow`]).
    dev_url: Option<&'a str>,
}

fn run_diff(args: &DiffArgs<'_>) -> Result<(), String> {
    let project_root = std::env::current_dir()
        .map_err(|e| format!("failed to resolve the current directory: {e}"))?;
    diff_at(&project_root, args)
}

/// The body of [`run_diff`], taking an explicit `project_root` (rather than the
/// process CWD) so the command wiring is testable without mutating the current
/// directory.
fn diff_at(project_root: &Path, args: &DiffArgs<'_>) -> Result<(), String> {
    diff_at_with(project_root, args, &write_file_atomic)
}

/// Write `text` to `path` through a temporary file and a rename, so a failed
/// write never leaves a part of the file. Keeps the permissions of the old
/// file. A symbolic link stays a link: the write goes to its target.
fn write_file_atomic(path: &Path, text: &str) -> std::io::Result<()> {
    let target = std::fs::canonicalize(path).unwrap_or_else(|_| path.to_path_buf());
    let dir = target.parent().unwrap_or_else(|| Path::new("."));
    let mut tmp = tempfile::NamedTempFile::new_in(dir)?;
    std::io::Write::write_all(&mut tmp, text.as_bytes())?;
    if let Ok(meta) = std::fs::metadata(&target) {
        tmp.as_file().set_permissions(meta.permissions())?;
    }
    tmp.as_file().sync_all()?;
    tmp.persist(&target).map(|_| ()).map_err(|e| e.error)
}

/// [`diff_at`] with the `src/schema.rs` writer as a parameter, so a test can
/// make the write fail.
fn diff_at_with(
    project_root: &Path,
    args: &DiffArgs<'_>,
    write_schema_rs: &dyn Fn(&Path, &str) -> std::io::Result<()>,
) -> Result<(), String> {
    let DiffArgs {
        from,
        snapshot: snapshot_path,
        backend,
        write_migration,
        name,
        allow_destructive,
        dev_url,
    } = *args;
    // When any input falls back to a project-relative default (the models
    // source, the default snapshot path, the auto-detected backend, or the
    // `migrations/` directory a `--dev-url` replays), the command needs the
    // current directory to be the project root — mirroring `run_snapshot`.
    let backend_given = backend.is_some();
    if (from.is_none() || snapshot_path.is_none() || backend.is_none() || dev_url.is_some())
        && crate::generate::ensure_project_root(project_root).is_err()
    {
        return Err(project_root_required_error(!backend_given));
    }

    // (a) Resolve the desired backend: `--backend`, else the project's configured
    //     backend.
    let backend: Backend = backend.map_or_else(
        || map_detected_backend(crate::generate::detect_backend(project_root)),
        Backend::from,
    );

    // (b) Load the snapshot. (c) PROVIDER-LOCK GUARD — before parsing/diffing.
    let snapshot_path = snapshot_path.map_or_else(
        || project_root.join(SNAPSHOT_DEFAULT_PATH),
        Path::to_path_buf,
    );
    let snapshot = load_diff_snapshot(&snapshot_path, backend, dev_url.is_some())?;

    // (d) Parse the desired state with the SAME backend tag.
    let models_path = match from {
        Some(path) => path.to_path_buf(),
        None => resolve_default_models_path(project_root)?,
    };
    let desired = parse_models_path(&models_path, backend).map_err(|e| e.to_string())?;
    for d in &desired.diagnostics {
        eprintln!("warning: {}", d.message);
    }

    // The baseline: the snapshot, or with `--dev-url` the replayed migrations.
    let (baseline_tables, other_relations) = match dev_url {
        Some(url) => {
            let replay = replay_baseline(project_root, url, backend, snapshot.as_ref(), &desired)?;
            (replay.tables, replay.other_relations)
        }
        None => (
            snapshot
                .as_ref()
                .map(|s| s.tables.clone())
                .unwrap_or_default(),
            std::collections::BTreeSet::new(),
        ),
    };

    // (e) Diff (pure) then guard (policy).
    let opts = diff::DiffOptions {
        allow_destructive,
        ..Default::default()
    };
    let plan = diff::diff_schema(&baseline_tables, &desired, opts);
    if plan.is_empty() {
        println!("No schema changes — models match the baseline.");
        if write_migration {
            let sync = plan_schema_rs(project_root, &desired, &plan)?;
            let adopted = adopted_snapshot(snapshot.as_ref(), &baseline_tables, backend, &desired);
            commit_empty_plan(
                &snapshot_path,
                adopted.as_ref(),
                sync.as_ref(),
                write_schema_rs,
            )?;
        }
        return Ok(());
    }
    diff::guard_plan(&plan, opts).map_err(|e| e.to_string())?;
    if let Some(clash) = relation_clash(&plan, &baseline_tables, &other_relations) {
        return Err(clash);
    }

    // A SQLite table-recreate needs the full desired and baseline tables, and the
    // skipped decimal columns. Postgres ignores the context. The baseline is the
    // renamed baseline: the plan's non-rename changes use the new names.
    let renamed = rename::renamed_baseline(&baseline_tables, &plan.changes);
    let ctx =
        diff::SchemaContext::from_tables(&desired.tables, &renamed).with_skipped_decimals(&desired);
    let up = diff::emit_up_sql_with_context(&plan, &ctx).map_err(|e| e.to_string())?;
    let down = diff::emit_down_sql_with_context(&plan, &ctx).map_err(|e| e.to_string())?;

    if !write_migration {
        print!("{}", diff::describe_plan(&plan));
        println!("\n-- up.sql\n{up}\n-- down.sql\n{down}");
        return Ok(());
    }

    // Read `src/schema.rs` before the first write, so a read error stops here.
    let sync = plan_schema_rs(project_root, &desired, &plan)?;

    // (f) Write the migration dir, matching the generator's naming convention.
    let ts = crate::generate::timestamp_now();
    let suffix = crate::generate::naming::snake(name.unwrap_or("schema_update"));
    let dir = write_migration_dir(project_root, &ts, &suffix, &up, &down)?;

    // (g) Advance the checked-in snapshot to the state this migration converges
    // the database on. The snapshot now moves at GENERATION time (here), not at
    // apply time — `schema migrate` only applies files. The target is the guarded
    // `plan` PROJECTED onto the baseline (not `desired.tables` wholesale), so a
    // change the guard reduced or refused is never baked into the baseline: the
    // snapshot tracks exactly what the emitted SQL produces. A plain `schema diff`
    // (no `--write-migration`) returned above (step e), so it never reaches here
    // and never touches the snapshot.
    //
    // The migration files and the checked-in snapshot must advance together: if
    // the snapshot write fails (read-only dir, full disk, ...) after the
    // migration is on disk, retrying `schema diff --write-migration` would
    // regenerate the same SQL as a second migration (baseline never advanced)
    // and applying both would fail on duplicate DDL. Roll the migration dir back
    // on a snapshot-write failure so the pre-command state is left intact.
    let mut target_tables = project_plan_target(&baseline_tables, &plan);
    mark_managed(&mut target_tables, &desired);

    // (h) Then write `src/schema.rs` (see `commit_outputs`).
    commit_outputs(
        &snapshot_path,
        Some(&SchemaSnapshot::new(backend, target_tables)),
        Some(&dir),
        sync.as_ref(),
        write_schema_rs,
    )?;

    println!(
        "wrote migration {} ({} change(s))",
        dir.display(),
        plan.changes.len()
    );
    println!("advanced schema snapshot at {}", snapshot_path.display());
    report_schema_rs(sync.as_ref());
    Ok(())
}

/// `--write-migration` with no SQL change. A newly managed model needs no
/// migration, but the snapshot records its `managed` flag (`adopted`, see
/// [`adopted_snapshot`]) and `src/schema.rs` can need its block.
fn commit_empty_plan(
    snapshot_path: &Path,
    adopted: Option<&SchemaSnapshot>,
    sync: Option<&(PathBuf, schema_rs::SchemaRsSync)>,
    write_schema_rs: &dyn Fn(&Path, &str) -> std::io::Result<()>,
) -> Result<(), String> {
    commit_outputs(snapshot_path, adopted, None, sync, write_schema_rs)?;
    if adopted.is_some() {
        println!("recorded the managed models in {}", snapshot_path.display());
    }
    report_schema_rs(sync);
    Ok(())
}

/// The snapshot to write for an empty plan: the snapshot with the new
/// `managed` flags, when a flag changed. With `--dev-url` and no snapshot,
/// a new snapshot from the replayed `baseline`, as a non-empty plan makes.
fn adopted_snapshot(
    snapshot: Option<&SchemaSnapshot>,
    baseline: &[Table],
    backend: Backend,
    desired: &parse::ParsedSchema,
) -> Option<SchemaSnapshot> {
    snapshot.map_or_else(
        || {
            let mut tables = baseline.to_vec();
            mark_managed(&mut tables, desired);
            Some(SchemaSnapshot::new(backend, tables))
        },
        |s| {
            let mut tables = s.tables.clone();
            mark_managed(&mut tables, desired);
            (tables != s.tables).then(|| SchemaSnapshot::new(s.backend, tables))
        },
    )
}

/// Mark each table that a managed model declares as managed. A model that
/// becomes managed is then recorded in the snapshot, and a later delete of
/// the model drops its table. The flag never goes back to unmanaged.
fn mark_managed(tables: &mut [Table], desired: &parse::ParsedSchema) {
    for table in tables {
        if desired
            .tables
            .iter()
            .any(|t| t.managed && t.name == table.name)
        {
            table.managed = true;
        }
    }
}

/// Write the snapshot (when given), then `src/schema.rs` (when given). If a
/// write fails, restore the old snapshot and remove `migration`. Then a new
/// run starts from the same state.
fn commit_outputs(
    snapshot_path: &Path,
    snapshot: Option<&SchemaSnapshot>,
    migration: Option<&Path>,
    sync: Option<&(PathBuf, schema_rs::SchemaRsSync)>,
    write_schema_rs: &dyn Fn(&Path, &str) -> std::io::Result<()>,
) -> Result<(), String> {
    let remove_migration = || {
        if let Some(dir) = migration {
            let _ = std::fs::remove_dir_all(dir);
        }
    };
    let old_snapshot = std::fs::read(snapshot_path).ok();
    if let Some(snapshot) = snapshot
        && let Err(e) = snapshot::write_snapshot(snapshot_path, snapshot)
    {
        remove_migration();
        return Err(e.to_string());
    }
    if let Err(e) = apply_schema_rs(sync, write_schema_rs) {
        remove_migration();
        if snapshot.is_some() && restore_file(snapshot_path, old_snapshot.as_deref()).is_err() {
            return Err(format!(
                "{e}; could not restore the snapshot at {} — check it",
                snapshot_path.display()
            ));
        }
        return Err(format!(
            "{e}; the migration and the snapshot were not changed"
        ));
    }
    Ok(())
}

/// The `src/schema.rs` sync for `plan`: the path and the result, when the
/// file exists and the text changes. `None` when the project has no
/// `src/schema.rs` or the text stays the same. Prints the skipped tables of
/// an existing file.
///
/// # Errors
///
/// Returns an error when the file exists but cannot be read. The caller
/// stops before it writes anything.
fn plan_schema_rs(
    project_root: &Path,
    desired: &parse::ParsedSchema,
    plan: &MigrationPlan,
) -> Result<Option<(PathBuf, schema_rs::SchemaRsSync)>, String> {
    let path = project_root.join(schema_rs::SCHEMA_RS_PATH);
    let existing = match std::fs::read_to_string(&path) {
        Ok(existing) => existing,
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
            if desired.tables.iter().any(|t| t.managed) {
                eprintln!(
                    "note: no {} — the diesel schema was not written",
                    schema_rs::SCHEMA_RS_PATH
                );
            }
            return Ok(None);
        }
        Err(e) => return Err(format!("failed to read {}: {e}", path.display())),
    };
    let sync = schema_rs::sync_for_plan(&existing, desired, plan);
    for (table, reason) in &sync.skipped {
        eprintln!(
            "warning: {} — did not write `{table}`: {reason}",
            schema_rs::SCHEMA_RS_PATH
        );
    }
    if sync.text == existing {
        return Ok(None);
    }
    // The sync edits text. Never write a file that does not parse.
    if let Err(reason) = schema_rs::validate(&sync.text) {
        eprintln!(
            "warning: did not update {}: {reason}; edit it by hand",
            schema_rs::SCHEMA_RS_PATH
        );
        return Ok(None);
    }
    Ok(Some((path, sync)))
}

/// Write the `src/schema.rs` sync from [`plan_schema_rs`], if any.
fn apply_schema_rs(
    sync: Option<&(PathBuf, schema_rs::SchemaRsSync)>,
    write: &dyn Fn(&Path, &str) -> std::io::Result<()>,
) -> Result<(), String> {
    let Some((path, sync)) = sync else {
        return Ok(());
    };
    write(path, &sync.text).map_err(|e| format!("failed to write {}: {e}", path.display()))
}

/// Put `path` back to the bytes in `old`. Remove it when `old` is `None`.
fn restore_file(path: &Path, old: Option<&[u8]>) -> std::io::Result<()> {
    old.map_or_else(
        || std::fs::remove_file(path),
        |bytes| std::fs::write(path, bytes),
    )
}

/// Print what the `src/schema.rs` sync changed.
fn report_schema_rs(sync: Option<&(PathBuf, schema_rs::SchemaRsSync)>) {
    let Some((path, sync)) = sync else { return };
    let mut parts = Vec::new();
    if !sync.written.is_empty() {
        parts.push(format!("wrote {}", sync.written.join(", ")));
    }
    if !sync.removed.is_empty() {
        parts.push(format!("removed {}", sync.removed.join(", ")));
    }
    for (old, new) in &sync.renamed {
        parts.push(format!("renamed {old} to {new}"));
    }
    if parts.is_empty() {
        parts.push("updated the table lists".to_owned());
    }
    println!("updated {} ({})", path.display(), parts.join("; "));
}

/// Load the diff snapshot, provider-locked to `backend`. A missing snapshot is
/// a friendly "run `autumn schema snapshot` first" error, except with
/// `--dev-url` (`optional`), where it is `None`.
fn load_diff_snapshot(
    snapshot_path: &Path,
    backend: Backend,
    optional: bool,
) -> Result<Option<SchemaSnapshot>, String> {
    match snapshot::load_snapshot(snapshot_path) {
        Ok(snapshot) => {
            snapshot
                .ensure_backend_matches(backend)
                .map_err(|e| e.to_string())?;
            Ok(Some(snapshot))
        }
        Err(snapshot::SnapshotError::Io { source, .. })
            if source.kind() == std::io::ErrorKind::NotFound =>
        {
            if optional {
                Ok(None)
            } else {
                Err(format!(
                    "no schema snapshot at {} — run `autumn schema snapshot` first to create the diff baseline",
                    snapshot_path.display()
                ))
            }
        }
        Err(other) => Err(other.to_string()),
    }
}

/// The `--dev-url` baseline: the migrations replayed on the dev database, with
/// a warning when they do not match the snapshot, and with managed flags from
/// the snapshot and the models (see [`shadow::adopt_managed_flags`]).
fn replay_baseline(
    project_root: &Path,
    url: &str,
    backend: Backend,
    snapshot: Option<&SchemaSnapshot>,
    desired: &parse::ParsedSchema,
) -> Result<shadow::Replay, String> {
    let mut replayed = shadow::replay(backend, url, &project_root.join("migrations"))?;
    shadow::adopt_model_spelling(&mut replayed.tables, desired);
    if let Some(snapshot) = snapshot {
        warn_snapshot_drift(&snapshot.tables, &replayed.tables);
    }
    shadow::adopt_managed_flags(
        &mut replayed.tables,
        snapshot.map(|s| &s.tables[..]),
        desired,
    );
    Ok(replayed)
}

/// The refusal for a plan target (a new or renamed table or index) whose name
/// is already taken. Tables, indexes, views, sequences and other relations
/// share one namespace, so the migration would fail. Taken names are the
/// baseline's tables and indexes plus `others` (the non-table relations of a
/// `--dev-url` replay), less the names the plan frees first: the old name of a
/// renamed table or index, and an index dropped just before its re-add. Two
/// targets in the plan with one name collide as well.
fn relation_clash(
    plan: &MigrationPlan,
    baseline: &[Table],
    others: &std::collections::BTreeSet<String>,
) -> Option<String> {
    // Names compare case-insensitively: SQLite resolves them that way, and on
    // Postgres a false match only refuses (the safe side).
    let key = |name: &str| name.to_ascii_lowercase();
    let added: std::collections::BTreeSet<String> = plan
        .changes
        .iter()
        .filter_map(|c| match c {
            SchemaChange::AddIndex { index, .. } => Some(key(&index.name)),
            _ => None,
        })
        .collect();
    let freed: std::collections::BTreeSet<String> = plan
        .changes
        .iter()
        .filter_map(|c| match c {
            SchemaChange::RenameTable { from, .. } | SchemaChange::RenameIndex { from, .. } => {
                Some(key(from))
            }
            SchemaChange::DropIndex { index, .. } if added.contains(&key(&index.name)) => {
                Some(key(&index.name))
            }
            _ => None,
        })
        .collect();
    let existing: std::collections::BTreeSet<String> = others
        .iter()
        .map(|n| key(n))
        .chain(baseline.iter().flat_map(|t| {
            std::iter::once(key(&t.name)).chain(t.indexes.iter().map(|i| key(&i.name)))
        }))
        .collect();
    let taken = |name: &str| {
        let k = key(name);
        !freed.contains(&k) && existing.contains(&k)
    };
    // A name that two targets of this plan share collides too.
    let mut seen: std::collections::BTreeSet<String> = std::collections::BTreeSet::new();
    plan.changes.iter().find_map(|c| {
        let names: Vec<&String> = match c {
            // A new table brings its own indexes.
            SchemaChange::CreateTable(t) => std::iter::once(&t.name)
                .chain(t.indexes.iter().map(|i| &i.name))
                .collect(),
            SchemaChange::RenameTable { to, .. } => vec![to],
            SchemaChange::AddIndex { index, .. } | SchemaChange::RenameIndex { index, .. } => {
                vec![&index.name]
            }
            _ => return None,
        };
        let name = names
            .into_iter()
            .find(|n| taken(n) || !seen.insert(key(n)))?;
        Some(format!(
            "`{name}` is already a table, index, view, sequence, type or other relation \
             in the schema; choose another name"
        ))
    })
}

/// Warn on stderr when the snapshot does not match the schema the migrations
/// make (the `--dev-url` replay). The diff still uses the replayed schema.
fn warn_snapshot_drift(snapshot: &[Table], replayed: &[Table]) {
    if let Some(text) = snapshot_drift(snapshot, replayed) {
        eprintln!("warning: the snapshot does not match the migrations.\n{text}");
    }
}

/// The differences between the snapshot and the replayed schema, in both
/// directions, or `None` when they match. Same comparison as doctor's
/// database-schema-drift check.
fn snapshot_drift(snapshot: &[Table], replayed: &[Table]) -> Option<String> {
    let drift = doctor::compute_db_schema_drift(snapshot, replayed);
    if drift.is_clean() {
        return None;
    }
    let mut text = String::new();
    if !drift.forward.is_empty() {
        text.push_str("Changes that turn the snapshot into the migrated schema:\n");
        text.push_str(&diff::describe_plan(&drift.forward));
    }
    if !drift.reverse.is_empty() {
        text.push_str("Changes that turn the migrated schema into the snapshot:\n");
        text.push_str(&diff::describe_plan(&drift.reverse));
    }
    Some(text)
}

/// Project the guarded migration `plan` onto the `baseline` tables to compute the
/// state the database will be in once the migration applies — the new snapshot
/// baseline `schema diff --write-migration` advances to.
///
/// Each [`SchemaChange`] carries the full desired objects/values (they come from
/// the desired state), so applying them onto a clone of the baseline is verbatim
/// and convergent: the result equals what the emitted `up.sql` produces. This is
/// deliberately NOT a snapshot of `desired.tables` wholesale — the guard may have
/// reduced or refused parts of the delta, and the snapshot must track only what
/// the migration actually does. Table order is irrelevant here: `write_snapshot`
/// re-sorts tables by name canonically.
fn project_plan_target(baseline: &[Table], plan: &MigrationPlan) -> Vec<Table> {
    // Renames come first in a plan and depend only on the baseline, so apply
    // them before the name-keyed working set is built.
    let baseline = rename::renamed_baseline(baseline, &plan.changes);
    // Name-keyed working set cloned from the baseline.
    let mut tables: BTreeMap<String, Table> = baseline
        .iter()
        .map(|t| (t.name.clone(), t.clone()))
        .collect();

    // Exhaustive match with NO wildcard arm: a future `SchemaChange` variant
    // forces a compile error here rather than silently mis-projecting the target.
    for change in &plan.changes {
        match change {
            SchemaChange::CreateTable(table) => {
                tables.insert(table.name.clone(), table.clone());
            }
            SchemaChange::DropTable(table) => {
                tables.remove(&table.name);
            }
            SchemaChange::AddColumn { table, column } => {
                if let Some(t) = tables.get_mut(table) {
                    t.columns.push(column.clone());
                }
            }
            SchemaChange::DropColumn { table, column } => {
                if let Some(t) = tables.get_mut(table) {
                    t.columns.retain(|c| c.name != column.name);
                    // Postgres `ALTER TABLE … DROP COLUMN` automatically drops any
                    // index (and the constraint it backs) that depends on the
                    // column. Ordinary model-managed indexes on the column already
                    // carry an explicit `DropIndex` in this plan; this ALSO prunes
                    // the RETAINED (`definition`-carrying) expression / partial /
                    // constraint-owned indexes that have NO `DropIndex` — otherwise
                    // the projected snapshot would claim an index the database just
                    // cascade-dropped and every subsequent `schema doctor` /
                    // `pull --dry-run` would report false, perpetual drift. Uses
                    // the same dependency test the down emitter uses to restore
                    // these indexes on rollback, so projection and rollback agree.
                    t.indexes
                        .retain(|i| !diff::index_depends_on_column(i, &column.name, plan.backend));
                }
            }
            SchemaChange::AlterColumnType {
                table, column, to, ..
            } => {
                if let Some(c) = column_mut(&mut tables, table, column) {
                    c.ty = to.clone();
                }
            }
            SchemaChange::SetNotNull { table, column } => {
                if let Some(c) = column_mut(&mut tables, table, column) {
                    c.nullable = false;
                }
            }
            SchemaChange::DropNotNull { table, column } => {
                if let Some(c) = column_mut(&mut tables, table, column) {
                    c.nullable = true;
                }
            }
            SchemaChange::SetDefault {
                table, column, to, ..
            } => {
                if let Some(c) = column_mut(&mut tables, table, column) {
                    c.default = Some(to.clone());
                }
            }
            SchemaChange::AddForeignKey {
                table,
                column,
                foreign_key,
            } => {
                if let Some(c) = column_mut(&mut tables, table, column) {
                    c.references = Some(foreign_key.clone());
                }
            }
            SchemaChange::AddIndex { table, index } => {
                if let Some(t) = tables.get_mut(table) {
                    t.indexes.push(index.clone());
                }
            }
            SchemaChange::DropIndex { table, index } => {
                if let Some(t) = tables.get_mut(table) {
                    t.indexes.retain(|i| i.name != index.name);
                }
            }
            SchemaChange::AddCheck { table, check } => {
                if let Some(t) = tables.get_mut(table) {
                    t.checks.push(check.clone());
                }
            }
            // The renames were applied above by `renamed_baseline`, and
            // the non-emittable marker variants: `guard_plan` refuses these before
            // we get here (a guarded plan never carries one), so projecting them is
            // a no-op — never a panic.
            SchemaChange::RenameTable { .. }
            | SchemaChange::RenameColumn { .. }
            | SchemaChange::RenameIndex { .. }
            | SchemaChange::RenameConflict { .. }
            | SchemaChange::PrimaryKeyChange { .. }
            | SchemaChange::ForeignKeyChange { .. }
            | SchemaChange::IdentityChange { .. }
            | SchemaChange::DropTableBlockedByInboundFk { .. }
            | SchemaChange::AlterColumnTypeBlockedByFk { .. }
            | SchemaChange::AddForeignKeyToExistingColumn { .. }
            | SchemaChange::CreateTableBlockedBySkippedField { .. } => {}
        }
    }

    tables.into_values().collect()
}

/// Find a mutable column by name within a named table in the working set — the
/// column-facet helper [`project_plan_target`] uses to apply the `Alter*` /
/// `Set*` / `Drop*Null` column changes. Returns `None` when the table or column
/// is absent (a guarded plan never carries such a change, so a miss is a no-op).
fn column_mut<'a>(
    tables: &'a mut BTreeMap<String, Table>,
    table: &str,
    column: &str,
) -> Option<&'a mut Column> {
    tables
        .get_mut(table)
        .and_then(|t| t.columns.iter_mut().find(|c| c.name == column))
}

/// Write a `migrations/<ts>_<suffix>/{up,down}.sql` pair, creating the leaf
/// directory **exclusively**.
///
/// `create_dir_all` on the leaf would silently succeed if the directory already
/// existed and the following `fs::write`s would then OVERWRITE any `up.sql` /
/// `down.sql` already there — and because `timestamp_now` has 1-second
/// resolution, two `--write-migration --name <same>` runs in the same second
/// resolve to the same directory. So the parent `migrations/` is created
/// idempotently but the leaf is created with `create_dir`, which fails with
/// `AlreadyExists` if it is already present; that case is mapped to a clear
/// refusal and **neither** SQL file is written, so an already-generated (and
/// possibly committed or applied) migration is never clobbered.
fn write_migration_dir(
    project_root: &Path,
    ts: &str,
    suffix: &str,
    up: &str,
    down: &str,
) -> Result<std::path::PathBuf, String> {
    write_migration_dir_with(project_root, ts, suffix, up, down, |path, contents| {
        std::fs::write(path, contents)
    })
}

/// Inner implementation of [`write_migration_dir`] that takes the file-write
/// operation as a closure. Extracted so a test can deterministically force a
/// mid-write failure (e.g. the `down.sql` write) and assert the
/// partial-migration cleanup below; production always passes `std::fs::write`.
///
/// Transactional-on-failure: the leaf directory is created **exclusively** (so
/// this call, and only this call, owns it — a pre-existing COMPLETE migration
/// still trips the `AlreadyExists` collision guard and is never touched). Once
/// this call has created the dir, if EITHER SQL write fails the just-created dir
/// is best-effort removed (`remove_dir_all`, its own error ignored) and the
/// ORIGINAL write error is returned, so a partial `migrations/<ts>_<suffix>/`
/// containing only `up.sql` is never left behind to block a same-timestamp
/// retry.
fn write_migration_dir_with<W>(
    project_root: &Path,
    ts: &str,
    suffix: &str,
    up: &str,
    down: &str,
    write: W,
) -> Result<std::path::PathBuf, String>
where
    W: Fn(&Path, &str) -> std::io::Result<()>,
{
    let migrations_dir = project_root.join("migrations");
    let dir = migrations_dir.join(format!("{ts}_{suffix}"));
    std::fs::create_dir_all(&migrations_dir).map_err(|e| {
        format!(
            "failed to create migrations directory {}: {e}",
            migrations_dir.display()
        )
    })?;
    std::fs::create_dir(&dir).map_err(|e| {
        if e.kind() == std::io::ErrorKind::AlreadyExists {
            format!(
                "refusing to overwrite existing migration directory {}; a migration with this \
                 timestamp and name already exists",
                dir.display()
            )
        } else {
            format!(
                "failed to create migration directory {}: {e}",
                dir.display()
            )
        }
    })?;
    // From here on THIS call owns `dir` (exclusive `create_dir` above). If either
    // write fails, remove the dir we just created so no partial migration is left
    // to trip the collision guard on retry, and surface the original write error.
    let write_both = || -> Result<(), String> {
        write(&dir.join("up.sql"), up).map_err(|e| format!("failed to write up.sql: {e}"))?;
        write(&dir.join("down.sql"), down).map_err(|e| format!("failed to write down.sql: {e}"))?;
        Ok(())
    };
    if let Err(e) = write_both() {
        let _ = std::fs::remove_dir_all(&dir);
        return Err(e);
    }
    Ok(dir)
}

/// Parse a file or directory and render the tables as pretty JSON, surfacing any
/// per-field diagnostics on stderr. Split out (returning `Result`) so it is
/// testable without a process exit.
fn run_parse(path: &Path) -> Result<String, String> {
    // The parser IR is backend-tagged; `parse` defaults to Postgres (the fully
    // wired runtime backend). A backend flag is a later-slice concern.
    let backend = Backend::Postgres;
    let parsed = parse_models_path(path, backend).map_err(|e| e.to_string())?;

    for diag in &parsed.diagnostics {
        eprintln!("warning: {}", diag.message);
    }

    serde_json::to_string_pretty(&parsed.tables).map_err(|e| e.to_string())
}

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

    // -- `diff` command wiring (tempdir, explicit project root) --------------

    /// Write a project scaffold: a `Cargo.toml` (so `ensure_project_root`
    /// passes), a single-file `src/models.rs`, and a snapshot at the default
    /// path. Returns the tempdir (kept alive by the caller).
    fn scaffold_project(models: &str, snapshot_json: &str) -> tempfile::TempDir {
        let root = tempfile::tempdir().expect("tempdir");
        std::fs::write(root.path().join("Cargo.toml"), "[package]\nname=\"x\"\n")
            .expect("Cargo.toml");
        let src = root.path().join("src");
        std::fs::create_dir_all(&src).expect("mkdir src");
        std::fs::write(src.join("models.rs"), models).expect("models.rs");
        let snap = root.path().join(".autumn");
        std::fs::create_dir_all(&snap).expect("mkdir .autumn");
        std::fs::write(snap.join("schema-snapshot.json"), snapshot_json).expect("snapshot");
        root
    }

    const POST_MODEL: &str = r#"
        #[autumn_web::model(managed)]
        pub struct Post {
            #[id]
            pub id: i64,
            pub title: String,
        }
    "#;

    /// A version-1 snapshot for a `posts(id, title)` table, matching `POST_MODEL`.
    fn posts_snapshot(backend: &str) -> String {
        format!(
            r#"{{
  "snapshot_version": 1,
  "backend": "{backend}",
  "tables": [
    {{
      "name": "posts",
      "columns": [
        {{ "name": "id", "ty": "Int64", "nullable": false, "primary_key": true, "unique": false, "default": null, "references": null, "serial": "BigSerial" }},
        {{ "name": "title", "ty": "Text", "nullable": false, "primary_key": false, "unique": false, "default": null, "references": null }},
        {{ "name": "created_at", "ty": "Timestamp", "nullable": false, "primary_key": false, "unique": false, "default": "Now", "references": null }}
      ],
      "primary_key": ["id"],
      "indexes": [],
      "checks": [],
      "backend": "{backend}",
      "managed": true
    }}
  ]
}}
"#
        )
    }

    #[test]
    fn diff_backend_mismatch_errors_before_diff() {
        // A Sqlite-tagged snapshot diffed with `--backend pg` must fail with the
        // provider-lock guard BEFORE any parse/diff work.
        let root = scaffold_project(POST_MODEL, &posts_snapshot("Sqlite"));
        let err = diff_at(
            root.path(),
            &DiffArgs {
                backend: Some(BackendArg::Pg),
                ..DiffArgs::default()
            },
        )
        .unwrap_err();
        assert!(
            err.contains("does not match") && err.to_lowercase().contains("backend"),
            "backend mismatch surfaced: {err}"
        );
    }

    #[test]
    fn diff_missing_snapshot_is_friendly_error() {
        let root = tempfile::tempdir().expect("tempdir");
        std::fs::write(root.path().join("Cargo.toml"), "[package]\nname=\"x\"\n")
            .expect("Cargo.toml");
        let src = root.path().join("src");
        std::fs::create_dir_all(&src).expect("mkdir");
        std::fs::write(src.join("models.rs"), POST_MODEL).expect("models.rs");
        // No snapshot file written.
        let err = diff_at(
            root.path(),
            &DiffArgs {
                backend: Some(BackendArg::Pg),
                ..DiffArgs::default()
            },
        )
        .unwrap_err();
        assert!(
            err.contains("autumn schema snapshot"),
            "friendly 'run snapshot first' message: {err}"
        );
    }

    #[test]
    fn diff_no_op_writes_nothing() {
        // Models match the snapshot → no-op → no migration directory AND the
        // checked-in snapshot is left byte-for-byte unchanged (a no-op diff must
        // not advance the baseline).
        let root = scaffold_project(POST_MODEL, &posts_snapshot("Postgres"));
        let snapshot_path = root.path().join(".autumn/schema-snapshot.json");
        let before = std::fs::read_to_string(&snapshot_path).expect("snapshot before");
        diff_at(
            root.path(),
            &DiffArgs {
                backend: Some(BackendArg::Pg),
                write_migration: true,
                ..DiffArgs::default()
            },
        )
        .expect("no-op diff ok");
        assert!(
            !root.path().join("migrations").exists(),
            "a no-op must not create a migrations directory"
        );
        let after = std::fs::read_to_string(&snapshot_path).expect("snapshot after");
        assert_eq!(
            before, after,
            "a no-op diff must leave the snapshot byte-for-byte unchanged"
        );
    }

    #[test]
    fn write_migration_with_renamed_from_renames_and_advances_snapshot() {
        let models = r#"
            #[autumn_web::model(managed)]
            #[renamed_from("posts")]
            pub struct Article {
                #[id]
                pub id: i64,
                #[renamed_from("title")]
                pub headline: String,
            }
        "#;
        let root = scaffold_project(models, &posts_snapshot("Postgres"));
        diff_at(
            root.path(),
            &DiffArgs {
                backend: Some(BackendArg::Pg),
                write_migration: true,
                name: Some("rename_posts"),
                ..DiffArgs::default()
            },
        )
        .expect("a rename needs no --allow-destructive");

        let migrations = root.path().join("migrations");
        let dirs: Vec<_> = std::fs::read_dir(&migrations)
            .expect("read migrations dir")
            .map(|e| e.expect("entry").path())
            .collect();
        assert_eq!(dirs.len(), 1, "{dirs:?}");
        let up = std::fs::read_to_string(dirs[0].join("up.sql")).expect("up.sql");
        assert_eq!(
            up.trim(),
            "ALTER TABLE posts RENAME TO articles;\n\n\
             ALTER TABLE articles RENAME COLUMN title TO headline;"
        );

        let snapshot_path = root.path().join(".autumn/schema-snapshot.json");
        let advanced = snapshot::load_snapshot(&snapshot_path).expect("load snapshot");
        let names: Vec<&str> = advanced.tables.iter().map(|t| t.name.as_str()).collect();
        assert_eq!(names, vec!["articles"]);
        assert!(
            advanced.tables[0]
                .columns
                .iter()
                .any(|c| c.name == "headline")
        );

        // The hints stay in the source; the next diff is a no-op.
        diff_at(
            root.path(),
            &DiffArgs {
                backend: Some(BackendArg::Pg),
                write_migration: true,
                name: Some("again"),
                ..DiffArgs::default()
            },
        )
        .expect("second diff");
        assert_eq!(std::fs::read_dir(&migrations).expect("read").count(), 1);
    }

    #[test]
    fn snapshot_drift_sees_replay_only_facets_both_ways() {
        let mut posts = autumn_schema_core::Table::new("posts", Backend::Postgres);
        posts
            .columns
            .push(Column::new("id", autumn_schema_core::ColumnType::Int64));
        assert_eq!(snapshot_drift(&[posts.clone()], &[posts.clone()]), None);

        // A hand-written CHECK in the migrations that the snapshot lacks.
        let mut replayed = posts.clone();
        replayed.checks.push(autumn_schema_core::CheckConstraint {
            name: Some("id_positive".to_owned()),
            expression: "id > 0".to_owned(),
        });
        let text = snapshot_drift(&[posts.clone()], &[replayed.clone()]).expect("drift");
        assert!(text.contains("id_positive"), "{text}");
        // And the other way: the snapshot has it, the migrations do not.
        let text = snapshot_drift(&[replayed], &[posts]).expect("drift");
        assert!(text.contains("id_positive"), "{text}");
    }

    #[test]
    fn a_new_table_or_rename_on_a_replayed_view_name_is_refused() {
        let others: std::collections::BTreeSet<String> =
            ["report".to_owned(), "posts_id_seq".to_owned()].into();
        let plan = |change| MigrationPlan {
            backend: Backend::Postgres,
            changes: vec![change],
        };
        let create = plan(SchemaChange::CreateTable(autumn_schema_core::Table::new(
            "report",
            Backend::Postgres,
        )));
        assert!(relation_clash(&create, &[], &others).is_some_and(|e| e.contains("report")));
        let rename = plan(SchemaChange::RenameTable {
            from: "old".to_owned(),
            to: "posts_id_seq".to_owned(),
        });
        assert!(relation_clash(&rename, &[], &others).is_some());
        let mut with_index = autumn_schema_core::Table::new("new_table", Backend::Postgres);
        with_index.indexes.push(autumn_schema_core::Index::new(
            "report",
            vec!["id".to_owned()],
            false,
        ));
        assert!(
            relation_clash(&plan(SchemaChange::CreateTable(with_index)), &[], &others).is_some()
        );
        let fine = plan(SchemaChange::RenameTable {
            from: "old".to_owned(),
            to: "new".to_owned(),
        });
        assert_eq!(relation_clash(&fine, &[], &others), None);
    }

    #[test]
    fn a_new_index_named_like_a_baseline_index_on_another_table_is_refused() {
        use autumn_schema_core::{Index, Table};
        let mut audit = Table::new("audit", Backend::Postgres);
        audit
            .indexes
            .push(Index::new("idx_reports_slug", vec!["x".to_owned()], false));
        let baseline = vec![audit, Table::new("reports", Backend::Postgres)];
        let none = std::collections::BTreeSet::new();
        let add = |name: &str| SchemaChange::AddIndex {
            table: "reports".to_owned(),
            index: Index::new(name, vec!["slug".to_owned()], false),
        };
        let plan = |changes| MigrationPlan {
            backend: Backend::Postgres,
            changes,
        };
        let err =
            relation_clash(&plan(vec![add("idx_reports_slug")]), &baseline, &none).expect("clash");
        assert!(
            err.contains("idx_reports_slug") && err.contains("already"),
            "{err}"
        );
        // A table name is taken too.
        assert!(relation_clash(&plan(vec![add("audit")]), &baseline, &none).is_some());
        // A replaced index (dropped before its re-add) frees its name.
        let replaced = plan(vec![
            SchemaChange::DropIndex {
                table: "audit".to_owned(),
                index: Index::new("idx_reports_slug", vec!["x".to_owned()], false),
            },
            add("idx_reports_slug"),
        ]);
        assert_eq!(relation_clash(&replaced, &baseline, &none), None);
        assert_eq!(
            relation_clash(&plan(vec![add("idx_new")]), &baseline, &none),
            None
        );
        // Names match case-insensitively (SQLite resolves them that way).
        let mixed: std::collections::BTreeSet<String> = ["Reports".to_owned()].into();
        let create = plan(vec![SchemaChange::CreateTable(Table::new(
            "reports",
            Backend::Sqlite,
        ))]);
        assert!(relation_clash(&create, &[], &mixed).is_some());
        assert!(relation_clash(&plan(vec![add("IDX_REPORTS_SLUG")]), &baseline, &none).is_some());
        // Two targets in one plan cannot share a name either.
        let twice = plan(vec![
            SchemaChange::RenameIndex {
                table: "reports".to_owned(),
                from: "idx_old".to_owned(),
                index: Index::new("idx_users_mails", vec!["x".to_owned()], false),
            },
            SchemaChange::CreateTable(Table::new("idx_users_mails", Backend::Postgres)),
        ]);
        assert!(
            relation_clash(&twice, &baseline, &none).is_some_and(|e| e.contains("idx_users_mails"))
        );
    }

    #[test]
    fn dev_url_makes_the_snapshot_optional() {
        let root = scaffold_project(POST_MODEL, "unused");
        std::fs::remove_file(root.path().join(".autumn/schema-snapshot.json")).expect("rm");
        let err = diff_at(
            root.path(),
            &DiffArgs {
                backend: Some(BackendArg::Pg),
                dev_url: Some("postgres://user:pw_do_not_leak@127.0.0.1:1/dev"),
                ..DiffArgs::default()
            },
        )
        .unwrap_err();
        assert!(!err.contains("autumn schema snapshot"), "{err}");
        assert!(
            err.contains("127.0.0.1") && !err.contains("pw_do_not_leak"),
            "{err}"
        );
    }

    #[cfg(feature = "sqlite")]
    #[test]
    fn sqlite_dev_url_reads_names_that_differ_only_in_case_as_the_model_names() {
        let root = scaffold_project(POST_MODEL, "unused");
        std::fs::remove_file(root.path().join(".autumn/schema-snapshot.json")).expect("rm");
        let init = root.path().join("migrations/20000101000000_init");
        std::fs::create_dir_all(&init).expect("mkdir");
        std::fs::write(
            init.join("up.sql"),
            "CREATE TABLE \"Posts\" (id INTEGER PRIMARY KEY AUTOINCREMENT, \
             \"Title\" TEXT NOT NULL, created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP);",
        )
        .expect("up.sql");
        std::fs::write(init.join("down.sql"), "DROP TABLE \"Posts\";").expect("down.sql");
        diff_at(
            root.path(),
            &DiffArgs {
                backend: Some(BackendArg::Sqlite),
                dev_url: Some("sqlite::memory:"),
                write_migration: true,
                ..DiffArgs::default()
            },
        )
        .expect("`Posts` and `Title` are the model's `posts` and `title`");
        let written: Vec<String> = std::fs::read_dir(root.path().join("migrations"))
            .expect("migrations")
            .filter_map(|e| {
                let p = e.ok()?.path();
                (!p.ends_with("20000101000000_init"))
                    .then(|| std::fs::read_to_string(p.join("up.sql")).unwrap_or_default())
            })
            .collect();
        assert!(written.is_empty(), "no new migration: {written:?}");
    }

    #[test]
    fn write_migration_creates_up_and_down() {
        // Add a `body` column to the model that the snapshot lacks → one AddColumn.
        let models = r#"
            #[autumn_web::model(managed)]
            pub struct Post {
                #[id]
                pub id: i64,
                pub title: String,
                pub body: Option<String>,
            }
        "#;
        let root = scaffold_project(models, &posts_snapshot("Postgres"));
        diff_at(
            root.path(),
            &DiffArgs {
                backend: Some(BackendArg::Pg),
                write_migration: true,
                name: Some("add_body"),
                ..DiffArgs::default()
            },
        )
        .expect("write migration ok");

        // Exactly one migration directory named `<ts>_add_body` with the pair.
        let migrations = root.path().join("migrations");
        let entries: Vec<_> = std::fs::read_dir(&migrations)
            .expect("read migrations dir")
            .map(|e| e.expect("entry").file_name().to_string_lossy().into_owned())
            .collect();
        assert_eq!(entries.len(), 1, "one migration dir: {entries:?}");
        assert!(
            entries[0].ends_with("_add_body"),
            "named suffix: {entries:?}"
        );
        let dir = migrations.join(&entries[0]);
        let up = std::fs::read_to_string(dir.join("up.sql")).expect("up.sql");
        let down = std::fs::read_to_string(dir.join("down.sql")).expect("down.sql");
        assert!(
            up.contains("ALTER TABLE posts ADD COLUMN body TEXT NULL;"),
            "up: {up}"
        );
        assert!(
            down.contains("ALTER TABLE posts DROP COLUMN body;"),
            "down: {down}"
        );

        // The checked-in snapshot ADVANCED to the plan's target state: it now
        // carries the `body` column the migration adds. Load it back and check.
        let snapshot_path = root.path().join(".autumn/schema-snapshot.json");
        let advanced = snapshot::load_snapshot(&snapshot_path).expect("load advanced snapshot");
        let posts = advanced
            .tables
            .iter()
            .find(|t| t.name == "posts")
            .expect("posts table in advanced snapshot");
        assert!(
            posts.columns.iter().any(|c| c.name == "body"),
            "advanced snapshot must contain the new `body` column: {:?}",
            posts.columns.iter().map(|c| &c.name).collect::<Vec<_>>()
        );
    }

    // The migration files and the checked-in snapshot must advance together
    // (Codex P1): if the snapshot write fails after `write_migration_dir` has
    // already left a complete migration on disk, the baseline never advances, so
    // retrying `schema diff --write-migration` regenerates the same SQL as a
    // SECOND migration and applying both fails on duplicate DDL. `diff_at` rolls
    // the migration dir back on a snapshot-write failure. Reproduce that failure
    // portably by making the snapshot's parent dir (`.autumn`) read-only so the
    // earlier snapshot LOAD still succeeds but `write_snapshot`'s tempfile
    // create+rename cannot, then assert the command errors and left no migration.
    #[cfg(unix)]
    #[test]
    fn write_migration_rolls_back_on_snapshot_write_failure() {
        use std::os::unix::fs::PermissionsExt;

        // Add a `body` column the snapshot lacks → a non-empty plan that would
        // write a migration and then advance the snapshot.
        let models = r#"
            #[autumn_web::model(managed)]
            pub struct Post {
                #[id]
                pub id: i64,
                pub title: String,
                pub body: Option<String>,
            }
        "#;
        let root = scaffold_project(models, &posts_snapshot("Postgres"));
        let autumn_dir = root.path().join(".autumn");

        // Make the snapshot's parent directory read-only so a new tempfile cannot
        // be created in it (the earlier read of the existing snapshot still works).
        std::fs::set_permissions(&autumn_dir, std::fs::Permissions::from_mode(0o555))
            .expect("chmod .autumn read-only");

        // Skipped when the process can bypass the permission bits (e.g. running as
        // root, where a read-only dir is still writable), which would otherwise
        // make this non-deterministic — mirrors `classify_unreadable_dir_is_an_error`.
        let can_still_write = tempfile::NamedTempFile::new_in(&autumn_dir).is_ok();
        if can_still_write {
            std::fs::set_permissions(&autumn_dir, std::fs::Permissions::from_mode(0o755))
                .expect("restore chmod");
            return;
        }

        let result = diff_at(
            root.path(),
            &DiffArgs {
                backend: Some(BackendArg::Pg),
                write_migration: true,
                name: Some("add_body"),
                ..DiffArgs::default()
            },
        );

        // Restore permissions so the tempdir can be cleaned up (and the migrations
        // assertion below can read the dir regardless of outcome).
        std::fs::set_permissions(&autumn_dir, std::fs::Permissions::from_mode(0o755))
            .expect("restore chmod");

        // (a) The command surfaced the snapshot-write failure as an error.
        assert!(
            result.is_err(),
            "a snapshot-write failure must surface as an error, got: {result:?}"
        );

        // (b) The just-created migration was rolled back: no migration directory
        // is left behind, so a retry regenerates a single migration (not a second).
        let migrations = root.path().join("migrations");
        let leftover: Vec<_> = std::fs::read_dir(&migrations)
            .map(|rd| {
                rd.map(|e| e.expect("entry").file_name().to_string_lossy().into_owned())
                    .collect()
            })
            .unwrap_or_default();
        assert!(
            leftover.is_empty(),
            "the migration dir must be rolled back on a snapshot-write failure, found: {leftover:?}"
        );
    }

    /// The core drift scenario from #2041: `schema diff --write-migration`
    /// advances the snapshot to the GENERATED plan's target, so a later,
    /// still-ungenerated model edit diffs on top of it (only the *new* delta),
    /// never re-generating the already-generated change.
    #[test]
    #[allow(clippy::too_many_lines)] // linear scenario reads clearest end-to-end
    fn write_migration_advances_snapshot_so_later_edits_diff_on_top() {
        // Change A: add `body`. Baseline is the posts(id,title,created_at) snapshot.
        let models_a = r#"
            #[autumn_web::model(managed)]
            pub struct Post {
                #[id]
                pub id: i64,
                pub title: String,
                pub body: Option<String>,
            }
        "#;
        let root = scaffold_project(models_a, &posts_snapshot("Postgres"));
        let snapshot_path = root.path().join(".autumn/schema-snapshot.json");

        // Generate change A. The snapshot advances to include `body`.
        diff_at(
            root.path(),
            &DiffArgs {
                backend: Some(BackendArg::Pg),
                write_migration: true,
                name: Some("add_body"),
                ..DiffArgs::default()
            },
        )
        .expect("write migration A ok");

        let after_a = snapshot::load_snapshot(&snapshot_path).expect("load snapshot after A");
        let posts_a = after_a
            .tables
            .iter()
            .find(|t| t.name == "posts")
            .expect("posts after A");
        assert!(
            posts_a.columns.iter().any(|c| c.name == "body"),
            "snapshot advanced to include `body`"
        );
        // The not-yet-added `draft` column is NOT in the snapshot yet.
        assert!(
            !posts_a.columns.iter().any(|c| c.name == "draft"),
            "snapshot must not contain the not-yet-added `draft` column"
        );

        // Convergence: re-running with the SAME models is now a no-op — no second
        // migration directory is written (the snapshot already matches).
        diff_at(
            root.path(),
            &DiffArgs {
                backend: Some(BackendArg::Pg),
                write_migration: true,
                name: Some("add_body_again"),
                ..DiffArgs::default()
            },
        )
        .expect("second (converged) diff ok");
        let dir_count = std::fs::read_dir(root.path().join("migrations"))
            .expect("read migrations")
            .count();
        assert_eq!(
            dir_count, 1,
            "a converged re-run must not write a second migration"
        );

        // Now edit the models to add `draft` on top of the already-generated
        // `body`. Because the snapshot advanced to change A, the next diff
        // generates ONLY `draft` — proving the baseline moved to the plan target,
        // not that it stayed at the original baseline.
        std::fs::write(
            root.path().join("src").join("models.rs"),
            r#"
                #[autumn_web::model(managed)]
                pub struct Post {
                    #[id]
                    pub id: i64,
                    pub title: String,
                    pub body: Option<String>,
                    pub draft: Option<String>,
                }
            "#,
        )
        .expect("edit models to add draft");

        diff_at(
            root.path(),
            &DiffArgs {
                backend: Some(BackendArg::Pg),
                write_migration: true,
                name: Some("add_draft"),
                ..DiffArgs::default()
            },
        )
        .expect("write migration B (draft) ok");

        // Exactly two migration dirs; the newest carries `draft` but NOT `body`.
        let mut names: Vec<String> = std::fs::read_dir(root.path().join("migrations"))
            .expect("read migrations")
            .map(|e| e.expect("entry").file_name().to_string_lossy().into_owned())
            .collect();
        names.sort();
        assert_eq!(names.len(), 2, "two migration dirs now: {names:?}");
        let draft_dir = names
            .iter()
            .find(|n| n.ends_with("_add_draft"))
            .expect("add_draft dir");
        let up = std::fs::read_to_string(
            root.path()
                .join("migrations")
                .join(draft_dir)
                .join("up.sql"),
        )
        .expect("draft up.sql");
        assert!(
            up.contains("draft"),
            "the second migration still generates `draft`: {up}"
        );
        assert!(
            !up.contains("body"),
            "the second migration must NOT re-generate the already-added `body`: {up}"
        );
    }

    // -- `src/schema.rs` sync (Decision 1) -----------------------------------

    /// The `posts` block that `POST_MODEL` renders.
    const POSTS_BLOCK: &str = "diesel::table! {
    posts (id) {
        id -> Int8,
        title -> Text,
        created_at -> Timestamp,
    }
}
";

    fn write_schema_rs(root: &Path, text: &str) -> PathBuf {
        let path = root.join(schema_rs::SCHEMA_RS_PATH);
        std::fs::write(&path, text).expect("schema.rs");
        path
    }

    fn write_args<'a>() -> DiffArgs<'a> {
        DiffArgs {
            backend: Some(BackendArg::Pg),
            write_migration: true,
            ..DiffArgs::default()
        }
    }

    #[test]
    fn write_migration_writes_the_new_column_to_schema_rs() {
        let models = r#"
            #[autumn_web::model(managed)]
            pub struct Post {
                #[id]
                pub id: i64,
                pub title: String,
                pub body: Option<String>,
            }
        "#;
        let root = scaffold_project(models, &posts_snapshot("Postgres"));
        let path = write_schema_rs(root.path(), POSTS_BLOCK);
        diff_at(root.path(), &write_args()).expect("write migration ok");
        let schema = std::fs::read_to_string(path).expect("schema.rs");
        assert!(
            schema.contains("        body -> Nullable<Text>,\n"),
            "{schema}"
        );
    }

    #[test]
    fn write_migration_removes_the_block_of_a_dropped_table() {
        let root = scaffold_project("", &posts_snapshot("Postgres"));
        let path = write_schema_rs(
            root.path(),
            &format!(
                "{POSTS_BLOCK}\ndiesel::table! {{\n    users (id) {{\n        id -> Int8,\n    }}\n}}\n"
            ),
        );
        diff_at(
            root.path(),
            &DiffArgs {
                allow_destructive: true,
                ..write_args()
            },
        )
        .expect("drop ok");
        let schema = std::fs::read_to_string(path).expect("schema.rs");
        assert!(!schema.contains("posts"), "{schema}");
        assert!(schema.contains("users (id)"), "{schema}");
    }

    #[test]
    fn a_no_op_write_migration_still_syncs_a_stale_schema_rs() {
        let root = scaffold_project(POST_MODEL, &posts_snapshot("Postgres"));
        let snapshot_path = root.path().join(SNAPSHOT_DEFAULT_PATH);
        let before = std::fs::read_to_string(&snapshot_path).expect("snapshot");
        let path = write_schema_rs(
            root.path(),
            "diesel::table! {\n    posts (id) {\n        id -> Int8,\n    }\n}\n",
        );
        diff_at(root.path(), &write_args()).expect("no-op ok");
        assert_eq!(
            std::fs::read_to_string(path).expect("schema.rs"),
            POSTS_BLOCK
        );
        assert!(!root.path().join("migrations").exists());
        assert_eq!(std::fs::read_to_string(&snapshot_path).unwrap(), before);
    }

    #[test]
    fn a_preview_diff_and_a_missing_schema_rs_write_no_schema_rs() {
        let models = r#"
            #[autumn_web::model(managed)]
            pub struct Post {
                #[id]
                pub id: i64,
                pub title: String,
                pub body: Option<String>,
            }
        "#;
        let root = scaffold_project(models, &posts_snapshot("Postgres"));
        let path = write_schema_rs(root.path(), POSTS_BLOCK);
        diff_at(
            root.path(),
            &DiffArgs {
                backend: Some(BackendArg::Pg),
                ..DiffArgs::default()
            },
        )
        .expect("preview ok");
        assert_eq!(std::fs::read_to_string(&path).unwrap(), POSTS_BLOCK);

        // No `src/schema.rs`: the command does not make one.
        std::fs::remove_file(&path).unwrap();
        diff_at(root.path(), &write_args()).expect("write ok");
        assert!(!path.exists());
    }

    /// Adding `managed` to a model records it in the snapshot, so a later
    /// delete of the model drops the table.
    #[test]
    fn adopting_a_model_records_it_as_managed_in_the_snapshot() {
        let root = scaffold_project(
            POST_MODEL,
            &posts_snapshot("Postgres").replace("\"managed\": true", "\"managed\": false"),
        );
        diff_at(root.path(), &write_args()).expect("adopt ok");
        let snapshot =
            snapshot::load_snapshot(&root.path().join(SNAPSHOT_DEFAULT_PATH)).expect("load");
        assert!(snapshot.tables[0].managed, "{:?}", snapshot.tables[0]);
        assert!(!root.path().join("migrations").exists());
    }

    /// An edit to the macros alone (the dropped table has no block) is
    /// written.
    #[test]
    fn a_macro_only_schema_rs_change_is_written() {
        let root = scaffold_project("", &posts_snapshot("Postgres"));
        let path = write_schema_rs(
            root.path(),
            "diesel::allow_tables_to_appear_in_same_query!(posts, users);\n",
        );
        diff_at(
            root.path(),
            &DiffArgs {
                allow_destructive: true,
                ..write_args()
            },
        )
        .expect("drop ok");
        assert_eq!(std::fs::read_to_string(path).unwrap(), "");
    }

    /// A `src/schema.rs` that exists but cannot be read stops the command
    /// before it writes anything.
    #[cfg(unix)]
    #[test]
    fn an_unreadable_schema_rs_stops_the_command_before_any_write() {
        let models = r#"
            #[autumn_web::model(managed)]
            pub struct Post {
                #[id]
                pub id: i64,
                pub title: String,
                pub body: Option<String>,
            }
        "#;
        let root = scaffold_project(models, &posts_snapshot("Postgres"));
        let snapshot_path = root.path().join(SNAPSHOT_DEFAULT_PATH);
        let before = std::fs::read_to_string(&snapshot_path).expect("snapshot");
        // A directory at the path: reading it fails with an error that is
        // not "not found", also as root.
        std::fs::create_dir_all(root.path().join(schema_rs::SCHEMA_RS_PATH)).unwrap();
        let err = diff_at(root.path(), &write_args()).unwrap_err();
        assert!(err.contains("schema.rs"), "{err}");
        assert!(!root.path().join("migrations").exists());
        assert_eq!(std::fs::read_to_string(&snapshot_path).unwrap(), before);
    }

    /// A `src/schema.rs` the sync cannot edit into valid Rust is left as is;
    /// the migration and the snapshot still advance.
    #[test]
    fn an_update_that_would_not_parse_is_not_written() {
        let models = r#"
            #[autumn_web::model(managed)]
            pub struct Post {
                #[id]
                pub id: i64,
                pub title: String,
                pub body: Option<String>,
            }
        "#;
        let root = scaffold_project(models, &posts_snapshot("Postgres"));
        let broken = "diesel::table! {\n    posts (id) {\n        id -> Int8,\n";
        let path = write_schema_rs(root.path(), broken);
        diff_at(root.path(), &write_args()).expect("write ok");
        assert_eq!(std::fs::read_to_string(path).unwrap(), broken);
        assert!(root.path().join("migrations").exists());
    }

    /// `--dev-url --write-migration` with no snapshot and no SQL change
    /// still creates the snapshot from the replayed baseline, with the new
    /// managed flags.
    #[test]
    fn an_empty_dev_url_write_creates_the_snapshot() {
        let root = tempfile::tempdir().expect("tempdir");
        let snapshot_path = root.path().join(SNAPSHOT_DEFAULT_PATH);
        let desired = parse::parse_model_source(POST_MODEL, Backend::Postgres).expect("parse");
        let mut baseline = desired.tables.clone();
        baseline[0].managed = false;
        let adopted = adopted_snapshot(None, &baseline, Backend::Postgres, &desired);
        commit_empty_plan(&snapshot_path, adopted.as_ref(), None, &|_, _| Ok(())).expect("commit");
        let written = snapshot::load_snapshot(&snapshot_path).expect("snapshot");
        assert!(written.tables[0].managed, "{:?}", written.tables);
    }

    /// A failed `schema.rs` write leaves the migration and the snapshot as
    /// they were before the command.
    #[test]
    fn write_migration_rolls_back_on_schema_rs_write_failure() {
        let models = r#"
            #[autumn_web::model(managed)]
            pub struct Post {
                #[id]
                pub id: i64,
                pub title: String,
                pub body: Option<String>,
            }
        "#;
        let root = scaffold_project(models, &posts_snapshot("Postgres"));
        let snapshot_path = root.path().join(SNAPSHOT_DEFAULT_PATH);
        let before = std::fs::read_to_string(&snapshot_path).expect("snapshot");
        let path = write_schema_rs(root.path(), POSTS_BLOCK);
        let err = diff_at_with(root.path(), &write_args(), &|_, _| {
            Err(std::io::Error::other("disk full"))
        })
        .unwrap_err();
        assert!(
            err.contains("schema.rs") && err.contains("disk full"),
            "{err}"
        );
        let leftover: Vec<_> = std::fs::read_dir(root.path().join("migrations"))
            .map(|rd| rd.flatten().map(|e| e.file_name()).collect())
            .unwrap_or_default();
        assert!(leftover.is_empty(), "{leftover:?}");
        assert_eq!(std::fs::read_to_string(&snapshot_path).unwrap(), before);
        assert_eq!(std::fs::read_to_string(&path).unwrap(), POSTS_BLOCK);
    }

    /// Round 3 / Finding X: a second `--write-migration` resolving to an
    /// already-existing `migrations/<ts>_<suffix>/` directory (two runs in the
    /// same 1-second-resolution timestamp) is REFUSED rather than silently
    /// overwriting the first migration's SQL. Drives `write_migration_dir`
    /// directly with a fixed timestamp so the collision is deterministic.
    #[test]
    fn write_migration_refuses_to_overwrite_existing_dir() {
        let root = tempfile::tempdir().expect("tempdir");
        let dir = write_migration_dir(
            root.path(),
            "20260101120000",
            "add_body",
            "UP-ONE",
            "DOWN-ONE",
        )
        .expect("first write ok");
        assert_eq!(
            std::fs::read_to_string(dir.join("up.sql")).expect("up.sql"),
            "UP-ONE"
        );

        // A second write to the same <ts>_<suffix> is refused with a clear error.
        let err = write_migration_dir(
            root.path(),
            "20260101120000",
            "add_body",
            "UP-TWO",
            "DOWN-TWO",
        )
        .expect_err("second write to the same dir must be refused");
        assert!(
            err.contains("refusing to overwrite") && err.contains("add_body"),
            "clear refusal naming the colliding dir: {err}"
        );

        // The first migration's contents are left intact (not clobbered).
        assert_eq!(
            std::fs::read_to_string(dir.join("up.sql")).expect("up.sql"),
            "UP-ONE",
            "first up.sql untouched"
        );
        assert_eq!(
            std::fs::read_to_string(dir.join("down.sql")).expect("down.sql"),
            "DOWN-ONE",
            "first down.sql untouched"
        );
    }

    /// Round 9 / P2: if the `up.sql` write succeeds but the `down.sql` write
    /// fails (disk-full / I/O error), the just-created leaf directory must be
    /// removed so no partial migration is left behind — otherwise a retry with
    /// the same timestamp+name would hit the exclusive-create collision guard
    /// and be wrongly refused. Injects a `write` closure that fails only on
    /// `down.sql` to force the failure deterministically, then asserts the dir
    /// is gone and a clean retry succeeds.
    #[test]
    fn write_migration_cleans_up_partial_dir_on_write_failure() {
        let root = tempfile::tempdir().expect("tempdir");
        let leaf = root
            .path()
            .join("migrations")
            .join("20260101120000_add_body");

        let err = write_migration_dir_with(
            root.path(),
            "20260101120000",
            "add_body",
            "UP",
            "DOWN",
            |path, contents| {
                if path.file_name().and_then(|n| n.to_str()) == Some("down.sql") {
                    Err(std::io::Error::other("simulated disk full"))
                } else {
                    std::fs::write(path, contents)
                }
            },
        )
        .expect_err("down.sql write failure must propagate");
        assert!(
            err.contains("down.sql") && err.contains("simulated disk full"),
            "original write error is surfaced: {err}"
        );

        // The partially-written dir (which had only up.sql) was cleaned up.
        assert!(
            !leaf.exists(),
            "partial migration dir must be removed on write failure"
        );

        // A retry with the SAME timestamp+name now succeeds: the collision guard
        // sees no leftover dir, so the user is not stuck needing manual cleanup.
        let dir = write_migration_dir(root.path(), "20260101120000", "add_body", "UP", "DOWN")
            .expect("retry after cleanup succeeds");
        assert_eq!(
            std::fs::read_to_string(dir.join("up.sql")).expect("up.sql"),
            "UP"
        );
        assert_eq!(
            std::fs::read_to_string(dir.join("down.sql")).expect("down.sql"),
            "DOWN"
        );
    }

    #[test]
    fn diff_destructive_refused_without_flag_then_allowed() {
        // Snapshot has a `title` the model dropped → DropColumn → refused by default.
        let models = r#"
            #[autumn_web::model(managed)]
            pub struct Post {
                #[id]
                pub id: i64,
            }
        "#;
        let root = scaffold_project(models, &posts_snapshot("Postgres"));
        let err = diff_at(
            root.path(),
            &DiffArgs {
                backend: Some(BackendArg::Pg),
                ..DiffArgs::default()
            },
        )
        .unwrap_err();
        assert!(
            err.contains("--allow-destructive"),
            "refusal names the flag: {err}"
        );

        // With the flag it succeeds (prints the plan).
        diff_at(
            root.path(),
            &DiffArgs {
                backend: Some(BackendArg::Pg),
                allow_destructive: true,
                ..DiffArgs::default()
            },
        )
        .expect("allowed with --allow-destructive");
    }

    #[test]
    fn default_resolver_prefers_src_models_dir() {
        let root = tempfile::tempdir().expect("tempdir");
        let dir = root.path().join("src").join("models");
        std::fs::create_dir_all(&dir).expect("mkdir");
        // A `src/models.rs` file also present must not win over the directory.
        std::fs::write(root.path().join("src").join("models.rs"), "").expect("write file");

        let resolved = resolve_default_models_path(root.path()).expect("resolve");
        assert_eq!(resolved, dir);
    }

    #[test]
    fn default_resolver_falls_back_to_single_file() {
        let root = tempfile::tempdir().expect("tempdir");
        let src = root.path().join("src");
        std::fs::create_dir_all(&src).expect("mkdir");
        let file = src.join("models.rs");
        std::fs::write(&file, "").expect("write file");

        let resolved = resolve_default_models_path(root.path()).expect("resolve");
        assert_eq!(resolved, file);
    }

    #[test]
    fn fully_explicit_invocation_needs_no_project_root() {
        let from = Path::new("models.rs");
        let out = Path::new("snap.json");
        // `--from` + `--out` + `--backend`: nothing is defaulted, so the command
        // can run from any directory.
        assert!(!needs_project_root(
            Some(from),
            Some(out),
            false,
            Some(BackendArg::Sqlite)
        ));
        // `--from` + `--stdout` + `--backend` likewise.
        assert!(!needs_project_root(
            Some(from),
            None,
            true,
            Some(BackendArg::Sqlite)
        ));
    }

    #[test]
    fn missing_backend_requires_project_root_even_with_explicit_paths() {
        let from = Path::new("models.rs");
        let out = Path::new("snap.json");
        // Explicit `--from` and `--out`/`--stdout`, but no `--backend`: detection
        // needs the root's config, so the root is still required.
        assert!(needs_project_root(Some(from), Some(out), false, None));
        assert!(needs_project_root(Some(from), None, true, None));
    }

    #[test]
    fn default_paths_require_project_root() {
        let from = Path::new("models.rs");
        let out = Path::new("snap.json");
        // Missing `--from` defaults the source to `src/models`.
        assert!(needs_project_root(
            None,
            Some(out),
            false,
            Some(BackendArg::Sqlite)
        ));
        // Missing `--out` while not writing to stdout defaults the output file.
        assert!(needs_project_root(
            Some(from),
            None,
            false,
            Some(BackendArg::Sqlite)
        ));
        // Bare `snapshot` (everything defaulted) certainly needs the root.
        assert!(needs_project_root(None, None, false, None));
    }

    #[test]
    fn root_required_error_names_backend_when_detection_is_a_reason() {
        let msg = project_root_required_error(true);
        assert!(
            msg.contains("--backend"),
            "guides the user to --backend: {msg}"
        );
        assert!(
            msg.contains("auto-detect"),
            "explains the auto-detection failure: {msg}"
        );
    }

    #[test]
    fn root_required_error_is_generic_when_backend_is_explicit() {
        // Backend was given, so only a default path forced the requirement — the
        // generic project-root message is used (no misleading --backend hint).
        let msg = project_root_required_error(false);
        assert!(!msg.contains("--backend"), "no --backend hint: {msg}");
        assert!(
            msg.contains("Autumn project"),
            "the generic project-root message: {msg}"
        );
    }

    /// A minimal `clap::Parser` wrapper so the `SchemaAction` subcommand args
    /// (a `Subcommand`, not a `Parser`) can be exercised at parse time.
    #[derive(clap::Parser, Debug)]
    struct TestCli {
        #[command(subcommand)]
        action: SchemaAction,
    }

    #[test]
    fn out_and_stdout_are_mutually_exclusive() {
        use clap::Parser;
        // clap rejects `--out` together with `--stdout` before `run_snapshot`
        // ever runs, so the misleading "printed but `--out` file unwritten" case
        // is impossible.
        let err = TestCli::try_parse_from([
            "autumn",
            "snapshot",
            "--from",
            "models.rs",
            "--out",
            "snap.json",
            "--stdout",
        ])
        .unwrap_err();
        assert_eq!(err.kind(), clap::error::ErrorKind::ArgumentConflict);

        // Each mode alone still parses.
        assert!(
            TestCli::try_parse_from(["autumn", "snapshot", "--from", "models.rs", "--stdout"])
                .is_ok()
        );
        assert!(
            TestCli::try_parse_from([
                "autumn",
                "snapshot",
                "--from",
                "models.rs",
                "--out",
                "snap.json"
            ])
            .is_ok()
        );
    }

    /// Findings 2 & 3 (#2036): the schema-command backend derives from the
    /// profile-resolved URL when one is configured (its scheme is authoritative),
    /// and only falls back to the project default when no URL is present — so
    /// `--profile <name>` picks the apply path / provider-lock of the database it
    /// acts against, not the ambient project default.
    #[test]
    fn backend_for_url_prefers_url_scheme_over_project_default() {
        // A tempdir with no config → the project default is Postgres.
        let root = tempfile::tempdir().expect("tempdir");
        // A configured SQLite URL is authoritative regardless of that default.
        assert_eq!(
            backend_for_url(root.path(), None, Some("sqlite://app.db")),
            Backend::Sqlite
        );
        // A configured Postgres URL likewise.
        assert_eq!(
            backend_for_url(root.path(), None, Some("postgres://u@h/db")),
            Backend::Postgres
        );
        // No URL → falls back to the profile-aware project default (Postgres here).
        assert_eq!(backend_for_url(root.path(), None, None), Backend::Postgres);
    }

    #[test]
    fn existing_models_path_prefers_dir_then_file_then_none() {
        let root = tempfile::tempdir().expect("tempdir");
        assert!(existing_models_path(root.path()).is_none());

        let src = root.path().join("src");
        std::fs::create_dir_all(&src).expect("mkdir src");
        let file = src.join("models.rs");
        std::fs::write(&file, "").expect("write file");
        assert_eq!(existing_models_path(root.path()), Some(file));

        let dir = src.join("models");
        std::fs::create_dir_all(&dir).expect("mkdir models");
        assert_eq!(existing_models_path(root.path()), Some(dir));
    }

    #[test]
    fn load_existing_snapshot_absent_and_corrupt_are_none_valid_is_some() {
        let root = tempfile::tempdir().expect("tempdir");
        let path = root.path().join("snap.json");
        // Missing file → treated as absent (fresh baseline).
        assert!(load_existing_snapshot(&path).is_none());
        // Corrupt/unreadable-as-absent → also None (a pull re-establishes it).
        std::fs::write(&path, "{ not json").expect("write corrupt");
        assert!(load_existing_snapshot(&path).is_none());
        // A valid snapshot loads and surfaces its backend to the provider-lock.
        let snap = SchemaSnapshot::new(Backend::Sqlite, Vec::new());
        snapshot::write_snapshot(&path, &snap).expect("write valid");
        let loaded = load_existing_snapshot(&path).expect("some");
        assert_eq!(loaded.backend, Backend::Sqlite);
    }

    #[test]
    fn default_resolver_errors_when_neither_exists() {
        let root = tempfile::tempdir().expect("tempdir");
        std::fs::create_dir_all(root.path().join("src")).expect("mkdir");

        let err = resolve_default_models_path(root.path()).unwrap_err();
        assert!(
            err.contains("--from"),
            "error tells the user to pass --from"
        );
    }

    // -- Part A: `project_plan_target` prunes column-dependent retained indexes --

    /// A `posts(id, email)` table carrying a RETAINED expression/constraint index
    /// on `email` (a `definition`-carrying index with NO paired `DropIndex`, the
    /// shape `schema pull` preserves for a brownfield `UNIQUE`/expression index).
    fn users_table_with_email_unique_index() -> Table {
        use autumn_schema_core::{ColumnType, Index};
        let mut id = Column::new("id".to_string(), ColumnType::Int64);
        id.primary_key = true;
        let email = Column::new("email".to_string(), ColumnType::Text);
        let mut table = Table::new("users", Backend::Postgres);
        table.managed = true;
        table.primary_key = vec!["id".to_string()];
        table.columns = vec![id, email];
        table.indexes = vec![Index {
            name: "users_email_key".to_string(),
            columns: vec!["email".to_string()],
            unique: true,
            // A constraint-owned unique index, retained verbatim by `schema pull`.
            definition: Some("CREATE UNIQUE INDEX users_email_key ON users (email)".to_string()),
            is_partial: false,
            key_columns: Vec::new(),
        }];
        table
    }

    fn drop_email_plan() -> MigrationPlan {
        use autumn_schema_core::{Column, ColumnType};
        MigrationPlan {
            backend: Backend::Postgres,
            changes: vec![SchemaChange::DropColumn {
                table: "users".to_string(),
                column: Column::new("email".to_string(), ColumnType::Text),
            }],
        }
    }

    #[test]
    fn project_plan_target_prunes_retained_index_depending_on_dropped_column() {
        // Dropping `email` cascade-drops the retained unique index in Postgres, so
        // the projected snapshot must contain NEITHER the column NOR the index —
        // otherwise `doctor` / `pull --dry-run` would report false perpetual drift.
        let baseline = vec![users_table_with_email_unique_index()];
        let projected = project_plan_target(&baseline, &drop_email_plan());
        let users = projected
            .iter()
            .find(|t| t.name == "users")
            .expect("users table retained");
        assert!(
            users.columns.iter().all(|c| c.name != "email"),
            "dropped column must be gone: {:?}",
            users.columns
        );
        assert!(
            users.indexes.is_empty(),
            "retained index depending on the dropped column must be pruned: {:?}",
            users.indexes
        );
    }

    #[test]
    fn project_plan_target_keeps_retained_index_on_unrelated_column() {
        use autumn_schema_core::{Column, ColumnType, Index};
        // A retained expression index on `handle`; a DropColumn of the UNRELATED
        // `email` must leave it intact (no over-pruning).
        let mut table = Table::new("users", Backend::Postgres);
        table.managed = true;
        let mut id = Column::new("id".to_string(), ColumnType::Int64);
        id.primary_key = true;
        table.primary_key = vec!["id".to_string()];
        table.columns = vec![
            id,
            Column::new("email".to_string(), ColumnType::Text),
            Column::new("handle".to_string(), ColumnType::Text),
        ];
        table.indexes = vec![Index {
            name: "users_lower_handle_idx".to_string(),
            columns: vec!["handle".to_string()],
            unique: false,
            definition: Some(
                "CREATE INDEX users_lower_handle_idx ON users (lower(handle))".to_string(),
            ),
            is_partial: false,
            key_columns: Vec::new(),
        }];
        let projected = project_plan_target(&[table], &drop_email_plan());
        let users = projected
            .iter()
            .find(|t| t.name == "users")
            .expect("users table");
        assert_eq!(
            users.indexes.len(),
            1,
            "an index on an unrelated column must survive: {:?}",
            users.indexes
        );
        assert_eq!(users.indexes[0].name, "users_lower_handle_idx");
    }

    #[test]
    fn project_plan_target_false_drift_regression_is_clean() {
        // Snapshot = the projected target (post-Part-A). The live DB (like
        // Postgres) also lacks the cascade-dropped column AND index. The
        // authoritative drift check must then report Clean — proving the false
        // perpetual drift is gone.
        let baseline = vec![users_table_with_email_unique_index()];
        let snapshot = project_plan_target(&baseline, &drop_email_plan());

        // The database after `ALTER TABLE users DROP COLUMN email` cascades:
        // no `email` column, no dependent index.
        let mut db_users = users_table_with_email_unique_index();
        db_users.columns.retain(|c| c.name != "email");
        db_users.indexes.clear();
        let db = vec![db_users];

        let drift = doctor::compute_db_schema_drift(&snapshot, &db);
        assert!(
            drift.is_clean(),
            "projected snapshot must match the cascade-dropped DB (no false drift); \
             forward={:?} reverse={:?}",
            drift.forward.changes,
            drift.reverse.changes
        );
    }

    // -- `pull --dry-run` report formatting (pure) ---------------------------

    /// A `posts(id BIGINT PK, created_at TIMESTAMP [default])` table. `default`
    /// controls the `created_at` default; `extra` adds a trailing `TEXT` column.
    fn dry_run_posts(default: Option<autumn_schema_core::ColumnDefault>, extra: bool) -> Table {
        use autumn_schema_core::{Column, ColumnType};
        let mut id = Column::new("id".to_string(), ColumnType::Int64);
        id.primary_key = true;
        let mut created_at = Column::new("created_at".to_string(), ColumnType::Timestamp);
        created_at.default = default;
        let mut table = Table::new("posts", Backend::Postgres);
        table.managed = true;
        table.primary_key = vec!["id".to_string()];
        table.columns = vec![id, created_at];
        if extra {
            table
                .columns
                .push(Column::new("extra".to_string(), ColumnType::Text));
        }
        table
    }

    #[test]
    fn dry_run_report_surfaces_reverse_only_removal_alongside_forward_change() {
        // Snapshot: posts(id, created_at DEFAULT now()). Live DB: added an `extra`
        // column AND dropped the `created_at` default. The forward plan catches the
        // added column; the reverse-only pass must still surface the dropped
        // default — the two must both appear, not just the forward change.
        let snapshot = vec![dry_run_posts(
            Some(autumn_schema_core::ColumnDefault::Now),
            false,
        )];
        let db = vec![dry_run_posts(None, true)];
        let drift = doctor::compute_db_schema_drift(&snapshot, &db);

        // Precondition: the forward plan is non-empty (the added column) so we are
        // exercising the "both directions non-empty" path, not the old fallback.
        assert!(
            !drift.forward.is_empty(),
            "forward plan should carry the added column: {:?}",
            drift.forward.changes
        );

        let report = format_pull_dry_run_report(
            &drift.forward,
            &drift.reverse,
            Path::new(".autumn/schema-snapshot.json"),
        );

        assert!(
            report.contains("ADD COLUMN posts.extra"),
            "dry-run report must include the forward added-column change:\n{report}"
        );
        assert!(
            report.contains("pulling would also remove from the snapshot")
                && report.contains("default on `posts.created_at`"),
            "dry-run report must ALSO surface the reverse-only dropped default even though a \
             forward change exists:\n{report}"
        );
    }

    #[test]
    fn dry_run_report_reverse_only_removal_with_empty_forward() {
        // Only a dropped default (no forward change) — the reverse-only removal must
        // still be surfaced (never a false "up to date").
        let snapshot = vec![dry_run_posts(
            Some(autumn_schema_core::ColumnDefault::Now),
            false,
        )];
        let db = vec![dry_run_posts(None, false)];
        let drift = doctor::compute_db_schema_drift(&snapshot, &db);
        assert!(
            drift.forward.is_empty(),
            "forward pass must miss the dropped default"
        );

        let report = format_pull_dry_run_report(
            &drift.forward,
            &drift.reverse,
            Path::new(".autumn/schema-snapshot.json"),
        );
        assert!(
            report.contains("default on `posts.created_at`"),
            "dry-run report must surface the reverse-only dropped default:\n{report}"
        );
        assert!(
            !report.contains("is up to date"),
            "a dropped default is drift, never 'up to date':\n{report}"
        );
    }

    #[test]
    fn dry_run_report_clean_when_both_directions_empty() {
        let snapshot = vec![dry_run_posts(
            Some(autumn_schema_core::ColumnDefault::Now),
            false,
        )];
        let db = snapshot.clone();
        let drift = doctor::compute_db_schema_drift(&snapshot, &db);
        let report = format_pull_dry_run_report(
            &drift.forward,
            &drift.reverse,
            Path::new(".autumn/schema-snapshot.json"),
        );
        assert!(
            report.contains("is up to date"),
            "identical snapshot/DB must report up to date:\n{report}"
        );
    }
}