// TODO(v5): remove entire test module
#![allow(deprecated)]

//! Tests of live query functionality
#![allow(clippy::disallowed_methods)]

use std::sync::{
    atomic::{AtomicBool, Ordering},
    Arc, Mutex,
};

use anyhow::Error;
use assert_matches::assert_matches as assert_matches_;
use culpa::throws;
use serde::{Deserialize, Serialize};
use serde_json::json;

mod common;
use common::*;

#[allow(dead_code)]
struct LiveQueryTest {
    ditto: Ditto,
    collection_name: String,
    collection: Collection,
}

#[throws]
fn setup() -> LiveQueryTest {
    let ditto = common::get_builder_ditto(None)?;
    let collection_name = uuid::Uuid::new_v4().to_string();
    let collection = ditto.store().collection(&collection_name)?;

    LiveQueryTest {
        ditto,
        collection_name,
        collection,
    }
}

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

    let doc1_id = DocumentId::new(&"1")?;
    let doc1 = json!({
        "_id": doc1_id,
        "make": "Honda",
        "color": "red",
    });
    let expected_doc = doc1.clone();
    let doc1_id_ret = test.collection.upsert(doc1)?;
    assert_eq!(doc1_id_ret, doc1_id);

    let doc2 = json!({
        "make": "Honda",
        "color": "crimson"
    });
    let _doc2_id = test.collection.upsert(doc2)?;

    let (state_tx, mut state_rx) = tokio::sync::watch::channel((0, false)); // (count, finished)
    let live_query = test.collection.find("color == 'red'").observe_local(
        move |docs: Vec<BoxedDocument>, event| {
            state_tx.send_modify(|(count, _finished)| {
                *count += 1;
            });

            let LiveQueryEvent::Initial = event else {
                panic!("Unexpected event received");
            };

            let (count, finished) = *state_tx.borrow();
            assert_eq!(count, 1);
            assert!(!finished);

            assert_eq!(docs.len(), 1);
            assert_eq!(doc1_id, docs[0].id());

            let first_doc = docs[0].typed::<serde_json::Value>().unwrap();
            assert_eq!(first_doc, expected_doc);

            state_tx.send_modify(|(_count, finished)| *finished = true);
        },
    )?;

    state_rx.wait_for(|(_count, finished)| *finished).await?;
    drop(live_query);
}

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

    let (state_tx, mut state_rx) = tokio::sync::watch::channel((0, false)); // (count, finished)
    let honda_id = DocumentId::new(&"honda")?;
    let collection = test.collection.clone();

    let live_query = test.collection.find("color == 'Red'").observe_local(
        move |docs: Vec<BoxedDocument>, event| {
            state_tx.send_modify(|(count, _finished)| {
                *count += 1;
            });

            match event {
                // Step 1: On initial call, insert a new document into the collection
                LiveQueryEvent::Initial => {
                    let (counter, _finished) = *state_tx.borrow();
                    assert_eq!(counter, 1);

                    assert!(docs.is_empty());
                    let doc1 = json!({
                        "_id": honda_id,
                        "make": "Honda",
                        "color": "Red",
                    });

                    // The Collection here has a reference to the underlying core ditto Instance
                    // which may be freed by the time this insert is called unless
                    // something holds the test open
                    let _id = collection.upsert(doc1).unwrap();
                }
                // Step 2: After inserting document in initial callback, assert, we receive it as an
                // update
                LiveQueryEvent::Update {
                    old_documents,
                    insertions,
                    ..
                } => {
                    let (count, _finished) = *state_tx.borrow();
                    assert_eq!(count, 2);

                    let expected_insertions = [0_usize; 1];
                    let expected_doc = json!({
                        "_id": honda_id,
                        "make": "Honda",
                        "color": "Red",
                    });
                    let first_doc = docs[0].typed::<serde_json::Value>().unwrap();
                    assert_eq!(docs.len(), 1);
                    assert!(old_documents.is_empty());
                    assert_eq!(insertions, expected_insertions.into());
                    assert_eq!(first_doc, expected_doc);

                    state_tx.send_modify(|(_count, finished)| *finished = true);
                }
            }
        },
    )?;

    state_rx.wait_for(|(_count, finished)| *finished).await?;
    drop(live_query);
}

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

    let doc = json!({
        "make": "Honda",
        "color": "red",
    });
    let doc_id = test.collection.upsert(doc)?;

    let (state_tx, mut state_rx) = tokio::sync::watch::channel((0, false)); // (count, finished)
    let collection = test.collection.clone();
    let live_query = test
        .collection
        .find("color == 'red'")
        .observe_local(move |docs, event| {
            state_tx.send_modify(|(count, _finished)| *count += 1);

            match event {
                LiveQueryEvent::Initial => {
                    let (count, _finished) = *state_tx.borrow();
                    assert_eq!(count, 1);
                    assert_eq_ignoring_id(
                        &**docs.first().unwrap(),
                        json!({ "make": "Honda", "color": "red" }),
                    )
                    .unwrap();

                    collection
                        .find_by_id(doc_id.clone())
                        .update(|maybe_doc| {
                            let doc = maybe_doc.unwrap();
                            doc.set("make", "Ferrari").unwrap();
                        })
                        .unwrap();
                }
                LiveQueryEvent::Update {
                    old_documents,
                    updates,
                    ..
                } => {
                    let (count, _finished) = *state_tx.borrow();
                    assert_eq!(count, 2);
                    assert_eq_ignoring_id(
                        &**docs.first().unwrap(),
                        json!({ "make": "Ferrari", "color": "red" }),
                    )
                    .unwrap();

                    assert_eq!(docs.len(), 1);
                    assert_eq!(old_documents.len(), 1);
                    assert_eq!(*updates, [0]);

                    state_tx.send_modify(|(_count, finished)| *finished = true);
                }
            }
        });

    state_rx.wait_for(|(_count, finished)| *finished).await?;
    drop(live_query);
}

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

    let doc_id = test.collection.upsert(json!({
        "make": "Honda",
        "color": "red",
    }))?;

    let (state_tx, mut state_rx) = tokio::sync::watch::channel((0, false)); // (count, finished)
    let collection = test.collection.clone();
    let live_query = test
        .collection
        .find("color == 'red'")
        .observe_local(move |docs, event| {
            state_tx.send_modify(|(count, _finished)| *count += 1);

            match event {
                LiveQueryEvent::Initial => {
                    let (count, _finished) = *state_tx.borrow();
                    assert_eq!(count, 1);
                    assert_eq_ignoring_id(
                        &**docs.first().unwrap(),
                        json!({ "make": "Honda", "color": "red" }),
                    )
                    .unwrap();

                    collection.find_by_id(&doc_id).remove().unwrap();
                }
                LiveQueryEvent::Update {
                    old_documents,
                    deletions,
                    ..
                } => {
                    let (count, _finished) = *state_tx.borrow();
                    assert_eq!(count, 2);

                    assert_eq!(old_documents.len(), 1);
                    assert_eq!(*deletions, [0]);
                    let result = collection
                        .find_by_id(old_documents.first().unwrap().id())
                        .exec();
                    assert!(result.is_err());

                    state_tx.send_modify(|(_count, finished)| *finished = true);
                }
            }
        })?;

    state_rx.wait_for(|(_count, finished)| *finished).await?;
    drop(live_query);
}

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

    let doc_id = test
        .collection
        .upsert(json!({ "make": "Honda", "color": "red" }))?;

    let (state_tx, mut state_rx) = tokio::sync::watch::channel((0, false)); // (count, finished)
    let collection = test.collection.clone();
    let live_query = test
        .collection
        .find("color == 'red'")
        .observe_local(move |docs, event| {
            state_tx.send_modify(|(count, _finished)| *count += 1);

            match event {
                LiveQueryEvent::Initial => {
                    let (count, _finished) = *state_tx.borrow();
                    assert_eq!(count, 1);
                    assert_eq!(docs.len(), 1);
                    assert_eq_ignoring_id(
                        &**docs.first().unwrap(),
                        json!({ "make": "Honda", "color": "red" }),
                    )
                    .unwrap();

                    collection.find_by_id(&doc_id).evict().unwrap();
                }
                LiveQueryEvent::Update {
                    old_documents,
                    deletions,
                    ..
                } => {
                    let (count, _finished) = *state_tx.borrow();
                    assert_eq!(count, 2);
                    assert_eq!(docs.len(), 0);

                    assert_eq!(old_documents.len(), 1);
                    assert_eq!(*deletions, [0]);
                    let result = collection
                        .find_by_id(old_documents.first().unwrap().id())
                        .exec();
                    assert!(result.is_err());

                    state_tx.send_modify(|(_count, finished)| *finished = true);
                }
            }
        })?;

    state_rx.wait_for(|(_count, finished)| *finished).await?;
    drop(live_query);
}

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

    let doc_id_1 = test
        .collection
        .upsert(json!({ "make": "Honda", "color": "red" }))?;
    let doc_id_2 = test
        .collection
        .upsert(json!({ "make": "Ferrari", "color": "red" }))?;

    let (state_tx, mut state_rx) = tokio::sync::watch::channel((0, false)); // (count, finished)
    let collection = test.collection.clone();
    let live_query = test
        .collection
        .find("color == 'red'")
        .observe_local(move |docs, event| {
            let doc_id_1 = doc_id_1.clone();
            let doc_id_2 = doc_id_2.clone();
            state_tx.send_modify(|(count, _finished)| *count += 1);

            match event {
                LiveQueryEvent::Initial => {
                    let (count, _finished) = *state_tx.borrow();
                    assert_eq!(count, 1);
                    assert_eq!(docs.len(), 2);

                    assert!(docs.iter().any(|doc| eq_ignoring_id(
                        &**doc,
                        json!({ "make": "Honda", "color": "red" })
                    )
                    .unwrap()));
                    assert!(docs.iter().any(|doc| eq_ignoring_id(
                        &**doc,
                        json!({ "make": "Ferrari", "color": "red" })
                    )
                    .unwrap()));

                    test.ditto
                        .store()
                        .with_batched_write(|mut batch| {
                            let mut collection = batch.collection(&test.collection_name);
                            collection
                                .upsert(json!({ "make": "Ford", "color": "red" }))
                                .unwrap();
                            collection.find_by_id(doc_id_1).evict().unwrap();
                            collection
                                .find_by_id(doc_id_2)
                                .update(|maybe_doc| {
                                    let doc = maybe_doc.unwrap();
                                    doc.set("make", "Aston Martin").unwrap();
                                })
                                .unwrap();

                            batch.commit_changes()
                        })
                        .unwrap();
                }
                LiveQueryEvent::Update {
                    old_documents,
                    insertions,
                    deletions,
                    updates,
                    ..
                } => {
                    let (count, _finished) = *state_tx.borrow();
                    assert_eq!(count, 2);
                    assert_eq!(docs.len(), 2);

                    assert_eq!(old_documents.len(), 2);
                    assert_eq!(insertions.len(), 1);
                    assert_eq!(deletions.len(), 1);
                    assert_eq!(updates.len(), 1);
                    let result = collection.find_by_id(&doc_id_1).exec();
                    assert!(result.is_err());

                    assert!(docs.iter().any(|doc| eq_ignoring_id(
                        &**doc,
                        json!({ "make": "Ford", "color": "red" })
                    )
                    .unwrap()));
                    assert!(docs.iter().any(|doc| eq_ignoring_id(
                        &**doc,
                        json!({ "make": "Aston Martin", "color": "red" })
                    )
                    .unwrap()));

                    state_tx.send_modify(|(_count, finished)| *finished = true);
                }
            }
        })?;

    state_rx.wait_for(|(_count, finished)| *finished).await?;
    drop(live_query);
}

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

    let collection_1_name = test.collection_name.clone();
    let collection_1 = test.collection.clone();
    let collection_2_name = uuid::Uuid::new_v4().to_string();
    let collection_2 = test.ditto.store().collection(&collection_2_name)?;

    let c1_doc1_id = collection_1.upsert(json!({ "make": "Honda", "color": "red"}))?;
    let _c1_doc2_id = collection_1.upsert(json!({ "make": "Honda", "color": "crimson" }))?;
    let c2_doc1_id = collection_2.upsert(json!({ "make": "Bentley", "color": "red" }))?;
    let c2_doc2_id = collection_2.upsert(json!({ "make": "Jaguar", "color": "red" }))?;

    // Inline helper to assert test cases
    let assert_payload = |docs: &[BoxedDocument],
                          event: LiveQueryEvent,
                          doc_count: usize,
                          insertions_count: usize,
                          deletions_count: usize,
                          updates_count: usize,
                          old_docs_count: usize| {
        let LiveQueryEvent::Update {
            old_documents,
            insertions,
            deletions,
            updates,
            ..
        } = event
        else {
            panic!("Unexpected event type");
        };
        assert_eq!(docs.len(), doc_count);
        assert_eq!(old_documents.len(), old_docs_count);
        assert_eq!(insertions.len(), insertions_count);
        assert_eq!(deletions.len(), deletions_count);
        assert_eq!(updates.len(), updates_count);
    };

    let mut doc2_id = None;
    let mut doc3_id = None;
    let mut doc4_id = None;
    let mut doc5_id = None;
    let mut doc6_id = None;
    let mut doc7_id = None;

    let (state_tx, mut state_rx) = tokio::sync::watch::channel((0, false)); // (count, finished)
    let live_query = test
        .collection
        .find("color == 'red'")
        .observe_local(move |docs, event| {
            state_tx.send_modify(|(count, _finished)| *count += 1);

            match event {
                LiveQueryEvent::Initial => {
                    let (count, _finished) = *state_tx.borrow();
                    assert_eq!(count, 1);
                    assert_eq_ignoring_id(
                        &**docs.first().unwrap(),
                        json!({ "make": "Honda", "color": "red" }),
                    )
                    .unwrap();

                    doc2_id = Some(
                        collection_1
                            .upsert(json!({ "make": "Toyota", "color": "red" }))
                            .unwrap(),
                    );
                }
                LiveQueryEvent::Update { .. } => {
                    let (count, _finished) = *state_tx.borrow();

                    match count {
                        2 => {
                            assert_payload(&docs, event, 2, 1, 0, 0, 1);
                            assert!(docs.iter().any(|doc| eq_ignoring_id(
                                &**doc,
                                json!({ "make": "Toyota", "color": "red" })
                            )
                            .unwrap()));

                            doc3_id = Some(
                                collection_1
                                    .upsert(json!({ "make": "Ford", "color": "red"}))
                                    .unwrap(),
                            );
                        }
                        3 => {
                            assert_payload(&docs, event, 3, 1, 0, 0, 2);
                            assert!(docs.iter().any(|doc| eq_ignoring_id(
                                &**doc,
                                json!({ "make": "Ford", "color": "red" })
                            )
                            .unwrap()));

                            collection_1
                                .find_by_id(doc2_id.as_ref().unwrap())
                                .update(|maybe_doc| {
                                    let doc = maybe_doc.unwrap();
                                    doc.set("make", "Maserati").unwrap();
                                })
                                .unwrap();
                        }
                        4 => {
                            assert_payload(&docs, event, 3, 0, 0, 1, 3);
                            assert!(docs.iter().any(|doc| eq_ignoring_id(
                                &**doc,
                                json!({ "make": "Maserati", "color": "red" })
                            )
                            .unwrap()));

                            doc4_id = Some(
                                collection_1
                                    .upsert(json!({ "make": "Mazda", "color": "orange" }))
                                    .unwrap(),
                            );

                            doc5_id = Some(
                                collection_1
                                    .upsert(json!({ "make": "Aston Martin", "color": "yellow" }))
                                    .unwrap(),
                            );

                            doc6_id = Some(
                                collection_1
                                    .upsert(json!({ "make": "Chevrolet", "color": "red" }))
                                    .unwrap(),
                            );
                        }
                        5 => {
                            assert_payload(&docs, event, 4, 1, 0, 0, 3);
                            assert!(docs.iter().any(|doc| eq_ignoring_id(
                                &**doc,
                                json!({ "make": "Chevrolet", "color": "red" })
                            )
                            .unwrap()));

                            collection_1
                                .find_by_id(&c1_doc1_id)
                                .update(|maybe_doc| {
                                    let doc = maybe_doc.unwrap();
                                    doc.set("color", "blue").unwrap();
                                })
                                .unwrap();
                        }
                        6 => {
                            assert_payload(&docs, event, 3, 0, 1, 0, 4);
                            assert!(docs.iter().all(|doc| !eq_ignoring_id(
                                &**doc,
                                json!({ "make": "Honda", "color": "red" })
                            )
                            .unwrap()));

                            let results = test
                                .ditto
                                .store()
                                .with_batched_write(|mut tx| {
                                    let mut collection_1 = tx.collection(&test.collection_name);
                                    collection_1
                                        .upsert(json!({ "make": "Hyundai", "color": "red" }))
                                        .unwrap();
                                    collection_1
                                        .upsert(json!({ "make": "Jeep", "color": "pink" }))
                                        .unwrap();
                                    collection_1
                                        .find_by_id(doc2_id.clone().unwrap())
                                        .evict()
                                        .unwrap();
                                    collection_1
                                        .find_by_id(doc4_id.clone().unwrap())
                                        .evict()
                                        .unwrap();
                                    collection_1
                                        .find_by_id(doc3_id.clone().unwrap())
                                        .update(|maybe_doc| {
                                            let doc = maybe_doc.unwrap();
                                            doc.set("make", "Audi").unwrap();
                                        })
                                        .unwrap();
                                    collection_1
                                        .find_by_id(doc6_id.clone().unwrap())
                                        .update(|maybe_doc| {
                                            let doc = maybe_doc.unwrap();
                                            doc.set("color", "white").unwrap();
                                        })
                                        .unwrap();
                                    collection_1
                                        .find_by_id(doc5_id.clone().unwrap())
                                        .update(|maybe_doc| {
                                            let doc = maybe_doc.unwrap();
                                            doc.set("color", "red").unwrap();
                                        })
                                        .unwrap();

                                    let mut collection_2 = tx.collection(&collection_2_name);
                                    collection_2
                                        .upsert(json!({ "make": "BMW", "color": "black" }))
                                        .unwrap();
                                    collection_2.find_by_id(c2_doc1_id.clone()).evict().unwrap();
                                    collection_2
                                        .find_by_id(c2_doc2_id.clone())
                                        .update(|maybe_doc| {
                                            let doc = maybe_doc.unwrap();
                                            doc.set("make", "Citroen").unwrap();
                                        })
                                        .unwrap();
                                    tx.commit_changes()
                                })
                                .unwrap();

                            use dittolive_ditto::store::batch::DocChangeKind;
                            for (i, result) in results.iter().enumerate() {
                                match i {
                                    0 => {
                                        let DocChangeKind::Inserted = result.kind else {
                                            panic!("Incorrect result");
                                        };
                                        assert_eq!(
                                            result.collection_name.to_string(),
                                            collection_1_name
                                        );
                                        let doc = collection_1
                                            .find_by_id(result.doc_id.clone())
                                            .exec()
                                            .unwrap();
                                        assert_eq_ignoring_id(
                                            &*doc,
                                            json!({ "make": "Hyundai", "color": "red" }),
                                        )
                                        .unwrap();
                                    }
                                    1 => {
                                        let DocChangeKind::Inserted = result.kind else {
                                            panic!("Incorrect result");
                                        };
                                        assert_eq!(
                                            result.collection_name.to_string(),
                                            collection_1_name
                                        );
                                        let doc = collection_1
                                            .find_by_id(result.doc_id.clone())
                                            .exec()
                                            .unwrap();
                                        assert_eq_ignoring_id(
                                            &*doc,
                                            json!({ "make": "Jeep", "color": "pink" }),
                                        )
                                        .unwrap();
                                    }
                                    2 => {
                                        let DocChangeKind::Evicted = result.kind else {
                                            panic!("Incorrect result");
                                        };
                                        assert_eq!(
                                            result.collection_name.to_string(),
                                            collection_1_name
                                        );
                                        let result =
                                            collection_1.find_by_id(result.doc_id.clone()).exec();
                                        assert!(result.is_err());
                                    }
                                    3 => {
                                        let DocChangeKind::Evicted = result.kind else {
                                            panic!("Incorrect result");
                                        };
                                        assert_eq!(
                                            result.collection_name.to_string(),
                                            collection_1_name
                                        );
                                        let result =
                                            collection_1.find_by_id(result.doc_id.clone()).exec();
                                        assert!(result.is_err());
                                    }
                                    4 => {
                                        let DocChangeKind::Updated = result.kind else {
                                            panic!("Incorrect result");
                                        };
                                        assert_eq!(
                                            result.collection_name.to_string(),
                                            collection_1_name
                                        );
                                        let doc = collection_1
                                            .find_by_id(result.doc_id.clone())
                                            .exec()
                                            .unwrap();
                                        assert_eq_ignoring_id(
                                            &*doc,
                                            json!({ "make": "Audi", "color": "red" }),
                                        )
                                        .unwrap();
                                    }
                                    5 => {
                                        let DocChangeKind::Updated = result.kind else {
                                            panic!("Incorrect result");
                                        };
                                        assert_eq!(
                                            result.collection_name.to_string(),
                                            collection_1_name
                                        );
                                        let doc = collection_1
                                            .find_by_id(result.doc_id.clone())
                                            .exec()
                                            .unwrap();
                                        assert_eq_ignoring_id(
                                            &*doc,
                                            json!({ "make": "Chevrolet", "color": "white" }),
                                        )
                                        .unwrap();
                                    }
                                    6 => {
                                        let DocChangeKind::Updated = result.kind else {
                                            panic!("Incorrect result");
                                        };
                                        assert_eq!(
                                            result.collection_name.to_string(),
                                            collection_1_name
                                        );
                                        let doc = collection_1
                                            .find_by_id(result.doc_id.clone())
                                            .exec()
                                            .unwrap();
                                        assert_eq_ignoring_id(
                                            &*doc,
                                            json!({ "make": "Aston Martin", "color": "red" }),
                                        )
                                        .unwrap();
                                    }
                                    7 => {
                                        let DocChangeKind::Inserted = result.kind else {
                                            panic!("Incorrect result");
                                        };
                                        assert_eq!(
                                            result.collection_name.to_string(),
                                            collection_2_name
                                        );
                                        let doc = collection_2
                                            .find_by_id(result.doc_id.clone())
                                            .exec()
                                            .unwrap();
                                        assert_eq_ignoring_id(
                                            &*doc,
                                            json!({ "make": "BMW", "color": "black" }),
                                        )
                                        .unwrap();
                                    }
                                    8 => {
                                        let DocChangeKind::Evicted = result.kind else {
                                            panic!("Incorrect result");
                                        };
                                        assert_eq!(
                                            result.collection_name.to_string(),
                                            collection_2_name
                                        );
                                        let result =
                                            collection_2.find_by_id(result.doc_id.clone()).exec();
                                        assert!(result.is_err());
                                    }
                                    9 => {
                                        let DocChangeKind::Updated = result.kind else {
                                            panic!("Incorrect result");
                                        };
                                        assert_eq!(
                                            result.collection_name.to_string(),
                                            collection_2_name
                                        );
                                        let doc = collection_2
                                            .find_by_id(result.doc_id.clone())
                                            .exec()
                                            .unwrap();
                                        assert_eq_ignoring_id(
                                            &*doc,
                                            json!({ "make": "Citroen", "color": "red" }),
                                        )
                                        .unwrap();
                                    }
                                    _ => {
                                        panic!("Too many results!");
                                    }
                                }
                            }
                        }
                        7 => {
                            assert_payload(&docs, event, 3, 2, 2, 1, 3);
                            assert!(docs.iter().any(|doc| eq_ignoring_id(
                                &**doc,
                                json!({ "make": "Audi", "color": "red" })
                            )
                            .unwrap()));
                            assert!(docs.iter().any(|doc| eq_ignoring_id(
                                &**doc,
                                json!({ "make": "Aston Martin", "color": "red" })
                            )
                            .unwrap()));
                            assert!(docs.iter().any(|doc| eq_ignoring_id(
                                &**doc,
                                json!({ "make": "Hyundai", "color": "red" })
                            )
                            .unwrap()));

                            doc7_id = Some(
                                collection_1
                                    .upsert(json!({ "make": "Ferrari", "color": "red"}))
                                    .unwrap(),
                            );
                        }
                        8 => {
                            assert_payload(&docs, event, 4, 1, 0, 0, 3);
                            assert!(docs.iter().any(|doc| eq_ignoring_id(
                                &**doc,
                                json!({ "make": "Ferrari", "color": "red" })
                            )
                            .unwrap()));

                            let results2 = test
                                .ditto
                                .store()
                                .with_batched_write(|mut tx| {
                                    let mut collection = tx.collection(&collection_1_name);
                                    collection
                                        .find_by_id(doc7_id.clone().unwrap())
                                        .update(|maybe_doc| {
                                            let doc = maybe_doc.unwrap();
                                            doc.set("make", "Porsche").unwrap();
                                        })
                                        .unwrap();
                                    collection
                                        .find_by_id(doc7_id.clone().unwrap())
                                        .update(|maybe_doc| {
                                            let doc = maybe_doc.unwrap();
                                            doc.set("color", "brown").unwrap();
                                        })
                                        .unwrap();
                                    collection
                                        .find_by_id(doc7_id.clone().unwrap())
                                        .evict()
                                        .unwrap();

                                    tx.commit_changes()
                                })
                                .unwrap();

                            use dittolive_ditto::store::batch::DocChangeKind;
                            for (i, result) in results2.iter().enumerate() {
                                match i {
                                    0 => {
                                        let DocChangeKind::Updated = result.kind else {
                                            panic!("Incorrect result");
                                        };
                                        assert_eq!(
                                            result.collection_name.to_string(),
                                            collection_1_name
                                        );
                                        let result =
                                            collection_1.find_by_id(result.doc_id.clone()).exec();
                                        assert!(result.is_err());
                                    }
                                    1 => {
                                        let DocChangeKind::Updated = result.kind else {
                                            panic!("Incorrect result");
                                        };
                                        assert_eq!(
                                            result.collection_name.to_string(),
                                            collection_1_name
                                        );
                                        let result =
                                            collection_1.find_by_id(result.doc_id.clone()).exec();
                                        assert!(result.is_err());
                                    }
                                    2 => {
                                        let DocChangeKind::Evicted = result.kind else {
                                            panic!("Incorrect result");
                                        };
                                        assert_eq!(
                                            result.collection_name.to_string(),
                                            collection_1_name
                                        );
                                        let result =
                                            collection_1.find_by_id(result.doc_id.clone()).exec();
                                        assert!(result.is_err());
                                    }
                                    _ => {
                                        panic!("Too many results");
                                    }
                                }
                            }
                        }
                        9 => {
                            assert_payload(&docs, event, 3, 0, 1, 0, 4);
                            assert!(!docs.iter().any(|doc| eq_ignoring_id(
                                &**doc,
                                json!({ "make": "Ferrari", "color": "red" })
                            )
                            .unwrap()));
                            state_tx.send_modify(|(_count, finished)| *finished = true);
                        }
                        _ => {
                            panic!("Received update live query callback event unexpectedly");
                        }
                    }
                }
            }
        })?;

    state_rx.wait_for(|(_count, finished)| *finished).await?;
    drop(live_query);
}

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

    let (state_tx, mut state_rx) = tokio::sync::watch::channel((0, false)); // (count, finished)
    let collection = test.collection.clone();
    let handler = move |docs: Vec<BoxedDocument>, event| {
        state_tx.send_modify(|(count, _finished)| *count += 1);

        match event {
            LiveQueryEvent::Initial => {
                let (count, _finished) = *state_tx.borrow();
                assert_eq!(count, 1);
                assert!(docs.is_empty());

                let doc1 = json!({ "make": "Honda", "color": "red" });
                collection.upsert(doc1).unwrap();
            }
            LiveQueryEvent::Update {
                old_documents,
                insertions,
                ..
            } => {
                let (count, _finished) = *state_tx.borrow();
                assert_eq!(count, 2);
                assert_eq!(docs.len(), 1);
                assert_eq!(old_documents.len(), 0);
                assert_eq!(*insertions, [0]);

                let expected_doc = json!({ "make": "Honda", "color": "red" });
                assert_eq_ignoring_id(&**docs.first().unwrap(), expected_doc).unwrap();

                state_tx.send_modify(|(_count, finished)| *finished = true);
            }
        }
    };
    let live_query = test
        .collection
        .find_with_args("color == $args.color", json!({"color": "red"}))
        .observe_local(handler)?;

    state_rx.wait_for(|(_count, finished)| *finished).await?;
    drop(live_query);
}

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

    #[derive(Debug, Serialize, Deserialize)]
    struct Car {
        make: String,
        color: String,
    }

    let (state_tx, mut state_rx) = tokio::sync::watch::channel((0, false)); // (count, finished)
    let collection = test.collection.clone();
    let live_query = test
        .collection
        .find("color == 'red'")
        .observe_local(move |docs, event| {
            state_tx.send_modify(|(count, _finished)| *count += 1);
            let typed = docs
                .into_iter()
                .map(|it| it.typed::<Car>().unwrap())
                .collect::<Vec<_>>();

            match event {
                LiveQueryEvent::Initial => {
                    let (count, _finished) = *state_tx.borrow();
                    assert_eq!(count, 1);
                    assert_eq!(typed.len(), 0);

                    let car = Car {
                        make: "Aston Martin".to_string(),
                        color: "red".to_string(),
                    };
                    collection.upsert(car).unwrap();
                }
                LiveQueryEvent::Update {
                    old_documents,
                    insertions,
                    ..
                } => {
                    let (count, _finished) = *state_tx.borrow();
                    assert_eq!(count, 2);
                    assert_eq!(typed.len(), 1);
                    assert_eq!(old_documents.len(), 0);
                    assert_eq!(*insertions, [0]);
                    assert_eq!(typed[0].make, "Aston Martin");
                    assert_eq!(typed[0].color, "red");

                    state_tx.send_modify(|(_count, finished)| *finished = true);
                }
            }
        })
        .unwrap();

    state_rx.wait_for(|(_count, finished)| *finished).await?;
    drop(live_query);
}

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

    #[derive(Debug, Clone, Serialize, Deserialize)]
    struct Car {
        make: String,
        color: String,
    }

    let car = Car {
        make: "Honda".to_string(),
        color: "red".to_string(),
    };
    let doc_id = test.collection.upsert(car.clone())?;

    let (state_tx, mut state_rx) = tokio::sync::watch::channel((0, false)); // (count, finished)
    let live_query = test
        .collection
        .clone()
        .find("color == 'red'")
        .observe_local(move |docs, event| {
            state_tx.send_modify(|(count, _finished)| *count += 1);

            let typed = docs
                .into_iter()
                .map(|it| it.typed::<Car>().unwrap())
                .collect::<Vec<_>>();

            let (count, _finished) = *state_tx.borrow();
            match event {
                LiveQueryEvent::Initial => {
                    assert_eq!(count, 1);
                    assert_eq!(typed.len(), 1);
                    assert_eq!(typed[0].make, "Honda");
                    assert_eq!(typed[0].color, "red");

                    let mut car = car.clone();
                    car.make = "Ferrari".to_string();
                    test.collection
                        .find_by_id(&doc_id)
                        .update_doc(&car)
                        .unwrap();
                }
                LiveQueryEvent::Update {
                    old_documents,
                    updates,
                    ..
                } => {
                    let typed_old_docs = old_documents
                        .into_iter()
                        .map(|it| it.typed::<Car>().unwrap())
                        .collect::<Vec<_>>();

                    assert_eq!(count, 2);
                    assert_eq!(typed.len(), 1);
                    assert_eq!(typed_old_docs.len(), 1);
                    assert_eq!(typed_old_docs[0].make, "Honda");
                    assert_eq!(typed_old_docs[0].color, "red");
                    assert_eq!(*updates, [0]);
                    assert_eq!(typed[0].make, "Ferrari");
                    assert_eq!(typed[0].color, "red");
                    state_tx.send_modify(|(_count, finished)| *finished = true);
                }
            }
        });

    state_rx.wait_for(|(_count, finished)| *finished).await?;
    drop(live_query);
}

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

    let _doc_id = test
        .collection
        .upsert(json!({ "make": "Jaguar", "color": "blue" }))?;

    let (state_tx, mut state_rx) = tokio::sync::watch::channel((0, false)); // (count, finished)
    let live_query = test
        .collection
        .clone()
        .find_all()
        .observe_local(move |docs, event| {
            state_tx.send_modify(|(count, _finished)| *count += 1);
            let (count, _finished) = *state_tx.borrow();

            match event {
                LiveQueryEvent::Initial => {
                    assert_eq!(count, 1);
                    assert_eq!(docs.len(), 1);
                    assert_eq_ignoring_id(&*docs[0], json!({ "make": "Jaguar", "color": "blue" }))
                        .unwrap();

                    test.collection
                        .upsert(json!({ "make": "Aston Martin", "color": "red" }))
                        .unwrap();
                }
                LiveQueryEvent::Update {
                    old_documents,
                    insertions,
                    ..
                } => {
                    assert_eq!(count, 2);
                    assert_eq!(docs.len(), 2);
                    assert_eq!(old_documents.len(), 1);
                    assert_eq!(insertions.len(), 1);
                    assert!(docs.iter().any(|doc| eq_ignoring_id(
                        &**doc,
                        json!({ "make": "Jaguar", "color": "blue" })
                    )
                    .unwrap()));
                    assert!(docs.iter().any(|doc| eq_ignoring_id(
                        &**doc,
                        json!({ "make": "Aston Martin", "color": "red" })
                    )
                    .unwrap()));
                    state_tx.send_modify(|(_count, finished)| *finished = true);
                }
            }
        });

    state_rx.wait_for(|(_count, finished)| *finished).await?;
    drop(live_query);
}

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

    let _doc_id1 = test
        .collection
        .upsert(json!({ "make": "Honda", "color": "red", "year": 1981 }))?;
    let _doc_id2 = test
        .collection
        .upsert(json!({ "make": "Honda", "color": "red", "year": 1988 }))?;

    let (state_tx, mut state_rx) = tokio::sync::watch::channel((0, false)); // (count, finished)
    let live_query = test
        .collection
        .clone()
        .find("color == 'red'")
        .sort(vec![COrderByParam {
            query_c_str: "year\0".try_into()?,
            direction: QuerySortDirection::Descending,
        }])
        .observe_local(move |docs, event| {
            state_tx.send_modify(|(count, _finished)| *count += 1);
            let (count, _finished) = *state_tx.borrow();

            match event {
                LiveQueryEvent::Initial => {
                    assert_eq!(count, 1);
                    assert_eq!(docs.len(), 2);
                    assert_eq!(docs[0].get::<i32>("year").unwrap(), 1988);
                    assert_eq!(docs[1].get::<i32>("year").unwrap(), 1981);

                    test.collection
                        .upsert(json!({ "make": "Honda", "color": "red", "year": 1985 }))
                        .unwrap();
                }
                LiveQueryEvent::Update {
                    old_documents,
                    insertions,
                    ..
                } => {
                    assert_eq!(count, 2);
                    assert_eq!(docs.len(), 3);
                    assert_eq!(old_documents.len(), 2);
                    assert_eq!(*insertions, [1]);
                    assert_eq!(docs[0].get::<i32>("year").unwrap(), 1988);
                    assert_eq!(docs[1].get::<i32>("year").unwrap(), 1985);
                    assert_eq!(docs[2].get::<i32>("year").unwrap(), 1981);

                    state_tx.send_modify(|(_count, finished)| *finished = true);
                }
            }
        });

    state_rx.wait_for(|(_count, finished)| *finished).await?;
    drop(live_query);
}

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

    let _doc_id1 = test
        .collection
        .upsert(json!({ "make": "Honda", "color": "red", "year": 1981 }))?;
    let doc_id2 = test
        .collection
        .upsert(json!({ "make": "Honda", "color": "red", "year": 1988 }))?;

    let (state_tx, mut state_rx) = tokio::sync::watch::channel((0, false)); // (count, finished)
    let live_query = test
        .collection
        .clone()
        .find("color == 'red'")
        .sort(vec![COrderByParam {
            query_c_str: "year\0".try_into()?,
            direction: QuerySortDirection::Descending,
        }])
        .observe_local(move |docs, event| {
            state_tx.send_modify(|(count, _finished)| *count += 1);
            let (count, _finished) = *state_tx.borrow();

            match event {
                LiveQueryEvent::Initial => {
                    assert_eq!(count, 1);
                    assert_eq!(docs.len(), 2);
                    assert_eq!(docs[0].get::<i32>("year").unwrap(), 1988);
                    assert_eq!(docs[1].get::<i32>("year").unwrap(), 1981);

                    test.collection
                        .find_by_id(&doc_id2)
                        .update(|maybe_doc| {
                            let doc = maybe_doc.unwrap();
                            doc.set("year", 1979).unwrap();
                        })
                        .unwrap();
                }
                LiveQueryEvent::Update {
                    old_documents,
                    moves,
                    updates,
                    ..
                } => {
                    assert_eq!(count, 2);
                    assert_eq!(docs.len(), 2);
                    assert_eq!(old_documents.len(), 2);
                    assert_eq!(moves.len(), 2);
                    assert_eq!(moves[0].from, 1);
                    assert_eq!(moves[0].to, 0);
                    assert_eq!(moves[1].from, 0);
                    assert_eq!(moves[1].to, 1);
                    assert_eq!(*updates, [1]);

                    assert_eq!(docs[0].get::<i32>("year").unwrap(), 1981);
                    assert_eq!(docs[1].get::<i32>("year").unwrap(), 1979);

                    state_tx.send_modify(|(_count, finished)| *finished = true);
                }
            }
        });

    state_rx.wait_for(|(_count, finished)| *finished).await?;
    drop(live_query);
}

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

    let (state_tx, mut state_rx) = tokio::sync::watch::channel((0, false)); // (count, finished)
    let live_query1 =
        test.collection
            .find("color == 'red'")
            .observe_local(move |_docs, _event| {
                state_tx.send_modify(|(count, _finished)| *count += 1);
                let (count, _finished) = *state_tx.borrow();

                if count == 2 {
                    state_tx.send_modify(|(_count, finished)| *finished = true);
                }
            })?;

    let live_query2 = test
        .collection
        .find("make == 'honda'")
        .observe_local(move |_docs, _event| {})?;

    let live_queries = vec![live_query1, live_query2];

    let hash1 = test.ditto.store().queries_hash(&live_queries)?;
    assert_eq!(test.ditto.store().queries_hash(&live_queries)?, hash1);
    assert!(test.ditto.store().queries_hash(&live_queries[1..2])? != hash1);
    assert!(test.ditto.store().queries_hash(&[])? != hash1);

    let hash1_mnemonic = test.ditto.store().queries_hash_mnemonic(&live_queries)?;
    assert_eq!(
        test.ditto.store().queries_hash_mnemonic(&live_queries)?,
        hash1_mnemonic
    );
    assert!(
        test.ditto
            .store()
            .queries_hash_mnemonic(&live_queries[1..2])?
            != hash1_mnemonic
    );
    assert!(test.ditto.store().queries_hash_mnemonic(&[])? != hash1_mnemonic);

    test.collection.upsert(json!({ "color": "red" }))?;
    let hash2 = test.ditto.store().queries_hash(&live_queries)?;
    let hash2_mnemonic = test.ditto.store().queries_hash_mnemonic(&live_queries)?;
    assert!(hash1 != hash2);
    assert!(hash1_mnemonic != hash2_mnemonic);

    state_rx.wait_for(|(_count, finished)| *finished).await?;
    drop(live_queries);
}

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

    let long_text1_orig = "a".repeat(50_000);
    let long_text2_orig = "b".repeat(50_000);

    let long_text1 = long_text1_orig.clone();
    let long_text2 = long_text2_orig.clone();

    let (state_tx, mut state_rx) = tokio::sync::watch::channel((0, false)); // (count, finished)
    let live_query = test
        .collection
        .find_all()
        .observe_local(move |docs, event| {
            if let LiveQueryEvent::Update { .. } = event {
                let text1 = docs[0].get::<String>("text").unwrap();
                let text2 = docs[1].get::<String>("text").unwrap();
                let text_set = std::collections::HashSet::from([text1, text2]);
                if text_set.contains(&long_text1) && text_set.contains(&long_text2) {
                    state_tx.send_modify(|(_count, finished)| *finished = true);
                }
            }
        });

    let result = test.ditto.store().with_batched_write(move |mut tx| {
        let mut collection = tx.collection(&test.collection_name);
        collection
            .upsert(json!({ "text": long_text1_orig }))
            .unwrap();
        collection
            .upsert(json!({ "text": long_text2_orig }))
            .unwrap();

        tx.commit_changes()
    })?;
    assert_eq!(result.len(), 2);

    state_rx.wait_for(|(_count, finished)| *finished).await?;
    drop(live_query);
}

// test_with_next_signal // QueryBuilder `with_next_signal` not implemented
// test_local_with_next_signal // QueryBuilder `with_next_signal` not implemented

#[throws]
#[test]
fn test_immediate_stop() {
    let test = setup()?;
    let live_query = test
        .collection
        .find_all()
        .observe_local(move |_docs, _event| {})?;
    drop(live_query);
}

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

    let (state_tx, mut state_rx) = tokio::sync::watch::channel((0, false)); // (count, finished)
    let live_query = test
        .collection
        .clone()
        .find_all()
        .observe_local(move |docs, event| {
            state_tx.send_modify(|(count, _finished)| *count += 1);
            let (count, _finished) = *state_tx.borrow();

            match count {
                1 => {
                    let LiveQueryEvent::Initial = event else {
                        panic!("Incorrect live query event type");
                    };
                    assert_eq!(docs.len(), 0);
                    assert_eq!(event.hash(&docs).unwrap(), 3244421341483603138);
                    assert_eq!(
                        event.hash_mnemonic(&docs).unwrap(),
                        "love-prize-janet--chemist-orchid-ford"
                    );

                    test.collection
                        .upsert_with_strategy(
                            json!({ "_id": "1", "a": "test" }),
                            WriteStrategy::InsertDefaultIfAbsent,
                        )
                        .unwrap();
                }
                2 => {
                    let LiveQueryEvent::Update { old_documents, .. } = &event else {
                        panic!("Incorrect live query event type");
                    };

                    assert_eq!(docs.len(), 1);
                    assert_eq!(old_documents.len(), 0);
                    assert_eq!(event.hash(&docs).unwrap(), 10778274233543275271);
                    assert_eq!(
                        event.hash_mnemonic(&docs).unwrap(),
                        "marble-benny-memo--society-forum-nikita"
                    );

                    test.collection
                        .upsert_with_strategy(
                            json!({ "_id": "2", "a": "nother" }),
                            WriteStrategy::InsertDefaultIfAbsent,
                        )
                        .unwrap();
                }
                3 => {
                    let LiveQueryEvent::Update { old_documents, .. } = &event else {
                        panic!("Incorrect live query event type");
                    };

                    assert_eq!(docs.len(), 2);
                    assert_eq!(old_documents.len(), 1);
                    assert_eq!(event.hash(&docs).unwrap(), 5198709967650152102);
                    assert_eq!(
                        event.hash_mnemonic(&docs).unwrap(),
                        "chief-signal-ruby--david-donor-micro"
                    );

                    state_tx.send_modify(|(_count, finished)| *finished = true);
                }
                n => {
                    panic!("Unexpected number of calls ({n}) to observe_local");
                }
            }
        })?;

    state_rx.wait_for(|(_count, finished)| *finished).await?;
    drop(live_query);
}

///////////////////////////////////////////////////////////////////////

/// Test a live query for a single document, with document IDs of different
/// types.
#[test]
fn single_document_with_different_ids() {
    let ditto = common::get_builder_ditto(None).unwrap();
    single_document_with_different_ids_impl(ditto);

    let ditto = common::get_revamp_ditto(None).unwrap();
    single_document_with_different_ids_impl(ditto);

    let ditto = common::get_revamp_ditto_async_sync(None).unwrap();
    single_document_with_different_ids_impl(ditto);
}

fn single_document_with_different_ids_impl(ditto: dittolive_ditto::prelude::Ditto) {
    let store = ditto.store();
    let uuid = uuid::Uuid::new_v4().to_string();
    let collection = store.collection(&uuid).unwrap();

    let raw_doc_ids: Vec<serde_json::Value> = vec![
        "boring_old_string".to_string().into(),
        1.into(),
        0.into(),
        123.into(),
        9999.into(),
        false.into(),
        true.into(),
        (0..64)
            .map(|_| rand::random::<u8>())
            .collect::<Vec<u8>>()
            .into(),
        vec!["a", "abc", "z89{{}}@£!fv>?!,[](){{}}000"].into(),
        json!({"a": "b", "__num__@£$%^&{})(|,,./!?": -7123}),
    ];

    raw_doc_ids.iter().for_each(|raw_id| {
        let doc_id = DocumentId::new(raw_id).unwrap();
        let doc = TestType {
            make: String::from("Honda"),
            color: TestColor::Red,
            id: Some(doc_id.clone()),
            ..Default::default()
        };

        // make copies for the test
        let expected_doc_id = doc_id.clone();
        let expected_doc = doc.clone();

        let returned_id = collection.upsert(doc).unwrap();
        assert_eq!(returned_id, doc_id);

        let finished = Arc::new(AtomicBool::new(false));
        let finished_clone = Arc::clone(&finished);
        let counter = Arc::new(Mutex::new(0));
        assert_eq!(*counter.lock().unwrap(), 0);

        let handler = move |doc: Option<BoxedDocument>, event: SingleDocumentLiveQueryEvent| {
            let counter_mtx = &*counter; // move (copy) and reborrow
            if let Ok(mut counter) = counter_mtx.lock() {
                *counter += 1;
            }
            if event.is_initial() {
                let doc = doc.expect("doc should be Some");
                if let Ok(counter) = counter_mtx.lock() {
                    assert_eq!(*counter, 1);
                }
                assert_eq!(expected_doc_id, doc.id());
                let first_doc = doc.typed::<TestType>().unwrap();
                assert_eq!(first_doc, expected_doc);
                finished_clone.store(true, Ordering::SeqCst);
            } else {
                unimplemented!("Unexpected non-initial event received");
            }
        };
        let lq = collection
            .find_by_id(doc_id)
            .observe_local(handler)
            .unwrap();

        // wait for the LQ thread to finish
        while !finished.load(Ordering::SeqCst) {
            std::thread::yield_now();
        }

        drop(lq);
    });
}

#[test]
fn single_document_lq() -> Result<(), Box<dyn std::error::Error>> {
    let ditto = common::get_builder_ditto(None).unwrap();
    single_document_lq_impl(ditto)?;

    let ditto = common::get_revamp_ditto(None).unwrap();
    single_document_lq_impl(ditto)?;

    Ok(())
}

fn single_document_lq_impl(
    ditto: dittolive_ditto::prelude::Ditto,
) -> Result<(), Box<dyn std::error::Error>> {
    let store = ditto.store();
    let uuid = uuid::Uuid::new_v4().to_string();
    let collection = store.collection(&uuid).unwrap(); // Gets moved into Handler

    let finished = Arc::new(AtomicBool::new(false));
    let finished_clone = Arc::clone(&finished);
    let counter = Arc::new(Mutex::new(0));
    assert_eq!(*counter.lock().unwrap(), 0);
    let doc_id = DocumentId::new(&"a".to_string()).unwrap();
    let doc_id_clone = doc_id.clone();
    let collection_clone = collection.clone(); // Ref is not allowed as must be 'static

    let handler = move |doc: Option<BoxedDocument>, event: SingleDocumentLiveQueryEvent| {
        let counter_mtx = &*counter; // move (copy) and reborrow
        let mut counter = counter_mtx.lock().unwrap();
        *counter += 1;

        match *counter {
            1 => {
                assert!(event.is_initial());
                assert!(doc.is_none());
                let id = collection_clone
                    .upsert(json!({ "a": "b", "_id": doc_id_clone }))
                    .unwrap();
                assert_eq!(id, doc_id_clone);
            }
            2 => {
                assert!(!event.is_initial());
                assert_eq!(doc.unwrap().get::<String>("a").unwrap(), "b");
                finished_clone.store(true, Ordering::SeqCst);
            }
            _ => unreachable!(
                "Live query callback should not have been called {} times",
                *counter
            ),
        }
    };
    let lq = collection.find_by_id(doc_id).observe_local(handler)?;

    // wait for the LQ thread to finish
    while !finished.load(Ordering::SeqCst) {
        std::thread::yield_now();
    }

    drop(lq); // In theory, the clone should drop here
    drop(collection);
    drop(ditto);

    Ok(())
}

#[test]
fn hash_and_hash_mnemonic() -> Result<(), Box<dyn std::error::Error>> {
    let ditto = common::get_builder_ditto(None).unwrap();
    hash_and_hash_mnemonic_impl(ditto)?;

    let ditto = common::get_revamp_ditto(None).unwrap();
    hash_and_hash_mnemonic_impl(ditto)?;

    Ok(())
}

fn hash_and_hash_mnemonic_impl(
    ditto: dittolive_ditto::prelude::Ditto,
) -> Result<(), Box<dyn std::error::Error>> {
    let store = ditto.store();
    let collection = store.collection("test").unwrap(); // Gets moved into Handler

    let finished = Arc::new(AtomicBool::new(false));
    let finished_clone = Arc::clone(&finished);
    let counter = Arc::new(Mutex::new(0));
    assert_eq!(*counter.lock().unwrap(), 0);
    let doc_id = DocumentId::new(&"1".to_string()).unwrap();
    let collection_clone = collection.clone(); // Ref is not allowed as must be 'static

    let handler = move |docs: Vec<BoxedDocument>, event: LiveQueryEvent| {
        let counter_mtx = &*counter; // move (copy) and reborrow
        let mut counter = counter_mtx.lock().unwrap();
        *counter += 1;

        match *counter {
            1 => {
                assert_matches_!(event, LiveQueryEvent::Initial);
                assert!(docs.is_empty());
                assert_eq!(event.hash(&docs).unwrap(), 3244421341483603138);
                assert_eq!(
                    event.hash_mnemonic(&docs).unwrap(),
                    "love-prize-janet--chemist-orchid-ford"
                );
                let id = collection_clone
                    .upsert_with_strategy(
                        json!({ "a": "test", "_id": doc_id }),
                        WriteStrategy::InsertDefaultIfAbsent,
                    )
                    .unwrap();
                assert_eq!(id, doc_id);
            }
            2 => {
                assert_matches_!(event, LiveQueryEvent::Update { .. });
                assert_eq!(event.hash(&docs).unwrap(), 10778274233543275271);
                assert_eq!(
                    event.hash_mnemonic(&docs).unwrap(),
                    "marble-benny-memo--society-forum-nikita"
                );
                assert_eq!(docs[0].get::<String>("a").unwrap(), "test");
                finished_clone.store(true, Ordering::SeqCst);
            }
            _ => unreachable!(
                "Live query callback should not have been called {} times",
                *counter
            ),
        }
    };
    let lq = collection.find("a == 'test'").observe_local(handler)?;

    // wait for the LQ thread to finish
    while !finished.load(Ordering::SeqCst) {
        std::thread::yield_now();
    }

    drop(lq); // In theory, the clone should drop here
    drop(collection);
    drop(ditto);

    Ok(())
}

#[test]
fn single_doc_hash_and_hash_mnemonic() -> Result<(), Box<dyn std::error::Error>> {
    let ditto = common::get_builder_ditto(None).unwrap();
    single_doc_hash_and_hash_mnemonic_impl(ditto)?;

    let ditto = common::get_revamp_ditto(None).unwrap();
    single_doc_hash_and_hash_mnemonic_impl(ditto)?;

    Ok(())
}

fn single_doc_hash_and_hash_mnemonic_impl(
    ditto: dittolive_ditto::prelude::Ditto,
) -> Result<(), Box<dyn std::error::Error>> {
    let store = ditto.store();
    let collection = store.collection("test").unwrap(); // Gets moved into Handler

    let finished = Arc::new(AtomicBool::new(false));
    let finished_clone = Arc::clone(&finished);
    let counter = Arc::new(Mutex::new(0));
    assert_eq!(*counter.lock().unwrap(), 0);
    let doc_id = DocumentId::new(&"1".to_string()).unwrap();
    let doc_id_clone = doc_id.clone();
    let collection_clone = collection.clone(); // Ref is not allowed as must be 'static

    let handler = move |doc: Option<BoxedDocument>, event: SingleDocumentLiveQueryEvent| {
        let counter_mtx = &*counter; // move (copy) and reborrow
        let mut counter = counter_mtx.lock().unwrap();
        *counter += 1;

        match *counter {
            1 => {
                assert!(event.is_initial());
                assert!(doc.is_none());
                assert_eq!(event.hash(&doc).unwrap(), 3244421341483603138);
                assert_eq!(
                    event.hash_mnemonic(&doc).unwrap(),
                    "love-prize-janet--chemist-orchid-ford"
                );
                let id = collection_clone
                    .upsert_with_strategy(
                        json!({ "a": "test", "_id": doc_id_clone }),
                        WriteStrategy::InsertDefaultIfAbsent,
                    )
                    .unwrap();
                assert_eq!(id, doc_id_clone);
            }
            2 => {
                assert!(!event.is_initial());
                assert_eq!(event.hash(&doc).unwrap(), 10778274233543275271);
                assert_eq!(
                    event.hash_mnemonic(&doc).unwrap(),
                    "marble-benny-memo--society-forum-nikita"
                );
                assert_eq!(doc.unwrap().get::<String>("a").unwrap(), "test");
                finished_clone.store(true, Ordering::SeqCst);
            }
            _ => unreachable!(
                "Live query callback should not have been called {} times",
                *counter
            ),
        }
    };
    let lq = collection.find_by_id(doc_id).observe_local(handler)?;

    // wait for the LQ thread to finish
    while !finished.load(Ordering::SeqCst) {
        std::thread::yield_now();
    }

    drop(lq); // In theory, the clone should drop here
    drop(collection);
    drop(ditto);

    Ok(())
}

#[test]
fn live_query_mut() -> Result<(), Box<dyn std::error::Error>> {
    let ditto = common::get_builder_ditto(None).unwrap();
    live_query_mut_impl(ditto)?;

    let ditto = common::get_revamp_ditto(None).unwrap();
    live_query_mut_impl(ditto)?;

    Ok(())
}

fn live_query_mut_impl(
    ditto: dittolive_ditto::prelude::Ditto,
) -> Result<(), Box<dyn std::error::Error>> {
    let store = ditto.store();
    let collection = store.collection("test").unwrap();
    let mut v = vec![];
    let _lq = collection.find_all().observe_local(move |_, _| {
        // Should be able to use mutable state in a LQ handler
        v.push(1);
    })?;
    Ok(())
}

#[test]
fn sorting_with_an_update() -> Result<(), Box<dyn std::error::Error>> {
    let ditto = common::get_builder_ditto(None).unwrap();
    sorting_with_an_update_impl(ditto)?;

    let ditto = common::get_revamp_ditto(None).unwrap();
    sorting_with_an_update_impl(ditto)?;

    Ok(())
}

fn sorting_with_an_update_impl(
    ditto: dittolive_ditto::prelude::Ditto,
) -> Result<(), Box<dyn std::error::Error>> {
    let store = ditto.store();
    let uuid = uuid::Uuid::new_v4().to_string();
    let collection = store.collection(&uuid).unwrap();

    let doc1_id = collection.upsert(json!({ "a": 4 })).unwrap();
    let doc2_id = collection.upsert(json!({ "a": 7 })).unwrap();

    let finished = Arc::new(AtomicBool::new(false));
    let finished_clone = Arc::clone(&finished);
    let counter = Arc::new(Mutex::new(0));
    assert_eq!(*counter.lock().unwrap(), 0);

    let collection_clone = collection.clone();

    let handler = move |docs: Vec<BoxedDocument>, event| {
        let counter_mtx = &*counter; // move (copy) and reborrow
        let mut counter = counter_mtx.lock().unwrap();
        *counter += 1;

        match *counter {
            1 => {
                assert_matches_!(event, LiveQueryEvent::Initial);
                assert_eq!(docs.len(), 2);
                assert_eq!(docs[0].id(), doc1_id);
                assert_eq!(docs[1].id(), doc2_id);
                assert_eq!(docs[0].get::<i64>("a").unwrap(), 4);
                assert_eq!(docs[1].get::<i64>("a").unwrap(), 7);

                let id = collection_clone
                    .upsert(json!({ "_id": doc2_id, "a": 2 }))
                    .unwrap();
                assert_eq!(id, doc2_id);
            }
            2 => {
                assert_matches_!(event, LiveQueryEvent::Update { .. });
                match event {
                    LiveQueryEvent::Update {
                        old_documents,
                        updates,
                        moves,
                        ..
                    } => {
                        assert_eq!(docs.len(), 2);
                        assert_eq!(docs[0].id(), doc2_id);
                        assert_eq!(docs[1].id(), doc1_id);
                        assert_eq!(old_documents.len(), 2);
                        assert_eq!(updates.len(), 1);
                        assert_eq!(moves.len(), 2);
                        assert_eq!(docs[0].get::<i64>("a").unwrap(), 2);
                        assert_eq!(docs[1].get::<i64>("a").unwrap(), 4);
                        assert_eq!(updates[0], 0);
                        assert_eq!(moves[0].from, 1);
                        assert_eq!(moves[0].to, 0);
                        assert_eq!(moves[1].from, 0);
                        assert_eq!(moves[1].to, 1);
                    }
                    _ => unreachable!(),
                };

                finished_clone.store(true, Ordering::SeqCst);
            }
            _ => unreachable!(
                "Live query callback should not have been called {} times",
                *counter
            ),
        }
    };
    let lq = collection
        .find_all()
        .sort(vec![COrderByParam {
            query_c_str: ::std::ffi::CString::new("a").unwrap().as_c_str().into(),
            direction: QuerySortDirection::Ascending,
        }])
        .observe_local(handler)?;

    // wait for the LQ thread to finish
    while !finished.load(Ordering::SeqCst) {
        std::thread::yield_now();
    }

    drop(lq);
    Ok(())
}
