//! 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_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");
}