//! Tier-3 integration tests for the undo/snapshot ring wired into
//! the db worker (ADR-0006 Amendment 1, §8 step 3).
//!
//! These drive the real `Database` worker: a mutation takes a
//! pre-op snapshot, `undo` restores it through the live connection,
//! `redo` re-applies, a batch collapses to a single undo step, and
//! `--no-undo` (undo disabled) takes no snapshots at all.

use std::path::Path;

use rdbms_playground::db::Database;
use rdbms_playground::dsl::{ColumnSpec, RowFilter, Type, Value};
use rdbms_playground::persistence::Persistence;
use rdbms_playground::project;

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

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

/// Open a fresh temp project with undo enabled (or not).
fn open_project(
    data: &tempfile::TempDir,
    undo_enabled: bool,
) -> (project::Project, Database, std::path::PathBuf) {
    let project = project::open_or_create(None, Some(data.path())).expect("open project");
    let path = project.path().to_path_buf();
    let persistence = Persistence::new(path.clone());
    let db = Database::open_with_persistence_and_undo(project.db_path(), persistence, undo_enabled)
        .expect("open db");
    (project, db, path)
}

async fn make_customers(db: &Database) {
    db.create_table(
        "Customers".to_string(),
        vec![
            ColumnSpec::new("id".to_string(), Type::Serial),
            ColumnSpec::new("Name".to_string(), Type::Text),
        ],
        vec!["id".to_string()],
        Some("create table Customers with pk id(serial)".to_string()),
    )
    .await
    .unwrap();
}

async fn insert_named(db: &Database, name: &str) {
    db.insert(
        "Customers".to_string(),
        None,
        vec![Value::Text(name.to_string())],
        Some(format!("insert into Customers ('{name}')")),
    )
    .await
    .unwrap();
}

async fn row_count(db: &Database) -> usize {
    db.query_data("Customers".to_string(), None, None, None)
        .await
        .unwrap()
        .rows
        .len()
}

fn snapshots_dir(path: &Path) -> std::path::PathBuf {
    path.join(".snapshots")
}

#[test]
fn mutation_snapshots_and_undo_restores_through_the_worker() {
    let data = tempdir();
    let (_p, db, path) = open_project(&data, true);

    rt().block_on(async {
        make_customers(&db).await;
        insert_named(&db, "Alice").await;
        insert_named(&db, "Bob").await;
        assert_eq!(row_count(&db).await, 2);

        // Destructive op: delete Bob (id = 2).
        db.delete(
            "Customers".to_string(),
            RowFilter::eq("id", Value::Number("2".to_string())),
            Some("delete from Customers where id = 2".to_string()),
        )
        .await
        .unwrap();
        assert_eq!(row_count(&db).await, 1);

        // The pending undo names the delete.
        let peek = db.peek_undo().await.unwrap().expect("an undo entry");
        assert_eq!(peek.command, "delete from Customers where id = 2");

        // Undo restores Bob.
        let undone = db.undo().await.unwrap().expect("undo applied");
        assert_eq!(undone.command, "delete from Customers where id = 2");
        assert_eq!(row_count(&db).await, 2, "Bob restored by undo");
    });

    assert!(snapshots_dir(&path).exists(), "snapshots dir created");
}

#[test]
fn redo_reapplies_the_undone_command() {
    let data = tempdir();
    let (_p, db, _path) = open_project(&data, true);

    rt().block_on(async {
        make_customers(&db).await;
        insert_named(&db, "Alice").await;
        insert_named(&db, "Bob").await;
        db.delete(
            "Customers".to_string(),
            RowFilter::eq("id", Value::Number("2".to_string())),
            Some("delete from Customers where id = 2".to_string()),
        )
        .await
        .unwrap();
        assert_eq!(row_count(&db).await, 1);

        db.undo().await.unwrap();
        assert_eq!(row_count(&db).await, 2);

        let redone = db.redo().await.unwrap().expect("redo applied");
        assert_eq!(redone.command, "delete from Customers where id = 2");
        assert_eq!(row_count(&db).await, 1, "delete re-applied by redo");
    });
}

#[test]
fn new_work_after_undo_clears_redo() {
    let data = tempdir();
    let (_p, db, _path) = open_project(&data, true);

    rt().block_on(async {
        make_customers(&db).await;
        insert_named(&db, "Alice").await;
        db.delete(
            "Customers".to_string(),
            RowFilter::AllRows,
            Some("delete from Customers --all-rows".to_string()),
        )
        .await
        .unwrap();
        db.undo().await.unwrap();
        assert!(db.peek_redo().await.unwrap().is_some(), "redo available");

        // New destructive work.
        insert_named(&db, "Carol").await;
        assert!(
            db.peek_redo().await.unwrap().is_none(),
            "new work discards the redo stack"
        );
    });
}

#[test]
fn undo_disabled_takes_no_snapshots() {
    let data = tempdir();
    let (_p, db, path) = open_project(&data, false);

    rt().block_on(async {
        make_customers(&db).await;
        insert_named(&db, "Alice").await;
        db.delete(
            "Customers".to_string(),
            RowFilter::AllRows,
            Some("delete from Customers --all-rows".to_string()),
        )
        .await
        .unwrap();

        // Nothing to undo, and no snapshot machinery on disk.
        assert!(db.undo().await.unwrap().is_none(), "undo is a no-op when disabled");
        assert!(db.peek_undo().await.unwrap().is_none());
    });

    assert!(
        !snapshots_dir(&path).exists(),
        "no .snapshots dir when undo is disabled"
    );
}

#[test]
fn batch_records_a_single_undo_step() {
    let data = tempdir();
    let (_p, db, _path) = open_project(&data, true);

    rt().block_on(async {
        make_customers(&db).await; // one undo entry (the create)

        // A batch of three inserts → one boundary snapshot.
        db.begin_batch(Some("replay history.log".to_string()))
            .await
            .unwrap();
        insert_named(&db, "Alice").await;
        insert_named(&db, "Bob").await;
        insert_named(&db, "Carol").await;
        db.end_batch().await.unwrap();
        assert_eq!(row_count(&db).await, 3);

        // The single batch undo names the batch command.
        let peek = db.peek_undo().await.unwrap().expect("batch undo entry");
        assert_eq!(peek.command, "replay history.log");

        // One undo rolls the whole batch back to the pre-batch state
        // (table exists, no rows).
        db.undo().await.unwrap();
        assert_eq!(row_count(&db).await, 0, "whole batch undone in one step");

        // The next undo is the create_table (table gone).
        let next = db.peek_undo().await.unwrap().expect("create entry");
        assert_eq!(next.command, "create table Customers with pk id(serial)");
    });
}

#[test]
fn empty_undo_and_redo_are_no_ops() {
    let data = tempdir();
    let (_p, db, _path) = open_project(&data, true);

    rt().block_on(async {
        assert!(db.undo().await.unwrap().is_none());
        assert!(db.redo().await.unwrap().is_none());
        assert!(db.peek_undo().await.unwrap().is_none());
        assert!(db.peek_redo().await.unwrap().is_none());
    });
}