use super::{
    AedbConfig, AedbError, AedbInstance, ColumnDef, ColumnType, ConsistencyMode, DdlOperation,
    Expr, Mutation, Query, QueryError, QueryOptions, ReadAssertion, Row, TransactionEnvelope,
    Value, WriteClass, WriteIntent, create_table,
};
use crate::commit::tx::ReadSet;
use crate::commit::validation::CompareOp;
use tempfile::tempdir;

#[tokio::test]
async fn envelope_mutation_count_cap_rejects_oversized_envelope() {
    let dir = tempdir().expect("temp");
    let config = AedbConfig {
        max_mutations_per_envelope: 4,
        ..AedbConfig::default()
    };
    let db = AedbInstance::open_anonymous(config, dir.path()).expect("open");
    db.create_project("p").await.expect("project");

    let too_many: Vec<Mutation> = (0..5u8)
        .map(|i| Mutation::KvSet {
            project_id: "p".into(),
            scope_id: "app".into(),
            key: vec![b'k', i],
            value: vec![i],
        })
        .collect();
    let err = db
        .commit_envelope(TransactionEnvelope {
            caller: None,
            idempotency_key: None,
            write_class: WriteClass::Standard,
            assertions: Vec::new(),
            read_set: ReadSet::default(),
            write_intent: WriteIntent {
                mutations: too_many,
            },
            base_seq: 0,
        })
        .await
        .expect_err("envelope mutation count cap must reject");
    match err {
        AedbError::Validation(msg) => {
            assert!(
                msg.contains("max_mutations_per_envelope") && msg.contains("mutations=5"),
                "informative limit message expected, got: {msg}"
            );
        }
        other => panic!("expected Validation error, got: {other:?}"),
    }

    let ok_batch: Vec<Mutation> = (0..4u8)
        .map(|i| Mutation::KvSet {
            project_id: "p".into(),
            scope_id: "app".into(),
            key: vec![b'k', i],
            value: vec![i],
        })
        .collect();
    db.commit_envelope(TransactionEnvelope {
        caller: None,
        idempotency_key: None,
        write_class: WriteClass::Standard,
        assertions: Vec::new(),
        read_set: ReadSet::default(),
        write_intent: WriteIntent {
            mutations: ok_batch,
        },
        base_seq: 0,
    })
    .await
    .expect("envelope within mutation cap commits");
}

#[tokio::test]
async fn envelope_assertion_count_cap_rejects_oversized_envelope() {
    let dir = tempdir().expect("temp");
    let config = AedbConfig {
        max_read_assertions_per_envelope: 2,
        ..AedbConfig::default()
    };
    let db = AedbInstance::open_anonymous(config, dir.path()).expect("open");
    db.create_project("p").await.expect("project");

    let too_many: Vec<ReadAssertion> = (0..3u8)
        .map(|i| ReadAssertion::KeyExists {
            project_id: "p".into(),
            scope_id: "app".into(),
            key: vec![b'k', i],
            expected: false,
        })
        .collect();
    let err = db
        .commit_envelope(TransactionEnvelope {
            caller: None,
            idempotency_key: None,
            write_class: WriteClass::Standard,
            assertions: too_many,
            read_set: ReadSet::default(),
            write_intent: WriteIntent {
                mutations: vec![Mutation::KvSet {
                    project_id: "p".into(),
                    scope_id: "app".into(),
                    key: b"k".to_vec(),
                    value: b"v".to_vec(),
                }],
            },
            base_seq: 0,
        })
        .await
        .expect_err("envelope assertion count cap must reject");
    match err {
        AedbError::Validation(msg) => {
            assert!(
                msg.contains("max_read_assertions_per_envelope") && msg.contains("assertions=3"),
                "informative limit message expected, got: {msg}"
            );
        }
        other => panic!("expected Validation error, got: {other:?}"),
    }
}

#[tokio::test]
async fn list_with_total_respects_scan_bounds_for_count_queries() {
    let dir = tempdir().expect("temp");
    let config = AedbConfig {
        max_scan_rows: 3,
        ..AedbConfig::default()
    };
    let db = AedbInstance::open_anonymous(config, dir.path()).expect("open");
    db.create_project("p").await.expect("project");
    db.create_scope("p", "app").await.expect("scope");
    create_table(
        &db,
        "p",
        "app",
        "items",
        vec![
            ColumnDef {
                name: "id".into(),
                col_type: ColumnType::Integer,
                nullable: false,
            },
            ColumnDef {
                name: "name".into(),
                col_type: ColumnType::Text,
                nullable: false,
            },
        ],
        vec!["id"],
    )
    .await;

    for id in 1_i64..=5_i64 {
        db.commit(Mutation::Upsert {
            project_id: "p".into(),
            scope_id: "app".into(),
            table_name: "items".into(),
            primary_key: vec![Value::Integer(id)],
            row: Row::from_values(vec![
                Value::Integer(id),
                Value::Text(format!("item-{id}").into()),
            ]),
        })
        .await
        .expect("insert item");
    }

    let err = db
        .list_with_total(
            "p",
            "app",
            Query::select(&["id"]).from("items"),
            None,
            None,
            2,
            ConsistencyMode::AtLatest,
        )
        .await
        .expect_err("count query should respect scan bounds");
    assert!(matches!(
        err,
        QueryError::ScanBoundExceeded {
            estimated_rows: 5,
            max_scan_rows: 3
        }
    ));
}

#[tokio::test]
async fn delete_where_rejects_deep_predicate_trees() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
    db.create_project("p").await.expect("project");
    db.create_scope("p", "app").await.expect("scope");
    create_table(
        &db,
        "p",
        "app",
        "items",
        vec![ColumnDef {
            name: "id".into(),
            col_type: ColumnType::Integer,
            nullable: false,
        }],
        vec!["id"],
    )
    .await;

    let mut predicate = Expr::IsNotNull("id".into());
    for _ in 0..40 {
        predicate = predicate.and(Expr::IsNotNull("id".into()));
    }

    let err = db
        .commit(Mutation::DeleteWhere {
            project_id: "p".into(),
            scope_id: "app".into(),
            table_name: "items".into(),
            predicate,
            limit: None,
        })
        .await
        .expect_err("deep predicate should be rejected");
    assert!(matches!(err, AedbError::Validation(ref msg) if msg.contains("expression depth")));
}

#[tokio::test]
async fn delete_where_respects_scan_budget() {
    let dir = tempdir().expect("temp");
    let config = AedbConfig {
        max_scan_rows: 3,
        ..AedbConfig::default()
    };
    let db = AedbInstance::open_anonymous(config, dir.path()).expect("open");
    db.create_project("p").await.expect("project");
    db.create_scope("p", "app").await.expect("scope");
    create_table(
        &db,
        "p",
        "app",
        "items",
        vec![ColumnDef {
            name: "id".into(),
            col_type: ColumnType::Integer,
            nullable: false,
        }],
        vec!["id"],
    )
    .await;

    for id in 1_i64..=5_i64 {
        db.commit(Mutation::Insert {
            project_id: "p".into(),
            scope_id: "app".into(),
            table_name: "items".into(),
            primary_key: vec![Value::Integer(id)],
            row: Row::from_values(vec![Value::Integer(id)]),
        })
        .await
        .expect("insert item");
    }

    let err = db
        .commit(Mutation::DeleteWhere {
            project_id: "p".into(),
            scope_id: "app".into(),
            table_name: "items".into(),
            predicate: Expr::IsNotNull("id".into()),
            limit: None,
        })
        .await
        .expect_err("delete_where should honor scan budget");
    assert!(
        matches!(err, AedbError::Validation(ref msg) if msg.contains("mutation scan bound exceeded"))
    );

    for id in 1_i64..=5_i64 {
        let row = db
            .query_no_auth(
                "p",
                "app",
                Query::select(&["id"])
                    .from("items")
                    .where_(Expr::Eq("id".into(), Value::Integer(id)))
                    .limit(1),
                QueryOptions::default(),
            )
            .await
            .expect("query item by id");
        assert_eq!(row.rows.len(), 1, "row {id} should still exist");
    }
}

#[tokio::test]
async fn count_compare_assertions_respect_scan_budget() {
    let dir = tempdir().expect("temp");
    let config = AedbConfig {
        max_scan_rows: 3,
        ..AedbConfig::default()
    };
    let db = AedbInstance::open_anonymous(config, dir.path()).expect("open");
    db.create_project("p").await.expect("project");
    db.create_scope("p", "app").await.expect("scope");
    create_table(
        &db,
        "p",
        "app",
        "items",
        vec![ColumnDef {
            name: "id".into(),
            col_type: ColumnType::Integer,
            nullable: false,
        }],
        vec!["id"],
    )
    .await;

    for id in 1_i64..=5_i64 {
        db.commit(Mutation::Insert {
            project_id: "p".into(),
            scope_id: "app".into(),
            table_name: "items".into(),
            primary_key: vec![Value::Integer(id)],
            row: Row::from_values(vec![Value::Integer(id)]),
        })
        .await
        .expect("insert item");
    }

    let err = db
        .commit_envelope(TransactionEnvelope {
            caller: None,
            idempotency_key: None,
            write_class: WriteClass::Standard,
            assertions: vec![ReadAssertion::CountCompare {
                project_id: "p".into(),
                scope_id: "app".into(),
                table_name: "items".into(),
                filter: None,
                op: CompareOp::Eq,
                threshold: 5,
            }],
            read_set: Default::default(),
            write_intent: WriteIntent {
                mutations: vec![Mutation::KvSet {
                    project_id: "p".into(),
                    scope_id: "app".into(),
                    key: b"guard".to_vec(),
                    value: b"1".to_vec(),
                }],
            },
            base_seq: 0,
        })
        .await
        .expect_err("assertion scan should be bounded");
    assert!(
        matches!(err, AedbError::Validation(ref msg) if msg.contains("assertion scan bound exceeded"))
    );
}

#[tokio::test]
async fn table_values_respect_max_table_value_bytes() {
    let dir = tempdir().expect("temp");
    let config = AedbConfig {
        max_table_value_bytes: 8,
        ..AedbConfig::default()
    };
    let db = AedbInstance::open_anonymous(config, dir.path()).expect("open");
    db.create_project("p").await.expect("project");
    create_table(
        &db,
        "p",
        "app",
        "docs",
        vec![
            ColumnDef {
                name: "id".into(),
                col_type: ColumnType::Integer,
                nullable: false,
            },
            ColumnDef {
                name: "body".into(),
                col_type: ColumnType::Text,
                nullable: false,
            },
        ],
        vec!["id"],
    )
    .await;

    let err = db
        .commit(Mutation::Upsert {
            project_id: "p".into(),
            scope_id: "app".into(),
            table_name: "docs".into(),
            primary_key: vec![Value::Integer(1)],
            row: Row::from_values(vec![
                Value::Integer(1),
                Value::Text("too-large-body".into()),
            ]),
        })
        .await
        .expect_err("oversized cell should be rejected");
    assert!(matches!(err, AedbError::Validation(ref msg) if msg.contains("max_table_value_bytes")));
}

#[tokio::test]
async fn cascade_delete_respects_max_depth() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
    db.create_project("p").await.expect("project");
    db.create_scope("p", "app").await.expect("scope");

    for depth in 0..=9 {
        let table = format!("t{depth}");
        create_table(
            &db,
            "p",
            "app",
            &table,
            vec![
                ColumnDef {
                    name: "id".into(),
                    col_type: ColumnType::Integer,
                    nullable: false,
                },
                ColumnDef {
                    name: "parent_id".into(),
                    col_type: ColumnType::Integer,
                    nullable: true,
                },
            ],
            vec!["id"],
        )
        .await;
        if depth > 0 {
            db.commit_ddl(DdlOperation::AlterTable {
                project_id: "p".into(),
                scope_id: "app".into(),
                table_name: table.clone(),
                alteration: crate::catalog::schema::TableAlteration::AddForeignKey(
                    crate::catalog::schema::ForeignKey {
                        name: format!("fk_t{depth}_to_t{}", depth - 1),
                        columns: vec!["parent_id".into()],
                        references_project_id: "p".into(),
                        references_scope_id: "app".into(),
                        references_table: format!("t{}", depth - 1),
                        references_columns: vec!["id".into()],
                        on_delete: crate::catalog::schema::ForeignKeyAction::Cascade,
                        on_update: crate::catalog::schema::ForeignKeyAction::Cascade,
                    },
                ),
            })
            .await
            .expect("add fk");
        }
        db.commit(Mutation::Insert {
            project_id: "p".into(),
            scope_id: "app".into(),
            table_name: table,
            primary_key: vec![Value::Integer(depth as i64)],
            row: Row::from_values(vec![
                Value::Integer(depth as i64),
                if depth == 0 {
                    Value::Null
                } else {
                    Value::Integer((depth - 1) as i64)
                },
            ]),
        })
        .await
        .expect("insert row");
    }

    let err = db
        .commit(Mutation::Delete {
            project_id: "p".into(),
            scope_id: "app".into(),
            table_name: "t0".into(),
            primary_key: vec![Value::Integer(0)],
        })
        .await
        .expect_err("cascade depth should be bounded");
    assert!(
        matches!(err, AedbError::Validation(ref msg) if msg.contains("cascade delete depth exceeded"))
    );
}
