//! Iteration-3 integration tests: rebuild from text on a
//! missing `.db` (ADR-0015 §7).
//!
//! These tests:
//!
//! 1. Build a populated project via Iteration 2's write-through
//!    path so YAML and CSVs end up on disk.
//! 2. Delete `playground.db`.
//! 3. Re-open the project and call `rebuild_from_text`.
//! 4. Verify the schema, relationships, and row data round-trip.

use std::fs;

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

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")
}

#[test]
fn rebuild_restores_schema_only_project() {
    let data = tempdir();

    // Phase 1: populate via write-through.
    let project_path = {
        let project = project::open_or_create(None, Some(data.path())).unwrap();
        let path = project.path().to_path_buf();
        let db = Database::open_with_persistence(
            project.db_path(),
            Persistence::new(path.clone()),
        )
        .unwrap();
        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();
        });
        drop(db);
        drop(project);
        path
    };

    // Phase 2: delete the .db so the next open triggers rebuild.
    fs::remove_file(project_path.join(PLAYGROUND_DB)).unwrap();

    // Phase 3: reopen and rebuild.
    let project = project::Project::open(&project_path).unwrap();
    let db = Database::open_with_persistence(
        project.db_path(),
        Persistence::new(project.path().to_path_buf()),
    )
    .unwrap();
    rt().block_on(async {
        db.rebuild_from_text(project.path().to_path_buf(), None)
            .await
            .expect("rebuild");
    });

    // Phase 4: confirm Customers exists with the right shape.
    let desc = rt()
        .block_on(async { db.describe_table("Customers".to_string()).await })
        .expect("describe_table");
    assert_eq!(desc.name, "Customers");
    let cols: Vec<&str> = desc.columns.iter().map(|c| c.name.as_str()).collect();
    assert_eq!(cols, vec!["id", "Name"]);
}

#[test]
fn rebuild_restores_rows_from_csv() {
    let data = tempdir();
    let project_path = {
        let project = project::open_or_create(None, Some(data.path())).unwrap();
        let path = project.path().to_path_buf();
        let db = Database::open_with_persistence(
            project.db_path(),
            Persistence::new(path.clone()),
        )
        .unwrap();
        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".to_string()),
            )
            .await
            .unwrap();
            db.insert(
                "Customers".to_string(),
                None,
                vec![Value::Text("Alice".to_string())],
                Some("insert".to_string()),
            )
            .await
            .unwrap();
            db.insert(
                "Customers".to_string(),
                None,
                vec![Value::Text("Bob".to_string())],
                Some("insert".to_string()),
            )
            .await
            .unwrap();
        });
        drop(db);
        drop(project);
        path
    };

    fs::remove_file(project_path.join(PLAYGROUND_DB)).unwrap();

    let project = project::Project::open(&project_path).unwrap();
    let db = Database::open_with_persistence(
        project.db_path(),
        Persistence::new(project.path().to_path_buf()),
    )
    .unwrap();
    rt().block_on(async {
        db.rebuild_from_text(project.path().to_path_buf(), None)
            .await
            .expect("rebuild");
    });

    let rows = rt()
        .block_on(async { db.query_data("Customers".to_string(), None, None).await })
        .expect("query_data");
    assert_eq!(rows.rows.len(), 2);
    let names: Vec<Option<String>> = rows.rows.iter().map(|r| r[1].clone()).collect();
    assert_eq!(names[0].as_deref(), Some("Alice"));
    assert_eq!(names[1].as_deref(), Some("Bob"));
}

#[test]
fn rebuild_restores_relationships_and_cascade_behaviour() {
    let data = tempdir();
    let project_path = {
        let project = project::open_or_create(None, Some(data.path())).unwrap();
        let path = project.path().to_path_buf();
        let db = Database::open_with_persistence(
            project.db_path(),
            Persistence::new(path.clone()),
        )
        .unwrap();
        rt().block_on(async {
            db.create_table(
                "Customers".to_string(),
                vec![ColumnSpec::new("id".to_string(), Type::Serial)],
                vec!["id".to_string()],
                Some("create".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".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("rel".to_string()),
            )
            .await
            .unwrap();
            db.insert(
                "Customers".to_string(),
                Some(vec!["id".to_string()]),
                vec![Value::Number("1".to_string())],
                Some("insert".to_string()),
            )
            .await
            .unwrap();
            db.insert(
                "Orders".to_string(),
                Some(vec!["CustId".to_string()]),
                vec![Value::Number("1".to_string())],
                Some("insert".to_string()),
            )
            .await
            .unwrap();
        });
        drop(db);
        drop(project);
        path
    };

    fs::remove_file(project_path.join(PLAYGROUND_DB)).unwrap();

    let project = project::Project::open(&project_path).unwrap();
    let db = Database::open_with_persistence(
        project.db_path(),
        Persistence::new(project.path().to_path_buf()),
    )
    .unwrap();
    rt().block_on(async {
        db.rebuild_from_text(project.path().to_path_buf(), None)
            .await
            .expect("rebuild");
    });

    // Relationship is back: cascade-delete from Customers
    // should also clean Orders.
    let result = rt()
        .block_on(async {
            db.delete(
                "Customers".to_string(),
                rdbms_playground::dsl::RowFilter::AllRows,
                Some("delete".to_string()),
            )
            .await
        })
        .expect("delete");
    assert_eq!(result.rows_affected, 1);
    assert_eq!(result.cascade.len(), 1, "expected one cascade entry: {result:?}");
    assert_eq!(result.cascade[0].child_table, "Orders");
}

#[test]
fn rebuild_reports_fatal_error_on_bad_csv_row() {
    let data = tempdir();

    // Create a project, populate, then corrupt the CSV.
    let project_path = {
        let project = project::open_or_create(None, Some(data.path())).unwrap();
        let path = project.path().to_path_buf();
        let db = Database::open_with_persistence(
            project.db_path(),
            Persistence::new(path.clone()),
        )
        .unwrap();
        rt().block_on(async {
            db.create_table(
                "Numbers".to_string(),
                vec![
                    ColumnSpec::new("id".to_string(), Type::Serial),
                    ColumnSpec::new("n".to_string(), Type::Int),
                ],
                vec!["id".to_string()],
                Some("create".to_string()),
            )
            .await
            .unwrap();
            db.insert(
                "Numbers".to_string(),
                Some(vec!["n".to_string()]),
                vec![Value::Number("1".to_string())],
                Some("insert".to_string()),
            )
            .await
            .unwrap();
        });
        drop(db);
        drop(project);
        path
    };

    // Hand-corrupt the CSV: replace the int with a non-number.
    let csv_path = project_path.join("data").join("Numbers.csv");
    let body = fs::read_to_string(&csv_path).unwrap();
    let corrupt = body.replace(",1\n", ",not-a-number\n");
    fs::write(&csv_path, corrupt).unwrap();

    fs::remove_file(project_path.join(PLAYGROUND_DB)).unwrap();

    let project = project::Project::open(&project_path).unwrap();
    let db = Database::open_with_persistence(
        project.db_path(),
        Persistence::new(project.path().to_path_buf()),
    )
    .unwrap();
    let err = rt()
        .block_on(async {
            db.rebuild_from_text(project.path().to_path_buf(), None).await
        })
        .expect_err("must fail with row-level error");
    let msg = format!("{err}");
    assert!(msg.contains("row 2"), "msg should name the row: {msg}");
    assert!(msg.contains("Numbers"), "msg should name the table: {msg}");
    assert!(msg.contains("integer"), "msg should explain the type mismatch: {msg}");
}

#[test]
fn rebuild_preserves_created_at_from_yaml() {
    let data = tempdir();
    let project_path = {
        let project = project::open_or_create(None, Some(data.path())).unwrap();
        let path = project.path().to_path_buf();
        let db = Database::open_with_persistence(
            project.db_path(),
            Persistence::new(path.clone()),
        )
        .unwrap();
        rt().block_on(async {
            db.create_table(
                "T".to_string(),
                vec![ColumnSpec::new("id".to_string(), Type::Serial)],
                vec!["id".to_string()],
                Some("create".to_string()),
            )
            .await
            .unwrap();
        });
        drop(db);
        drop(project);
        path
    };

    // Substitute a recognizable timestamp into project.yaml.
    let yaml_path = project_path.join("project.yaml");
    let body = fs::read_to_string(&yaml_path).unwrap();
    let edited = body
        .lines()
        .map(|l| {
            if l.trim_start().starts_with("created_at:") {
                "  created_at: 2020-01-02T03:04:05Z".to_string()
            } else {
                l.to_string()
            }
        })
        .collect::<Vec<_>>()
        .join("\n");
    fs::write(&yaml_path, format!("{edited}\n")).unwrap();

    // Delete the .db, rebuild from text.
    fs::remove_file(project_path.join(PLAYGROUND_DB)).unwrap();
    let project = project::Project::open(&project_path).unwrap();
    let db = Database::open_with_persistence(
        project.db_path(),
        Persistence::new(project.path().to_path_buf()),
    )
    .unwrap();
    rt().block_on(async {
        db.rebuild_from_text(project.path().to_path_buf(), None)
            .await
            .expect("rebuild");
    });

    // Trigger any successful command so project.yaml is
    // rewritten from the now-rebuilt db state.
    rt().block_on(async {
        db.describe_table("T".to_string())
            .await
            .unwrap();
        // describe is read-only; force a rewrite by adding a column.
        db.add_column(
            "T".to_string(),
            ColumnSpec::new("Note", Type::Text),
            Some("add column".to_string()),
        )
        .await
        .unwrap();
    });

    let final_yaml = fs::read_to_string(&yaml_path).unwrap();
    assert!(
        final_yaml.contains("created_at: 2020-01-02T03:04:05Z"),
        "yaml should preserve the edited created_at:\n{final_yaml}",
    );
}

/// Indexes round-trip through `project.yaml` and a full rebuild
/// (ADR-0025): create an index, drop the `.db`, rebuild from
/// text, confirm the index is back.
#[test]
fn rebuild_restores_indexes() {
    let data = tempdir();
    let project_path = {
        let project = project::open_or_create(None, Some(data.path())).unwrap();
        let path = project.path().to_path_buf();
        let db = Database::open_with_persistence(
            project.db_path(),
            Persistence::new(path.clone()),
        )
        .unwrap();
        rt().block_on(async {
            db.create_table(
                "Customers".to_string(),
                vec![
                    ColumnSpec::new("id".to_string(), Type::Serial),
                    ColumnSpec::new("Email".to_string(), Type::Text),
                ],
                vec!["id".to_string()],
                Some("create table Customers with pk id(serial)".to_string()),
            )
            .await
            .unwrap();
            db.add_index(
                Some("idx_email".to_string()),
                "Customers".to_string(),
                vec!["Email".to_string()],
                Some("add index as idx_email on Customers (Email)".to_string()),
            )
            .await
            .unwrap();
        });
        drop(db);
        drop(project);
        path
    };

    // The index must be recorded in project.yaml — the `.db` is
    // a derived artifact and gets discarded next.
    let yaml = fs::read_to_string(project_path.join(project::PROJECT_YAML)).unwrap();
    assert!(yaml.contains("idx_email"), "yaml should record the index:\n{yaml}");

    fs::remove_file(project_path.join(PLAYGROUND_DB)).unwrap();

    let project = project::Project::open(&project_path).unwrap();
    let db = Database::open_with_persistence(
        project.db_path(),
        Persistence::new(project.path().to_path_buf()),
    )
    .unwrap();
    rt().block_on(async {
        db.rebuild_from_text(project.path().to_path_buf(), None)
            .await
            .expect("rebuild");
    });

    let desc = rt()
        .block_on(async { db.describe_table("Customers".to_string()).await })
        .expect("describe_table");
    assert_eq!(desc.indexes.len(), 1, "index should survive rebuild");
    assert_eq!(desc.indexes[0].name, "idx_email");
    assert_eq!(desc.indexes[0].columns, vec!["Email".to_string()]);
}