//! 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_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).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_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_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).await })
        .expect("query_data");
    assert!(
        data_result.rows.is_empty(),
        "post-failure insert should not have run"
    );
}

#[test]
fn replay_rejects_typed_slot_violation_at_parse_time() {
    // Schema-aware replay (handoff-13 §2.1 fix): run_replay
    // re-snapshots the schema per line and parses with
    // parse_command_with_schema. So a wrong-type value in a
    // value list is caught at *parse* time during replay —
    // surfaced through the `replay.error_parse` wrapper ("parse
    // error …") — exactly as the interactive path would, rather
    // than only at bind time.
    //
    // `'not a number'` (a string) lands in the int `count`
    // slot. The schemaless parser would accept it (a string is
    // a value literal) and only bind-time would reject; the
    // schema-aware parser rejects it at parse time. Asserting
    // the error went through the parse wrapper proves the
    // schema was threaded.
    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.contains("parse error"),
        "typed-slot violation should be caught at parse time, got: {error}",
    );

    // The earlier two lines stayed applied; the failing insert
    // did not run.
    let data_result = rt()
        .block_on(async { db.query_data("T".to_string(), 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_refuses_nested_replay() {
    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
    });
    let failed = assert_failed_at(&events, 2);
    let AppEvent::ReplayFailed { error, .. } = failed else {
        unreachable!()
    };
    assert!(
        error.contains("nested `replay`"),
        "expected nested-replay refusal: {error}"
    );
}

#[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}"
    );
}