//! Integration tests for compound-primary-key foreign-key
//! references (T3 / ADR-0043) — the DSL `add 1:n relationship`
//! surface end to end.
//!
//! Covers: parse of the parenthesized multi-column endpoint;
//! worker round-trip (declare a 2-column FK, FK is enforced,
//! per-pair type-mismatch refused, arity mismatch refused);
//! persistence round-trip (`columns: [a, b]`); display; and undo.

use rdbms_playground::db::Database;
use rdbms_playground::dsl::{
    parse_command, ColumnSpec, Command, ReferentialAction, SqlForeignKey, Type, Value,
};
use rdbms_playground::persistence::Persistence;
use rdbms_playground::project;

// ---- parse layer ------------------------------------------------

#[test]
fn parenthesized_compound_endpoint_parses_to_column_lists() {
    let cmd = parse_command(
        "add 1:n relationship from Parent.(a, b) to Child.(x, y)",
    )
    .expect("parses");
    match cmd {
        Command::AddRelationship {
            parent_table,
            parent_columns,
            child_table,
            child_columns,
            ..
        } => {
            assert_eq!(parent_table, "Parent");
            assert_eq!(parent_columns, vec!["a".to_string(), "b".to_string()]);
            assert_eq!(child_table, "Child");
            assert_eq!(child_columns, vec!["x".to_string(), "y".to_string()]);
        }
        other => panic!("expected AddRelationship, got {other:?}"),
    }
}

#[test]
fn single_column_endpoint_still_parses_unparenthesized() {
    let cmd = parse_command("add 1:n relationship from Parent.id to Child.pid")
        .expect("parses");
    match cmd {
        Command::AddRelationship {
            parent_columns,
            child_columns,
            ..
        } => {
            assert_eq!(parent_columns, vec!["id".to_string()]);
            assert_eq!(child_columns, vec!["pid".to_string()]);
        }
        other => panic!("expected AddRelationship, got {other:?}"),
    }
}

// ---- SQL surface (advanced mode) --------------------------------

#[test]
fn sql_table_level_compound_fk_parses_to_lists() {
    let cmd = parse_command(
        "create table City (country int, region_code int, \
         foreign key (country, region_code) references Region(country, code))",
    )
    .expect("parses");
    match cmd {
        Command::SqlCreateTable { foreign_keys, .. } => {
            assert_eq!(foreign_keys.len(), 1);
            assert_eq!(
                foreign_keys[0].child_columns,
                vec!["country".to_string(), "region_code".to_string()],
            );
            assert_eq!(
                foreign_keys[0].parent_columns,
                Some(vec!["country".to_string(), "code".to_string()]),
            );
        }
        other => panic!("expected SqlCreateTable, got {other:?}"),
    }
}

#[test]
fn sql_bare_compound_reference_parses_with_no_parent_columns() {
    // `FOREIGN KEY (a, b) REFERENCES P` (no parent cols) — auto-expanded
    // to the parent's full PK at execution (F-D).
    let cmd = parse_command(
        "create table City (country int, region_code int, \
         foreign key (country, region_code) references Region)",
    )
    .expect("parses");
    match cmd {
        Command::SqlCreateTable { foreign_keys, .. } => {
            assert_eq!(
                foreign_keys[0].child_columns,
                vec!["country".to_string(), "region_code".to_string()],
            );
            assert_eq!(foreign_keys[0].parent_columns, None);
        }
        other => panic!("expected SqlCreateTable, got {other:?}"),
    }
}

#[test]
fn sql_create_table_compound_fk_executes_and_enforces() {
    let (_p, db, _dir) = open_project_db();
    let rt = rt();
    rt.block_on(async {
        // Parent with a compound PK.
        db.create_table(
            "Region".to_string(),
            vec![
                ColumnSpec::new("country", Type::Int),
                ColumnSpec::new("code", Type::Int),
            ],
            vec!["country".to_string(), "code".to_string()],
            None,
        )
        .await
        .expect("create Region");
        // Child via the SQL path with a multi-column FK referencing the
        // full compound PK (resolve_create_table_fks).
        db.sql_create_table(
            "City".to_string(),
            vec![
                ColumnSpec::new("country", Type::Int),
                ColumnSpec::new("region_code", Type::Int),
            ],
            vec![],
            vec![],
            vec![],
            vec![SqlForeignKey {
                name: None,
                child_columns: vec!["country".to_string(), "region_code".to_string()],
                parent_table: "Region".to_string(),
                parent_columns: Some(vec!["country".to_string(), "code".to_string()]),
                on_delete: ReferentialAction::NoAction,
                on_update: ReferentialAction::NoAction,
                inline: false,
            }],
            false,
            None,
        )
        .await
        .expect("create City with compound FK");

        db.insert(
            "Region".to_string(),
            Some(vec!["country".to_string(), "code".to_string()]),
            vec![Value::Number("1".to_string()), Value::Number("10".to_string())],
            None,
        )
        .await
        .expect("insert parent");
        let bad = db
            .insert(
                "City".to_string(),
                Some(vec!["country".to_string(), "region_code".to_string()]),
                vec![Value::Number("9".to_string()), Value::Number("9".to_string())],
                None,
            )
            .await;
        assert!(bad.is_err(), "compound FK violation refused by the engine");
    });
}

// ---- worker round-trip ------------------------------------------

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

fn open_project_db() -> (project::Project, Database, tempfile::TempDir) {
    let dir = tempfile::tempdir().expect("create tempdir");
    let project =
        project::open_or_create(None, Some(dir.path())).expect("open or create project");
    let persistence = Persistence::new(project.path().to_path_buf());
    let db = Database::open_with_persistence(project.db_path(), persistence)
        .expect("open db with persistence");
    (project, db, dir)
}

/// `Region(country int, code int)` compound PK + `City(country int,
/// region_code int, name text)` — the child FK columns matching the
/// parent PK by type (int → int).
async fn seed_compound(db: &Database) {
    db.create_table(
        "Region".to_string(),
        vec![
            ColumnSpec::new("country", Type::Int),
            ColumnSpec::new("code", Type::Int),
        ],
        vec!["country".to_string(), "code".to_string()],
        None,
    )
    .await
    .expect("create Region");
    db.create_table(
        "City".to_string(),
        vec![
            ColumnSpec::new("country", Type::Int),
            ColumnSpec::new("region_code", Type::Int),
            ColumnSpec::new("name", Type::Text),
        ],
        vec!["country".to_string()],
        None,
    )
    .await
    .expect("create City");
}

#[test]
fn compound_fk_declares_enforces_and_round_trips() {
    let (_p, db, _dir) = open_project_db();
    let rt = rt();
    rt.block_on(async {
        seed_compound(&db).await;
        // Declare the compound FK: City.(country, region_code) →
        // Region.(country, code).
        db.add_relationship(
            Some("city_region".to_string()),
            "Region".to_string(),
            vec!["country".to_string(), "code".to_string()],
            "City".to_string(),
            vec!["country".to_string(), "region_code".to_string()],
            ReferentialAction::NoAction,
            ReferentialAction::NoAction,
            false,
            None,
        )
        .await
        .expect("add compound relationship");

        // The FK is enforced: a parent row exists for (1, 10); a
        // child referencing it inserts, one referencing (9, 9) is
        // refused by the engine.
        db.insert(
            "Region".to_string(),
            Some(vec!["country".to_string(), "code".to_string()]),
            vec![Value::Number("1".to_string()), Value::Number("10".to_string())],
            None,
        )
        .await
        .expect("insert parent row");
        db.insert(
            "City".to_string(),
            Some(vec![
                "country".to_string(),
                "region_code".to_string(),
                "name".to_string(),
            ]),
            vec![Value::Number("1".to_string()), Value::Number("10".to_string()), Value::Text("Metropolis".to_string())],
            None,
        )
        .await
        .expect("child row referencing an existing parent key inserts");
        let violation = db
            .insert(
                "City".to_string(),
                Some(vec![
                    "country".to_string(),
                    "region_code".to_string(),
                    "name".to_string(),
                ]),
                vec![Value::Number("9".to_string()), Value::Number("9".to_string()), Value::Text("Nowhere".to_string())],
                None,
            )
            .await;
        assert!(
            violation.is_err(),
            "a child row with no matching compound parent key must be refused",
        );

        // describe shows the compound endpoints symmetrically.
        let city = db.describe_table("City".to_string(), None).await.unwrap();
        let outbound = &city.outbound_relationships[0];
        assert_eq!(
            outbound.local_columns,
            vec!["country".to_string(), "region_code".to_string()],
        );
        assert_eq!(
            outbound.other_columns,
            vec!["country".to_string(), "code".to_string()],
        );
    });
}

#[test]
fn compound_fk_create_fk_makes_both_child_columns() {
    let (_p, db, _dir) = open_project_db();
    let rt = rt();
    rt.block_on(async {
        // Region(country, code) compound PK; City has neither FK column.
        db.create_table(
            "Region".to_string(),
            vec![
                ColumnSpec::new("country", Type::Int),
                ColumnSpec::new("code", Type::Int),
            ],
            vec!["country".to_string(), "code".to_string()],
            None,
        )
        .await
        .expect("create Region");
        db.create_table(
            "City".to_string(),
            vec![ColumnSpec::new("name", Type::Text)],
            vec![],
            None,
        )
        .await
        .expect("create City");
        // --create-fk creates both missing child columns, typed to the
        // matching parent PK columns' fk_target_type (int → int).
        db.add_relationship(
            None,
            "Region".to_string(),
            vec!["country".to_string(), "code".to_string()],
            "City".to_string(),
            vec!["c_country".to_string(), "c_code".to_string()],
            ReferentialAction::NoAction,
            ReferentialAction::NoAction,
            true,
            None,
        )
        .await
        .expect("add compound relationship with --create-fk");
        let city = db.describe_table("City".to_string(), None).await.unwrap();
        for col in ["c_country", "c_code"] {
            assert!(
                city.columns.iter().any(|c| c.name == col),
                "--create-fk created `{col}`: {:?}",
                city.columns.iter().map(|c| &c.name).collect::<Vec<_>>(),
            );
        }
    });
}

#[test]
fn compound_fk_arity_mismatch_is_refused() {
    let (_p, db, _dir) = open_project_db();
    let rt = rt();
    rt.block_on(async {
        seed_compound(&db).await;
        // Two parent columns, one child column → arity mismatch.
        let err = db
            .add_relationship(
                None,
                "Region".to_string(),
                vec!["country".to_string(), "code".to_string()],
                "City".to_string(),
                vec!["country".to_string()],
                ReferentialAction::NoAction,
                ReferentialAction::NoAction,
                false,
                None,
            )
            .await;
        assert!(err.is_err(), "mismatched child/parent arity must be refused");
    });
}

#[test]
fn inline_fk_referencing_compound_pk_points_at_table_level_form() {
    // ADR-0043 D4 residual: an *inline* single-column FK cannot express a
    // multi-column reference, so referencing a parent's compound PK must
    // refuse with a pointer to the table-level `FOREIGN KEY (...)` form —
    // not the generic arity message. The grammar marks the FK `inline`.
    let (_p, db, _dir) = open_project_db();
    let rt = rt();
    rt.block_on(async {
        db.create_table(
            "Region".to_string(),
            vec![
                ColumnSpec::new("country", Type::Int),
                ColumnSpec::new("code", Type::Int),
            ],
            vec!["country".to_string(), "code".to_string()],
            None,
        )
        .await
        .expect("create Region");

        // Parse the inline form so the `inline` flag is set by the grammar.
        let cmd = parse_command(
            "create table City (country int references Region(country, code))",
        )
        .expect("parses");
        let Command::SqlCreateTable {
            name,
            columns,
            primary_key,
            unique_constraints,
            check_constraints,
            foreign_keys,
            if_not_exists,
        } = cmd
        else {
            panic!("expected SqlCreateTable");
        };
        let err = db
            .sql_create_table(
                name,
                columns,
                primary_key,
                unique_constraints,
                check_constraints,
                foreign_keys,
                if_not_exists,
                None,
            )
            .await
            .expect_err("inline FK referencing a compound PK must be refused");
        let msg = format!("{err}");
        assert!(
            msg.contains("FOREIGN KEY"),
            "expected a pointer to the table-level `FOREIGN KEY (...)` form, got: {msg}"
        );
    });
}

#[test]
fn compound_fk_type_mismatch_per_pair_is_refused() {
    let (_p, db, _dir) = open_project_db();
    let rt = rt();
    rt.block_on(async {
        db.create_table(
            "Region".to_string(),
            vec![
                ColumnSpec::new("country", Type::Int),
                ColumnSpec::new("code", Type::Int),
            ],
            vec!["country".to_string(), "code".to_string()],
            None,
        )
        .await
        .expect("create Region");
        // `bad` is `text` — incompatible with the `int` PK column it
        // would pair with (per-pair type-compat, ADR-0011).
        db.create_table(
            "City".to_string(),
            vec![
                ColumnSpec::new("country", Type::Int),
                ColumnSpec::new("bad", Type::Text),
            ],
            vec![],
            None,
        )
        .await
        .expect("create City");
        let err = db
            .add_relationship(
                None,
                "Region".to_string(),
                vec!["country".to_string(), "code".to_string()],
                "City".to_string(),
                vec!["country".to_string(), "bad".to_string()],
                ReferentialAction::NoAction,
                ReferentialAction::NoAction,
                false,
                None,
            )
            .await;
        assert!(err.is_err(), "a type-incompatible column pair must be refused");
    });
}

#[test]
fn compound_fk_survives_rebuild_from_text() {
    // The riskiest round-trip: comma-encoded metadata + yaml
    // `columns: [a, b]` → rebuild reconstructs the compound FK DDL.
    let dir = tempfile::tempdir().expect("tempdir");
    let project = project::open_or_create(None, Some(dir.path())).expect("open project");
    let path = project.path().to_path_buf();
    let rt = rt();
    {
        let db = Database::open_with_persistence(
            project.db_path(),
            Persistence::new(path.clone()),
        )
        .expect("open db");
        rt.block_on(async {
            seed_compound(&db).await;
            db.add_relationship(
                Some("city_region".to_string()),
                "Region".to_string(),
                vec!["country".to_string(), "code".to_string()],
                "City".to_string(),
                vec!["country".to_string(), "region_code".to_string()],
                ReferentialAction::NoAction,
                ReferentialAction::NoAction,
                false,
                Some("add 1:n relationship".to_string()),
            )
            .await
            .expect("add compound relationship");
        });
    }
    // Reopen and rebuild the database purely from the persisted
    // project.yaml + data/.
    let db = Database::open_with_persistence(project.db_path(), Persistence::new(path.clone()))
        .expect("reopen db");
    rt.block_on(async {
        db.rebuild_from_text(path.clone(), None)
            .await
            .expect("rebuild from text");
        // The compound FK is reconstructed and still enforced.
        db.insert(
            "Region".to_string(),
            Some(vec!["country".to_string(), "code".to_string()]),
            vec![Value::Number("1".to_string()), Value::Number("10".to_string())],
            None,
        )
        .await
        .expect("insert parent after rebuild");
        let bad = db
            .insert(
                "City".to_string(),
                Some(vec!["country".to_string(), "region_code".to_string()]),
                vec![Value::Number("9".to_string()), Value::Number("9".to_string())],
                None,
            )
            .await;
        assert!(bad.is_err(), "compound FK still enforced after rebuild from text");
        // Endpoints survived the round-trip intact.
        let city = db.describe_table("City".to_string(), None).await.unwrap();
        assert_eq!(
            city.outbound_relationships[0].other_columns,
            vec!["country".to_string(), "code".to_string()],
        );
    });
}

#[test]
fn compound_fk_undo_removes_the_relationship() {
    let dir = tempfile::tempdir().expect("tempdir");
    let project = project::open_or_create(None, Some(dir.path())).expect("open project");
    let db = Database::open_with_persistence_and_undo(
        project.db_path(),
        Persistence::new(project.path().to_path_buf()),
        true,
    )
    .expect("open db with undo");
    let rt = rt();
    rt.block_on(async {
        seed_compound(&db).await;
        db.add_relationship(
            Some("city_region".to_string()),
            "Region".to_string(),
            vec!["country".to_string(), "code".to_string()],
            "City".to_string(),
            vec!["country".to_string(), "region_code".to_string()],
            ReferentialAction::NoAction,
            ReferentialAction::NoAction,
            false,
            // A user-command source records one undo snapshot.
            Some("add 1:n relationship".to_string()),
        )
        .await
        .expect("add compound relationship");
        assert_eq!(
            db.describe_table("City".to_string(), None)
                .await
                .unwrap()
                .outbound_relationships
                .len(),
            1,
        );
        // One undo step removes the whole relationship (ADR-0013/0006).
        db.undo().await.unwrap().expect("undo applied");
        assert!(
            db.describe_table("City".to_string(), None)
                .await
                .unwrap()
                .outbound_relationships
                .is_empty(),
            "undo removed the compound relationship in one step",
        );
    });
}

#[test]
fn compound_fk_partial_pk_reference_is_refused() {
    let (_p, db, _dir) = open_project_db();
    let rt = rt();
    rt.block_on(async {
        seed_compound(&db).await;
        // Referencing only one column of Region's 2-column PK (F-A:
        // must reference the full PK).
        let err = db
            .add_relationship(
                None,
                "Region".to_string(),
                vec!["country".to_string()],
                "City".to_string(),
                vec!["country".to_string()],
                ReferentialAction::NoAction,
                ReferentialAction::NoAction,
                false,
                None,
            )
            .await;
        assert!(err.is_err(), "a partial-PK reference must be refused (F-A)");
    });
}

#[test]
fn show_relationship_carries_compound_columns_into_diagram_data() {
    // ADR-0044 §2.4: the `show relationship` diagram payload carries
    // both paired columns on each side so the renderer can route the
    // bus + pairing line.
    let (_p, db, _dir) = open_project_db();
    let rt = rt();
    rt.block_on(async {
        seed_compound(&db).await;
        db.add_relationship(
            Some("city_region".to_string()),
            "Region".to_string(),
            vec!["country".to_string(), "code".to_string()],
            "City".to_string(),
            vec!["country".to_string(), "region_code".to_string()],
            ReferentialAction::NoAction,
            ReferentialAction::NoAction,
            false,
            None,
        )
        .await
        .expect("add compound relationship");

        let data = db
            .show_relationship("city_region".to_string())
            .await
            .expect("ok")
            .expect("found");
        // child = FK holder (City), parent = referenced (Region).
        assert_eq!(data.child.name, "City");
        assert_eq!(data.parent.name, "Region");
        assert_eq!(
            data.rel.child_columns,
            vec!["country".to_string(), "region_code".to_string()],
        );
        assert_eq!(
            data.rel.parent_columns,
            vec!["country".to_string(), "code".to_string()],
        );
    });
}