use std::collections::HashSet;

use anyhow::Error;
use culpa::throws;
use dittolive_ditto::{
    error::{CoreApiErrorKind, ErrorKind},
    prelude::*,
};
use serde_json::json;

mod common;

#[allow(dead_code)]
struct StoreDqlTest {
    ditto: Ditto,
    collection_name: String,
    bmw: serde_json::Value,
    porsche: serde_json::Value,
    bmw_id: DocumentId,
    porsche_id: DocumentId,
}

#[throws]
async fn setup() -> StoreDqlTest {
    let ditto = common::get_builder_ditto(None)?;
    ditto.disable_sync_with_v3()?;

    let uuid = uuid::Uuid::new_v4();
    let collection_name = format!("cars_{}", uuid.to_string().replace('-', "_"));

    let bmw_id = DocumentId::new(&"BMW".to_string()).unwrap();
    let bmw = json!({
        "_id": bmw_id, "color": "blue", "mileage": 123, "driver": null
    });

    let porsche_id = DocumentId::new(&"Porsche".to_string()).unwrap();
    let porsche = json!({
        "_id": porsche_id, "color": "silver", "mileage": 1234
    });

    ditto
        .store()
        .execute_v2((
            format!("INSERT INTO {collection_name} DOCUMENTS (:bmw), (:porsche)"),
            json!({
                "bmw": bmw,
                "porsche": porsche
            }),
        ))
        .await?;

    StoreDqlTest {
        ditto,
        collection_name,
        bmw,
        porsche,
        bmw_id,
        porsche_id,
    }
}

#[throws]
#[tokio::test]
async fn test_execute_returns_all_documents_for_select_star() {
    let test = setup().await?;

    let query_result = test
        .ditto
        .store()
        .execute_v2(format!("SELECT * FROM {}", test.collection_name))
        .await?;

    assert_eq!(query_result.item_count(), 2);
    let mut items = query_result
        .iter()
        .map(|it| it.deserialize_value::<serde_json::Value>().unwrap())
        .collect::<Vec<_>>();
    items.sort_by(|a, b| a["_id"].as_str().unwrap().cmp(b["_id"].as_str().unwrap()));
    assert_eq!(items[0], test.bmw);
    assert_eq!(items[1], test.porsche);
}

#[throws]
#[tokio::test]
async fn test_execute_returns_matching_documents_for_select_star_where() {
    let test = setup().await?;

    let query_result = test
        .ditto
        .store()
        .execute_v2(format!(
            "SELECT * FROM {} WHERE _id = 'BMW'",
            test.collection_name
        ))
        .await?;

    assert_eq!(query_result.item_count(), 1);
    let items = query_result
        .iter()
        .map(|it| it.deserialize_value::<serde_json::Value>().unwrap())
        .collect::<Vec<_>>();
    let keys = items[0].as_object().unwrap().keys().collect::<Vec<_>>();
    assert_eq!(keys, vec!["_id", "color", "driver", "mileage"]);
    assert_eq!(items[0]["color"], json!("blue"));
    assert_eq!(items[0]["mileage"], json!(123));
    assert_eq!(items[0]["driver"], json!(null));
}

#[throws]
#[tokio::test]
async fn test_execute_returns_parameter_for_show() {
    let test = setup().await?;

    let query_result = test
        .ditto
        .store()
        .execute_v2("SHOW example_parameter")
        .await?;

    assert_eq!(query_result.item_count(), 1);
    let items = query_result
        .iter()
        .map(|it| it.deserialize_value::<serde_json::Value>().unwrap())
        .collect::<Vec<_>>();
    assert!(!items[0]["example_parameter"].is_null());
}

#[throws]
#[tokio::test]
async fn test_execute_show_throws_when_given_unknown_parameter() {
    let test = setup().await?;

    let result = test
        .ditto
        .store()
        .execute_v2("SHOW nonexistent_parameter")
        .await;

    let Err(error) = result else {
        panic!("Expected error");
    };

    let error_kind = error.kind();

    use dittolive_ditto::error::ErrorKind;
    let ErrorKind::CoreApi(core_error) = error_kind else {
        panic!("Expected CoreApi error, got {error_kind}");
    };

    use dittolive_ditto::error::CoreApiErrorKind;
    assert!(core_error == CoreApiErrorKind::ParameterQuery);

    assert!(error
        .to_string()
        .contains("unknown parameter 'nonexistent_parameter'"));
}

#[throws]
#[tokio::test]
async fn test_execute_returns_many_parameters_for_show_all() {
    let test = setup().await?;

    let query_result = test.ditto.store().execute_v2("SHOW ALL").await?;
    assert_eq!(query_result.item_count(), 1);

    let items = query_result
        .iter()
        .map(|it| it.deserialize_value::<serde_json::Value>().unwrap())
        .collect::<Vec<_>>();
    assert!(!items[0]["example_parameter"].is_null());
}

#[throws]
#[tokio::test]
async fn test_execute_alter_system_has_visible_effect() {
    let test = setup().await?;

    let set_examples = vec![
        "ALTER SYSTEM SET example_parameter = 456",
        "ALTER SYSTEM SET example_parameter TO 456",
    ];

    for example in set_examples {
        let query_result = test
            .ditto
            .store()
            .execute_v2("ALTER SYSTEM SET example_parameter = 321")
            .await?;
        assert_eq!(query_result.item_count(), 1);

        let item = query_result
            .get_item(0)
            .unwrap()
            .deserialize_value::<serde_json::Value>()?;
        assert_eq!(item["example_parameter"], json!(321));

        let query_result = test.ditto.store().execute_v2(example).await?;
        assert_eq!(query_result.item_count(), 1);

        let item = query_result
            .get_item(0)
            .unwrap()
            .deserialize_value::<serde_json::Value>()?;
        assert_eq!(item["example_parameter"], json!(456));
    }
}

#[throws]
#[tokio::test]
async fn test_execute_alter_system_supports_bool() {
    let test = setup().await?;

    // Check default true state
    let query_result = test
        .ditto
        .store()
        .execute_v2("SHOW example_bool_parameter")
        .await?;
    assert_eq!(query_result.item_count(), 1);
    let item = query_result
        .get_item(0)
        .unwrap()
        .deserialize_value::<serde_json::Value>()?;
    assert_eq!(item["example_bool_parameter"], json!(true));

    // Set false
    let query_result = test
        .ditto
        .store()
        .execute_v2("ALTER SYSTEM SET example_bool_parameter = false")
        .await?;
    assert_eq!(query_result.item_count(), 1);
    let item = query_result
        .get_item(0)
        .unwrap()
        .deserialize_value::<serde_json::Value>()?;
    assert_eq!(item["example_bool_parameter"], json!(false));

    // Check false state
    let query_result = test
        .ditto
        .store()
        .execute_v2("SHOW example_bool_parameter")
        .await?;
    assert_eq!(query_result.item_count(), 1);
    let item = query_result
        .get_item(0)
        .unwrap()
        .deserialize_value::<serde_json::Value>()?;
    assert_eq!(item["example_bool_parameter"], json!(false));
}

#[throws]
#[tokio::test]
async fn test_execute_alter_system_supports_durations() {
    let test = setup().await?;

    let query_result = test
        .ditto
        .store()
        .execute_v2("SHOW example_duration_parameter")
        .await?;
    assert_eq!(query_result.item_count(), 1);
    let item = query_result
        .get_item(0)
        .unwrap()
        .deserialize_value::<serde_json::Value>()?;
    assert_eq!(item["example_duration_parameter"], json!(200));

    let query_result = test
        .ditto
        .store()
        .execute_v2("ALTER SYSTEM SET example_duration_parameter = 500")
        .await?;
    assert_eq!(query_result.item_count(), 1);
    let item = query_result
        .get_item(0)
        .unwrap()
        .deserialize_value::<serde_json::Value>()?;
    assert_eq!(item["example_duration_parameter"], json!(500));

    let query_result = test
        .ditto
        .store()
        .execute_v2("SHOW example_duration_parameter")
        .await?;
    assert_eq!(query_result.item_count(), 1);
    let item = query_result
        .get_item(0)
        .unwrap()
        .deserialize_value::<serde_json::Value>()?;
    assert_eq!(item["example_duration_parameter"], json!(500));

    let query_result = test
        .ditto
        .store()
        .execute_v2("ALTER SYSTEM SET example_duration_parameter = duration_cast('1s')")
        .await?;
    assert_eq!(query_result.item_count(), 1);
    let item = query_result
        .get_item(0)
        .unwrap()
        .deserialize_value::<serde_json::Value>()?;
    assert_eq!(item["example_duration_parameter"], json!(1000));

    let query_result = test
        .ditto
        .store()
        .execute_v2("SHOW example_duration_parameter")
        .await?;
    assert_eq!(query_result.item_count(), 1);
    let item = query_result
        .get_item(0)
        .unwrap()
        .deserialize_value::<serde_json::Value>()?;
    assert_eq!(item["example_duration_parameter"], json!(1000));

    let query_result = test
        .ditto
        .store()
        .execute_v2("ALTER SYSTEM RESET example_duration_parameter")
        .await?;
    assert_eq!(query_result.item_count(), 1);
    let item = query_result
        .get_item(0)
        .unwrap()
        .deserialize_value::<serde_json::Value>()?;
    assert_eq!(item["example_duration_parameter"], json!(200));

    let query_result = test
        .ditto
        .store()
        .execute_v2("SHOW example_duration_parameter")
        .await?;
    assert_eq!(query_result.item_count(), 1);
    let item = query_result
        .get_item(0)
        .unwrap()
        .deserialize_value::<serde_json::Value>()?;
    assert_eq!(item["example_duration_parameter"], json!(200));
}

#[throws]
#[tokio::test]
async fn test_execute_alter_system_supports_strings() {
    let test = setup().await?;

    let query_result = test
        .ditto
        .store()
        .execute_v2("SHOW example_string_parameter")
        .await?;
    assert_eq!(query_result.item_count(), 1);
    let item = query_result
        .get_item(0)
        .unwrap()
        .deserialize_value::<serde_json::Value>()?;
    assert_eq!(item["example_string_parameter"], json!("default"));

    let query_result = test
        .ditto
        .store()
        .execute_v2("ALTER SYSTEM SET example_string_parameter = 'test'")
        .await?;
    assert_eq!(query_result.item_count(), 1);
    let item = query_result
        .get_item(0)
        .unwrap()
        .deserialize_value::<serde_json::Value>()?;
    assert_eq!(item["example_string_parameter"], json!("test"));

    let query_result = test
        .ditto
        .store()
        .execute_v2("SHOW example_string_parameter")
        .await?;
    assert_eq!(query_result.item_count(), 1);
    let item = query_result
        .get_item(0)
        .unwrap()
        .deserialize_value::<serde_json::Value>()?;
    assert_eq!(item["example_string_parameter"], json!("test"));

    let query_result = test
        .ditto
        .store()
        .execute_v2("ALTER SYSTEM RESET example_string_parameter")
        .await?;
    assert_eq!(query_result.item_count(), 1);
    let item = query_result
        .get_item(0)
        .unwrap()
        .deserialize_value::<serde_json::Value>()?;
    assert_eq!(item["example_string_parameter"], json!("default"));

    let query_result = test
        .ditto
        .store()
        .execute_v2("SHOW example_string_parameter")
        .await?;
    assert_eq!(query_result.item_count(), 1);
    let item = query_result
        .get_item(0)
        .unwrap()
        .deserialize_value::<serde_json::Value>()?;
    assert_eq!(item["example_string_parameter"], json!("default"));
}

#[throws]
#[tokio::test]
async fn test_execute_alter_system_supports_maps() {
    let test = setup().await?;

    let query_result = test
        .ditto
        .store()
        .execute_v2("SHOW example_map_parameter")
        .await?;
    assert_eq!(query_result.item_count(), 1);
    let item = query_result
        .get_item(0)
        .unwrap()
        .deserialize_value::<serde_json::Value>()?;
    assert_eq!(item["example_map_parameter"], json!({}));

    let query_result = test
        .ditto
        .store()
        .execute_v2("ALTER SYSTEM SET example_map_parameter = {'foo': 'bar'}")
        .await?;
    assert_eq!(query_result.item_count(), 1);
    let item = query_result
        .get_item(0)
        .unwrap()
        .deserialize_value::<serde_json::Value>()?;
    assert_eq!(item["example_map_parameter"], json!({"foo": "bar"}));

    let query_result = test
        .ditto
        .store()
        .execute_v2("SHOW example_map_parameter")
        .await?;
    assert_eq!(query_result.item_count(), 1);
    let item = query_result
        .get_item(0)
        .unwrap()
        .deserialize_value::<serde_json::Value>()?;
    assert_eq!(item["example_map_parameter"], json!({"foo": "bar"}));

    let query_result = test
        .ditto
        .store()
        .execute_v2("ALTER SYSTEM RESET example_map_parameter")
        .await?;
    assert_eq!(query_result.item_count(), 1);
    let item = query_result
        .get_item(0)
        .unwrap()
        .deserialize_value::<serde_json::Value>()?;
    assert_eq!(item["example_map_parameter"], json!({}));

    let query_result = test
        .ditto
        .store()
        .execute_v2("SHOW example_map_parameter")
        .await?;
    assert_eq!(query_result.item_count(), 1);
    let item = query_result
        .get_item(0)
        .unwrap()
        .deserialize_value::<serde_json::Value>()?;
    assert_eq!(item["example_map_parameter"], json!({}));
}

#[throws]
#[tokio::test]
async fn test_execute_alter_system_supports_arrays() {
    let test = setup().await?;

    let query_result = test
        .ditto
        .store()
        .execute_v2("SHOW example_array_parameter")
        .await?;
    assert_eq!(query_result.item_count(), 1);
    let item = query_result
        .get_item(0)
        .unwrap()
        .deserialize_value::<serde_json::Value>()?;
    assert_eq!(item["example_array_parameter"], json!([]));

    let query_result = test
        .ditto
        .store()
        .execute_v2("ALTER SYSTEM SET example_array_parameter = ['foo', 'bar']")
        .await?;
    assert_eq!(query_result.item_count(), 1);
    let item = query_result
        .get_item(0)
        .unwrap()
        .deserialize_value::<serde_json::Value>()?;
    assert_eq!(item["example_array_parameter"], json!(["foo", "bar"]));

    let query_result = test
        .ditto
        .store()
        .execute_v2("SHOW example_array_parameter")
        .await?;
    assert_eq!(query_result.item_count(), 1);
    let item = query_result
        .get_item(0)
        .unwrap()
        .deserialize_value::<serde_json::Value>()?;
    assert_eq!(item["example_array_parameter"], json!(["foo", "bar"]));

    let query_result = test
        .ditto
        .store()
        .execute_v2("ALTER SYSTEM RESET example_array_parameter")
        .await?;
    assert_eq!(query_result.item_count(), 1);
    let item = query_result
        .get_item(0)
        .unwrap()
        .deserialize_value::<serde_json::Value>()?;
    assert_eq!(item["example_array_parameter"], json!([]));

    let query_result = test
        .ditto
        .store()
        .execute_v2("SHOW example_array_parameter")
        .await?;
    assert_eq!(query_result.item_count(), 1);
    let item = query_result
        .get_item(0)
        .unwrap()
        .deserialize_value::<serde_json::Value>()?;
    assert_eq!(item["example_array_parameter"], json!([]));
}

#[throws]
#[tokio::test]
async fn test_execute_alter_system_set_throws_when_given_unknown_parameter() {
    let test = setup().await?;

    let result = test
        .ditto
        .store()
        .execute_v2("ALTER SYSTEM SET nonexistent_parameter = 123")
        .await;
    let Err(error) = result else {
        panic!("Expected error, got Ok");
    };

    assert!(error
        .to_string()
        .contains("query error: unknown parameter 'nonexistent_parameter'"));
    let ErrorKind::CoreApi(CoreApiErrorKind::ParameterQuery) = error.kind() else {
        panic!("Unexpected error kind: {error}");
    };
}

#[throws]
#[tokio::test]
async fn test_execute_alter_system_set_throws_when_given_wrong_type() {
    let test = setup().await?;

    let result = test
        .ditto
        .store()
        .execute_v2("ALTER SYSTEM SET example_bool_parameter = 42")
        .await;
    let Err(error) = result else {
        panic!("Expected error, got Ok");
    };

    assert!(
        error.to_string().contains(
            "query error: type error: value U64(42) doesn't match system parameter \
             example_bool_parameter type of bool"
        ),
        "unexpected error string: {}",
        error
    );
    let ErrorKind::CoreApi(CoreApiErrorKind::ParameterQuery) = error.kind() else {
        panic!("Unexpected error kind: {error}");
    };
}

#[throws]
#[tokio::test]
async fn test_execute_alter_system_reset_throws_when_given_unknown_parameter() {
    let test = setup().await?;

    let result = test
        .ditto
        .store()
        .execute_v2("ALTER SYSTEM RESET nonexistent_parameter")
        .await;
    let Err(error) = result else {
        panic!("Expected error, got Ok");
    };

    assert!(error
        .to_string()
        .contains("query error: unknown parameter 'nonexistent_parameter'"));
    let ErrorKind::CoreApi(CoreApiErrorKind::ParameterQuery) = error.kind() else {
        panic!("Unexpected error kind: {error}");
    };
}

#[throws]
#[tokio::test]
async fn test_execute_alter_system_set_throws_when_given_out_of_bounds_value() {
    let test = setup().await?;

    let result = test
        .ditto
        .store()
        .execute_v2("ALTER SYSTEM SET example_parameter = 99999")
        .await;
    let Err(error) = result else {
        panic!("Expected error, got Ok");
    };

    assert!(
        error.to_string().contains(
            "query error: parameter validation for example_parameter failed: value U64(99999) is \
             outside of acceptable range"
        ),
        "unexpected error string: {}",
        error
    );
    let ErrorKind::CoreApi(CoreApiErrorKind::ParameterQuery) = error.kind() else {
        panic!("Unexpected error kind: {error}");
    };
}

#[throws]
#[tokio::test]
async fn test_execute_returns_mutated_document_ids_for_mutating_queries() {
    let test = setup().await?;

    // [(Query, Args, Expected)]
    // See 'fn setup' for BMW and Porsche seed data
    let mutating_query_examples = vec![
        (
            format!(
                "EVICT FROM {} WHERE _id = 'BMW' OR _id = 'Porsche'",
                test.collection_name
            ),
            None,
            vec!["BMW", "Porsche"],
        ),
        (
            format!(
                "INSERT INTO {} DOCUMENTS (:doc1), (:doc2)",
                test.collection_name
            ),
            Some(
                json!({ "doc1": { "_id": "Mustang",  "blah": 123}, "doc2": { "_id": "Ford", "blub": 446}}),
            ),
            vec!["Ford", "Mustang"],
        ),
    ];

    for (query, args, expected) in mutating_query_examples {
        let query_result = test
            .ditto
            .store()
            .execute_v2((query, args.unwrap_or(json!({}))))
            .await?;
        let mutated = query_result
            .mutated_document_ids()
            .into_iter()
            .map(|it| it.to_string())
            .collect::<HashSet<_>>();
        let expected = expected
            .into_iter()
            .map(|it| it.to_string())
            .collect::<HashSet<_>>();
        assert_eq!(mutated, expected);
    }
}

#[throws]
#[tokio::test]
async fn test_execute_throws_when_given_an_invalid_query() {
    let test = setup().await?;

    let result = test.ditto.store().execute_v2("make me a coffee!").await;
    let Err(error) = result else {
        panic!("Expected error, got Ok");
    };

    assert!(error.to_string().contains(r#"Invalid query: `DQL parser error: Unexpected token make at line 1 column 1 near "make me a coffee!""#));
    let ErrorKind::CoreApi(CoreApiErrorKind::DqlQueryCompilation) = error.kind() else {
        panic!("Unexpected error kind: {error}");
    };
}

#[throws]
#[tokio::test]
async fn test_execute_throws_when_given_an_empty_query() {
    let test = setup().await?;

    let result = test.ditto.store().execute_v2("").await;
    let Err(error) = result else {
        panic!("Expected error, got Ok");
    };

    assert!(error.to_string().contains(r#"Invalid query: `DQL parser error: Unexpected end of input; expected a DQL statement`. For more information on Ditto's query language see: https://ditto.com/link/dql-guide"#));
    let ErrorKind::CoreApi(CoreApiErrorKind::DqlQueryCompilation) = error.kind() else {
        panic!("Unexpected error kind: {error}");
    };
}

#[throws]
#[tokio::test]
async fn test_execute_does_not_throw_when_querying_a_collection_that_does_not_exist() {
    let test = setup().await?;

    let query_result = test
        .ditto
        .store()
        .execute_v2("SELECT * FROM non_existant_collection")
        .await?;
    assert_eq!(query_result.item_count(), 0);
}

#[throws]
#[tokio::test]
async fn test_execute_performs_a_query_with_a_placeholder() {
    let test = setup().await?;

    let query_result = test
        .ditto
        .store()
        .execute_v2((
            format!("SELECT * FROM {} WHERE _id = :id", test.collection_name),
            json!({ "id": test.bmw_id }),
        ))
        .await?;
    assert_eq!(query_result.item_count(), 1);
}

#[throws]
#[tokio::test]
async fn test_execute_throws_for_a_query_with_a_numerical_placeholder() {
    let test = setup().await?;

    let result = test
        .ditto
        .store()
        .execute_v2((
            format!("SELECT * FROM {} WHERE _id = :1", test.collection_name),
            json!({ "1": test.bmw_id }),
        ))
        .await;
    let Err(error) = result else {
        panic!("Expected error, got Ok");
    };

    // This error includes the collection name, which is a new UUID each run
    // Therefore we check string containment excluding that part of the message
    assert!(error.to_string().contains(
        r#"Invalid query: `DQL parser error: Unexpected token 1 at line 1 column 70 near"#
    ));
    assert!(error.to_string().contains(r#"WHERE _id = :1"; expected identifier`. For more information on Ditto's query language see: https://ditto.com/link/dql-guide"#));
    let ErrorKind::CoreApi(CoreApiErrorKind::DqlQueryCompilation) = error.kind() else {
        panic!("Unexpected error kind: {error}");
    };
}

#[throws]
#[tokio::test]
async fn test_execute_throws_for_a_query_with_a_placeholder_for_the_collection_name() {
    let test = setup().await?;

    let result = test
        .ditto
        .store()
        .execute_v2(("SELECT * FROM :collection", json!({ "collection": "blah" })))
        .await;
    let Err(error) = result else {
        panic!("Expected error, got Ok");
    };

    assert!(error.to_string().contains(r#"Invalid query: `DQL parser error: Unexpected token : at line 1 column 15 near "SELECT * FROM :collection""#));
    let ErrorKind::CoreApi(CoreApiErrorKind::DqlQueryCompilation) = error.kind() else {
        panic!("Unexpected error kind: {error}");
    };
}

#[throws]
#[tokio::test]
async fn test_execute_accepts_queries_using_valid_types_for_placeholders() {
    let test = setup().await?;

    let valid_placeholders = vec![
        json!({ "mileage": 1230 }),
        json!({ "mileage": 123.456 }),
        json!({ "mileage": -123 }),
        json!({ "mileage": null }),
    ];

    for valid_placeholder in valid_placeholders {
        let query_result = test
            .ditto
            .store()
            .execute_v2((
                format!(
                    "SELECT * FROM {} WHERE mileage = :mileage",
                    test.collection_name
                ),
                valid_placeholder,
            ))
            .await?;
        assert_eq!(query_result.item_count(), 0);
    }
}

#[throws]
#[tokio::test]
async fn test_execute_matches_nil_values_in_arguments() {
    let test = setup().await?;

    let query_result = test
        .ditto
        .store()
        .execute_v2((
            format!(
                "SELECT * FROM {} WHERE driver = :driver OR (driver IS NOT MISSING AND driver IS \
                 NULL AND :driver IS NULL)",
                test.collection_name
            ),
            json!({ "driver": null }),
        ))
        .await?;

    assert_eq!(query_result.item_count(), 1);
    let items = query_result
        .iter()
        .map(|it| it.deserialize_value::<serde_json::Value>().unwrap())
        .collect::<Vec<_>>();
    let keys = items[0].as_object().unwrap().keys().collect::<Vec<_>>();
    assert_eq!(keys, vec!["_id", "color", "driver", "mileage"]);
    assert_eq!(items[0]["color"], json!("blue"));
    assert_eq!(items[0]["mileage"], json!(123));
    assert_eq!(items[0]["driver"], json!(null));
}

#[throws]
#[tokio::test]
async fn test_execute_throws_for_queries_that_have_placeholders_but_miss_the_corresponding_query_arguments(
) {
    let test = setup().await?;

    let result = test
        .ditto
        .store()
        .execute_v2((
            format!(
                "SELECT * FROM {} WHERE mileage = :mileage",
                test.collection_name
            ),
            json!({}),
        ))
        .await;
    let Err(error) = result else {
        panic!("Expected error, got Ok");
    };

    assert!(error.to_string().contains(r#"Invalid query: `Parameter mileage was not provided`. For more information on Ditto's query language see: https://ditto.com/link/dql-guide"#));
    let ErrorKind::CoreApi(CoreApiErrorKind::DqlQueryCompilation) = error.kind() else {
        panic!("Unexpected error kind: {error}");
    };
}

#[throws]
#[tokio::test]
async fn test_execute_ignores_type_mismatches_between_query_arguments_and_document_fields() {
    let test = setup().await?;

    let invalid_query_arguments = vec![
        // String
        json!({ "mileage": "all the way" }),
        // Boolean
        json!({ "mileage": true }),
        json!({ "mileage": false }),
        // Array
        json!({ "mileage": [1, 2, 3] }),
        json!({ "mileage": [1, "two", 3.0] }),
        // Object
        json!({ "mileage": { "a": 1, "b": 2 } }),
        json!({ "mileage": { "a": 1, "b": "two" } }),
        json!({ "mileage": { "0": 1, "1": 2 } }),
        // Binary
        json!({ "mileage": [0x00, 0x00, 0x00, 0x7b] }),
    ];

    for query_arguments in invalid_query_arguments {
        let query_result = test
            .ditto
            .store()
            .execute_v2((
                format!(
                    "SELECT * FROM {} WHERE mileage = :mileage",
                    test.collection_name
                ),
                query_arguments,
            ))
            .await?;
        assert_eq!(query_result.item_count(), 0);
    }
}

#[throws]
#[tokio::test]
async fn test_observers_returns_currently_active_store_observers() {
    let test = setup().await?;

    let query_string_1 = format!(
        "SELECT * FROM {} WHERE mileage = 1234",
        test.collection_name
    );
    let query_string_2 = format!(
        "SELECT * FROM {} WHERE mileage = 2345",
        test.collection_name
    );
    let query_string_3 = format!(
        "SELECT * FROM {} WHERE mileage = 3456",
        test.collection_name
    );
    let query_string_4 = format!(
        "SELECT * FROM {} WHERE mileage = 4567",
        test.collection_name
    );
    let query_string_5 = format!(
        "SELECT * FROM {} WHERE mileage = 5678",
        test.collection_name
    );
    let query_string_6 = format!(
        "SELECT * FROM {} WHERE mileage = 6789",
        test.collection_name
    );

    let store_observer_1 = test
        .ditto
        .store()
        .register_observer_v2(query_string_1, |_| {})?;
    let store_observer_2 = test
        .ditto
        .store()
        .register_observer_v2(query_string_2, |_| {})?;
    let store_observer_3 = test
        .ditto
        .store()
        .register_observer_v2(query_string_3, |_| {})?;

    let expected_observers =
        HashSet::from([&*store_observer_1, &*store_observer_2, &*store_observer_3]);
    let actual_observers_lock = test.ditto.store().observers();
    let actual_observers = actual_observers_lock
        .keys
        .iter()
        .map(|(key, _value)| &**key)
        .collect::<HashSet<_>>();

    // Note: Can't use assert_eq! because it requires Debug on values
    assert!(expected_observers == actual_observers);
    drop(actual_observers);
    drop(actual_observers_lock); // Release read lock, otherwise deadlock :(

    let store_observer_4 = test
        .ditto
        .store()
        .register_observer_v2(query_string_4, |_| {})?;
    let store_observer_5 = test
        .ditto
        .store()
        .register_observer_v2(query_string_5, |_| {})?;
    let store_observer_6 = test
        .ditto
        .store()
        .register_observer_v2(query_string_6, |_| {})?;

    store_observer_1.cancel();
    let expected_observers = HashSet::from([
        &*store_observer_2,
        &*store_observer_3,
        &*store_observer_4,
        &*store_observer_5,
        &*store_observer_6,
    ]);
    let actual_observers = test.ditto.store().observers();
    let actual_observers = actual_observers
        .keys
        .iter()
        .map(|(key, _value)| &**key)
        .collect::<HashSet<_>>();

    // Note: Can't use assert_eq! because it requires Debug on values
    assert!(expected_observers == actual_observers);
}

#[throws]
#[tokio::test]
async fn test_observers_returns_distinct_replication_observers_for_the_exact_same_query() {
    let test = setup().await?;

    let query_string = format!(
        "SELECT * FROM {} WHERE mileage = 1234",
        test.collection_name
    );

    let store_observer_1 = test
        .ditto
        .store()
        .register_observer_v2(&*query_string, |_| {})?;
    let store_observer_2 = test
        .ditto
        .store()
        .register_observer_v2(&*query_string, |_| {})?;

    assert!(store_observer_1 != store_observer_2);
    assert_eq!(test.ditto.store().observers().keys.len(), 2);
}

#[throws]
#[tokio::test]
async fn test_register_observer_registers_and_returns_a_store_observer_for_a_given_query() {
    let test = setup().await?;

    let query_string = format!("SELECT * FROM {} WHERE _id = :id", test.collection_name);
    let query_arguments = json!({ "id": "BMW" });
    let store_observer = test
        .ditto
        .store()
        .register_observer_v2((query_string, query_arguments), |_| {})?;

    assert_eq!(test.ditto.store().observers().keys.len(), 1);
    assert!(*test.ditto.store().observers().keys.iter().next().unwrap().0 == store_observer);
}

#[throws]
#[tokio::test]
async fn test_register_observer_registers_and_returns_a_different_store_observer_for_the_same_query(
) {
    let test = setup().await?;

    let query_string = format!("SELECT * FROM {} WHERE _id = :id", test.collection_name);
    let query_arguments = json!({ "id": test.bmw_id });

    let store_observer_1 = test
        .ditto
        .store()
        .register_observer_v2((&query_string, &query_arguments), |_| {})?;
    let store_observer_2 = test
        .ditto
        .store()
        .register_observer_v2((query_string, query_arguments), |_| {})?;
    assert!(store_observer_1 != store_observer_2);
}

#[throws]
#[tokio::test]
async fn test_register_observer_throws_when_given_an_invalid_query() {
    let test = setup().await?;

    let result = test
        .ditto
        .store()
        .register_observer_v2("make me a coffee!", |_| {});
    let Err(_error) = result else {
        panic!("Expected error, got Ok");
    };

    // TODO(nimo): Fix error message and add error assertion
    // assert_eq!(error.to_string(), "");
    // let ErrorKind::CoreApi(CoreApiErrorKind::ParameterQuery) = error.kind() else {
    //     panic!("Unexpected error kind: {error}");
    // };
}

#[throws]
#[tokio::test]
async fn test_register_observer_throws_when_given_an_empty_query() {
    let test = setup().await?;

    let result = test.ditto.store().register_observer_v2("", |_| {});
    let Err(_error) = result else {
        panic!("Expected error, got Ok");
    };

    // TODO(nimo): Fix error message and add error assertion
    // assert_eq!(error.to_string(), "");
    // let ErrorKind::CoreApi(CoreApiErrorKind::ParameterQuery) = error.kind() else {
    //     panic!("Unexpected error kind: {error}");
    // };
}

#[throws]
#[tokio::test]
async fn test_register_observer_throws_when_given_a_non_select_query() {
    let test = setup().await?;

    let non_select_query_examples = vec![
        (
            (
                format!("EXPLAIN SELECT * FROM {}", test.collection_name),
                None,
            ),
            "Unsupported feature: non-SELECT statement in store observer",
        ),
        (
            (format!("EVICT FROM {}", test.collection_name), None),
            "Unsupported feature: non-SELECT statement in store observer",
        ),
        (
            (
                format!(
                    "INSERT INTO {} DOCUMENTS (:doc1), (:doc2)",
                    test.collection_name
                ),
                Some(json!({ "doc1": { "blah": 123 }, "doc2": { "blub": 446 }})),
            ),
            "Unsupported feature: non-SELECT statement in store observer",
        ),
    ];

    for (query, _expected) in non_select_query_examples {
        let result = test.ditto.store().register_observer_v2(query, |_| {});
        let Err(_error) = result else {
            panic!("Expected error, got Ok");
        };

        // TODO(nimo): Fix error message and add error assertion
        // assert_eq!(error.to_string(), "");
        // let ErrorKind::CoreApi(CoreApiErrorKind::ParameterQuery) = error.kind() else {
        //     panic!("Unexpected error kind: {error}");
        // };
    }
}

#[throws]
#[tokio::test]
async fn test_unregister_observer_unregisters_and_cancels_a_previously_registered_subscription_returning_true(
) {
    let test = setup().await?;

    let query_string = format!("SELECT * FROM {} WHERE _id = :id", test.collection_name);
    let query_arguments = json!({ "id": test.bmw_id });
    let store_observer = test
        .ditto
        .store()
        .register_observer_v2((query_string, query_arguments), |_| {})?;

    assert_eq!(test.ditto.store().observers().keys.len(), 1);

    let observers = test.ditto.store().observers();
    let fetched_observer = observers.keys.iter().next().unwrap().0;
    assert!(*fetched_observer == store_observer);
    drop(observers);

    store_observer.cancel();
    assert!(store_observer.is_cancelled());
    assert_eq!(test.ditto.store().observers().keys.len(), 0);
}

#[throws]
#[tokio::test]
async fn test_unregister_observer_has_no_effect_and_returns_false_when_unregistering_the_same_store_observer_multiple_times(
) {
    let test = setup().await?;

    let query_string = format!("SELECT * FROM {} WHERE _id = :id", test.collection_name);
    let query_arguments = json!({ "id": test.bmw_id });
    let store_observer = test
        .ditto
        .store()
        .register_observer_v2((query_string, query_arguments), |_| {})?;

    assert_eq!(test.ditto.store().observers().keys.len(), 1);

    let observers = test.ditto.store().observers();
    let fetched_observer = observers.keys.iter().next().unwrap().0;
    assert!(*fetched_observer == store_observer);
    drop(observers);

    store_observer.cancel();
    assert!(store_observer.is_cancelled());
    assert_eq!(test.ditto.store().observers().keys.len(), 0);

    store_observer.cancel();
    assert!(store_observer.is_cancelled());
    assert_eq!(test.ditto.store().observers().keys.len(), 0);

    store_observer.cancel();
    assert!(store_observer.is_cancelled());
    assert_eq!(test.ditto.store().observers().keys.len(), 0);
}
