//! 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::completion::{SchemaCache, TableColumn};
use rdbms_playground::db::{Database, DbError, UpdateResult};
use rdbms_playground::dsl::{ColumnSpec, Command, Type, Value, parse_command};
use rdbms_playground::event::AppEvent;
use rdbms_playground::input_render::{
    AmbientHint, InputState, ambient_hint_in_mode, classify_input_with_schema_in_mode,
};
use rdbms_playground::mode::Mode;
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_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 update") {
        Command::SqlUpdate { sql, target_table, returning, set_literals } => rt.block_on(
            db.run_sql_update_with_literals(
                sql,
                Some(input.to_string()),
                target_table,
                returning,
                set_literals,
            ),
        ),
        other => panic!("expected Command::SqlUpdate, got {other:?}"),
    }
}

#[test]
fn parse_path_lowers_sql_update_to_command() {
    let command = parse_command("update Orders set total = 0 where id = 1")
        .expect("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, "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, "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, "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, "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,
        "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, "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 = "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:?}");
}

// =================================================================
// ADR-0036 Phase 2 — `SET` literal value validation
// =================================================================

#[test]
fn sql_update_validates_set_literals_like_the_dsl() {
    // ADR-0036 Phase 2: advanced-mode SQL `UPDATE` now validates each
    // literal `SET col = <literal>` value against its column type before
    // the (still verbatim) update runs, sharing the DSL's per-type
    // validators. `2025/01/15` is a malformed date (slashes, not dashes):
    // the DSL update rejects it at bind time, and advanced-mode SQL now
    // refuses it too (it used to splice the literal into text and let a
    // STRICT TEXT column accept anything).
    let (project, db, _dir) = open_project_db();
    let rt = rt();
    create_cols(&db, &rt, "t", &[("id", Type::Int), ("d", Type::Date)], &["id"]);
    seed(&db, &rt, "insert into t (id, d) values (1, '2025-01-15')", "t");

    // SQL path (advanced mode, full replay pipeline) — REJECTS the bad date.
    std::fs::write(
        project.path().join("bad.commands"),
        "update t set d = '2025/01/15' where id = 1\n",
    )
    .expect("write script");
    let events = rt.block_on(run_replay(&db, project.path(), "bad.commands"));
    assert!(
        matches!(events.last(), Some(AppEvent::ReplayFailed { .. })),
        "advanced-mode SQL validates the `date` SET literal and refuses \
         2025/01/15 (ADR-0036 Phase 2); events: {events:?}"
    );

    // A well-formed date still updates (the verbatim path is unaffected).
    std::fs::write(
        project.path().join("ok.commands"),
        "update t set d = '2025-02-20' where id = 1\n",
    )
    .expect("write script");
    let ok = rt.block_on(run_replay(&db, project.path(), "ok.commands"));
    assert!(
        matches!(ok.last(), Some(AppEvent::ReplayCompleted { count, .. }) if *count == 1),
        "a well-formed date still updates; events: {ok:?}"
    );
}

#[test]
fn sql_update_captures_set_literal_classification() {
    // ADR-0036 Phase 2 seam (the "one new seam to keep honest"): each
    // top-level `SET` RHS is classified — a bare literal (string / signed
    // number / bool / null) is captured as `Some`, while an expression
    // (arithmetic / scalar subquery / function call / column ref) is
    // `None` and left to the engine. Critically, a comma *inside* a
    // function call and a `where` *inside* a subquery must NOT be mistaken
    // for an assignment separator / SET-list terminator (paren-depth
    // guard), and the trailing top-level `WHERE` predicate is not captured.
    let cmd = parse_command(
        "update t set a = '2025-01-15', b = price * qty, c = -5, \
         d = (select max(n) from o where n < 100), e = true, \
         f = coalesce(g, 0), h = null where id = 7",
    )
    .expect("advanced-mode SQL update parses");
    match cmd {
        Command::SqlUpdate { set_literals, .. } => {
            assert_eq!(
                set_literals,
                vec![
                    ("a".to_string(), Some(Value::Text("2025-01-15".to_string()))),
                    ("b".to_string(), None),
                    ("c".to_string(), Some(Value::Number("-5".to_string()))),
                    ("d".to_string(), None),
                    ("e".to_string(), Some(Value::Bool(true))),
                    ("f".to_string(), None),
                    ("h".to_string(), Some(Value::Null)),
                ],
                "literals captured; arithmetic / subquery (with inner WHERE) / \
                 function call (with inner comma) skipped; trailing WHERE excluded",
            );
        }
        other => panic!("expected Command::SqlUpdate, got {other:?}"),
    }
}

#[test]
fn sql_update_validates_every_assignment_not_just_the_first() {
    // A malformed literal in the *second* assignment is caught — the
    // validation loop covers every `SET` literal, not only the first
    // (ADR-0036 Phase 2). The first assignment (`v = 'ok'`) is well-formed.
    let (project, db, _dir) = open_project_db();
    let rt = rt();
    create_cols(
        &db,
        &rt,
        "t",
        &[("id", Type::Int), ("v", Type::Text), ("d", Type::Date)],
        &["id"],
    );
    seed(&db, &rt, "insert into t (id, v, d) values (1, 'a', '2025-01-01')", "t");
    std::fs::write(
        project.path().join("multi.commands"),
        "update t set v = 'ok', d = '2025/01/15' where id = 1\n",
    )
    .expect("write script");
    let events = rt.block_on(run_replay(&db, project.path(), "multi.commands"));
    assert!(
        matches!(events.last(), Some(AppEvent::ReplayFailed { .. })),
        "the malformed date in the second assignment is caught; events: {events:?}"
    );
}

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

// =================================================================
// ADR-0036 Phase 3a — live typed-slot hints + highlighting for
// advanced-mode `SET col = <rhs>` (boundary-aware lookahead).
// =================================================================

/// Build a `SchemaCache` for the advanced-mode typing-surface tests
/// (mirrors `tests/typing_surface`'s `build_schema`).
fn schema_cache(tables: &[(&str, &[(&str, Type)])]) -> SchemaCache {
    let mut cache = SchemaCache::default();
    for (table, cols) in tables {
        let table_cols: Vec<TableColumn> = cols
            .iter()
            .map(|(n, t)| TableColumn {
                name: (*n).to_string(),
                user_type: *t,
                not_null: false,
                has_default: false,
            })
            .collect();
        cache.tables.push((*table).to_string());
        for c in &table_cols {
            if !cache.columns.contains(&c.name) {
                cache.columns.push(c.name.clone());
            }
        }
        cache.table_columns.insert((*table).to_string(), table_cols);
    }
    cache
}

#[test]
fn advanced_update_set_value_offers_typed_slot_hint_for_column() {
    // ADR-0036 Phase 3a: at a `SET col = ` value position the
    // advanced-mode SQL UPDATE now drives the same column-typed slot
    // hint the DSL gives — "for `Email`: type a quoted string …" —
    // instead of the type-blind sql_expr surface.
    let schema = schema_cache(&[(
        "Customers",
        &[("id", Type::Serial), ("Name", Type::Text), ("Email", Type::Text)],
    )]);
    let input = "update Customers set Email=";
    let hint = ambient_hint_in_mode(input, input.len(), None, &schema, Mode::Advanced);
    let Some(AmbientHint::Prose(prose)) = hint else {
        panic!("expected a Prose hint at the typed value slot, got {hint:?}");
    };
    assert!(prose.contains("Email"), "hint names the column `Email`: {prose:?}");
    assert!(
        prose.contains("quoted string"),
        "text-column hint says `quoted string`: {prose:?}"
    );
}

#[test]
fn advanced_update_set_date_value_hint_says_yyyy_mm_dd() {
    let schema = schema_cache(&[("Things", &[("k", Type::Int), ("dt", Type::Date)])]);
    let input = "update Things set dt=";
    let hint = ambient_hint_in_mode(input, input.len(), None, &schema, Mode::Advanced);
    let Some(AmbientHint::Prose(prose)) = hint else {
        panic!("expected a Prose hint at the date value slot, got {hint:?}");
    };
    assert!(
        prose.contains("YYYY-MM-DD"),
        "date-column hint references the YYYY-MM-DD format: {prose:?}"
    );
}

#[test]
fn advanced_update_set_int_value_type_mismatch_is_caught_live() {
    // A decimal literal at an `int` column now fails to parse in
    // advanced mode (the typed slot's integer validator fires while
    // typing) — previously the verbatim sql_expr surface accepted it
    // and only Phase 2's execution-time validation caught it.
    let schema = schema_cache(&[("Things", &[("k", Type::Int), ("note", Type::Text)])]);
    let bad = classify_input_with_schema_in_mode(
        "update Things set k = 3.14 where k = 0",
        &schema,
        Mode::Advanced,
    );
    assert!(
        !matches!(bad, InputState::Valid),
        "a decimal at an int column is rejected live (typed slot), got {bad:?}"
    );
    // A well-formed integer literal still parses cleanly.
    let ok = classify_input_with_schema_in_mode(
        "update Things set k = 5 where k = 0",
        &schema,
        Mode::Advanced,
    );
    assert!(matches!(ok, InputState::Valid), "a valid int literal parses: {ok:?}");
}

#[test]
fn advanced_update_set_expression_still_parses_via_sql_expr() {
    // Regression guard: the boundary-aware lookahead must fall through
    // to sql_expr for anything that is not a lone literal — arithmetic,
    // a literal-prefixed expression, a function call, a scalar subquery.
    // None of these may be stolen by the typed slot.
    let schema = schema_cache(&[
        ("Things", &[("k", Type::Int), ("note", Type::Text)]),
        ("other", &[("n", Type::Int)]),
    ]);
    for input in [
        "update Things set k = 3 + 2 where k = 0",      // literal-prefixed expression
        "update Things set k = (select max(n) from other) where k = 0", // scalar subquery
        "update Things set note = upper(note) where k = 0", // function call
        "update Things set k = -5 where k = 0",         // signed number → sql_expr
    ] {
        let state = classify_input_with_schema_in_mode(input, &schema, Mode::Advanced);
        assert!(
            matches!(state, InputState::Valid),
            "{input:?} must still parse via sql_expr, got {state:?}"
        );
    }
}