//! Sub-phase 4e/4f Tier-3 end-to-end tests for advanced-mode SQL
//! `ALTER TABLE` (ADR-0035 §4e + §4f).
//!
//! These drive the **full advanced-mode pipeline** via `run_replay`: a
//! literal `alter table …` line is parsed in Advanced mode, routed to
//! `Command::SqlAlterTable`, decomposed by the runtime to the existing
//! column executor, and persisted. 4e proves the decomposition for
//! add/drop/rename column and the **raw-text DEFAULT/CHECK ADD COLUMN**
//! path; 4f adds `ALTER COLUMN <col> TYPE <type>`, decomposed to
//! `change_column_type` with `ChangeColumnMode::ForceConversion` — the
//! §7 advanced policy (lossy converts with a note, no force flag;
//! static-refused / incompatible still refuse). The drop/rename refusals
//! (PK / FK / index / table-CHECK) and the internal-table guard live in
//! the shared executors and are covered by `tests/column_op_guards.rs` —
//! the SQL surface reaches the same code.

use rdbms_playground::db::Database;
use rdbms_playground::dsl::{Type, Value};
use rdbms_playground::event::AppEvent;
use rdbms_playground::persistence::Persistence;
use rdbms_playground::project;
use rdbms_playground::runtime::run_replay;

fn rt() -> tokio::runtime::Runtime {
    tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .expect("tokio rt")
}

fn open() -> (project::Project, Database, tempfile::TempDir) {
    let dir = tempfile::tempdir().expect("create tempdir");
    let project =
        project::open_or_create(None, Some(dir.path())).expect("open or create project");
    let db = Database::open_with_persistence(
        project.db_path(),
        Persistence::new(project.path().to_path_buf()),
    )
    .expect("db");
    (project, db, dir)
}

fn open_with_undo() -> (project::Project, Database, tempfile::TempDir) {
    let dir = tempfile::tempdir().expect("create tempdir");
    let project =
        project::open_or_create(None, Some(dir.path())).expect("open or create project");
    let db = Database::open_with_persistence_and_undo(
        project.db_path(),
        Persistence::new(project.path().to_path_buf()),
        true,
    )
    .expect("db");
    (project, db, dir)
}

/// Run a single-conversion script through the full pipeline and report
/// whether it aborted with a `ReplayFailed` (i.e. the command was
/// refused). Used to assert the SQL `ALTER COLUMN TYPE` path reaches the
/// shared executor's static / incompatible refusals.
fn replay_is_refused(script: &str) -> bool {
    let (project, db, _d) = open();
    let r = rt();
    std::fs::write(project.path().join("conv.commands"), script).expect("write script");
    let events = r.block_on(run_replay(&db, project.path(), "conv.commands"));
    matches!(events.last(), Some(AppEvent::ReplayFailed { .. }))
}

/// The current user-facing type of column `name` in table `T`.
fn col_type(db: &Database, r: &tokio::runtime::Runtime, name: &str) -> Option<Type> {
    r.block_on(db.describe_table("T".to_string(), None))
        .expect("describe")
        .columns
        .into_iter()
        .find(|c| c.name == name)
        .and_then(|c| c.user_type)
}

fn column_names(db: &Database, r: &tokio::runtime::Runtime) -> Vec<String> {
    r.block_on(db.describe_table("T".to_string(), None))
        .expect("describe")
        .columns
        .into_iter()
        .map(|c| c.name)
        .collect()
}

#[test]
fn e2e_alter_table_add_rename_drop_and_raw_default_check() {
    let (project, db, _d) = open();
    let r = rt();
    // A script exercising all three actions through the full pipeline.
    // `v` is added (simple) so there is a non-PK column to rename/drop;
    // a row is inserted before the ADD so the DEFAULT backfill is
    // exercised by the rebuild.
    std::fs::write(
        project.path().join("alter.commands"),
        "create table T with pk id(int)\n\
         add column T: v (text)\n\
         insert into T (id, v) values (1, 'a')\n\
         alter table T add column qty int default 0 check (qty >= 0)\n\
         alter table T rename column v to label\n\
         alter table T add column note text\n\
         alter table T drop column note\n",
    )
    .expect("write script");

    let events = r.block_on(run_replay(&db, project.path(), "alter.commands"));
    match events.last().expect("at least one event") {
        AppEvent::ReplayCompleted { count, .. } => {
            assert_eq!(*count, 7, "all seven lines replayed; events: {events:?}");
        }
        other => panic!("expected ReplayCompleted, got {other:?} (events: {events:?})"),
    }

    // Final schema: id, label (renamed from v), qty; `note` added then
    // dropped.
    let cols = column_names(&db, &r);
    assert_eq!(cols, vec!["id".to_string(), "label".to_string(), "qty".to_string()]);

    // The DEFAULT backfilled the pre-existing row to qty = 0.
    let rows = r
        .block_on(db.query_data("T".to_string(), None, None, None))
        .expect("query")
        .rows;
    assert_eq!(rows.len(), 1);
    // qty is the third column; the rebuild backfilled the default.
    assert_eq!(rows[0][2].as_deref(), Some("0"), "DEFAULT 0 backfilled the existing row");

    // The CHECK (qty >= 0) is enforced: a negative qty is refused.
    assert!(
        r.block_on(db.insert(
            "T".to_string(),
            Some(vec!["id".to_string(), "qty".to_string()]),
            vec![Value::Number("2".to_string()), Value::Number("-1".to_string())],
            Some("insert".to_string()),
        ))
        .is_err(),
        "the raw-text CHECK (qty >= 0) added via ALTER is enforced"
    );
    // A non-negative qty is accepted.
    r.block_on(db.insert(
        "T".to_string(),
        Some(vec!["id".to_string(), "qty".to_string()]),
        vec![Value::Number("3".to_string()), Value::Number("7".to_string())],
        Some("insert".to_string()),
    ))
    .expect("qty = 7 satisfies the CHECK");
}

#[test]
fn e2e_alter_add_column_survives_rebuild() {
    // The column added via SQL ALTER (with a raw CHECK) round-trips
    // through the text artifacts and survives a rebuild.
    let (project, db, _d) = open();
    let r = rt();
    std::fs::write(
        project.path().join("alter.commands"),
        "create table T with pk id(int)\n\
         alter table T add column qty int check (qty >= 0)\n",
    )
    .expect("write script");
    r.block_on(run_replay(&db, project.path(), "alter.commands"));
    assert!(column_names(&db, &r).contains(&"qty".to_string()));

    r.block_on(db.rebuild_from_text(project.path().to_path_buf(), Some("rebuild".to_string())))
        .expect("rebuild");
    // The CHECK survives the rebuild — a negative qty is still refused.
    assert!(column_names(&db, &r).contains(&"qty".to_string()));
    assert!(
        r.block_on(db.insert(
            "T".to_string(),
            Some(vec!["id".to_string(), "qty".to_string()]),
            vec![Value::Number("1".to_string()), Value::Number("-5".to_string())],
            Some("insert".to_string()),
        ))
        .is_err(),
        "the ALTER-added CHECK is intact after rebuild"
    );
}

// --- 4f: ALTER COLUMN … TYPE (ADR-0035 §4f) -----------------------------

#[test]
fn e2e_alter_column_type_clean_and_lossy_convert() {
    // The key 4f assertion: the SQL ALTER COLUMN TYPE path wires
    // `ForceConversion`. A lossy `real → int` (3.7 → 3) is therefore
    // *performed*, not refused — under `Default` mode the replay line
    // would refuse and abort (count < 6). A clean `int → text` stringifies.
    let (project, db, _d) = open();
    let r = rt();
    std::fs::write(
        project.path().join("conv.commands"),
        "create table T with pk id(int)\n\
         add column T: v (real)\n\
         add column T: w (int)\n\
         insert into T (id, v, w) values (1, 3.7, 42)\n\
         alter table T alter column v type int\n\
         alter table T alter column w type text\n",
    )
    .expect("write script");

    let events = r.block_on(run_replay(&db, project.path(), "conv.commands"));
    match events.last().expect("at least one event") {
        AppEvent::ReplayCompleted { count, .. } => {
            assert_eq!(*count, 6, "all six lines replayed; events: {events:?}");
        }
        other => panic!("expected ReplayCompleted, got {other:?} (events: {events:?})"),
    }

    let rows = r
        .block_on(db.query_data("T".to_string(), None, None, None))
        .expect("query")
        .rows;
    assert_eq!(rows.len(), 1);
    // v (col 1): lossy real→int performed → 3.7 stored as 3.
    assert_eq!(rows[0][1].as_deref(), Some("3"), "lossy real→int performed (3.7→3)");
    // w (col 2): clean int→text stringified → "42".
    assert_eq!(rows[0][2].as_deref(), Some("42"), "clean int→text stringified");

    // The columns now carry the new user-facing types (round-tripped
    // through the metadata).
    assert_eq!(col_type(&db, &r, "v"), Some(Type::Int));
    assert_eq!(col_type(&db, &r, "w"), Some(Type::Text));
}

#[test]
fn e2e_alter_column_type_int_to_serial_is_allowed() {
    // ADR-0035 §7's "static-refused (→serial …)" summary is looser than
    // the code: `int → serial` IS allowed (ADR-0018 §8 — auto-fills nulls,
    // adds UNIQUE on a non-PK column). The SQL path reaches that supported
    // conversion; the pre-existing non-null value is preserved.
    let (project, db, _d) = open();
    let r = rt();
    std::fs::write(
        project.path().join("conv.commands"),
        "create table T with pk id(int)\n\
         add column T: n (int)\n\
         insert into T (id, n) values (1, 100)\n\
         alter table T alter column n type serial\n",
    )
    .expect("write script");
    let events = r.block_on(run_replay(&db, project.path(), "conv.commands"));
    match events.last().expect("at least one event") {
        AppEvent::ReplayCompleted { count, .. } => {
            assert_eq!(*count, 4, "all four lines replayed; events: {events:?}");
        }
        other => panic!("expected ReplayCompleted, got {other:?} (events: {events:?})"),
    }
    assert_eq!(col_type(&db, &r, "n"), Some(Type::Serial), "int→serial converted the column");
    let rows = r
        .block_on(db.query_data("T".to_string(), None, None, None))
        .expect("query")
        .rows;
    assert_eq!(rows[0][1].as_deref(), Some("100"), "the existing value is preserved");
}

#[test]
fn e2e_alter_column_type_incompatible_is_refused() {
    // text "abc" → int has no valid per-cell conversion → refused (no
    // force flag overrides incompatibles). The SQL path reaches the
    // shared executor's incompatible refusal.
    assert!(
        replay_is_refused(
            "create table T with pk id(int)\n\
             add column T: v (text)\n\
             insert into T (id, v) values (1, 'abc')\n\
             alter table T alter column v type int\n",
        ),
        "an incompatible text→int conversion is refused via the SQL path"
    );
}

#[test]
fn e2e_alter_column_type_static_refusals() {
    // Static refusals are shared by both modes (ADR-0017 §3); the SQL
    // ALTER COLUMN TYPE path reaches them.
    assert!(
        replay_is_refused(
            "create table T with pk id(int)\n\
             add column T: v (text)\n\
             alter table T alter column v type serial\n",
        ),
        "text→serial is refused (only int→serial is allowed)"
    );
    assert!(
        replay_is_refused(
            "create table T with pk id(int)\n\
             add column T: v (text)\n\
             alter table T alter column v type blob\n",
        ),
        "↔ blob is statically refused"
    );
}

#[test]
fn e2e_alter_column_type_on_fk_column_is_refused() {
    // The column is the child side of a relationship (outbound FK);
    // changing its type is refused for v1 (ADR-0017 §4.2). The SQL ALTER
    // COLUMN TYPE path reaches the same executor precondition.
    assert!(
        replay_is_refused(
            "create table P with pk id(int)\n\
             create table C with pk cid(int)\n\
             add column C: pid (int)\n\
             add 1:n relationship from P.id to C.pid\n\
             alter table C alter column pid type text\n",
        ),
        "changing the type of a child-side FK column is refused via the SQL path"
    );
}

#[test]
fn e2e_alter_column_type_survives_rebuild() {
    // The user_type metadata update is the existing path, so the
    // converted type round-trips through the text artifacts and survives
    // a rebuild.
    let (project, db, _d) = open();
    let r = rt();
    std::fs::write(
        project.path().join("conv.commands"),
        "create table T with pk id(int)\n\
         add column T: v (real)\n\
         alter table T alter column v type int\n",
    )
    .expect("write script");
    r.block_on(run_replay(&db, project.path(), "conv.commands"));
    assert_eq!(col_type(&db, &r, "v"), Some(Type::Int), "converted before rebuild");

    r.block_on(db.rebuild_from_text(project.path().to_path_buf(), Some("rebuild".to_string())))
        .expect("rebuild");
    assert_eq!(col_type(&db, &r, "v"), Some(Type::Int), "the converted type survives rebuild");
}

#[test]
fn e2e_alter_column_type_is_one_undo_step() {
    // The runtime decomposes SqlAlterTable::AlterColumnType into ONE
    // change_column_type call, so the whole conversion is one undo step
    // (the executor's rebuild is one snapshot) — like the simple
    // `change column`. Driven through the full SQL pipeline (run_replay
    // fires the worker snapshot hook per command), then undone in one.
    let (project, db, _d) = open_with_undo();
    let r = rt();
    std::fs::write(
        project.path().join("conv.commands"),
        "create table T with pk id(int)\n\
         add column T: v (real)\n\
         insert into T (id, v) values (1, 3.7)\n\
         alter table T alter column v type int\n",
    )
    .expect("write script");
    r.block_on(run_replay(&db, project.path(), "conv.commands"));
    assert_eq!(col_type(&db, &r, "v"), Some(Type::Int), "the SQL ALTER COLUMN TYPE converted v");

    // A single undo reverts the whole conversion.
    assert!(
        r.block_on(db.undo()).expect("undo").is_some(),
        "the conversion was one undo step"
    );
    assert_eq!(col_type(&db, &r, "v"), Some(Type::Real), "one undo restored the pre-conversion type");
}

// --- 4g: ADD/DROP constraint + ADD foreign key (ADR-0035 §4g) -----------

/// True if inserting `(id, qty)` into table `T` succeeds.
fn insert_t_qty_ok(db: &Database, r: &tokio::runtime::Runtime, id: i64, qty: i64) -> bool {
    r.block_on(db.insert(
        "T".to_string(),
        Some(vec!["id".to_string(), "qty".to_string()]),
        vec![Value::Number(id.to_string()), Value::Number(qty.to_string())],
        Some("insert".to_string()),
    ))
    .is_ok()
}

#[test]
fn e2e_add_named_check_enforced_and_survives_rebuild_with_its_name() {
    // ADD a named table-CHECK; it is enforced; it round-trips through a
    // rebuild *with its name* — proven by DROP CONSTRAINT <name> still
    // resolving after the rebuild (the name reached project.yaml and back).
    let (project, db, _d) = open();
    let r = rt();
    std::fs::write(
        project.path().join("c.commands"),
        "create table T with pk id(int)\n\
         add column T: qty (int)\n\
         alter table T add constraint qty_positive check (qty >= 0)\n",
    )
    .expect("write");
    let events = r.block_on(run_replay(&db, project.path(), "c.commands"));
    assert!(
        matches!(events.last(), Some(AppEvent::ReplayCompleted { count, .. }) if *count == 3),
        "events: {events:?}"
    );
    // Enforced: qty = -1 refused, qty = 5 accepted.
    assert!(!insert_t_qty_ok(&db, &r, 1, -1), "the CHECK rejects qty = -1");
    assert!(insert_t_qty_ok(&db, &r, 2, 5), "qty = 5 satisfies the CHECK");

    // Rebuild from text, then DROP CONSTRAINT by name must still work →
    // the name survived the round-trip.
    r.block_on(db.rebuild_from_text(project.path().to_path_buf(), Some("rebuild".to_string())))
        .expect("rebuild");
    assert!(!insert_t_qty_ok(&db, &r, 3, -2), "the CHECK is intact after rebuild");
    std::fs::write(
        project.path().join("drop.commands"),
        "alter table T drop constraint qty_positive\n",
    )
    .expect("write");
    let events = r.block_on(run_replay(&db, project.path(), "drop.commands"));
    assert!(
        matches!(events.last(), Some(AppEvent::ReplayCompleted { count, .. }) if *count == 1),
        "DROP CONSTRAINT resolved the name after rebuild; events: {events:?}"
    );
    // After the drop the CHECK no longer applies: qty = -1 is accepted.
    assert!(insert_t_qty_ok(&db, &r, 4, -1), "the CHECK was dropped");
}

#[test]
fn e2e_add_check_with_violating_data_is_refused() {
    assert!(
        replay_is_refused(
            "create table T with pk id(int)\n\
             add column T: qty (int)\n\
             insert into T (id, qty) values (1, -5)\n\
             alter table T add check (qty >= 0)\n",
        ),
        "adding a CHECK that existing rows violate is refused"
    );
}

#[test]
fn e2e_add_composite_unique_enforced_and_survives_rebuild() {
    let (project, db, _d) = open();
    let r = rt();
    std::fs::write(
        project.path().join("u.commands"),
        "create table T with pk id(int)\n\
         add column T: a (int)\n\
         add column T: b (int)\n\
         insert into T (id, a, b) values (1, 1, 2)\n\
         alter table T add unique (a, b)\n",
    )
    .expect("write");
    let events = r.block_on(run_replay(&db, project.path(), "u.commands"));
    assert!(
        matches!(events.last(), Some(AppEvent::ReplayCompleted { count, .. }) if *count == 5),
        "events: {events:?}"
    );
    let dup_ok = |id: i64, a: i64, b: i64| {
        r.block_on(db.insert(
            "T".to_string(),
            Some(vec!["id".to_string(), "a".to_string(), "b".to_string()]),
            vec![
                Value::Number(id.to_string()),
                Value::Number(a.to_string()),
                Value::Number(b.to_string()),
            ],
            Some("insert".to_string()),
        ))
        .is_ok()
    };
    assert!(!dup_ok(2, 1, 2), "the composite UNIQUE rejects the duplicate (1, 2)");
    assert!(dup_ok(3, 1, 3), "(1, 3) is distinct and accepted");

    // Survives rebuild (the unique_constraints yaml path).
    r.block_on(db.rebuild_from_text(project.path().to_path_buf(), Some("rebuild".to_string())))
        .expect("rebuild");
    assert!(!dup_ok(4, 1, 2), "the composite UNIQUE is intact after rebuild");
}

#[test]
fn e2e_add_unique_with_duplicate_data_is_refused() {
    assert!(
        replay_is_refused(
            "create table T with pk id(int)\n\
             add column T: a (int)\n\
             insert into T (id, a) values (1, 7)\n\
             insert into T (id, a) values (2, 7)\n\
             alter table T add unique (a)\n",
        ),
        "adding a UNIQUE that existing rows violate is refused"
    );
}

#[test]
fn e2e_drop_composite_unique_by_derived_name() {
    // ADR-0035 Amendment 1: a composite UNIQUE is anonymous, addressed by
    // its derived name `unique_<cols>`. DROP CONSTRAINT <derived-name>
    // removes it via the rebuild primitive and the UNIQUE stops enforcing.
    let (project, db, _d) = open();
    let r = rt();
    std::fs::write(
        project.path().join("u.commands"),
        "create table T with pk id(int)\n\
         add column T: a (int)\n\
         add column T: b (int)\n\
         alter table T add unique (a, b)\n",
    )
    .expect("write");
    let events = r.block_on(run_replay(&db, project.path(), "u.commands"));
    assert!(
        matches!(events.last(), Some(AppEvent::ReplayCompleted { count, .. }) if *count == 4),
        "events: {events:?}"
    );
    let dup_ok = |id: i64, a: i64, b: i64| {
        r.block_on(db.insert(
            "T".to_string(),
            Some(vec!["id".to_string(), "a".to_string(), "b".to_string()]),
            vec![
                Value::Number(id.to_string()),
                Value::Number(a.to_string()),
                Value::Number(b.to_string()),
            ],
            Some("insert".to_string()),
        ))
        .is_ok()
    };
    assert!(dup_ok(1, 1, 2), "first (1, 2) accepted");
    assert!(!dup_ok(2, 1, 2), "duplicate (1, 2) rejected while the UNIQUE stands");

    // Drop the UNIQUE by its derived name through the existing DROP
    // CONSTRAINT grammar.
    r.block_on(db.alter_drop_constraint(
        "T".to_string(),
        "unique_a_b".to_string(),
        Some("alter table T drop constraint unique_a_b".to_string()),
    ))
    .expect("drop constraint unique_a_b resolves the composite UNIQUE");

    // The UNIQUE no longer enforces: the previously-rejected duplicate is
    // now accepted.
    assert!(dup_ok(3, 1, 2), "duplicate (1, 2) accepted after the UNIQUE was dropped");

    // And it stays gone across a rebuild from text.
    r.block_on(db.rebuild_from_text(project.path().to_path_buf(), Some("rebuild".to_string())))
        .expect("rebuild");
    assert!(dup_ok(4, 1, 2), "still no UNIQUE after rebuild");
}

#[test]
fn e2e_drop_composite_unique_ambiguous_name_is_refused() {
    // Two distinct composite UNIQUEs can derive the same name —
    // `unique (a, b_c)` and `unique (a_b, c)` both → `unique_a_b_c`. The
    // drop must refuse as ambiguous, never guess which to drop.
    let (project, db, _d) = open();
    let r = rt();
    std::fs::write(
        project.path().join("u.commands"),
        "create table T with pk id(int)\n\
         add column T: a (int)\n\
         add column T: b_c (int)\n\
         add column T: a_b (int)\n\
         add column T: c (int)\n\
         alter table T add unique (a, b_c)\n\
         alter table T add unique (a_b, c)\n",
    )
    .expect("write");
    let events = r.block_on(run_replay(&db, project.path(), "u.commands"));
    assert!(
        matches!(events.last(), Some(AppEvent::ReplayCompleted { count, .. }) if *count == 7),
        "setup events: {events:?}"
    );
    let err = r
        .block_on(db.alter_drop_constraint(
            "T".to_string(),
            "unique_a_b_c".to_string(),
            Some("alter table T drop constraint unique_a_b_c".to_string()),
        ))
        .expect_err("an ambiguous derived name is refused, not guessed");
    let msg = err.friendly_message();
    assert!(
        msg.to_lowercase().contains("ambiguous") || msg.to_lowercase().contains("more than one"),
        "refusal explains the ambiguity; got: {msg}"
    );
}

#[test]
fn e2e_drop_composite_unique_is_one_undo_step() {
    // Dropping a composite UNIQUE rebuilds the table = one undo step; undo
    // restores the constraint (ADR-0035 Amendment 1). The drop is the last
    // mutation, so a single undo targets it (checked via describe, so no
    // extra mutation shifts the undo target).
    let (project, db, _d) = open_with_undo();
    let r = rt();
    std::fs::write(
        project.path().join("u.commands"),
        "create table T with pk id(int)\n\
         add column T: a (int)\n\
         add column T: b (int)\n\
         alter table T add unique (a, b)\n",
    )
    .expect("write");
    r.block_on(run_replay(&db, project.path(), "u.commands"));
    let has_unique = || {
        !r.block_on(db.describe_table("T".to_string(), None))
            .expect("describe")
            .unique_constraints
            .is_empty()
    };
    assert!(has_unique(), "the composite UNIQUE exists before the drop");

    r.block_on(db.alter_drop_constraint(
        "T".to_string(),
        "unique_a_b".to_string(),
        Some("alter table T drop constraint unique_a_b".to_string()),
    ))
    .expect("drop the composite UNIQUE");
    assert!(!has_unique(), "the composite UNIQUE is gone after the drop");

    assert!(
        r.block_on(db.undo()).expect("undo").is_some(),
        "the DROP CONSTRAINT was one undo step"
    );
    assert!(has_unique(), "one undo restored the composite UNIQUE");
}

#[test]
fn e2e_add_foreign_key_creates_an_enforced_relationship() {
    let (project, db, _d) = open();
    let r = rt();
    std::fs::write(
        project.path().join("fk.commands"),
        "create table P with pk id(int)\n\
         create table C with pk cid(int)\n\
         add column C: pid (int)\n\
         insert into P (id) values (1)\n\
         alter table C add constraint c_to_p foreign key (pid) references P(id)\n",
    )
    .expect("write");
    let events = r.block_on(run_replay(&db, project.path(), "fk.commands"));
    assert!(
        matches!(events.last(), Some(AppEvent::ReplayCompleted { count, .. }) if *count == 5),
        "events: {events:?}"
    );
    // FK enforced: a child row referencing a missing parent is rejected;
    // one referencing the existing parent (id = 1) is accepted.
    let insert_c = |cid: i64, pid: i64| {
        r.block_on(db.insert(
            "C".to_string(),
            Some(vec!["cid".to_string(), "pid".to_string()]),
            vec![Value::Number(cid.to_string()), Value::Number(pid.to_string())],
            Some("insert".to_string()),
        ))
    };
    assert!(insert_c(10, 1).is_ok(), "a child referencing parent id=1 is accepted");
    assert!(insert_c(11, 999).is_err(), "a child referencing a missing parent is rejected");
}

#[test]
fn e2e_drop_constraint_removes_a_named_foreign_key() {
    let (project, db, _d) = open();
    let r = rt();
    std::fs::write(
        project.path().join("fk.commands"),
        "create table P with pk id(int)\n\
         create table C with pk cid(int)\n\
         add column C: pid (int)\n\
         alter table C add constraint c_to_p foreign key (pid) references P(id)\n\
         alter table C drop constraint c_to_p\n",
    )
    .expect("write");
    let events = r.block_on(run_replay(&db, project.path(), "fk.commands"));
    assert!(
        matches!(events.last(), Some(AppEvent::ReplayCompleted { count, .. }) if *count == 5),
        "events: {events:?}"
    );
    // The FK is gone: a child referencing a missing parent now succeeds.
    assert!(
        r.block_on(db.insert(
            "C".to_string(),
            Some(vec!["cid".to_string(), "pid".to_string()]),
            vec![Value::Number("1".to_string()), Value::Number("999".to_string())],
            Some("insert".to_string()),
        ))
        .is_ok(),
        "after DROP CONSTRAINT the FK no longer applies"
    );
}

#[test]
fn e2e_constraint_name_collision_is_refused() {
    // A named CHECK cannot reuse a relationship (FK) name on the same
    // table — both are `DROP CONSTRAINT <name>` targets, so a collision
    // would make the drop ambiguous.
    assert!(
        replay_is_refused(
            "create table P with pk id(int)\n\
             create table C with pk cid(int)\n\
             add column C: pid (int)\n\
             alter table C add constraint dup foreign key (pid) references P(id)\n\
             alter table C add constraint dup check (cid > 0)\n",
        ),
        "a CHECK reusing an existing FK name on the table is refused"
    );
}

#[test]
fn e2e_drop_unknown_constraint_is_refused() {
    assert!(
        replay_is_refused(
            "create table T with pk id(int)\n\
             alter table T drop constraint nope\n",
        ),
        "dropping a non-existent constraint is refused"
    );
}

#[test]
fn e2e_add_constraint_is_one_undo_step() {
    // ADD CONSTRAINT CHECK is one rebuild = one undo step; driven through
    // the full SQL pipeline, then undone in one.
    let (project, db, _d) = open_with_undo();
    let r = rt();
    std::fs::write(
        project.path().join("c.commands"),
        "create table T with pk id(int)\n\
         add column T: qty (int)\n\
         insert into T (id, qty) values (1, 5)\n\
         alter table T add constraint qty_positive check (qty >= 0)\n",
    )
    .expect("write");
    r.block_on(run_replay(&db, project.path(), "c.commands"));
    assert!(!insert_t_qty_ok(&db, &r, 2, -1), "the CHECK is enforced");

    assert!(
        r.block_on(db.undo()).expect("undo").is_some(),
        "the ADD CONSTRAINT was one undo step"
    );
    // After undo the CHECK is gone: qty = -1 is accepted.
    assert!(insert_t_qty_ok(&db, &r, 3, -1), "one undo removed the CHECK");
}

#[test]
fn e2e_named_check_metadata_survives_a_fresh_rebuild() {
    // A FRESH rebuild (deleted .db, reconstructed from project.yaml) must
    // repopulate the table-CHECK metadata — not just re-emit the CHECK
    // into the recreated DDL. Otherwise the CHECK is enforced but its
    // metadata (incl. the name) is lost: `describe` / `DROP CONSTRAINT` /
    // a later save would drop it (ADR-0035 §4g; fixes a latent 4a.3 gap).
    use rdbms_playground::dsl::ColumnSpec;
    use rdbms_playground::project::PLAYGROUND_DB;
    let dir = tempfile::tempdir().expect("tempdir");
    let r = rt();
    let project_path = {
        let project = project::open_or_create(None, Some(dir.path())).expect("open");
        let path = project.path().to_path_buf();
        let db = Database::open_with_persistence(project.db_path(), Persistence::new(path.clone()))
            .expect("db");
        r.block_on(db.sql_create_table(
            "T".to_string(),
            vec![ColumnSpec::new("id", Type::Int), ColumnSpec::new("qty", Type::Int)],
            vec!["id".to_string()],
            vec![],
            vec![],
            vec![],
            false,
            Some("create table T (id int primary key, qty int)".to_string()),
        ))
        .expect("create");
        r.block_on(db.alter_add_table_check(
            "T".to_string(),
            Some("qty_positive".to_string()),
            "qty >= 0".to_string(),
            Some("alter table T add constraint qty_positive check (qty >= 0)".to_string()),
        ))
        .expect("add named check");
        drop(db);
        path
    };
    // Delete the .db → the next open + rebuild reconstructs from yaml.
    std::fs::remove_file(project_path.join(PLAYGROUND_DB)).unwrap();
    let project = project::Project::open(&project_path).unwrap();
    let db = Database::open_with_persistence(
        project.db_path(),
        Persistence::new(project.path().to_path_buf()),
    )
    .unwrap();
    r.block_on(db.rebuild_from_text(project.path().to_path_buf(), None)).expect("rebuild");

    // The named CHECK metadata survived: DROP CONSTRAINT by name resolves.
    r.block_on(db.alter_drop_constraint(
        "T".to_string(),
        "qty_positive".to_string(),
        Some("drop".to_string()),
    ))
    .expect("DROP CONSTRAINT after a fresh rebuild — the CHECK metadata was reconstructed");
}

// --- 4i (b): describe shows table-level constraints ---------------------

#[test]
fn e2e_describe_shows_table_level_constraints() {
    // ADR-0035 §4i (b): `describe` surfaces composite UNIQUE and
    // table-level CHECK constraints (named + unnamed) — the executor
    // populates them on TableDescription from the metadata.
    let (project, db, _d) = open();
    let r = rt();
    std::fs::write(
        project.path().join("d.commands"),
        "create table T (a integer primary key, b integer, unique (a, b), check (a < b))\n\
         alter table T add constraint a_ne_b check (a <> b)\n",
    )
    .expect("write script");
    let events = r.block_on(run_replay(&db, project.path(), "d.commands"));
    assert!(
        matches!(events.last(), Some(AppEvent::ReplayCompleted { .. })),
        "events: {events:?}"
    );

    let desc = r.block_on(db.describe_table("T".to_string(), None)).expect("describe");
    assert_eq!(
        desc.unique_constraints,
        vec![vec!["a".to_string(), "b".to_string()]],
        "composite UNIQUE surfaced"
    );
    let checks: Vec<(Option<String>, String)> = desc
        .check_constraints
        .iter()
        .map(|c| (c.name.clone(), c.expr.clone()))
        .collect();
    assert!(
        checks.iter().any(|(n, e)| n.is_none() && e.contains("a < b")),
        "unnamed table CHECK surfaced: {checks:?}"
    );
    assert!(
        checks
            .iter()
            .any(|(n, e)| n.as_deref() == Some("a_ne_b") && e.contains("a <> b")),
        "named table CHECK surfaced with its name: {checks:?}"
    );
}

// --- 4h: ALTER TABLE … RENAME TO (ADR-0035 §6) --------------------------

/// Path to a table's CSV in the project data dir.
fn csv_path(project: &project::Project, table: &str) -> std::path::PathBuf {
    project
        .path()
        .join(project::DATA_DIR)
        .join(format!("{table}.csv"))
}

/// Drop the current db handle, delete the `.db`, reopen, and rebuild from
/// the text artifacts (`project.yaml` + CSVs) only — the FRESH rebuild
/// that re-emits DDL from stored metadata via `schema_to_ddl`. This is
/// where the CHECK-text drift (Finding-1) and the FK / metadata
/// reconciliation actually round-trip, unlike an in-place rebuild whose
/// wipe leaves the user tables untouched.
fn fresh_rebuild(
    old: Database,
    project: &project::Project,
    r: &tokio::runtime::Runtime,
) -> Database {
    use rdbms_playground::project::PLAYGROUND_DB;
    // Drop only the db handle (release the .db file) and reuse the live
    // `project` — re-opening the Project would re-acquire its lock file,
    // which the still-alive `project` already holds. The Database does not
    // hold the project lock; only the Project does.
    drop(old);
    std::fs::remove_file(project.path().join(PLAYGROUND_DB)).expect("remove db");
    let db = Database::open_with_persistence(
        project.db_path(),
        Persistence::new(project.path().to_path_buf()),
    )
    .expect("db");
    r.block_on(db.rebuild_from_text(project.path().to_path_buf(), None))
        .expect("rebuild");
    db
}

fn table_names(db: &Database, r: &tokio::runtime::Runtime) -> Vec<String> {
    r.block_on(db.list_tables()).expect("list_tables")
}

#[test]
fn e2e_rename_table_with_rows_csv_follows_and_survives_rebuild() {
    // The CSV file follows the rename (data/<new>.csv written, <old>.csv
    // removed), rows are intact including a NULL (NULL-vs-empty fidelity),
    // and the renamed table round-trips through a FRESH rebuild.
    let (project, db, _d) = open();
    let r = rt();
    std::fs::write(
        project.path().join("rn.commands"),
        "create table Orders with pk id(int)\n\
         add column Orders: note (text)\n\
         insert into Orders (id, note) values (1, 'first')\n\
         insert into Orders (id) values (2)\n\
         alter table Orders rename to Purchases\n",
    )
    .expect("write script");
    let events = r.block_on(run_replay(&db, project.path(), "rn.commands"));
    match events.last().expect("event") {
        AppEvent::ReplayCompleted { count, .. } => {
            assert_eq!(*count, 5, "all five lines replayed; events: {events:?}");
        }
        other => panic!("expected ReplayCompleted, got {other:?} (events: {events:?})"),
    }

    let tables = table_names(&db, &r);
    assert!(
        tables.contains(&"Purchases".to_string()) && !tables.contains(&"Orders".to_string()),
        "the table is now Purchases, not Orders: {tables:?}"
    );
    assert!(csv_path(&project, "Purchases").exists(), "data/Purchases.csv written");
    assert!(!csv_path(&project, "Orders").exists(), "data/Orders.csv removed");

    let rows = r
        .block_on(db.query_data("Purchases".to_string(), None, None, None))
        .expect("query")
        .rows;
    assert_eq!(rows.len(), 2);
    assert_eq!(rows[0][1].as_deref(), Some("first"));
    assert_eq!(rows[1][1], None, "the NULL note survived the rename");

    // FRESH rebuild — the renamed table + its rows reconstruct from text.
    let db = fresh_rebuild(db, &project, &r);
    let tables = table_names(&db, &r);
    assert!(
        tables.contains(&"Purchases".to_string()) && !tables.contains(&"Orders".to_string()),
        "Purchases round-tripped through a fresh rebuild: {tables:?}"
    );
    let rows = r
        .block_on(db.query_data("Purchases".to_string(), None, None, None))
        .expect("query")
        .rows;
    assert_eq!(rows.len(), 2);
    assert_eq!(rows[1][1], None, "NULL preserved across the rebuild");
}

#[test]
fn e2e_rename_table_with_table_qualified_check_survives_fresh_rebuild() {
    // Finding-1 regression. A CHECK that qualifies a column with the table
    // name (`T.age`, and a table-level `T.lo < T.hi`) drifts on rename:
    // the engine rewrites the LIVE CHECK, but the STORED text would stay
    // `T.…` and break a FRESH rebuild (`schema_to_ddl` → "no such table
    // T"). The §2.9 rewrite keeps the stored text in step.
    let (project, db, _d) = open();
    let r = rt();
    std::fs::write(
        project.path().join("rn.commands"),
        "create table T (id integer primary key, age integer check (T.age > 0), lo integer, hi integer, check (T.lo < T.hi))\n\
         insert into T (id, age, lo, hi) values (1, 5, 1, 9)\n\
         alter table T rename to U\n",
    )
    .expect("write script");
    let events = r.block_on(run_replay(&db, project.path(), "rn.commands"));
    assert!(
        matches!(events.last(), Some(AppEvent::ReplayCompleted { .. })),
        "create with table-qualified CHECKs + rename replayed; events: {events:?}"
    );

    // The live CHECKs still enforce under the new name.
    let bad_age = r.block_on(db.insert(
        "U".to_string(),
        Some(vec!["id".into(), "age".into(), "lo".into(), "hi".into()]),
        vec![
            Value::Number("2".into()),
            Value::Number("0".into()),
            Value::Number("1".into()),
            Value::Number("9".into()),
        ],
        Some("i".into()),
    ));
    assert!(bad_age.is_err(), "age > 0 still enforced after rename");

    // The headline: a FRESH rebuild reconstructs from the stored CHECK
    // text — which must now reference U, not T — and still enforces.
    let db = fresh_rebuild(db, &project, &r);
    assert!(
        table_names(&db, &r).contains(&"U".to_string()),
        "U rebuilt from the text artifacts (would fail on 'no such table T' without the rewrite)"
    );
    let bad_after = r.block_on(db.insert(
        "U".to_string(),
        Some(vec!["id".into(), "age".into(), "lo".into(), "hi".into()]),
        vec![
            Value::Number("3".into()),
            Value::Number("-1".into()),
            Value::Number("1".into()),
            Value::Number("9".into()),
        ],
        Some("i".into()),
    ));
    assert!(bad_after.is_err(), "the rewritten CHECK enforces after a fresh rebuild");
}

#[test]
fn e2e_rename_fk_parent_updates_metadata_and_still_enforces() {
    // Renaming an FK *parent* updates the relationship's parent end; the
    // child FK still enforces, and the metadata is consistent enough that
    // a fresh rebuild (which re-emits the FK DDL from metadata) succeeds.
    let (project, db, _d) = open();
    let r = rt();
    std::fs::write(
        project.path().join("rn.commands"),
        "create table P with pk id(int)\n\
         create table C (id integer primary key, p_id integer references P(id))\n\
         insert into P (id) values (1)\n\
         alter table P rename to Parent\n",
    )
    .expect("write script");
    let events = r.block_on(run_replay(&db, project.path(), "rn.commands"));
    assert!(
        matches!(events.last(), Some(AppEvent::ReplayCompleted { .. })),
        "events: {events:?}"
    );

    // The child's outbound relationship now points at the new parent name.
    let c = r.block_on(db.describe_table("C".to_string(), None)).expect("describe C");
    assert_eq!(c.outbound_relationships.len(), 1);
    assert_eq!(c.outbound_relationships[0].other_table, "Parent");

    // FK still enforces: a child row referencing a missing parent fails.
    assert!(
        r.block_on(db.insert(
            "C".to_string(),
            Some(vec!["id".into(), "p_id".into()]),
            vec![Value::Number("9".into()), Value::Number("99".into())],
            Some("i".into()),
        ))
        .is_err(),
        "FK to the renamed parent still enforces"
    );

    // Fresh rebuild re-emits the FK from metadata (parent_table = Parent).
    let db = fresh_rebuild(db, &project, &r);
    let tables = table_names(&db, &r);
    assert!(tables.contains(&"Parent".to_string()) && tables.contains(&"C".to_string()));
    assert!(
        r.block_on(db.insert(
            "C".to_string(),
            Some(vec!["id".into(), "p_id".into()]),
            vec![Value::Number("8".into()), Value::Number("77".into())],
            Some("i".into()),
        ))
        .is_err(),
        "FK still enforces after a fresh rebuild"
    );
}

#[test]
fn e2e_rename_fk_child_updates_metadata_and_still_enforces() {
    // Renaming an FK *child* updates the relationship's child end.
    let (project, db, _d) = open();
    let r = rt();
    std::fs::write(
        project.path().join("rn.commands"),
        "create table P with pk id(int)\n\
         create table C (id integer primary key, p_id integer references P(id))\n\
         insert into P (id) values (1)\n\
         alter table C rename to Child\n",
    )
    .expect("write script");
    let events = r.block_on(run_replay(&db, project.path(), "rn.commands"));
    assert!(
        matches!(events.last(), Some(AppEvent::ReplayCompleted { .. })),
        "events: {events:?}"
    );

    // The parent's inbound relationship now names the renamed child.
    let p = r.block_on(db.describe_table("P".to_string(), None)).expect("describe P");
    assert_eq!(p.inbound_relationships.len(), 1);
    assert_eq!(p.inbound_relationships[0].other_table, "Child");

    // FK still enforces under the new child name; survives a fresh rebuild.
    assert!(
        r.block_on(db.insert(
            "Child".to_string(),
            Some(vec!["id".into(), "p_id".into()]),
            vec![Value::Number("9".into()), Value::Number("99".into())],
            Some("i".into()),
        ))
        .is_err(),
        "FK from the renamed child still enforces"
    );
    let db = fresh_rebuild(db, &project, &r);
    assert!(table_names(&db, &r).contains(&"Child".to_string()));
}

#[test]
fn e2e_rename_self_referential_table_updates_both_ends() {
    // A self-referential FK has parent_table == child_table; both ends
    // must update on rename without a relationship-metadata PK conflict.
    let (project, db, _d) = open();
    let r = rt();
    std::fs::write(
        project.path().join("rn.commands"),
        "create table N (id integer primary key, parent_id integer references N(id))\n\
         insert into N (id, parent_id) values (1, null)\n\
         alter table N rename to Tree\n",
    )
    .expect("write script");
    let events = r.block_on(run_replay(&db, project.path(), "rn.commands"));
    assert!(
        matches!(events.last(), Some(AppEvent::ReplayCompleted { .. })),
        "events: {events:?}"
    );

    // Both ends of the self-reference now name `Tree`.
    let t = r.block_on(db.describe_table("Tree".to_string(), None)).expect("describe Tree");
    assert_eq!(t.outbound_relationships[0].other_table, "Tree");
    assert_eq!(t.inbound_relationships[0].other_table, "Tree");

    // The self-FK still enforces and survives a fresh rebuild.
    assert!(
        r.block_on(db.insert(
            "Tree".to_string(),
            Some(vec!["id".into(), "parent_id".into()]),
            vec![Value::Number("2".into()), Value::Number("99".into())],
            Some("i".into()),
        ))
        .is_err(),
        "self-FK to a missing parent row is rejected"
    );
    let db = fresh_rebuild(db, &project, &r);
    assert!(table_names(&db, &r).contains(&"Tree".to_string()));
    // A valid self-reference (parent_id = 1, which exists) is accepted.
    r.block_on(db.insert(
        "Tree".to_string(),
        Some(vec!["id".into(), "parent_id".into()]),
        vec![Value::Number("3".into()), Value::Number("1".into())],
        Some("i".into()),
    ))
    .expect("valid self-reference accepted after rebuild");
}

#[test]
fn e2e_rename_table_keeps_its_index_with_a_stale_name() {
    // Auto-named indexes embed the old table name and are left STALE on
    // rename (user decision); the index stays functional and survives a
    // fresh rebuild.
    let (project, db, _d) = open();
    let r = rt();
    std::fs::write(
        project.path().join("rn.commands"),
        "create table T with pk id(int)\n\
         add column T: email (text)\n\
         create index on T (email)\n\
         alter table T rename to Users\n",
    )
    .expect("write script");
    let events = r.block_on(run_replay(&db, project.path(), "rn.commands"));
    assert!(
        matches!(events.last(), Some(AppEvent::ReplayCompleted { .. })),
        "events: {events:?}"
    );

    let u = r.block_on(db.describe_table("Users".to_string(), None)).expect("describe Users");
    assert_eq!(u.indexes.len(), 1, "the index followed the rename");
    assert_eq!(
        u.indexes[0].name, "T_email_idx",
        "the auto-name is left stale (embeds the old table name) per the user decision"
    );
    assert_eq!(u.indexes[0].columns, vec!["email".to_string()]);

    // Survives a fresh rebuild (recreated from IndexSchema on table Users).
    let db = fresh_rebuild(db, &project, &r);
    let u = r.block_on(db.describe_table("Users".to_string(), None)).expect("describe Users");
    assert_eq!(u.indexes.len(), 1);
    assert_eq!(u.indexes[0].name, "T_email_idx");
}

#[test]
fn e2e_rename_table_is_one_undo_step() {
    // The rename is one user mutation = one whole-project snapshot = one
    // undo step. Undo restores the old name and its rows; redo reapplies.
    let (project, db, _d) = open_with_undo();
    let r = rt();
    std::fs::write(
        project.path().join("rn.commands"),
        "create table Orders with pk id(int)\n\
         insert into Orders (id) values (1)\n\
         alter table Orders rename to Purchases\n",
    )
    .expect("write script");
    r.block_on(run_replay(&db, project.path(), "rn.commands"));
    assert!(table_names(&db, &r).contains(&"Purchases".to_string()));

    // One undo reverts the rename.
    assert!(r.block_on(db.undo()).expect("undo").is_some(), "rename was one undo step");
    let tables = table_names(&db, &r);
    assert!(
        tables.contains(&"Orders".to_string()) && !tables.contains(&"Purchases".to_string()),
        "undo restored the old table name: {tables:?}"
    );
    assert_eq!(
        r.block_on(db.query_data("Orders".to_string(), None, None, None)).expect("query").rows.len(),
        1,
        "the row is back under the old name"
    );

    // Redo reapplies the rename.
    assert!(r.block_on(db.redo()).expect("redo").is_some());
    let tables = table_names(&db, &r);
    assert!(
        tables.contains(&"Purchases".to_string()) && !tables.contains(&"Orders".to_string()),
        "redo reapplied the rename: {tables:?}"
    );
}

#[test]
fn e2e_rename_table_refusals() {
    // The executor's guards: existing-target, same-name, non-existent
    // source, and an internal `__rdbms_*` target (defense in depth — the
    // parse validator also refuses it, but a synthesised command reaches
    // the worker directly).
    let (project, db, _d) = open();
    let r = rt();
    std::fs::write(
        project.path().join("setup.commands"),
        "create table T with pk id(int)\n\
         create table X with pk id(int)\n",
    )
    .expect("write");
    r.block_on(run_replay(&db, project.path(), "setup.commands"));

    assert!(
        r.block_on(db.rename_table("T".into(), "X".into(), Some("rn".into()))).is_err(),
        "rename to an existing other table is refused"
    );
    assert!(
        r.block_on(db.rename_table("T".into(), "T".into(), Some("rn".into()))).is_err(),
        "rename to the same name is refused"
    );
    assert!(
        r.block_on(db.rename_table("Ghost".into(), "G".into(), Some("rn".into()))).is_err(),
        "rename of a non-existent table is refused"
    );
    assert!(
        r.block_on(db.rename_table("T".into(), "__rdbms_evil".into(), Some("rn".into()))).is_err(),
        "rename to an internal table name is refused at the executor"
    );

    // Case-insensitive collisions are refused with engine-neutral wording
    // (not the raw engine "already another table" error) — the database
    // matches names case-insensitively (ADR-0035 §9).
    let case_only = r.block_on(db.rename_table("T".into(), "t".into(), Some("rn".into())));
    assert!(case_only.is_err(), "a case-only rename is refused");
    if let Err(e) = case_only {
        let msg = e.to_string();
        assert!(
            !msg.to_lowercase().contains("another table") && !msg.to_lowercase().contains("index"),
            "the refusal is engine-neutral, not a raw engine collision error: {msg}"
        );
    }
    assert!(
        r.block_on(db.rename_table("T".into(), "x".into(), Some("rn".into()))).is_err(),
        "rename to a name colliding case-insensitively with another table (X) is refused"
    );

    // The failed renames left the schema untouched.
    let tables = table_names(&db, &r);
    assert!(tables.contains(&"T".to_string()) && tables.contains(&"X".to_string()));
}