//! Sub-phase 3e integration tests for the advanced-mode SQL
//! `UPDATE` surface (ADR-0033 §2).
//!
//! Covers the parse path (the dev `sql_update` scaffold lowers to
//! `Command::SqlUpdate`, reconstructing valid `update …` SQL) and
//! the worker round-trip (execute, re-persist the target CSV,
//! append `history.log`). A SQL `UPDATE` without `WHERE` runs
//! across all rows with no rail (ADR-0030 §12).

use rdbms_playground::db::{Database, DbError, UpdateResult};
use rdbms_playground::dsl::{ColumnSpec, Command, Type, 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(),
        false,
    ))
    .unwrap_or_else(|e| panic!("seed {sql:?}: {e:?}"));
}

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

#[test]
fn parse_path_lowers_sql_update_to_command() {
    let command = parse_command("sql_update Orders set total = 0 where id = 1")
        .expect("sql_update parses in advanced mode");
    match command {
        Command::SqlUpdate { sql, target_table, .. } => {
            assert_eq!(sql, "update Orders set total = 0 where id = 1");
            assert_eq!(target_table, "Orders");
        }
        other => panic!("expected Command::SqlUpdate, got {other:?}"),
    }
}

#[test]
fn single_column_update_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, 'old'), (2, 'keep')", "t");
    let result = run_update(&db, &rt, "sql_update t set v = 'new' where id = 1")
        .expect("update runs");
    assert_eq!(result.rows_affected, 1, "one row updated");
    let csv = read_csv(&project, "t").expect("t.csv");
    assert!(csv.contains("new"), "updated value present: {csv:?}");
    assert!(csv.contains("keep"), "untouched row preserved: {csv:?}");
    assert!(!csv.contains("old"), "old value replaced: {csv:?}");
}

#[test]
fn multi_column_update_persists() {
    let (project, db, _dir) = open_project_db();
    let rt = rt();
    create_cols(
        &db,
        &rt,
        "t",
        &[("id", Type::Int), ("a", Type::Int), ("b", Type::Text)],
        &["id"],
    );
    seed(&db, &rt, "insert into t (id, a, b) values (1, 0, 'x')", "t");
    let result = run_update(&db, &rt, "sql_update t set a = 9, b = 'y' where id = 1")
        .expect("multi-col update runs");
    assert_eq!(result.rows_affected, 1);
    let csv = read_csv(&project, "t").expect("t.csv");
    assert!(csv.contains('9') && csv.contains('y'), "both columns updated: {csv:?}");
}

#[test]
fn update_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), ("active", Type::Bool)], &["id"]);
    seed(&db, &rt, "insert into t (id, active) values (1, true), (2, true)", "t");
    let result = run_update(&db, &rt, "sql_update t set active = false")
        .expect("unfiltered update runs");
    assert_eq!(result.rows_affected, 2, "all rows updated");
    let csv = read_csv(&project, "t").expect("t.csv");
    assert!(!csv.contains("true"), "no row left active: {csv:?}");
}

#[test]
fn update_with_sql_expr_in_set() {
    let (project, db, _dir) = open_project_db();
    let rt = rt();
    create_cols(
        &db,
        &rt,
        "t",
        &[("id", Type::Int), ("price", Type::Int), ("qty", Type::Int), ("total", Type::Int)],
        &["id"],
    );
    seed(&db, &rt, "insert into t (id, price, qty, total) values (1, 6, 7, 0)", "t");
    let result = run_update(&db, &rt, "sql_update t set total = price * qty where id = 1")
        .expect("expression update runs");
    assert_eq!(result.rows_affected, 1);
    let csv = read_csv(&project, "t").expect("t.csv");
    assert!(csv.contains("42"), "engine evaluated price*qty: {csv:?}");
}

#[test]
fn update_with_subquery_in_set() {
    // DA gate: the SET RHS admits a scalar subquery.
    let (project, db, _dir) = open_project_db();
    let rt = rt();
    create_cols(&db, &rt, "other", &[("n", Type::Int)], &["n"]);
    create_cols(&db, &rt, "t", &[("id", Type::Int), ("v", Type::Int)], &["id"]);
    seed(&db, &rt, "insert into other (n) values (3), (8), (5)", "other");
    seed(&db, &rt, "insert into t (id, v) values (1, 0)", "t");
    let result = run_update(
        &db,
        &rt,
        "sql_update t set v = (select max(n) from other) where id = 1",
    )
    .expect("subquery-set update runs");
    assert_eq!(result.rows_affected, 1);
    let csv = read_csv(&project, "t").expect("t.csv");
    assert!(csv.contains('8'), "subquery max landed: {csv:?}");
}

#[test]
fn update_matching_no_rows_is_ok() {
    // DA gate: an UPDATE matching nothing succeeds (0 affected),
    // the path doesn't crash, and the CSV is unchanged.
    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, 'keep')", "t");
    let result = run_update(&db, &rt, "sql_update t set v = 'x' where id = 999")
        .expect("no-match update is a success");
    assert_eq!(result.rows_affected, 0, "no rows matched");
    let csv = read_csv(&project, "t").expect("t.csv");
    assert!(csv.contains("keep") && !csv.contains('x'), "unchanged: {csv:?}");
}

#[test]
fn update_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, 'old')", "t");
    let input = "sql_update t set v = 'new' where id = 1";
    run_update(&db, &rt, input).expect("update 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:?}");
}

// =================================================================
// Sub-phase 3g — RETURNING (ADR-0033 §5)
// =================================================================

#[test]
fn update_returning_yields_modified_columns() {
    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, 'old'), (2, 'keep')", "t");
    let result = run_update(&db, &rt, "sql_update t set v = 'new' where id = 1 returning id, v")
        .expect("UPDATE … RETURNING runs");
    assert_eq!(result.rows_affected, 1, "one row updated");
    assert_eq!(result.data.columns, vec!["id".to_string(), "v".to_string()]);
    assert_eq!(result.data.rows.len(), 1);
    // RETURNING reflects the POST-update value.
    assert_eq!(result.data.rows[0][1], Some("new".to_string()), "modified value returned");
}

#[test]
fn update_returning_recovers_bare_column_type() {
    let (_project, db, _dir) = open_project_db();
    let rt = rt();
    create_cols(&db, &rt, "t", &[("id", Type::Int), ("active", Type::Bool)], &["id"]);
    seed(&db, &rt, "insert into t (id, active) values (1, false)", "t");
    let result = run_update(&db, &rt, "sql_update t set active = true where id = 1 returning active")
        .expect("UPDATE … RETURNING active runs");
    assert_eq!(result.data.column_types, vec![Some(Type::Bool)], "bool type recovered");
    assert_eq!(result.data.rows[0][0], Some("true".to_string()));
}

#[test]
fn update_returning_matching_no_rows_is_ok_and_empty() {
    // DA gate: RETURNING makes data.columns non-empty even when no
    // rows match (unlike the 3e column-less case). The operation
    // succeeds with zero rows and an empty result set — no panic, no
    // phantom row.
    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, 'keep')", "t");
    let result = run_update(&db, &rt, "sql_update t set v = 'x' where id = 999 returning id, v")
        .expect("no-match UPDATE … RETURNING is a success");
    assert_eq!(result.rows_affected, 0, "no rows matched");
    assert!(result.data.rows.is_empty(), "no rows returned");
    assert_eq!(result.data.columns, vec!["id".to_string(), "v".to_string()], "columns still present");
}