//! Sub-phase 3b integration tests for the advanced-mode SQL
//! `INSERT` surface (ADR-0033 §1).
//!
//! Covers:
//! - Worker round-trip: a validated `INSERT` runs against the
//!   database, returns the affected-row count, and re-persists the
//!   target table's CSV (ADR-0030 §11).
//! - Single-row, multi-row, explicit-column-list, and full-arity
//!   (no column list) forms.
//! - `history.log` records the literal submitted line.
//! - A failing INSERT (PK conflict) rolls back and does NOT
//!   re-persist the CSV.
//! - Parse path: `parse_command` (advanced mode) lowers the dev
//!   `sqlinsert` scaffold to `Command::SqlInsert`, reconstructing
//!   valid `insert …` SQL and extracting the target table.
//! - `__rdbms_*` target tables are rejected at the grammar layer.
//!
//! The dev `sqlinsert` entry word keeps the SQL INSERT path
//! isolated from the DSL `insert` word until sub-phase 3j; the
//! worker-level tests call `db.run_sql_insert` directly with the
//! real reconstructed SQL.

use rdbms_playground::db::{Database, DbError, InsertResult};
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()
}

/// Create a two-column table `T(a int pk, b text)` — no
/// auto-generated columns, so 3b's explicit-values requirement is
/// satisfied by every INSERT below.
fn create_t(db: &Database, rt: &tokio::runtime::Runtime) {
    rt.block_on(db.create_table(
        "T".to_string(),
        vec![
            ColumnSpec::new("a", Type::Int),
            ColumnSpec::new("b", Type::Text),
        ],
        vec!["a".to_string()],
        None,
    ))
    .expect("create table T");
}

#[test]
fn single_row_insert_persists_and_counts() {
    let (project, db, _dir) = open_project_db();
    let rt = rt();
    create_t(&db, &rt);
    let result = rt
        .block_on(db.run_sql_insert(
            "insert into T (a, b) values (1, 'Ada')".to_string(),
            Some("insert into T (a, b) values (1, 'Ada')".to_string()),
            "T".to_string(),
            Vec::new(),
            String::new(),
            false,
        ))
        .expect("insert runs");
    assert_eq!(result.rows_affected, 1, "one row inserted");
    let csv = read_csv(&project, "T").expect("T.csv written after insert");
    assert!(csv.contains("Ada"), "CSV reflects the inserted row: {csv:?}");
}

#[test]
fn multi_row_insert_persists_both_rows() {
    let (project, db, _dir) = open_project_db();
    let rt = rt();
    create_t(&db, &rt);
    let result = rt
        .block_on(db.run_sql_insert(
            "insert into T (a, b) values (1, 'first'), (2, 'second')".to_string(),
            None,
            "T".to_string(),
            Vec::new(),
            String::new(),
            false,
        ))
        .expect("multi-row insert runs");
    assert_eq!(result.rows_affected, 2, "two rows inserted");
    let csv = read_csv(&project, "T").expect("T.csv written");
    assert!(
        csv.contains("first") && csv.contains("second"),
        "CSV reflects both inserted rows: {csv:?}",
    );
}

#[test]
fn no_column_list_full_arity_insert_persists() {
    let (project, db, _dir) = open_project_db();
    let rt = rt();
    create_t(&db, &rt);
    let result = rt
        .block_on(db.run_sql_insert(
            "insert into T values (7, 'full-arity')".to_string(),
            None,
            "T".to_string(),
            Vec::new(),
            String::new(),
            false,
        ))
        .expect("full-arity insert runs");
    assert_eq!(result.rows_affected, 1);
    let csv = read_csv(&project, "T").expect("T.csv written");
    assert!(csv.contains("full-arity"), "CSV reflects the row: {csv:?}");
}

#[test]
fn insert_appends_literal_line_to_history() {
    let (project, db, _dir) = open_project_db();
    let rt = rt();
    create_t(&db, &rt);
    // ADR-0030 §11: the literal submitted line lands in history.log.
    let source = "insert into T (a, b) values (1, 'logged')";
    rt.block_on(db.run_sql_insert(
        "insert into T (a, b) values (1, 'logged')".to_string(),
        Some(source.to_string()),
        "T".to_string(),
        Vec::new(),
        String::new(),
        false,
    ))
    .expect("insert runs");
    let body = std::fs::read_to_string(project.path().join("history.log"))
        .expect("history.log present after an INSERT");
    assert!(
        body.contains(source),
        "history.log records the literal INSERT line: {body:?}",
    );
}

#[test]
fn failed_insert_rolls_back_and_does_not_repersist() {
    let (project, db, _dir) = open_project_db();
    let rt = rt();
    create_t(&db, &rt);
    // First insert succeeds and persists.
    rt.block_on(db.run_sql_insert(
        "insert into T (a, b) values (1, 'kept')".to_string(),
        None,
        "T".to_string(),
        Vec::new(),
        String::new(),
        false,
    ))
    .expect("first insert runs");
    // Second insert violates the primary key — it must fail and
    // leave persistence untouched (the transaction rolls back
    // before finalize_persistence).
    let outcome = rt.block_on(db.run_sql_insert(
        "insert into T (a, b) values (1, 'discarded')".to_string(),
        None,
        "T".to_string(),
        Vec::new(),
        String::new(),
        false,
    ));
    assert!(outcome.is_err(), "duplicate PK must fail: {outcome:?}");
    let csv = read_csv(&project, "T").expect("T.csv still present");
    assert!(
        csv.contains("kept") && !csv.contains("discarded"),
        "failed insert must not be persisted: {csv:?}",
    );
}

#[test]
fn failed_multi_row_insert_is_atomic() {
    // ADR-0033 §3b DA gate: a multi-row INSERT whose later tuple
    // violates a constraint fails as one statement — no partial
    // rows land, and the CSV is not rewritten.
    let (project, db, _dir) = open_project_db();
    let rt = rt();
    create_t(&db, &rt);
    rt.block_on(db.run_sql_insert(
        "insert into T (a, b) values (1, 'existing')".to_string(),
        None,
        "T".to_string(),
        Vec::new(),
        String::new(),
        false,
    ))
    .expect("seed row");
    // Row (2,…) is new but (1,…) collides on the PK — the whole
    // statement must fail with neither tuple applied.
    let outcome = rt.block_on(db.run_sql_insert(
        "insert into T (a, b) values (2, 'fresh'), (1, 'collides')".to_string(),
        None,
        "T".to_string(),
        Vec::new(),
        String::new(),
        false,
    ));
    assert!(outcome.is_err(), "multi-row PK conflict must fail: {outcome:?}");
    let csv = read_csv(&project, "T").expect("T.csv still present");
    assert!(
        csv.contains("existing") && !csv.contains("fresh") && !csv.contains("collides"),
        "no tuple from the failed multi-row insert may land: {csv:?}",
    );
}

#[test]
fn parse_path_lowers_sqlinsert_scaffold_to_command() {
    // Advanced-mode parse of the dev scaffold reconstructs valid
    // `insert …` SQL and extracts the target table.
    let command = parse_command("sqlinsert into Orders (id, total) values (1, 99.5)")
        .expect("sqlinsert parses in advanced mode");
    match command {
        Command::SqlInsert { sql, target_table, .. } => {
            assert_eq!(sql, "insert into Orders (id, total) values (1, 99.5)");
            assert_eq!(target_table, "Orders");
        }
        other => panic!("expected Command::SqlInsert, got {other:?}"),
    }
}

#[test]
fn parse_path_rejects_internal_target_table() {
    let result = parse_command("sqlinsert into __rdbms_playground_columns values (1)");
    assert!(
        result.is_err(),
        "an internal `__rdbms_*` target must be rejected: {result:?}",
    );
}

// =================================================================
// Sub-phase 3c — INSERT … SELECT
// =================================================================

/// Create a two-column table `name(a int pk, b text)`.
fn create_named(db: &Database, rt: &tokio::runtime::Runtime, name: &str) {
    rt.block_on(db.create_table(
        name.to_string(),
        vec![
            ColumnSpec::new("a", Type::Int),
            ColumnSpec::new("b", Type::Text),
        ],
        vec!["a".to_string()],
        None,
    ))
    .unwrap_or_else(|e| panic!("create table {name}: {e:?}"));
}

#[test]
fn parse_path_lowers_insert_select_to_command() {
    let command = parse_command("sqlinsert into archive select * from source")
        .expect("INSERT … SELECT parses in advanced mode");
    match command {
        Command::SqlInsert { sql, target_table, .. } => {
            assert_eq!(sql, "insert into archive select * from source");
            assert_eq!(target_table, "archive");
        }
        other => panic!("expected Command::SqlInsert, got {other:?}"),
    }
}

#[test]
fn parse_path_lowers_with_prefixed_insert_select() {
    // R4: a WITH-prefixed SELECT row source lowers verbatim.
    let command = parse_command(
        "sqlinsert into archive with t as (select * from orders) select * from t",
    )
    .expect("WITH-prefixed INSERT … SELECT parses");
    match command {
        Command::SqlInsert { sql, target_table, .. } => {
            assert_eq!(
                sql,
                "insert into archive with t as (select * from orders) select * from t",
            );
            assert_eq!(target_table, "archive");
        }
        other => panic!("expected Command::SqlInsert, got {other:?}"),
    }
}

#[test]
fn insert_select_copies_rows_and_persists() {
    let (project, db, _dir) = open_project_db();
    let rt = rt();
    create_named(&db, &rt, "source");
    create_named(&db, &rt, "archive");
    rt.block_on(db.run_sql_insert(
        "insert into source (a, b) values (1, 'one'), (2, 'two')".to_string(),
        None,
        "source".to_string(),
        Vec::new(),
        String::new(),
        false,
    ))
    .expect("seed source");
    let result = rt
        .block_on(db.run_sql_insert(
            "insert into archive select * from source".to_string(),
            Some("insert into archive select * from source".to_string()),
            "archive".to_string(),
            Vec::new(),
            String::new(),
            false,
        ))
        .expect("INSERT … SELECT runs");
    assert_eq!(result.rows_affected, 2, "both source rows copied");
    let csv = read_csv(&project, "archive").expect("archive.csv written");
    assert!(
        csv.contains("one") && csv.contains("two"),
        "archive CSV reflects the copied rows: {csv:?}",
    );
}

#[test]
fn insert_select_with_column_list_and_projection_persists() {
    let (project, db, _dir) = open_project_db();
    let rt = rt();
    create_named(&db, &rt, "source");
    create_named(&db, &rt, "target");
    rt.block_on(db.run_sql_insert(
        "insert into source (a, b) values (5, 'five')".to_string(),
        None,
        "source".to_string(),
        Vec::new(),
        String::new(),
        false,
    ))
    .expect("seed source");
    let result = rt
        .block_on(db.run_sql_insert(
            "insert into target (a, b) select a, b from source".to_string(),
            None,
            "target".to_string(),
            Vec::new(),
            String::new(),
            false,
        ))
        .expect("column-list + projection INSERT … SELECT runs");
    assert_eq!(result.rows_affected, 1);
    let csv = read_csv(&project, "target").expect("target.csv written");
    assert!(csv.contains("five"), "target CSV reflects the row: {csv:?}");
}

#[test]
fn with_prefixed_insert_select_runs_and_persists() {
    // R4 end-to-end: the CTE row source executes and lands rows.
    let (project, db, _dir) = open_project_db();
    let rt = rt();
    create_named(&db, &rt, "orders");
    create_named(&db, &rt, "archive");
    rt.block_on(db.run_sql_insert(
        "insert into orders (a, b) values (1, 'a'), (2, 'b')".to_string(),
        None,
        "orders".to_string(),
        Vec::new(),
        String::new(),
        false,
    ))
    .expect("seed orders");
    let result = rt
        .block_on(db.run_sql_insert(
            "insert into archive with t as (select * from orders) select * from t".to_string(),
            None,
            "archive".to_string(),
            Vec::new(),
            String::new(),
            false,
        ))
        .expect("WITH-prefixed INSERT … SELECT runs");
    assert_eq!(result.rows_affected, 2);
    let csv = read_csv(&project, "archive").expect("archive.csv written");
    assert!(
        csv.contains('a') && csv.contains('b'),
        "archive CSV reflects the CTE-sourced rows: {csv:?}",
    );
}

#[test]
fn insert_select_from_self_runs_as_plain_insert() {
    // DA gate: INSERT … SELECT where the source is the target
    // executes as a plain insert (InsertResult — no cascade
    // summary; cascade output is a DELETE-only concept, 3f).
    let (project, db, _dir) = open_project_db();
    let rt = rt();
    create_named(&db, &rt, "T");
    rt.block_on(db.run_sql_insert(
        "insert into T (a, b) values (1, 'x'), (2, 'y')".to_string(),
        None,
        "T".to_string(),
        Vec::new(),
        String::new(),
        false,
    ))
    .expect("seed");
    let result = rt
        .block_on(db.run_sql_insert(
            "insert into T select a + 10, b from T".to_string(),
            None,
            "T".to_string(),
            Vec::new(),
            String::new(),
            false,
        ))
        .expect("self-sourced INSERT … SELECT runs");
    assert_eq!(result.rows_affected, 2, "two rows copied with shifted PK");
    let csv = read_csv(&project, "T").expect("T.csv written");
    assert!(
        csv.contains("11") && csv.contains("12"),
        "the shifted-PK copies landed: {csv:?}",
    );
}

// =================================================================
// Sub-phase 3d — shortid auto-fill (worker)
// =================================================================

/// Full-stack: parse the dev `sqlinsert …` scaffold (so
/// `listed_columns` / `row_source` are extracted exactly as the
/// app does) and run it through the worker.
fn run_sqlinsert(
    db: &Database,
    rt: &tokio::runtime::Runtime,
    input: &str,
) -> Result<InsertResult, DbError> {
    match parse_command(input).expect("parse sqlinsert") {
        Command::SqlInsert {
            sql,
            target_table,
            listed_columns,
            row_source,
            returning,
        } => rt.block_on(db.run_sql_insert(
            sql,
            Some(input.to_string()),
            target_table,
            listed_columns,
            row_source,
            returning,
        )),
        other => panic!("expected Command::SqlInsert, got {other:?}"),
    }
}

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:?}"));
}

/// Data rows of a table's CSV (header skipped), each split on `,`.
/// The test data uses comma/quote-free values, so a plain split is
/// sufficient.
fn csv_rows(project: &project::Project, table: &str) -> Vec<Vec<String>> {
    read_csv(project, table)
        .unwrap_or_default()
        .lines()
        .skip(1)
        .filter(|l| !l.is_empty())
        .map(|l| l.split(',').map(str::to_string).collect())
        .collect()
}

#[test]
fn values_autofills_omitted_shortid_pk() {
    let (project, db, _dir) = open_project_db();
    let rt = rt();
    create_cols(&db, &rt, "t", &[("id", Type::ShortId), ("label", Type::Text)], &["id"]);
    let result = run_sqlinsert(&db, &rt, "sqlinsert into t (label) values ('x')")
        .expect("auto-fill insert runs");
    assert_eq!(result.rows_affected, 1);
    let rows = csv_rows(&project, "t");
    assert_eq!(rows.len(), 1, "one row: {rows:?}");
    assert!(!rows[0][0].is_empty(), "id auto-filled: {rows:?}");
    assert_eq!(rows[0][1], "x", "label preserved: {rows:?}");
}

#[test]
fn values_multirow_autofills_distinct_shortids() {
    let (project, db, _dir) = open_project_db();
    let rt = rt();
    create_cols(&db, &rt, "t", &[("id", Type::ShortId), ("label", Type::Text)], &["id"]);
    let result = run_sqlinsert(
        &db,
        &rt,
        "sqlinsert into t (label) values ('a'), ('b'), ('c')",
    )
    .expect("multi-row auto-fill runs");
    assert_eq!(result.rows_affected, 3);
    let rows = csv_rows(&project, "t");
    let ids: std::collections::HashSet<&String> = rows.iter().map(|r| &r[0]).collect();
    assert_eq!(ids.len(), 3, "three DISTINCT non-empty shortids: {rows:?}");
    assert!(rows.iter().all(|r| !r[0].is_empty()), "no empty id: {rows:?}");
}

#[test]
fn explicit_shortid_value_is_respected() {
    let (project, db, _dir) = open_project_db();
    let rt = rt();
    create_cols(&db, &rt, "t", &[("id", Type::ShortId), ("label", Type::Text)], &["id"]);
    // The user provided `id` explicitly — it must be honoured
    // verbatim (the override WARNING is sub-phase 3i).
    run_sqlinsert(
        &db,
        &rt,
        "sqlinsert into t (id, label) values ('hardcoded', 'x')",
    )
    .expect("explicit-id insert runs");
    let rows = csv_rows(&project, "t");
    assert_eq!(rows[0][0], "hardcoded", "explicit id preserved: {rows:?}");
}

#[test]
fn insert_select_autofills_distinct_shortids() {
    let (project, db, _dir) = open_project_db();
    let rt = rt();
    create_cols(&db, &rt, "source", &[("id", Type::ShortId), ("label", Type::Text)], &["id"]);
    create_cols(&db, &rt, "target", &[("id", Type::ShortId), ("label", Type::Text)], &["id"]);
    run_sqlinsert(&db, &rt, "sqlinsert into source (label) values ('a'), ('b')")
        .expect("seed source");
    let result = run_sqlinsert(
        &db,
        &rt,
        "sqlinsert into target (label) select label from source",
    )
    .expect("INSERT … SELECT auto-fill runs");
    assert_eq!(result.rows_affected, 2);
    let rows = csv_rows(&project, "target");
    let ids: std::collections::HashSet<&String> = rows.iter().map(|r| &r[0]).collect();
    assert_eq!(ids.len(), 2, "two DISTINCT fresh shortids: {rows:?}");
    assert!(rows.iter().all(|r| !r[0].is_empty()), "no empty id: {rows:?}");
}

#[test]
fn combined_serial_and_shortid_autofill() {
    let (project, db, _dir) = open_project_db();
    let rt = rt();
    // id: serial PK (engine rowid), code: shortid (worker), name.
    create_cols(
        &db,
        &rt,
        "t",
        &[("id", Type::Serial), ("code", Type::ShortId), ("name", Type::Text)],
        &["id"],
    );
    run_sqlinsert(&db, &rt, "sqlinsert into t (name) values ('x')")
        .expect("combined auto-fill runs");
    let rows = csv_rows(&project, "t");
    assert_eq!(rows.len(), 1, "{rows:?}");
    assert_eq!(rows[0][0], "1", "serial PK engine-filled: {rows:?}");
    assert!(!rows[0][1].is_empty(), "shortid worker-filled: {rows:?}");
    assert_eq!(rows[0][2], "x", "name preserved: {rows:?}");
}

#[test]
fn autofill_logs_original_source_not_rewritten_sql() {
    // ADR-0030 §11: even though the worker rewrites the executed
    // statement to bind synthesised shortids, history.log records
    // the user's original line verbatim.
    let (project, db, _dir) = open_project_db();
    let rt = rt();
    create_cols(&db, &rt, "t", &[("id", Type::ShortId), ("label", Type::Text)], &["id"]);
    let input = "sqlinsert into t (label) values ('x')";
    run_sqlinsert(&db, &rt, input).expect("auto-fill insert runs");
    let body = std::fs::read_to_string(project.path().join("history.log"))
        .expect("history.log present");
    assert!(body.contains(input), "original line logged: {body:?}");
    // The rewritten parameterised INSERT must not leak into history.
    assert!(
        !body.contains("INSERT INTO") && !body.contains("?1"),
        "rewritten SQL must not be logged: {body:?}",
    );
}

#[test]
fn shortid_autofill_respects_mixed_case_column_name() {
    // ADR-0009 / 3d DA gate: identifiers are case-preserving. The
    // omitted-shortid detection must match the case-preserved
    // schema name `MyId`, not a lowercased form.
    let (project, db, _dir) = open_project_db();
    let rt = rt();
    create_cols(&db, &rt, "t", &[("MyId", Type::ShortId), ("label", Type::Text)], &["MyId"]);
    run_sqlinsert(&db, &rt, "sqlinsert into t (label) values ('x')")
        .expect("mixed-case shortid auto-fill runs");
    let rows = csv_rows(&project, "t");
    assert_eq!(rows.len(), 1, "{rows:?}");
    assert!(!rows[0][0].is_empty(), "MyId auto-filled: {rows:?}");
}

#[test]
fn two_shortids_pk_and_nonpk_both_autofill_distinctly() {
    // Two shortid columns (one PK, one not), both omitted: each
    // gets its own distinct-per-row batch.
    let (project, db, _dir) = open_project_db();
    let rt = rt();
    create_cols(
        &db,
        &rt,
        "t",
        &[("id", Type::ShortId), ("code", Type::ShortId), ("label", Type::Text)],
        &["id"],
    );
    let result = run_sqlinsert(&db, &rt, "sqlinsert into t (label) values ('x'), ('y')")
        .expect("two-shortid auto-fill runs");
    assert_eq!(result.rows_affected, 2);
    let rows = csv_rows(&project, "t");
    let ids: std::collections::HashSet<&String> = rows.iter().map(|x| &x[0]).collect();
    let codes: std::collections::HashSet<&String> = rows.iter().map(|x| &x[1]).collect();
    assert_eq!(ids.len(), 2, "distinct ids: {rows:?}");
    assert_eq!(codes.len(), 2, "distinct codes: {rows:?}");
    assert!(
        rows.iter().all(|x| !x[0].is_empty() && !x[1].is_empty()),
        "both shortid columns filled on every row: {rows:?}",
    );
}

#[test]
fn two_shortids_one_provided_one_autofilled() {
    // The user provides one shortid and omits the other; the
    // provided value is honoured and only the omitted one fills.
    let (project, db, _dir) = open_project_db();
    let rt = rt();
    create_cols(
        &db,
        &rt,
        "t",
        &[("id", Type::ShortId), ("code", Type::ShortId), ("label", Type::Text)],
        &["id"],
    );
    run_sqlinsert(&db, &rt, "sqlinsert into t (id, label) values ('myid', 'x')")
        .expect("partial-shortid insert runs");
    let rows = csv_rows(&project, "t");
    assert_eq!(rows[0][0], "myid", "provided id preserved: {rows:?}");
    assert!(!rows[0][1].is_empty(), "omitted code auto-filled: {rows:?}");
}

#[test]
fn compound_pk_with_shortid_member_autofills() {
    // A shortid that is part of a compound PK still auto-fills when
    // omitted (membership in the PK is irrelevant to the fill).
    let (project, db, _dir) = open_project_db();
    let rt = rt();
    create_cols(
        &db,
        &rt,
        "t",
        &[("id", Type::ShortId), ("region", Type::Int), ("label", Type::Text)],
        &["id", "region"],
    );
    run_sqlinsert(&db, &rt, "sqlinsert into t (region, label) values (1, 'x')")
        .expect("compound-pk insert runs");
    let rows = csv_rows(&project, "t");
    assert!(
        !rows[0][0].is_empty(),
        "shortid PK member auto-filled: {rows:?}",
    );
    assert_eq!(rows[0][1], "1", "{rows:?}");
}

#[test]
fn autofill_does_not_mask_arity_mismatch() {
    // A column-list / value-count mismatch must still be rejected
    // even when a shortid auto-fill would otherwise kick in — the
    // auto-fill path must not silently drop the extra value. (3i
    // adds a friendly pre-flight; until then we defer to the
    // engine rather than mask the error.)
    let (project, db, _dir) = open_project_db();
    let rt = rt();
    create_cols(&db, &rt, "t", &[("id", Type::ShortId), ("label", Type::Text)], &["id"]);
    let outcome = run_sqlinsert(&db, &rt, "sqlinsert into t (label) values ('a', 'b')");
    assert!(
        outcome.is_err(),
        "arity mismatch must be rejected, not masked: {outcome:?}",
    );
    let rows = csv_rows(&project, "t");
    assert!(rows.is_empty(), "no row should land on a rejected insert: {rows:?}");
}

#[test]
fn autofill_insert_select_wider_projection_is_rejected() {
    // SELECT projects more columns than the list: the guard defers
    // to the engine instead of dropping the extra projection.
    let (project, db, _dir) = open_project_db();
    let rt = rt();
    create_cols(&db, &rt, "src", &[("a", Type::Text), ("b", Type::Text)], &["a"]);
    create_cols(&db, &rt, "t", &[("id", Type::ShortId), ("label", Type::Text)], &["id"]);
    run_sqlinsert(&db, &rt, "sqlinsert into src (a, b) values ('p', 'q')").expect("seed");
    let outcome = run_sqlinsert(&db, &rt, "sqlinsert into t (label) select a, b from src");
    assert!(outcome.is_err(), "wider projection must be rejected: {outcome:?}");
    assert!(csv_rows(&project, "t").is_empty(), "nothing should land");
}

#[test]
fn autofill_insert_select_narrower_projection_is_rejected() {
    // SELECT projects fewer columns than the list: the guard
    // prevents an out-of-range read and defers to the engine.
    let (project, db, _dir) = open_project_db();
    let rt = rt();
    create_cols(&db, &rt, "src", &[("a", Type::Text)], &["a"]);
    create_cols(
        &db,
        &rt,
        "t",
        &[("id", Type::ShortId), ("x", Type::Text), ("y", Type::Text)],
        &["id"],
    );
    run_sqlinsert(&db, &rt, "sqlinsert into src (a) values ('p')").expect("seed");
    let outcome = run_sqlinsert(&db, &rt, "sqlinsert into t (x, y) select a from src");
    assert!(outcome.is_err(), "narrower projection must be rejected: {outcome:?}");
    assert!(csv_rows(&project, "t").is_empty(), "nothing should land");
}

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

#[test]
fn insert_returning_star_returns_inserted_row() {
    let (_project, db, _dir) = open_project_db();
    let rt = rt();
    create_cols(&db, &rt, "t", &[("id", Type::Int), ("b", Type::Text)], &["id"]);
    let result = run_sqlinsert(&db, &rt, "sqlinsert into t (id, b) values (1, 'Ada') returning *")
        .expect("INSERT … RETURNING * runs");
    assert_eq!(result.rows_affected, 1, "one row inserted");
    assert_eq!(result.data.rows.len(), 1, "RETURNING yielded the inserted row");
    assert_eq!(result.data.columns, vec!["id".to_string(), "b".to_string()]);
    assert_eq!(result.data.rows[0][1], Some("Ada".to_string()));
}

#[test]
fn insert_multirow_returning_id_yields_distinct_rows() {
    let (_project, db, _dir) = open_project_db();
    let rt = rt();
    create_cols(&db, &rt, "t", &[("id", Type::Int), ("b", Type::Text)], &["id"]);
    let result = run_sqlinsert(
        &db,
        &rt,
        "sqlinsert into t (id, b) values (1, 'a'), (2, 'b'), (3, 'c') returning id",
    )
    .expect("multi-row INSERT … RETURNING id runs");
    assert_eq!(result.rows_affected, 3, "three rows inserted");
    assert_eq!(result.data.columns, vec!["id".to_string()]);
    let ids: std::collections::BTreeSet<_> =
        result.data.rows.iter().map(|r| r[0].clone()).collect();
    assert_eq!(ids.len(), 3, "three distinct ids returned: {:?}", result.data.rows);
}

#[test]
fn insert_returning_autofills_shortid_and_returns_it() {
    // The auto-fill × RETURNING interaction (3d × 3g): the worker
    // rewrites the INSERT to add the generated shortid, and the
    // rewrite must PRESERVE the RETURNING tail so the generated id
    // surfaces in the returned row.
    let (_project, db, _dir) = open_project_db();
    let rt = rt();
    create_cols(&db, &rt, "t", &[("id", Type::ShortId), ("label", Type::Text)], &["id"]);
    let result = run_sqlinsert(&db, &rt, "sqlinsert into t (label) values ('x') returning *")
        .expect("auto-fill INSERT … RETURNING * runs");
    assert_eq!(result.rows_affected, 1, "one row inserted (RETURNING-counted)");
    assert_eq!(result.data.rows.len(), 1, "RETURNING yielded the row");
    // `id` is the auto-filled shortid column; it must be non-empty in
    // the returned row (proving the rewrite kept RETURNING).
    let id_idx = result.data.columns.iter().position(|c| c == "id").expect("id column");
    let id_val = result.data.rows[0][id_idx].clone();
    assert!(id_val.is_some_and(|s| !s.is_empty()), "generated shortid surfaced via RETURNING");
}

#[test]
fn insert_returning_recovers_bare_column_type() {
    // 3g type recovery: a bare-column RETURNING ref recovers its
    // playground type via the column-origin path (a `bool` column
    // renders as the word, not 0/1).
    let (_project, db, _dir) = open_project_db();
    let rt = rt();
    create_cols(&db, &rt, "t", &[("id", Type::Int), ("active", Type::Bool)], &["id"]);
    let result = run_sqlinsert(&db, &rt, "sqlinsert into t (id, active) values (1, true) returning active")
        .expect("INSERT … 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()), "rendered as the bool word");
}

#[test]
fn insert_returning_computed_expression_is_typeless() {
    // 3g: a computed RETURNING projection has no base-table origin,
    // so its recovered type is None (renders with neutral alignment).
    let (_project, db, _dir) = open_project_db();
    let rt = rt();
    create_cols(&db, &rt, "t", &[("id", Type::Int), ("n", Type::Int)], &["id"]);
    let result = run_sqlinsert(&db, &rt, "sqlinsert into t (id, n) values (1, 5) returning n + 1")
        .expect("INSERT … RETURNING <expr> runs");
    assert_eq!(result.data.column_types, vec![None], "computed projection is typeless");
    assert_eq!(result.data.rows[0][0], Some("6".to_string()), "engine evaluated n + 1");
}

#[test]
fn insert_returning_recovers_multiple_bare_column_types() {
    // 3g type recovery spans the playground vocabulary. RETURNING
    // reuses the SELECT column-origin path (`resolve_select_column_
    // types`), exhaustively type-tested on the SELECT side; this
    // pins a representative spread reached via the RETURNING tail.
    let (_project, db, _dir) = open_project_db();
    let rt = rt();
    create_cols(
        &db,
        &rt,
        "t",
        &[
            ("id", Type::Int),
            ("txt", Type::Text),
            ("amount", Type::Decimal),
            ("ratio", Type::Real),
            ("flag", Type::Bool),
        ],
        &["id"],
    );
    let result = run_sqlinsert(
        &db,
        &rt,
        "sqlinsert into t (id, txt, amount, ratio, flag) values (1, 'a', 9.50, 1.5, true) returning id, txt, amount, ratio, flag",
    )
    .expect("INSERT … RETURNING <cols> runs");
    assert_eq!(
        result.data.column_types,
        vec![
            Some(Type::Int),
            Some(Type::Text),
            Some(Type::Decimal),
            Some(Type::Real),
            Some(Type::Bool),
        ],
        "each bare-column RETURNING ref recovered its playground type",
    );
}

#[test]
fn multirow_autofill_returning_yields_distinct_generated_ids() {
    // DA gate (3d × 3g): multi-row INSERT with an omitted shortid PK
    // AND RETURNING — the auto-fill rewrite produces N tuples with N
    // distinct generated ids, and RETURNING * must surface all N
    // rows each carrying its own generated id.
    let (_project, db, _dir) = open_project_db();
    let rt = rt();
    create_cols(&db, &rt, "t", &[("id", Type::ShortId), ("label", Type::Text)], &["id"]);
    let result = run_sqlinsert(
        &db,
        &rt,
        "sqlinsert into t (label) values ('a'), ('b'), ('c') returning *",
    )
    .expect("multi-row auto-fill INSERT … RETURNING * runs");
    assert_eq!(result.rows_affected, 3, "three rows inserted");
    assert_eq!(result.data.rows.len(), 3, "three rows returned");
    let id_idx = result.data.columns.iter().position(|c| c == "id").expect("id column");
    let ids: std::collections::BTreeSet<_> =
        result.data.rows.iter().map(|r| r[id_idx].clone()).collect();
    assert_eq!(ids.len(), 3, "three DISTINCT generated ids via RETURNING: {:?}", result.data.rows);
    assert!(ids.iter().all(|v| v.as_ref().is_some_and(|s| !s.is_empty())), "all ids non-empty");
}

#[test]
fn insert_select_returning_executes_and_returns_rows() {
    // DA gate: the grammar accepts INSERT … SELECT … RETURNING; this
    // pins that it also EXECUTES through run_returning (the SELECT row
    // source feeds the insert, and RETURNING yields the inserted rows).
    let (_project, db, _dir) = open_project_db();
    let rt = rt();
    create_cols(&db, &rt, "src", &[("id", Type::Int), ("b", Type::Text)], &["id"]);
    create_cols(&db, &rt, "dst", &[("id", Type::Int), ("b", Type::Text)], &["id"]);
    run_sqlinsert(&db, &rt, "sqlinsert into src (id, b) values (1, 'x'), (2, 'y')").expect("seed src");
    let result = run_sqlinsert(&db, &rt, "sqlinsert into dst select * from src returning id, b")
        .expect("INSERT … SELECT … RETURNING runs");
    assert_eq!(result.rows_affected, 2, "two rows copied");
    assert_eq!(result.data.rows.len(), 2, "RETURNING yielded both inserted rows");
    let bs: std::collections::BTreeSet<_> =
        result.data.rows.iter().map(|r| r[1].clone()).collect();
    assert!(bs.contains(&Some("x".to_string())) && bs.contains(&Some("y".to_string())));
}