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