//! DDL command nodes (ADR-0024 §migration Phase B).
//!
//! Five commands at four entry words: `drop` (drop table /
//! drop column / drop relationship), `add` (add column /
//! add 1:n relationship), `rename` (rename column), `change`
//! (change column). The chumsky-side declarations stay
//! reachable for any input the walker doesn't engage on, but
//! for these entry words the walker is authoritative.
//!
//! Each shape is laid out inline so per-use-site `role`
//! annotations carry meaning end-to-end (e.g.,
//! `parent_table` vs `child_table` for the endpoints clause).

use crate::dsl::action::ReferentialAction;
use crate::dsl::command::{
    ChangeColumnMode, ColumnSpec, Command, IndexSelector, RelationshipSelector,
};
use crate::dsl::grammar::{
    CommandNode, HintMode, IdentSource, Node, ValidationError, Word,
    shared::{REFERENTIAL_CLAUSES, TYPE_SLOT, TYPE_VALIDATOR},
};

/// `HintMode` annotation shared by every `NewName` ident slot:
/// the user is inventing a name, so the hint panel forces the
/// "Type a name [then …]" prose rather than offering schema
/// candidates (ADR-0024 §HintMode-per-node).
const NEW_NAME_HINT: HintMode = HintMode::ForceProse("hint.ambient_typing_name");
use crate::dsl::types::Type;
use crate::dsl::walker::outcome::{MatchedKind, MatchedPath};

// =================================================================
// Building blocks
// =================================================================

const TABLE_NAME_NEW_IDENT: Node = Node::Ident {
    source: IdentSource::NewName,
    role: "table_name",
    validator: None,
    highlight_override: None,
        writes_table: false,
        writes_column: false,
        writes_user_listed_column: false,
};
const TABLE_NAME_NEW: Node = Node::Hinted {
    mode: NEW_NAME_HINT,
    inner: &TABLE_NAME_NEW_IDENT,
};

// `writes_table: true` so that the column-name slots that
// follow the table name in `drop column` / `rename column` /
// `change column` / `add column` can narrow their candidates to
// this table's columns (handoff-12 §2.2). The walker writes
// `current_table` / `current_table_columns` on match; the
// completion engine reads the snapshot. `drop table` has no
// downstream column slot, so the write is harmless there.
const TABLE_NAME_EXISTING: Node = Node::Ident {
    source: IdentSource::Tables,
    role: "table_name",
    validator: None,
    highlight_override: None,
        writes_table: true,
        writes_column: false,
        writes_user_listed_column: false,
};

const COLUMN_NAME: Node = Node::Ident {
    source: IdentSource::Columns,
    role: "column_name",
    validator: None,
    highlight_override: None,
        writes_table: false,
        writes_column: false,
        writes_user_listed_column: false,
};

const COLUMN_NAME_NEW_IDENT: Node = Node::Ident {
    source: IdentSource::NewName,
    role: "column_name",
    validator: None,
    highlight_override: None,
        writes_table: false,
        writes_column: false,
        writes_user_listed_column: false,
};
const COLUMN_NAME_NEW: Node = Node::Hinted {
    mode: NEW_NAME_HINT,
    inner: &COLUMN_NAME_NEW_IDENT,
};

const RELATIONSHIP_NAME: Node = Node::Ident {
    source: IdentSource::Relationships,
    role: "relationship_name",
    validator: None,
    highlight_override: None,
        writes_table: false,
        writes_column: false,
        writes_user_listed_column: false,
};

const RELATIONSHIP_NAME_NEW_IDENT: Node = Node::Ident {
    source: IdentSource::NewName,
    role: "relationship_name",
    validator: None,
    highlight_override: None,
        writes_table: false,
        writes_column: false,
        writes_user_listed_column: false,
};
const RELATIONSHIP_NAME_NEW: Node = Node::Hinted {
    mode: NEW_NAME_HINT,
    inner: &RELATIONSHIP_NAME_NEW_IDENT,
};

const INDEX_NAME_EXISTING: Node = Node::Ident {
    source: IdentSource::Indexes,
    role: "index_name",
    validator: None,
    highlight_override: None,
    writes_table: false,
    writes_column: false,
    writes_user_listed_column: false,
};

const INDEX_NAME_NEW_IDENT: Node = Node::Ident {
    source: IdentSource::NewName,
    role: "index_name",
    validator: None,
    highlight_override: None,
    writes_table: false,
    writes_column: false,
    writes_user_listed_column: false,
};
const INDEX_NAME_NEW: Node = Node::Hinted {
    mode: NEW_NAME_HINT,
    inner: &INDEX_NAME_NEW_IDENT,
};

// The column list shared by `add index` / `drop index`: one or
// more existing column names, comma-separated, inside parens.
// `COLUMN_NAME` narrows to the `on <Table>` table's columns
// because that ident carries `writes_table: true`.
const INDEX_COLUMN_LIST: Node = Node::Repeated {
    inner: &COLUMN_NAME,
    separator: Some(&Node::Punct(',')),
    min: 1,
};

// `[to]` and `[table]` connectives.
const TO_OPT: Node = Node::Optional(&Node::Word(Word::keyword("to")));
const FROM_OPT: Node = Node::Optional(&Node::Word(Word::keyword("from")));
const IN_OPT: Node = Node::Optional(&Node::Word(Word::keyword("in")));
const TABLE_OPT: Node = Node::Optional(&Node::Word(Word::keyword("table")));

// =================================================================
// drop_table — `drop table <T>`
// =================================================================

const DROP_TABLE_NODES: &[Node] = &[
    Node::Word(Word::keyword("table")),
    TABLE_NAME_EXISTING,
];
const DROP_TABLE: Node = Node::Seq(DROP_TABLE_NODES);

// =================================================================
// drop_column — `drop column [from] [table] <T> : <col>`
// =================================================================

// `--cascade` (ADR-0025): opt-in to dropping any index that
// covers the column alongside the column itself. Without it, a
// covered column is refused with a friendly error.
const DROP_COLUMN_CASCADE_OPT: Node = Node::Optional(&Node::Flag("cascade"));

const DROP_COLUMN_NODES: &[Node] = &[
    Node::Word(Word::keyword("column")),
    FROM_OPT,
    TABLE_OPT,
    TABLE_NAME_EXISTING,
    Node::Punct(':'),
    COLUMN_NAME,
    DROP_COLUMN_CASCADE_OPT,
];
const DROP_COLUMN: Node = Node::Seq(DROP_COLUMN_NODES);

// =================================================================
// drop_relationship — `drop relationship (endpoints | name)`
// =================================================================

// `writes_table: true` on each endpoint's table ident so the
// `.<col>` slot that follows narrows to that table's columns
// (handoff-13 §2.2 follow-up). The two endpoints are walked
// sequentially, so `current_table` is correctly the parent
// table while walking `parent_column` and the child table
// while walking `child_column`.
const DR_PARENT_NODES: &[Node] = &[
    Node::Ident {
        source: IdentSource::Tables,
        role: "parent_table",
        validator: None,
        highlight_override: None,
        writes_table: true,
        writes_column: false,
        writes_user_listed_column: false,
    },
    Node::Punct('.'),
    Node::Ident {
        source: IdentSource::Columns,
        role: "parent_column",
        validator: None,
        highlight_override: None,
        writes_table: false,
        writes_column: false,
        writes_user_listed_column: false,
    },
];
const DR_PARENT: Node = Node::Seq(DR_PARENT_NODES);

const DR_CHILD_NODES: &[Node] = &[
    Node::Ident {
        source: IdentSource::Tables,
        role: "child_table",
        validator: None,
        highlight_override: None,
        writes_table: true,
        writes_column: false,
        writes_user_listed_column: false,
    },
    Node::Punct('.'),
    Node::Ident {
        source: IdentSource::Columns,
        role: "child_column",
        validator: None,
        highlight_override: None,
        writes_table: false,
        writes_column: false,
        writes_user_listed_column: false,
    },
];
const DR_CHILD: Node = Node::Seq(DR_CHILD_NODES);

const DR_ENDPOINTS_NODES: &[Node] = &[
    Node::Word(Word::keyword("from")),
    DR_PARENT,
    Node::Word(Word::keyword("to")),
    DR_CHILD,
];
const DR_ENDPOINTS: Node = Node::Seq(DR_ENDPOINTS_NODES);

const DR_SELECTOR_CHOICES: &[Node] = &[DR_ENDPOINTS, RELATIONSHIP_NAME];
const DR_SELECTOR: Node = Node::Choice(DR_SELECTOR_CHOICES);

const DROP_RELATIONSHIP_NODES: &[Node] = &[
    Node::Word(Word::keyword("relationship")),
    DR_SELECTOR,
];
const DROP_RELATIONSHIP: Node = Node::Seq(DROP_RELATIONSHIP_NODES);

// =================================================================
// drop_index — `drop index (<name> | on <T> (<col>, …))`
// =================================================================

const DI_POSITIONAL_NODES: &[Node] = &[
    Node::Word(Word::keyword("on")),
    TABLE_NAME_EXISTING,
    Node::Punct('('),
    INDEX_COLUMN_LIST,
    Node::Punct(')'),
];
const DI_POSITIONAL: Node = Node::Seq(DI_POSITIONAL_NODES);

// Positional form first — it opens with the `on` keyword, so a
// bare index name can't be mistaken for it (mirrors DR_SELECTOR).
const DI_SELECTOR_CHOICES: &[Node] = &[DI_POSITIONAL, INDEX_NAME_EXISTING];
const DI_SELECTOR: Node = Node::Choice(DI_SELECTOR_CHOICES);

const DROP_INDEX_NODES: &[Node] = &[
    Node::Word(Word::keyword("index")),
    DI_SELECTOR,
];
const DROP_INDEX: Node = Node::Seq(DROP_INDEX_NODES);

// =================================================================
// drop entry — `drop (table|column|relationship|index) ...`
// =================================================================

const DROP_CHOICES: &[Node] = &[DROP_COLUMN, DROP_RELATIONSHIP, DROP_TABLE, DROP_INDEX];
const DROP_SHAPE: Node = Node::Choice(DROP_CHOICES);

// =================================================================
// add_column — `add column [to] [table] <T> : <col> ( <type> )`
// =================================================================

const ADD_COLUMN_NODES: &[Node] = &[
    Node::Word(Word::keyword("column")),
    TO_OPT,
    TABLE_OPT,
    TABLE_NAME_EXISTING,
    Node::Punct(':'),
    COLUMN_NAME_NEW,
    Node::Punct('('),
    TYPE_SLOT,
    Node::Punct(')'),
];
const ADD_COLUMN: Node = Node::Seq(ADD_COLUMN_NODES);

// =================================================================
// add_relationship — `add 1:n relationship [as <name>]
//                    from <T>.<c> to <T>.<c>
//                    [on delete <a>] [on update <a>]
//                    [--create-fk]`
// =================================================================

// `writes_table: true` on each endpoint's table ident so the
// `.<col>` slot narrows to that table's columns (handoff-13
// §2.2 follow-up — mirrors DR_PARENT / DR_CHILD).
const AR_PARENT_NODES: &[Node] = &[
    Node::Ident {
        source: IdentSource::Tables,
        role: "parent_table",
        validator: None,
        highlight_override: None,
        writes_table: true,
        writes_column: false,
        writes_user_listed_column: false,
    },
    Node::Punct('.'),
    Node::Ident {
        source: IdentSource::Columns,
        role: "parent_column",
        validator: None,
        highlight_override: None,
        writes_table: false,
        writes_column: false,
        writes_user_listed_column: false,
    },
];
const AR_PARENT: Node = Node::Seq(AR_PARENT_NODES);

const AR_CHILD_NODES: &[Node] = &[
    Node::Ident {
        source: IdentSource::Tables,
        role: "child_table",
        validator: None,
        highlight_override: None,
        writes_table: true,
        writes_column: false,
        writes_user_listed_column: false,
    },
    Node::Punct('.'),
    Node::Ident {
        source: IdentSource::Columns,
        role: "child_column",
        validator: None,
        highlight_override: None,
        writes_table: false,
        writes_column: false,
        writes_user_listed_column: false,
    },
];
const AR_CHILD: Node = Node::Seq(AR_CHILD_NODES);

const AR_AS_NAME_NODES: &[Node] = &[
    Node::Word(Word::keyword("as")),
    RELATIONSHIP_NAME_NEW,
];
const AR_AS_NAME_OPT: Node = Node::Optional(&Node::Seq(AR_AS_NAME_NODES));

const AR_CREATE_FK_OPT: Node = Node::Optional(&Node::Flag("create-fk"));

const ADD_RELATIONSHIP_NODES: &[Node] = &[
    Node::Literal("1"),
    Node::Punct(':'),
    Node::Word(Word::keyword("n")),
    Node::Word(Word::keyword("relationship")),
    AR_AS_NAME_OPT,
    Node::Word(Word::keyword("from")),
    AR_PARENT,
    Node::Word(Word::keyword("to")),
    AR_CHILD,
    REFERENTIAL_CLAUSES,
    AR_CREATE_FK_OPT,
];
const ADD_RELATIONSHIP: Node = Node::Seq(ADD_RELATIONSHIP_NODES);

// =================================================================
// add_index — `add index [as <name>] on <T> (<col>, …)`
// =================================================================

const AI_AS_NAME_NODES: &[Node] = &[
    Node::Word(Word::keyword("as")),
    INDEX_NAME_NEW,
];
const AI_AS_NAME_OPT: Node = Node::Optional(&Node::Seq(AI_AS_NAME_NODES));

const ADD_INDEX_NODES: &[Node] = &[
    Node::Word(Word::keyword("index")),
    AI_AS_NAME_OPT,
    Node::Word(Word::keyword("on")),
    TABLE_NAME_EXISTING,
    Node::Punct('('),
    INDEX_COLUMN_LIST,
    Node::Punct(')'),
];
const ADD_INDEX: Node = Node::Seq(ADD_INDEX_NODES);

// =================================================================
// add entry — `add (column|1:n relationship|index) …`
// =================================================================

const ADD_CHOICES: &[Node] = &[ADD_COLUMN, ADD_RELATIONSHIP, ADD_INDEX];
const ADD_SHAPE: Node = Node::Choice(ADD_CHOICES);

// =================================================================
// rename_column — `rename column [in] [table] <T> : <col> to <new>`
// =================================================================

const NEW_COLUMN_NAME_IDENT: Node = Node::Ident {
    source: IdentSource::NewName,
    role: "new_column_name",
    validator: None,
    highlight_override: None,
    writes_table: false,
    writes_column: false,
    writes_user_listed_column: false,
};
const NEW_COLUMN_NAME: Node = Node::Hinted {
    mode: NEW_NAME_HINT,
    inner: &NEW_COLUMN_NAME_IDENT,
};

const RENAME_COLUMN_NODES: &[Node] = &[
    Node::Word(Word::keyword("column")),
    IN_OPT,
    TABLE_OPT,
    TABLE_NAME_EXISTING,
    Node::Punct(':'),
    COLUMN_NAME,
    Node::Word(Word::keyword("to")),
    NEW_COLUMN_NAME,
];
const RENAME_COLUMN: Node = Node::Seq(RENAME_COLUMN_NODES);

// =================================================================
// change_column — `change column [in] [table] <T> : <col>
//                  ( <type> ) [--force-conversion | --dont-convert]`
// =================================================================

const CHANGE_FLAG_CHOICES: &[Node] = &[
    Node::Flag("force-conversion"),
    Node::Flag("dont-convert"),
];
const CHANGE_FLAG_OPT: Node = Node::Repeated {
    inner: &Node::Choice(CHANGE_FLAG_CHOICES),
    separator: None,
    min: 0,
};

const CHANGE_COLUMN_NODES: &[Node] = &[
    Node::Word(Word::keyword("column")),
    IN_OPT,
    TABLE_OPT,
    TABLE_NAME_EXISTING,
    Node::Punct(':'),
    COLUMN_NAME,
    Node::Punct('('),
    TYPE_SLOT,
    Node::Punct(')'),
    CHANGE_FLAG_OPT,
];
const CHANGE_COLUMN: Node = Node::Seq(CHANGE_COLUMN_NODES);

// =================================================================
// AST builders
// =================================================================

/// First ident whose role matches.
fn ident<'a>(path: &'a MatchedPath, role: &str) -> Option<&'a str> {
    path.items.iter().find_map(|i| match &i.kind {
        MatchedKind::Ident { role: r, .. } if *r == role => Some(i.text.as_str()),
        _ => None,
    })
}

fn require_ident(path: &MatchedPath, role: &'static str) -> Result<String, ValidationError> {
    ident(path, role)
        .map(str::to_string)
        .ok_or_else(|| ValidationError {
            message_key: "parse.error_wrapper",
            args: vec![("detail", format!("missing {role}"))],
        })
}

/// Every ident whose role matches, in matched (left-to-right)
/// order. Used by the column-list commands.
fn collect_idents(path: &MatchedPath, role: &str) -> Vec<String> {
    path.items
        .iter()
        .filter_map(|i| match &i.kind {
            MatchedKind::Ident { role: r, .. } if *r == role => Some(i.text.clone()),
            _ => None,
        })
        .collect()
}

fn parse_action(words: &[&'static str]) -> ReferentialAction {
    // `set null`, `no action`, `cascade`, `restrict`.
    if words.contains(&"set") && words.contains(&"null") {
        ReferentialAction::SetNull
    } else if words.contains(&"no") && words.contains(&"action") {
        ReferentialAction::NoAction
    } else if words.contains(&"cascade") {
        ReferentialAction::Cascade
    } else if words.contains(&"restrict") {
        ReferentialAction::Restrict
    } else {
        ReferentialAction::default_action()
    }
}

fn build_drop(path: &MatchedPath) -> Result<Command, ValidationError> {
    // Discriminate by the second word matched (the entry was
    // `drop`, the next Word is `table` / `column` / `relationship`).
    let sub = path
        .items
        .iter()
        .filter_map(|i| match &i.kind {
            MatchedKind::Word(w) => Some(*w),
            _ => None,
        })
        .nth(1);
    match sub {
        Some("table") => Ok(Command::DropTable {
            name: require_ident(path, "table_name")?,
        }),
        Some("column") => Ok(Command::DropColumn {
            table: require_ident(path, "table_name")?,
            column: require_ident(path, "column_name")?,
            cascade: path
                .items
                .iter()
                .any(|i| matches!(&i.kind, MatchedKind::Flag("cascade"))),
        }),
        Some("index") => {
            // Positional form has `on` as the third Word.
            let has_on = path
                .items
                .iter()
                .any(|i| matches!(&i.kind, MatchedKind::Word("on")));
            if has_on {
                Ok(Command::DropIndex {
                    selector: IndexSelector::Columns {
                        table: require_ident(path, "table_name")?,
                        columns: collect_idents(path, "column_name"),
                    },
                })
            } else {
                Ok(Command::DropIndex {
                    selector: IndexSelector::Named {
                        name: require_ident(path, "index_name")?,
                    },
                })
            }
        }
        Some("relationship") => {
            // Endpoints form has `from` as the third Word.
            let has_from = path
                .items
                .iter()
                .any(|i| matches!(&i.kind, MatchedKind::Word("from")));
            if has_from {
                Ok(Command::DropRelationship {
                    selector: RelationshipSelector::Endpoints {
                        parent_table: require_ident(path, "parent_table")?,
                        parent_column: require_ident(path, "parent_column")?,
                        child_table: require_ident(path, "child_table")?,
                        child_column: require_ident(path, "child_column")?,
                    },
                })
            } else {
                Ok(Command::DropRelationship {
                    selector: RelationshipSelector::Named {
                        name: require_ident(path, "relationship_name")?,
                    },
                })
            }
        }
        _ => Err(ValidationError {
            message_key: "parse.error_wrapper",
            args: vec![("detail", "unknown drop subcommand".to_string())],
        }),
    }
}

fn build_add(path: &MatchedPath) -> Result<Command, ValidationError> {
    // Second matched Word distinguishes column vs the `1:n
    // relationship` form. The `1` literal counts as a Word
    // (the walker records Literal matches as MatchedKind::Word
    // for AST-builder uniformity).
    let second_word = path
        .items
        .iter()
        .filter_map(|i| match &i.kind {
            MatchedKind::Word(w) => Some(*w),
            _ => None,
        })
        .nth(1);
    match second_word {
        Some("column") => {
            let ty_text = require_ident(path, "type")?;
            let ty = ty_text
                .parse::<crate::dsl::types::Type>()
                .map_err(|_| ValidationError {
                    message_key: "parse.error_wrapper",
                    args: vec![("detail", "unknown type".to_string())],
                })?;
            Ok(Command::AddColumn {
                table: require_ident(path, "table_name")?,
                column: require_ident(path, "column_name")?,
                ty,
                // Constraint suffix is wired in once the
                // constraint grammar lands (ADR-0029).
                not_null: false,
                unique: false,
                default: None,
                check: None,
            })
        }
        Some("1") => build_add_relationship(path),
        Some("index") => Ok(Command::AddIndex {
            name: ident(path, "index_name").map(str::to_string),
            table: require_ident(path, "table_name")?,
            columns: collect_idents(path, "column_name"),
        }),
        _ => Err(ValidationError {
            message_key: "parse.error_wrapper",
            args: vec![("detail", "unknown add subcommand".to_string())],
        }),
    }
}

fn build_add_relationship(path: &MatchedPath) -> Result<Command, ValidationError> {
    // Collect all referential-clause actions in matched order
    // and validate at-most-2 + not-repeated. The `on <delete|
    // update> <action>` sequence shows up as a run of Word
    // matches in the path between `to <child>` and end.
    //
    // Strategy: walk through Word items in order; whenever we
    // see `on`, the next Word is the target, then the action
    // word(s) follow until the next `on`, `--create-fk`, or
    // end-of-path.
    let words: Vec<&'static str> = path
        .items
        .iter()
        .filter_map(|i| match &i.kind {
            MatchedKind::Word(w) => Some(*w),
            _ => None,
        })
        .collect();

    let mut on_delete: Option<ReferentialAction> = None;
    let mut on_update: Option<ReferentialAction> = None;
    let mut i = 0;
    while i < words.len() {
        if words[i] == "on" && i + 1 < words.len() {
            let target = words[i + 1];
            // Action runs from i+2 until the next `on` or end.
            let action_start = i + 2;
            let mut action_end = action_start;
            while action_end < words.len() && words[action_end] != "on" {
                action_end += 1;
            }
            let action = parse_action(&words[action_start..action_end]);
            let slot = match target {
                "delete" => &mut on_delete,
                "update" => &mut on_update,
                _ => {
                    i = action_end;
                    continue;
                }
            };
            if slot.is_some() {
                return Err(ValidationError {
                    message_key: "parse.custom.on_action_specified_twice",
                    args: vec![("target", target.to_string())],
                });
            }
            *slot = Some(action);
            i = action_end;
        } else {
            i += 1;
        }
    }

    let create_fk = path
        .items
        .iter()
        .any(|i| matches!(&i.kind, MatchedKind::Flag("create-fk")));

    Ok(Command::AddRelationship {
        name: ident(path, "relationship_name").map(str::to_string),
        parent_table: require_ident(path, "parent_table")?,
        parent_column: require_ident(path, "parent_column")?,
        child_table: require_ident(path, "child_table")?,
        child_column: require_ident(path, "child_column")?,
        on_delete: on_delete.unwrap_or_else(ReferentialAction::default_action),
        on_update: on_update.unwrap_or_else(ReferentialAction::default_action),
        create_fk,
    })
}

fn build_rename_column(path: &MatchedPath) -> Result<Command, ValidationError> {
    Ok(Command::RenameColumn {
        table: require_ident(path, "table_name")?,
        old: require_ident(path, "column_name")?,
        new: require_ident(path, "new_column_name")?,
    })
}

fn build_change_column(path: &MatchedPath) -> Result<Command, ValidationError> {
    let ty_text = require_ident(path, "type")?;
    let ty = ty_text
        .parse::<crate::dsl::types::Type>()
        .map_err(|_| ValidationError {
            message_key: "parse.error_wrapper",
            args: vec![("detail", "unknown type".to_string())],
        })?;

    // Flags: at most one of --force-conversion / --dont-convert.
    let flags: Vec<&'static str> = path
        .items
        .iter()
        .filter_map(|i| match &i.kind {
            MatchedKind::Flag(n) => Some(*n),
            _ => None,
        })
        .collect();
    let mode = match flags.as_slice() {
        [] => ChangeColumnMode::Default,
        [one] => match *one {
            "force-conversion" => ChangeColumnMode::ForceConversion,
            "dont-convert" => ChangeColumnMode::DontConvert,
            _ => ChangeColumnMode::Default,
        },
        _ => {
            // Two or more flags — mutual exclusion fires
            // whether they're the same flag twice or both
            // mutually-exclusive flags appear. Wording mirrors
            // the chumsky parser's `change_column_flags_exclusive`.
            return Err(ValidationError {
                message_key: "parse.custom.change_column_flags_exclusive",
                args: vec![],
            });
        }
    };

    Ok(Command::ChangeColumnType {
        table: require_ident(path, "table_name")?,
        column: require_ident(path, "column_name")?,
        ty,
        mode,
    })
}

// =================================================================
// CommandNodes
// =================================================================

pub static DROP: CommandNode = CommandNode {
    entry: Word::keyword("drop"),
    shape: DROP_SHAPE,
    ast_builder: build_drop,
    help_id: Some("ddl.drop"),
    usage_ids: &[
        "parse.usage.drop_table",
        "parse.usage.drop_column",
        "parse.usage.drop_relationship",
        "parse.usage.drop_index",
    ],};

pub static ADD: CommandNode = CommandNode {
    entry: Word::keyword("add"),
    shape: ADD_SHAPE,
    ast_builder: build_add,
    help_id: Some("ddl.add"),
    usage_ids: &[
        "parse.usage.add_column",
        "parse.usage.add_relationship",
        "parse.usage.add_index",
    ],};

pub static RENAME: CommandNode = CommandNode {
    entry: Word::keyword("rename"),
    shape: RENAME_COLUMN,
    ast_builder: build_rename_column,
    help_id: Some("ddl.rename"),
    usage_ids: &["parse.usage.rename_column"],};

pub static CHANGE: CommandNode = CommandNode {
    entry: Word::keyword("change"),
    shape: CHANGE_COLUMN,
    ast_builder: build_change_column,
    help_id: Some("ddl.change"),
    usage_ids: &["parse.usage.change_column"],};

// =================================================================
// create_table — `create table <Name> [with pk [<col>(<type>)[, ...]]]`
//                                          (Phase C)
// =================================================================

const COL_NAME_IDENT: Node = Node::Ident {
    source: IdentSource::NewName,
    role: "col_name",
    validator: None,
    highlight_override: None,
    writes_table: false,
    writes_column: false,
    writes_user_listed_column: false,
};
const COL_NAME: Node = Node::Hinted {
    mode: NEW_NAME_HINT,
    inner: &COL_NAME_IDENT,
};

// ADR-0029 column-constraint suffix — `not null`, `unique`,
// `default <literal>`. (`check (<expr>)` joins in a later
// ADR-0029 step.) One shared fragment: `create table` uses it
// here; `add column` and `add constraint` reuse it later.
const NOT_NULL_NODES: &[Node] = &[
    Node::Word(Word::keyword("not")),
    Node::Word(Word::keyword("null")),
];
const NOT_NULL_CONSTRAINT: Node = Node::Seq(NOT_NULL_NODES);

const UNIQUE_CONSTRAINT: Node = Node::Word(Word::keyword("unique"));

const DEFAULT_CONSTRAINT_NODES: &[Node] = &[
    Node::Word(Word::keyword("default")),
    super::shared::FALLBACK_VALUE_LITERAL,
];
const DEFAULT_CONSTRAINT: Node = Node::Seq(DEFAULT_CONSTRAINT_NODES);

const COLUMN_CONSTRAINT_CHOICES: &[Node] =
    &[NOT_NULL_CONSTRAINT, UNIQUE_CONSTRAINT, DEFAULT_CONSTRAINT];
const COLUMN_CONSTRAINT: Node = Node::Choice(COLUMN_CONSTRAINT_CHOICES);

/// Zero-or-more constraints — the suffix after a column's
/// `(type)` group (ADR-0029 §2.1). `min: 0` so an
/// unconstrained column still matches.
const COLUMN_CONSTRAINT_SUFFIX: Node = Node::Repeated {
    inner: &COLUMN_CONSTRAINT,
    separator: None,
    min: 0,
};

const COL_SPEC_NODES: &[Node] = &[
    COL_NAME,
    Node::Punct('('),
    Node::Ident {
        source: IdentSource::Types,
        role: "col_type",
        validator: Some(TYPE_VALIDATOR),
        highlight_override: None,
        writes_table: false,
        writes_column: false,
        writes_user_listed_column: false,
    },
    Node::Punct(')'),
    COLUMN_CONSTRAINT_SUFFIX,
];
const COL_SPEC: Node = Node::Seq(COL_SPEC_NODES);

const SPEC_LIST: Node = Node::Repeated {
    inner: &COL_SPEC,
    separator: Some(&Node::Punct(',')),
    min: 1,
};
const SPEC_LIST_OPT: Node = Node::Optional(&SPEC_LIST);

const WITH_PK_NODES: &[Node] = &[
    Node::Word(Word::keyword("with")),
    Node::Word(Word::keyword("pk")),
    SPEC_LIST_OPT,
];
const WITH_PK: Node = Node::Seq(WITH_PK_NODES);
const WITH_PK_OPT: Node = Node::Optional(&WITH_PK);

const CREATE_TABLE_NODES: &[Node] = &[
    Node::Word(Word::keyword("table")),
    TABLE_NAME_NEW,
    WITH_PK_OPT,
];
const CREATE_TABLE: Node = Node::Seq(CREATE_TABLE_NODES);

/// The friendly error for declaring a constraint a
/// primary-key column already implies (ADR-0029 §9).
fn redundant_pk_constraint(column: &str, constraint: &str) -> ValidationError {
    ValidationError {
        message_key: "parse.custom.constraint_redundant_on_pk",
        args: vec![
            ("column", column.to_string()),
            ("constraint", constraint.to_string()),
        ],
    }
}

fn build_create_table(path: &MatchedPath) -> Result<Command, ValidationError> {
    let name = require_ident(path, "table_name")?;

    // Walk the matched items, segmenting per column: a
    // `col_name` ident stashes the name, the following
    // `col_type` ident finalises the spec, and the constraint
    // tokens after it (ADR-0029 §2.1) attach to that spec.
    let mut columns: Vec<ColumnSpec> = Vec::new();
    let mut pending_name: Option<String> = None;
    let mut items = path.items.iter().peekable();
    while let Some(item) = items.next() {
        match &item.kind {
            MatchedKind::Ident { role: "col_name", .. } => {
                pending_name = Some(item.text.clone());
            }
            MatchedKind::Ident { role: "col_type", .. } => {
                let ty = item.text.parse::<Type>().map_err(|_| ValidationError {
                    message_key: "parse.error_wrapper",
                    args: vec![("detail", "unknown type".to_string())],
                })?;
                let col_name = pending_name.take().ok_or_else(|| ValidationError {
                    message_key: "parse.error_wrapper",
                    args: vec![("detail", "column type without a name".to_string())],
                })?;
                columns.push(ColumnSpec::new(col_name, ty));
            }
            // `not null` — the grammar's `Seq` guarantees a
            // `null` Word follows a matched `not` Word.
            MatchedKind::Word("not") => {
                if matches!(
                    items.peek().map(|i| &i.kind),
                    Some(MatchedKind::Word("null"))
                ) {
                    items.next();
                    if let Some(last) = columns.last_mut() {
                        last.not_null = true;
                    }
                }
            }
            MatchedKind::Word("unique") => {
                if let Some(last) = columns.last_mut() {
                    last.unique = true;
                }
            }
            // `default <literal>` — the `Seq` guarantees a value
            // item follows a matched `default` Word.
            MatchedKind::Word("default") => {
                let value = items
                    .next()
                    .and_then(crate::dsl::grammar::data::item_to_value)
                    .ok_or_else(|| ValidationError {
                        message_key: "parse.error_wrapper",
                        args: vec![("detail", "default needs a value".to_string())],
                    })?;
                if let Some(last) = columns.last_mut() {
                    last.default = Some(value);
                }
            }
            _ => {}
        }
    }

    // No PK clause OR `with pk` alone (no specs): if `with` was
    // matched, default to id(serial); otherwise reject with the
    // "tables need a primary key" friendly wording.
    if columns.is_empty() {
        let saw_with = path
            .items
            .iter()
            .any(|i| matches!(i.kind, MatchedKind::Word("with")));
        if saw_with {
            columns.push(ColumnSpec::new("id", Type::Serial));
        } else {
            return Err(ValidationError {
                message_key: "parse.custom.create_table_needs_pk",
                args: vec![],
            });
        }
    }

    // Every `with pk` column is part of the primary key
    // (ADR-0029 §2.1). A PK column is already NOT NULL, and a
    // single-column PK is already UNIQUE — declaring those
    // explicitly is a friendly error, not a silent no-op
    // (ADR-0029 §9).
    let single_column_pk = columns.len() == 1;
    for col in &columns {
        if col.not_null {
            return Err(redundant_pk_constraint(&col.name, "NOT NULL"));
        }
        if col.unique && single_column_pk {
            return Err(redundant_pk_constraint(&col.name, "UNIQUE"));
        }
    }

    let primary_key = columns.iter().map(|c| c.name.clone()).collect();

    Ok(Command::CreateTable {
        name,
        columns,
        primary_key,
    })
}

pub static CREATE: CommandNode = CommandNode {
    entry: Word::keyword("create"),
    shape: CREATE_TABLE,
    ast_builder: build_create_table,
    help_id: Some("ddl.create"),
    usage_ids: &["parse.usage.create_table"],};

// =================================================================
// Tests — `create table` column constraints (ADR-0029 §2.1, §9)
// =================================================================

#[cfg(test)]
mod constraint_tests {
    use super::Command;
    use crate::dsl::command::ColumnSpec;
    use crate::dsl::parser::parse_command;
    use crate::dsl::value::Value;

    /// Parse a `create table` and return its column specs.
    fn create_columns(input: &str) -> Vec<ColumnSpec> {
        match parse_command(input).expect("create table should parse") {
            Command::CreateTable { columns, .. } => columns,
            other => panic!("expected CreateTable, got {other:?}"),
        }
    }

    #[test]
    fn create_table_parses_a_text_default() {
        // `grade` is the (single) PK column; `default` is
        // allowed on a PK column (ADR-0029 §9).
        let cols = create_columns("create table T with pk grade(text) default 'A'");
        assert_eq!(cols.len(), 1);
        assert_eq!(cols[0].default, Some(Value::Text("A".to_string())));
        assert!(!cols[0].not_null && !cols[0].unique);
    }

    #[test]
    fn create_table_parses_a_numeric_default_on_a_compound_pk_member() {
        let cols = create_columns("create table T with pk a(int), b(int) default 7");
        assert_eq!(cols.len(), 2);
        assert_eq!(cols[1].default, Some(Value::Number("7".to_string())));
    }

    #[test]
    fn not_null_on_a_pk_column_is_a_redundancy_error() {
        // Every `create table` column is a primary-key column,
        // so `not null` is always redundant there (ADR-0029 §9).
        assert!(parse_command("create table T with pk id(serial) not null").is_err());
    }

    #[test]
    fn unique_on_a_single_column_pk_is_a_redundancy_error() {
        assert!(parse_command("create table T with pk code(text) unique").is_err());
    }

    #[test]
    fn unique_on_a_compound_pk_member_is_allowed() {
        // A compound PK does not make its members individually
        // unique, so an explicit `unique` is meaningful there.
        let cols = create_columns("create table T with pk a(int) unique, b(text)");
        assert_eq!(cols.len(), 2);
        assert!(cols[0].unique, "`a` carries an explicit UNIQUE");
        assert!(!cols[1].unique);
    }

    #[test]
    fn an_unconstrained_create_table_still_parses() {
        let cols = create_columns("create table T with pk id(serial), name(text)");
        assert_eq!(cols.len(), 2);
        assert!(cols.iter().all(|c| !c.not_null && !c.unique && c.default.is_none()));
    }
}