#![allow(clippy::bool_assert_comparison)]

use core::{fmt::Debug, hash::Hash};
use std::collections::HashMap;

use dittolive_ditto::{error::ErrorKind, prelude::*};
use ffi_sdk::FfiErrorCode;
use futures::FutureExt;
use serde_json::json;
use static_assertions::{assert_impl_all, assert_not_impl_any};

mod common;

// === Public API trait impls ===

assert_impl_all!(CreateTransactionOptions: Debug, Clone, PartialEq, Default, Send, Sync, Sized);
assert_not_impl_any!(CreateTransactionOptions: Eq, Ord, Hash); // semver hazards
                                                               //
assert_impl_all!(TransactionInfo: Debug, Clone, PartialEq, Send, Sync, Sized);
assert_not_impl_any!(TransactionInfo: Eq, Ord, Hash); // semver hazards

// TODO(cameron): `Copy`? `Eq`?
assert_impl_all!(TransactionCompletionAction: Debug, Clone, PartialEq, Send, Sync, Sized);
assert_not_impl_any!(TransactionCompletionAction: Eq, Ord, Hash); // semver hazards

// `Transaction` should implement basically nothing to prevent misuse
assert_impl_all!(Transaction: Send, Sync, Sized);
assert_not_impl_any!(Transaction: Clone, Default, PartialEq, Eq, Ord, Hash);

// === Utils ===

fn get_builder_ditto() -> Ditto {
    let ditto = common::get_inactive_builder_ditto(None).unwrap();
    ditto.disable_sync_with_v3().unwrap();
    ditto
}

fn get_revamp_ditto() -> Ditto {
    let ditto = common::get_inactive_revamp_ditto(None).unwrap();
    ditto.disable_sync_with_v3().unwrap();
    ditto
}

async fn len(ditto: &Ditto) -> usize {
    ditto
        .store()
        .execute_v2("SELECT * FROM tasks")
        .await
        .unwrap()
        .item_count()
}

async fn len_txn(txn: &Transaction) -> usize {
    txn.execute("SELECT * FROM tasks")
        .await
        .unwrap()
        .item_count()
}

// === Tests ===

#[tokio::test]
async fn potentially_modifies_database_if_return_commit() {
    let ditto = get_builder_ditto();
    potentially_modifies_database_if_return_commit_impl(ditto).await;

    let ditto = get_revamp_ditto();
    potentially_modifies_database_if_return_commit_impl(ditto).await;
}

async fn potentially_modifies_database_if_return_commit_impl(ditto: Ditto) {
    let action = ditto
        .store()
        .transaction(async |txn| {
            txn.execute((
                "INSERT INTO tasks DOCUMENTS (:doc)",
                json!({ "doc": { "foo": "bar"}}),
            ))
            .await?;

            assert_eq!(len_txn(txn).await, 1);

            Ok::<_, DittoError>(TransactionCompletionAction::Commit)
        })
        .await
        .unwrap();

    assert_eq!(len(&ditto).await, 1);
    assert_eq!(action, TransactionCompletionAction::Commit);
    ditto.close();
}

#[tokio::test]
async fn doesnt_modify_database_if_rolled_back() {
    let ditto = get_builder_ditto();
    doesnt_modify_database_if_rolled_back_impl(ditto).await;

    let ditto = get_revamp_ditto();
    doesnt_modify_database_if_rolled_back_impl(ditto).await;
}

async fn doesnt_modify_database_if_rolled_back_impl(ditto: Ditto) {
    let action = ditto
        .store()
        .transaction(async |txn| {
            txn.execute((
                "INSERT INTO tasks DOCUMENTS (:doc)",
                json!({ "doc": { "foo": "bar"}}),
            ))
            .await?;

            assert_eq!(len_txn(txn).await, 1);

            Ok::<_, DittoError>(TransactionCompletionAction::Rollback)
        })
        .await
        .unwrap();

    assert_eq!(len(&ditto).await, 0);
    assert_eq!(action, TransactionCompletionAction::Rollback);
}

#[tokio::test]
async fn returns_info_with_options_corresponding_to_the_options_passed_when_creating_the_transaction(
) {
    let ditto = get_builder_ditto();
    returns_info_with_options_corresponding_to_the_options_passed_when_creating_the_transaction_impl(ditto).await;

    let ditto = get_revamp_ditto();
    returns_info_with_options_corresponding_to_the_options_passed_when_creating_the_transaction_impl(ditto).await;
}

async fn returns_info_with_options_corresponding_to_the_options_passed_when_creating_the_transaction_impl(
    ditto: Ditto,
) {
    let mut options = CreateTransactionOptions::new();
    options.hint = Some("5 across");
    options.is_read_only = true;

    let _ = ditto
        .store()
        .transaction_with_options(options, async |txn| {
            assert_eq!(txn.info().is_read_only, true);
            assert_eq!(txn.info().hint.unwrap(), "5 across");
            let _ = txn.info().id;

            Ok::<_, DittoError>(TransactionCompletionAction::Commit)
        })
        .await
        .unwrap();

    let _ = ditto
        .store()
        .transaction(async |txn| {
            assert_eq!(txn.info().is_read_only, false);
            assert_eq!(txn.info().hint, None);
            let _ = txn.info().id;

            Ok::<_, DittoError>(TransactionCompletionAction::Commit)
        })
        .await
        .unwrap();
}

#[tokio::test]
async fn read_write_by_default() {
    let ditto = get_builder_ditto();
    read_write_by_default_impl(ditto).await;

    let ditto = get_revamp_ditto();
    read_write_by_default_impl(ditto).await;
}

async fn read_write_by_default_impl(ditto: Ditto) {
    let action = ditto
        .store()
        .transaction(async |txn| -> Result<_, DittoError> {
            assert_eq!(txn.info().is_read_only, false);

            txn.execute((
                "INSERT INTO tasks DOCUMENTS (:doc)",
                json!({ "doc": {"foo": "bar"}}),
            ))
            .await?;

            assert_eq!(len_txn(txn).await, 1);

            Ok(TransactionCompletionAction::Commit)
        })
        .await
        .unwrap();

    assert_eq!(action, TransactionCompletionAction::Commit);
    assert_eq!(len(&ditto).await, 1);
}

#[tokio::test]
async fn txn_object_is_made_unusable_after_transaction_ends() {
    // no-op because the lifetime of the `&Transaction` parameter guarantees that the transaction
    // object cannot escape the closure
    //
    // See the module-level docs for a doctest that guarantees this
}

#[tokio::test]
async fn performs_multiple_dql_query_executions_exclusively_and_atomically() {
    let ditto = get_builder_ditto();
    performs_multiple_dql_query_executions_exclusively_and_atomically_impl(ditto).await;

    let ditto = get_revamp_ditto();
    performs_multiple_dql_query_executions_exclusively_and_atomically_impl(ditto).await;
}

async fn performs_multiple_dql_query_executions_exclusively_and_atomically_impl(ditto: Ditto) {
    let (send, mut recv) = tokio::sync::mpsc::unbounded_channel();

    let fut1 = ditto
        .store()
        .transaction::<_, DittoError>(async |txn| {
            send.send(1).unwrap();

            txn.execute((
                "INSERT INTO tasks DOCUMENTS (:doc)",
                json!({"doc": {"foo": "bar"}}),
            ))
            .await?;

            send.send(2).unwrap();

            txn.execute((
                "INSERT INTO tasks DOCUMENTS (:doc)",
                json!({"doc": {"foo": "bar"}}),
            ))
            .await?;

            send.send(3).unwrap();

            Ok(TransactionCompletionAction::Commit)
        })
        .map(Result::unwrap);

    let fut2 = ditto
        .store()
        .transaction::<_, DittoError>(async |txn| {
            send.send(4).unwrap();

            txn.execute((
                "INSERT INTO tasks DOCUMENTS (:doc)",
                json!({"doc": {"foo": "bar"}}),
            ))
            .await?;

            send.send(5).unwrap();

            txn.execute((
                "INSERT INTO tasks DOCUMENTS (:doc)",
                json!({"doc": {"foo": "bar"}}),
            ))
            .await?;

            send.send(6).unwrap();

            Ok(TransactionCompletionAction::Commit)
        })
        .map(Result::unwrap);

    _ = futures::join!(fut1, fut2);

    drop(send);

    let mut events = vec![];
    recv.recv_many(&mut events, usize::MAX).await;

    match &events[..] {
        [1, 2, 3, 4, 5, 6] => {}
        [4, 5, 6, 1, 2, 3] => {}
        other => panic!("unexpected order: {other:?}"),
    }
}

#[tokio::test]
async fn performs_multiple_read_only_transactions_in_parallel() {
    let ditto = get_builder_ditto();
    performs_multiple_read_only_transactions_in_parallel_impl(ditto).await;

    let ditto = get_revamp_ditto();
    performs_multiple_read_only_transactions_in_parallel_impl(ditto).await;
}

async fn performs_multiple_read_only_transactions_in_parallel_impl(ditto: Ditto) {
    let mut options = CreateTransactionOptions::default();
    options.is_read_only = true;

    let (send, mut recv) = tokio::sync::mpsc::unbounded_channel();
    let (txn_1_made_progress, txn_1_progress) = tokio::sync::oneshot::channel();
    let (txn_2_made_progress, txn_2_progress) = tokio::sync::oneshot::channel();

    let fut1 = ditto
        .store()
        .transaction_with_options(options.clone(), async |txn: &Transaction| {
            send.send(1).unwrap();

            txn.execute("SELECT * FROM tasks").await?;
            send.send(2).unwrap();

            txn_1_made_progress.send(()).unwrap();
            txn_2_progress.await.unwrap();

            send.send(5).unwrap();

            txn.execute("SELECT * FROM tasks").await?;
            send.send(6).unwrap();

            Ok::<_, DittoError>(TransactionCompletionAction::Commit)
        })
        .map(Result::unwrap);

    let fut2 = ditto
        .store()
        .transaction_with_options(options, async |txn: &Transaction| {
            txn_1_progress.await.unwrap();

            send.send(3).unwrap();

            txn.execute("SELECT * FROM tasks").await?;
            send.send(4).unwrap();

            txn_2_made_progress.send(()).unwrap();

            Ok::<_, DittoError>(TransactionCompletionAction::Commit)
        })
        .map(Result::unwrap);

    _ = futures::join!(fut1, fut2);

    let mut events = vec![];
    recv.recv_many(&mut events, usize::MAX).await;

    assert_eq!(events, [1, 2, 3, 4, 5, 6]);
}

#[tokio::test]
async fn allows_some_execute_calls_to_fail_without_affecting_the_whole_transaction_if_error_is_caught(
) {
    let ditto = get_builder_ditto();
    allows_some_execute_calls_to_fail_without_affecting_the_whole_transaction_if_error_is_caught_impl(ditto).await;

    let ditto = get_revamp_ditto();
    allows_some_execute_calls_to_fail_without_affecting_the_whole_transaction_if_error_is_caught_impl(ditto).await;
}

async fn allows_some_execute_calls_to_fail_without_affecting_the_whole_transaction_if_error_is_caught_impl(
    ditto: Ditto,
) {
    let action = ditto
        .store()
        .transaction::<_, DittoError>(async |txn: &Transaction| {
            if txn.execute("not DQL").await.is_ok() {
                panic!("expected error");
            }

            Ok(TransactionCompletionAction::Commit)
        })
        .await
        .unwrap();

    assert_eq!(action, TransactionCompletionAction::Commit);
}

#[tokio::test]
async fn commits_if_returned_action_is_commit() {
    let ditto = get_builder_ditto();
    commits_if_returned_action_is_commit_impl(ditto).await;

    let ditto = get_revamp_ditto();
    commits_if_returned_action_is_commit_impl(ditto).await;
}

async fn commits_if_returned_action_is_commit_impl(ditto: Ditto) {
    let action = ditto
        .store()
        .transaction::<_, DittoError>(async |txn: &Transaction| {
            txn.execute((
                "INSERT INTO tasks DOCUMENTS (:doc)",
                json!({"doc": {"foo": "bar"}}),
            ))
            .await?;

            Ok(TransactionCompletionAction::Commit)
        })
        .await
        .unwrap();

    assert_eq!(action, TransactionCompletionAction::Commit);
    assert_eq!(len(&ditto).await, 1);
}

#[tokio::test]
async fn rolls_back_if_the_returned_action_is_rollback() {
    let ditto = get_builder_ditto();
    rolls_back_if_the_returned_action_is_rollback_impl(ditto).await;

    let ditto = get_revamp_ditto();
    rolls_back_if_the_returned_action_is_rollback_impl(ditto).await;
}

async fn rolls_back_if_the_returned_action_is_rollback_impl(ditto: Ditto) {
    let action = ditto
        .store()
        .transaction::<_, DittoError>(async |txn: &Transaction| {
            txn.execute((
                "INSERT INTO tasks DOCUMENTS (:doc)",
                json!({"doc": {"foo": "bar"}}),
            ))
            .await?;

            Ok(TransactionCompletionAction::Rollback)
        })
        .await
        .unwrap();

    assert_eq!(action, TransactionCompletionAction::Rollback);
    assert_eq!(len(&ditto).await, 0);
}

#[tokio::test]
async fn implicitly_rolls_back_if_scope_throws_an_error() {
    let ditto = get_builder_ditto();
    implicitly_rolls_back_if_scope_throws_an_error_impl(ditto).await;

    let ditto = get_revamp_ditto();
    implicitly_rolls_back_if_scope_throws_an_error_impl(ditto).await;
}

async fn implicitly_rolls_back_if_scope_throws_an_error_impl(ditto: Ditto) {
    let err = ditto
        .store()
        .transaction::<_, DittoError>(async |txn: &Transaction| {
            txn.execute((
                "INSERT INTO tasks DOCUMENTS (:doc)",
                json!({"doc": {"foo": "bar"}}),
            ))
            .await?;

            txn.execute("bleep bloop boop").await?;

            Ok(TransactionCompletionAction::Commit)
        })
        .await
        .unwrap_err();

    assert!(matches!(
        err.kind(),
        ErrorKind::CoreApi(FfiErrorCode::DqlQueryCompilation)
    ));
    assert_eq!(len(&ditto).await, 0);
}

#[tokio::test]
async fn throws_if_read_only_while_mutating_query_is_executed() {
    let ditto = get_builder_ditto();
    throws_if_read_only_while_mutating_query_is_executed_impl(ditto).await;

    let ditto = get_revamp_ditto();
    throws_if_read_only_while_mutating_query_is_executed_impl(ditto).await;
}

async fn throws_if_read_only_while_mutating_query_is_executed_impl(ditto: Ditto) {
    let mut options = CreateTransactionOptions::new();
    options.is_read_only = true;

    let err = ditto
        .store()
        .transaction_with_options::<_, DittoError>(options, async |txn: &Transaction| {
            txn.execute((
                "INSERT INTO tasks DOCUMENTS (:doc)",
                json!({"doc": {"foo": "bar"}}),
            ))
            .await?;

            Ok(TransactionCompletionAction::Commit)
        })
        .await
        .unwrap_err();

    assert!(matches!(
        err.kind(),
        ErrorKind::CoreApi(FfiErrorCode::StoreTransactionReadOnly)
    ));
    assert_eq!(len(&ditto).await, 0);
}

#[tokio::test]
async fn throws_if_read_only_while_mutating_query_is_executed_even_if_without_effect() {
    let ditto = get_builder_ditto();
    throws_if_read_only_while_mutating_query_is_executed_even_if_without_effect_impl(ditto).await;

    let ditto = get_revamp_ditto();
    throws_if_read_only_while_mutating_query_is_executed_even_if_without_effect_impl(ditto).await;
}

async fn throws_if_read_only_while_mutating_query_is_executed_even_if_without_effect_impl(
    ditto: Ditto,
) {
    let mut options = CreateTransactionOptions::new();
    options.is_read_only = true;

    let err = ditto
        .store()
        .transaction_with_options::<_, DittoError>(options, async |txn: &Transaction| {
            txn.execute(
                "UPDATE tasks SET color = 'black' WHERE completed = true AND completed = false",
            )
            .await?;

            Ok(TransactionCompletionAction::Commit)
        })
        .await
        .unwrap_err();

    assert!(matches!(
        err.kind(),
        ErrorKind::CoreApi(FfiErrorCode::StoreTransactionReadOnly)
    ));
    assert_eq!(len(&ditto).await, 0);
}

/// This test has some extra logic to make sure type inference and `?` work as expected
#[tokio::test]
async fn propagates_client_side_error_thrown_within_scope() {
    #[derive(Debug, thiserror::Error)]
    enum CustomerError {
        #[error("Foo")]
        Foo,
        #[error("{0}")]
        Ditto(#[from] DittoError),
    }

    fn customer_err_fn() -> Result<(), CustomerError> {
        Ok(())
    }

    let ditto = get_builder_ditto();

    let _ = ditto
        .store()
        .transaction(async |txn: &Transaction| {
            let qr = txn.execute("SELECT * FROM tasks").await?;

            if qr.item_count() > 0 {
                return Err(CustomerError::Foo);
            }

            customer_err_fn()?;

            Ok(TransactionCompletionAction::Commit)
        })
        .await;

    let _ = ditto
        .store()
        .transaction(async |txn| {
            assert_eq!(txn.info().is_read_only, false);
            assert_eq!(txn.info().hint, None);
            let _ = txn.info().id;

            Ok::<_, DittoError>(TransactionCompletionAction::Commit)
        })
        .await
        .unwrap();
}

#[tokio::test]
async fn allows_creating_and_inserting_attachment_inside_the_transaction_block() {
    let ditto = get_builder_ditto();
    allows_creating_and_inserting_attachment_inside_the_transaction_block_impl(ditto).await;

    let ditto = get_revamp_ditto();
    allows_creating_and_inserting_attachment_inside_the_transaction_block_impl(ditto).await;
}

async fn allows_creating_and_inserting_attachment_inside_the_transaction_block_impl(ditto: Ditto) {
    let _ = ditto
        .store()
        .transaction::<_, DittoError>(async |txn: &Transaction| {
            let attachment = ditto
                .store()
                .new_attachment_from_bytes(&[1, 2, 3, 4], HashMap::default())
                .await?;

            txn.execute((
                "INSERT INTO COLLECTION tasks (attachment ATTACHMENT) DOCUMENTS (:doc)",
                json!({"doc": {"attachment": attachment}}),
            ))
            .await?;

            Ok(TransactionCompletionAction::Commit)
        })
        .await
        .unwrap();

    let result = ditto
        .store()
        .execute_v2("SELECT * FROM COLLECTION tasks (attachment ATTACHMENT)")
        .await
        .unwrap();

    let attachment = result.iter().next().unwrap().value();
    let attachment_token = attachment.get("attachment").unwrap().as_object().unwrap();

    let fetcher = ditto
        .store()
        .fetch_attachment(attachment_token, |event| match event {
            DittoAttachmentFetchEvent::Deleted => panic!("wut"),
            DittoAttachmentFetchEvent::Progress { .. } => {}
            DittoAttachmentFetchEvent::Completed { attachment } => {
                assert_eq!(std::fs::read(attachment.path()).unwrap(), [1, 2, 3, 4]);
            }
        })
        .unwrap();

    fetcher.cancel();
}

#[tokio::test]
async fn allows_creating_a_new_read_only_transaction_inside_read_write_transaction_block() {
    let ditto = get_builder_ditto();
    allows_creating_a_new_read_only_transaction_inside_read_write_transaction_block_impl(ditto)
        .await;

    let ditto = get_revamp_ditto();
    allows_creating_a_new_read_only_transaction_inside_read_write_transaction_block_impl(ditto)
        .await;
}

async fn allows_creating_a_new_read_only_transaction_inside_read_write_transaction_block_impl(
    ditto: Ditto,
) {
    let mut options = CreateTransactionOptions::new();
    options.is_read_only = true;

    let _ = ditto
        .store()
        .transaction(async |rw_txn: &Transaction| {
            assert_eq!(len_txn(rw_txn).await, 0);
            let _ = ditto
                .store()
                .transaction_with_options(options, async |ro_txn: &Transaction| {
                    assert_eq!(len_txn(ro_txn).await, 0);

                    Ok::<_, DittoError>(TransactionCompletionAction::Commit)
                })
                .await
                .unwrap();
            Ok::<_, DittoError>(TransactionCompletionAction::Commit)
        })
        .await
        .unwrap();
}

#[tokio::test]
async fn allows_creating_a_new_read_write_transaction_inside_read_only_transaction_block() {
    let ditto = get_builder_ditto();
    allows_creating_a_new_read_write_transaction_inside_read_only_transaction_block_impl(ditto)
        .await;

    let ditto = get_revamp_ditto();
    allows_creating_a_new_read_write_transaction_inside_read_only_transaction_block_impl(ditto)
        .await;
}

async fn allows_creating_a_new_read_write_transaction_inside_read_only_transaction_block_impl(
    ditto: Ditto,
) {
    let mut options = CreateTransactionOptions::new();
    options.is_read_only = true;

    let _ = ditto
        .store()
        .transaction_with_options(options, async |ro_txn: &Transaction| {
            assert_eq!(len_txn(ro_txn).await, 0);
            let _ = ditto
                .store()
                .transaction(async |rw_txn: &Transaction| {
                    assert_eq!(len_txn(rw_txn).await, 0);

                    Ok::<_, DittoError>(TransactionCompletionAction::Commit)
                })
                .await
                .unwrap();
            Ok::<_, DittoError>(TransactionCompletionAction::Commit)
        })
        .await
        .unwrap();
}

#[tokio::test]
async fn allows_returning_arbitrary_values() {
    let ditto = get_builder_ditto();
    allows_returning_arbitrary_values_impl(ditto).await;

    let ditto = get_revamp_ditto();
    allows_returning_arbitrary_values_impl(ditto).await;
}

async fn allows_returning_arbitrary_values_impl(ditto: Ditto) {
    let hello = ditto
        .store()
        .transaction::<_, DittoError>(async |_| Ok(String::from("hello")))
        .await
        .unwrap();

    assert_eq!(hello, "hello");

    let world = ditto
        .store()
        .transaction_with_options::<_, DittoError>(Default::default(), async |_| {
            Ok(String::from("world"))
        })
        .await
        .unwrap();

    assert_eq!(world, "world");
}
