//! Iteration-2 integration tests: per-command write-through
//! to `project.yaml`, `data/<table>.csv`, and `history.log`
//! (ADR-0015 §3-§6).
//!
//! These tests exercise the full path from
//! `Database::open_with_persistence` through a successful
//! command into the on-disk text targets. They use
//! `Database::open_with_persistence(...)` so the worker
//! thread runs the persistence callbacks the runtime would.

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

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

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 project under a fresh data root and return the
/// `Database` (with persistence wired) plus the path so the
/// test can inspect on-disk state. The project is held alive
/// implicitly via the leaked `TempDir` returned alongside.
fn open_project(
    data: &tempfile::TempDir,
) -> (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(project.db_path(), persistence)
        .expect("open db with persistence");
    (project, db, path)
}

fn read_yaml(project_path: &Path) -> String {
    fs::read_to_string(project_path.join(PROJECT_YAML)).expect("project.yaml")
}

fn read_csv(project_path: &Path, table: &str) -> Option<String> {
    fs::read_to_string(project_path.join(DATA_DIR).join(format!("{table}.csv"))).ok()
}

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

    rt().block_on(async {
        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();
    });

    let yaml = read_yaml(&path);
    assert!(yaml.contains("- name: Customers"), "yaml missing table:\n{yaml}");
    assert!(yaml.contains("primary_key: [id]"), "yaml: {yaml}");
    assert!(yaml.contains("type: serial"), "yaml: {yaml}");
    assert!(yaml.contains("type: text"), "yaml: {yaml}");

    // ADR-0052: journaling moved to the dispatch layer (the worker no
    // longer writes history.log); this test verifies only the yaml state.
    // Journaling is covered by the history.rs/app.rs/replay tests.
}

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

    rt().block_on(async {
        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();
        db.insert(
            "Customers".to_string(),
            None,
            vec![Value::Text("Alice".to_string())],
            Some("insert into Customers ('Alice')".to_string()),
        )
        .await
        .unwrap();
    });

    let csv = read_csv(&path, "Customers").expect("Customers.csv missing");
    let lines: Vec<&str> = csv.trim_end().lines().collect();
    assert_eq!(lines[0], "id,Name");
    assert_eq!(lines[1], "1,Alice");

    // ADR-0052: journaling moved off the worker; this test verifies the
    // csv state only (journaling covered elsewhere).
}

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

    rt().block_on(async {
        db.create_table(
            "Customers".to_string(),
            vec![ColumnSpec::new("id".to_string(), Type::Serial)],
            vec!["id".to_string()],
            Some("create table Customers with pk id(serial)".to_string()),
        )
        .await
        .unwrap();
        db.insert(
            "Customers".to_string(),
            Some(vec!["id".to_string()]),
            vec![Value::Number("42".to_string())],
            Some("insert into Customers (id) values (42)".to_string()),
        )
        .await
        .unwrap();
        // The CSV exists before drop.
        assert!(read_csv(&path, "Customers").is_some());

        db.drop_table(
            "Customers".to_string(),
            Some("drop table Customers".to_string()),
        )
        .await
        .unwrap();
    });

    assert!(read_csv(&path, "Customers").is_none(), "CSV should be deleted");
    let yaml = read_yaml(&path);
    assert!(!yaml.contains("- name: Customers"), "table should be gone from yaml:\n{yaml}");
}

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

    rt().block_on(async {
        db.create_table(
            "Customers".to_string(),
            vec![ColumnSpec::new("id".to_string(), Type::Serial)],
            vec!["id".to_string()],
            Some("create table Customers with pk id(serial)".to_string()),
        )
        .await
        .unwrap();
        db.create_table(
            "Orders".to_string(),
            vec![
                ColumnSpec::new("id".to_string(), Type::Serial),
                ColumnSpec::new("CustId".to_string(), Type::Int),
            ],
            vec!["id".to_string()],
            Some("create table Orders with pk id(serial), CustId(int)".to_string()),
        )
        .await
        .unwrap();
        db.add_relationship(
            None,
            "Customers".to_string(),
            vec!["id".to_string()],
            "Orders".to_string(),
            vec!["CustId".to_string()],
            ReferentialAction::Cascade,
            ReferentialAction::NoAction,
            false,
            Some(
                "add 1:n relationship from Customers.id to Orders.CustId on delete cascade"
                    .to_string(),
            ),
        )
        .await
        .unwrap();
        // Customers has only a serial PK; long-form INSERT with
        // an explicit id keeps the test independent of short-form
        // semantics for "all-auto-generated" tables.
        db.insert(
            "Customers".to_string(),
            Some(vec!["id".to_string()]),
            vec![Value::Number("1".to_string())],
            Some("insert into Customers (id) values (1)".to_string()),
        )
        .await
        .unwrap();
        db.insert(
            "Orders".to_string(),
            Some(vec!["CustId".to_string()]),
            vec![Value::Number("1".to_string())],
            Some("insert into Orders (CustId) values (1)".to_string()),
        )
        .await
        .unwrap();

        // Cascade delete from Customers should also clean Orders.
        let result = db
            .delete(
                "Customers".to_string(),
                RowFilter::eq("id", Value::Number("1".to_string())),
                Some("delete from Customers where id=1".to_string()),
            )
            .await
            .unwrap();
        assert_eq!(result.rows_affected, 1);
    });

    // Both CSVs should be gone after the cascade leaves both
    // tables empty: empty table -> no CSV (the rule from
    // Persistence::write_table_data; see ADR-0015 §4 commentary).
    assert!(
        read_csv(&path, "Customers").is_none(),
        "Customers.csv should be gone after cascade leaves it empty",
    );
    assert!(
        read_csv(&path, "Orders").is_none(),
        "Orders.csv should be gone after cascade leaves it empty",
    );
}

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

    rt().block_on(async {
        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();
    });

    // Schema landed in YAML.
    let yaml = read_yaml(&path);
    assert!(yaml.contains("- name: Customers"), "yaml missing table:\n{yaml}");
    // ...but no CSV until there's data.
    assert!(
        read_csv(&path, "Customers").is_none(),
        "no CSV should exist for an empty table",
    );
}

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

    rt().block_on(async {
        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();
        db.insert(
            "Customers".to_string(),
            None,
            vec![Value::Text("Alice".to_string())],
            Some("insert into Customers ('Alice')".to_string()),
        )
        .await
        .unwrap();
        // CSV exists once there's data.
        assert!(read_csv(&path, "Customers").is_some());

        db.delete(
            "Customers".to_string(),
            RowFilter::AllRows,
            Some("delete from Customers --all-rows".to_string()),
        )
        .await
        .unwrap();
    });

    assert!(
        read_csv(&path, "Customers").is_none(),
        "CSV should be removed when the table becomes empty",
    );
}

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

    rt().block_on(async {
        db.create_table(
            "Customers".to_string(),
            vec![ColumnSpec::new("id".to_string(), Type::Serial)],
            vec!["id".to_string()],
            Some("create table Customers with pk id(serial)".to_string()),
        )
        .await
        .unwrap();
        let yaml_before = read_yaml(&path);

        // Same name again — should fail.
        let err = db
            .create_table(
                "Customers".to_string(),
                vec![ColumnSpec::new("id".to_string(), Type::Serial)],
                vec!["id".to_string()],
                Some("create table Customers with pk id(serial)".to_string()),
            )
            .await
            .expect_err("must fail");
        let _ = err;

        let yaml_after = read_yaml(&path);
        assert_eq!(yaml_before, yaml_after, "failed cmd must not change yaml");
    });

    // ADR-0052: journaling moved off the worker; this test now verifies
    // only that a failed command does not change the yaml state.
}

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

    rt().block_on(async {
        db.create_table(
            "Customers".to_string(),
            vec![ColumnSpec::new("id".to_string(), Type::Serial)],
            vec!["id".to_string()],
            None,
        )
        .await
        .unwrap();
        db.create_table(
            "Orders".to_string(),
            vec![
                ColumnSpec::new("id".to_string(), Type::Serial),
                ColumnSpec::new("CustId".to_string(), Type::Int),
            ],
            vec!["id".to_string()],
            None,
        )
        .await
        .unwrap();
        db.add_relationship(
            None,
            "Customers".to_string(),
            vec!["id".to_string()],
            "Orders".to_string(),
            vec!["CustId".to_string()],
            ReferentialAction::Cascade,
            ReferentialAction::NoAction,
            false,
            Some(
                "add 1:n relationship from Customers.id to Orders.CustId on delete cascade"
                    .to_string(),
            ),
        )
        .await
        .unwrap();
    });

    let yaml = read_yaml(&path);
    assert!(yaml.contains("- name: Customers_id_to_Orders_CustId"), "yaml: {yaml}");
    assert!(yaml.contains("on_delete: cascade"), "yaml: {yaml}");
    assert!(yaml.contains("on_update: no_action"), "yaml: {yaml}");
}