autumn-cli 0.10.0

//! Migration safety classification — pure SQL pattern analysis.
//!
//! Inspects the contents of `up.sql` files and classifies each SQL statement
//! into a risk tier. The result drives `autumn migrate check`'s exit code and
//! human-readable safety report printed to stderr.
//!
//! Classification is per backend (issue #1906). Postgres asks how risky a
//! statement is for a rolling deploy. `SQLite` asks that too, but first asks
//! whether the statement runs at all: its `ALTER TABLE` accepts only four
//! subcommands, and it has no `CONCURRENTLY`, `TRUNCATE`, sequences, types or
//! grants. A statement the backend cannot parse is
//! [`RiskLevel::Unsupported`].
//!
//! # Lexing
//!
//! A tokenizer ([`tokens`]) reads the SQL with the quote rules of the backend
//! (see the table at [`quoted_len`]). A `;`, `--` or `/*` in a comment or a
//! quoted span is text.
//!
//! Each dialect has its own comment rules. Postgres block comments nest, and
//! a `\r` also ends a `--` comment. `SQLite` block comments do not nest.
//!
//! # Known limitations
//!
//! - In a plain `'..'` literal, `\` does not escape a quote. This is Postgres
//!   with `standard_conforming_strings = on`, the default since 9.1.

use std::fmt;

use autumn_web::config::DatabaseBackend;

/// Risk level for a migration operation, ordered from least to most risky.
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
pub enum RiskLevel {
    /// Additive, backward-compatible schema change. Safe for rolling deploys.
    Safe,
    /// May acquire a table-level lock on large datasets.
    PotentiallyBlocking,
    /// Removes data or structure; old replicas may fail until they restart.
    Destructive,
    /// Cannot be easily reversed without a multi-step expand/contract cycle.
    Irreversible,
    /// Schema change is safe but requires a separate data backfill job.
    DataBackfill,
    /// Autumn cannot auto-classify this statement. Operator review required.
    ManualReview,
    /// The target backend does not accept this statement at all — it will fail
    /// at apply time (issue #1906). `SQLite` only.
    Unsupported,
}

impl fmt::Display for RiskLevel {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Safe => write!(f, "safe"),
            Self::PotentiallyBlocking => write!(f, "potentially-blocking"),
            Self::Destructive => write!(f, "destructive"),
            Self::Irreversible => write!(f, "irreversible"),
            Self::DataBackfill => write!(f, "data-backfill"),
            Self::ManualReview => write!(f, "manual-review"),
            Self::Unsupported => write!(f, "unsupported"),
        }
    }
}

/// A single safety finding for one SQL statement in a migration file.
#[derive(Debug, Clone)]
pub struct SafetyFinding {
    /// Short description of the risky operation (e.g. `DROP COLUMN`).
    pub operation: String,
    /// Risk classification.
    pub risk: RiskLevel,
    /// Why this is dangerous for a rolling deploy.
    pub why: &'static str,
    /// Recommended next action for the operator.
    pub next_action: &'static str,
}

/// Classify the SQL content of an `up.sql` file and return all safety findings.
///
/// Returns an empty `Vec` when the migration is fully additive and safe.
///
/// A statement annotated with `-- autumn-safety: reviewed` is skipped entirely,
/// allowing operators to acknowledge and suppress findings they have manually
/// reviewed and accepted.
///
/// Classifies against Postgres. Use [`classify_sql_for`] on a `SQLite` app.
// Retained as a Postgres-default convenience wrapper for the test suite;
// production always passes the app's detected backend.
#[cfg(test)]
pub fn classify_sql(sql: &str) -> Vec<SafetyFinding> {
    classify_sql_for(DatabaseBackend::Postgres, sql)
}

/// Classify the SQL content of an `up.sql` file against a specific database
/// `backend` (issue #1906).
///
/// The Postgres path is unchanged. The `SQLite` path applies `SQLite`'s own
/// dialect rules: statements Postgres merely finds slow are often outright
/// invalid on `SQLite` (`ALTER COLUMN`, `TRUNCATE`, `CONCURRENTLY`,
/// `ADD COLUMN NOT NULL` with no default), and Postgres-only remedies
/// (`CREATE INDEX CONCURRENTLY`) must never be recommended there. Such
/// statements are reported as [`RiskLevel::Unsupported`] — they fail at apply
/// time, so the deploy gate blocks them.
pub fn classify_sql_for(backend: DatabaseBackend, sql: &str) -> Vec<SafetyFinding> {
    // Strip block comments at the whole-SQL level before splitting so that a
    // semicolon inside a block comment (e.g. `/* note; end */`) does not produce
    // a spurious empty statement fragment.
    let stmts = statements(backend, sql);

    // Tables created within this migration have no existing rows or live traffic,
    // so a non-concurrent index build on them is safe.  Collect the names upfront
    // so we can suppress the false-positive PotentiallyBlocking finding for those
    // tables when we encounter their CREATE INDEX statements below.
    let newly_created: Vec<String> = stmts
        .iter()
        .filter(|s| !has_review_suppression(s))
        .filter_map(|s| extract_created_table_name(&normalize_statement(backend, s)))
        .collect();

    stmts
        .iter()
        .filter(|stmt| !has_review_suppression(stmt))
        .flat_map(|stmt| {
            let normalized = normalize_statement(backend, stmt);
            let mut findings = classify_statement(backend, &normalized, &newly_created);
            // Drop any index-build finding whose table was created earlier in
            // this same migration file (either backend's spelling of it).
            findings.retain(|f| {
                (f.operation != CREATE_INDEX_PG_OP && f.operation != CREATE_INDEX_SQLITE_OP)
                    || extract_index_table_name(&normalized)
                        .is_none_or(|t| !newly_created.iter().any(|c| same_table(backend, c, t)))
            });
            findings
        })
        .collect()
}

/// Returns `true` if the raw (un-normalized) statement carries an operator
/// acknowledgement marker (`-- autumn-safety: reviewed`).
///
/// The check is done on the raw text before comment-stripping so the marker
/// is not accidentally erased.
fn has_review_suppression(stmt: &str) -> bool {
    stmt.lines().any(|line| {
        let trimmed = line.trim();
        trimmed.starts_with("--") && trimmed.contains("autumn-safety: reviewed")
    })
}

/// True iff all findings are at the `Safe` risk level (or there are none).
pub fn is_safe(findings: &[SafetyFinding]) -> bool {
    findings.iter().all(|f| f.risk == RiskLevel::Safe)
}

/// True iff any finding exceeds the `Safe` risk level.
pub fn has_unsafe_findings(findings: &[SafetyFinding]) -> bool {
    findings.iter().any(|f| f.risk > RiskLevel::Safe)
}

/// True iff `sql` contains at least one non-empty, non-comment SQL statement.
///
/// Used to gate `autumn migrate down`: a `down.sql` that is blank or contains
/// only comments is treated as absent — the command refuses to proceed and
/// names the offending migration.
///
/// The comment rules are those of `backend`: a nested `/*` differs.
pub fn has_executable_sql(backend: DatabaseBackend, sql: &str) -> bool {
    statements(backend, sql).iter().any(|stmt| {
        stmt.lines()
            .any(|line| !line.trim().is_empty() && !line.trim().starts_with("--"))
    })
}

// ── internals ────────────────────────────────────────────────────────────────

/// Whether two table names from normalized statements name one table.
/// `SQLite` names ignore ASCII case, also in quotes. A quoted Postgres name
/// keeps its case.
fn same_table(backend: DatabaseBackend, a: &str, b: &str) -> bool {
    match backend {
        DatabaseBackend::Postgres => a == b,
        DatabaseBackend::Sqlite => a.eq_ignore_ascii_case(b),
    }
}

/// Extract the table name from a normalized `CREATE TABLE name …` statement.
/// Only unconditional creates that are known to create a fresh table are matched.
fn extract_created_table_name(normalized: &str) -> Option<String> {
    let rest = normalized.strip_prefix("create table ")?;
    if rest.starts_with("if not exists ") {
        return None;
    }
    let name = rest.split([' ', '(']).next()?;
    if name.is_empty() {
        None
    } else {
        Some(name.to_owned())
    }
}

/// Extract the target table name from a normalized `CREATE [UNIQUE] INDEX … ON name …` statement.
fn extract_index_table_name(normalized: &str) -> Option<&str> {
    let after_on = normalized.find(" on ").map(|i| &normalized[i + 4..])?;
    let name = after_on.split([' ', '(']).next()?;
    if name.is_empty() { None } else { Some(name) }
}

/// Extract the table name from a normalized `ALTER TABLE name …` statement.
fn extract_altered_table_name(normalized: &str) -> Option<&str> {
    let rest = normalized.strip_prefix("alter table ")?;
    let rest = rest.strip_prefix("if exists ").unwrap_or(rest);
    let rest = rest.strip_prefix("only ").unwrap_or(rest);
    let name = rest.split([' ', '(']).next()?;
    if name.is_empty() { None } else { Some(name) }
}

/// Strip block comments, then split `sql` into statements.
fn statements(backend: DatabaseBackend, sql: &str) -> Vec<String> {
    split_statements(backend, &strip_block_comments(backend, sql))
}

/// Split `sql` into statements at each `;`. The result drops each terminating
/// `;`, trims each statement and skips empty ones.
///
/// A `;` in a comment or a quoted span (see [`quoted_len`]) does not split.
pub fn split_statements(backend: DatabaseBackend, sql: &str) -> Vec<String> {
    let mut statements = Vec::new();
    let mut current = String::new();
    for (kind, text) in tokens(backend, sql) {
        if kind == TokenKind::Plain && text == ";" {
            let trimmed = current.trim();
            if !trimmed.is_empty() {
                statements.push(trimmed.to_owned());
            }
            current.clear();
        } else {
            // Comments stay: `has_review_suppression` reads them.
            current.push_str(text);
        }
    }
    let trimmed = current.trim();
    if !trimmed.is_empty() {
        statements.push(trimmed.to_owned());
    }
    statements
}

/// Split a normalized `ALTER TABLE` statement into individual subcommand strings.
///
/// Strips the `alter table <name>` prefix and splits the remaining text on
/// commas outside parentheses, trimming each segment. A space or comma in a
/// quoted span of the `backend` dialect does not split.
fn alter_table_subcommands(backend: DatabaseBackend, normalized: &str) -> Vec<&str> {
    let mut rest = normalized.strip_prefix("alter table ").unwrap_or("");
    // `SQLite` has no `IF EXISTS` or `ONLY` here. They stay in its first
    // subcommand, and the grammar rule rejects them.
    if backend == DatabaseBackend::Postgres {
        for keyword in ["if exists ", "only "] {
            rest = rest.strip_prefix(keyword).unwrap_or(rest);
        }
    }
    let name_len: usize = tokens(backend, rest)
        .take_while(|&(kind, text)| !(kind == TokenKind::Plain && text == " "))
        .map(|(_, text)| text.len())
        .sum();
    let subcommands = rest[name_len..].trim_start();

    let mut result = Vec::new();
    let mut depth: i32 = 0;
    let mut start = 0;
    let mut i = 0;
    for (kind, text) in tokens(backend, subcommands) {
        if kind == TokenKind::Plain {
            match text {
                "(" => depth += 1,
                ")" => depth -= 1,
                "," if depth == 0 => {
                    result.push(subcommands[start..i].trim());
                    start = i + 1;
                }
                _ => {}
            }
        }
        i += text.len();
    }
    let last = subcommands[start..].trim();
    if !last.is_empty() {
        result.push(last);
    }
    result
}

/// True iff `subcommand` is an ALTER TABLE subcommand that Autumn fully classifies.
///
/// A subcommand is "known" when a specific safety rule covers all risk scenarios
/// for it (including the case where it is safe and produces no finding).
fn is_known_alter_subcommand(backend: DatabaseBackend, subcommand: &str) -> bool {
    // On SQLite every subcommand is already classified: the four the grammar
    // accepts by the rules below, and every other one as Unsupported. So none
    // is left for the manual-review fallback.
    if backend == DatabaseBackend::Sqlite {
        return true;
    }
    // `add column` is known unless it carries an inline PRIMARY KEY constraint,
    // which Autumn does not specifically classify.  UNIQUE and REFERENCES are handled
    // by dedicated rules; NOT NULL is handled by the NOT NULL rule.
    let add_col_known =
        subcommand.starts_with("add column ") && !subcommand.contains(" primary key");
    add_col_known
        || subcommand.starts_with("drop column ")
        || subcommand.starts_with("rename column ") // RENAME COLUMN
        || subcommand.starts_with("rename to ") // RENAME TABLE (ALTER TABLE … RENAME TO …)
        || (subcommand.starts_with("alter column ") && subcommand.contains(" type "))
}

/// Rewrite `SQLite`'s optional-`COLUMN` `ALTER TABLE` spellings into the
/// canonical ones, so `ADD title TEXT` classifies exactly as
/// `ADD COLUMN title TEXT`, and `RENAME a TO b` as `RENAME COLUMN a TO b`.
/// `ADD CONSTRAINT` and `DROP CONSTRAINT` are left alone: they are
/// Postgres-only, and the grammar rule must still reject them.
fn canonical_sqlite_alter(normalized: &str) -> String {
    if !normalized.starts_with("alter table ") {
        return normalized.to_owned();
    }
    let subcommands = alter_table_subcommands(DatabaseBackend::Sqlite, normalized);
    if subcommands.is_empty() {
        return normalized.to_owned();
    }
    // The subcommands borrow from `normalized`, so the first one marks the end
    // of the head, whatever spacing the commas had.
    let head_len = subcommands[0].as_ptr() as usize - normalized.as_ptr() as usize;
    let head = &normalized[..head_len];
    let rewritten: Vec<String> = subcommands
        .iter()
        .map(|sub| {
            if let Some(rest) = sub.strip_prefix("rename ")
                && !rest.starts_with("column ")
                && !rest.starts_with("to ")
                && rest.contains(" to ")
            {
                return format!("rename column {rest}");
            }
            for verb in ["add", "drop"] {
                let prefix = format!("{verb} ");
                if let Some(rest) = sub.strip_prefix(&prefix)
                    && !rest.starts_with("column ")
                    && !rest.starts_with("constraint ")
                {
                    return format!("{verb} column {rest}");
                }
            }
            (*sub).to_owned()
        })
        .collect();
    format!("{head}{}", rewritten.join(", "))
}

/// Whether `subcommand` is one of the four `ALTER TABLE` forms `SQLite` parses:
/// `RENAME TO`, `RENAME [COLUMN] … TO …`, `ADD COLUMN` and `DROP COLUMN`.
/// `SQLite` also accepts `ADD`/`DROP` with the `COLUMN` keyword left out.
fn is_sqlite_alter_subcommand(subcommand: &str) -> bool {
    ["rename to ", "rename column ", "add column ", "drop column "]
        .iter()
        .any(|p| subcommand.starts_with(p))
        // Bare `RENAME old TO new`, `ADD <col> …`, `DROP <col>` — but not
        // `ADD CONSTRAINT`/`DROP CONSTRAINT`, which SQLite rejects.
        || (subcommand.starts_with("rename ") && subcommand.contains(" to "))
        || (subcommand.starts_with("add ") && !subcommand.starts_with("add constraint "))
        || (subcommand.starts_with("drop ") && !subcommand.starts_with("drop constraint "))
}

/// Operation label for a non-concurrent Postgres index build.
const CREATE_INDEX_PG_OP: &str = "CREATE INDEX (non-concurrent)";
/// Operation label for a `SQLite` index build. `SQLite` has no `CONCURRENTLY`,
/// so "non-concurrent" would be a meaningless qualifier there.
const CREATE_INDEX_SQLITE_OP: &str = "CREATE INDEX";

/// The value of an `add column` subcommand's `DEFAULT` clause, lowercased, or
/// `None` when the subcommand carries no default.
///
/// The keyword is matched at a word boundary, so `DEFAULT(0)` — valid SQL with
/// no space — is read, while a column named `defaulted` is not.
///
/// Three shapes are read whole rather than split on whitespace: a parenthesized
/// expression (`(lower(x))`), a quoted literal including one with spaces
/// (`'a b'`, with `''` as the escaped quote), and a signed number (`-1`).
/// A `DEFAULT` inside a foreign-key action (`ON UPDATE SET DEFAULT`) is not a
/// column default and is skipped.
fn add_column_default_token(subcommand: &str) -> Option<String> {
    let mut from = 0;
    let rest = loop {
        let at = subcommand[from..].find(" default")? + from;
        let after = &subcommand[at + " default".len()..];
        // `DEFAULT(0)` needs no space, but `defaulted` is a different word.
        let value = match after.strip_prefix(' ') {
            Some(v) => Some(v),
            None if after.starts_with('(') => Some(after),
            None => None,
        };
        // `SET DEFAULT` here is an FK action, not this column's default.
        if let Some(value) = value
            && !subcommand[..at].trim_end().ends_with(" set")
        {
            break value.trim_start();
        }
        from = at + " default".len();
    };
    if rest.starts_with('(') {
        let mut depth = 0usize;
        for (i, c) in rest.char_indices() {
            match c {
                '(' => depth += 1,
                // `depth` is at least 1 here: the first character is `(`.
                ')' => {
                    depth -= 1;
                    if depth == 0 {
                        return Some(rest[..=i].to_owned());
                    }
                }
                _ => {}
            }
        }
        return Some(rest.to_owned());
    }
    if rest.starts_with('\'') {
        let bytes = rest.as_bytes();
        let mut i = 1;
        while i < bytes.len() {
            if bytes[i] == b'\'' {
                if bytes.get(i + 1) == Some(&b'\'') {
                    i += 2; // `''` is one escaped quote
                    continue;
                }
                return Some(rest[..=i].to_owned());
            }
            i += 1;
        }
        return Some(rest.to_owned());
    }
    Some(
        rest.split_whitespace()
            .next()
            .unwrap_or_default()
            .trim_end_matches(',')
            .to_owned(),
    )
}

/// Whether `default` is a constant `SQLite` accepts as an `ADD COLUMN` default.
///
/// `SQLite`'s rule is that the default must reduce to a constant value. Its
/// parser keeps no node for parentheses, so `DEFAULT ('draft')` is the same as
/// `DEFAULT 'draft'` and is accepted; `DEFAULT (1+2)` and `DEFAULT (abs(-1))`
/// are not. `CURRENT_TIME`, `CURRENT_DATE` and `CURRENT_TIMESTAMP` are rejected
/// by name. An unparenthesized call such as `DEFAULT now()` is not even
/// grammatical — only a literal or signed number may follow `DEFAULT` bare — so
/// it counts as non-constant too.
fn is_sqlite_constant_default(default: &str) -> bool {
    let inner = default
        .strip_prefix('(')
        .and_then(|d| d.strip_suffix(')'))
        .unwrap_or(default)
        .trim();
    if matches!(inner, "current_time" | "current_date" | "current_timestamp") {
        return false;
    }
    // A quoted literal is constant whatever it contains.
    if inner.starts_with('\'') || inner.starts_with("x'") {
        return true;
    }
    // A number, including exponent form (`1e-3`), whose sign is part of the
    // literal rather than an operator.
    if is_numeric_literal(inner) {
        return true;
    }
    // Otherwise: a bare keyword literal. A call or an operator makes it an
    // expression.
    let unsigned = inner.trim_start_matches(['+', '-']).trim_start();
    !unsigned.contains(['(', ')', '+', '-', '*', '/', '|', '<', '>', '=', '\''])
}

/// Whether `value` is an `SQLite` numeric literal: an optionally signed integer
/// or real, in plain, exponent (`1e-3`) or hex (`0x1f`) form.
fn is_numeric_literal(value: &str) -> bool {
    let body = value.strip_prefix(['+', '-']).unwrap_or(value);
    if let Some(hex) = body.strip_prefix("0x").or_else(|| body.strip_prefix("0X")) {
        return !hex.is_empty() && hex.chars().all(|c| c.is_ascii_hexdigit());
    }
    let (mantissa, exponent) = body
        .split_once(['e', 'E'])
        .map_or((body, None), |(m, e)| (m, Some(e)));
    if !is_decimal(mantissa) {
        return false;
    }
    exponent.is_none_or(|e| {
        let digits = e.strip_prefix(['+', '-']).unwrap_or(e);
        !digits.is_empty() && digits.chars().all(|c| c.is_ascii_digit())
    })
}

/// Whether `value` is an unsigned decimal integer or real (`12`, `1.5`, `.5`).
fn is_decimal(value: &str) -> bool {
    let (int, frac) = value
        .split_once('.')
        .map_or((value, None), |(i, f)| (i, Some(f)));
    if int.is_empty() && frac.is_none_or(str::is_empty) {
        return false;
    }
    int.chars().all(|c| c.is_ascii_digit())
        && frac.is_none_or(|f| f.chars().all(|c| c.is_ascii_digit()))
}

/// A statement `SQLite` has no syntax for at all (issue #1906).
///
/// These are Postgres-only objects (sequences, types, extensions, materialized
/// views), Postgres-only statements (`TRUNCATE`, `COMMENT ON`, `GRANT`,
/// `REVOKE`) or the Postgres-only `CONCURRENTLY` index option. Each fails at
/// apply time on `SQLite`, so it is reported as [`RiskLevel::Unsupported`]
/// rather than classified for rolling-deploy risk.
fn sqlite_unsupported_statement(normalized: &str) -> Option<SafetyFinding> {
    let (operation, next_action) = if normalized.starts_with("truncate ") {
        (
            "TRUNCATE (unsupported on SQLite)",
            "SQLite has no TRUNCATE statement. Use `DELETE FROM <table>;` (add `DELETE FROM \
             sqlite_sequence WHERE name = '<table>';` to reset AUTOINCREMENT).",
        )
    } else if normalized.starts_with("create index concurrently ")
        || normalized.starts_with("create unique index concurrently ")
    {
        (
            "CREATE INDEX CONCURRENTLY (unsupported on SQLite)",
            "SQLite has no CONCURRENTLY option. Drop the keyword and build the index during a \
             low-traffic window.",
        )
    } else if normalized.starts_with("drop index concurrently ") {
        (
            "DROP INDEX CONCURRENTLY (unsupported on SQLite)",
            "SQLite has no CONCURRENTLY option. Drop the keyword — a plain DROP INDEX is a cheap \
             catalog edit on SQLite.",
        )
    } else if normalized.starts_with("create sequence")
        || normalized.starts_with("alter sequence")
        || normalized.starts_with("drop sequence")
    {
        (
            "SEQUENCE statement (unsupported on SQLite)",
            "SQLite has no sequences. Use an INTEGER PRIMARY KEY AUTOINCREMENT column instead.",
        )
    } else if normalized.starts_with("create type")
        || normalized.starts_with("alter type")
        || normalized.starts_with("drop type")
    {
        (
            "TYPE statement (unsupported on SQLite)",
            "SQLite has no user-defined types. Model an enum as a TEXT column with a CHECK \
             constraint.",
        )
    } else if normalized.starts_with("create extension") || normalized.starts_with("drop extension")
    {
        (
            "EXTENSION statement (unsupported on SQLite)",
            "SQLite has no extensions in this sense. Remove the statement, or gate the migration to \
             the Postgres backend.",
        )
    } else if normalized.starts_with("comment on") {
        (
            "COMMENT ON (unsupported on SQLite)",
            "SQLite has no COMMENT ON. Remove the statement — it carries no schema meaning.",
        )
    } else if normalized.starts_with("create materialized view")
        || normalized.starts_with("drop materialized view")
        || normalized.starts_with("refresh materialized view")
    {
        (
            "MATERIALIZED VIEW statement (unsupported on SQLite)",
            "SQLite has no materialized views. Use a plain view, or a real table refreshed by a \
             background job.",
        )
    } else if normalized.starts_with("grant ") || normalized.starts_with("revoke ") {
        (
            "GRANT/REVOKE (unsupported on SQLite)",
            "SQLite has no roles or grants — file permissions are the access control. Remove the \
             statement.",
        )
    } else {
        return sqlite_unsupported_clause(normalized);
    };
    Some(SafetyFinding {
        operation: operation.to_owned(),
        risk: RiskLevel::Unsupported,
        why: "SQLite has no syntax for this statement, so the migration fails at apply time.",
        next_action,
    })
}

/// A statement whose *shape* `SQLite` accepts but which carries a Postgres-only
/// clause, plus the DML forms `SQLite` has no grammar for (issue #1906).
///
/// Split out from [`sqlite_unsupported_statement`], which handles the
/// Postgres-only object statements.
fn sqlite_unsupported_clause(normalized: &str) -> Option<SafetyFinding> {
    let (operation, next_action) = if normalized.starts_with("merge into ") {
        (
            "MERGE (unsupported on SQLite)",
            "SQLite has no MERGE. Use INSERT … ON CONFLICT (upsert), naming the conflicting \
             columns rather than a constraint.",
        )
    } else if normalized.starts_with("drop index ")
        && normalized
            .trim_start_matches("drop index ")
            .trim_start_matches("if exists ")
            .starts_with("sqlite_autoindex")
    {
        (
            "DROP INDEX on an implicit index (unsupported on SQLite)",
            "SQLite refuses to drop the index behind a UNIQUE or PRIMARY KEY constraint, even \
             with IF EXISTS. Rebuild the table without the constraint instead.",
        )
    } else if is_create_index_statement(normalized) && has_postgres_index_clause(normalized) {
        (
            "CREATE INDEX with a Postgres-only clause (unsupported on SQLite)",
            "SQLite's CREATE INDEX takes no USING, INCLUDE, WITH, TABLESPACE or NULLS NOT \
             DISTINCT clause. Drop it — SQLite indexes are always B-trees. WHERE (partial), \
             COLLATE and DESC are supported.",
        )
    } else if normalized.starts_with("create table") && has_postgres_table_clause(normalized) {
        (
            "CREATE TABLE with a Postgres-only clause (unsupported on SQLite)",
            "SQLite has no declarative partitioning, identity columns, table inheritance or \
             tablespaces. Use INTEGER PRIMARY KEY AUTOINCREMENT for an identity column; model \
             partitioning in the application.",
        )
    } else if has_row_locking_clause(normalized) {
        (
            "SELECT … FOR UPDATE (unsupported on SQLite)",
            "SQLite has no row-level locking clause — a write transaction locks the whole \
             database. Drop the clause.",
        )
    } else if blank_quoted_spans(DatabaseBackend::Sqlite, normalized)
        .contains("on conflict on constraint ")
    {
        (
            "ON CONFLICT ON CONSTRAINT (unsupported on SQLite)",
            "SQLite's upsert target is a column list, not a constraint name. Write \
             `ON CONFLICT (<columns>)`.",
        )
    } else if normalized.starts_with("with ") && contains_data_modifying_cte(normalized) {
        (
            "Data-modifying CTE (unsupported on SQLite)",
            "SQLite allows only SELECT inside a CTE. Run the UPDATE/DELETE/INSERT as its own \
             statement.",
        )
    } else {
        return None;
    };
    Some(SafetyFinding {
        operation: operation.to_owned(),
        risk: RiskLevel::Unsupported,
        why: "SQLite has no syntax for this statement, so the migration fails at apply time.",
        next_action,
    })
}

/// Whether the statement carries a Postgres row-locking clause.
///
/// Searches the form with blank quoted spans: an INSERT of the text
/// `'waiting for update'` is data, not a locking clause.
fn has_row_locking_clause(normalized: &str) -> bool {
    let bare = blank_quoted_spans(DatabaseBackend::Sqlite, normalized);
    bare.contains(" for update") || bare.contains(" for share")
}

/// Whether the statement is a `CREATE [UNIQUE] INDEX`.
fn is_create_index_statement(normalized: &str) -> bool {
    normalized.starts_with("create index") || normalized.starts_with("create unique index")
}

/// Whether a `CREATE INDEX` carries a clause `SQLite` has no grammar for.
/// `WHERE` (partial), `COLLATE` and `DESC` are absent from the list: `SQLite`
/// supports all three.
fn has_postgres_index_clause(normalized: &str) -> bool {
    let sql = blank_quoted_spans(DatabaseBackend::Sqlite, normalized);
    sql.contains(" using ")
        || sql.contains(" include (")
        || sql.contains(" with (")
        || sql.contains(" tablespace ")
        || sql.contains(" nulls not distinct")
}

/// Whether a `CREATE TABLE` carries a clause `SQLite` has no grammar for.
fn has_postgres_table_clause(normalized: &str) -> bool {
    let sql = blank_quoted_spans(DatabaseBackend::Sqlite, normalized);
    sql.contains(" partition by ")
        || sql.contains(" partition of ")
        || sql.contains(" as identity")
        || sql.contains(" inherits (")
        || sql.contains(" tablespace ")
}

/// Whether a `WITH` statement writes inside a CTE. `SQLite` allows only
/// `SELECT` there.
fn contains_data_modifying_cte(normalized: &str) -> bool {
    // `AS ( DELETE FROM …` is ordinary formatting; close the paren up first.
    let sql = blank_quoted_spans(DatabaseBackend::Sqlite, normalized).replace("( ", "(");
    ["(update ", "(delete ", "(insert into "]
        .iter()
        .any(|p| sql.contains(p))
}

/// Blank the text in each quoted span, but keep its first and last character
/// and its character count. A keyword search then does not match a word in a
/// quoted span (`DEFAULT 'not unique yet'`, `"it's"`).
fn blank_quoted_spans(backend: DatabaseBackend, sql: &str) -> String {
    let mut out = String::with_capacity(sql.len());
    for (kind, text) in tokens(backend, sql) {
        if kind == TokenKind::Quoted {
            let last = text.chars().count() - 1;
            out.extend(
                text.chars()
                    .enumerate()
                    .map(|(i, c)| if i == 0 || i == last { c } else { ' ' }),
            );
        } else {
            out.push_str(text);
        }
    }
    out
}

/// Returns `true` when the normalized `add column` subcommand carries an inline
/// `UNIQUE` constraint keyword.  Trailing-space / end-of-string anchoring prevents
/// matching a column or table name that contains `unique` as a substring.
fn has_inline_unique_constraint(subcommand: &str) -> bool {
    subcommand.contains(" unique ") || subcommand.ends_with(" unique")
}

/// A token kind.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum TokenKind {
    /// One character of SQL, or one `SQLite` variable (see
    /// [`sqlite_variable_len`]).
    Plain,
    /// A `--` comment. The line end is not part of it.
    LineComment,
    /// A `/* */` comment. Postgres comments nest. `SQLite` comments do not.
    BlockComment,
    /// A quoted span: a literal, a quoted identifier or a dollar-quoted block.
    Quoted,
}

/// Split `sql` into tokens for `backend`. Each item is a kind and its text.
/// The texts join back to `sql`.
fn tokens(backend: DatabaseBackend, sql: &str) -> impl Iterator<Item = (TokenKind, &str)> {
    let mut rest = sql;
    let mut prev = None;
    std::iter::from_fn(move || {
        let (kind, len) = token_at(backend, rest, prev)?;
        let (text, tail) = rest.split_at(len);
        rest = tail;
        prev = text.chars().next_back();
        Some((kind, text))
    })
}

/// The kind and byte length of the token at the start of `rest`, or `None`
/// when `rest` is empty. `prev` is the character before `rest`.
fn token_at(
    backend: DatabaseBackend,
    rest: &str,
    prev: Option<char>,
) -> Option<(TokenKind, usize)> {
    let first = rest.chars().next()?;
    if let Some(len) = line_comment_len(backend, rest) {
        return Some((TokenKind::LineComment, len));
    }
    if let Some(len) = block_comment_len(backend, rest) {
        return Some((TokenKind::BlockComment, len));
    }
    if let Some(len) = quoted_len(backend, rest, prev) {
        return Some((TokenKind::Quoted, len));
    }
    if backend == DatabaseBackend::Sqlite
        && prev.is_none_or(|c| !is_identifier_char(c))
        && let Some(len) = sqlite_variable_len(rest)
    {
        return Some((TokenKind::Plain, len));
    }
    Some((TokenKind::Plain, first.len_utf8()))
}

/// Byte length of the quoted span at the start of `rest`, or `None`. `prev` is
/// the character before `rest`. In a span, `--`, `/*` and `;` are text.
///
/// The span forms follow the dialect:
///
/// | Form | Escaped quote | Postgres | `SQLite` |
/// | --- | --- | --- | --- |
/// | `'..'` literal | doubled | yes | yes |
/// | `E'..'` literal | doubled or `\'` | yes | no |
/// | `".."` identifier | doubled | yes | yes |
/// | `` `..` `` identifier | doubled | no | yes |
/// | `[..]` identifier | none | no | yes |
/// | `$tag$..$tag$` block | none | yes | no |
///
/// `E'` and `$` start a span only at the start of a word: in `xe'..'` and
/// `a$b$`, Postgres reads them as part of the identifier. An unclosed span
/// runs to the end of `rest`.
fn quoted_len(backend: DatabaseBackend, rest: &str, prev: Option<char>) -> Option<usize> {
    let postgres = backend == DatabaseBackend::Postgres;
    let word_start = prev.is_none_or(|c| !is_identifier_char(c));
    match rest.as_bytes().first()? {
        b'\'' => Some(doubled_quote_len(rest, '\'')),
        b'"' => Some(doubled_quote_len(rest, '"')),
        b'`' if !postgres => Some(doubled_quote_len(rest, '`')),
        b'[' if !postgres => Some(rest[1..].find(']').map_or(rest.len(), |p| p + 2)),
        b'e' | b'E' if postgres && word_start && rest[1..].starts_with('\'') => {
            Some(1 + escape_string_len(&rest[1..]))
        }
        b'$' if postgres && word_start => dollar_quote_len(rest),
        _ => None,
    }
}

/// Whether `c` can be part of an identifier. Postgres and `SQLite` agree.
const fn is_identifier_char(c: char) -> bool {
    c.is_ascii_alphanumeric() || matches!(c, '_' | '$') || !c.is_ascii()
}

/// Byte length of the span at the start of `rest` that `close` ends.
///
/// `rest` must start with a one-byte opening character. A doubled `close` is
/// an escaped `close`. An unclosed span runs to the end of `rest`.
fn doubled_quote_len(rest: &str, close: char) -> usize {
    let mut j = 1;
    while let Some(rel) = rest[j..].find(close) {
        let abs = j + rel;
        if rest[abs + 1..].starts_with(close) {
            j = abs + 2;
        } else {
            return abs + 1;
        }
    }
    rest.len()
}

/// Byte length of the `'` literal at the start of `rest`, where `\` escapes
/// the next character (the body of a Postgres `E'..'` string). A doubled `''`
/// is also an escaped quote. An unclosed literal runs to the end of `rest`.
fn escape_string_len(rest: &str) -> usize {
    let mut chars = rest.char_indices().skip(1);
    while let Some((i, c)) = chars.next() {
        match c {
            '\\' => {
                chars.next();
            }
            '\'' if rest[i + 1..].starts_with('\'') => {
                chars.next();
            }
            '\'' => return i + 1,
            _ => {}
        }
    }
    rest.len()
}

/// Byte length of the dollar-quoted block (`$$…$$`, `$tag$…$tag$`) at the
/// start of `rest`, or `None` when `rest` does not start one. An unclosed
/// block runs to the end of `rest`.
fn dollar_quote_len(rest: &str) -> Option<usize> {
    let after = rest.strip_prefix('$')?;
    let tag_body = &after[..after.find('$')?];
    // `$1` is a parameter, so a tag cannot start with a digit.
    if tag_body.starts_with(|c: char| c.is_ascii_digit())
        || !tag_body.chars().all(|c| is_identifier_char(c) && c != '$')
    {
        return None;
    }
    let tag_len = tag_body.len() + 2;
    let tag = &rest[..tag_len];
    Some(
        rest[tag_len..]
            .find(tag)
            .map_or(rest.len(), |p| p + 2 * tag_len),
    )
}

/// Byte length of the line comment at the start of `rest`, or `None`. The
/// line end is not part of the comment. Postgres ends the comment at `\n` or
/// `\r`. `SQLite` ends it at `\n` only.
fn line_comment_len(backend: DatabaseBackend, rest: &str) -> Option<usize> {
    let end: &[char] = match backend {
        DatabaseBackend::Postgres => &['\n', '\r'],
        DatabaseBackend::Sqlite => &['\n'],
    };
    rest.starts_with("--")
        .then(|| rest.find(end).unwrap_or(rest.len()))
}

/// Byte length of the block comment at the start of `rest`, or `None`.
/// Postgres comments nest, so each `/*` needs its own `*/`. `SQLite` comments
/// end at the first `*/`. An unclosed comment runs to the end of `rest`.
fn block_comment_len(backend: DatabaseBackend, rest: &str) -> Option<usize> {
    if !rest.starts_with("/*") {
        return None;
    }
    let nests = backend == DatabaseBackend::Postgres;
    let mut depth = 0usize;
    let mut i = 0;
    while i < rest.len() {
        if rest[i..].starts_with("/*") && (nests || depth == 0) {
            depth += 1;
            i += 2;
        } else if rest[i..].starts_with("*/") {
            depth -= 1;
            i += 2;
            if depth == 0 {
                return Some(i);
            }
        } else {
            i += rest[i..].chars().next().map_or(1, char::len_utf8);
        }
    }
    Some(rest.len())
}

/// Byte length of the `SQLite` variable at the start of `rest`, or `None`.
///
/// A variable is `$`, `@`, `:` or `#`, then name characters. A `(` after the
/// name runs to `)` or to a space, as in `$a(x)`. A quote in the variable is
/// text, as in `SQLite`.
fn sqlite_variable_len(rest: &str) -> Option<usize> {
    let body = rest.strip_prefix(['$', '@', ':', '#'])?;
    let name_len: usize = body
        .chars()
        .take_while(|&c| is_identifier_char(c))
        .map(char::len_utf8)
        .sum();
    if name_len == 0 {
        return None;
    }
    let after = &body[name_len..];
    let args_len = if after.starts_with('(') {
        after
            .find(|c: char| c == ')' || c.is_whitespace())
            .map_or(after.len(), |p| {
                p + usize::from(after[p..].starts_with(')'))
            })
    } else {
        0
    };
    Some(1 + name_len + args_len)
}

/// Replace each `/* ... */` block comment in `sql` with one space.
///
/// A `/*` in a `--` comment or a quoted span starts nothing. Line comments
/// stay because `has_review_suppression` reads them.
pub fn strip_block_comments(backend: DatabaseBackend, sql: &str) -> String {
    let mut result = String::with_capacity(sql.len());
    for (kind, text) in tokens(backend, sql) {
        if kind == TokenKind::BlockComment {
            result.push(' '); // keep the token boundary
        } else {
            result.push_str(text);
        }
    }
    result
}

/// Remove `--` line comments from `sql`, keeping newlines. A `--` in a quoted
/// span is text.
fn strip_line_comments(backend: DatabaseBackend, sql: &str) -> String {
    tokens(backend, sql)
        .filter(|(kind, _)| *kind != TokenKind::LineComment)
        .map(|(_, text)| text)
        .collect()
}

/// Strip comments, collapse whitespace, and lowercase a single statement.
///
/// Quoted spans stay as they are. A later scan of the result then reads the
/// same spans: Postgres dollar tags are case-sensitive (`$A$` is not `$a$`).
pub fn normalize_statement(backend: DatabaseBackend, stmt: &str) -> String {
    let stripped = strip_line_comments(backend, &strip_block_comments(backend, stmt));
    let mut out = String::with_capacity(stripped.len());
    for (kind, text) in tokens(backend, &stripped) {
        if kind == TokenKind::Quoted {
            out.push_str(text);
            continue;
        }
        for c in text.chars() {
            if !c.is_whitespace() {
                out.extend(c.to_lowercase());
            } else if !out.is_empty() && !out.ends_with(' ') {
                out.push(' ');
            }
        }
    }
    out.truncate(out.trim_end().len());
    out
}

/// Returns `true` when `sql` contains an executable concurrent index operation
/// (`CREATE [UNIQUE] INDEX CONCURRENTLY` or `DROP INDEX CONCURRENTLY`).
///
/// Uses the same comment-stripping and whitespace-normalization pipeline as
/// [`classify_sql_for`] so that concurrent index keywords mentioned only inside a
/// SQL comment (e.g. `-- CREATE INDEX CONCURRENTLY ...`) are not counted.
/// `CONCURRENTLY` is Postgres only, so the scan uses the Postgres dialect.
pub fn contains_concurrent_index(sql: &str) -> bool {
    let pg = DatabaseBackend::Postgres;
    statements(pg, sql).iter().any(|stmt| {
        let normalized = normalize_statement(pg, stmt);
        normalized.contains("create index concurrently ")
            || normalized.contains("create unique index concurrently ")
            || normalized.starts_with("drop index concurrently ")
    })
}

/// `PostgreSQL` STABLE/IMMUTABLE function prefixes that are safe as NOT NULL column
/// defaults: Postgres evaluates them once at statement time and stores the result as
/// a constant, so the PG 11+ fast-default path applies — no table rewrite needed.
const STABLE_FN_PREFIXES: &[&str] = &["now(", "current_timestamp(", "localtimestamp("];

/// Returns `true` when `default_expr` is a volatile function call that `PostgreSQL`
/// cannot optimise via the PG 11+ fast-constant-default path.
///
/// Only VOLATILE function calls are flagged.  Grouped constant expressions such as
/// `(0)` or `('draft')` have `(` as the first non-space character (no identifier
/// before the parenthesis) and are treated as constants — they use the fast path.
/// STABLE/IMMUTABLE functions (e.g. `now()`) are also exempt via `STABLE_FN_PREFIXES`.
fn is_volatile_function_default(default_expr: &str) -> bool {
    let Some(paren_pos) = default_expr.find('(') else {
        return false; // no parentheses — constant literal
    };
    // A function call has an identifier character immediately before `(`.
    // A grouped constant like `(0)` or `('x')` has nothing or whitespace before it.
    let is_fn_call = default_expr[..paren_pos]
        .chars()
        .next_back()
        .is_some_and(|c| c.is_alphanumeric() || c == '_');
    if !is_fn_call {
        return false; // parenthesized constant — uses the fast-default path
    }
    !STABLE_FN_PREFIXES
        .iter()
        .any(|p| default_expr.starts_with(p))
}

/// Apply all pattern checks to a single normalized (lowercase, single-spaced) statement.
#[allow(clippy::too_many_lines)]
fn classify_statement(
    backend: DatabaseBackend,
    normalized: &str,
    newly_created: &[String],
) -> Vec<SafetyFinding> {
    if normalized.is_empty() {
        return vec![];
    }
    let sqlite = backend == DatabaseBackend::Sqlite;
    // SQLite lets `ALTER TABLE … ADD <col>` and `DROP <col>` omit the `COLUMN`
    // keyword. Every rule below matches the `COLUMN` spelling, so canonicalize
    // first rather than duplicating each match.
    let canonical;
    let normalized = if sqlite {
        canonical = canonical_sqlite_alter(normalized);
        canonical.as_str()
    } else {
        normalized
    };

    // Statements SQLite has no syntax for at all. They fail at apply time, so
    // there is nothing further to classify about them.
    if sqlite && let Some(f) = sqlite_unsupported_statement(normalized) {
        return vec![f];
    }

    let mut findings = Vec::new();
    // Some SQLite ADD COLUMN restrictions apply only to a table that already has
    // rows: NOT NULL without a usable default, a non-constant default, and a
    // STORED generated column. A table this same migration creates has none, so
    // those statements apply cleanly — the suppression `classify_sql_for`
    // already makes for index builds. UNIQUE and PRIMARY KEY are NOT in that
    // set: SQLite rejects them on an empty table too.
    let fresh_table = sqlite
        && extract_altered_table_name(normalized)
            .is_some_and(|t| newly_created.iter().any(|c| same_table(backend, c, t)));

    // DROP TABLE — check first; it subsumes DROP COLUMN detection
    if normalized.starts_with("drop table") {
        findings.push(SafetyFinding {
            operation: "DROP TABLE".to_owned(),
            risk: RiskLevel::Destructive,
            why: "Drops the entire table and all its data. Old replicas that reference this \
                  table will error immediately.",
            next_action: "Use expand/contract: first deploy code that stops using the table, \
                          then drop it in a subsequent release.",
        });
        return findings;
    }

    // DROP VIEW
    if normalized.starts_with("drop view") {
        findings.push(SafetyFinding {
            operation: "DROP VIEW".to_owned(),
            risk: RiskLevel::Destructive,
            why: "Drops the view. Old replicas that query this view will error immediately \
                  during a rolling deploy.",
            next_action: "Use expand/contract: first deploy code that no longer references the \
                          view, then drop it in a subsequent release.",
        });
        return findings;
    }

    // DROP SEQUENCE
    if normalized.starts_with("drop sequence") {
        findings.push(SafetyFinding {
            operation: "DROP SEQUENCE".to_owned(),
            risk: RiskLevel::Destructive,
            why: "Dropping a sequence breaks any column that uses it as a default \
                  (`nextval(seq)`) and any application code that calls `nextval` directly. \
                  Old replicas will error immediately on INSERT.",
            next_action: "Use expand/contract: first deploy code that no longer relies on this \
                          sequence, then drop it in a subsequent release.",
        });
        return findings;
    }

    // TRUNCATE TABLE
    if normalized.starts_with("truncate ") {
        findings.push(SafetyFinding {
            operation: "TRUNCATE".to_owned(),
            risk: RiskLevel::Destructive,
            why: "Truncating a table deletes all data and acquires an AccessExclusiveLock, \
                  blocking all concurrent reads and writes.",
            next_action: "If you need to empty the table, delete rows in small batches, or perform \
                          the truncate during a coordinated maintenance window.",
        });
        return findings;
    }

    // DROP COLUMN
    if normalized.contains(" drop column ") {
        findings.push(SafetyFinding {
            operation: "DROP COLUMN".to_owned(),
            risk: RiskLevel::Destructive,
            why: if sqlite {
                "Removes a column and its data. SQLite also refuses the statement outright in \
                 several cases. The column must not be a primary key, UNIQUE, or named by any \
                 index — a partial index's WHERE clause included. It must not appear in a CHECK \
                 constraint, a generated column, a foreign key, a view or a trigger."
            } else {
                "Removes a column and its data. Old replicas that SELECT or INSERT this column \
                 will error until they restart."
            },
            next_action: if sqlite {
                "DROP INDEX every index that names this column earlier in the same migration. \
                 For a primary-key, UNIQUE, CHECK or generated-column reference, rebuild the \
                 table instead (`autumn schema diff --write-migration` emits the rebuild)."
            } else {
                "Use expand/contract: first deploy code that no longer reads or writes \
                 this column, then drop it in the next release."
            },
        });
    }

    // RENAME COLUMN
    if normalized.contains(" rename column ") {
        findings.push(SafetyFinding {
            operation: "RENAME COLUMN".to_owned(),
            risk: RiskLevel::Irreversible,
            why: "Renaming a column breaks queries from old replicas that still reference the \
                  old name, causing errors during a rolling deploy.",
            next_action: "Use expand/contract: add the new column, dual-write, backfill existing \
                          rows, update all code, then drop the old column.",
        });
    }

    // RENAME TABLE
    if normalized.contains("alter table")
        && normalized.contains(" rename to ")
        && !normalized.contains(" rename column ")
    {
        findings.push(SafetyFinding {
            operation: "RENAME TABLE".to_owned(),
            risk: RiskLevel::Irreversible,
            why: "Renaming a table breaks all queries from old replicas that reference the \
                  original name.",
            next_action: "Create a view under the old name while the new name rolls out, or \
                          coordinate a maintenance window.",
        });
    }

    // SQLite's ALTER TABLE grammar is exactly RENAME TO, RENAME [COLUMN],
    // ADD COLUMN and DROP COLUMN. Every other subcommand — ALTER COLUMN in any
    // spelling, ADD/DROP CONSTRAINT, SET SCHEMA, OWNER TO — is a parse error, so
    // one rule covers them all rather than a list that can fall behind Postgres.
    if sqlite && normalized.starts_with("alter table") {
        let subcommands = alter_table_subcommands(backend, normalized);
        // SQLite takes exactly one action per ALTER TABLE; Postgres allows a
        // comma-separated list.
        if subcommands.len() > 1 {
            findings.push(SafetyFinding {
                operation: "Multi-action ALTER TABLE (unsupported on SQLite)".to_owned(),
                risk: RiskLevel::Unsupported,
                why: "SQLite takes one action per ALTER TABLE. A comma-separated list is a parse \
                      error, so this statement fails at apply time.",
                next_action: "Split the statement: one ALTER TABLE per action.",
            });
        }
        for subcommand in subcommands {
            if is_sqlite_alter_subcommand(subcommand) {
                continue;
            }
            let is_alter_column = subcommand.starts_with("alter column ");
            findings.push(SafetyFinding {
                operation: if is_alter_column {
                    "ALTER COLUMN (unsupported on SQLite)".to_owned()
                } else {
                    "ALTER TABLE subcommand (unsupported on SQLite)".to_owned()
                },
                risk: RiskLevel::Unsupported,
                why: "SQLite's ALTER TABLE supports only RENAME, ADD COLUMN and DROP COLUMN. \
                      Any other subcommand is a parse error, so this statement fails at apply \
                      time.",
                next_action: "Rebuild the table: create the new-shape table, copy the rows, drop \
                              the old table, rename the new one. `autumn schema diff \
                              --write-migration` emits that rebuild for you.",
            });
            break; // one finding per statement is enough
        }
    }

    // ALTER COLUMN TYPE (Postgres)
    if !sqlite
        && let Some(i) = normalized.find("alter column")
        && normalized[i..].contains(" type ")
    {
        findings.push(SafetyFinding {
            operation: "ALTER COLUMN TYPE".to_owned(),
            risk: RiskLevel::Destructive,
            why: "Changing a column's type rewrites the column data and may be incompatible \
                  with values read by old replicas or application code.",
            next_action: "Add a new column with the target type, migrate data, update code to \
                          use the new column, then drop the old one.",
        });
    }

    // ADD COLUMN NOT NULL — checked per subcommand so that a DEFAULT on one column
    // in a multi-column ALTER TABLE does not suppress the check for other columns.
    //
    // Two unsafe cases:
    //  1. No DEFAULT at all — Postgres must validate every existing row under lock.
    //  2. Volatile DEFAULT (contains a function call) — Postgres must evaluate the
    //     function per-row and cannot use the fast constant-default path (PG 11+),
    //     so the table is still rewritten under the exclusive lock.
    if normalized.starts_with("alter table") {
        for subcommand in alter_table_subcommands(backend, normalized) {
            if subcommand.starts_with("add column ") && subcommand.contains("not null") {
                // SQLite drops a literal `DEFAULT NULL` before applying the
                // NOT NULL check, so it is not a default there.
                let default_token = add_column_default_token(subcommand);
                let has_default = if sqlite {
                    default_token.as_deref().is_some_and(|d| d != "null")
                } else {
                    subcommand.contains(" default ")
                };
                // A DEFAULT is "volatile" when it is a VOLATILE function call —
                // i.e. one that Postgres must evaluate per-row, preventing the PG11+
                // fast-constant-default path.  STABLE functions (e.g. `now()`) are
                // evaluated once at statement time and do not require a table rewrite.
                let has_volatile_default = has_default
                    && subcommand.find(" default ").is_some_and(|pos| {
                        let default_expr = subcommand[pos + " default ".len()..].trim();
                        is_volatile_function_default(default_expr)
                    });

                if !has_default {
                    findings.push(SafetyFinding {
                        operation: "ADD COLUMN NOT NULL (no default)".to_owned(),
                        // SQLite does not merely slow down here: it rejects the
                        // statement ("Cannot add a NOT NULL column with default
                        // value NULL"), so this can never apply.
                        risk: if sqlite && !fresh_table {
                            RiskLevel::Unsupported
                        } else if sqlite {
                            // A table created in this same migration has no rows,
                            // so SQLite accepts the statement; Postgres still
                            // validates nothing on an empty table either.
                            RiskLevel::Safe
                        } else {
                            RiskLevel::PotentiallyBlocking
                        },
                        why: if sqlite {
                            "SQLite rejects ALTER TABLE … ADD COLUMN … NOT NULL without a \
                             DEFAULT outright — the statement fails at apply time."
                        } else {
                            "Adding a NOT NULL column without a DEFAULT forces Postgres to \
                             validate every existing row under an exclusive lock. On a large \
                             table this may time out."
                        },
                        next_action: if sqlite {
                            "Give the column a constant DEFAULT, or add it as nullable and \
                             backfill. SQLite cannot add the NOT NULL constraint afterwards \
                             (no ALTER COLUMN) — that needs a table rebuild."
                        } else {
                            "Provide a constant DEFAULT value, or add the column as \
                             nullable first, backfill existing rows, then add the NOT \
                             NULL constraint in a later migration."
                        },
                    });
                    break; // one finding per statement is sufficient
                } else if !sqlite && has_volatile_default {
                    findings.push(SafetyFinding {
                        operation: "ADD COLUMN NOT NULL (volatile default)".to_owned(),
                        risk: RiskLevel::PotentiallyBlocking,
                        why: "A volatile function-call DEFAULT (e.g. random(), gen_random_uuid()) \
                              is evaluated per-row: Postgres cannot use the PG11+ fast-constant \
                              path and must rewrite the entire table under an exclusive lock.",
                        next_action: "Use a constant literal DEFAULT instead (e.g. DEFAULT 0, \
                                      DEFAULT ''), or add the column nullable, backfill, then \
                                      add the NOT NULL constraint in a later migration.",
                    });
                    break; // one finding per statement is sufficient
                }
            }
        }
    }

    // ADD COLUMN with inline UNIQUE — implicitly builds a non-concurrent unique index
    // ADD COLUMN with inline REFERENCES — scans existing rows to validate the FK
    if normalized.starts_with("alter table") {
        for subcommand in alter_table_subcommands(backend, normalized) {
            if !subcommand.starts_with("add column ") {
                continue;
            }
            if has_inline_unique_constraint(&blank_quoted_spans(backend, subcommand)) {
                findings.push(if sqlite {
                    // SQLite: "Cannot add a UNIQUE column".
                    SafetyFinding {
                        operation: "ADD COLUMN UNIQUE (unsupported on SQLite)".to_owned(),
                        risk: RiskLevel::Unsupported,
                        why: "SQLite rejects ALTER TABLE … ADD COLUMN with an inline UNIQUE \
                              constraint — the statement fails at apply time.",
                        next_action: "Add the column without UNIQUE, then CREATE UNIQUE INDEX on \
                                      it in the same migration.",
                    }
                } else {
                    SafetyFinding {
                        operation: "ADD COLUMN UNIQUE (inline constraint)".to_owned(),
                        risk: RiskLevel::PotentiallyBlocking,
                        why: "An inline UNIQUE constraint implicitly builds a non-concurrent \
                              unique index under an exclusive table lock, blocking all reads and \
                              writes during the build.",
                        next_action: "Add the column without UNIQUE first, then create the unique \
                                      index in a separate migration using \
                                      `CREATE UNIQUE INDEX CONCURRENTLY`.",
                    }
                });
            }
            if sqlite && subcommand.contains(" primary key") {
                findings.push(SafetyFinding {
                    operation: "ADD COLUMN PRIMARY KEY (unsupported on SQLite)".to_owned(),
                    risk: RiskLevel::Unsupported,
                    why: "SQLite rejects ALTER TABLE … ADD COLUMN with an inline PRIMARY KEY \
                          constraint — the statement fails at apply time.",
                    next_action: "Rebuild the table with the new primary key \
                                  (`autumn schema diff --write-migration` emits the rebuild).",
                });
            }
            // SQLite accepts an added REFERENCES column but requires its default
            // to be NULL, and never scans existing rows to validate the key — so
            // the Postgres row-validation finding does not apply there.
            if sqlite {
                // A generated column is the one ADD COLUMN form left; SQLite
                // accepts VIRTUAL and rejects STORED.
                if !fresh_table && subcommand.contains(" stored") && subcommand.contains(" as (") {
                    findings.push(SafetyFinding {
                        operation: "ADD COLUMN GENERATED … STORED (unsupported on SQLite)"
                            .to_owned(),
                        risk: RiskLevel::Unsupported,
                        why: "SQLite cannot add a STORED generated column to an existing table \
                              — the statement fails at apply time. VIRTUAL is accepted.",
                        next_action: "Use a VIRTUAL generated column, or rebuild the table with \
                                      the STORED column in its CREATE TABLE.",
                    });
                }
                // SQLite's ADD COLUMN default must reduce to a constant.
                // NOTE: no rule for `REFERENCES` with a non-NULL default. SQLite
                // enforces that only when `PRAGMA foreign_keys` is ON, and
                // Autumn's migration connection deliberately leaves it OFF (see
                // `establish_sqlite_migration_connection`), so the statement does
                // apply on the path this classifier grades.
                if !fresh_table
                    && let Some(default) = add_column_default_token(subcommand)
                    && !is_sqlite_constant_default(&default)
                {
                    findings.push(SafetyFinding {
                        operation: "ADD COLUMN (non-constant default)".to_owned(),
                        risk: RiskLevel::Unsupported,
                        why: "SQLite's ADD COLUMN default must reduce to a constant: \
                              CURRENT_TIME/CURRENT_DATE/CURRENT_TIMESTAMP, a call, and any other \
                              expression are rejected at apply time.",
                        next_action: "Use a literal DEFAULT, or add the column nullable and \
                                      backfill it with an UPDATE.",
                    });
                }
            } else if subcommand.contains(" references ") {
                findings.push(SafetyFinding {
                    operation: "ADD COLUMN REFERENCES (inline FK)".to_owned(),
                    risk: RiskLevel::PotentiallyBlocking,
                    why: "An inline REFERENCES constraint scans all existing rows to validate the \
                          foreign key, acquiring locks that can block writes on the referenced \
                          table.",
                    next_action: "Add the column without the constraint first, then add the FK \
                                  using `ADD CONSTRAINT ... FOREIGN KEY ... NOT VALID` and \
                                  validate separately with `VALIDATE CONSTRAINT`.",
                });
            }
        }
    }

    // Unclassified ALTER TABLE subcommand — fires when any subcommand in the statement
    // is not covered by the specific rules above. Checking all subcommands individually
    // prevents a known-safe subcommand (e.g. ADD COLUMN) from hiding an unknown one
    // (e.g. DROP CONSTRAINT) in the same multi-action ALTER TABLE.
    if normalized.starts_with("alter table") {
        let subcommands = alter_table_subcommands(backend, normalized);
        let all_known = subcommands
            .iter()
            .all(|s| is_known_alter_subcommand(backend, s));
        if !all_known {
            findings.push(SafetyFinding {
                operation: "Unclassified ALTER TABLE".to_owned(),
                risk: RiskLevel::ManualReview,
                why: "Autumn cannot automatically assess the safety of this ALTER TABLE \
                      subcommand for a rolling deploy. Some operations (e.g. DROP CONSTRAINT, \
                      ALTER COLUMN SET NOT NULL, ADD CONSTRAINT) acquire table locks or validate \
                      existing rows.",
                next_action: "Review the statement manually. If it is safe, you may suppress \
                              this finding by adding `-- autumn-safety: reviewed` above the \
                              statement.",
            });
        }
    }

    // CREATE INDEX / CREATE UNIQUE INDEX without CONCURRENTLY
    let is_create_index =
        normalized.starts_with("create index") || normalized.starts_with("create unique index");
    // The trailing space matters: `concurrently` must be the SQL option, not the
    // start of the index's name (`CREATE INDEX concurrently_idx ON …`).
    let is_concurrent = normalized.starts_with("create index concurrently ")
        || normalized.starts_with("create unique index concurrently ");
    if is_create_index && !is_concurrent {
        findings.push(SafetyFinding {
            operation: if sqlite {
                CREATE_INDEX_SQLITE_OP
            } else {
                CREATE_INDEX_PG_OP
            }
            .to_owned(),
            risk: RiskLevel::PotentiallyBlocking,
            why: if sqlite {
                "Building an index takes SQLite's single write lock for the whole build. \
                 Writers block until it finishes; under WAL, readers do not."
            } else {
                "Non-concurrent index creation holds an exclusive table lock for the entire \
                 build, blocking all reads and writes."
            },
            next_action: if sqlite {
                "SQLite has no online index build. Build the index during a low-traffic window, \
                 and keep the migration's transaction short."
            } else {
                "Use CREATE INDEX CONCURRENTLY instead. Note: concurrent index \
                 creation cannot run inside a transaction block."
            },
        });
    }

    // Data backfill — bulk DML inside a migration requires a separate job
    if normalized.starts_with("update ")
        || normalized.starts_with("insert into ")
        || normalized.starts_with("delete from ")
        || normalized.starts_with("merge into ")
    {
        findings.push(SafetyFinding {
            operation: "Bulk DML (data backfill)".to_owned(),
            risk: RiskLevel::DataBackfill,
            why: "Running bulk UPDATE or INSERT inside a migration locks rows for the duration \
                  of the transaction. On large tables this can time out or block application \
                  traffic for seconds to minutes.",
            next_action: "Run the data backfill as a separate idempotent background job or \
                          one-off task (`autumn task`) after the schema migration has deployed. \
                          Add a NOT VALID constraint first if you need the constraint enforced \
                          before the backfill completes.",
        });
    }

    // CTE-prefixed bulk DML — WITH … UPDATE / DELETE / INSERT
    // A CTE starts with `with` so the plain DML checks above don't fire.
    // Check both the outer statement (`) update/delete/insert`) and CTE bodies
    // (`(update/delete/insert`) to catch data-modifying CTEs followed by SELECT.
    if normalized.starts_with("with ")
        && (normalized.contains(") update ")
            || normalized.contains(") delete ")
            || normalized.contains(") insert into ")
            || normalized.contains("(update ")
            || normalized.contains("(delete ")
            || normalized.contains("(insert into "))
    {
        findings.push(SafetyFinding {
            operation: "Bulk DML (data backfill via CTE)".to_owned(),
            risk: RiskLevel::DataBackfill,
            why: "A CTE that writes (UPDATE, DELETE, INSERT) locks rows for the duration of the \
                  transaction. On large tables this can time out or block application traffic.",
            next_action: "Run the data backfill as a separate idempotent background job or \
                          one-off task (`autumn task`) after the schema migration has deployed.",
        });
    }

    // DROP INDEX (non-concurrent) — holds an exclusive table lock
    // Use token-aware check: `concurrently` must be the SQL option immediately after
    // `drop index`, not a substring of the index name (e.g. idx_concurrently).
    // Postgres only: on SQLite a DROP INDEX is a cheap catalog edit, and the
    // generated SQLite `DROP COLUMN` path is *required* to emit one first —
    // flagging it would fail the deploy gate on every such migration.
    if !sqlite
        && normalized.starts_with("drop index")
        && !normalized.starts_with("drop index concurrently ")
    {
        findings.push(SafetyFinding {
            operation: "DROP INDEX (non-concurrent)".to_owned(),
            risk: RiskLevel::PotentiallyBlocking,
            why: "Non-concurrent DROP INDEX acquires an AccessExclusiveLock on the table, \
                  blocking all reads and writes for the duration of the drop.",
            next_action: "Use DROP INDEX CONCURRENTLY to avoid the exclusive table lock. \
                          Add `run_in_transaction = false` to the migration's `metadata.toml`.",
        });
    }

    // ALTER TYPE RENAME VALUE — renaming an enum label breaks old replicas that
    // still INSERT or compare against the old label during a rolling deploy.
    if !sqlite && normalized.starts_with("alter type") && normalized.contains(" rename value ") {
        findings.push(SafetyFinding {
            operation: "ALTER TYPE RENAME VALUE".to_owned(),
            risk: RiskLevel::Irreversible,
            why: "Renaming an enum label breaks old replicas that still insert, compare, or \
                  decode the old label. Errors will appear immediately during a rolling deploy.",
            next_action: "Use expand/contract: add a new enum value, migrate all writes to use \
                          it, then remove the old value in a subsequent release.",
        });
        return findings;
    }

    // ALTER TYPE RENAME TO — renaming the type itself breaks references in old replicas.
    if !sqlite && normalized.starts_with("alter type") && normalized.contains(" rename to ") {
        findings.push(SafetyFinding {
            operation: "ALTER TYPE RENAME".to_owned(),
            risk: RiskLevel::Irreversible,
            why: "Renaming a type breaks all references to its old name in old replicas, \
                  causing errors during a rolling deploy.",
            next_action: "Coordinate a maintenance window or use expand/contract by creating \
                          the new type, migrating columns/code, then dropping the old one.",
        });
        return findings;
    }

    // Generic catch-all — DDL/DML not matched by any rule above
    let is_known_start = normalized.starts_with("drop table")
        || normalized.starts_with("drop index")
        || normalized.starts_with("alter table") // unclassified subcommands handled above
        || normalized.starts_with("create table")
        || normalized.starts_with("create index")
        || normalized.starts_with("create unique index")
        || normalized.starts_with("update ")
        || normalized.starts_with("insert into ")
        || normalized.starts_with("delete from ")
        || normalized.starts_with("merge into ")
        || normalized.starts_with("truncate ")
        || normalized.starts_with("comment on")
        || normalized.starts_with("create sequence")
        || normalized.starts_with("alter sequence")
        || normalized.starts_with("drop sequence")
        || normalized.starts_with("create type")
        // `alter type` is intentionally absent — unclassified forms fall through to ManualReview
        // `drop type` is intentionally absent — falls through to ManualReview
        || normalized.starts_with("create extension")
        || normalized.starts_with("create view")
        || normalized.starts_with("drop view")
        || normalized.starts_with("select ");

    let starts_with_ddl_keyword = normalized.starts_with("create ")
        || normalized.starts_with("drop ")
        || normalized.starts_with("alter ")
        || normalized.starts_with("truncate ")
        || normalized.starts_with("grant ")
        || normalized.starts_with("revoke ");

    if starts_with_ddl_keyword && !is_known_start {
        findings.push(SafetyFinding {
            operation: "Unclassified DDL".to_owned(),
            risk: RiskLevel::ManualReview,
            why: "Autumn cannot automatically assess the safety of this statement for a rolling \
                  deploy. Operator review is required before applying this migration in \
                  production.",
            next_action: "Review the statement manually. If it is safe, you may suppress this \
                          finding by adding `-- autumn-safety: reviewed` above the statement.",
        });
    }

    findings
}

// ── tests ─────────────────────────────────────────────────────────────────────

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

    // ── RiskLevel ordering ────────────────────────────────────────────────────

    #[test]
    fn risk_level_ordering() {
        assert!(RiskLevel::Safe < RiskLevel::PotentiallyBlocking);
        assert!(RiskLevel::PotentiallyBlocking < RiskLevel::Destructive);
        assert!(RiskLevel::Destructive < RiskLevel::Irreversible);
        assert!(RiskLevel::Irreversible < RiskLevel::DataBackfill);
        assert!(RiskLevel::DataBackfill < RiskLevel::ManualReview);
    }

    #[test]
    fn risk_level_display() {
        assert_eq!(RiskLevel::Safe.to_string(), "safe");
        assert_eq!(
            RiskLevel::PotentiallyBlocking.to_string(),
            "potentially-blocking"
        );
        assert_eq!(RiskLevel::Destructive.to_string(), "destructive");
        assert_eq!(RiskLevel::Irreversible.to_string(), "irreversible");
        assert_eq!(RiskLevel::DataBackfill.to_string(), "data-backfill");
        assert_eq!(RiskLevel::ManualReview.to_string(), "manual-review");
    }

    // ── safe migrations ───────────────────────────────────────────────────────

    #[test]
    fn empty_sql_has_no_findings() {
        assert!(classify_sql("").is_empty());
    }

    #[test]
    fn create_table_is_safe() {
        let sql = "CREATE TABLE posts (\n id BIGSERIAL PRIMARY KEY,\n title TEXT NOT NULL\n);";
        let findings = classify_sql(sql);
        assert!(
            findings.is_empty(),
            "CREATE TABLE should be safe: {findings:?}"
        );
    }

    #[test]
    fn add_nullable_column_is_safe() {
        let sql = "ALTER TABLE posts ADD COLUMN subtitle TEXT NULL;";
        let findings = classify_sql(sql);
        assert!(
            findings.is_empty(),
            "ADD COLUMN NULL should be safe: {findings:?}"
        );
    }

    #[test]
    fn add_not_null_column_with_default_is_safe() {
        let sql = "ALTER TABLE posts ADD COLUMN status TEXT NOT NULL DEFAULT 'draft';";
        let findings = classify_sql(sql);
        assert!(
            findings.is_empty(),
            "ADD COLUMN NOT NULL DEFAULT should be safe: {findings:?}"
        );
    }

    #[test]
    fn add_not_null_column_name_containing_default_is_blocking() {
        // Column named `defaulted_at` must not be mistaken for having a DEFAULT clause.
        let sql = "ALTER TABLE posts ADD COLUMN defaulted_at TIMESTAMP NOT NULL;";
        let findings = classify_sql(sql);
        assert_eq!(
            findings.len(),
            1,
            "column name containing 'default' must not suppress finding"
        );
        assert_eq!(findings[0].risk, RiskLevel::PotentiallyBlocking);
    }

    #[test]
    fn create_concurrent_index_is_safe() {
        let sql = "CREATE INDEX CONCURRENTLY idx_posts_title ON posts (title);";
        let findings = classify_sql(sql);
        assert!(
            findings.is_empty(),
            "CREATE INDEX CONCURRENTLY should be safe: {findings:?}"
        );
    }

    #[test]
    fn create_unique_index_concurrently_is_safe() {
        let sql = "CREATE UNIQUE INDEX CONCURRENTLY idx_posts_slug ON posts (slug);";
        let findings = classify_sql(sql);
        assert!(
            findings.is_empty(),
            "CREATE UNIQUE INDEX CONCURRENTLY should be safe: {findings:?}"
        );
    }

    // ── destructive patterns ──────────────────────────────────────────────────

    #[test]
    fn drop_view_is_destructive() {
        let findings = classify_sql("DROP VIEW active_posts;");
        assert_eq!(findings.len(), 1);
        assert_eq!(findings[0].risk, RiskLevel::Destructive);
        assert_eq!(findings[0].operation, "DROP VIEW");
    }

    #[test]
    fn drop_view_cascade_is_destructive() {
        let findings = classify_sql("DROP VIEW active_posts CASCADE;");
        assert_eq!(findings.len(), 1);
        assert_eq!(findings[0].risk, RiskLevel::Destructive);
    }

    #[test]
    fn drop_table_is_destructive() {
        let findings = classify_sql("DROP TABLE posts;");
        assert_eq!(findings.len(), 1);
        assert_eq!(findings[0].risk, RiskLevel::Destructive);
        assert_eq!(findings[0].operation, "DROP TABLE");
    }

    #[test]
    fn drop_sequence_is_destructive() {
        let findings = classify_sql("DROP SEQUENCE posts_id_seq;");
        assert_eq!(findings.len(), 1);
        assert_eq!(findings[0].risk, RiskLevel::Destructive);
        assert_eq!(findings[0].operation, "DROP SEQUENCE");
    }

    #[test]
    fn drop_sequence_cascade_is_destructive() {
        let findings = classify_sql("DROP SEQUENCE posts_id_seq CASCADE;");
        assert_eq!(findings.len(), 1);
        assert_eq!(findings[0].risk, RiskLevel::Destructive);
    }

    #[test]
    fn drop_column_is_destructive() {
        let findings = classify_sql("ALTER TABLE posts DROP COLUMN title;");
        assert_eq!(findings.len(), 1);
        assert_eq!(findings[0].risk, RiskLevel::Destructive);
        assert_eq!(findings[0].operation, "DROP COLUMN");
    }

    #[test]
    fn drop_column_case_insensitive() {
        let findings = classify_sql("alter table posts drop column title;");
        assert_eq!(findings.len(), 1);
        assert_eq!(findings[0].risk, RiskLevel::Destructive);
    }

    #[test]
    fn alter_column_type_is_destructive() {
        let findings = classify_sql("ALTER TABLE posts ALTER COLUMN score TYPE BIGINT;");
        assert_eq!(findings.len(), 1);
        assert_eq!(findings[0].risk, RiskLevel::Destructive);
        assert_eq!(findings[0].operation, "ALTER COLUMN TYPE");
    }

    // ── irreversible patterns ─────────────────────────────────────────────────

    #[test]
    fn rename_column_is_irreversible() {
        let findings = classify_sql("ALTER TABLE posts RENAME COLUMN body TO content;");
        assert_eq!(findings.len(), 1);
        assert_eq!(findings[0].risk, RiskLevel::Irreversible);
        assert_eq!(findings[0].operation, "RENAME COLUMN");
    }

    #[test]
    fn rename_table_is_irreversible() {
        let findings = classify_sql("ALTER TABLE posts RENAME TO articles;");
        assert_eq!(findings.len(), 1);
        assert_eq!(findings[0].risk, RiskLevel::Irreversible);
        assert_eq!(findings[0].operation, "RENAME TABLE");
    }

    #[test]
    fn alter_type_rename_value_is_irreversible() {
        let findings = classify_sql("ALTER TYPE status RENAME VALUE 'draft' TO 'pending';");
        assert_eq!(
            findings.len(),
            1,
            "ALTER TYPE RENAME VALUE must be flagged: {findings:?}"
        );
        assert_eq!(findings[0].risk, RiskLevel::Irreversible);
        assert_eq!(findings[0].operation, "ALTER TYPE RENAME VALUE");
    }

    #[test]
    fn alter_type_rename_to_is_irreversible() {
        let findings = classify_sql("ALTER TYPE order_status RENAME TO status;");
        assert_eq!(
            findings.len(),
            1,
            "ALTER TYPE RENAME TO must be flagged: {findings:?}"
        );
        assert_eq!(findings[0].risk, RiskLevel::Irreversible);
        assert_eq!(findings[0].operation, "ALTER TYPE RENAME");
    }

    #[test]
    fn alter_type_add_value_requires_manual_review() {
        // ADD VALUE is not specifically classified — operator must review.
        let findings = classify_sql("ALTER TYPE status ADD VALUE 'archived';");
        assert_eq!(
            findings.len(),
            1,
            "unclassified ALTER TYPE must require manual review: {findings:?}"
        );
        assert_eq!(findings[0].risk, RiskLevel::ManualReview);
    }

    #[test]
    fn rename_constraint_requires_manual_review() {
        // RENAME CONSTRAINT is a schema change that Autumn cannot auto-classify —
        // it must not silently pass as safe.
        let findings = classify_sql("ALTER TABLE users RENAME CONSTRAINT old_name TO new_name;");
        assert_eq!(
            findings.len(),
            1,
            "RENAME CONSTRAINT must not silently pass: {findings:?}"
        );
        assert_eq!(findings[0].risk, RiskLevel::ManualReview);
        assert_eq!(findings[0].operation, "Unclassified ALTER TABLE");
    }

    // ── potentially blocking patterns ─────────────────────────────────────────

    #[test]
    fn multi_column_add_only_flags_clause_without_default() {
        // ADD COLUMN score has DEFAULT — ADD COLUMN slug does NOT. Only slug should be flagged.
        let sql = "ALTER TABLE posts \
                   ADD COLUMN score INT NOT NULL DEFAULT 0, \
                   ADD COLUMN slug TEXT NOT NULL;";
        let findings = classify_sql(sql);
        assert_eq!(
            findings.len(),
            1,
            "only the column without a DEFAULT should be flagged"
        );
        assert_eq!(findings[0].risk, RiskLevel::PotentiallyBlocking);
    }

    #[test]
    fn mixed_known_and_unknown_alter_table_subcommands_flagged() {
        // ADD COLUMN is safe, but DROP CONSTRAINT is unclassified — should get ManualReview.
        let sql = "ALTER TABLE posts ADD COLUMN subtitle TEXT, DROP CONSTRAINT posts_title_key;";
        let findings = classify_sql(sql);
        assert!(
            findings.iter().any(|f| f.risk == RiskLevel::ManualReview),
            "unknown subcommand must produce ManualReview: {findings:?}"
        );
    }

    #[test]
    fn add_not_null_column_without_default_is_blocking() {
        let findings = classify_sql("ALTER TABLE posts ADD COLUMN score INTEGER NOT NULL;");
        assert_eq!(findings.len(), 1);
        assert_eq!(findings[0].risk, RiskLevel::PotentiallyBlocking);
        assert_eq!(findings[0].operation, "ADD COLUMN NOT NULL (no default)");
    }

    #[test]
    fn add_not_null_column_with_volatile_default_is_blocking() {
        let findings = classify_sql(
            "ALTER TABLE posts ADD COLUMN token UUID NOT NULL DEFAULT gen_random_uuid();",
        );
        assert_eq!(findings.len(), 1);
        assert_eq!(findings[0].risk, RiskLevel::PotentiallyBlocking);
        assert_eq!(
            findings[0].operation,
            "ADD COLUMN NOT NULL (volatile default)"
        );
    }

    #[test]
    fn add_not_null_column_with_now_default_is_safe() {
        // now() is STABLE: Postgres evaluates it once at statement time and stores the
        // constant, so the PG11+ fast-default path applies — no table rewrite needed.
        let findings = classify_sql(
            "ALTER TABLE posts ADD COLUMN created_at TIMESTAMP NOT NULL DEFAULT now();",
        );
        assert!(
            findings.is_empty(),
            "DEFAULT now() is stable and must not be flagged as volatile: {findings:?}"
        );
    }

    #[test]
    fn add_not_null_column_with_parenthesized_constant_default_is_safe() {
        // `DEFAULT (0)` and `DEFAULT ('draft')` are parenthesized constants, not function
        // calls.  They use the PG11+ fast-default path and must not be flagged as volatile.
        let findings_int =
            classify_sql("ALTER TABLE posts ADD COLUMN score INT NOT NULL DEFAULT (0);");
        assert!(
            findings_int.is_empty(),
            "DEFAULT (0) must be safe: {findings_int:?}"
        );
        let findings_str =
            classify_sql("ALTER TABLE posts ADD COLUMN status TEXT NOT NULL DEFAULT ('draft');");
        assert!(
            findings_str.is_empty(),
            "DEFAULT ('draft') must be safe: {findings_str:?}"
        );
    }

    #[test]
    fn add_not_null_column_with_constant_default_is_safe() {
        // Constant literals use the PG11+ fast path — no table rewrite.
        let findings =
            classify_sql("ALTER TABLE posts ADD COLUMN active BOOLEAN NOT NULL DEFAULT false;");
        assert!(
            findings.is_empty(),
            "constant DEFAULT false must be safe: {findings:?}"
        );
    }

    #[test]
    fn add_column_with_inline_unique_is_potentially_blocking() {
        let findings = classify_sql("ALTER TABLE posts ADD COLUMN slug TEXT UNIQUE;");
        assert_eq!(
            findings.len(),
            1,
            "inline UNIQUE must be flagged: {findings:?}"
        );
        assert_eq!(findings[0].risk, RiskLevel::PotentiallyBlocking);
        assert_eq!(
            findings[0].operation,
            "ADD COLUMN UNIQUE (inline constraint)"
        );
    }

    #[test]
    fn add_column_with_inline_references_is_potentially_blocking() {
        let findings =
            classify_sql("ALTER TABLE posts ADD COLUMN user_id INT REFERENCES users(id);");
        assert_eq!(
            findings.len(),
            1,
            "inline REFERENCES must be flagged: {findings:?}"
        );
        assert_eq!(findings[0].risk, RiskLevel::PotentiallyBlocking);
        assert_eq!(findings[0].operation, "ADD COLUMN REFERENCES (inline FK)");
    }

    #[test]
    fn add_column_with_primary_key_requires_manual_review() {
        // PRIMARY KEY inline on ADD COLUMN is not specifically classified.
        let findings = classify_sql("ALTER TABLE posts ADD COLUMN id BIGSERIAL PRIMARY KEY;");
        assert!(
            findings.iter().any(|f| f.risk == RiskLevel::ManualReview),
            "inline PRIMARY KEY must require manual review: {findings:?}"
        );
    }

    #[test]
    fn add_column_without_constraints_does_not_trigger_manual_review() {
        let findings = classify_sql("ALTER TABLE posts ADD COLUMN subtitle TEXT;");
        assert!(
            findings.iter().all(|f| f.risk != RiskLevel::ManualReview),
            "simple ADD COLUMN must not trigger ManualReview: {findings:?}"
        );
    }

    #[test]
    fn create_non_concurrent_index_is_blocking() {
        let findings = classify_sql("CREATE INDEX idx_posts_title ON posts (title);");
        assert_eq!(findings.len(), 1);
        assert_eq!(findings[0].risk, RiskLevel::PotentiallyBlocking);
        assert_eq!(findings[0].operation, "CREATE INDEX (non-concurrent)");
    }

    #[test]
    fn create_unique_index_without_concurrently_is_blocking() {
        let findings = classify_sql("CREATE UNIQUE INDEX idx_posts_slug ON posts (slug);");
        assert_eq!(findings.len(), 1);
        assert_eq!(findings[0].risk, RiskLevel::PotentiallyBlocking);
        assert_eq!(findings[0].operation, "CREATE INDEX (non-concurrent)");
    }

    #[test]
    fn non_concurrent_index_on_newly_created_table_is_safe() {
        // The table is created in the same migration — no existing rows to lock.
        // This is the shape emitted by `autumn generate ... --index`.
        let sql = "CREATE TABLE posts (id BIGSERIAL PRIMARY KEY, title TEXT NOT NULL);\n\
                   CREATE INDEX idx_posts_title ON posts (title);";
        let findings = classify_sql(sql);
        assert!(
            findings.is_empty(),
            "non-concurrent index on a table created in the same migration must be safe: \
             {findings:?}"
        );
    }

    #[test]
    fn unique_non_concurrent_index_on_newly_created_table_is_safe() {
        let sql = "CREATE TABLE posts (id BIGSERIAL PRIMARY KEY, slug TEXT NOT NULL);\n\
                   CREATE UNIQUE INDEX idx_posts_slug ON posts (slug);";
        let findings = classify_sql(sql);
        assert!(
            findings.is_empty(),
            "non-concurrent unique index on a newly created table must be safe: {findings:?}"
        );
    }

    #[test]
    fn non_concurrent_index_on_different_table_is_still_blocking() {
        // CREATE TABLE `posts` does not exempt an index on a different table `comments`.
        let sql = "CREATE TABLE posts (id BIGSERIAL PRIMARY KEY);\n\
                   CREATE INDEX idx_comments_post_id ON comments (post_id);";
        let findings = classify_sql(sql);
        assert_eq!(
            findings.len(),
            1,
            "non-concurrent index on a pre-existing table must still be flagged: {findings:?}"
        );
        assert_eq!(findings[0].risk, RiskLevel::PotentiallyBlocking);
    }

    #[test]
    fn if_not_exists_table_not_treated_as_newly_created() {
        let sql = "CREATE TABLE IF NOT EXISTS posts (id BIGSERIAL PRIMARY KEY, title TEXT NOT NULL);\n\
                   CREATE INDEX idx_posts_title ON posts (title);";
        let findings = classify_sql(sql);
        assert_eq!(
            findings.len(),
            1,
            "non-concurrent index on IF NOT EXISTS table must still be flagged"
        );
        assert_eq!(findings[0].risk, RiskLevel::PotentiallyBlocking);
    }

    #[test]
    fn truncate_table_is_destructive() {
        let findings = classify_sql("TRUNCATE TABLE posts;");
        assert_eq!(findings.len(), 1);
        assert_eq!(findings[0].risk, RiskLevel::Destructive);
        assert_eq!(findings[0].operation, "TRUNCATE");
    }

    // ── multi-statement SQL ───────────────────────────────────────────────────

    #[test]
    fn multiple_safe_statements_produce_no_findings() {
        let sql = "\
            ALTER TABLE posts ADD COLUMN subtitle TEXT NULL;\n\
            CREATE INDEX CONCURRENTLY idx_posts_subtitle ON posts (subtitle);";
        let findings = classify_sql(sql);
        assert!(findings.is_empty());
    }

    #[test]
    fn mixed_safe_and_unsafe_statements_produces_findings_for_unsafe() {
        let sql = "\
            ALTER TABLE posts ADD COLUMN subtitle TEXT NULL;\n\
            ALTER TABLE posts DROP COLUMN body;";
        let findings = classify_sql(sql);
        assert_eq!(findings.len(), 1);
        assert_eq!(findings[0].risk, RiskLevel::Destructive);
    }

    #[test]
    fn multiple_unsafe_statements_produce_multiple_findings() {
        let sql = "\
            ALTER TABLE posts DROP COLUMN body;\n\
            CREATE INDEX idx_posts_title ON posts (title);";
        let findings = classify_sql(sql);
        assert_eq!(findings.len(), 2);
        assert!(findings.iter().any(|f| f.risk == RiskLevel::Destructive));
        assert!(
            findings
                .iter()
                .any(|f| f.risk == RiskLevel::PotentiallyBlocking)
        );
    }

    // ── line comments are ignored ─────────────────────────────────────────────

    #[test]
    fn sql_with_line_comments_is_classified_correctly() {
        let sql = "-- Removing old column\nALTER TABLE posts DROP COLUMN body;";
        let findings = classify_sql(sql);
        assert_eq!(findings.len(), 1);
        assert_eq!(findings[0].risk, RiskLevel::Destructive);
    }

    #[test]
    fn line_comment_with_semicolon_does_not_hide_following_statement() {
        // A `;` inside a `--` comment must not be treated as a statement separator.
        // Before the fix, `-- rollout complete; safe\nDROP TABLE posts` would be split
        // and the fragment `safe\nDROP TABLE posts` no longer starts with `drop table`.
        let sql = "-- rollout complete; safe to proceed\nDROP TABLE posts;";
        let findings = classify_sql(sql);
        assert_eq!(
            findings.len(),
            1,
            "DROP TABLE must be found after a line comment containing ';': {findings:?}"
        );
        assert_eq!(findings[0].risk, RiskLevel::Destructive);
    }

    #[test]
    fn autumn_safety_comment_does_not_double_classify() {
        // Autumn-generated SQL includes a leading safety comment; ensure the
        // comment text itself doesn't trigger a second finding.
        let sql = "-- autumn-safety: destructive\nALTER TABLE posts DROP COLUMN body;";
        let findings = classify_sql(sql);
        assert_eq!(findings.len(), 1);
    }

    #[test]
    fn autumn_safety_reviewed_suppresses_manual_review_finding() {
        // Operator acknowledges a CREATE FUNCTION is safe for their deploy.
        let sql = "-- autumn-safety: reviewed\nCREATE FUNCTION noop() RETURNS void LANGUAGE sql AS $$SELECT 1$$;";
        let findings = classify_sql(sql);
        assert!(
            findings.is_empty(),
            "reviewed marker must suppress ManualReview finding: {findings:?}"
        );
    }

    #[test]
    fn autumn_safety_reviewed_suppresses_unclassified_alter_table() {
        let sql = "-- autumn-safety: reviewed\nALTER TABLE users DROP CONSTRAINT users_email_key;";
        let findings = classify_sql(sql);
        assert!(
            findings.is_empty(),
            "reviewed marker must suppress finding: {findings:?}"
        );
    }

    #[test]
    fn autumn_safety_destructive_does_not_suppress() {
        // Only the `reviewed` marker suppresses; other autumn-safety values are informational.
        let sql = "-- autumn-safety: destructive\nALTER TABLE posts DROP COLUMN body;";
        let findings = classify_sql(sql);
        assert_eq!(
            findings.len(),
            1,
            "non-reviewed marker must not suppress findings"
        );
    }

    // ── helper predicates ─────────────────────────────────────────────────────

    #[test]
    fn is_safe_returns_true_for_empty() {
        assert!(is_safe(&[]));
    }

    #[test]
    fn is_safe_returns_false_for_unsafe_findings() {
        let f = SafetyFinding {
            operation: "DROP COLUMN".to_owned(),
            risk: RiskLevel::Destructive,
            why: "",
            next_action: "",
        };
        assert!(!is_safe(&[f]));
    }

    #[test]
    fn has_unsafe_findings_returns_false_for_empty() {
        assert!(!has_unsafe_findings(&[]));
    }

    #[test]
    fn has_unsafe_findings_returns_true_for_blocking() {
        let f = SafetyFinding {
            operation: "CREATE INDEX (non-concurrent)".to_owned(),
            risk: RiskLevel::PotentiallyBlocking,
            why: "",
            next_action: "",
        };
        assert!(has_unsafe_findings(&[f]));
    }

    // ── finding fields carry useful guidance ─────────────────────────────────

    #[test]
    fn drop_column_finding_names_the_risk_and_next_action() {
        let findings = classify_sql("ALTER TABLE posts DROP COLUMN body;");
        let f = &findings[0];
        assert!(
            !f.why.is_empty(),
            "why must explain the rolling-deploy risk"
        );
        assert!(
            !f.next_action.is_empty(),
            "next_action must tell the operator what to do"
        );
    }

    #[test]
    fn non_concurrent_index_finding_mentions_concurrently() {
        let findings = classify_sql("CREATE INDEX idx ON posts (title);");
        let f = &findings[0];
        assert!(
            f.next_action.to_lowercase().contains("concurrently"),
            "next_action should recommend CONCURRENTLY: {}",
            f.next_action
        );
    }

    // ── data backfill patterns ────────────────────────────────────────────────

    #[test]
    fn merge_into_is_data_backfill() {
        let sql = "MERGE INTO posts AS target \
                   USING staging AS src ON target.id = src.id \
                   WHEN MATCHED THEN UPDATE SET title = src.title \
                   WHEN NOT MATCHED THEN INSERT (id, title) VALUES (src.id, src.title);";
        let findings = classify_sql(sql);
        assert_eq!(
            findings.len(),
            1,
            "MERGE INTO must be classified as a data backfill: {findings:?}"
        );
        assert_eq!(findings[0].risk, RiskLevel::DataBackfill);
        assert_eq!(findings[0].operation, "Bulk DML (data backfill)");
    }

    #[test]
    fn bulk_update_is_data_backfill() {
        let findings = classify_sql("UPDATE posts SET status = 'draft' WHERE status IS NULL;");
        assert_eq!(findings.len(), 1);
        assert_eq!(findings[0].risk, RiskLevel::DataBackfill);
        assert_eq!(findings[0].operation, "Bulk DML (data backfill)");
    }

    #[test]
    fn insert_select_is_data_backfill() {
        let findings =
            classify_sql("INSERT INTO post_tags (post_id, tag) SELECT id, 'untagged' FROM posts;");
        assert_eq!(findings.len(), 1);
        assert_eq!(findings[0].risk, RiskLevel::DataBackfill);
    }

    #[test]
    fn bulk_delete_is_data_backfill() {
        let findings = classify_sql("DELETE FROM posts WHERE created_at < '2020-01-01';");
        assert_eq!(findings.len(), 1);
        assert_eq!(findings[0].risk, RiskLevel::DataBackfill);
        assert_eq!(findings[0].operation, "Bulk DML (data backfill)");
    }

    #[test]
    fn data_backfill_finding_recommends_separate_job() {
        let findings = classify_sql("UPDATE posts SET slug = LOWER(title);");
        let f = &findings[0];
        assert!(!f.why.is_empty());
        assert!(
            f.next_action.to_lowercase().contains("background job")
                || f.next_action.to_lowercase().contains("task"),
            "next_action should recommend a separate job or task: {}",
            f.next_action
        );
    }

    #[test]
    fn cte_update_is_data_backfill() {
        let sql = "WITH batch AS (SELECT id FROM posts WHERE status IS NULL LIMIT 1000) \
                   UPDATE posts SET status = 'draft' FROM batch WHERE posts.id = batch.id;";
        let findings = classify_sql(sql);
        assert_eq!(findings.len(), 1);
        assert_eq!(findings[0].risk, RiskLevel::DataBackfill);
        assert_eq!(findings[0].operation, "Bulk DML (data backfill via CTE)");
    }

    #[test]
    fn cte_delete_is_data_backfill() {
        let sql = "WITH doomed AS (SELECT id FROM posts WHERE archived = true) DELETE FROM posts USING doomed WHERE posts.id = doomed.id;";
        let findings = classify_sql(sql);
        assert_eq!(findings.len(), 1);
        assert_eq!(findings[0].risk, RiskLevel::DataBackfill);
    }

    #[test]
    fn cte_select_only_is_safe() {
        // A read-only CTE should not produce a DataBackfill finding.
        let sql = "WITH recent AS (SELECT id FROM posts ORDER BY created_at DESC LIMIT 10) \
                   SELECT * FROM recent;";
        let findings = classify_sql(sql);
        assert!(
            findings.iter().all(|f| f.risk != RiskLevel::DataBackfill),
            "read-only CTE must not produce DataBackfill finding: {findings:?}"
        );
    }

    #[test]
    fn cte_body_write_with_outer_select_is_data_backfill() {
        // data-modifying CTE where the outer statement is SELECT, not UPDATE/DELETE.
        let sql = "WITH touched AS \
                   (UPDATE posts SET migrated = true WHERE migrated IS NULL RETURNING id) \
                   SELECT count(*) FROM touched;";
        let findings = classify_sql(sql);
        assert!(
            findings.iter().any(|f| f.risk == RiskLevel::DataBackfill),
            "data-modifying CTE with outer SELECT must still be flagged: {findings:?}"
        );
    }

    // ── manual review patterns ────────────────────────────────────────────────

    #[test]
    fn create_function_requires_manual_review() {
        let sql = "CREATE FUNCTION update_modified() RETURNS trigger AS $$ BEGIN NEW.updated_at = now(); RETURN NEW; END; $$ LANGUAGE plpgsql;";
        let findings = classify_sql(sql);
        assert_eq!(findings.len(), 1);
        assert_eq!(findings[0].risk, RiskLevel::ManualReview);
        assert_eq!(findings[0].operation, "Unclassified DDL");
    }

    #[test]
    fn truncate_is_destructive() {
        let findings = classify_sql("TRUNCATE TABLE staging_data;");
        assert_eq!(findings.len(), 1);
        assert_eq!(findings[0].risk, RiskLevel::Destructive);
    }

    #[test]
    fn drop_constraint_requires_manual_review() {
        let findings = classify_sql("ALTER TABLE users DROP CONSTRAINT users_email_key;");
        assert_eq!(findings.len(), 1);
        assert_eq!(findings[0].risk, RiskLevel::ManualReview);
        assert_eq!(findings[0].operation, "Unclassified ALTER TABLE");
    }

    #[test]
    fn alter_column_set_not_null_requires_manual_review() {
        let findings = classify_sql("ALTER TABLE users ALTER COLUMN email SET NOT NULL;");
        assert_eq!(findings.len(), 1);
        assert_eq!(findings[0].risk, RiskLevel::ManualReview);
        assert_eq!(findings[0].operation, "Unclassified ALTER TABLE");
    }

    #[test]
    fn known_ddl_does_not_trigger_manual_review() {
        // CREATE TABLE is safe — must not get a ManualReview finding on top
        let findings = classify_sql("CREATE TABLE comments (id BIGSERIAL PRIMARY KEY);");
        assert!(
            findings.iter().all(|f| f.risk != RiskLevel::ManualReview),
            "known DDL should not also produce ManualReview: {findings:?}"
        );
    }

    #[test]
    fn add_column_does_not_trigger_unclassified_alter_table() {
        // A safe ADD COLUMN must not also generate a ManualReview finding.
        let findings = classify_sql("ALTER TABLE posts ADD COLUMN subtitle TEXT NULL;");
        assert!(
            findings.iter().all(|f| f.risk != RiskLevel::ManualReview),
            "safe ADD COLUMN should not produce ManualReview: {findings:?}"
        );
    }

    // ── contains_concurrent_index ─────────────────────────────────────────────

    #[test]
    fn contains_concurrent_index_true_for_executable_statement() {
        assert!(contains_concurrent_index(
            "CREATE INDEX CONCURRENTLY idx ON posts (title);"
        ));
        assert!(contains_concurrent_index(
            "CREATE UNIQUE INDEX CONCURRENTLY idx ON posts (slug);"
        ));
    }

    #[test]
    fn contains_concurrent_index_false_for_non_concurrent() {
        assert!(!contains_concurrent_index(
            "CREATE INDEX idx ON posts (title);"
        ));
    }

    #[test]
    fn contains_concurrent_index_false_for_comment_only_mention() {
        let sql = "-- TODO: use CREATE INDEX CONCURRENTLY later\n\
                   CREATE INDEX idx ON posts (title);";
        assert!(
            !contains_concurrent_index(sql),
            "a CONCURRENTLY reference only in a comment must return false"
        );
    }

    #[test]
    fn contains_concurrent_index_true_for_multiline_statement() {
        let sql = "CREATE INDEX\n  CONCURRENTLY idx_posts_title ON posts (title);";
        assert!(
            contains_concurrent_index(sql),
            "multi-line CONCURRENTLY statement must be detected"
        );
    }

    #[test]
    fn contains_concurrent_index_true_for_drop_index_concurrently() {
        assert!(
            contains_concurrent_index("DROP INDEX CONCURRENTLY idx_posts_title;"),
            "DROP INDEX CONCURRENTLY must be detected"
        );
    }

    // ── block comment stripping ───────────────────────────────────────────────

    #[test]
    fn block_comment_before_drop_table_is_still_classified() {
        let sql = "/* cleanup old table */ DROP TABLE posts;";
        let findings = classify_sql(sql);
        assert_eq!(
            findings.len(),
            1,
            "DROP TABLE must be found after block comment"
        );
        assert_eq!(findings[0].risk, RiskLevel::Destructive);
    }

    #[test]
    fn block_comment_before_create_index_is_still_classified() {
        let sql = "/* needs index */ CREATE INDEX idx ON posts (title);";
        let findings = classify_sql(sql);
        assert_eq!(
            findings.len(),
            1,
            "CREATE INDEX must be found after block comment"
        );
        assert_eq!(findings[0].risk, RiskLevel::PotentiallyBlocking);
    }

    #[test]
    fn multiline_block_comment_is_stripped() {
        let sql = "/*\n * Remove legacy column\n */\nALTER TABLE posts DROP COLUMN body;";
        let findings = classify_sql(sql);
        assert_eq!(
            findings.len(),
            1,
            "DROP COLUMN must be found after multi-line block comment"
        );
        assert_eq!(findings[0].risk, RiskLevel::Destructive);
    }

    #[test]
    fn block_comment_only_mention_of_keyword_is_not_classified() {
        // Only mentions DROP TABLE inside a block comment; actual statement is safe.
        let sql = "/* was: DROP TABLE posts; */ ALTER TABLE posts ADD COLUMN active BOOL;";
        let findings = classify_sql(sql);
        assert!(
            findings.iter().all(|f| f.risk != RiskLevel::Destructive),
            "Destructive keyword inside block comment must not produce a Destructive finding"
        );
    }

    #[test]
    fn block_comment_with_semicolon_does_not_hide_following_statement() {
        // The semicolon inside the block comment must not split the statement early.
        let sql = "/* cleanup; safe to drop */ DROP TABLE posts;";
        let findings = classify_sql(sql);
        assert_eq!(
            findings.len(),
            1,
            "DROP TABLE after a block comment containing ';' must still be classified"
        );
        assert_eq!(findings[0].risk, RiskLevel::Destructive);
    }

    #[test]
    fn block_open_inside_line_comment_does_not_hide_statements() {
        // Issue #3437: `/*` in a `--` comment has no meaning.
        for backend in [DatabaseBackend::Postgres, DatabaseBackend::Sqlite] {
            for sql in [
                "--/*\nDROP TABLE t;",
                "-- load /var/data/*.csv later\nDROP TABLE t;",
            ] {
                let findings = classify_sql_for(backend, sql);
                assert_eq!(findings.len(), 1, "{backend:?} {sql:?}");
                assert_eq!(findings[0].risk, RiskLevel::Destructive);
            }
        }
    }

    #[test]
    fn block_open_inside_literal_does_not_hide_statements() {
        let sql = "INSERT INTO t VALUES ('/*');\nDROP TABLE t;";
        let findings = classify_sql(sql);
        assert!(findings.iter().any(|f| f.risk == RiskLevel::Destructive));
    }

    #[test]
    fn line_comment_marker_inside_literal_does_not_hide_clause() {
        let sql = "ALTER TABLE t ADD COLUMN a text DEFAULT '--', DROP COLUMN b;";
        let findings = classify_sql(sql);
        assert!(
            findings.iter().any(|f| f.operation == "DROP COLUMN"),
            "{findings:?}"
        );
    }

    #[test]
    fn review_marker_survives_block_stripping() {
        let sql = "-- autumn-safety: reviewed /* ok\nDROP TABLE t;";
        assert!(classify_sql(sql).is_empty());
    }

    #[test]
    fn block_comment_after_line_comment_is_still_stripped() {
        let sql = "-- note\n/* DROP TABLE a; */ SELECT 1;";
        assert!(classify_sql(sql).is_empty());
    }

    #[test]
    fn has_executable_sql_sees_statement_after_line_comment_with_block_open() {
        assert!(has_executable_sql(
            DatabaseBackend::Postgres,
            "-- x /* y\nDROP TABLE t;"
        ));
        assert!(!has_executable_sql(
            DatabaseBackend::Postgres,
            "-- x /* y\n-- z"
        ));
    }

    #[test]
    fn concurrent_index_scan_strips_block_comments_first() {
        assert!(contains_concurrent_index(
            "CREATE INDEX /* a; b */ CONCURRENTLY idx ON t (c);"
        ));
        assert!(!contains_concurrent_index(
            "/* x; CREATE INDEX CONCURRENTLY i ON t (c) */"
        ));
    }

    #[test]
    fn apostrophe_in_dollar_quote_does_not_keep_later_comment() {
        let sql = "ALTER TABLE t ADD COLUMN a text DEFAULT $$it's$$ -- drop column b\n;";
        assert!(
            classify_sql(sql)
                .iter()
                .all(|f| f.operation != "DROP COLUMN")
        );
    }

    #[test]
    fn unclosed_spans_run_to_end_of_input() {
        let sql = "DROP TABLE a; SELECT 'oops --/* DROP TABLE b;";
        assert_eq!(classify_sql(sql).len(), 1);
        assert_eq!(
            strip_line_comments(DatabaseBackend::Postgres, "SELECT 'a -- b"),
            "SELECT 'a -- b"
        );
        assert_eq!(
            strip_block_comments(
                DatabaseBackend::Postgres,
                "DROP TABLE a; /* open\nDROP TABLE b;"
            ),
            "DROP TABLE a;  "
        );
        assert_eq!(strip_block_comments(DatabaseBackend::Postgres, "/*/"), " ");
        let dollar = "SELECT $$ /* x\nDROP TABLE t;";
        assert_eq!(
            strip_block_comments(DatabaseBackend::Postgres, dollar),
            dollar
        );
    }

    #[test]
    fn block_comments_nest_on_postgres_only() {
        let sql = "/* a /* b */ DROP TABLE t; */";
        assert!(classify_sql(sql).is_empty());
        assert_eq!(ops(DatabaseBackend::Sqlite, sql), ["DROP TABLE"]);
    }

    #[test]
    fn multibyte_text_does_not_break_scanning() {
        let sql = "-- é/*\nDROP TABLE tést; /* ü */ SELECT '日本--', $é$ ; $é$;";
        let findings = classify_sql(sql);
        assert_eq!(findings.len(), 1, "{findings:?}");
        assert_eq!(findings[0].risk, RiskLevel::Destructive);
    }

    #[test]
    fn lexer_helpers_measure_spans() {
        assert_eq!(doubled_quote_len("'a''b'x", '\''), 6);
        assert_eq!(doubled_quote_len("`a", '`'), 2);
        assert_eq!(escape_string_len(r"'a\'b'x"), 6);
        assert_eq!(escape_string_len(r"'a''b'x"), 6);
        assert_eq!(escape_string_len(r"'a\"), 3);
        let lite = DatabaseBackend::Sqlite;
        assert_eq!(quoted_len(lite, "[a b", None), Some(4));
        assert_eq!(quoted_len(lite, "$$x$$", None), None);
        // Postgres has no backtick or bracket quotes.
        let pg = DatabaseBackend::Postgres;
        assert_eq!(quoted_len(pg, "`a`", None), None);
        assert_eq!(quoted_len(pg, "[a]", None), None);
        assert_eq!(quoted_len(pg, "E'a'", Some('x')), None);
        assert_eq!(quoted_len(pg, "$b$x$b$", Some('a')), None);
        assert_eq!(blank_quoted_spans(pg, "a 'bc d"), "a '   d");
        assert_eq!(dollar_quote_len("$1 AND"), None);
        assert_eq!(dollar_quote_len("$a b$ x"), None);
        assert_eq!(dollar_quote_len("$$x$$y"), Some(5));
        assert_eq!(
            line_comment_len(DatabaseBackend::Postgres, "--a\nb"),
            Some(3)
        );
        assert_eq!(line_comment_len(DatabaseBackend::Postgres, "a--"), None);
    }

    #[test]
    fn quote_in_line_comment_opens_no_literal() {
        let findings = classify_sql("-- don't\nDROP TABLE t;");
        assert_eq!(findings.len(), 1);
        assert_eq!(
            strip_line_comments(DatabaseBackend::Postgres, "-- don't\nx"),
            "\nx"
        );
    }

    #[test]
    fn dollar_quote_hides_line_comment_marker() {
        let sql = "DO $$ BEGIN -- x\n END $$";
        assert_eq!(
            normalize_statement(DatabaseBackend::Postgres, sql),
            "do $$ BEGIN -- x\n END $$"
        );
    }

    #[test]
    fn has_executable_sql_follows_the_dialect() {
        // A `SQLite` comment ends at the first `*/`. A Postgres comment nests.
        let sql = "/* a /* */ DROP TABLE t;";
        assert!(has_executable_sql(DatabaseBackend::Sqlite, sql));
        assert!(!has_executable_sql(DatabaseBackend::Postgres, sql));
    }

    #[test]
    fn quoted_table_names_follow_the_dialect_case_rule() {
        // `SQLite` names ignore case, also in quotes. Postgres quoted names do not.
        let lite = DatabaseBackend::Sqlite;
        let sql =
            r#"CREATE TABLE "Users" (id int); ALTER TABLE "users" ADD COLUMN z int NOT NULL;"#;
        assert!(is_safe(&classify_sql_for(lite, sql)), "{sql}");
        let sql = r#"CREATE TABLE "Users" (id int); CREATE INDEX i ON "users" (id);"#;
        assert_eq!(ops(lite, sql), Vec::<String>::new());
        assert_eq!(
            ops(DatabaseBackend::Postgres, sql),
            ["CREATE INDEX (non-concurrent)"]
        );
    }

    #[test]
    fn has_executable_sql_edge_inputs() {
        assert!(has_executable_sql(
            DatabaseBackend::Postgres,
            "SELECT '--';"
        ));
        assert!(!has_executable_sql(DatabaseBackend::Postgres, ""));
        assert!(!has_executable_sql(DatabaseBackend::Postgres, "/* only */"));
    }

    #[test]
    fn identifier_quoting_forms_do_not_hide_later_statements() {
        for (backend, sql) in [
            (
                DatabaseBackend::Postgres,
                "SELECT ARRAY[$$]$$]; DROP TABLE t;",
            ),
            (
                DatabaseBackend::Postgres,
                "SELECT ARRAY['a]']; DROP TABLE t;",
            ),
            (
                DatabaseBackend::Sqlite,
                "CREATE TABLE `a\"b` (id int); DROP TABLE t;",
            ),
            (
                DatabaseBackend::Sqlite,
                "CREATE TABLE [a\"b] (id int); DROP TABLE t;",
            ),
        ] {
            let f = classify_sql_for(backend, sql);
            assert!(
                f.iter().any(|f| f.risk == RiskLevel::Destructive),
                "{sql}: {f:?}"
            );
        }
    }

    /// Operations found in `sql` for `backend`.
    fn ops(backend: DatabaseBackend, sql: &str) -> Vec<String> {
        classify_sql_for(backend, sql)
            .into_iter()
            .map(|f| f.operation)
            .collect()
    }

    #[test]
    fn markers_in_quoted_identifiers_hide_no_statement() {
        // Issue #3518. Each identifier holds a `/*`, `--`, `'` or `;`.
        let pg = DatabaseBackend::Postgres;
        let lite = DatabaseBackend::Sqlite;
        let mut cases = Vec::new();
        for backend in [pg, lite] {
            for ident in [
                r#""a/*b""#,
                r#""a--b""#,
                r#""a'b""#,
                r#""a;drop table y""#,
                r#""a""/*b""#,
            ] {
                cases.push((backend, ident));
            }
        }
        for ident in [
            "`a/*b`",
            "`a--b`",
            "`a'b`",
            "`a;drop table y`",
            "`a``/*b`",
            "[a/*b]",
            "[a--b]",
            "[a'b]",
            "[a;drop table y]",
        ] {
            cases.push((lite, ident));
        }
        for (backend, ident) in cases {
            let sql = format!("CREATE TABLE {ident} (id int); DROP TABLE x;");
            assert_eq!(ops(backend, &sql), ["DROP TABLE"], "{backend:?} {sql}");
            let sql = format!("CREATE TABLE {ident} (id int);\nDROP TABLE x;");
            assert_eq!(ops(backend, &sql), ["DROP TABLE"], "{backend:?} {sql}");
        }
    }

    #[test]
    fn escape_string_hides_no_statement() {
        // Issue #3518. In a Postgres `E'..'` string, `\'` is a quote. Outside
        // `E'..'`, `\` is text.
        for lit in [
            r"E'\''",
            r"e'\''",
            r"E'a\'b'",
            r"E'\\'",
            r"E'\'--/*;'",
            r"E'it''s; x'",
            r"E'\é'",
            r#""x"E'\''"#,
            r"U&'\0041'",
            r"'C:\'",
            r"xe'\'",
        ] {
            let sql = format!("SELECT {lit}; DROP TABLE x;");
            assert_eq!(
                ops(DatabaseBackend::Postgres, &sql),
                ["DROP TABLE"],
                "{sql}"
            );
        }
    }

    #[test]
    fn dollar_quote_rules_follow_the_dialect() {
        // Issue #3518. `a$b$` is one identifier. `SQLite` has no dollar quotes.
        let sql = "CREATE TABLE t (a$b$ int); DROP TABLE x;";
        for backend in [DatabaseBackend::Postgres, DatabaseBackend::Sqlite] {
            assert_eq!(ops(backend, sql), ["DROP TABLE"], "{backend:?}");
        }
        let sql = "SELECT $y$z; DROP TABLE x;";
        assert_eq!(ops(DatabaseBackend::Sqlite, sql), ["DROP TABLE"]);
        // Postgres still keeps a function body whole.
        let sql = "CREATE FUNCTION f() RETURNS int AS $b$ SELECT 1; $b$ LANGUAGE sql;";
        assert_eq!(ops(DatabaseBackend::Postgres, sql), ["Unclassified DDL"]);
        // `SQLite` has no `E'..'` strings, so `\` is text there.
        let sql = r"SELECT E'\'; DROP TABLE x;";
        assert_eq!(ops(DatabaseBackend::Sqlite, sql), ["DROP TABLE"]);
    }

    #[test]
    fn quote_in_identifier_hides_no_clause() {
        // Issue #3518. A `'` in an identifier opens no literal.
        let pg = DatabaseBackend::Postgres;
        let lite = DatabaseBackend::Sqlite;
        let sql = r#"ALTER TABLE t ADD COLUMN "it's" text UNIQUE;"#;
        assert_eq!(ops(pg, sql), ["ADD COLUMN UNIQUE (inline constraint)"]);
        let sql = "ALTER TABLE t ADD COLUMN [it's] text UNIQUE;";
        assert_eq!(
            ops(lite, sql),
            ["ADD COLUMN UNIQUE (unsupported on SQLite)"]
        );
        let sql = r#"CREATE TABLE "it's" (id int) PARTITION BY RANGE (id);"#;
        assert_eq!(
            ops(lite, sql),
            ["CREATE TABLE with a Postgres-only clause (unsupported on SQLite)"]
        );
        // A `\'` in an escape string and a word in a dollar quote are text.
        for sql in [
            r"ALTER TABLE t ADD COLUMN a text DEFAULT E'\' unique ';",
            "ALTER TABLE t ADD COLUMN a text DEFAULT $$ not unique $$;",
        ] {
            assert_eq!(ops(pg, sql), Vec::<String>::new(), "{sql}");
        }
    }

    #[test]
    fn comma_in_quoted_span_splits_no_alter_subcommand() {
        // Issue #3518. A `,` in a quoted span is text on both backends.
        for (backend, sql) in [
            (
                DatabaseBackend::Postgres,
                r#"ALTER TABLE t ADD COLUMN "a,b" text DEFAULT 'x, y', DROP COLUMN c;"#,
            ),
            (
                DatabaseBackend::Postgres,
                r"ALTER TABLE t ADD COLUMN a text DEFAULT E'\', y', DROP COLUMN c;",
            ),
            (
                DatabaseBackend::Postgres,
                "ALTER TABLE t ADD COLUMN a text DEFAULT $$x, y$$, DROP COLUMN c;",
            ),
            (DatabaseBackend::Sqlite, "ALTER TABLE t DROP COLUMN [a,b];"),
        ] {
            assert_eq!(ops(backend, sql), ["DROP COLUMN"], "{backend:?} {sql}");
        }
    }

    #[test]
    fn quoted_table_name_merges_no_alter_subcommand() {
        // Issue #3518. A space in a quoted table name does not end the name.
        let pg = DatabaseBackend::Postgres;
        for table in [r#""my table""#, r#"s."my table""#] {
            for head in ["", "IF EXISTS ", "ONLY ", "IF EXISTS ONLY "] {
                let sql = format!(
                    "ALTER TABLE {head}{table} ADD COLUMN a int, ADD COLUMN b int NOT NULL;"
                );
                assert_eq!(ops(pg, &sql), ["ADD COLUMN NOT NULL (no default)"], "{sql}");
            }
        }
        let sql = r#"ALTER TABLE "my table" ADD COLUMN b int UNIQUE;"#;
        assert_eq!(ops(pg, sql), ["ADD COLUMN UNIQUE (inline constraint)"]);
        let sql = "ALTER TABLE [my table] ADD COLUMN b int UNIQUE;";
        assert_eq!(
            ops(DatabaseBackend::Sqlite, sql),
            ["ADD COLUMN UNIQUE (unsupported on SQLite)"]
        );
    }

    #[test]
    fn paren_in_quoted_span_merges_no_alter_subcommand() {
        // Issue #3518. A `(` in a literal does not change the depth.
        let sql = "ALTER TABLE t ADD COLUMN a text DEFAULT '(', ALTER COLUMN b SET NOT NULL;";
        assert_eq!(
            ops(DatabaseBackend::Postgres, sql),
            ["Unclassified ALTER TABLE"]
        );
    }

    #[test]
    fn postgres_lexer_rules_hide_no_statement() {
        // Issue #3518. Each case follows a Postgres lexer rule.
        for sql in [
            // Dollar tags are case-sensitive.
            "ALTER TABLE t ADD COLUMN c text DEFAULT $A$ $a$ ' $A$, ADD COLUMN z int NOT NULL;",
            // Block comments nest.
            "/*\nCREATE TABLE old (id int); /* legacy */\n*/\nALTER TABLE t ADD COLUMN z int NOT NULL;",
            "/* /* */ ' */ ALTER TABLE t ADD COLUMN z int NOT NULL; -- '",
            // A tag can hold any non-ASCII character.
            "SELECT $€$/*$€$; ALTER TABLE t ADD COLUMN z int NOT NULL;",
            // A bare CR ends a line comment.
            "-- note\rALTER TABLE t ADD COLUMN z int NOT NULL;",
        ] {
            assert_eq!(
                ops(DatabaseBackend::Postgres, sql),
                ["ADD COLUMN NOT NULL (no default)"],
                "{sql:?}"
            );
        }
        // A tag cannot start with a digit: `$1` is a parameter.
        assert_eq!(dollar_quote_len("$1$x$1$"), None);
        assert_eq!(dollar_quote_len("$t1$x$t1$"), Some(9));
    }

    #[test]
    fn sqlite_lexer_rules_hide_no_statement() {
        // Issue #3518. A `SQLite` variable such as `$a(..)` runs to `)` or a space.
        for sql in [
            "SELECT $a([); DROP TABLE x; -- ]",
            "SELECT :a(\"); DROP TABLE x; -- \"",
            "SELECT @a(`); DROP TABLE x; -- `",
            "SELECT $a('); DROP TABLE x; SELECT '';",
        ] {
            assert_eq!(ops(DatabaseBackend::Sqlite, sql), ["DROP TABLE"], "{sql}");
        }
        // Block comments do not nest, and only `\n` ends a line comment.
        let sql = "/* a /* b */ DROP TABLE x;";
        assert_eq!(ops(DatabaseBackend::Sqlite, sql), ["DROP TABLE"]);
        let sql = "-- a\rDROP TABLE x;\nSELECT 1;";
        assert_eq!(ops(DatabaseBackend::Sqlite, sql), Vec::<String>::new());
        assert_eq!(sqlite_variable_len("$a(x y"), Some(4));
        assert_eq!(sqlite_variable_len(":a(x)y"), Some(5));
        assert_eq!(sqlite_variable_len("$("), None);
        // In `a$b(`, `$b` is part of the name `a$b`.
        let texts: Vec<_> = tokens(DatabaseBackend::Sqlite, "a$b(")
            .map(|(_, text)| text)
            .collect();
        assert_eq!(texts, ["a", "$", "b", "("]);
    }

    mod quoted_identifier_proptest {
        use super::*;
        use proptest::prelude::*;

        proptest! {
            /// Issue #3518. No identifier text hides a later statement.
            #[test]
            fn quoted_identifier_hides_no_statement(
                body in "[a-z;'\"`\\[\\]$/*\\- \n]{0,12}",
                form in 0..4usize,
            ) {
                let (backend, ident) = match form {
                    0 => (DatabaseBackend::Postgres, format!("\"{}\"", body.replace('"', "\"\""))),
                    1 => (DatabaseBackend::Sqlite, format!("\"{}\"", body.replace('"', "\"\""))),
                    2 => (DatabaseBackend::Sqlite, format!("`{}`", body.replace('`', "``"))),
                    _ => (DatabaseBackend::Sqlite, format!("[{}]", body.replace(']', ""))),
                };
                let sql = format!("CREATE TABLE {ident} (id int); DROP TABLE x;");
                prop_assert_eq!(ops(backend, &sql), ["DROP TABLE"], "{}", sql);
            }

            /// The token texts join back to the input.
            #[test]
            fn tokens_join_back_to_the_input(sql in any::<String>(), pg in any::<bool>()) {
                let backend = if pg { DatabaseBackend::Postgres } else { DatabaseBackend::Sqlite };
                let joined: String = tokens(backend, &sql).map(|(_, text)| text).collect();
                prop_assert_eq!(joined, sql);
            }
        }
    }

    #[test]
    fn block_comment_between_keywords_preserves_token_boundary() {
        // `DROP/* note */TABLE posts` must not concatenate to `DROPTABLE posts`,
        // which would miss both the `drop table` rule and the `drop ` catch-all.
        let findings = classify_sql("DROP/* note */TABLE posts;");
        assert_eq!(
            findings.len(),
            1,
            "block comment between keywords must not merge them: {findings:?}"
        );
        assert_eq!(findings[0].risk, RiskLevel::Destructive);
    }

    // ── dollar-quoted function bodies ─────────────────────────────────────────

    #[test]
    fn dollar_quoted_function_with_semicolons_is_one_statement() {
        // The semicolons inside $$ ... $$ must not produce extra statement fragments.
        let sql = "CREATE FUNCTION bump() RETURNS void AS $$ BEGIN \
                   UPDATE posts SET hits = hits + 1; RETURN; END; $$ LANGUAGE plpgsql;";
        let findings = classify_sql(sql);
        assert_eq!(
            findings.len(),
            1,
            "dollar-quoted body with semicolons must produce exactly one finding: {findings:?}"
        );
        assert_eq!(findings[0].risk, RiskLevel::ManualReview);
    }

    #[test]
    fn autumn_safety_reviewed_suppresses_function_with_dml_in_body() {
        // Without dollar-quote-aware splitting the DML fragment would escape suppression.
        let sql = "-- autumn-safety: reviewed\n\
                   CREATE FUNCTION migrate_posts() RETURNS void AS $$\n\
                   BEGIN\n  UPDATE posts SET migrated = true;\n  RETURN;\nEND;\n\
                   $$ LANGUAGE plpgsql;";
        let findings = classify_sql(sql);
        assert!(
            findings.is_empty(),
            "reviewed marker must suppress a dollar-quoted function containing DML: {findings:?}"
        );
    }

    #[test]
    fn tagged_dollar_quote_with_semicolons_is_one_statement() {
        let sql = "CREATE FUNCTION foo() RETURNS void AS $func$ \
                   BEGIN UPDATE posts SET x = 1; END; $func$ LANGUAGE plpgsql;";
        let findings = classify_sql(sql);
        assert_eq!(
            findings.len(),
            1,
            "tagged dollar-quote body with semicolons must not split: {findings:?}"
        );
    }

    #[test]
    fn split_statements_keeps_semicolon_in_string_literal_intact() {
        // Regression test (PR review, issue #1023): a `DEFAULT 'hello; world'`
        // clause -- exactly what `sql_default_literal` produces for a
        // `--default` value containing a semicolon -- must not be split at
        // the semicolon inside the quotes.
        let sql = "CREATE TABLE posts (\n    \
             id BIGSERIAL PRIMARY KEY,\n    \
             title TEXT NOT NULL DEFAULT 'hello; world',\n    \
             created_at TIMESTAMP NOT NULL DEFAULT NOW()\n\
             );";
        let statements = split_statements(DatabaseBackend::Postgres, sql);
        assert_eq!(
            statements.len(),
            1,
            "a semicolon inside a single-quoted literal must not split the statement: {statements:?}"
        );
        assert!(statements[0].contains("'hello; world'"));
    }

    #[test]
    fn split_statements_handles_doubled_escaped_quote_in_literal() {
        // `''` is the standard SQL-escaped single quote; a semicolon after one
        // must still be treated as inside the (still-open) literal.
        let sql = "INSERT INTO posts (title) VALUES ('it''s; still one literal');";
        let statements = split_statements(DatabaseBackend::Postgres, sql);
        assert_eq!(statements.len(), 1, "got: {statements:?}");
    }

    #[test]
    fn split_statements_multiple_string_literals_with_semicolons() {
        let sql = "CREATE TABLE posts (a TEXT DEFAULT 'x;y', b TEXT DEFAULT 'p;q');";
        let statements = split_statements(DatabaseBackend::Postgres, sql);
        assert_eq!(statements.len(), 1, "got: {statements:?}");
    }

    // ── DROP INDEX ────────────────────────────────────────────────────────────

    #[test]
    fn drop_index_non_concurrent_is_potentially_blocking() {
        let findings = classify_sql("DROP INDEX idx_posts_title;");
        assert_eq!(findings.len(), 1);
        assert_eq!(findings[0].risk, RiskLevel::PotentiallyBlocking);
        assert_eq!(findings[0].operation, "DROP INDEX (non-concurrent)");
    }

    #[test]
    fn drop_index_with_concurrently_in_name_is_still_blocking() {
        // "concurrently" appears in the index name, not as the SQL token.
        let findings = classify_sql("DROP INDEX idx_concurrently;");
        assert_eq!(
            findings.len(),
            1,
            "index named idx_concurrently must still be flagged as non-concurrent: {findings:?}"
        );
        assert_eq!(findings[0].operation, "DROP INDEX (non-concurrent)");
    }

    #[test]
    fn drop_index_concurrently_is_safe_from_classifier() {
        // CONCURRENTLY avoids the table lock; the opt-out check in migrate.rs handles
        // the metadata.toml requirement separately.
        let findings = classify_sql("DROP INDEX CONCURRENTLY idx_posts_title;");
        assert!(
            findings
                .iter()
                .all(|f| f.risk != RiskLevel::PotentiallyBlocking
                    || f.operation.contains("CONCURRENTLY")),
            "DROP INDEX CONCURRENTLY must not produce a non-concurrent finding"
        );
    }

    // ── DROP TYPE ─────────────────────────────────────────────────────────────

    #[test]
    fn drop_type_requires_manual_review() {
        let findings = classify_sql("DROP TYPE status;");
        assert_eq!(findings.len(), 1);
        assert_eq!(findings[0].risk, RiskLevel::ManualReview);
    }

    #[test]
    fn drop_type_cascade_requires_manual_review() {
        let findings = classify_sql("DROP TYPE status CASCADE;");
        assert_eq!(findings.len(), 1);
        assert_eq!(findings[0].risk, RiskLevel::ManualReview);
    }

    // ── has_executable_sql ────────────────────────────────────────────────────

    #[test]
    fn has_executable_sql_empty_string_is_false() {
        assert!(!has_executable_sql(DatabaseBackend::Postgres, ""));
    }

    #[test]
    fn has_executable_sql_whitespace_only_is_false() {
        assert!(!has_executable_sql(
            DatabaseBackend::Postgres,
            "   \n\t\n  "
        ));
    }

    #[test]
    fn has_executable_sql_line_comment_only_is_false() {
        assert!(!has_executable_sql(
            DatabaseBackend::Postgres,
            "-- nothing here\n-- just comments"
        ));
    }

    #[test]
    fn has_executable_sql_block_comment_only_is_false() {
        assert!(!has_executable_sql(
            DatabaseBackend::Postgres,
            "/* block comment only */"
        ));
    }

    #[test]
    fn has_executable_sql_real_sql_is_true() {
        assert!(has_executable_sql(
            DatabaseBackend::Postgres,
            "DROP TABLE posts;"
        ));
    }

    #[test]
    fn has_executable_sql_comment_plus_sql_is_true() {
        assert!(has_executable_sql(
            DatabaseBackend::Postgres,
            "-- undo the migration\nDROP TABLE posts;"
        ));
    }

    #[test]
    fn has_executable_sql_no_trailing_semicolon_is_true() {
        assert!(has_executable_sql(
            DatabaseBackend::Postgres,
            "DROP TABLE posts"
        ));
    }

    // ── SQLite dialect (#1906) ────────────────────────────────────────────────

    /// One finding for `operation`, or `None`.
    fn sqlite_finding(sql: &str, operation: &str) -> Option<SafetyFinding> {
        classify_sql_for(DatabaseBackend::Sqlite, sql)
            .into_iter()
            .find(|f| f.operation == operation)
    }

    #[test]
    fn sqlite_drop_index_is_safe() {
        // SQLite has no `CONCURRENTLY`, and DROP INDEX is a cheap catalog edit.
        // The generated SQLite `DROP COLUMN` path must emit DROP INDEX first, so
        // flagging it would make every such migration fail the deploy gate.
        assert!(
            classify_sql_for(DatabaseBackend::Sqlite, "DROP INDEX idx_posts_title;").is_empty(),
            "SQLite DROP INDEX must not be flagged"
        );
        // Postgres keeps its existing finding.
        assert!(!classify_sql("DROP INDEX idx_posts_title;").is_empty());
    }

    #[test]
    fn sqlite_create_index_advice_never_names_concurrently() {
        let f = sqlite_finding(
            "CREATE INDEX idx_posts_title ON posts (title);",
            "CREATE INDEX",
        )
        .expect("SQLite still reports a blocking index build");
        assert_eq!(f.risk, RiskLevel::PotentiallyBlocking);
        assert!(
            !f.next_action.to_lowercase().contains("concurrently"),
            "SQLite has no CREATE INDEX CONCURRENTLY: {}",
            f.next_action
        );
        assert!(
            !f.why.to_lowercase().contains("postgres"),
            "why must not cite Postgres on SQLite: {}",
            f.why
        );
    }

    #[test]
    fn sqlite_create_index_concurrently_is_unsupported() {
        let f = sqlite_finding(
            "CREATE INDEX CONCURRENTLY idx_posts_title ON posts (title);",
            "CREATE INDEX CONCURRENTLY (unsupported on SQLite)",
        )
        .expect("CONCURRENTLY is not SQLite syntax");
        assert_eq!(f.risk, RiskLevel::Unsupported);
    }

    #[test]
    fn sqlite_add_column_not_null_without_default_is_unsupported() {
        // SQLite rejects the statement outright — it is not merely blocking.
        let f = sqlite_finding(
            "ALTER TABLE posts ADD COLUMN title TEXT NOT NULL;",
            "ADD COLUMN NOT NULL (no default)",
        )
        .expect("finding expected");
        assert_eq!(f.risk, RiskLevel::Unsupported);
        // Postgres keeps the potentially-blocking classification.
        assert_eq!(
            classify_sql("ALTER TABLE posts ADD COLUMN title TEXT NOT NULL;")[0].risk,
            RiskLevel::PotentiallyBlocking
        );
    }

    #[test]
    fn sqlite_add_column_not_null_with_constant_default_is_safe() {
        assert!(
            classify_sql_for(
                DatabaseBackend::Sqlite,
                "ALTER TABLE posts ADD COLUMN views INTEGER NOT NULL DEFAULT 0;"
            )
            .is_empty(),
            "a constant default is the supported SQLite form"
        );
    }

    #[test]
    fn sqlite_add_column_with_expression_default_is_unsupported() {
        // SQLite forbids CURRENT_TIMESTAMP and parenthesized expressions as an
        // ADD COLUMN default.
        for sql in [
            "ALTER TABLE posts ADD COLUMN seen_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP;",
            "ALTER TABLE posts ADD COLUMN slug TEXT NOT NULL DEFAULT (lower(title));",
        ] {
            let f = sqlite_finding(sql, "ADD COLUMN (non-constant default)")
                .unwrap_or_else(|| panic!("finding expected for {sql}"));
            assert_eq!(f.risk, RiskLevel::Unsupported, "{sql}");
        }
    }

    #[test]
    fn sqlite_alter_column_is_unsupported() {
        for sql in [
            "ALTER TABLE posts ALTER COLUMN title TYPE TEXT;",
            "ALTER TABLE posts ALTER COLUMN title SET NOT NULL;",
        ] {
            let f = sqlite_finding(sql, "ALTER COLUMN (unsupported on SQLite)")
                .unwrap_or_else(|| panic!("finding expected for {sql}"));
            assert_eq!(f.risk, RiskLevel::Unsupported, "{sql}");
            assert!(
                f.next_action.contains("rebuild"),
                "must point at the table-rebuild procedure: {}",
                f.next_action
            );
        }
    }

    #[test]
    fn sqlite_add_column_inline_unique_is_unsupported() {
        let f = sqlite_finding(
            "ALTER TABLE posts ADD COLUMN slug TEXT UNIQUE;",
            "ADD COLUMN UNIQUE (unsupported on SQLite)",
        )
        .expect("SQLite rejects ADD COLUMN with UNIQUE");
        assert_eq!(f.risk, RiskLevel::Unsupported);
    }

    #[test]
    fn sqlite_add_column_nullable_references_is_safe() {
        // SQLite does not scan existing rows for a new FK, so the Postgres
        // "validates every row" finding is wrong here.
        assert!(
            classify_sql_for(
                DatabaseBackend::Sqlite,
                "ALTER TABLE comments ADD COLUMN post_id INTEGER REFERENCES posts(id);"
            )
            .is_empty(),
            "a nullable FK column is the supported SQLite form"
        );
    }

    #[test]
    fn sqlite_truncate_is_unsupported() {
        let f = sqlite_finding("TRUNCATE TABLE posts;", "TRUNCATE (unsupported on SQLite)")
            .expect("SQLite has no TRUNCATE statement");
        assert_eq!(f.risk, RiskLevel::Unsupported);
        assert!(f.next_action.contains("DELETE FROM"), "{}", f.next_action);
    }

    #[test]
    fn sqlite_postgres_only_objects_are_unsupported() {
        for sql in [
            "CREATE SEQUENCE post_ids;",
            "ALTER SEQUENCE post_ids RESTART WITH 1;",
            "DROP SEQUENCE post_ids;",
            "CREATE TYPE mood AS ENUM ('ok');",
            "ALTER TYPE mood RENAME VALUE 'ok' TO 'fine';",
            "DROP TYPE mood;",
            "CREATE EXTENSION IF NOT EXISTS pgcrypto;",
            "COMMENT ON TABLE posts IS 'hi';",
            "CREATE MATERIALIZED VIEW mv AS SELECT 1;",
        ] {
            let findings = classify_sql_for(DatabaseBackend::Sqlite, sql);
            assert!(
                findings.iter().any(|f| f.risk == RiskLevel::Unsupported),
                "expected an Unsupported finding for {sql}, got {findings:?}"
            );
        }
    }

    #[test]
    fn sqlite_drop_column_stays_destructive_with_sqlite_guidance() {
        let f = sqlite_finding("ALTER TABLE posts DROP COLUMN title;", "DROP COLUMN")
            .expect("finding expected");
        assert_eq!(f.risk, RiskLevel::Destructive);
        assert!(
            f.next_action.contains("DROP INDEX"),
            "SQLite refuses to drop an indexed column: {}",
            f.next_action
        );
    }

    #[test]
    fn sqlite_unsupported_blocks_the_deploy_gate() {
        let findings = classify_sql_for(
            DatabaseBackend::Sqlite,
            "ALTER TABLE posts ALTER COLUMN a TYPE TEXT;",
        );
        assert!(has_unsafe_findings(&findings));
        assert!(!is_safe(&findings));
    }

    #[test]
    fn unsupported_is_the_highest_risk_level() {
        assert!(RiskLevel::ManualReview < RiskLevel::Unsupported);
    }

    #[test]
    fn classify_sql_still_classifies_as_postgres() {
        // The bare entry point must stay byte-identical to the Postgres path.
        let sql = "ALTER TABLE posts ADD COLUMN title TEXT NOT NULL;\nDROP INDEX idx_posts_a;\n\
                   CREATE INDEX idx_posts_b ON posts (b);";
        let bare = classify_sql(sql);
        let explicit = classify_sql_for(DatabaseBackend::Postgres, sql);
        assert_eq!(bare.len(), explicit.len());
        for (a, b) in bare.iter().zip(explicit.iter()) {
            assert_eq!(a.operation, b.operation);
            assert_eq!(a.risk, b.risk);
            assert_eq!(a.why, b.why);
            assert_eq!(a.next_action, b.next_action);
        }
    }

    // ── review regressions: SQLite dialect (#1906) ────────────────────────

    /// Every operation reported for `sql` on `SQLite`.
    fn sqlite_ops(sql: &str) -> Vec<String> {
        classify_sql_for(DatabaseBackend::Sqlite, sql)
            .into_iter()
            .map(|f| f.operation)
            .collect()
    }

    #[test]
    fn sqlite_add_column_not_null_default_null_is_unsupported() {
        // SQLite drops a literal DEFAULT NULL before the NOT NULL check, so it
        // is not a default there.
        let f = sqlite_finding(
            "ALTER TABLE posts ADD COLUMN title TEXT NOT NULL DEFAULT NULL;",
            "ADD COLUMN NOT NULL (no default)",
        )
        .expect("finding expected");
        assert_eq!(f.risk, RiskLevel::Unsupported);
    }

    #[test]
    fn sqlite_add_column_bare_function_default_is_unsupported() {
        // Only a literal or signed number may follow a bare DEFAULT.
        for sql in [
            "ALTER TABLE posts ADD COLUMN a TEXT NOT NULL DEFAULT now();",
            "ALTER TABLE posts ADD COLUMN b TEXT NOT NULL DEFAULT gen_random_uuid();",
        ] {
            let f = sqlite_finding(sql, "ADD COLUMN (non-constant default)")
                .unwrap_or_else(|| panic!("finding expected for {sql}"));
            assert_eq!(f.risk, RiskLevel::Unsupported, "{sql}");
        }
    }

    #[test]
    fn sqlite_add_column_parenthesized_literal_default_is_accepted() {
        // SQLite's parser keeps no node for parentheses, so `DEFAULT ('draft')`
        // is the same constant as `DEFAULT 'draft'`.
        for sql in [
            "ALTER TABLE posts ADD COLUMN a TEXT NOT NULL DEFAULT ('draft');",
            "ALTER TABLE posts ADD COLUMN b INTEGER NOT NULL DEFAULT (0);",
            "ALTER TABLE posts ADD COLUMN c INTEGER NOT NULL DEFAULT -1;",
            "ALTER TABLE posts ADD COLUMN d TEXT NOT NULL DEFAULT 'a b';",
        ] {
            assert!(
                classify_sql_for(DatabaseBackend::Sqlite, sql).is_empty(),
                "{sql} must be accepted, got {:?}",
                sqlite_ops(sql)
            );
        }
    }

    #[test]
    fn sqlite_reads_a_default_with_no_space_before_the_value() {
        // `DEFAULT(0)` is valid SQL. Missing it made the NOT NULL rule fire and
        // block a statement SQLite applies cleanly.
        assert!(
            classify_sql_for(
                DatabaseBackend::Sqlite,
                "ALTER TABLE posts ADD COLUMN views INTEGER NOT NULL DEFAULT(0);"
            )
            .is_empty(),
            "got {:?}",
            sqlite_ops("ALTER TABLE posts ADD COLUMN views INTEGER NOT NULL DEFAULT(0);")
        );
        // A column whose name merely starts with `default` is not a keyword.
        let f = sqlite_finding(
            "ALTER TABLE posts ADD COLUMN defaulted TEXT NOT NULL;",
            "ADD COLUMN NOT NULL (no default)",
        )
        .expect("finding expected");
        assert_eq!(f.risk, RiskLevel::Unsupported);
    }

    #[test]
    fn sqlite_numeric_defaults_including_exponent_form_are_accepted() {
        for sql in [
            "ALTER TABLE s ADD COLUMN a REAL DEFAULT 1e-3;",
            "ALTER TABLE s ADD COLUMN b REAL DEFAULT -1.5E+10;",
            "ALTER TABLE s ADD COLUMN c REAL DEFAULT .5;",
            "ALTER TABLE s ADD COLUMN d INTEGER DEFAULT 0x1f;",
            "ALTER TABLE s ADD COLUMN e REAL DEFAULT (1e-3);",
        ] {
            assert!(
                classify_sql_for(DatabaseBackend::Sqlite, sql).is_empty(),
                "{sql} is a numeric literal, got {:?}",
                sqlite_ops(sql)
            );
        }
        // Still an expression, not a literal.
        let expr = "ALTER TABLE s ADD COLUMN f REAL DEFAULT 1e-3+1;";
        assert!(
            classify_sql_for(DatabaseBackend::Sqlite, expr)
                .iter()
                .any(|f| f.risk == RiskLevel::Unsupported),
            "{expr} -> {:?}",
            sqlite_ops(expr)
        );
    }

    #[test]
    fn sqlite_add_column_parenthesized_expression_default_is_unsupported() {
        for sql in [
            "ALTER TABLE posts ADD COLUMN a TEXT NOT NULL DEFAULT (1+2);",
            "ALTER TABLE posts ADD COLUMN b TEXT NOT NULL DEFAULT (abs(-1));",
            "ALTER TABLE posts ADD COLUMN c TEXT NOT NULL DEFAULT (datetime('now'));",
        ] {
            let f = sqlite_finding(sql, "ADD COLUMN (non-constant default)")
                .unwrap_or_else(|| panic!("finding expected for {sql}"));
            assert_eq!(f.risk, RiskLevel::Unsupported, "{sql}");
        }
    }

    #[test]
    fn sqlite_add_column_references_applies_without_a_foreign_keys_pragma() {
        // Autumn's migration connection leaves `PRAGMA foreign_keys` OFF, so
        // SQLite accepts a non-NULL default on an added REFERENCES column.
        let sql = "ALTER TABLE comments ADD COLUMN post_id INTEGER NOT NULL DEFAULT 0 \
                   REFERENCES posts(id);";
        assert!(
            classify_sql_for(DatabaseBackend::Sqlite, sql).is_empty(),
            "got {:?}",
            sqlite_ops(sql)
        );
    }

    #[test]
    fn sqlite_fk_action_set_default_is_not_a_column_default() {
        let sql = "ALTER TABLE t ADD COLUMN c INTEGER REFERENCES u(id) \
                   ON UPDATE SET DEFAULT ON DELETE CASCADE;";
        assert!(
            classify_sql_for(DatabaseBackend::Sqlite, sql).is_empty(),
            "got {:?}",
            sqlite_ops(sql)
        );
    }

    #[test]
    fn sqlite_unique_inside_a_string_literal_is_not_a_constraint() {
        let sql = "ALTER TABLE t ADD COLUMN d TEXT DEFAULT 'not unique yet';";
        assert!(
            classify_sql_for(DatabaseBackend::Sqlite, sql).is_empty(),
            "got {:?}",
            sqlite_ops(sql)
        );
    }

    #[test]
    fn sqlite_add_column_generated_stored_is_unsupported() {
        let f = sqlite_finding(
            "ALTER TABLE posts ADD COLUMN n INT GENERATED ALWAYS AS (id + 1) STORED;",
            "ADD COLUMN GENERATED … STORED (unsupported on SQLite)",
        )
        .expect("finding expected");
        assert_eq!(f.risk, RiskLevel::Unsupported);
        // VIRTUAL is accepted.
        assert!(
            classify_sql_for(
                DatabaseBackend::Sqlite,
                "ALTER TABLE posts ADD COLUMN n INT GENERATED ALWAYS AS (id + 1) VIRTUAL;"
            )
            .is_empty()
        );
    }

    #[test]
    fn sqlite_row_dependent_add_column_rules_skip_a_table_this_migration_creates() {
        // Verified against SQLite 3.45: on an EMPTY table these three apply
        // cleanly. UNIQUE and PRIMARY KEY do not — see the test below.
        let sql = "CREATE TABLE posts (id INTEGER PRIMARY KEY);\n\
                   ALTER TABLE posts ADD COLUMN a INT NOT NULL;\n\
                   ALTER TABLE posts ADD COLUMN b TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP;\n\
                   ALTER TABLE posts ADD COLUMN c INT GENERATED ALWAYS AS (id + 1) STORED;";
        assert!(
            !classify_sql_for(DatabaseBackend::Sqlite, sql)
                .iter()
                .any(|f| f.risk == RiskLevel::Unsupported),
            "got {:?}",
            sqlite_ops(sql)
        );
        // On a table the migration does not create, all three are unsupported.
        let existing = "ALTER TABLE posts ADD COLUMN a INT NOT NULL;\n\
                        ALTER TABLE posts ADD COLUMN b TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP;\n\
                        ALTER TABLE posts ADD COLUMN c INT GENERATED ALWAYS AS (id + 1) STORED;";
        assert_eq!(
            classify_sql_for(DatabaseBackend::Sqlite, existing)
                .iter()
                .filter(|f| f.risk == RiskLevel::Unsupported)
                .count(),
            3,
            "got {:?}",
            sqlite_ops(existing)
        );
    }

    #[test]
    fn sqlite_concurrently_must_be_the_keyword_not_a_name_prefix() {
        for sql in [
            "DROP INDEX concurrently_old_idx;",
            "CREATE INDEX concurrently_idx ON posts (x);",
        ] {
            assert!(
                !classify_sql_for(DatabaseBackend::Sqlite, sql)
                    .iter()
                    .any(|f| f.risk == RiskLevel::Unsupported),
                "{sql} names an index, it does not use CONCURRENTLY: {:?}",
                sqlite_ops(sql)
            );
        }
    }

    #[test]
    fn sqlite_rejects_every_alter_table_subcommand_outside_its_grammar() {
        for sql in [
            "ALTER TABLE t ADD CONSTRAINT ck CHECK (id > 0);",
            "ALTER TABLE t DROP CONSTRAINT ck;",
            "ALTER TABLE t SET SCHEMA other;",
            "ALTER TABLE t OWNER TO app;",
        ] {
            let findings = classify_sql_for(DatabaseBackend::Sqlite, sql);
            assert!(
                findings.iter().any(|f| f.risk == RiskLevel::Unsupported),
                "{sql} -> {:?}",
                sqlite_ops(sql)
            );
        }
        // The four forms SQLite does parse stay classified as before.
        for sql in [
            "ALTER TABLE t ADD COLUMN c TEXT;",
            "ALTER TABLE t DROP COLUMN c;",
            "ALTER TABLE t RENAME COLUMN a TO b;",
            "ALTER TABLE t RENAME TO u;",
        ] {
            assert!(
                !classify_sql_for(DatabaseBackend::Sqlite, sql)
                    .iter()
                    .any(|f| f.risk == RiskLevel::Unsupported),
                "{sql} -> {:?}",
                sqlite_ops(sql)
            );
        }
    }

    #[test]
    fn sqlite_unique_and_primary_key_are_rejected_even_on_a_fresh_table() {
        // Verified against SQLite 3.45: "Cannot add a UNIQUE column" and "Cannot
        // add a PRIMARY KEY column" fire on an empty table too, unlike the
        // NOT NULL / non-constant-default / STORED restrictions.
        let sql = "CREATE TABLE posts (id INTEGER PRIMARY KEY);\n\
                   ALTER TABLE posts ADD COLUMN a TEXT UNIQUE;\n\
                   ALTER TABLE posts ADD COLUMN b INTEGER PRIMARY KEY;";
        let ops = sqlite_ops(sql);
        assert!(
            ops.contains(&"ADD COLUMN UNIQUE (unsupported on SQLite)".to_owned()),
            "got {ops:?}"
        );
        assert!(
            ops.contains(&"ADD COLUMN PRIMARY KEY (unsupported on SQLite)".to_owned()),
            "got {ops:?}"
        );
    }

    #[test]
    fn sqlite_bare_add_and_drop_forms_get_the_same_rules() {
        // SQLite lets the COLUMN keyword be omitted.
        let f = sqlite_finding("ALTER TABLE posts DROP title;", "DROP COLUMN")
            .expect("a bare DROP is still destructive");
        assert_eq!(f.risk, RiskLevel::Destructive);

        let f = sqlite_finding(
            "ALTER TABLE posts ADD title TEXT NOT NULL;",
            "ADD COLUMN NOT NULL (no default)",
        )
        .expect("a bare ADD still hits the NOT NULL rule");
        assert_eq!(f.risk, RiskLevel::Unsupported);

        // The canonicalization must not swallow the Postgres-only forms.
        for sql in [
            "ALTER TABLE t ADD CONSTRAINT ck CHECK (id > 0);",
            "ALTER TABLE t DROP CONSTRAINT ck;",
        ] {
            assert!(
                classify_sql_for(DatabaseBackend::Sqlite, sql)
                    .iter()
                    .any(|f| f.risk == RiskLevel::Unsupported),
                "{sql} -> {:?}",
                sqlite_ops(sql)
            );
        }
    }

    #[test]
    fn sqlite_writing_cte_is_caught_through_formatting_whitespace() {
        let sql = "WITH removed AS ( DELETE FROM posts RETURNING id) SELECT * FROM removed;";
        assert!(
            classify_sql_for(DatabaseBackend::Sqlite, sql)
                .iter()
                .any(|f| f.risk == RiskLevel::Unsupported),
            "got {:?}",
            sqlite_ops(sql)
        );
    }

    #[test]
    fn sqlite_bare_rename_is_a_rename_column() {
        // SQLite accepts `RENAME old TO new` without `COLUMN` (#2580).
        let f = sqlite_finding(
            "ALTER TABLE posts RENAME title TO headline;",
            "RENAME COLUMN",
        )
        .expect("a bare column rename must be flagged");
        assert_eq!(f.risk, RiskLevel::Irreversible);
    }

    #[test]
    fn sqlite_bare_rename_is_not_a_rename_table() {
        let ops = sqlite_ops("ALTER TABLE posts RENAME title TO headline;");
        assert!(
            !ops.iter().any(|o| o.starts_with("RENAME TABLE")),
            "{ops:?}"
        );
        // `RENAME TO` still renames the table.
        assert!(sqlite_finding("ALTER TABLE posts RENAME TO articles;", "RENAME TABLE").is_some());
    }

    #[test]
    fn alter_table_subcommands_ignores_a_comma_in_a_string_literal() {
        let subs = alter_table_subcommands(
            DatabaseBackend::Sqlite,
            "alter table posts add column l text default 'a,b'",
        );
        assert_eq!(subs, ["add column l text default 'a,b'"]);
        let subs = alter_table_subcommands(
            DatabaseBackend::Sqlite,
            "alter table posts add column l text default 'it''s,x'",
        );
        assert_eq!(subs.len(), 1, "{subs:?}");
        let subs = alter_table_subcommands(
            DatabaseBackend::Sqlite,
            "alter table posts drop column \"a,b\"",
        );
        assert_eq!(subs.len(), 1, "{subs:?}");
    }

    #[test]
    fn alter_table_subcommands_splits_after_an_escaped_quote() {
        let subs = alter_table_subcommands(
            DatabaseBackend::Sqlite,
            "alter table posts add column a text default 'it''s', add column b text",
        );
        assert_eq!(
            subs,
            ["add column a text default 'it''s'", "add column b text"]
        );
        let subs = alter_table_subcommands(
            DatabaseBackend::Sqlite,
            "alter table posts add column a text default 'it''s,x'",
        );
        assert_eq!(subs, ["add column a text default 'it''s,x'"]);
    }

    // ── #3527 (issue #3521): Postgres ALTER TABLE split ──────────────────────

    fn pg_subs(sql: &str) -> Vec<&str> {
        alter_table_subcommands(DatabaseBackend::Postgres, sql)
    }

    #[test]
    fn pg_split_ignores_a_paren_in_a_string_literal() {
        for lit in ["'('", "')'", "'/* ( */'", "'it''s ('"] {
            let sql = format!("alter table t add column a text default {lit}, add column b int");
            assert_eq!(pg_subs(&sql).len(), 2, "{lit}");
        }
    }

    #[test]
    fn pg_split_ignores_a_paren_in_a_dollar_quote() {
        for lit in ["$$($$", "$tag$($tag$", "$$ ( ' $$"] {
            let sql = format!("alter table t add column a text default {lit}, add column b int");
            assert_eq!(pg_subs(&sql).len(), 2, "{lit}");
        }
    }

    #[test]
    fn pg_split_honours_backslash_escapes_in_e_strings() {
        let subs = pg_subs("alter table t add column a text default e'it\\'s', add column b int");
        assert_eq!(subs.len(), 2, "{subs:?}");
        let subs = pg_subs("alter table t add column a text default e'\\'(', add column b int");
        assert_eq!(subs.len(), 2, "{subs:?}");
    }

    #[test]
    fn pg_split_ignores_a_quote_inside_a_quoted_name() {
        for name in [r#""a'b""#, r#""$$""#, r#""e'""#, r#""a""'b""#, r#""a(b""#] {
            let sql = format!("alter table t add column {name} int, drop column c");
            let subs = pg_subs(&sql);
            assert_eq!(subs.len(), 2, "{name}: {subs:?}");
        }
    }

    #[test]
    fn quote_in_a_quoted_name_does_not_hide_a_drop_column() {
        let sql = r#"ALTER TABLE t ADD COLUMN "a'b" int, DROP COLUMN c;"#;
        assert_eq!(ops(DatabaseBackend::Postgres, sql), ["DROP COLUMN"]);
    }

    #[test]
    fn dollar_tag_cannot_start_with_a_digit() {
        assert_eq!(dollar_quote_len("$1$ x $1$"), None);
        assert_eq!(dollar_quote_len("$a1$ x $a1$"), Some(11));
        let subs =
            pg_subs("alter table t add column a int default $1$, add column b int, drop column c$");
        assert_eq!(subs.len(), 3, "{subs:?}");
    }

    #[test]
    fn pg_split_reads_a_quoted_table_name_with_spaces() {
        for table in [r#""my table""#, r#""x add column y""#, r#"s."my table""#] {
            let sql = format!(
                "alter table {table} add column a text default 'ok', add column b int not null"
            );
            assert_eq!(
                pg_subs(&sql),
                [
                    "add column a text default 'ok'",
                    "add column b int not null"
                ],
                "{table}"
            );
        }
    }

    #[test]
    fn dollar_tag_case_does_not_close_a_literal_early() {
        for lit in [
            "$tag$ $TAG$ $tag$",
            "$tag$ x$TAG$y $tag$",
            "$Tag$ $tag$ $Tag$",
            "$Tag$ $X$Tag$",
            "$a$ $A$A$ $a$",
            "'$A$', $B$ $b$ $B$",
        ] {
            let sql = format!(
                "ALTER TABLE t ADD COLUMN a text DEFAULT {lit}, ADD COLUMN b int NOT NULL;"
            );
            assert!(
                ops(DatabaseBackend::Postgres, &sql)
                    .contains(&"ADD COLUMN NOT NULL (no default)".to_owned()),
                "{lit}"
            );
        }
    }

    #[test]
    fn normalize_keeps_quoted_spans_verbatim() {
        let pg = DatabaseBackend::Postgres;
        for (sql, want) in [
            ("SELECT $Tag$ ABC", "select $Tag$ ABC"),
            ("SELECT $Tag$", "select $Tag$"),
            ("SELECT E'A''B'", "select E'A''B'"),
            ("SELECT E'A\\'B' X", "select E'A\\'B' x"),
            ("SELECT E'AB", "select E'AB"),
            (
                "SELECT $Tag$ ABC $Tag$ $$ X $$",
                "select $Tag$ ABC $Tag$ $$ X $$",
            ),
        ] {
            assert_eq!(normalize_statement(pg, sql), want, "{sql}");
        }
    }

    #[test]
    fn pg_split_handles_unclosed_and_escaped_names_and_e_strings() {
        assert_eq!(
            pg_subs(r#"alter table t drop column "a, drop column b"#).len(),
            1
        );
        assert!(pg_subs(r#"alter table "t, add column a int"#).is_empty());
        let subs = pg_subs("alter table t add column a text default e'x''(', drop column b");
        assert_eq!(subs.len(), 2, "{subs:?}");
        assert_eq!(
            pg_subs("alter table t add column a text default e'x(, y").len(),
            1
        );
    }

    #[test]
    fn sqlite_table_name_with_a_double_quote_inside_backticks() {
        let sql = "ALTER TABLE `a\"b` ADD COLUMN c INT NOT NULL;";
        assert_eq!(
            ops(DatabaseBackend::Sqlite, sql),
            ["ADD COLUMN NOT NULL (no default)"]
        );
    }

    #[test]
    fn pg_split_keeps_a_comma_inside_a_literal_together() {
        let subs = pg_subs("alter table t add column a text default 'x,y', drop column b");
        assert_eq!(subs, ["add column a text default 'x,y'", "drop column b"]);
    }

    #[test]
    fn pg_split_does_not_treat_a_name_ending_in_e_or_dollar_as_a_quote() {
        let subs = pg_subs("alter table t add column name text default ('a'), drop column b");
        assert_eq!(subs.len(), 2, "{subs:?}");
        let subs = pg_subs("alter table t add column a$b$ int, drop column b");
        assert_eq!(subs.len(), 2, "{subs:?}");
    }

    #[test]
    fn pg_split_survives_an_unclosed_literal() {
        for sql in [
            "alter table t add column a text default '(, x",
            "alter table t add column a text default $$(, x",
        ] {
            assert_eq!(pg_subs(sql).len(), 1, "{sql}");
        }
    }

    #[test]
    fn paren_in_a_literal_does_not_hide_a_finding() {
        let sql = "ALTER TABLE t ADD COLUMN a text DEFAULT '(', ADD COLUMN b int NOT NULL;";
        assert_eq!(
            ops(DatabaseBackend::Postgres, sql),
            ["ADD COLUMN NOT NULL (no default)"]
        );
        for lit in ["'('", "')'", "$$($$", "'/* ( */'"] {
            let sql = format!(
                "ALTER TABLE t ADD COLUMN a text DEFAULT {lit}, DROP CONSTRAINT important_check;"
            );
            let f = classify_sql_for(DatabaseBackend::Postgres, &sql);
            assert!(
                f.iter().any(|f| f.risk == RiskLevel::ManualReview),
                "{lit}: {f:?}"
            );
        }
    }

    #[test]
    fn alter_table_subcommands_ignores_a_comma_in_backtick_and_bracket_names() {
        for sql in [
            "alter table posts drop `a,b`",
            "alter table posts drop [a,b]",
        ] {
            assert_eq!(
                alter_table_subcommands(DatabaseBackend::Sqlite, sql).len(),
                1,
                "{sql}"
            );
        }
    }

    #[test]
    fn sqlite_rewrite_keeps_the_head_for_odd_comma_spacing() {
        assert_eq!(
            canonical_sqlite_alter("alter table posts add a text,add b text"),
            "alter table posts add column a text, add column b text"
        );
        assert_eq!(
            canonical_sqlite_alter("alter table posts add a text , add b text"),
            "alter table posts add column a text, add column b text"
        );
    }

    #[test]
    fn sqlite_rename_edge_spellings() {
        for sql in [
            "ALTER TABLE posts RENAME COLUMN a TO b;",
            "ALTER TABLE posts RENAME \"to\" TO headline;",
            "ALTER TABLE posts RENAME \"a to b\" TO c;",
        ] {
            assert!(
                sqlite_finding(sql, "RENAME COLUMN").is_some(),
                "{sql} -> {:?}",
                sqlite_ops(sql)
            );
        }
        let sql = "ALTER TABLE posts RENAME TO \"a to b\";";
        assert!(sqlite_finding(sql, "RENAME TABLE").is_some());
        assert!(sqlite_finding(sql, "RENAME COLUMN").is_none());
    }

    #[test]
    fn postgres_dollar_quote_cannot_hide_a_drop_column() {
        // The apostrophe in `$$..$$` must not merge the DROP COLUMN into the
        // ADD COLUMN.
        let f = classify_sql_for(
            DatabaseBackend::Postgres,
            "ALTER TABLE t ADD COLUMN a text DEFAULT $$it's$$, DROP COLUMN b;",
        );
        assert!(f.iter().any(|f| f.operation == "DROP COLUMN"), "{f:?}");
    }

    #[test]
    fn postgres_alter_classification_is_unchanged() {
        let ops = |sql: &str| -> Vec<String> {
            classify_sql_for(DatabaseBackend::Postgres, sql)
                .into_iter()
                .map(|f| f.operation)
                .collect()
        };
        let quoted = ops("ALTER TABLE posts ADD COLUMN l text DEFAULT 'a,b';");
        assert!(
            !quoted.iter().any(|o| o.starts_with("Multi-action")),
            "{quoted:?}"
        );
        // Postgres has no canonicalization, so a bare rename stays unrewritten.
        let bare = ops("ALTER TABLE posts RENAME title TO headline;");
        assert!(!bare.iter().any(|o| o == "RENAME COLUMN"), "{bare:?}");
    }

    #[test]
    fn sqlite_comma_in_a_default_literal_is_not_multi_action() {
        let sql = "ALTER TABLE posts ADD COLUMN label TEXT DEFAULT 'a,b';";
        assert!(
            classify_sql_for(DatabaseBackend::Sqlite, sql).is_empty(),
            "{:?}",
            sqlite_ops(sql)
        );
        // A real second action still trips the rule.
        assert!(
            sqlite_finding(
                "ALTER TABLE posts ADD COLUMN a TEXT DEFAULT 'x,y', ADD COLUMN b TEXT;",
                "Multi-action ALTER TABLE (unsupported on SQLite)",
            )
            .is_some()
        );
    }

    #[test]
    fn sqlite_rejects_a_multi_action_alter_table() {
        // SQLite takes one action per ALTER TABLE; both actions are individually
        // valid, so only the statement-level rule catches this.
        let f = sqlite_finding(
            "ALTER TABLE posts ADD COLUMN a TEXT, ADD COLUMN b TEXT;",
            "Multi-action ALTER TABLE (unsupported on SQLite)",
        )
        .expect("finding expected");
        assert_eq!(f.risk, RiskLevel::Unsupported);
        // One action is fine.
        assert!(
            classify_sql_for(
                DatabaseBackend::Sqlite,
                "ALTER TABLE posts ADD COLUMN a TEXT;"
            )
            .is_empty()
        );
        // Postgres allows the list.
        assert!(classify_sql("ALTER TABLE posts ADD COLUMN a TEXT, ADD COLUMN b TEXT;").is_empty());
    }

    #[test]
    fn sqlite_alter_table_never_falls_through_to_manual_review() {
        // Every subcommand is classified, so the Postgres catch-all must not fire.
        for sql in [
            "ALTER TABLE t ALTER COLUMN c SET NOT NULL;",
            "ALTER TABLE t ADD COLUMN c INTEGER PRIMARY KEY;",
            "ALTER TABLE t VALIDATE CONSTRAINT ck;",
        ] {
            assert!(
                !sqlite_ops(sql).contains(&"Unclassified ALTER TABLE".to_owned()),
                "{sql} -> {:?}",
                sqlite_ops(sql)
            );
        }
    }

    #[test]
    fn sqlite_rejects_postgres_only_index_and_table_clauses() {
        for sql in [
            "CREATE INDEX i ON posts USING btree (a);",
            "CREATE UNIQUE INDEX i ON posts (a) INCLUDE (b);",
            "CREATE INDEX i ON posts (a) WITH (fillfactor=70);",
            "CREATE TABLE p (id int) PARTITION BY RANGE (id);",
            "CREATE TABLE r (id int GENERATED BY DEFAULT AS IDENTITY);",
        ] {
            assert!(
                classify_sql_for(DatabaseBackend::Sqlite, sql)
                    .iter()
                    .any(|f| f.risk == RiskLevel::Unsupported),
                "{sql} -> {:?}",
                sqlite_ops(sql)
            );
        }
        // A partial / COLLATE / DESC index is valid SQLite.
        let ok = "CREATE INDEX i ON posts (title COLLATE NOCASE DESC) WHERE deleted_at IS NULL;";
        assert!(
            !classify_sql_for(DatabaseBackend::Sqlite, ok)
                .iter()
                .any(|f| f.risk == RiskLevel::Unsupported),
            "{ok} -> {:?}",
            sqlite_ops(ok)
        );
    }

    #[test]
    fn sqlite_unsupported_clauses_ignore_string_literal_contents() {
        for sql in [
            "INSERT INTO audit_log(message) VALUES ('waiting for update');",
            "INSERT INTO audit_log(message) VALUES ('on conflict on constraint x');",
        ] {
            assert!(
                !classify_sql_for(DatabaseBackend::Sqlite, sql)
                    .iter()
                    .any(|f| f.risk == RiskLevel::Unsupported),
                "{sql} is ordinary data, got {:?}",
                sqlite_ops(sql)
            );
        }
    }

    #[test]
    fn sqlite_rejects_postgres_only_dml_forms() {
        for sql in [
            "MERGE INTO t USING u ON t.id = u.id WHEN MATCHED THEN UPDATE SET a = 1;",
            "SELECT id FROM t FOR UPDATE;",
            "INSERT INTO t (a) VALUES (1) ON CONFLICT ON CONSTRAINT t_pkey DO NOTHING;",
            "WITH d AS (DELETE FROM t RETURNING id) SELECT * FROM d;",
        ] {
            assert!(
                classify_sql_for(DatabaseBackend::Sqlite, sql)
                    .iter()
                    .any(|f| f.risk == RiskLevel::Unsupported),
                "{sql} -> {:?}",
                sqlite_ops(sql)
            );
        }
    }

    #[test]
    fn sqlite_cannot_drop_an_implicit_index() {
        for sql in [
            "DROP INDEX sqlite_autoindex_posts_1;",
            "DROP INDEX IF EXISTS sqlite_autoindex_posts_1;",
        ] {
            let findings = classify_sql_for(DatabaseBackend::Sqlite, sql);
            assert!(
                findings.iter().any(|f| f.risk == RiskLevel::Unsupported),
                "{sql} -> {:?}",
                sqlite_ops(sql)
            );
        }
    }

    #[test]
    fn sqlite_covers_every_postgres_only_object_statement() {
        for sql in [
            "GRANT SELECT ON posts TO app;",
            "REVOKE SELECT ON posts FROM app;",
            "DROP EXTENSION pgcrypto;",
            "DROP MATERIALIZED VIEW mv;",
            "REFRESH MATERIALIZED VIEW mv;",
            "CREATE UNIQUE INDEX CONCURRENTLY i ON posts (a);",
            "DROP INDEX CONCURRENTLY i;",
            "ALTER SEQUENCE s RESTART;",
        ] {
            assert!(
                classify_sql_for(DatabaseBackend::Sqlite, sql)
                    .iter()
                    .any(|f| f.risk == RiskLevel::Unsupported),
                "{sql} -> {:?}",
                sqlite_ops(sql)
            );
        }
    }

    #[test]
    fn finding_prose_has_no_stray_whitespace_runs() {
        // A Rust `\` string continuation eats the newline and the indentation;
        // losing one leaves a space run in text printed straight to stderr.
        let samples = [
            "ALTER TABLE t ALTER COLUMN c TYPE TEXT;",
            "ALTER TABLE t ADD COLUMN c TEXT NOT NULL;",
            "ALTER TABLE t ADD COLUMN c TEXT UNIQUE;",
            "ALTER TABLE t DROP COLUMN c;",
            "TRUNCATE TABLE t;",
            "CREATE INDEX i ON t (a);",
            "CREATE SEQUENCE s;",
            "CREATE TYPE m AS ENUM ('a');",
            "CREATE EXTENSION pgcrypto;",
            "COMMENT ON TABLE t IS 'x';",
            "CREATE MATERIALIZED VIEW mv AS SELECT 1;",
            "GRANT SELECT ON t TO app;",
            "MERGE INTO t USING u ON t.id = u.id WHEN MATCHED THEN UPDATE SET a = 1;",
            "CREATE INDEX i ON t USING btree (a);",
            "CREATE TABLE p (id int) PARTITION BY RANGE (id);",
            "SELECT id FROM t FOR UPDATE;",
            "WITH d AS (DELETE FROM t RETURNING id) SELECT * FROM d;",
            "DROP INDEX sqlite_autoindex_t_1;",
            "ALTER TABLE t ADD COLUMN n INT GENERATED ALWAYS AS (id + 1) STORED;",
        ];
        for backend in [DatabaseBackend::Postgres, DatabaseBackend::Sqlite] {
            for sql in samples {
                for f in classify_sql_for(backend, sql) {
                    assert!(!f.why.contains("  "), "{sql}: why = {:?}", f.why);
                    assert!(
                        !f.next_action.contains("  "),
                        "{sql}: next_action = {:?}",
                        f.next_action
                    );
                }
            }
        }
    }
}