//! Executor-level guards on the shared column operations (ADR-0035 §4e).
//!
//! These guards live in `do_add_column` / `do_drop_column` /
//! `do_rename_column`, so they apply to BOTH the simple-mode DSL
//! commands (exercised here) and the advanced-mode SQL `ALTER TABLE`
//! (which reaches the same executors). Two guards:
//!   1. internal `__rdbms_*` tables are refused as "no such table";
//!   2. dropping/renaming a column a table-level CHECK references is
//!      refused up-front (the 4a.3 deferral; it also fixes a latent
//!      rename-drift bug that would break a later rebuild).

use rdbms_playground::db::Database;
use rdbms_playground::dsl::command::Constraint;
use rdbms_playground::dsl::{ChangeColumnMode, ColumnSpec, ReferentialAction, Type};
use rdbms_playground::persistence::Persistence;
use rdbms_playground::project;

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 persistence = Persistence::new(project.path().to_path_buf());
    let db = Database::open_with_persistence_and_undo(project.db_path(), persistence, true)
        .expect("open db with persistence");
    (project, db, dir)
}

/// `T (id int pk, a int, b int, c text)` with a table-level CHECK
/// `a < b`.
fn make_t_with_check(db: &Database, r: &tokio::runtime::Runtime) {
    r.block_on(db.sql_create_table(
        "T".to_string(),
        vec![
            ColumnSpec::new("id", Type::Int),
            ColumnSpec::new("a", Type::Int),
            ColumnSpec::new("b", Type::Int),
            ColumnSpec::new("c", Type::Text),
        ],
        vec!["id".to_string()],
        vec![],
        vec!["a < b".to_string()],
        vec![],
        false,
        Some("create table T (id int primary key, a int, b int, c text, check (a < b))".to_string()),
    ))
    .expect("create T with table CHECK");
}

#[test]
fn simple_column_ops_refuse_internal_tables() {
    let (_p, db, _d) = open();
    let r = rt();
    let internal = "__rdbms_playground_columns".to_string();
    assert!(
        r.block_on(db.add_column(
            internal.clone(),
            ColumnSpec::new("x", Type::Int),
            Some("add column".to_string())
        ))
        .is_err(),
        "add column on an internal table is refused"
    );
    assert!(
        r.block_on(db.drop_column(internal.clone(), "table_name".to_string(), false, None))
            .is_err(),
        "drop column on an internal table is refused"
    );
    assert!(
        r.block_on(db.rename_column(
            internal.clone(),
            "table_name".to_string(),
            "tn".to_string(),
            None
        ))
        .is_err(),
        "rename column on an internal table is refused"
    );
    // `change column` (the simple surface; also the SQL `ALTER COLUMN
    // TYPE` decomposition target — ADR-0035 §4f) is refused too: the
    // guard lives in `do_change_column_type`. It refuses up-front as
    // "no such table" (the sibling-executor contract), not via the
    // incidental "no user-facing type metadata" path internal tables
    // happen to hit.
    let err = r
        .block_on(db.change_column_type(
            internal.clone(),
            "table_name".to_string(),
            Type::Int,
            ChangeColumnMode::Default,
            None,
        ))
        .expect_err("change column type on an internal table is refused");
    assert!(
        format!("{err:?}").contains("NoSuchTable"),
        "expected a no-such-table refusal from the internal-table guard, got: {err:?}"
    );
    // `add constraint` (the simple surface; also the SQL `ALTER TABLE …
    // ADD CONSTRAINT` decomposition target — ADR-0035 §4g) is refused:
    // the guard lives in `do_add_constraint`.
    let err = r
        .block_on(db.add_constraint(
            internal,
            "table_name".to_string(),
            Constraint::NotNull,
            None,
        ))
        .expect_err("add constraint on an internal table is refused");
    assert!(
        format!("{err:?}").contains("NoSuchTable"),
        "expected a no-such-table refusal from the internal-table guard, got: {err:?}"
    );
}

#[test]
fn add_relationship_refuses_internal_tables() {
    // The guard lives in `do_add_relationship` (ADR-0035 §4g) and covers
    // both the parent and the child endpoint — so the simple `add 1:n
    // relationship` and the SQL `ALTER TABLE … ADD FOREIGN KEY` (which
    // reaches the same executor) cannot touch an internal table.
    let (_p, db, _d) = open();
    let r = rt();
    let internal = "__rdbms_playground_relationships".to_string();
    // Internal *parent* — refused up-front.
    let err = r
        .block_on(db.add_relationship(
            None,
            internal.clone(),
            "name".to_string(),
            "C".to_string(),
            "x".to_string(),
            ReferentialAction::NoAction,
            ReferentialAction::NoAction,
            false,
            None,
        ))
        .expect_err("relationship with an internal parent is refused");
    assert!(
        format!("{err:?}").contains("NoSuchTable"),
        "expected a no-such-table refusal (internal parent), got: {err:?}"
    );
    // Internal *child* — also refused (a real parent exists).
    r.block_on(db.sql_create_table(
        "P".to_string(),
        vec![ColumnSpec::new("id", Type::Int)],
        vec!["id".to_string()],
        vec![],
        vec![],
        vec![],
        false,
        Some("create table P (id int primary key)".to_string()),
    ))
    .expect("create P");
    let err = r
        .block_on(db.add_relationship(
            None,
            "P".to_string(),
            "id".to_string(),
            internal,
            "x".to_string(),
            ReferentialAction::NoAction,
            ReferentialAction::NoAction,
            false,
            None,
        ))
        .expect_err("relationship with an internal child is refused");
    assert!(
        format!("{err:?}").contains("NoSuchTable"),
        "expected a no-such-table refusal (internal child), got: {err:?}"
    );
}

#[test]
fn drop_column_referenced_by_a_table_check_is_refused() {
    let (_p, db, _d) = open();
    let r = rt();
    make_t_with_check(&db, &r);
    // `a` is referenced by the CHECK `a < b` → refused (both surfaces;
    // here via the simple `drop column`).
    assert!(
        r.block_on(db.drop_column("T".to_string(), "a".to_string(), false, None))
            .is_err(),
        "dropping a CHECK-referenced column is refused"
    );
    // `c` is not referenced → the drop succeeds.
    r.block_on(db.drop_column("T".to_string(), "c".to_string(), false, None))
        .expect("dropping an unreferenced column succeeds");
}

#[test]
fn rename_column_referenced_by_a_table_check_is_refused() {
    let (_p, db, _d) = open();
    let r = rt();
    make_t_with_check(&db, &r);
    // `a` is referenced → refused (without this guard, a native rename
    // would silently drift the CHECK metadata and break rebuild).
    assert!(
        r.block_on(db.rename_column("T".to_string(), "a".to_string(), "z".to_string(), None))
            .is_err(),
        "renaming a CHECK-referenced column is refused"
    );
    // `c` is not referenced → rename succeeds.
    r.block_on(db.rename_column("T".to_string(), "c".to_string(), "note".to_string(), None))
        .expect("renaming an unreferenced column succeeds");
}

/// `T (id int pk, price int, discount int CHECK(discount < price),
/// qty int CHECK(qty >= 0))` — column-level CHECKs (ADR-0035 §4e).
fn make_t_with_column_checks(db: &Database, r: &tokio::runtime::Runtime) {
    let mut discount = ColumnSpec::new("discount", Type::Int);
    discount.check_sql = Some("discount < price".to_string());
    let mut qty = ColumnSpec::new("qty", Type::Int);
    qty.check_sql = Some("qty >= 0".to_string());
    r.block_on(db.sql_create_table(
        "T".to_string(),
        vec![
            ColumnSpec::new("id", Type::Int),
            ColumnSpec::new("price", Type::Int),
            discount,
            qty,
        ],
        vec!["id".to_string()],
        vec![],
        vec![],
        vec![],
        false,
        Some("create table T (...)".to_string()),
    ))
    .expect("create T with column CHECKs");
}

#[test]
fn rename_column_with_a_column_level_check_is_refused() {
    // A native RENAME would leave the stored column-level CHECK text
    // stale (drift → broken rebuild), so it is refused — including a
    // column's own self-check.
    let (_p, db, _d) = open();
    let r = rt();
    make_t_with_column_checks(&db, &r);
    // `qty`'s own check `qty >= 0` references qty → refused.
    assert!(
        r.block_on(db.rename_column("T".to_string(), "qty".to_string(), "amount".to_string(), None))
            .is_err(),
        "renaming a column with its own column-level CHECK is refused"
    );
    // `price` is referenced by `discount`'s check `discount < price`.
    assert!(
        r.block_on(db.rename_column("T".to_string(), "price".to_string(), "cost".to_string(), None))
            .is_err(),
        "renaming a column referenced by another column's CHECK is refused"
    );
    // `id` is referenced by no CHECK → rename succeeds.
    r.block_on(db.rename_column("T".to_string(), "id".to_string(), "pk".to_string(), None))
        .expect("renaming an unreferenced column succeeds");
}

#[test]
fn drop_column_referenced_by_another_columns_check_is_refused_but_own_check_drops() {
    let (p, db, _d) = open();
    let r = rt();
    make_t_with_column_checks(&db, &r);
    // `price` is referenced by `discount`'s check → refused.
    assert!(
        r.block_on(db.drop_column("T".to_string(), "price".to_string(), false, None))
            .is_err(),
        "dropping a column another column's CHECK references is refused"
    );
    // `qty` has only its OWN check → it drops with the column.
    r.block_on(db.drop_column("T".to_string(), "qty".to_string(), false, None))
        .expect("dropping a column whose only CHECK is its own succeeds");
    // Rebuild still works (the remaining `discount < price` CHECK's
    // columns survive).
    r.block_on(db.rebuild_from_text(p.path().to_path_buf(), Some("rebuild".to_string())))
        .expect("rebuild succeeds after dropping the self-checked column");
}

#[test]
fn rebuild_survives_after_dropping_an_unreferenced_column() {
    // Guard is not over-broad: a table that carries a CHECK still
    // rebuilds after an unrelated column is dropped (the CHECK's
    // referenced columns remain).
    let (p, db, _d) = open();
    let r = rt();
    make_t_with_check(&db, &r);
    r.block_on(db.drop_column("T".to_string(), "c".to_string(), false, None))
        .expect("drop unreferenced column");
    r.block_on(db.rebuild_from_text(p.path().to_path_buf(), Some("rebuild".to_string())))
        .expect("rebuild succeeds — the CHECK still references existing columns");
    // The CHECK is intact: it still enforces a < b.
    assert!(
        r.block_on(db.insert(
            "T".to_string(),
            Some(vec!["id".to_string(), "a".to_string(), "b".to_string()]),
            vec![
                rdbms_playground::dsl::Value::Number("1".to_string()),
                rdbms_playground::dsl::Value::Number("5".to_string()),
                rdbms_playground::dsl::Value::Number("3".to_string()),
            ],
            Some("insert".to_string()),
        ))
        .is_err(),
        "CHECK a < b still enforced after the rebuild (5 < 3 is false)"
    );
}