//! Sub-phase 4d integration tests for advanced-mode SQL
//! `CREATE [UNIQUE] INDEX [IF NOT EXISTS]` (ADR-0035 §4d).
//!
//! `SqlCreateIndex` executes through the same `do_add_index` machinery
//! as the simple `add index`, plus the `unique` flag and the
//! `IF NOT EXISTS` no-op-with-note (`CreateIndexOutcome::Skipped`).
//! Parsing (text → `Command::SqlCreateIndex`) is covered by the
//! `sql_create_index_tests` in `src/dsl/grammar/ddl.rs`.

use rdbms_playground::db::{CreateIndexOutcome, Database};
use rdbms_playground::dsl::{ColumnSpec, Type, Value};
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(undo: bool) -> (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, undo)
        .expect("open db with persistence");
    (project, db, dir)
}

/// Create `T (id int primary key, email text)`.
fn make_t(db: &Database, r: &tokio::runtime::Runtime) {
    r.block_on(db.sql_create_table(
        "T".to_string(),
        vec![ColumnSpec::new("id", Type::Int), ColumnSpec::new("email", Type::Text)],
        vec!["id".to_string()],
        vec![],
        vec![],
        vec![],
        false,
        Some("create table T (id int primary key, email text)".to_string()),
    ))
    .expect("create T");
}

fn insert_row(db: &Database, r: &tokio::runtime::Runtime, id: i64, email: &str) -> bool {
    r.block_on(db.insert(
        "T".to_string(),
        Some(vec!["id".to_string(), "email".to_string()]),
        vec![Value::Number(id.to_string()), Value::Text(email.to_string())],
        Some(format!("insert into T (id, email) values ({id}, '{email}')")),
    ))
    .is_ok()
}

fn index(db: &Database, r: &tokio::runtime::Runtime, name: &str) -> Option<(Vec<String>, bool)> {
    r.block_on(db.describe_table("T".to_string(), None))
        .expect("describe")
        .indexes
        .into_iter()
        .find(|i| i.name == name)
        .map(|i| (i.columns, i.unique))
}

#[test]
fn create_plain_index() {
    let (_p, db, _d) = open(false);
    let r = rt();
    make_t(&db, &r);
    let out = r
        .block_on(db.sql_create_index(
            Some("ix".to_string()),
            "T".to_string(),
            vec!["email".to_string()],
            false,
            false,
            Some("create index ix on T (email)".to_string()),
        ))
        .expect("create index");
    assert!(matches!(out, CreateIndexOutcome::Created(_)));
    assert_eq!(index(&db, &r, "ix"), Some((vec!["email".to_string()], false)));
}

#[test]
fn create_unique_index_round_trips_and_survives_rebuild_and_enforces() {
    let (p, db, _d) = open(false);
    let r = rt();
    make_t(&db, &r);
    r.block_on(db.sql_create_index(
        Some("ux".to_string()),
        "T".to_string(),
        vec!["email".to_string()],
        true,
        false,
        Some("create unique index ux on T (email)".to_string()),
    ))
    .expect("create unique index");
    // Reported as unique.
    assert_eq!(index(&db, &r, "ux"), Some((vec!["email".to_string()], true)));
    // Persisted to project.yaml as a unique index.
    let yaml = std::fs::read_to_string(p.path().join("project.yaml")).expect("read project.yaml");
    assert!(yaml.contains("unique: true"), "project.yaml:\n{yaml}");

    // Uniqueness is enforced by the engine.
    assert!(insert_row(&db, &r, 1, "a@x"));
    assert!(!insert_row(&db, &r, 2, "a@x"), "duplicate email refused by the unique index");

    // Rebuild from the text artifacts: the index comes back UNIQUE
    // (the rebuild re-emits CREATE UNIQUE INDEX), not demoted to plain.
    r.block_on(db.rebuild_from_text(p.path().to_path_buf(), Some("rebuild".to_string())))
        .expect("rebuild");
    assert_eq!(
        index(&db, &r, "ux"),
        Some((vec!["email".to_string()], true)),
        "the unique flag survived rebuild"
    );
    // Still enforced after rebuild.
    assert!(!insert_row(&db, &r, 3, "a@x"), "uniqueness enforced after rebuild too");
}

#[test]
fn create_unique_index_on_duplicate_data_is_refused() {
    let (_p, db, _d) = open(false);
    let r = rt();
    make_t(&db, &r);
    assert!(insert_row(&db, &r, 1, "dup@x"));
    assert!(insert_row(&db, &r, 2, "dup@x"));
    // A unique index can't be created over columns that already hold
    // duplicate values — the engine refuses at creation.
    let res = r.block_on(db.sql_create_index(
        Some("ux".to_string()),
        "T".to_string(),
        vec!["email".to_string()],
        true,
        false,
        Some("create unique index ux on T (email)".to_string()),
    ));
    assert!(res.is_err(), "unique index over duplicate data is refused");
}

#[test]
fn if_not_exists_on_an_existing_name_is_a_noop_and_journalled() {
    let (_p, db, _d) = open(false);
    let r = rt();
    make_t(&db, &r);
    r.block_on(db.sql_create_index(
        Some("ix".to_string()),
        "T".to_string(),
        vec!["email".to_string()],
        false,
        false,
        Some("create index ix on T (email)".to_string()),
    ))
    .expect("first create");
    // A second IF NOT EXISTS create of the same name is a no-op.
    let line = "create index if not exists ix on T (email)";
    let out = r
        .block_on(db.sql_create_index(
            Some("ix".to_string()),
            "T".to_string(),
            vec!["email".to_string()],
            false,
            true,
            Some(line.to_string()),
        ))
        .expect("IF NOT EXISTS on an existing index name succeeds as a no-op");
    match out {
        CreateIndexOutcome::Skipped(name) => assert_eq!(name, "ix"),
        CreateIndexOutcome::Created(_) => panic!("expected Skipped, got Created"),
    }
    // ADR-0052: journaling moved to the dispatch layer; this test now
    // asserts only the no-op `Skipped` outcome.
}

#[test]
fn unnamed_if_not_exists_skips_when_the_auto_named_index_exists() {
    // The unnamed form resolves the auto-name `<T>_<cols>_idx`; the skip
    // pre-check must resolve the SAME name (shared `resolve_index_name`).
    // First an unnamed create (auto-named T_email_idx), then an unnamed
    // IF NOT EXISTS create of the same columns → skip on the auto-name.
    let (_p, db, _d) = open(false);
    let r = rt();
    make_t(&db, &r);
    r.block_on(db.sql_create_index(
        None,
        "T".to_string(),
        vec!["email".to_string()],
        false,
        false,
        Some("create index on T (email)".to_string()),
    ))
    .expect("unnamed create");
    let out = r
        .block_on(db.sql_create_index(
            None,
            "T".to_string(),
            vec!["email".to_string()],
            false,
            true,
            Some("create index if not exists on T (email)".to_string()),
        ))
        .expect("unnamed IF NOT EXISTS over the auto-named index is a no-op");
    match out {
        CreateIndexOutcome::Skipped(name) => assert_eq!(name, "T_email_idx"),
        CreateIndexOutcome::Created(_) => panic!("expected Skipped on the auto-name, got Created"),
    }
}

#[test]
fn if_not_exists_short_circuits_only_a_name_collision() {
    // `IF NOT EXISTS` skips only when the *name* already exists. A
    // *different*-named create over already-indexed columns is not a
    // name collision, so it still hits the ADR-0025 redundant-set guard
    // (the playground's pedagogical refusal, not raw-SQL semantics).
    let (_p, db, _d) = open(false);
    let r = rt();
    make_t(&db, &r);
    r.block_on(db.sql_create_index(
        Some("ix".to_string()),
        "T".to_string(),
        vec!["email".to_string()],
        false,
        false,
        Some("create index ix on T (email)".to_string()),
    ))
    .expect("first create");
    // Same columns, a *new* name, with IF NOT EXISTS → not a name
    // collision, so the redundant-set refusal still fires.
    let res = r.block_on(db.sql_create_index(
        Some("ix2".to_string()),
        "T".to_string(),
        vec!["email".to_string()],
        false,
        true,
        Some("create index if not exists ix2 on T (email)".to_string()),
    ));
    assert!(
        res.is_err(),
        "IF NOT EXISTS does not bypass the redundant-column-set guard for a new name"
    );
}

#[test]
fn plain_duplicate_name_errors() {
    let (_p, db, _d) = open(false);
    let r = rt();
    make_t(&db, &r);
    r.block_on(db.sql_create_index(
        Some("ix".to_string()),
        "T".to_string(),
        vec!["email".to_string()],
        false,
        false,
        Some("create index ix on T (email)".to_string()),
    ))
    .expect("first create");
    // Same name again, *without* IF NOT EXISTS → error.
    let res = r.block_on(db.sql_create_index(
        Some("ix".to_string()),
        "T".to_string(),
        vec!["id".to_string()],
        false,
        false,
        Some("create index ix on T (id)".to_string()),
    ));
    assert!(res.is_err(), "duplicate index name without IF NOT EXISTS errors");
}

#[test]
fn plain_and_unique_over_the_same_columns_are_not_duplicates() {
    // The redundant-set guard keys on (columns, unique): a plain and a
    // unique index over the same columns are distinct (different
    // semantics). They need distinct explicit names (the auto-name would
    // collide).
    let (_p, db, _d) = open(false);
    let r = rt();
    make_t(&db, &r);
    r.block_on(db.sql_create_index(
        Some("ix_plain".to_string()),
        "T".to_string(),
        vec!["email".to_string()],
        false,
        false,
        Some("create index ix_plain on T (email)".to_string()),
    ))
    .expect("plain");
    r.block_on(db.sql_create_index(
        Some("ix_unique".to_string()),
        "T".to_string(),
        vec!["email".to_string()],
        true,
        false,
        Some("create unique index ix_unique on T (email)".to_string()),
    ))
    .expect("unique over the same columns is allowed (distinct kind)");
    assert_eq!(index(&db, &r, "ix_plain").map(|(_, u)| u), Some(false));
    assert_eq!(index(&db, &r, "ix_unique").map(|(_, u)| u), Some(true));

    // But an *exact* duplicate (same columns AND same uniqueness) is
    // still refused.
    let res = r.block_on(db.sql_create_index(
        Some("ix_plain2".to_string()),
        "T".to_string(),
        vec!["email".to_string()],
        false,
        false,
        Some("create index ix_plain2 on T (email)".to_string()),
    ));
    assert!(res.is_err(), "a second plain index over the same columns is redundant");
}

#[test]
fn create_index_on_an_internal_table_is_refused_on_both_surfaces() {
    // Internal `__rdbms_*` tables are hidden from the user; indexing one
    // is refused as "no such table" — via the SQL surface and the simple
    // `add index` surface alike (the guard lives in the shared
    // `do_add_index`, ADR-0035 §4d).
    let (_p, db, _d) = open(false);
    let r = rt();
    make_t(&db, &r);
    // SQL CREATE INDEX on an internal table → error.
    let sql = r.block_on(db.sql_create_index(
        Some("bad".to_string()),
        "__rdbms_playground_columns".to_string(),
        vec!["table_name".to_string()],
        false,
        false,
        Some("create index bad on __rdbms_playground_columns (table_name)".to_string()),
    ));
    assert!(sql.is_err(), "SQL CREATE INDEX on an internal table is refused");
    // Simple `add index` on an internal table → error (same guard).
    let dsl = r.block_on(db.add_index(
        Some("bad2".to_string()),
        "__rdbms_playground_columns".to_string(),
        vec!["table_name".to_string()],
        Some("add index as bad2 on __rdbms_playground_columns (table_name)".to_string()),
    ));
    assert!(dsl.is_err(), "simple add index on an internal table is refused");
}

#[test]
fn create_index_is_one_undo_step() {
    let (_p, db, _d) = open(true); // undo enabled
    let r = rt();
    make_t(&db, &r);
    r.block_on(db.sql_create_index(
        Some("ix".to_string()),
        "T".to_string(),
        vec!["email".to_string()],
        true,
        false,
        Some("create unique index ix on T (email)".to_string()),
    ))
    .expect("create index");
    assert!(index(&db, &r, "ix").is_some());
    // One undo removes the index.
    assert!(r.block_on(db.undo()).expect("undo").is_some(), "the create was one undo step");
    assert!(index(&db, &r, "ix").is_none(), "undo removed the index");
}