//! Tier-3 integration tests for the `seed` command (ADR-0048, the
//! Phase-1 walking skeleton). Covers the parse path (grammar → AST),
//! the worker round-trip (rows generated + persisted to CSV),
//! reproducibility via a fixed `--seed`, and the single `history.log`
//! line for the whole command (ADR-0048 D15 / U3).

use rdbms_playground::db::Database;
use rdbms_playground::dsl::{ColumnSpec, Command, ReferentialAction, 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()
}

/// `People(id serial pk, name text, email text)` — `id` is autogen
/// (excluded from generation, so no PK collisions), `name`/`email`
/// are generated.
fn create_people(db: &Database, rt: &tokio::runtime::Runtime) {
    rt.block_on(db.create_table(
        "People".to_string(),
        vec![
            ColumnSpec::new("id", Type::Serial),
            ColumnSpec::new("name", Type::Text),
            ColumnSpec::new("email", Type::Text),
        ],
        vec!["id".to_string()],
        None,
    ))
    .expect("create People");
}

/// Data rows in a CSV = non-empty lines minus the header.
fn data_row_count(csv: &str) -> usize {
    csv.lines()
        .filter(|l| !l.trim().is_empty())
        .count()
        .saturating_sub(1)
}

#[test]
fn seed_parses_with_and_without_count() {
    match parse_command("seed People 5").expect("`seed People 5` parses") {
        Command::Seed {
            table,
            target_column,
            count,
            overrides,
            rng_seed,
        } => {
            assert_eq!(table, "People");
            assert_eq!(target_column, None);
            assert_eq!(count, Some(5));
            assert!(overrides.is_empty());
            assert_eq!(rng_seed, None);
        }
        other => panic!("expected Command::Seed, got {other:?}"),
    }
    match parse_command("seed People").expect("`seed People` parses") {
        Command::Seed { table, count, .. } => {
            assert_eq!(table, "People");
            assert_eq!(count, None, "omitted count is None (executor defaults to 20)");
        }
        other => panic!("expected Command::Seed, got {other:?}"),
    }
}

#[test]
fn seed_parses_the_reproducibility_flag() {
    // `--seed <n>` after a count.
    match parse_command("seed People 5 --seed 42").expect("count + --seed parses") {
        Command::Seed {
            table,
            count,
            rng_seed,
            ..
        } => {
            assert_eq!(table, "People");
            assert_eq!(count, Some(5));
            assert_eq!(rng_seed, Some(42), "the value after --seed is the rng seed");
        }
        other => panic!("expected Command::Seed, got {other:?}"),
    }
    // `--seed <n>` with no count — the only number is the seed value,
    // not the count.
    match parse_command("seed People --seed 7").expect("--seed without count parses") {
        Command::Seed {
            count, rng_seed, ..
        } => {
            assert_eq!(count, None, "no positional count");
            assert_eq!(rng_seed, Some(7));
        }
        other => panic!("expected Command::Seed, got {other:?}"),
    }
}

// — Phase 2 (SD2): set-clause + column-fill parse path (ADR-0048 D2/D1) —

use rdbms_playground::dsl::command::{SeedOverride, SeedOverrideKind};
use rdbms_playground::dsl::value::Value;

/// Pull the `overrides` out of a parsed `seed` command (panics on a
/// non-seed command), for the builder-fold assertions below.
fn seed_overrides(input: &str) -> (Option<String>, Vec<SeedOverride>) {
    match parse_command(input).unwrap_or_else(|e| panic!("`{input}` should parse: {e:?}")) {
        Command::Seed {
            target_column,
            overrides,
            ..
        } => (target_column, overrides),
        other => panic!("expected Command::Seed, got {other:?}"),
    }
}

#[test]
fn seed_set_fixed_value_override_parses() {
    let (_t, ov) = seed_overrides("seed users 5 set status = 'active'");
    assert_eq!(ov.len(), 1);
    assert_eq!(ov[0].column, "status");
    assert_eq!(ov[0].kind, SeedOverrideKind::Fixed(Value::Text("active".into())));
}

#[test]
fn seed_set_pick_list_override_parses() {
    let (_t, ov) = seed_overrides("seed users set role in ('admin', 'editor', 'viewer')");
    assert_eq!(ov.len(), 1);
    assert_eq!(ov[0].column, "role");
    assert_eq!(
        ov[0].kind,
        SeedOverrideKind::PickList(vec![
            Value::Text("admin".into()),
            Value::Text("editor".into()),
            Value::Text("viewer".into()),
        ])
    );
}

#[test]
fn seed_set_generator_override_parses() {
    let (_t, ov) = seed_overrides("seed users set work_addr as email");
    assert_eq!(ov.len(), 1);
    assert_eq!(ov[0].column, "work_addr");
    assert_eq!(ov[0].kind, SeedOverrideKind::Generator("email".into()));
}

#[test]
fn seed_set_numeric_range_override_parses() {
    let (_t, ov) = seed_overrides("seed products set price between 10 and 100");
    assert_eq!(ov.len(), 1);
    assert_eq!(ov[0].column, "price");
    assert_eq!(
        ov[0].kind,
        SeedOverrideKind::Range {
            low: Value::Number("10".into()),
            high: Value::Number("100".into()),
        }
    );
}

#[test]
fn seed_set_date_range_override_parses_with_quoted_dates() {
    // ADR-0048 D2 amendment: dates in the range form are quoted strings.
    let (_t, ov) =
        seed_overrides("seed users set signup between '2023-01-01' and '2024-12-31'");
    assert_eq!(
        ov[0].kind,
        SeedOverrideKind::Range {
            low: Value::Text("2023-01-01".into()),
            high: Value::Text("2024-12-31".into()),
        }
    );
}

#[test]
fn seed_multiple_overrides_combine() {
    let (_t, ov) = seed_overrides(
        "seed users 20 set role in ('admin', 'user'), status = 'active', signup between '2023-01-01' and '2024-12-31'",
    );
    assert_eq!(ov.len(), 3, "three comma-separated overrides: {ov:?}");
    assert_eq!(ov[0].column, "role");
    assert!(matches!(ov[0].kind, SeedOverrideKind::PickList(_)));
    assert_eq!(ov[1].column, "status");
    assert!(matches!(ov[1].kind, SeedOverrideKind::Fixed(_)));
    assert_eq!(ov[2].column, "signup");
    assert!(matches!(ov[2].kind, SeedOverrideKind::Range { .. }));
}

#[test]
fn seed_count_is_not_confused_by_a_range_value() {
    // No positional count, but `between 18 and 80` carries NumberLits —
    // they must not be read as the count (bounded to before `set`).
    match parse_command("seed users set age between 18 and 80").expect("parses") {
        Command::Seed { count, overrides, .. } => {
            assert_eq!(count, None, "the count is None, not 18");
            assert_eq!(overrides.len(), 1);
        }
        other => panic!("expected seed, got {other:?}"),
    }
}

#[test]
fn seed_set_combines_with_count_and_flag() {
    match parse_command("seed users 30 set status = 'x' --seed 42").expect("parses") {
        Command::Seed {
            count,
            overrides,
            rng_seed,
            ..
        } => {
            assert_eq!(count, Some(30));
            assert_eq!(rng_seed, Some(42));
            assert_eq!(overrides.len(), 1);
        }
        other => panic!("expected seed, got {other:?}"),
    }
}

#[test]
fn seed_column_fill_target_parses() {
    let (target, ov) = seed_overrides("seed users.work_addr");
    assert_eq!(target.as_deref(), Some("work_addr"));
    assert!(ov.is_empty());
}

#[test]
fn seed_column_fill_with_set_parses() {
    let (target, ov) = seed_overrides("seed users.work_addr set work_addr as email");
    assert_eq!(target.as_deref(), Some("work_addr"));
    assert_eq!(ov.len(), 1);
    assert_eq!(ov[0].kind, SeedOverrideKind::Generator("email".into()));
}

#[test]
fn seed_bare_word_set_value_is_rejected() {
    // A bare (unquoted) word is not a value — D2 requires quoting. The
    // typed value slot rejects `active` at the grammar level (it is not a
    // quoted string / number), so the command does not parse.
    assert!(
        parse_command("seed users set status = active").is_err(),
        "a bare-word `set` value must be rejected (quoting required, D2)"
    );
    // The quoted form parses.
    assert!(parse_command("seed users set status = 'active'").is_ok());
}

#[test]
fn seed_populates_a_table_and_persists_rows() {
    let (project, db, _dir) = open_project_db();
    let rt = rt();
    create_people(&db, &rt);

    let result = rt
        .block_on(db.seed("People".into(), None, Some(7), Vec::new(), Some(42), Some("seed People 7".into())))
        .expect("seed succeeds");
    assert_eq!(result.produced, 7);

    let csv = read_csv(&project, "People").expect("People CSV exists after seed");
    assert_eq!(
        data_row_count(&csv),
        7,
        "CSV should hold 7 generated rows:\n{csv}"
    );
    // The generated `email` column produces address-shaped values.
    assert!(csv.contains('@'), "seeded emails should appear in the CSV:\n{csv}");
}

#[test]
fn seed_count_defaults_to_twenty() {
    let (project, db, _dir) = open_project_db();
    let rt = rt();
    create_people(&db, &rt);

    let result = rt
        .block_on(db.seed("People".into(), None, None, Vec::new(), Some(1), Some("seed People".into())))
        .expect("seed succeeds");
    assert_eq!(result.produced, 20, "omitted count defaults to 20");
    let csv = read_csv(&project, "People").expect("People CSV exists");
    assert_eq!(data_row_count(&csv), 20);
}

#[test]
fn seed_is_reproducible_with_a_fixed_seed() {
    let (p1, db1, _d1) = open_project_db();
    let (p2, db2, _d2) = open_project_db();
    let rt = rt();
    create_people(&db1, &rt);
    create_people(&db2, &rt);

    rt.block_on(db1.seed("People".into(), None, Some(4), Vec::new(), Some(123), Some("seed People 4".into())))
        .expect("seed run 1");
    rt.block_on(db2.seed("People".into(), None, Some(4), Vec::new(), Some(123), Some("seed People 4".into())))
        .expect("seed run 2");

    let csv1 = read_csv(&p1, "People").expect("csv 1");
    let csv2 = read_csv(&p2, "People").expect("csv 2");
    assert_eq!(csv1, csv2, "the same --seed must reproduce identical data");
}

#[test]
fn seed_writes_exactly_one_history_line() {
    let (project, db, _dir) = open_project_db();
    let rt = rt();
    create_people(&db, &rt);

    rt.block_on(db.seed("People".into(), None, Some(5), Vec::new(), Some(1), Some("seed People 5".into())))
        .expect("seed succeeds");

    let history = std::fs::read_to_string(project.path().join("history.log"))
        .expect("history.log exists");
    let seed_lines = history.lines().filter(|l| l.contains("seed People 5")).count();
    assert_eq!(
        seed_lines, 1,
        "a seed of 5 rows must write exactly one history line:\n{history}"
    );
}

// — FK sampling, empty-parent error, block guard (ADR-0048 D14 / D1) —

/// `Users(id serial pk, name text)` + `Orders(id serial pk, user_id
/// int, total decimal)` with `Orders.user_id -> Users.id`.
fn create_users_and_orders(db: &Database, rt: &tokio::runtime::Runtime, add_fk: bool) {
    rt.block_on(async {
        db.create_table(
            "Users".to_string(),
            vec![
                ColumnSpec::new("id", Type::Serial),
                ColumnSpec::new("name", Type::Text),
            ],
            vec!["id".to_string()],
            None,
        )
        .await
        .expect("create Users");
        db.create_table(
            "Orders".to_string(),
            vec![
                ColumnSpec::new("id", Type::Serial),
                ColumnSpec::new("user_id", Type::Int),
                ColumnSpec::new("total", Type::Decimal),
            ],
            vec!["id".to_string()],
            None,
        )
        .await
        .expect("create Orders");
        if add_fk {
            db.add_relationship(
                None,
                "Users".to_string(),
                vec!["id".to_string()],
                "Orders".to_string(),
                vec!["user_id".to_string()],
                ReferentialAction::NoAction,
                ReferentialAction::NoAction,
                false,
                None,
            )
            .await
            .expect("add Orders->Users FK");
        }
    });
}

/// `user_id` is column index 1 of `Orders(id, user_id, total)`.
fn order_user_ids(csv: &str) -> Vec<String> {
    let mut lines = csv.lines().filter(|l| !l.trim().is_empty());
    lines.next(); // header
    lines
        .map(|l| l.split(',').nth(1).unwrap_or_default().to_string())
        .collect()
}

#[test]
fn seed_fills_foreign_keys_from_existing_parents() {
    let (project, db, _dir) = open_project_db();
    let rt = rt();
    create_users_and_orders(&db, &rt, true);

    // 5 parents → serial ids 1..=5.
    rt.block_on(db.seed("Users".into(), None, Some(5), Vec::new(), Some(1), Some("seed Users 5".into())))
        .expect("seed Users");
    let res = rt
        .block_on(db.seed("Orders".into(), None, Some(10), Vec::new(), Some(2), Some("seed Orders 10".into())))
        .expect("seed Orders");
    assert_eq!(res.produced, 10, "every child row must insert (valid FK)");

    let csv = read_csv(&project, "Orders").expect("Orders CSV");
    let valid: std::collections::HashSet<String> = (1..=5).map(|i| i.to_string()).collect();
    let user_ids = order_user_ids(&csv);
    assert_eq!(user_ids.len(), 10);
    for uid in &user_ids {
        assert!(
            valid.contains(uid),
            "user_id `{uid}` does not reference an existing parent:\n{csv}"
        );
    }
}

#[test]
fn seed_refuses_when_a_parent_table_is_empty() {
    let (_project, db, _dir) = open_project_db();
    let rt = rt();
    create_users_and_orders(&db, &rt, true);

    // Users is empty — no valid FK can be fabricated.
    let err = rt
        .block_on(db.seed("Orders".into(), None, Some(3), Vec::new(), Some(1), Some("seed Orders 3".into())))
        .expect_err("seed must refuse an empty parent");
    let msg = err.to_string();
    assert!(msg.contains("Users"), "error should name the empty parent: {msg}");
    let lower = msg.to_lowercase();
    assert!(
        lower.contains("no rows") || lower.contains("first"),
        "error should explain how to fix it: {msg}"
    );
}

#[test]
fn seed_refuses_a_not_null_blob_column() {
    let (_project, db, _dir) = open_project_db();
    let rt = rt();
    let mut payload = ColumnSpec::new("payload", Type::Blob);
    payload.not_null = true;
    rt.block_on(db.create_table(
        "Files".to_string(),
        vec![ColumnSpec::new("id", Type::Serial), payload],
        vec!["id".to_string()],
        None,
    ))
    .expect("create Files");

    let err = rt
        .block_on(db.seed("Files".into(), None, Some(2), Vec::new(), Some(1), Some("seed Files 2".into())))
        .expect_err("seed must refuse a NOT NULL blob");
    let msg = err.to_string();
    assert!(
        msg.contains("payload") && msg.to_lowercase().contains("blob"),
        "error should name the un-generatable blob column: {msg}"
    );
}

#[test]
fn seed_omits_a_nullable_blob_column() {
    let (project, db, _dir) = open_project_db();
    let rt = rt();
    rt.block_on(db.create_table(
        "Files".to_string(),
        vec![
            ColumnSpec::new("id", Type::Serial),
            ColumnSpec::new("name", Type::Text),
            // nullable blob → omitted (→ NULL), seed still succeeds.
            ColumnSpec::new("payload", Type::Blob),
        ],
        vec!["id".to_string()],
        None,
    ))
    .expect("create Files");

    let res = rt
        .block_on(db.seed("Files".into(), None, Some(3), Vec::new(), Some(1), Some("seed Files 3".into())))
        .expect("seed succeeds despite the nullable blob");
    assert_eq!(res.produced, 3);
    let csv = read_csv(&project, "Files").expect("Files CSV");
    assert_eq!(data_row_count(&csv), 3);
}

// — uniqueness, junction distinct-combos, IN-CHECK (D10 / D14 / D17) —

/// The `n`th comma-separated field of each data row (the generated
/// values here never contain commas).
fn nth_column_values(csv: &str, n: usize) -> Vec<String> {
    csv.lines()
        .filter(|l| !l.trim().is_empty())
        .skip(1)
        .map(|l| l.split(',').nth(n).unwrap_or_default().trim().to_string())
        .collect()
}

#[test]
fn seed_keeps_unique_columns_distinct() {
    let (project, db, _dir) = open_project_db();
    let rt = rt();
    let mut label = ColumnSpec::new("label", Type::Text);
    label.unique = true;
    rt.block_on(db.create_table(
        "Tags".to_string(),
        vec![ColumnSpec::new("id", Type::Serial), label],
        vec!["id".to_string()],
        None,
    ))
    .expect("create Tags");

    let res = rt
        .block_on(db.seed("Tags".into(), None, Some(8), Vec::new(), Some(3), Some("seed Tags 8".into())))
        .expect("seed");
    assert_eq!(res.produced, 8);

    let csv = read_csv(&project, "Tags").expect("Tags CSV");
    let labels = nth_column_values(&csv, 1);
    let distinct: std::collections::HashSet<&String> = labels.iter().collect();
    assert_eq!(distinct.len(), labels.len(), "UNIQUE column has duplicates:\n{csv}");
}

#[test]
fn seed_sequences_identifier_int_columns() {
    let (project, db, _dir) = open_project_db();
    let rt = rt();
    // `code` is an identifier-named int (D10) but not a constraint —
    // uniqueness comes from the identifier rule.
    rt.block_on(db.create_table(
        "Items".to_string(),
        vec![
            ColumnSpec::new("id", Type::Serial),
            ColumnSpec::new("code", Type::Int),
            ColumnSpec::new("name", Type::Text),
        ],
        vec!["id".to_string()],
        None,
    ))
    .expect("create Items");

    let res = rt
        .block_on(db.seed("Items".into(), None, Some(5), Vec::new(), Some(1), Some("seed Items 5".into())))
        .expect("seed");
    assert_eq!(res.produced, 5);

    let csv = read_csv(&project, "Items").expect("Items CSV");
    let codes: Vec<i64> = nth_column_values(&csv, 1)
        .iter()
        .map(|s| s.parse().expect("code is an int"))
        .collect();
    let distinct: std::collections::HashSet<i64> = codes.iter().copied().collect();
    assert_eq!(distinct.len(), 5, "identifier ints must be unique: {codes:?}");
}

#[test]
fn seed_junction_produces_distinct_combinations_and_caps() {
    let (project, db, _dir) = open_project_db();
    let rt = rt();
    rt.block_on(async {
        // Two parents, 2 rows each → 2x2 = 4 possible (a, b) pairs.
        for t in ["P1", "P2"] {
            db.create_table(
                t.to_string(),
                vec![
                    ColumnSpec::new("id", Type::Serial),
                    ColumnSpec::new("name", Type::Text),
                ],
                vec!["id".to_string()],
                None,
            )
            .await
            .expect("create parent");
            db.seed(t.into(), None, Some(2), Vec::new(), Some(1), Some(format!("seed {t} 2")))
                .await
                .expect("seed parent");
        }
        // Junction with a compound PK over its two FK columns.
        db.create_table(
            "J".to_string(),
            vec![ColumnSpec::new("a", Type::Int), ColumnSpec::new("b", Type::Int)],
            vec!["a".to_string(), "b".to_string()],
            None,
        )
        .await
        .expect("create J");
        db.add_relationship(
            None,
            "P1".into(),
            vec!["id".into()],
            "J".into(),
            vec!["a".into()],
            ReferentialAction::NoAction,
            ReferentialAction::NoAction,
            false,
            None,
        )
        .await
        .expect("fk a");
        db.add_relationship(
            None,
            "P2".into(),
            vec!["id".into()],
            "J".into(),
            vec!["b".into()],
            ReferentialAction::NoAction,
            ReferentialAction::NoAction,
            false,
            None,
        )
        .await
        .expect("fk b");

        // Requesting 10 caps at the 4 available distinct combinations.
        let res = db
            .seed("J".into(), None, Some(10), Vec::new(), Some(7), Some("seed J 10".into()))
            .await
            .expect("seed J");
        assert_eq!(res.produced, 4, "junction caps at available combos");
        assert_eq!(res.requested, 10, "the requested count is reported for the cap note");
    });

    let csv = read_csv(&project, "J").expect("J CSV");
    let pairs: Vec<String> = csv
        .lines()
        .filter(|l| !l.trim().is_empty())
        .skip(1)
        .map(str::to_string)
        .collect();
    let distinct: std::collections::HashSet<&String> = pairs.iter().collect();
    assert_eq!(distinct.len(), pairs.len(), "junction rows must be distinct:\n{csv}");
}

#[test]
fn seed_draws_enum_values_from_an_in_check() {
    let (project, db, _dir) = open_project_db();
    let rt = rt();
    let mut status = ColumnSpec::new("status", Type::Text);
    status.check_sql = Some("status IN ('active', 'closed')".to_string());
    rt.block_on(db.create_table(
        "Tickets".to_string(),
        vec![ColumnSpec::new("id", Type::Serial), status],
        vec!["id".to_string()],
        None,
    ))
    .expect("create Tickets");

    // Every generated status must satisfy the CHECK, so all rows insert.
    let res = rt
        .block_on(db.seed("Tickets".into(), None, Some(12), Vec::new(), Some(2), Some("seed Tickets 12".into())))
        .expect("seed");
    assert_eq!(res.produced, 12, "all rows insert — values satisfy the CHECK");

    let csv = read_csv(&project, "Tickets").expect("Tickets CSV");
    for v in nth_column_values(&csv, 1) {
        assert!(
            matches!(v.as_str(), "active" | "closed"),
            "status `{v}` was not drawn from the IN check:\n{csv}"
        );
    }
    // The IN-check column is derived, not generic, so it is NOT flagged.
    assert!(
        res.advisory_columns.is_empty(),
        "an IN-check column should not be flagged: {:?}",
        res.advisory_columns
    );
}

#[test]
fn seed_advises_on_enum_ish_columns() {
    let (_project, db, _dir) = open_project_db();
    let rt = rt();
    // `status` has no CHECK and no name heuristic → generic text, so it
    // is flagged for the advisory (D12/D13).
    rt.block_on(db.create_table(
        "Tasks".to_string(),
        vec![
            ColumnSpec::new("id", Type::Serial),
            ColumnSpec::new("status", Type::Text),
        ],
        vec!["id".to_string()],
        None,
    ))
    .expect("create Tasks");

    let res = rt
        .block_on(db.seed("Tasks".into(), None, Some(3), Vec::new(), Some(1), Some("seed Tasks 3".into())))
        .expect("seed");
    assert!(
        res.advisory_columns.contains(&"status".to_string()),
        "enum-ish `status` should be flagged: {:?}",
        res.advisory_columns
    );
}

#[test]
fn seed_refuses_an_excessive_count() {
    let (_project, db, _dir) = open_project_db();
    let rt = rt();
    create_people(&db, &rt);
    let err = rt
        .block_on(db.seed("People".into(), None, Some(1_000_000), Vec::new(), Some(1), Some("seed People 1000000".into())))
        .expect_err("an excessive count must be refused");
    assert!(
        err.to_string().to_lowercase().contains("maximum"),
        "error should mention the maximum: {err}"
    );
}

#[test]
fn seed_preview_is_capped_but_count_is_full() {
    let (_project, db, _dir) = open_project_db();
    let rt = rt();
    create_people(&db, &rt);

    let res = rt
        .block_on(db.seed("People".into(), None, Some(25), Vec::new(), Some(1), Some("seed People 25".into())))
        .expect("seed");
    assert_eq!(res.produced, 25, "the full count is produced");
    assert_eq!(res.data.rows.len(), 20, "the preview is capped at 20 rows");
}

#[test]
fn seed_is_available_in_advanced_mode() {
    use rdbms_playground::dsl::parser::parse_command_in_mode;
    use rdbms_playground::mode::Mode;
    // D5/A1: seed is a canonical command available in BOTH modes.
    let r = parse_command_in_mode("seed People 5", Mode::Advanced);
    assert!(
        matches!(r, Ok(Command::Seed { .. })),
        "seed must parse in advanced mode: {r:?}"
    );
    // The Phase 2 surfaces (set clause + column-fill) also parse in
    // advanced mode — same grammar, no mode gate.
    assert!(
        matches!(
            parse_command_in_mode("seed People 5 set status = 'active'", Mode::Advanced),
            Ok(Command::Seed { .. })
        ),
        "set clause must parse in advanced mode"
    );
    assert!(
        matches!(
            parse_command_in_mode("seed People.email set email as email", Mode::Advanced),
            Ok(Command::Seed {
                target_column: Some(_),
                ..
            })
        ),
        "column-fill must parse in advanced mode"
    );
}

// — DA-pass coverage: undo (D15), replay (D16), atomicity, zero count,
//   complex-CHECK advisory (D17), FK reproducibility (D4) —

#[test]
fn seed_is_one_undo_step() {
    // Undo must be explicitly enabled on the Database.
    let dir = tempfile::tempdir().expect("tempdir");
    let project = project::open_or_create(None, Some(dir.path())).expect("project");
    let persistence = Persistence::new(project.path().to_path_buf());
    let db = Database::open_with_persistence_and_undo(project.db_path(), persistence, true)
        .expect("open db with undo");
    let rt = rt();
    create_people(&db, &rt);
    rt.block_on(db.seed("People".into(), None, Some(6), Vec::new(), Some(1), Some("seed People 6".into())))
        .expect("seed");
    assert_eq!(data_row_count(&read_csv(&project, "People").unwrap()), 6);

    // One undo removes the whole seed batch (ADR-0048 D15).
    rt.block_on(db.undo()).unwrap().expect("undo applied");
    let rows = read_csv(&project, "People").map_or(0, |c| data_row_count(&c));
    assert_eq!(rows, 0, "one undo must remove every seeded row in a single step");
}

#[test]
fn seed_column_fill_is_one_undo_step() {
    // ADR-0048 D15: column-fill's bulk UPDATE is one undo step too.
    let dir = tempfile::tempdir().expect("tempdir");
    let project = project::open_or_create(None, Some(dir.path())).expect("project");
    let persistence = Persistence::new(project.path().to_path_buf());
    let db = Database::open_with_persistence_and_undo(project.db_path(), persistence, true)
        .expect("open db with undo");
    let rt = rt();
    create_members(&db, &rt);
    run_seed(&db, &rt, "seed Members 5 --seed 1").expect("seed");
    // Fill `status` across all 5 rows with a constant, then undo once.
    run_seed(&db, &rt, "seed Members.status set status = 'flagged' --seed 2")
        .expect("column-fill");
    let before = named_column_values(&read_csv(&project, "Members").unwrap(), "status");
    assert!(before.iter().all(|s| s == "flagged"), "all rows filled: {before:?}");

    rt.block_on(db.undo()).unwrap().expect("undo applied");
    let after = named_column_values(&read_csv(&project, "Members").unwrap(), "status");
    assert!(
        after.iter().all(|s| s != "flagged"),
        "one undo reverts the whole column-fill in a single step: {after:?}"
    );
    assert_eq!(after.len(), 5, "undo restores the original rows, not removes them");
}

#[test]
fn replay_reruns_a_seed_line_as_a_data_write() {
    use rdbms_playground::runtime::run_replay;
    let (project, db, _dir) = open_project_db();
    let rt = rt();
    create_people(&db, &rt);
    std::fs::write(project.path().join("seed.script"), "seed People 5\n").expect("write script");

    // D16: seed is a data-write — replay re-runs it (it is NOT in the
    // app-lifecycle skip-list), so the rows appear.
    let _events = rt.block_on(run_replay(&db, project.path(), "seed.script"));
    assert_eq!(
        data_row_count(&read_csv(&project, "People").unwrap()),
        5,
        "replay must re-run the seed line"
    );
}

#[test]
fn seed_rolls_back_atomically_on_a_constraint_failure() {
    let (project, db, _dir) = open_project_db();
    let rt = rt();
    // A CHECK that generic text cannot satisfy → every generated row
    // violates it, so the whole batch must roll back (P1.3d atomicity).
    let mut code = ColumnSpec::new("note", Type::Text);
    code.check_sql = Some("length(note) > 100".to_string());
    rt.block_on(db.create_table(
        "Bad".to_string(),
        vec![ColumnSpec::new("id", Type::Serial), code],
        vec!["id".to_string()],
        None,
    ))
    .expect("create Bad");

    let res = rt.block_on(db.seed("Bad".into(), None, Some(5), Vec::new(), Some(1), Some("seed Bad 5".into())));
    assert!(res.is_err(), "seed must fail when generated rows violate the CHECK");
    let rows = read_csv(&project, "Bad").map_or(0, |c| data_row_count(&c));
    assert_eq!(rows, 0, "a failed seed must leave the table unchanged (atomic)");
}

#[test]
fn seed_zero_is_a_no_op() {
    let (project, db, _dir) = open_project_db();
    let rt = rt();
    create_people(&db, &rt);
    let res = rt
        .block_on(db.seed("People".into(), None, Some(0), Vec::new(), Some(1), Some("seed People 0".into())))
        .expect("seed 0 succeeds");
    assert_eq!(res.produced, 0);
    let rows = read_csv(&project, "People").map_or(0, |c| data_row_count(&c));
    assert_eq!(rows, 0, "seed 0 inserts nothing");
}

#[test]
fn seed_advises_on_a_complex_check_column() {
    let (_project, db, _dir) = open_project_db();
    let rt = rt();
    // A complex (non-IN) CHECK seed can't derive values from → the
    // column is filled generically AND flagged (D17/D13). `length` keeps
    // generic words valid so the seed still succeeds.
    let mut label = ColumnSpec::new("label", Type::Text);
    label.check_sql = Some("length(label) >= 1".to_string());
    rt.block_on(db.create_table(
        "Widgets".to_string(),
        vec![ColumnSpec::new("id", Type::Serial), label],
        vec!["id".to_string()],
        None,
    ))
    .expect("create Widgets");

    let res = rt
        .block_on(db.seed("Widgets".into(), None, Some(3), Vec::new(), Some(1), Some("seed Widgets 3".into())))
        .expect("seed");
    assert!(
        res.advisory_columns.contains(&"label".to_string()),
        "a column with an underivable CHECK should be flagged: {:?}",
        res.advisory_columns
    );
}

#[test]
fn seed_foreign_keys_are_reproducible_with_a_fixed_seed() {
    let rt = rt();
    let seed_one = |db: &Database| {
        create_users_and_orders(db, &rt, true);
        rt.block_on(db.seed("Users".into(), None, Some(4), Vec::new(), Some(1), Some("seed Users 4".into())))
            .expect("seed users");
        rt.block_on(db.seed("Orders".into(), None, Some(8), Vec::new(), Some(99), Some("seed Orders 8".into())))
            .expect("seed orders");
    };
    let (p1, db1, _d1) = open_project_db();
    let (p2, db2, _d2) = open_project_db();
    seed_one(&db1);
    seed_one(&db2);
    // With ORDER BY on the FK sample, the same --seed reproduces the
    // sampled FK values (D4).
    assert_eq!(
        read_csv(&p1, "Orders").unwrap(),
        read_csv(&p2, "Orders").unwrap(),
        "FK sampling must be reproducible with a fixed --seed"
    );
}

#[test]
fn seed_shortid_columns_are_reproducible_with_a_fixed_seed() {
    let rt = rt();
    let make = |db: &Database| {
        rt.block_on(db.create_table(
            "Contacts".to_string(),
            vec![
                ColumnSpec::new("code", Type::ShortId),
                ColumnSpec::new("name", Type::Text),
            ],
            vec!["code".to_string()],
            None,
        ))
        .expect("create Contacts");
        rt.block_on(db.seed("Contacts".into(), None, Some(5), Vec::new(), Some(42), Some("seed Contacts 5".into())))
            .expect("seed");
    };
    let (p1, db1, _d1) = open_project_db();
    let (p2, db2, _d2) = open_project_db();
    make(&db1);
    make(&db2);

    let csv1 = read_csv(&p1, "Contacts").unwrap();
    let csv2 = read_csv(&p2, "Contacts").unwrap();
    assert_eq!(csv1, csv2, "shortid values must reproduce under a fixed --seed");

    // The shortid PK is populated with distinct 10-char base58 ids.
    let codes = nth_column_values(&csv1, 0);
    assert_eq!(codes.len(), 5);
    let distinct: std::collections::HashSet<&String> = codes.iter().collect();
    assert_eq!(distinct.len(), 5, "shortid PK values must be distinct: {codes:?}");
    for code in &codes {
        assert_eq!(code.len(), 10, "shortid should be 10 chars: {code}");
    }
}

// =================================================================
// Phase 2 (SD2) executor: set-clause overrides + column-fill,
// exercised full-stack (parse → worker) — ADR-0048 D2 / D1.
// =================================================================

/// Parse `input` as a `seed` command and run it through the worker —
/// the full stack minus UI render (grammar → builder → executor).
fn run_seed(
    db: &Database,
    rt: &tokio::runtime::Runtime,
    input: &str,
) -> Result<rdbms_playground::db::SeedResult, rdbms_playground::db::DbError> {
    match parse_command(input).unwrap_or_else(|e| panic!("`{input}` should parse: {e:?}")) {
        Command::Seed {
            table,
            target_column,
            count,
            overrides,
            rng_seed,
        } => rt.block_on(db.seed(
            table,
            target_column,
            count,
            overrides,
            rng_seed,
            Some(input.to_string()),
        )),
        other => panic!("expected a seed command, got {other:?}"),
    }
}

/// Values of the column named `col` (by header lookup) across the CSV's
/// data rows.
fn named_column_values(csv: &str, col: &str) -> Vec<String> {
    let header = csv.lines().next().unwrap_or_default();
    let idx = header
        .split(',')
        .position(|h| h.trim() == col)
        .unwrap_or_else(|| panic!("column `{col}` not in header `{header}`"));
    nth_column_values(csv, idx)
}

/// `Members(id serial pk, name text, status text, role text, age int)`.
/// `status`/`role` are enum-ish names (advisory targets without an
/// override); `name`/`age` exercise the generator / range overrides.
fn create_members(db: &Database, rt: &tokio::runtime::Runtime) {
    rt.block_on(db.create_table(
        "Members".to_string(),
        vec![
            ColumnSpec::new("id", Type::Serial),
            ColumnSpec::new("name", Type::Text),
            ColumnSpec::new("status", Type::Text),
            ColumnSpec::new("role", Type::Text),
            ColumnSpec::new("age", Type::Int),
        ],
        vec!["id".to_string()],
        None,
    ))
    .expect("create Members");
}

#[test]
fn seed_set_fixed_value_fills_every_row() {
    let (project, db, _d) = open_project_db();
    let rt = rt();
    create_members(&db, &rt);
    run_seed(&db, &rt, "seed Members 6 set status = 'active' --seed 1").expect("seed");
    let csv = read_csv(&project, "Members").unwrap();
    let statuses = named_column_values(&csv, "status");
    assert_eq!(statuses.len(), 6);
    assert!(statuses.iter().all(|s| s == "active"), "every status pinned: {statuses:?}");
}

#[test]
fn seed_set_pick_list_draws_only_from_the_list() {
    let (project, db, _d) = open_project_db();
    let rt = rt();
    create_members(&db, &rt);
    run_seed(&db, &rt, "seed Members 20 set role in ('admin', 'user') --seed 2").expect("seed");
    let csv = read_csv(&project, "Members").unwrap();
    let roles = named_column_values(&csv, "role");
    assert!(
        roles.iter().all(|r| r == "admin" || r == "user"),
        "roles only from the list: {roles:?}"
    );
}

#[test]
fn seed_set_as_generator_forces_the_shape() {
    let (project, db, _d) = open_project_db();
    let rt = rt();
    create_members(&db, &rt);
    // Force the `name` column (a person-name heuristic) to emails.
    run_seed(&db, &rt, "seed Members 5 set name as email --seed 3").expect("seed");
    let csv = read_csv(&project, "Members").unwrap();
    let names = named_column_values(&csv, "name");
    assert!(names.iter().all(|n| n.contains('@')), "name forced to email shape: {names:?}");
}

#[test]
fn seed_set_numeric_range_stays_within_bounds() {
    let (project, db, _d) = open_project_db();
    let rt = rt();
    create_members(&db, &rt);
    run_seed(&db, &rt, "seed Members 30 set age between 30 and 40 --seed 4").expect("seed");
    let csv = read_csv(&project, "Members").unwrap();
    for a in named_column_values(&csv, "age") {
        let n: i64 = a.parse().unwrap_or_else(|_| panic!("age `{a}` not an int"));
        assert!((30..=40).contains(&n), "age {n} out of [30,40]");
    }
}

#[test]
fn seed_override_drops_the_column_from_the_advisory() {
    let (_p, db, _d) = open_project_db();
    let rt = rt();
    create_members(&db, &rt);
    // Without an override, `status` (enum-ish) is flagged in the advisory.
    let plain = run_seed(&db, &rt, "seed Members 3 --seed 5").expect("seed");
    assert!(
        plain.advisory_columns.iter().any(|c| c == "status"),
        "status should be advised without an override: {:?}",
        plain.advisory_columns
    );
    // With an override on status, it must not appear in the advisory.
    let overridden =
        run_seed(&db, &rt, "seed Members 3 set status in ('a', 'b') --seed 5").expect("seed");
    assert!(
        !overridden.advisory_columns.iter().any(|c| c == "status"),
        "overridden status must drop from advisory: {:?}",
        overridden.advisory_columns
    );
}

#[test]
fn seed_unknown_generator_is_a_friendly_error() {
    let (_p, db, _d) = open_project_db();
    let rt = rt();
    create_members(&db, &rt);
    let err = run_seed(&db, &rt, "seed Members 3 set name as bogus").unwrap_err();
    let msg = format!("{err}");
    assert!(
        msg.contains("unknown generator") && msg.contains("bogus"),
        "should name the unknown generator: {msg}"
    );
}

#[test]
fn seed_incompatible_range_is_a_friendly_error() {
    let (_p, db, _d) = open_project_db();
    let rt = rt();
    create_members(&db, &rt);
    // A numeric range on a text column (`name`) is rejected.
    let err = run_seed(&db, &rt, "seed Members 3 set name between 1 and 10").unwrap_err();
    let msg = format!("{err}");
    assert!(msg.contains("between"), "range error should mention `between`: {msg}");
}

#[test]
fn seed_with_set_is_reproducible() {
    let (p1, db1, _d1) = open_project_db();
    let (p2, db2, _d2) = open_project_db();
    let rt = rt();
    create_members(&db1, &rt);
    create_members(&db2, &rt);
    let cmd = "seed Members 10 set role in ('a', 'b', 'c'), age between 20 and 60 --seed 77";
    run_seed(&db1, &rt, cmd).expect("seed 1");
    run_seed(&db2, &rt, cmd).expect("seed 2");
    assert_eq!(
        read_csv(&p1, "Members").unwrap(),
        read_csv(&p2, "Members").unwrap(),
        "the same --seed + set clause must reproduce identical data"
    );
}

// — column-fill (ADR-0048 D1 form 2) —

#[test]
fn seed_column_fill_updates_existing_rows_without_adding() {
    let (project, db, _d) = open_project_db();
    let rt = rt();
    create_members(&db, &rt);
    run_seed(&db, &rt, "seed Members 5 --seed 1").expect("initial seed");
    let before = data_row_count(&read_csv(&project, "Members").unwrap());
    assert_eq!(before, 5);

    let res = run_seed(&db, &rt, "seed Members.status set status in ('x', 'y') --seed 2")
        .expect("column-fill");
    assert_eq!(res.produced, 5, "column-fill touches the 5 existing rows");
    let csv = read_csv(&project, "Members").unwrap();
    assert_eq!(data_row_count(&csv), 5, "no new rows added");
    let statuses = named_column_values(&csv, "status");
    assert!(
        statuses.iter().all(|s| s == "x" || s == "y"),
        "every existing row's status refilled from the list: {statuses:?}"
    );
}

#[test]
fn seed_column_fill_refuses_a_pk_target() {
    let (_p, db, _d) = open_project_db();
    let rt = rt();
    create_members(&db, &rt);
    run_seed(&db, &rt, "seed Members 3 --seed 1").expect("seed");
    let err = run_seed(&db, &rt, "seed Members.id").unwrap_err();
    assert!(format!("{err}").contains("primary key"), "PK target refused: {err}");
}

#[test]
fn seed_column_fill_empty_table_is_a_noop() {
    let (_p, db, _d) = open_project_db();
    let rt = rt();
    create_members(&db, &rt);
    // No rows yet → friendly no-op, not an error.
    let res = run_seed(&db, &rt, "seed Members.status set status in ('a', 'b')").expect("no-op");
    assert_eq!(res.produced, 0, "empty table → nothing filled");
}

#[test]
fn seed_column_fill_set_may_only_target_the_filled_column() {
    let (_p, db, _d) = open_project_db();
    let rt = rt();
    create_members(&db, &rt);
    run_seed(&db, &rt, "seed Members 3 --seed 1").expect("seed");
    let err = run_seed(&db, &rt, "seed Members.status set role = 'x'").unwrap_err();
    assert!(
        format!("{err}").contains("can only adjust"),
        "set targeting another column is refused: {err}"
    );
}

#[test]
fn seed_column_fill_rejects_a_row_count() {
    let (_p, db, _d) = open_project_db();
    let rt = rt();
    create_members(&db, &rt);
    // `seed T.col 5` parses, but a count is meaningless for column-fill.
    let err = rt
        .block_on(db.seed(
            "Members".into(),
            Some("status".into()),
            Some(5),
            Vec::new(),
            Some(1),
            Some("seed Members.status 5".into()),
        ))
        .unwrap_err();
    assert!(format!("{err}").contains("no row count"), "count refused: {err}");
}

#[test]
fn seed_column_fill_fk_target_samples_the_parent() {
    let (project, db, _d) = open_project_db();
    let rt = rt();
    create_users_and_orders(&db, &rt, true);
    run_seed(&db, &rt, "seed Users 4 --seed 1").expect("seed users");
    run_seed(&db, &rt, "seed Orders 8 --seed 2").expect("seed orders");
    // Re-fill the FK column across existing orders; every value must be a
    // valid parent key (the UPDATE would fail FK enforcement otherwise).
    let res = run_seed(&db, &rt, "seed Orders.user_id --seed 3").expect("column-fill FK");
    assert_eq!(res.produced, 8);
    let csv = read_csv(&project, "Orders").unwrap();
    let user_ids = named_column_values(&csv, "user_id");
    assert!(user_ids.iter().all(|v| (1..=4).contains(&v.parse::<i64>().unwrap())));
}

#[test]
fn seed_fixed_override_on_unique_column_is_a_friendly_error() {
    // DA finding (user-chosen: friendly error). A fixed value can't fill a
    // UNIQUE column for more than one row — refuse up front rather than
    // silently capping to 1.
    let (_p, db, _d) = open_project_db();
    let rt = rt();
    rt.block_on(db.create_table(
        "U".to_string(),
        vec![
            ColumnSpec::new("id", Type::Serial),
            {
                let mut c = ColumnSpec::new("email", Type::Text);
                c.unique = true;
                c
            },
        ],
        vec!["id".to_string()],
        None,
    ))
    .expect("create U");
    let err = run_seed(&db, &rt, "seed U 5 set email = 'x@y.com'").unwrap_err();
    let msg = format!("{err}");
    assert!(
        msg.contains("UNIQUE") && msg.contains("distinct"),
        "fixed value on a UNIQUE column should be a friendly capacity error: {msg}"
    );
    // A short pick-list (< count) is likewise refused...
    let err2 = run_seed(&db, &rt, "seed U 5 set email in ('a@b.c', 'd@e.f')").unwrap_err();
    assert!(format!("{err2}").contains("distinct"), "short list refused: {err2}");
    // ...but a pick-list with enough distinct values succeeds.
    let ok = run_seed(
        &db,
        &rt,
        "seed U 3 set email in ('a@b.c', 'd@e.f', 'g@h.i') --seed 1",
    )
    .expect("a list >= count fills cleanly");
    assert_eq!(ok.produced, 3);
    // A generator is unbounded — also fine.
    assert_eq!(
        run_seed(&db, &rt, "seed U 4 set email as email --seed 2")
            .expect("generator fills a unique column")
            .produced,
        4
    );
}

#[test]
fn seed_column_fill_fixed_on_unique_column_is_a_friendly_error() {
    let (_p, db, _d) = open_project_db();
    let rt = rt();
    rt.block_on(db.create_table(
        "U".to_string(),
        vec![
            ColumnSpec::new("id", Type::Serial),
            {
                let mut c = ColumnSpec::new("email", Type::Text);
                c.unique = true;
                c
            },
        ],
        vec!["id".to_string()],
        None,
    ))
    .expect("create U");
    run_seed(&db, &rt, "seed U 4 set email as email --seed 1").expect("seed 4 rows");
    // Filling the UNIQUE column on 4 rows with one fixed value is refused.
    let err = run_seed(&db, &rt, "seed U.email set email = 'same@x.com'").unwrap_err();
    assert!(
        format!("{err}").contains("UNIQUE"),
        "column-fill of a fixed value on a UNIQUE column should refuse: {err}"
    );
}