//! Sub-phase 3f integration tests for the advanced-mode SQL
//! `DELETE` surface (ADR-0033 §1/§7).
//!
//! Covers the parse path (the dev `sql_delete` scaffold lowers to
//! `Command::SqlDelete`, reconstructing valid `delete from …` SQL)
//! and the worker round-trip (execute, detect FK cascade by
//! row-count diffing per ADR-0033 Amendment 2, re-persist the
//! target *and every cascade-affected child* CSV, append
//! `history.log`). A SQL `DELETE` without `WHERE` runs across all
//! rows with no rail (ADR-0030 §12).
//!
//! The cascade tests pin the Amendment-2 decision: the SQL path
//! uses the *same* count-diff mechanism as the DSL `do_delete`, so
//! the two produce identical `DeleteResult.cascade` on identical
//! schema/data (the `cascade_parity_with_dsl` test asserts this
//! directly). The R2 invariant — a WHERE that itself contains a
//! subquery — is correct by construction because the verbatim SQL
//! executes once and the diff observes the result; no predicate
//! bytes are extracted.

use rdbms_playground::db::{Database, DbError, DeleteResult};
use rdbms_playground::dsl::{
    ColumnSpec, Command, ReferentialAction, RowFilter, Type, Value, parse_command,
};
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_db() -> (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(project.db_path(), persistence)
        .expect("open db with persistence");
    (project, db, dir)
}

fn read_csv(project: &project::Project, table: &str) -> Option<String> {
    std::fs::read_to_string(project.path().join("data").join(format!("{table}.csv"))).ok()
}

fn create_cols(
    db: &Database,
    rt: &tokio::runtime::Runtime,
    name: &str,
    cols: &[(&str, Type)],
    pk: &[&str],
) {
    rt.block_on(db.create_table(
        name.to_string(),
        cols.iter().map(|(n, t)| ColumnSpec::new(*n, *t)).collect(),
        pk.iter().map(|s| (*s).to_string()).collect(),
        None,
    ))
    .unwrap_or_else(|e| panic!("create table {name}: {e:?}"));
}

/// Seed via the SQL INSERT worker path (no shortid columns here, so
/// it executes verbatim).
fn seed(db: &Database, rt: &tokio::runtime::Runtime, sql: &str, target: &str) {
    rt.block_on(db.run_sql_insert(
        sql.to_string(),
        None,
        target.to_string(),
        Vec::new(),
        String::new(),
    ))
    .unwrap_or_else(|e| panic!("seed {sql:?}: {e:?}"));
}

/// Full-stack: parse the dev `sql_delete …` scaffold and run it.
fn run_delete(
    db: &Database,
    rt: &tokio::runtime::Runtime,
    input: &str,
) -> Result<DeleteResult, DbError> {
    match parse_command(input).expect("parse sql_delete") {
        Command::SqlDelete { sql, target_table } => {
            rt.block_on(db.run_sql_delete(sql, Some(input.to_string()), target_table))
        }
        other => panic!("expected Command::SqlDelete, got {other:?}"),
    }
}

/// `Customers (id int pk, Name text)` parent and `Orders (id int
/// pk, CustId int)` child, with `Customers.id → Orders.CustId`
/// `ON DELETE CASCADE`. Seeds Alice (1) with two orders (10, 11)
/// and Bob (2) with one order (12).
fn cascade_fixture(db: &Database, rt: &tokio::runtime::Runtime) {
    create_cols(db, rt, "Customers", &[("id", Type::Int), ("Name", Type::Text)], &["id"]);
    create_cols(db, rt, "Orders", &[("id", Type::Int), ("CustId", Type::Int)], &["id"]);
    rt.block_on(db.add_relationship(
        Some("places".to_string()),
        "Customers".to_string(),
        "id".to_string(),
        "Orders".to_string(),
        "CustId".to_string(),
        ReferentialAction::Cascade,
        ReferentialAction::NoAction,
        false,
        None,
    ))
    .expect("add cascade relationship");
    seed(db, rt, "insert into Customers (id, Name) values (1, 'Alice'), (2, 'Bob')", "Customers");
    seed(db, rt, "insert into Orders (id, CustId) values (10, 1), (11, 1), (12, 2)", "Orders");
}

#[test]
fn parse_path_lowers_sql_delete_to_command() {
    let command = parse_command("sql_delete from Orders where id = 1")
        .expect("sql_delete parses in advanced mode");
    match command {
        Command::SqlDelete { sql, target_table } => {
            assert_eq!(sql, "delete from Orders where id = 1");
            assert_eq!(target_table, "Orders");
        }
        other => panic!("expected Command::SqlDelete, got {other:?}"),
    }
}

#[test]
fn delete_with_where_persists() {
    let (project, db, _dir) = open_project_db();
    let rt = rt();
    create_cols(&db, &rt, "t", &[("id", Type::Int), ("v", Type::Text)], &["id"]);
    seed(&db, &rt, "insert into t (id, v) values (1, 'gone'), (2, 'keep')", "t");
    let result = run_delete(&db, &rt, "sql_delete from t where id = 1").expect("delete runs");
    assert_eq!(result.rows_affected, 1, "one row deleted");
    assert!(result.cascade.is_empty(), "no children, no cascade");
    let csv = read_csv(&project, "t").expect("t.csv");
    assert!(csv.contains("keep"), "untouched row preserved: {csv:?}");
    assert!(!csv.contains("gone"), "deleted row removed from CSV: {csv:?}");
}

#[test]
fn delete_without_where_runs_across_all_rows() {
    // ADR-0030 §12: no `--all-rows` rail.
    let (project, db, _dir) = open_project_db();
    let rt = rt();
    create_cols(&db, &rt, "t", &[("id", Type::Int), ("v", Type::Text)], &["id"]);
    seed(&db, &rt, "insert into t (id, v) values (1, 'a'), (2, 'b'), (3, 'c')", "t");
    let result = run_delete(&db, &rt, "sql_delete from t").expect("unfiltered delete runs");
    assert_eq!(result.rows_affected, 3, "all rows deleted");
    // Empty tables produce no CSV (CLAUDE.md persistence note), so the
    // file is either absent or has only a header — either way, no data.
    let csv = read_csv(&project, "t").unwrap_or_default();
    assert!(!csv.contains('a') && !csv.contains('b') && !csv.contains('c'), "no rows left: {csv:?}");
    let remaining = rt
        .block_on(db.query_data("t".to_string(), None, None, None))
        .expect("query t");
    assert!(remaining.rows.is_empty(), "table empty after unfiltered delete");
}

#[test]
fn cascade_delete_reports_summary_and_repersists_child() {
    let (project, db, _dir) = open_project_db();
    let rt = rt();
    cascade_fixture(&db, &rt);
    // Delete Alice (customer 1) — cascades to her two orders (10, 11).
    let result = run_delete(&db, &rt, "sql_delete from Customers where id = 1")
        .expect("cascading delete runs");
    assert_eq!(result.rows_affected, 1, "one parent row deleted");
    assert_eq!(result.cascade.len(), 1, "one cascade relationship reported");
    let effect = &result.cascade[0];
    assert_eq!(effect.relationship_name, "places");
    assert_eq!(effect.child_table, "Orders");
    // rows_changed == 2 pins the before-execute ordering: counted
    // after the delete it would be 0. Alice had exactly two orders.
    assert_eq!(effect.rows_changed, 2, "both of Alice's orders cascaded");
    // The child CSV must be re-persisted to reflect the cascade.
    let orders_csv = read_csv(&project, "Orders").expect("Orders.csv");
    assert!(orders_csv.contains("12"), "Bob's order (12) preserved: {orders_csv:?}");
    assert!(!orders_csv.contains("10") && !orders_csv.contains("11"),
        "Alice's cascaded orders gone from CSV: {orders_csv:?}");
}

#[test]
fn cascade_parity_with_dsl() {
    // ADR-0033 §2 / Amendment 2: the SQL DELETE cascade summary must
    // match the DSL `do_delete` output on the same schema/data —
    // because they use the identical count-diff mechanism. Run the
    // same operation through both paths on two identical fixtures and
    // compare the cascade vectors directly (CascadeEffect: PartialEq).
    let rt = rt();

    let (_p_sql, db_sql, _d_sql) = open_project_db();
    cascade_fixture(&db_sql, &rt);
    let sql_result = run_delete(&db_sql, &rt, "sql_delete from Customers where id = 1")
        .expect("SQL delete runs");

    let (_p_dsl, db_dsl, _d_dsl) = open_project_db();
    cascade_fixture(&db_dsl, &rt);
    let dsl_result = rt
        .block_on(db_dsl.delete(
            "Customers".to_string(),
            RowFilter::eq("id", Value::Number("1".to_string())),
            Some("delete from Customers where id = 1".to_string()),
        ))
        .expect("DSL delete runs");

    assert_eq!(sql_result.rows_affected, dsl_result.rows_affected, "row counts match");
    assert_eq!(sql_result.cascade, dsl_result.cascade, "cascade summaries identical");
}

#[test]
fn r2_where_with_subquery() {
    // R2 invariant (ADR-0033 §7 / Amendment 2): a WHERE containing a
    // subquery. Plan shape: `DELETE FROM orders WHERE customer_id IN
    // (SELECT id FROM customers WHERE …)`. The verbatim statement
    // executes once; no predicate extraction. Orders has no children,
    // so cascade is empty — the point is the subquery resolves.
    let (project, db, _dir) = open_project_db();
    let rt = rt();
    cascade_fixture(&db, &rt);
    let result = run_delete(
        &db,
        &rt,
        "sql_delete from Orders where CustId in (select id from Customers where Name = 'Alice')",
    )
    .expect("subquery-WHERE delete runs");
    assert_eq!(result.rows_affected, 2, "Alice's two orders deleted");
    assert!(result.cascade.is_empty(), "Orders has no cascade children");
    let orders_csv = read_csv(&project, "Orders").expect("Orders.csv");
    assert!(orders_csv.contains("12"), "Bob's order preserved: {orders_csv:?}");
    assert!(!orders_csv.contains("10") && !orders_csv.contains("11"),
        "Alice's orders deleted: {orders_csv:?}");
}

#[test]
fn r2_cascade_with_subquery_where() {
    // The strongest R2 case: the parent is the DELETE target AND the
    // WHERE subquery reads the very child table that will be cascade-
    // deleted. The engine evaluates the subquery against pre-delete
    // state, deletes the matched parent, then cascades — and the
    // count-diff observes the child rows that vanished. Pins both the
    // subquery correctness and the before-execute ordering together.
    let (project, db, _dir) = open_project_db();
    let rt = rt();
    cascade_fixture(&db, &rt);
    // order 11 belongs to Alice (CustId 1); the subquery yields 1, so
    // Alice is deleted and BOTH her orders (10, 11) cascade.
    let result = run_delete(
        &db,
        &rt,
        "sql_delete from Customers where id in (select CustId from Orders where id = 11)",
    )
    .expect("cascade + subquery-WHERE delete runs");
    assert_eq!(result.rows_affected, 1, "Alice deleted");
    assert_eq!(result.cascade.len(), 1, "one cascade relationship");
    assert_eq!(result.cascade[0].rows_changed, 2, "both Alice orders cascaded");
    let orders_csv = read_csv(&project, "Orders").expect("Orders.csv");
    assert!(orders_csv.contains("12") && !orders_csv.contains("10") && !orders_csv.contains("11"),
        "only Bob's order remains: {orders_csv:?}");
}

#[test]
fn delete_appends_literal_line_to_history() {
    let (project, db, _dir) = open_project_db();
    let rt = rt();
    create_cols(&db, &rt, "t", &[("id", Type::Int), ("v", Type::Text)], &["id"]);
    seed(&db, &rt, "insert into t (id, v) values (1, 'x')", "t");
    let input = "sql_delete from t where id = 1";
    run_delete(&db, &rt, input).expect("delete runs");
    let body = std::fs::read_to_string(project.path().join("history.log"))
        .expect("history.log present");
    assert!(body.contains(input), "history records the literal line: {body:?}");
}

#[test]
fn cascade_to_two_children_reports_both() {
    // DA gate (untested branch): a parent with TWO cascade children
    // must emit a CascadeEffect per affected child, and re-persist
    // both. The single-relationship tests never exercise the loop
    // emitting more than one effect.
    let (project, db, _dir) = open_project_db();
    let rt = rt();
    create_cols(&db, &rt, "Customers", &[("id", Type::Int), ("Name", Type::Text)], &["id"]);
    create_cols(&db, &rt, "Orders", &[("id", Type::Int), ("CustId", Type::Int)], &["id"]);
    create_cols(&db, &rt, "Reviews", &[("id", Type::Int), ("CustId", Type::Int)], &["id"]);
    for (child, name) in [("Orders", "places"), ("Reviews", "writes")] {
        rt.block_on(db.add_relationship(
            Some(name.to_string()),
            "Customers".to_string(),
            "id".to_string(),
            child.to_string(),
            "CustId".to_string(),
            ReferentialAction::Cascade,
            ReferentialAction::NoAction,
            false,
            None,
        ))
        .unwrap_or_else(|e| panic!("add rel {name}: {e:?}"));
    }
    seed(&db, &rt, "insert into Customers (id, Name) values (1, 'Alice')", "Customers");
    seed(&db, &rt, "insert into Orders (id, CustId) values (10, 1), (11, 1)", "Orders");
    seed(&db, &rt, "insert into Reviews (id, CustId) values (20, 1)", "Reviews");

    let result = run_delete(&db, &rt, "sql_delete from Customers where id = 1")
        .expect("cascade-to-two delete runs");
    assert_eq!(result.rows_affected, 1);
    assert_eq!(result.cascade.len(), 2, "both cascade relationships reported");
    let by_child: std::collections::HashMap<&str, i64> = result
        .cascade
        .iter()
        .map(|e| (e.child_table.as_str(), e.rows_changed))
        .collect();
    assert_eq!(by_child.get("Orders"), Some(&2), "two orders cascaded");
    assert_eq!(by_child.get("Reviews"), Some(&1), "one review cascaded");
    // Both child CSVs re-persisted to the post-cascade (empty) state.
    let orders = rt.block_on(db.query_data("Orders".to_string(), None, None, None)).unwrap();
    let reviews = rt.block_on(db.query_data("Reviews".to_string(), None, None, None)).unwrap();
    assert!(orders.rows.is_empty() && reviews.rows.is_empty(), "both children emptied");
    let _ = &project;
}

#[test]
fn delete_childless_parent_reports_no_cascade() {
    // DA gate (untested branch): a cascade relationship EXISTS, but
    // the deleted parent row has no children. The `diff > 0` guard
    // must yield an empty cascade and must NOT touch the child's CSV
    // (a `>= 0` regression would report a phantom 0-row cascade).
    let (project, db, _dir) = open_project_db();
    let rt = rt();
    cascade_fixture(&db, &rt);
    // Carol (3) exists with no orders; deleting her cascades nothing.
    seed(&db, &rt, "insert into Customers (id, Name) values (3, 'Carol')", "Customers");
    let result = run_delete(&db, &rt, "sql_delete from Customers where id = 3")
        .expect("childless-parent delete runs");
    assert_eq!(result.rows_affected, 1, "Carol deleted");
    assert!(result.cascade.is_empty(), "no children → no cascade effect reported");
    // All three orders untouched.
    let orders_csv = read_csv(&project, "Orders").expect("Orders.csv");
    assert!(
        orders_csv.contains("10") && orders_csv.contains("11") && orders_csv.contains("12"),
        "no order touched by a childless-parent delete: {orders_csv:?}"
    );
}

#[test]
fn delete_violating_fk_fails_and_persists_nothing() {
    // DA gate (untested error path): with an ON DELETE NO ACTION
    // child, deleting a referenced parent is rejected by the engine.
    // `do_sql_delete` runs persistence+history INSIDE the tx AFTER a
    // successful execute, so a rejected delete must roll back: the
    // parent row survives and history records no line.
    let (project, db, _dir) = open_project_db();
    let rt = rt();
    create_cols(&db, &rt, "Customers", &[("id", Type::Int), ("Name", Type::Text)], &["id"]);
    create_cols(&db, &rt, "Orders", &[("id", Type::Int), ("CustId", Type::Int)], &["id"]);
    rt.block_on(db.add_relationship(
        Some("places".to_string()),
        "Customers".to_string(),
        "id".to_string(),
        "Orders".to_string(),
        "CustId".to_string(),
        ReferentialAction::NoAction, // on delete: reject if referenced
        ReferentialAction::NoAction,
        false,
        None,
    ))
    .expect("add NO ACTION relationship");
    seed(&db, &rt, "insert into Customers (id, Name) values (1, 'Alice')", "Customers");
    seed(&db, &rt, "insert into Orders (id, CustId) values (10, 1)", "Orders");

    let input = "sql_delete from Customers where id = 1";
    let result = run_delete(&db, &rt, input);
    assert!(result.is_err(), "delete of a referenced parent must be rejected");
    // Rolled back: Alice survives.
    let customers = rt.block_on(db.query_data("Customers".to_string(), None, None, None)).unwrap();
    assert_eq!(customers.rows.len(), 1, "parent row preserved after rejected delete");
    // No history line for the failed statement (written only on success).
    let history = std::fs::read_to_string(project.path().join("history.log")).unwrap_or_default();
    assert!(!history.contains(input), "failed delete not logged: {history:?}");
}

#[test]
fn internal_target_table_rejected_at_parse() {
    // ADR-0030 §6 / ADR-0033 §1: the `__rdbms_*` metadata tables are
    // rejected at the target slot — the parse fails, the statement
    // never reaches the worker.
    assert!(
        parse_command("sql_delete from __rdbms_playground_columns").is_err(),
        "internal table must be rejected at the DELETE target slot"
    );
}