//! Regression: SQL identifiers are case-insensitive, so a user may refer
//! to a table by any capitalization. The engine resolves the name
//! case-insensitively, but our metadata tables and CSV files are keyed by
//! the *stored* case — so before the fix, an operation naming the table in
//! a different case drifted the metadata / silently skipped the CSV write
//! (losing data on reload). Every table-naming executor now canonicalizes
//! the name to its stored case first. These tests pin that behaviour
//! across schema and data operations, including fresh-rebuild round-trips
//! (which reconstruct purely from the text artifacts, so any drift shows).

use rdbms_playground::db::Database;
use rdbms_playground::dsl::{ColumnSpec, 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)
}

/// Drop the db handle, delete the `.db`, reopen, and rebuild purely from
/// the text artifacts (`project.yaml` + CSVs) — where any metadata/CSV
/// drift from a case-variant operation would surface.
fn fresh_rebuild(
    old: Database,
    project: &project::Project,
    r: &tokio::runtime::Runtime,
) -> Database {
    use rdbms_playground::project::PLAYGROUND_DB;
    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 replay(project: &project::Project, db: &Database, r: &tokio::runtime::Runtime, script: &str) {
    std::fs::write(project.path().join("ci.commands"), script).expect("write script");
    let events = r.block_on(run_replay(db, project.path(), "ci.commands"));
    assert!(
        matches!(events.last(), Some(AppEvent::ReplayCompleted { .. })),
        "script replayed cleanly; events: {events:?}"
    );
}

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

#[test]
fn rename_column_with_case_variant_table_keeps_metadata_in_step() {
    // The engine renames the column on the real table; the user-type
    // metadata must follow even when the table is named in a different
    // case (without canonicalization the metadata UPDATE misses and
    // `amount` loses its `int` user-type).
    let (_p, db, _d) = open();
    let r = rt();
    r.block_on(db.create_table(
        "Items".to_string(),
        vec![ColumnSpec::new("id", Type::Int), ColumnSpec::new("qty", Type::Int)],
        vec!["id".to_string()],
        Some("create".to_string()),
    ))
    .expect("create Items");

    r.block_on(db.rename_column(
        "items".to_string(), // ← case variant of `Items`
        "qty".to_string(),
        "amount".to_string(),
        Some("rename".to_string()),
    ))
    .expect("rename column via a case-variant table name");

    let desc = r
        .block_on(db.describe_table("Items".to_string()))
        .expect("describe Items");
    let amount = desc
        .columns
        .iter()
        .find(|c| c.name == "amount")
        .expect("the column was renamed to `amount`");
    assert_eq!(
        amount.user_type,
        Some(Type::Int),
        "the user-type metadata followed the case-variant rename (no drift)"
    );
}

#[test]
fn insert_with_case_variant_table_persists_and_survives_rebuild() {
    // The data-loss case: a wrong-case INSERT executes on the real table
    // (engine is case-insensitive), but the CSV write must target the
    // stored case — otherwise the row is silently absent from the CSV and
    // lost on a fresh rebuild.
    let (project, db, _d) = open();
    let r = rt();
    replay(
        &project,
        &db,
        &r,
        "create table Items with pk id(int)\n\
         add column Items: note (text)\n\
         insert into items (id, note) values (1, 'kept')\n",
    );

    let db = fresh_rebuild(db, &project, &r);
    let rows = r
        .block_on(db.query_data("Items".to_string(), None, None))
        .expect("query")
        .rows;
    assert_eq!(rows.len(), 1, "the wrong-case insert survived the rebuild (no data loss)");
    assert_eq!(rows[0][1].as_deref(), Some("kept"));
}

#[test]
fn add_column_with_case_variant_table_survives_rebuild() {
    let (project, db, _d) = open();
    let r = rt();
    replay(
        &project,
        &db,
        &r,
        "create table Items with pk id(int)\n\
         alter table items add column qty int check (qty >= 0)\n",
    );

    let db = fresh_rebuild(db, &project, &r);
    let desc = r.block_on(db.describe_table("Items".to_string())).expect("describe");
    let qty = desc.columns.iter().find(|c| c.name == "qty").expect("qty added");
    assert_eq!(qty.user_type, Some(Type::Int), "qty's user-type survived the rebuild");
    // The CHECK is intact too (a negative qty is refused under the real table).
    assert!(
        r.block_on(db.insert(
            "Items".to_string(),
            Some(vec!["id".into(), "qty".into()]),
            vec![Value::Number("1".into()), Value::Number("-3".into())],
            Some("i".into()),
        ))
        .is_err(),
        "the CHECK added via a case-variant ALTER is enforced"
    );
}

#[test]
fn drop_table_with_case_variant_name_clears_table_and_csv() {
    let (project, db, _d) = open();
    let r = rt();
    replay(
        &project,
        &db,
        &r,
        "create table Items with pk id(int)\n\
         add column Items: note (text)\n\
         insert into Items (id, note) values (1, 'x')\n\
         drop table items\n",
    );
    assert!(!tables(&db, &r).contains(&"Items".to_string()), "the table was dropped");
    let csv = project.path().join(project::DATA_DIR).join("Items.csv");
    assert!(!csv.exists(), "the CSV was removed despite the case-variant drop");

    // A fresh rebuild yields no Items (the metadata/yaml has no orphan).
    let db = fresh_rebuild(db, &project, &r);
    assert!(!tables(&db, &r).contains(&"Items".to_string()));
}

#[test]
fn rename_table_accepts_case_variant_source() {
    // `alter table orders rename to Sales` when the table is stored as
    // `Orders` now resolves the source case-insensitively and renames it.
    let (project, db, _d) = open();
    let r = rt();
    replay(
        &project,
        &db,
        &r,
        "create table Orders with pk id(int)\n\
         insert into Orders (id) values (1)\n\
         alter table orders rename to Sales\n",
    );
    let t = tables(&db, &r);
    assert!(
        t.contains(&"Sales".to_string()) && !t.contains(&"Orders".to_string()),
        "the case-variant source resolved and the table was renamed: {t:?}"
    );
    let db = fresh_rebuild(db, &project, &r);
    assert!(tables(&db, &r).contains(&"Sales".to_string()));
}

#[test]
fn add_relationship_with_case_variant_tables_survives_rebuild() {
    // The relationship metadata must store the canonical table names, or
    // `describe` (which matches by stored case) would not show it, and a
    // rebuild would emit a relationship against the wrong-case name.
    let (project, db, _d) = open();
    let r = rt();
    replay(
        &project,
        &db,
        &r,
        "create table Parent with pk id(int)\n\
         create table Child with pk id(int)\n\
         add column Child: parent_id (int)\n\
         add 1:n relationship from parent.id to child.parent_id\n",
    );
    // The parent's inbound relationship is visible under the stored case.
    let p = r.block_on(db.describe_table("Parent".to_string())).expect("describe Parent");
    assert_eq!(p.inbound_relationships.len(), 1, "relationship recorded under the stored case");
    assert_eq!(p.inbound_relationships[0].other_table, "Child");

    let db = fresh_rebuild(db, &project, &r);
    let p = r.block_on(db.describe_table("Parent".to_string())).expect("describe Parent");
    assert_eq!(p.inbound_relationships.len(), 1, "relationship survived the rebuild");
    assert_eq!(p.inbound_relationships[0].other_table, "Child");
}