rdbms-playground/src/dsl/parser.rs

//! Grammar-based DSL parser built on chumsky.
//!
//! The parser produces a `Command` AST directly — there is no
//! intermediate token tree to translate. Composable rules
//! (identifier, type keyword, padded keyword) are defined once
//! and reused across command variants, which is the point of
//! choosing a grammar approach (see Phase 2/3 selection).
//!
//! Errors from chumsky are mapped to the local `ParseError` type
//! so callers do not depend on chumsky's API surface — that
//! keeps the parser swappable if we ever revisit the choice.

use chumsky::error::RichReason;
use chumsky::prelude::*;

use crate::dsl::action::ReferentialAction;
use crate::dsl::command::{
    ChangeColumnMode, ColumnSpec, Command, RelationshipSelector, RowFilter,
};
use crate::dsl::types::Type;
use crate::dsl::value::Value;

#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum ParseError {
    #[error("could not parse command: {message}")]
    Invalid { message: String, position: usize },
    #[error("empty input")]
    Empty,
}

impl ParseError {
    #[must_use]
    pub const fn position(&self) -> Option<usize> {
        match self {
            Self::Invalid { position, .. } => Some(*position),
            Self::Empty => None,
        }
    }
}

/// Parse a single DSL command.
pub fn parse_command(input: &str) -> Result<Command, ParseError> {
    let trimmed = input.trim();
    if trimmed.is_empty() {
        return Err(ParseError::Empty);
    }
    match command_parser().parse(trimmed).into_result() {
        Ok(cmd) => Ok(cmd),
        Err(errs) => Err(into_parse_error(&errs, trimmed)),
    }
}

fn into_parse_error(errs: &[Rich<'_, char>], input: &str) -> ParseError {
    // Prefer custom-reason errors over chumsky's structural
    // ones — those carry our friendly messages from `try_map`
    // (e.g. "unknown type 'varchar' (expected one of: ...)").
    let chosen = errs
        .iter()
        .find(|e| has_custom_reason(e.reason()))
        .unwrap_or_else(|| errs.first().expect("parser failure with no error"));
    let span = chosen.span();
    let position = span.start;
    let message = humanise(chosen, input);
    ParseError::Invalid { message, position }
}

const fn has_custom_reason<T, C>(reason: &RichReason<'_, T, C>) -> bool {
    matches!(reason, RichReason::Custom(_))
}

fn humanise(err: &Rich<'_, char>, input: &str) -> String {
    // For custom errors, the underlying message is what we want
    // to show, not chumsky's "found ... expected ..." rendering.
    if let Some(msg) = first_custom_message(err.reason()) {
        return msg;
    }
    let span = err.span();
    let snippet: String = input
        .chars()
        .skip(span.start)
        .take((span.end - span.start).max(1))
        .collect();
    if snippet.is_empty() {
        format!("{err}")
    } else {
        format!("{err} (near `{snippet}`)")
    }
}

fn first_custom_message<T>(reason: &RichReason<'_, T, String>) -> Option<String> {
    match reason {
        RichReason::Custom(msg) => Some(msg.clone()),
        RichReason::ExpectedFound { .. } => None,
    }
}

/// The top-level command parser.
fn command_parser<'a>()
-> impl Parser<'a, &'a str, Command, extra::Err<Rich<'a, char>>> + Clone {
    let create_table = keyword_ci("create")
        .ignore_then(keyword_ci("table"))
        .ignore_then(identifier())
        .then(with_pk_clause())
        .try_map(|(name, pk_specs), span| {
            if pk_specs.is_empty() {
                return Err(Rich::custom(
                    span,
                    "tables need at least one column. Add `with pk` for a default \
                     `id INTEGER PRIMARY KEY`, or `with pk <name>:<type>` to choose. \
                     Use a comma-separated list for compound primary keys."
                        .to_string(),
                ));
            }
            let columns: Vec<ColumnSpec> = pk_specs
                .iter()
                .map(|(n, t)| ColumnSpec {
                    name: n.clone(),
                    ty: *t,
                })
                .collect();
            let primary_key = pk_specs.into_iter().map(|(n, _)| n).collect();
            Ok(Command::CreateTable {
                name,
                columns,
                primary_key,
            })
        });

    let drop_table = keyword_ci("drop")
        .ignore_then(keyword_ci("table"))
        .ignore_then(identifier())
        .map(|name| Command::DropTable { name });

    // Both `to` and `table` are independently optional —
    // `add column to table T: c (text)`,
    // `add column to T: c (text)`,
    // `add column table T: c (text)`,
    // and `add column T: c (text)` all parse identically.
    // Matches the convention elsewhere in the DSL where bare
    // identifiers are accepted in unambiguous positions.
    let add_column = keyword_ci("add")
        .ignore_then(keyword_ci("column"))
        .ignore_then(optional_keyword("to"))
        .ignore_then(optional_keyword("table"))
        .ignore_then(identifier())
        .then_ignore(just(':').padded())
        .then(identifier())
        .then_ignore(just('(').padded())
        .then(type_keyword())
        .then_ignore(just(')').padded())
        .map(|((table, column), ty)| Command::AddColumn { table, column, ty });

    // `drop column [from] [table] <T>: <col>`. Both
    // prepositions independently optional, matching the
    // `add column` shape for symmetry.
    let drop_column = keyword_ci("drop")
        .ignore_then(keyword_ci("column"))
        .ignore_then(optional_keyword("from"))
        .ignore_then(optional_keyword("table"))
        .ignore_then(identifier())
        .then_ignore(just(':').padded())
        .then(identifier())
        .map(|(table, column)| Command::DropColumn { table, column });

    // `rename column [in] [table] <T>: <old> to <new>`.
    let rename_column = keyword_ci("rename")
        .ignore_then(keyword_ci("column"))
        .ignore_then(optional_keyword("in"))
        .ignore_then(optional_keyword("table"))
        .ignore_then(identifier())
        .then_ignore(just(':').padded())
        .then(identifier())
        .then_ignore(keyword_ci("to"))
        .then(identifier())
        .map(|((table, old), new)| Command::RenameColumn { table, old, new });

    // `change column [in] [table] <T>: <col> (<newtype>) [flags]`
    // where `flags` is at most one of `--force-conversion` /
    // `--dont-convert` (mutually exclusive at parse time per
    // ADR-0017 §5).
    let change_column = keyword_ci("change")
        .ignore_then(keyword_ci("column"))
        .ignore_then(optional_keyword("in"))
        .ignore_then(optional_keyword("table"))
        .ignore_then(identifier())
        .then_ignore(just(':').padded())
        .then(identifier())
        .then_ignore(just('(').padded())
        .then(type_keyword())
        .then_ignore(just(')').padded())
        .then(change_column_flags())
        .map(|(((table, column), ty), mode)| Command::ChangeColumnType {
            table,
            column,
            ty,
            mode,
        });

    let add_relationship = add_relationship_parser();
    let drop_relationship = drop_relationship_parser();

    let show_data = keyword_ci("show")
        .ignore_then(keyword_ci("data"))
        .ignore_then(identifier())
        .map(|name| Command::ShowData { name });

    let show_table = keyword_ci("show")
        .ignore_then(keyword_ci("table"))
        .ignore_then(identifier())
        .map(|name| Command::ShowTable { name });

    let insert_cmd = insert_parser();
    let update_cmd = update_parser();
    let delete_cmd = delete_parser();

    choice((
        create_table,
        // `drop column` and `drop relationship` come before
        // `drop table` because both are more specific —
        // chumsky's `choice` tries each in order.
        drop_column,
        drop_relationship,
        drop_table,
        add_column,
        add_relationship,
        rename_column,
        change_column,
        // Order: `show data` before `show table` because both
        // start with `show` and the longer keyword is checked
        // first via this ordering.
        show_data,
        show_table,
        insert_cmd,
        update_cmd,
        delete_cmd,
    ))
    .padded()
    .then_ignore(end())
}

/// INSERT, accepting three shapes:
///   `insert into T (cols) values (vals)` — explicit columns
///   `insert into T values (vals)`        — implicit column order
///   `insert into T (vals)`               — short form, omits `values`
///
/// The short form is disambiguated from the column-list form by
/// trying both alternatives in order; chumsky's `choice`
/// backtracks, and only the all-literals form parses without
/// `values`.
fn insert_parser<'a>()
-> impl Parser<'a, &'a str, Command, extra::Err<Rich<'a, char>>> + Clone {
    let column_list = just('(')
        .padded()
        .ignore_then(
            identifier()
                .separated_by(just(',').padded())
                .at_least(1)
                .collect::<Vec<_>>(),
        )
        .then_ignore(just(')').padded());

    let value_list = just('(')
        .padded()
        .ignore_then(
            value_literal()
                .separated_by(just(',').padded())
                .at_least(1)
                .collect::<Vec<_>>(),
        )
        .then_ignore(just(')').padded());

    let with_columns_and_values = column_list
        .clone()
        .then_ignore(keyword_ci("values"))
        .then(value_list.clone())
        .map(|(cols, vals)| (Some(cols), vals));

    let with_values_keyword_only = keyword_ci("values")
        .ignore_then(value_list.clone())
        .map(|vals| (None, vals));

    let bare_value_list = value_list.map(|vals| (None, vals));

    keyword_ci("insert")
        .ignore_then(keyword_ci("into"))
        .ignore_then(identifier())
        .then(choice((
            with_columns_and_values,
            with_values_keyword_only,
            bare_value_list,
        )))
        .map(|(table, (columns, values))| Command::Insert {
            table,
            columns,
            values,
        })
}

/// `update <T> set <col>=<val>[, <col>=<val>...] (where <col>=<val> | --all-rows)`.
fn update_parser<'a>()
-> impl Parser<'a, &'a str, Command, extra::Err<Rich<'a, char>>> + Clone {
    let assignment = identifier()
        .then_ignore(just('=').padded())
        .then(value_literal());

    let assignments = assignment
        .separated_by(just(',').padded())
        .at_least(1)
        .collect::<Vec<_>>();

    keyword_ci("update")
        .ignore_then(identifier())
        .then_ignore(keyword_ci("set"))
        .then(assignments)
        .then(filter_clause())
        .map(|((table, assignments), filter)| Command::Update {
            table,
            assignments,
            filter,
        })
}

/// `delete from <T> (where <col>=<val> | --all-rows)`.
fn delete_parser<'a>()
-> impl Parser<'a, &'a str, Command, extra::Err<Rich<'a, char>>> + Clone {
    keyword_ci("delete")
        .ignore_then(keyword_ci("from"))
        .ignore_then(identifier())
        .then(filter_clause())
        .map(|(table, filter)| Command::Delete { table, filter })
}

/// Parse the row-filter portion of UPDATE/DELETE: either
/// `where <col>=<val>` or the `--all-rows` flag, with the two
/// being mutually exclusive (specifying both is a parse error).
fn filter_clause<'a>()
-> impl Parser<'a, &'a str, RowFilter, extra::Err<Rich<'a, char>>> + Clone {
    let where_clause = keyword_ci("where")
        .ignore_then(identifier())
        .then_ignore(just('=').padded())
        .then(value_literal())
        .map(|(column, value)| RowFilter::Where { column, value });

    let all_rows = just("--all-rows").padded().to(RowFilter::AllRows);

    where_clause.or(all_rows).labelled("where clause or --all-rows")
}

/// Parse a value literal: number, single-quoted string, `null`,
/// `true`, or `false`.
fn value_literal<'a>()
-> impl Parser<'a, &'a str, Value, extra::Err<Rich<'a, char>>> + Clone {
    choice((
        keyword_ci("null").to(Value::Null),
        keyword_ci("true").to(Value::Bool(true)),
        keyword_ci("false").to(Value::Bool(false)),
        number_literal(),
        string_literal(),
    ))
    .padded()
}

fn number_literal<'a>()
-> impl Parser<'a, &'a str, Value, extra::Err<Rich<'a, char>>> + Clone {
    let sign = just('-').or_not();
    let digits = any()
        .filter(|c: &char| c.is_ascii_digit())
        .repeated()
        .at_least(1)
        .collect::<String>();
    let fraction = just('.')
        .ignore_then(
            any()
                .filter(|c: &char| c.is_ascii_digit())
                .repeated()
                .at_least(1)
                .collect::<String>(),
        )
        .or_not();
    sign.then(digits)
        .then(fraction)
        .map(|((s, whole), frac)| {
            let mut out = String::new();
            if s.is_some() {
                out.push('-');
            }
            out.push_str(&whole);
            if let Some(f) = frac {
                out.push('.');
                out.push_str(&f);
            }
            Value::Number(out)
        })
}

fn string_literal<'a>()
-> impl Parser<'a, &'a str, Value, extra::Err<Rich<'a, char>>> + Clone {
    // Single-quoted SQL string. `''` inside the literal escapes
    // a literal single quote.
    let body = just('\'')
        .ignore_then(
            choice((
                just("''").to('\''),
                any().filter(|c: &char| *c != '\''),
            ))
            .repeated()
            .collect::<String>(),
        )
        .then_ignore(just('\''));
    body.map(Value::Text)
}

/// `add 1:n relationship [<name>] from <P>.<col> to <C>.<col>
///   [on delete <action>] [on update <action>] [--create-fk]`.
fn add_relationship_parser<'a>()
-> impl Parser<'a, &'a str, Command, extra::Err<Rich<'a, char>>> + Clone {
    let one_to_n = just('1').padded().ignore_then(just(':').padded()).ignore_then(
        any()
            .filter(|c: &char| *c == 'n' || *c == 'N')
            .padded(),
    );

    let optional_name = keyword_ci("as").ignore_then(identifier()).or_not();

    keyword_ci("add")
        .ignore_then(one_to_n)
        .ignore_then(keyword_ci("relationship"))
        .ignore_then(optional_name)
        .then_ignore(keyword_ci("from"))
        .then(qualified_column())
        .then_ignore(keyword_ci("to"))
        .then(qualified_column())
        .then(referential_clauses())
        .then(create_fk_flag())
        .map(
            |((((name, parent), child), (on_delete, on_update)), create_fk)| {
                Command::AddRelationship {
                    name,
                    parent_table: parent.0,
                    parent_column: parent.1,
                    child_table: child.0,
                    child_column: child.1,
                    on_delete,
                    on_update,
                    create_fk,
                }
            },
        )
}

/// `drop relationship <name>` or
/// `drop relationship from <P>.<col> to <C>.<col>`.
fn drop_relationship_parser<'a>()
-> impl Parser<'a, &'a str, Command, extra::Err<Rich<'a, char>>> + Clone {
    let endpoints_form = keyword_ci("from")
        .ignore_then(qualified_column())
        .then_ignore(keyword_ci("to"))
        .then(qualified_column())
        .map(|(parent, child)| RelationshipSelector::Endpoints {
            parent_table: parent.0,
            parent_column: parent.1,
            child_table: child.0,
            child_column: child.1,
        });

    let named_form = identifier().map(|name| RelationshipSelector::Named { name });

    keyword_ci("drop")
        .ignore_then(keyword_ci("relationship"))
        .ignore_then(choice((endpoints_form, named_form)))
        .map(|selector| Command::DropRelationship { selector })
}

/// Parse `<Table>.<Column>` returning (table, column).
fn qualified_column<'a>()
-> impl Parser<'a, &'a str, (String, String), extra::Err<Rich<'a, char>>> + Clone {
    identifier()
        .then_ignore(just('.').padded())
        .then(identifier())
}

/// Optional `on delete <action>` and/or `on update <action>`,
/// in either order. Default to `NoAction` when omitted.
fn referential_clauses<'a>() -> impl Parser<
    'a,
    &'a str,
    (ReferentialAction, ReferentialAction),
    extra::Err<Rich<'a, char>>,
> + Clone {
    let target = keyword_ci("delete")
        .to(ReferentialActionTarget::Delete)
        .or(keyword_ci("update").to(ReferentialActionTarget::Update));
    let clause = keyword_ci("on")
        .ignore_then(target)
        .then(action_keyword())
        .map(|(t, a)| (t, a));
    clause
        .repeated()
        .at_most(2)
        .collect::<Vec<_>>()
        .try_map(|clauses, span| {
            let mut on_delete = None;
            let mut on_update = None;
            for (target, action) in clauses {
                let slot = match target {
                    ReferentialActionTarget::Delete => &mut on_delete,
                    ReferentialActionTarget::Update => &mut on_update,
                };
                if slot.is_some() {
                    return Err(Rich::custom(
                        span,
                        format!("`on {target}` specified twice"),
                    ));
                }
                *slot = Some(action);
            }
            Ok((
                on_delete.unwrap_or_else(ReferentialAction::default_action),
                on_update.unwrap_or_else(ReferentialAction::default_action),
            ))
        })
}

#[derive(Debug, Clone, Copy)]
enum ReferentialActionTarget {
    Delete,
    Update,
}

impl std::fmt::Display for ReferentialActionTarget {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(match self {
            Self::Delete => "delete",
            Self::Update => "update",
        })
    }
}

/// Parse a referential-action keyword: `cascade`, `restrict`,
/// `set null`, or `no action`. The two-word forms come first in
/// the alternatives so they're tried before the one-word forms;
/// because the first words are unique to each phrase
/// (`set`/`no` for two-word, `cascade`/`restrict` for one-word)
/// there is no ambiguity.
fn action_keyword<'a>()
-> impl Parser<'a, &'a str, ReferentialAction, extra::Err<Rich<'a, char>>> + Clone {
    choice((
        keyword_ci("set")
            .ignore_then(keyword_ci("null"))
            .to(ReferentialAction::SetNull),
        keyword_ci("no")
            .ignore_then(keyword_ci("action"))
            .to(ReferentialAction::NoAction),
        keyword_ci("cascade").to(ReferentialAction::Cascade),
        keyword_ci("restrict").to(ReferentialAction::Restrict),
    ))
}

fn create_fk_flag<'a>()
-> impl Parser<'a, &'a str, bool, extra::Err<Rich<'a, char>>> + Clone {
    just("--create-fk")
        .padded()
        .or_not()
        .map(|opt| opt.is_some())
}

/// Optional flags for `change column …` (ADR-0017 §5).
/// Allows zero or one of the two mutually-exclusive flags;
/// emits a custom parse error if both are present, naming both
/// flags so the user knows what the conflict is.
fn change_column_flags<'a>()
-> impl Parser<'a, &'a str, ChangeColumnMode, extra::Err<Rich<'a, char>>> + Clone {
    let force = just("--force-conversion")
        .padded()
        .to(ChangeColumnMode::ForceConversion);
    let dont = just("--dont-convert")
        .padded()
        .to(ChangeColumnMode::DontConvert);
    choice((force, dont))
        .repeated()
        .collect::<Vec<_>>()
        .try_map(|flags, span| match flags.as_slice() {
            [] => Ok(ChangeColumnMode::Default),
            [single] => Ok(*single),
            _ => Err(Rich::custom(
                span,
                "`--force-conversion` and `--dont-convert` are mutually \
                 exclusive — pick one."
                    .to_string(),
            )),
        })
}

/// Parse the optional `with pk [<spec>]` clause that may follow
/// `create table <Name>`. Returns the list of (name, type) pairs
/// that form the primary key. An absent clause returns an empty
/// vector; a present `with pk` (no spec) returns the default
/// `id:serial`. Compound PK is a comma-separated list of specs.
fn with_pk_clause<'a>()
-> impl Parser<'a, &'a str, Vec<(String, Type)>, extra::Err<Rich<'a, char>>> + Clone {
    let single = identifier()
        .then_ignore(just(':').padded())
        .then(type_keyword())
        .map(|(name, ty)| (name, ty));

    let spec_list = single
        .clone()
        .separated_by(just(',').padded())
        .at_least(1)
        .collect::<Vec<_>>();

    keyword_ci("with")
        .ignore_then(keyword_ci("pk"))
        .ignore_then(spec_list.or_not())
        .map(|maybe_specs| {
            // `with pk` alone defaults to a serial id PK.
            maybe_specs.unwrap_or_else(|| vec![("id".to_string(), Type::Serial)])
        })
        .or_not()
        .map(Option::unwrap_or_default)
}

/// Identifier: a letter or underscore followed by letters,
/// digits, or underscores. Returned as an owned `String` so the
/// `Command` AST has no lifetime tying it to the input.
fn identifier<'a>()
-> impl Parser<'a, &'a str, String, extra::Err<Rich<'a, char>>> + Clone {
    any()
        .filter(|c: &char| c.is_ascii_alphabetic() || *c == '_')
        .then(
            any()
                .filter(|c: &char| c.is_ascii_alphanumeric() || *c == '_')
                .repeated()
                .collect::<Vec<_>>(),
        )
        .map(|(first, rest)| {
            let mut s = String::with_capacity(rest.len() + 1);
            s.push(first);
            s.extend(rest);
            s
        })
        .padded()
}

/// One of the supported type keywords, mapped to `Type`. The
/// `try_map` yields a `Custom` Rich error on unknown input,
/// which carries the friendly "unknown type 'X' (expected one
/// of: ...)" message — surfaced via `humanise()`. Note: no
/// `.labelled` here, because that would replace the custom
/// message with a generic "expected type".
fn type_keyword<'a>()
-> impl Parser<'a, &'a str, Type, extra::Err<Rich<'a, char>>> + Clone {
    let alphabetic = any()
        .filter(|c: &char| c.is_ascii_alphabetic())
        .repeated()
        .at_least(1)
        .collect::<String>();
    alphabetic.padded().try_map(|word, span| {
        word.parse::<Type>()
            .map_err(|e| Rich::custom(span, e.to_string()))
    })
}

/// `keyword_ci(kw).or_not()` packaged for readability.
fn optional_keyword<'a>(
    kw: &'static str,
) -> impl Parser<'a, &'a str, (), extra::Err<Rich<'a, char>>> + Clone {
    keyword_ci(kw).or_not().map(|_| ())
}

/// Case-insensitive keyword matcher. Consumes leading and
/// trailing whitespace and, importantly, requires a word
/// boundary so `create` does not match a prefix of `created`.
fn keyword_ci<'a>(
    kw: &'static str,
) -> impl Parser<'a, &'a str, (), extra::Err<Rich<'a, char>>> + Clone {
    let alphabetic = any()
        .filter(|c: &char| c.is_ascii_alphabetic())
        .repeated()
        .at_least(1)
        .collect::<String>();
    alphabetic.padded().try_map(move |word, span| {
        if word.eq_ignore_ascii_case(kw) {
            Ok(())
        } else {
            Err(Rich::custom(
                span,
                format!("expected '{kw}', found '{word}'"),
            ))
        }
    })
}

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

    fn ok(input: &str) -> Command {
        parse_command(input).unwrap_or_else(|e| panic!("expected ok for {input:?}, got {e:?}"))
    }

    fn err(input: &str) -> ParseError {
        parse_command(input).expect_err("expected parse error")
    }

    fn col(name: &str, ty: Type) -> ColumnSpec {
        ColumnSpec {
            name: name.to_string(),
            ty,
        }
    }

    #[test]
    fn bare_create_table_errors_with_helpful_message() {
        let e = err("create table Customers");
        match e {
            ParseError::Invalid { message, .. } => {
                assert!(
                    message.contains("with pk"),
                    "error should mention `with pk`:\n{message}"
                );
            }
            ParseError::Empty => panic!("unexpected empty error"),
        }
    }

    #[test]
    fn create_table_with_pk_default_is_id_serial() {
        assert_eq!(
            ok("create table Customers with pk"),
            Command::CreateTable {
                name: "Customers".to_string(),
                columns: vec![col("id", Type::Serial)],
                primary_key: vec!["id".to_string()],
            }
        );
    }

    #[test]
    fn create_table_with_named_typed_pk() {
        assert_eq!(
            ok("create table Customers with pk email:text"),
            Command::CreateTable {
                name: "Customers".to_string(),
                columns: vec![col("email", Type::Text)],
                primary_key: vec!["email".to_string()],
            }
        );
    }

    #[test]
    fn create_table_with_compound_pk() {
        assert_eq!(
            ok("create table OrderLines with pk order_id:int,product_id:int"),
            Command::CreateTable {
                name: "OrderLines".to_string(),
                columns: vec![
                    col("order_id", Type::Int),
                    col("product_id", Type::Int),
                ],
                primary_key: vec!["order_id".to_string(), "product_id".to_string()],
            }
        );
    }

    #[test]
    fn create_table_pk_accepts_any_user_type() {
        // Pedagogical freedom — the grammar imposes no
        // "sensible PK type" filter. Every user-facing type is
        // accepted; learners discover for themselves.
        for ty in Type::all() {
            let input = format!("create table T with pk col:{}", ty.keyword());
            let cmd = ok(&input);
            if let Command::CreateTable {
                columns,
                primary_key,
                ..
            } = cmd
            {
                assert_eq!(columns[0].ty, *ty);
                assert_eq!(primary_key, vec!["col".to_string()]);
            } else {
                panic!("expected CreateTable for {input}");
            }
        }
    }

    #[test]
    fn create_table_pk_tolerates_whitespace() {
        assert_eq!(
            ok("create table T  with  pk   id : serial"),
            Command::CreateTable {
                name: "T".to_string(),
                columns: vec![col("id", Type::Serial)],
                primary_key: vec!["id".to_string()],
            }
        );
        assert_eq!(
            ok("create table T with pk a : int , b : int"),
            Command::CreateTable {
                name: "T".to_string(),
                columns: vec![col("a", Type::Int), col("b", Type::Int)],
                primary_key: vec!["a".to_string(), "b".to_string()],
            }
        );
    }

    #[test]
    fn create_table_keywords_are_case_insensitive() {
        assert_eq!(
            ok("CREATE TABLE Customers WITH PK email:TEXT"),
            Command::CreateTable {
                name: "Customers".to_string(),
                columns: vec![col("email", Type::Text)],
                primary_key: vec!["email".to_string()],
            }
        );
    }

    // --- drop column / rename column / change column ---

    #[test]
    fn drop_column_simple() {
        assert_eq!(
            ok("drop column from table Customers: Email"),
            Command::DropColumn {
                table: "Customers".to_string(),
                column: "Email".to_string(),
            }
        );
    }

    #[test]
    fn drop_column_accepts_bare_identifiers() {
        // Both prepositions independently optional, matching
        // `add column`'s shape.
        assert_eq!(
            ok("drop column Customers: Email"),
            Command::DropColumn {
                table: "Customers".to_string(),
                column: "Email".to_string(),
            }
        );
        assert_eq!(
            ok("drop column from Customers: Email"),
            Command::DropColumn {
                table: "Customers".to_string(),
                column: "Email".to_string(),
            }
        );
        assert_eq!(
            ok("drop column table Customers: Email"),
            Command::DropColumn {
                table: "Customers".to_string(),
                column: "Email".to_string(),
            }
        );
    }

    #[test]
    fn rename_column_simple() {
        assert_eq!(
            ok("rename column in table Customers: OldName to NewName"),
            Command::RenameColumn {
                table: "Customers".to_string(),
                old: "OldName".to_string(),
                new: "NewName".to_string(),
            }
        );
    }

    #[test]
    fn rename_column_accepts_bare_identifiers() {
        assert_eq!(
            ok("rename column Customers: A to B"),
            Command::RenameColumn {
                table: "Customers".to_string(),
                old: "A".to_string(),
                new: "B".to_string(),
            }
        );
        assert_eq!(
            ok("rename column in Customers: A to B"),
            Command::RenameColumn {
                table: "Customers".to_string(),
                old: "A".to_string(),
                new: "B".to_string(),
            }
        );
        assert_eq!(
            ok("rename column table Customers: A to B"),
            Command::RenameColumn {
                table: "Customers".to_string(),
                old: "A".to_string(),
                new: "B".to_string(),
            }
        );
    }

    #[test]
    fn change_column_simple() {
        assert_eq!(
            ok("change column in table Customers: Score (int)"),
            Command::ChangeColumnType {
                table: "Customers".to_string(),
                column: "Score".to_string(),
                ty: Type::Int,
                mode: ChangeColumnMode::Default,
            }
        );
    }

    #[test]
    fn change_column_accepts_bare_identifiers() {
        assert_eq!(
            ok("change column Customers: Score (real)"),
            Command::ChangeColumnType {
                table: "Customers".to_string(),
                column: "Score".to_string(),
                ty: Type::Real,
                mode: ChangeColumnMode::Default,
            }
        );
    }

    #[test]
    fn change_column_keywords_are_case_insensitive() {
        assert_eq!(
            ok("CHANGE COLUMN IN TABLE Customers: Score (TEXT)"),
            Command::ChangeColumnType {
                table: "Customers".to_string(),
                column: "Score".to_string(),
                ty: Type::Text,
                mode: ChangeColumnMode::Default,
            }
        );
    }

    #[test]
    fn change_column_with_force_conversion_flag() {
        assert_eq!(
            ok("change column Customers: Score (int) --force-conversion"),
            Command::ChangeColumnType {
                table: "Customers".to_string(),
                column: "Score".to_string(),
                ty: Type::Int,
                mode: ChangeColumnMode::ForceConversion,
            }
        );
    }

    #[test]
    fn change_column_with_dont_convert_flag() {
        assert_eq!(
            ok("change column Customers: Score (int) --dont-convert"),
            Command::ChangeColumnType {
                table: "Customers".to_string(),
                column: "Score".to_string(),
                ty: Type::Int,
                mode: ChangeColumnMode::DontConvert,
            }
        );
    }

    #[test]
    fn change_column_rejects_both_flags() {
        let e = err(
            "change column Customers: Score (int) --force-conversion --dont-convert",
        );
        match e {
            ParseError::Invalid { message, .. } => {
                assert!(
                    message.contains("--force-conversion")
                        && message.contains("--dont-convert"),
                    "expected both flag names in error: {message}"
                );
                assert!(
                    message.contains("mutually exclusive") || message.contains("pick one"),
                    "{message}"
                );
            }
            ParseError::Empty => panic!("unexpected empty error"),
        }
    }

    #[test]
    fn change_column_rejects_both_flags_in_either_order() {
        // Both orderings — and same-flag-twice — should reject
        // with a uniform "pick one" signal.
        let e = err("change column T: c (int) --dont-convert --force-conversion");
        match e {
            ParseError::Invalid { message, .. } => {
                assert!(message.contains("mutually exclusive"), "{message}");
            }
            ParseError::Empty => panic!("unexpected empty error"),
        }
    }

    #[test]
    fn drop_table_simple() {
        assert_eq!(
            ok("drop table Customers"),
            Command::DropTable {
                name: "Customers".to_string()
            }
        );
    }

    #[test]
    fn add_column_simple() {
        assert_eq!(
            ok("add column to table Customers: Name (text)"),
            Command::AddColumn {
                table: "Customers".to_string(),
                column: "Name".to_string(),
                ty: Type::Text,
            }
        );
    }

    #[test]
    fn add_column_with_each_supported_type() {
        for ty in Type::all() {
            let input = format!("add column to table T: C ({})", ty.keyword());
            assert_eq!(
                ok(&input),
                Command::AddColumn {
                    table: "T".to_string(),
                    column: "C".to_string(),
                    ty: *ty,
                }
            );
        }
    }

    #[test]
    fn add_column_accepts_bare_table_name() {
        // `to table` are both optional; bare table identifier
        // is accepted in this unambiguous position.
        assert_eq!(
            ok("add column Customers: Name (text)"),
            Command::AddColumn {
                table: "Customers".to_string(),
                column: "Name".to_string(),
                ty: Type::Text,
            }
        );
    }

    #[test]
    fn add_column_accepts_to_alone() {
        // `to` without `table`.
        assert_eq!(
            ok("add column to Customers: Name (text)"),
            Command::AddColumn {
                table: "Customers".to_string(),
                column: "Name".to_string(),
                ty: Type::Text,
            }
        );
    }

    #[test]
    fn add_column_accepts_table_alone() {
        // `table` without `to`.
        assert_eq!(
            ok("add column table Customers: Name (text)"),
            Command::AddColumn {
                table: "Customers".to_string(),
                column: "Name".to_string(),
                ty: Type::Text,
            }
        );
    }

    #[test]
    fn add_column_tolerates_whitespace_around_punctuation() {
        assert_eq!(
            ok("add column to table T:Name(text)"),
            Command::AddColumn {
                table: "T".to_string(),
                column: "Name".to_string(),
                ty: Type::Text,
            }
        );
    }

    #[test]
    fn empty_input_is_an_explicit_empty_error() {
        assert_eq!(parse_command(""), Err(ParseError::Empty));
        assert_eq!(parse_command("   "), Err(ParseError::Empty));
    }

    #[test]
    fn unknown_command_errors() {
        let e = err("frobulate Customers");
        assert!(matches!(e, ParseError::Invalid { .. }), "got {e:?}");
    }

    #[test]
    fn unknown_type_errors_with_alternatives_listed() {
        let e = err("add column to table T: Name (varchar)");
        match e {
            ParseError::Invalid { message, .. } => {
                assert!(
                    message.contains("varchar"),
                    "error should mention the bad type: {message}"
                );
                assert!(
                    message.contains("expected one of"),
                    "error should list valid alternatives: {message}"
                );
                assert!(
                    message.contains("text") && message.contains("shortid"),
                    "error should name the alternatives: {message}"
                );
            }
            ParseError::Empty => panic!("unexpected empty error"),
        }
    }

    #[test]
    fn unknown_pk_type_errors_with_alternatives_listed() {
        let e = err("create table T with pk id:varchar");
        match e {
            ParseError::Invalid { message, .. } => {
                assert!(message.contains("varchar"), "{message}");
                assert!(message.contains("expected one of"), "{message}");
            }
            ParseError::Empty => panic!("unexpected empty error"),
        }
    }

    #[test]
    fn trailing_garbage_errors() {
        let e = err("create table Customers with pk and pickles");
        assert!(matches!(e, ParseError::Invalid { .. }), "got {e:?}");
    }

    #[test]
    fn identifier_must_start_with_letter_or_underscore() {
        let e = err("create table 1Customers with pk");
        assert!(matches!(e, ParseError::Invalid { .. }), "got {e:?}");
    }

    fn rel(
        name: Option<&str>,
        parent: (&str, &str),
        child: (&str, &str),
        on_delete: ReferentialAction,
        on_update: ReferentialAction,
        create_fk: bool,
    ) -> Command {
        Command::AddRelationship {
            name: name.map(String::from),
            parent_table: parent.0.to_string(),
            parent_column: parent.1.to_string(),
            child_table: child.0.to_string(),
            child_column: child.1.to_string(),
            on_delete,
            on_update,
            create_fk,
        }
    }

    #[test]
    fn add_relationship_minimal() {
        assert_eq!(
            ok("add 1:n relationship from Customers.Id to Orders.CustId"),
            rel(
                None,
                ("Customers", "Id"),
                ("Orders", "CustId"),
                ReferentialAction::NoAction,
                ReferentialAction::NoAction,
                false,
            )
        );
    }

    #[test]
    fn add_relationship_with_name() {
        assert_eq!(
            ok("add 1:n relationship as cust_orders from Customers.Id to Orders.CustId"),
            rel(
                Some("cust_orders"),
                ("Customers", "Id"),
                ("Orders", "CustId"),
                ReferentialAction::NoAction,
                ReferentialAction::NoAction,
                false,
            )
        );
    }

    #[test]
    fn add_relationship_with_on_delete() {
        assert_eq!(
            ok("add 1:n relationship from Customers.Id to Orders.CustId on delete cascade"),
            rel(
                None,
                ("Customers", "Id"),
                ("Orders", "CustId"),
                ReferentialAction::Cascade,
                ReferentialAction::NoAction,
                false,
            )
        );
    }

    #[test]
    fn add_relationship_with_on_delete_set_null() {
        assert_eq!(
            ok("add 1:n relationship from Customers.Id to Orders.CustId on delete set null"),
            rel(
                None,
                ("Customers", "Id"),
                ("Orders", "CustId"),
                ReferentialAction::SetNull,
                ReferentialAction::NoAction,
                false,
            )
        );
    }

    #[test]
    fn add_relationship_with_both_actions_in_either_order() {
        let expected = rel(
            None,
            ("Customers", "Id"),
            ("Orders", "CustId"),
            ReferentialAction::Cascade,
            ReferentialAction::SetNull,
            false,
        );
        assert_eq!(
            ok("add 1:n relationship from Customers.Id to Orders.CustId on delete cascade on update set null"),
            expected
        );
        assert_eq!(
            ok("add 1:n relationship from Customers.Id to Orders.CustId on update set null on delete cascade"),
            expected
        );
    }

    #[test]
    fn add_relationship_repeated_clause_errors() {
        let e = err(
            "add 1:n relationship from C.id to O.cid on delete cascade on delete restrict",
        );
        match e {
            ParseError::Invalid { message, .. } => {
                assert!(message.contains("specified twice"), "{message}");
            }
            ParseError::Empty => panic!("unexpected empty error"),
        }
    }

    #[test]
    fn add_relationship_with_create_fk_flag() {
        assert_eq!(
            ok("add 1:n relationship from Customers.Id to Orders.CustId --create-fk"),
            rel(
                None,
                ("Customers", "Id"),
                ("Orders", "CustId"),
                ReferentialAction::NoAction,
                ReferentialAction::NoAction,
                true,
            )
        );
    }

    #[test]
    fn add_relationship_with_name_actions_and_flag() {
        assert_eq!(
            ok("add 1:n relationship as cust_orders from Customers.Id to Orders.CustId on delete cascade on update no action --create-fk"),
            rel(
                Some("cust_orders"),
                ("Customers", "Id"),
                ("Orders", "CustId"),
                ReferentialAction::Cascade,
                ReferentialAction::NoAction,
                true,
            )
        );
    }

    #[test]
    fn add_relationship_keywords_are_case_insensitive() {
        assert_eq!(
            ok("ADD 1:N RELATIONSHIP FROM Customers.Id TO Orders.CustId ON DELETE CASCADE"),
            rel(
                None,
                ("Customers", "Id"),
                ("Orders", "CustId"),
                ReferentialAction::Cascade,
                ReferentialAction::NoAction,
                false,
            )
        );
    }

    #[test]
    fn add_relationship_unknown_action_errors() {
        let e = err("add 1:n relationship from C.id to O.cid on delete obliterate");
        assert!(matches!(e, ParseError::Invalid { .. }), "got {e:?}");
    }

    #[test]
    fn insert_with_explicit_column_list() {
        assert_eq!(
            ok("insert into Customers (Name, Email) values ('Alice', 'a@b.com')"),
            Command::Insert {
                table: "Customers".to_string(),
                columns: Some(vec!["Name".to_string(), "Email".to_string()]),
                values: vec![
                    Value::Text("Alice".to_string()),
                    Value::Text("a@b.com".to_string()),
                ],
            }
        );
    }

    #[test]
    fn insert_short_form_omitting_values_keyword() {
        // User typed `insert into T (vals)` without `values`.
        // Equivalent to `insert into T values (vals)`.
        assert_eq!(
            ok("insert into Customers ('Alice')"),
            Command::Insert {
                table: "Customers".to_string(),
                columns: None,
                values: vec![Value::Text("Alice".to_string())],
            }
        );
    }

    #[test]
    fn insert_short_form_without_column_list() {
        assert_eq!(
            ok("insert into Customers values ('Alice', 'a@b.com')"),
            Command::Insert {
                table: "Customers".to_string(),
                columns: None,
                values: vec![
                    Value::Text("Alice".to_string()),
                    Value::Text("a@b.com".to_string()),
                ],
            }
        );
    }

    #[test]
    fn insert_accepts_mixed_value_kinds() {
        assert_eq!(
            ok("insert into T values (1, 3.14, 'hi', true, null)"),
            Command::Insert {
                table: "T".to_string(),
                columns: None,
                values: vec![
                    Value::Number("1".to_string()),
                    Value::Number("3.14".to_string()),
                    Value::Text("hi".to_string()),
                    Value::Bool(true),
                    Value::Null,
                ],
            }
        );
    }

    #[test]
    fn insert_supports_negative_numbers() {
        assert_eq!(
            ok("insert into T values (-5, -3.14)"),
            Command::Insert {
                table: "T".to_string(),
                columns: None,
                values: vec![
                    Value::Number("-5".to_string()),
                    Value::Number("-3.14".to_string()),
                ],
            }
        );
    }

    #[test]
    fn string_literal_supports_escaped_single_quote() {
        // SQL convention: '' inside a quoted string is a literal '.
        assert_eq!(
            ok("insert into T values ('don''t panic')"),
            Command::Insert {
                table: "T".to_string(),
                columns: None,
                values: vec![Value::Text("don't panic".to_string())],
            }
        );
    }

    #[test]
    fn update_with_where() {
        assert_eq!(
            ok("update Customers set Name='Alice' where id=1"),
            Command::Update {
                table: "Customers".to_string(),
                assignments: vec![("Name".to_string(), Value::Text("Alice".to_string()))],
                filter: RowFilter::Where {
                    column: "id".to_string(),
                    value: Value::Number("1".to_string()),
                },
            }
        );
    }

    #[test]
    fn update_with_multiple_assignments() {
        assert_eq!(
            ok("update Customers set Name='Alice', Email='a@b.com' where id=1"),
            Command::Update {
                table: "Customers".to_string(),
                assignments: vec![
                    ("Name".to_string(), Value::Text("Alice".to_string())),
                    ("Email".to_string(), Value::Text("a@b.com".to_string())),
                ],
                filter: RowFilter::Where {
                    column: "id".to_string(),
                    value: Value::Number("1".to_string()),
                },
            }
        );
    }

    #[test]
    fn update_with_all_rows_flag() {
        assert_eq!(
            ok("update Customers set Active=false --all-rows"),
            Command::Update {
                table: "Customers".to_string(),
                assignments: vec![("Active".to_string(), Value::Bool(false))],
                filter: RowFilter::AllRows,
            }
        );
    }

    #[test]
    fn update_without_where_or_flag_errors() {
        let e = err("update Customers set Active=false");
        assert!(matches!(e, ParseError::Invalid { .. }), "got {e:?}");
    }

    #[test]
    fn delete_with_where() {
        assert_eq!(
            ok("delete from Customers where id=1"),
            Command::Delete {
                table: "Customers".to_string(),
                filter: RowFilter::Where {
                    column: "id".to_string(),
                    value: Value::Number("1".to_string()),
                },
            }
        );
    }

    #[test]
    fn delete_with_all_rows_flag() {
        assert_eq!(
            ok("delete from Customers --all-rows"),
            Command::Delete {
                table: "Customers".to_string(),
                filter: RowFilter::AllRows,
            }
        );
    }

    #[test]
    fn delete_without_where_or_flag_errors() {
        let e = err("delete from Customers");
        assert!(matches!(e, ParseError::Invalid { .. }), "got {e:?}");
    }

    #[test]
    fn show_data_command() {
        assert_eq!(
            ok("show data Customers"),
            Command::ShowData {
                name: "Customers".to_string()
            }
        );
    }

    #[test]
    fn drop_relationship_by_name() {
        assert_eq!(
            ok("drop relationship cust_orders"),
            Command::DropRelationship {
                selector: RelationshipSelector::Named {
                    name: "cust_orders".to_string()
                }
            }
        );
    }

    #[test]
    fn show_table_simple() {
        assert_eq!(
            ok("show table Customers"),
            Command::ShowTable {
                name: "Customers".to_string()
            }
        );
    }

    #[test]
    fn drop_relationship_by_endpoints() {
        assert_eq!(
            ok("drop relationship from Customers.Id to Orders.CustId"),
            Command::DropRelationship {
                selector: RelationshipSelector::Endpoints {
                    parent_table: "Customers".to_string(),
                    parent_column: "Id".to_string(),
                    child_table: "Orders".to_string(),
                    child_column: "CustId".to_string(),
                }
            }
        );
    }

    #[test]
    fn identifier_allows_underscores_and_digits_after_start() {
        assert_eq!(
            ok("create table customer_v2 with pk"),
            Command::CreateTable {
                name: "customer_v2".to_string(),
                columns: vec![col("id", Type::Serial)],
                primary_key: vec!["id".to_string()],
            }
        );
    }
}