//! Integration tests for the `replay <path>` command (U4).
//!
//! Exercises the runtime's `run_replay` directly rather than
//! booting a Tokio event loop — the inner replay logic is the
//! interesting unit, and `spawn_replay` is just the mpsc shim
//! around it.
//!
//! Covers (per handoff §A3):
//! - Happy path: 3-line file dispatches 3 commands, project
//!   state reflects the dispatched DDL/DML.
//! - Blank lines and `# comments` are skipped silently.
//! - Per-line failure: the runtime reports the line number of
//!   the offending entry and stops without dispatching the rest.
//!   Earlier successful commands are NOT rolled back.
//! - Empty file → ReplayCompleted with count 0.
//! - Missing file → ReplayFailed with line_number 0.
//! - Nested replay (`replay foo` inside the file being replayed)
//!   is refused with a clear message.
//! - history.log invariant: replaying a file produces the same
//!   per-command history entries as if the user had typed each
//!   line interactively.

use std::fs;
use std::path::Path;

use tokio::runtime::Runtime;

use rdbms_playground::db::Database;
use rdbms_playground::event::AppEvent;
use rdbms_playground::persistence::Persistence;
use rdbms_playground::project;
use rdbms_playground::runtime::run_replay;

fn rt() -> Runtime {
    tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .expect("tokio rt")
}

fn tempdir() -> tempfile::TempDir {
    tempfile::tempdir().expect("create tempdir")
}

/// Open a fresh project + persistence-wired database under
/// `data_root`, returning both. Used as the canonical test
/// harness — most tests only need to write a script file and
/// call `run_replay`.
fn open_project_db(data_root: &Path) -> (project::Project, Database) {
    let project = project::open_or_create(None, Some(data_root))
        .expect("open_or_create");
    let db = Database::open_with_persistence(
        project.db_path(),
        Persistence::new(project.path().to_path_buf()),
    )
    .expect("open db");
    (project, db)
}

fn write_script(project_path: &Path, name: &str, body: &str) {
    fs::write(project_path.join(name), body).expect("write script");
}

fn assert_completed(events: &[AppEvent], expected_count: usize) {
    let last = events.last().expect("at least one event");
    match last {
        AppEvent::ReplayCompleted { count, .. } => {
            assert_eq!(
                *count, expected_count,
                "ReplayCompleted count mismatch (events: {events:?})"
            );
        }
        other => panic!("expected ReplayCompleted, got {other:?}"),
    }
}

fn assert_failed_at(events: &[AppEvent], expected_line: usize) -> &AppEvent {
    let last = events.last().expect("at least one event");
    match last {
        AppEvent::ReplayFailed { line_number, .. } => {
            assert_eq!(
                *line_number, expected_line,
                "ReplayFailed line_number mismatch (events: {events:?})"
            );
            last
        }
        other => panic!("expected ReplayFailed, got {other:?}"),
    }
}

#[test]
fn replay_runs_advanced_sql_create_table_as_a_write() {
    // ADR-0035 §10: `create` is a schema-write entry word (not in the
    // ADR-0034 app-lifecycle skip set), so an advanced-mode SQL
    // `CREATE TABLE` line replays as a write — re-applied, not skipped
    // — and executes structurally (the table is rebuilt from the line).
    let data = tempdir();
    let (project, db) = open_project_db(data.path());
    write_script(
        project.path(),
        "ddl.commands",
        "create table Widget (id serial primary key, name text)\n\
         insert into Widget (name) values ('gadget')\n",
    );

    let events = rt().block_on(async { run_replay(&db, project.path(), "ddl.commands").await });
    assert_completed(&events, 2);

    // The SQL DDL line actually created the structural table…
    let desc = rt()
        .block_on(async { db.describe_table("Widget".to_string(), None).await })
        .expect("describe");
    let names: Vec<String> = desc.columns.iter().map(|c| c.name.clone()).collect();
    assert_eq!(names, vec!["id".to_string(), "name".to_string()]);
    // …and the following insert (serial id auto-filled) ran against it.
    let rows = rt()
        .block_on(async { db.query_data("Widget".to_string(), None, None, None).await })
        .expect("query")
        .rows;
    assert_eq!(rows.len(), 1);
}

#[test]
fn replay_three_lines_dispatches_three_commands() {
    let data = tempdir();
    let (project, db) = open_project_db(data.path());
    write_script(
        project.path(),
        "seed.commands",
        "create table T with pk id(int)\n\
         add column T: name (text)\n\
         insert into T (1, 'Alice')\n",
    );

    let events = rt().block_on(async {
        run_replay(&db, project.path(), "seed.commands").await
    });
    assert_completed(&events, 3);

    // The dispatched commands actually mutated state.
    let data_result = rt()
        .block_on(async { db.query_data("T".to_string(), None, None, None).await })
        .expect("query_data");
    assert_eq!(data_result.rows.len(), 1, "row inserted");
    assert_eq!(data_result.rows[0][1].as_deref(), Some("Alice"));
}

#[test]
fn replay_of_actual_history_log_runs_ok_commands_and_skips_err() {
    // ADR-0034 §3 + Problem 3 (handoff-34 §4): `replay history.log`
    // must work. The journal is the pipe format
    // `<iso8601>|<status>|<source>`; replay extracts `<source>`, runs
    // `ok` records, and skips `err` ones (like blank / `#` lines — a
    // skipped failure is not a replay failure).
    //
    // This is the ADR-0034 headline reproduction. It is RED before the
    // fix: today `run_replay` feeds the whole `2026-…|ok|…` line to the
    // parser, which dies on line 1 (the timestamp is not a command).
    let data = tempdir();
    let (project, db) = open_project_db(data.path());
    write_script(
        project.path(),
        "history.log",
        "2026-05-24T10:00:00Z|ok|create table T with pk id(int)\n\
         2026-05-24T10:00:01Z|ok|add column T: v (text)\n\
         2026-05-24T10:00:02Z|err|insert into T values (1, 2, 3, 4)\n\
         2026-05-24T10:00:03Z|ok|insert into T (id, v) values (1, 'alpha')\n",
    );

    let events =
        rt().block_on(async { run_replay(&db, project.path(), "history.log").await });
    // Three `ok` records replayed; the `err` record is skipped (not
    // counted, not a failure).
    assert_completed(&events, 3);

    let data_result = rt()
        .block_on(async { db.query_data("T".to_string(), None, None, None).await })
        .expect("query_data");
    assert_eq!(data_result.rows.len(), 1, "only the ok INSERT applied");
    assert_eq!(data_result.rows[0][1].as_deref(), Some("alpha"));
}

#[test]
fn replay_skips_app_lifecycle_commands_silently() {
    // ADR-0034: a real `history.log` contains app-lifecycle commands
    // (`save as` / `load` / `new` / `export` / `mode` / `rebuild` /
    // `undo` / `redo` …).
    // Replay skips them — they are session navigation, not schema/data
    // reconstruction, and the worker dispatch cannot run them (it would
    // panic on a parsed app command, or abort on the modal forms that
    // don't parse). These skip SILENTLY (no warning).
    let data = tempdir();
    let (project, db) = open_project_db(data.path());
    // Every silent-skip app-lifecycle form, including the modal forms
    // that don't parse on the command line (`save as` / `load` / `new`),
    // the bare incomplete form (`mode`), and the safety-critical `quit`
    // (a journalled quit must NOT quit during replay). None may abort;
    // none warns.
    write_script(
        project.path(),
        "history.log",
        "2026-05-24T10:00:00Z|ok|create table T with pk id(int)\n\
         2026-05-24T10:00:01Z|ok|save as backup\n\
         2026-05-24T10:00:02Z|ok|load other\n\
         2026-05-24T10:00:03Z|ok|new scratch\n\
         2026-05-24T10:00:04Z|ok|mode advanced\n\
         2026-05-24T10:00:05Z|ok|mode\n\
         2026-05-24T10:00:06Z|ok|messages verbose\n\
         2026-05-24T10:00:07Z|ok|export out.zip\n\
         2026-05-24T10:00:08Z|ok|rebuild\n\
         2026-05-24T10:00:09Z|ok|help\n\
         2026-05-24T10:00:10Z|ok|quit\n\
         2026-05-24T10:00:11Z|ok|undo\n\
         2026-05-24T10:00:12Z|ok|redo\n\
         2026-05-24T10:00:13Z|ok|add column T: v (text)\n\
         2026-05-24T10:00:14Z|ok|insert into T (id, v) values (1, 'alpha')\n",
    );
    let events =
        rt().block_on(async { run_replay(&db, project.path(), "history.log").await });
    // Three data/schema commands ran; every app-lifecycle line was
    // skipped silently (no panic, no abort, no warnings, no quit).
    match events.last().expect("event") {
        AppEvent::ReplayCompleted { count, warnings, .. } => {
            assert_eq!(*count, 3, "only the 3 write commands ran; events: {events:?}");
            assert!(warnings.is_empty(), "these skips are silent; got {warnings:?}");
        }
        other => panic!("expected ReplayCompleted, got {other:?}"),
    }
    let data_result = rt()
        .block_on(async { db.query_data("T".to_string(), None, None, None).await })
        .expect("query_data");
    assert!(
        data_result.columns.iter().any(|c| c == "v"),
        "the add-column line applied; columns: {:?}",
        data_result.columns,
    );
    assert_eq!(data_result.rows.len(), 1, "the insert applied");
    assert_eq!(data_result.rows[0][1].as_deref(), Some("alpha"));
}

#[test]
fn replay_skips_import_with_a_warning() {
    // ADR-0034: `import` is skipped like other app commands, but warns
    // — skipping it can leave the replayed state incomplete (the
    // imported data is not reconstructed).
    let data = tempdir();
    let (project, db) = open_project_db(data.path());
    write_script(
        project.path(),
        "history.log",
        "2026-05-24T10:00:00Z|ok|create table T with pk id(int)\n\
         2026-05-24T10:00:01Z|ok|import shared.zip as Imported\n",
    );
    let events =
        rt().block_on(async { run_replay(&db, project.path(), "history.log").await });
    match events.last().expect("event") {
        AppEvent::ReplayCompleted { count, warnings, .. } => {
            assert_eq!(*count, 1, "only the create ran; events: {events:?}");
            assert!(
                warnings.iter().any(|w| w.contains("import shared.zip")),
                "expected an import skip warning; got {warnings:?}",
            );
        }
        other => panic!("expected ReplayCompleted, got {other:?}"),
    }
}

#[test]
fn replay_skips_blank_lines_and_comments() {
    let data = tempdir();
    let (project, db) = open_project_db(data.path());
    write_script(
        project.path(),
        "seed.commands",
        "# this is a comment\n\
         \n\
         create table T with pk id(int)\n\
         \n\
         # another comment\n\
              # comment with leading whitespace\n\
         add column T: name (text)\n\
         \n",
    );

    let events = rt().block_on(async {
        run_replay(&db, project.path(), "seed.commands").await
    });
    // Only two non-blank, non-comment lines.
    assert_completed(&events, 2);
}

#[test]
fn replay_empty_file_completes_with_zero_commands() {
    let data = tempdir();
    let (project, db) = open_project_db(data.path());
    write_script(project.path(), "empty.commands", "");

    let events = rt().block_on(async {
        run_replay(&db, project.path(), "empty.commands").await
    });
    assert_completed(&events, 0);
}

#[test]
fn replay_only_comments_completes_with_zero_commands() {
    let data = tempdir();
    let (project, db) = open_project_db(data.path());
    write_script(
        project.path(),
        "comments.commands",
        "# just\n# comments\n\n",
    );

    let events = rt().block_on(async {
        run_replay(&db, project.path(), "comments.commands").await
    });
    assert_completed(&events, 0);
}

#[test]
fn replay_constraint_failure_shows_real_names_not_placeholders() {
    // F2 follow-up (ADR-0035 Amendment 1): a replayed command that hits a
    // UNIQUE violation renders with the REAL table/column/value (enriched
    // like the interactive path) — never a literal `{table}` / `{column}`
    // / `{value}` placeholder. Before the fix, replay rendered via a
    // contextless `friendly_message()` and leaked the markers.
    let data = tempdir();
    let (project, db) = open_project_db(data.path());
    write_script(
        project.path(),
        "dup.commands",
        "create table T with pk id(int)\n\
         add column T: email (text)\n\
         add constraint unique to T.email\n\
         insert into T (id, email) values (1, 'a@b.com')\n\
         insert into T (id, email) values (2, 'a@b.com')\n",
    );
    let events = rt().block_on(async { run_replay(&db, project.path(), "dup.commands").await });
    let failed = assert_failed_at(&events, 5);
    let AppEvent::ReplayFailed { error, .. } = failed else {
        unreachable!()
    };
    // No unsubstituted placeholders (the safety net + enrichment).
    assert!(
        !error.contains("{table}") && !error.contains("{column}") && !error.contains("{value}"),
        "no unsubstituted placeholders; got: {error}"
    );
    // The real table + column are shown (resolved from the engine
    // message). The offending value is NOT shown: replay parses in
    // advanced mode → `SqlInsert`, whose values are raw SQL text (ADR-0033
    // verbatim execution), not retained typed values — so it degrades to
    // the neutral "that value" rather than leaking `{value}`.
    assert!(error.contains("T.email"), "names the real table.column; got: {error}");
}

#[test]
fn replay_missing_file_fails_with_line_number_zero() {
    let data = tempdir();
    let (project, db) = open_project_db(data.path());

    let events = rt().block_on(async {
        run_replay(&db, project.path(), "no-such-file.commands").await
    });
    let failed = assert_failed_at(&events, 0);
    let AppEvent::ReplayFailed { error, .. } = failed else {
        unreachable!()
    };
    assert!(
        error.contains("could not open"),
        "expected `could not open` in error: {error}"
    );
}

#[test]
fn replay_aborts_on_first_parse_failure_and_reports_line() {
    let data = tempdir();
    let (project, db) = open_project_db(data.path());
    write_script(
        project.path(),
        "bad.commands",
        // Line 1: ok. Line 2: ok. Line 3: parse error
        // (`broken keyword X` — not a recognised command).
        "create table T with pk id(int)\n\
         add column T: name (text)\n\
         this is not a command\n\
         insert into T (1, 'should not happen')\n",
    );

    let events = rt().block_on(async {
        run_replay(&db, project.path(), "bad.commands").await
    });
    let failed = assert_failed_at(&events, 3);
    let AppEvent::ReplayFailed { error, command, .. } = failed else {
        unreachable!()
    };
    assert!(error.contains("parse error"), "got: {error}");
    assert_eq!(command, "this is not a command");

    // The failing line stops dispatch — no row was inserted —
    // but earlier commands stayed applied (table T exists with
    // the `name` column).
    let desc = rt()
        .block_on(async { db.describe_table("T".to_string(), None).await })
        .expect("describe_table");
    assert!(
        desc.columns.iter().any(|c| c.name == "name"),
        "earlier add column should have stayed applied"
    );
    let data_result = rt()
        .block_on(async { db.query_data("T".to_string(), None, None, None).await })
        .expect("query_data");
    assert!(
        data_result.rows.is_empty(),
        "post-failure insert should not have run"
    );
}

#[test]
fn replay_rejects_wrong_type_value_in_a_hand_built_script() {
    // Replay parses each line with the SAME schema-aware parser the
    // interactive path uses, in **advanced mode** (the full surface),
    // and executes the result — so a replayed line behaves exactly as
    // if it had been typed interactively in advanced mode. Nothing is
    // skipped or simplified during replay (handoff-13 §2.1: the schema
    // is threaded so the parser is fully schema-aware).
    //
    // A real journal only ever contains commands that already executed
    // successfully (history.log is success-only; ADR-0034's deferred
    // journal replays `ok` lines only), so a wrong-type line like this
    // never arises from a genuine replay. It only arises from a
    // *hand-built* `.commands` script — the robustness case this test
    // exercises: replay must reject the bad line and stop, leaving
    // state intact, with the same error a user would see typing it.
    //
    // Where the rejection lands depends on the grammar the line
    // matches, exactly as interactively: `insert into T values (…)` is
    // SQL in advanced mode, and SQL defers column-type checking to the
    // engine, so `'not a number'` in the int `count` column is rejected
    // at **execute** time (the engine's column-type enforcement) rather
    // than at parse time. Either way the line fails and is not applied.
    // (Before sub-phase 3j, `insert` was a DSL-only entry word, so even
    // advanced-mode parsing hit the DSL typed-slot rail and this was a
    // parse-time rejection — ADR-0033 Amendment 3.)
    let data = tempdir();
    let (project, db) = open_project_db(data.path());
    write_script(
        project.path(),
        "typed.commands",
        "create table T with pk id(int)\n\
         add column T: count (int)\n\
         insert into T values (1, 'not a number')\n",
    );

    let events = rt().block_on(async {
        run_replay(&db, project.path(), "typed.commands").await
    });
    let failed = assert_failed_at(&events, 3);
    let AppEvent::ReplayFailed { error, .. } = failed else {
        unreachable!()
    };
    assert!(
        !error.is_empty(),
        "the rejected line must carry a reported error",
    );

    // The earlier two lines stayed applied; the failing insert
    // did not run — state is intact.
    let data_result = rt()
        .block_on(async { db.query_data("T".to_string(), None, None, None).await })
        .expect("query_data");
    assert!(
        data_result.rows.is_empty(),
        "the rejected insert must not have dispatched",
    );
}

#[test]
fn replay_aborts_on_first_runtime_failure_and_reports_line() {
    let data = tempdir();
    let (project, db) = open_project_db(data.path());
    write_script(
        project.path(),
        "bad.commands",
        // Line 2 references a table that doesn't exist; the
        // engine refuses, replay stops and reports line 2.
        "create table T with pk id(int)\n\
         add column NotATable: x (text)\n\
         insert into T (1)\n",
    );

    let events = rt().block_on(async {
        run_replay(&db, project.path(), "bad.commands").await
    });
    let _ = assert_failed_at(&events, 2);
}

#[test]
fn replay_skips_nested_replay_with_a_warning() {
    // ADR-0034: a nested `replay` is no longer refused (which would
    // force a user to hand-edit a journal that happens to contain a
    // `replay` they once ran). It is skipped — sidestepping the
    // infinite-loop footgun by construction — and warned about,
    // because the nested file's commands are not reconstructed.
    let data = tempdir();
    let (project, db) = open_project_db(data.path());
    write_script(project.path(), "inner.commands", "create table T with pk id(int)\n");
    write_script(
        project.path(),
        "outer.commands",
        "create table U with pk id(int)\nreplay inner.commands\n",
    );

    let events = rt().block_on(async {
        run_replay(&db, project.path(), "outer.commands").await
    });
    // The outer `create table U` ran; the nested `replay` was
    // skipped (count 1), with a warning.
    match events.last().expect("event") {
        AppEvent::ReplayCompleted { count, warnings, .. } => {
            assert_eq!(*count, 1, "only the outer create ran; events: {events:?}");
            assert!(
                warnings.iter().any(|w| w.contains("nested") && w.contains("replay inner.commands")),
                "expected a nested-replay skip warning; got {warnings:?}",
            );
        }
        other => panic!("expected ReplayCompleted (nested replay skipped), got {other:?}"),
    }
    // The nested file's table was NOT created (the replay was skipped).
    let cols = rt().block_on(async { db.query_data("T".to_string(), None, None, None).await });
    assert!(cols.is_err(), "inner.commands' table T must not exist (nested replay skipped)");
}

#[test]
fn replay_history_log_records_subcommands_only() {
    // Per handoff §A3: replaying produces the same per-command
    // history.log entries as if each line had been typed
    // interactively. The replay invocation itself MUST NOT
    // appear in history.log (otherwise `replay history.log`
    // would re-trigger itself recursively).
    let data = tempdir();
    let (project, db) = open_project_db(data.path());
    write_script(
        project.path(),
        "seed.commands",
        "create table T with pk id(int)\nadd column T: name (text)\n",
    );

    let events = rt().block_on(async {
        run_replay(&db, project.path(), "seed.commands").await
    });
    assert_completed(&events, 2);

    let history = fs::read_to_string(project.path().join("history.log"))
        .expect("history.log exists");
    // Per-command entries landed.
    assert!(
        history.lines().any(|l| l.contains("create table T with pk id(int)")),
        "history.log missing create line:\n{history}"
    );
    assert!(
        history.lines().any(|l| l.contains("add column T: name (text)")),
        "history.log missing add column line:\n{history}"
    );
    // The replay invocation itself did NOT land — that's
    // the App layer's responsibility (Action::Replay never
    // reaches the per-command persistence path).
    assert!(
        !history.lines().any(|l| l.contains("replay seed.commands")),
        "history.log unexpectedly contains the replay invocation:\n{history}"
    );
}