dittolive-ditto 1.1.10-experimental.all-product-headers-yaml-issue.1

//! Tests of live query functionality
#[macro_use]
extern crate serde_json;

use assert_matches::assert_matches as assert_matches_;

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

use common::*;

mod common;

/// Test inserting a document inside of a LQ Callback
/// triggers the correct Initial LiveQueryEvent
#[test]
fn initial_lq() -> Result<(), Box<dyn std::error::Error>> {
    let ditto = common::get_ditto().unwrap();
    let store = ditto.store();
    let uuid = uuid::Uuid::new_v4().to_string();
    let collection = store.collection(&uuid).unwrap();

    let doc1_id = DocumentId::new(&"1".to_string()).unwrap();
    let doc1 = TestType {
        make: String::from("Honda"),
        color: TestColor::Red,
        id: Some(doc1_id.clone()),
        ..Default::default()
    };
    let expected_doc = doc1.clone(); // make a copy for the test

    let doc1_id_ret = collection.upsert(doc1).unwrap();
    assert_eq!(doc1_id_ret, doc1_id);

    let doc2 = TestType {
        make: String::from("Honda"),
        color: TestColor::Crimson,
        ..Default::default()
    };
    let _doc2_id = collection.upsert(doc2).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 handler = move |docs: Vec<BoxedDocument>, event| {
        let counter_mtx = &*counter; // move (copy) and reborrow
        if let Ok(mut counter) = counter_mtx.lock() {
            *counter += 1;
        }
        match event {
            LiveQueryEvent::Initial => {
                if let Ok(counter) = counter_mtx.lock() {
                    assert_eq!(*counter, 1);
                }
                assert_eq!(docs.len(), 1);
                assert_eq!(doc1_id, docs[0].id());
                let first_doc = docs[0].typed::<TestType>().unwrap();
                assert_eq!(first_doc, expected_doc);
                finished_clone.store(true, Ordering::SeqCst);
            }
            LiveQueryEvent::Update { .. } => {
                unimplemented!("Unexpected Update Received");
            }
        }
    };
    let lq = collection.find("color == \'Red\'").observe(handler)?; // String values in queries must be single quoted

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

    lq.stop();

    Ok(())
}

/// Test inserting a new document triggers the correct Update LiveQueryEvent
#[test]
fn insert_lq() -> Result<(), Box<dyn std::error::Error>> {
    let ditto = common::get_ditto().unwrap();
    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 honda_id = DocumentId::new(&"honda".to_string()).unwrap();
    let collection_clone = collection.clone(); // Ref is not allowed as must be 'static

    let handler = move |docs: Vec<BoxedDocument>, event| {
        let counter_mtx = &*counter; // move (copy) and reborrow
        if let Ok(mut counter) = counter_mtx.lock() {
            *counter += 1;
        }
        match event {
            LiveQueryEvent::Initial => {
                // this is a bit overkill but it keeps our Handler Fn
                if let Ok(counter) = counter_mtx.lock() {
                    assert_eq!(*counter, 1);
                }
                assert!(docs.is_empty());
                // insert a matching doc
                let doc1 = TestType {
                    make: String::from("Honda"),
                    color: TestColor::Red,
                    id: Some(honda_id.clone()),
                    ..Default::default()
                };
                // 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_clone.upsert(doc1).unwrap();
            }
            LiveQueryEvent::Update {
                old_documents,
                insertions,
                ..
            } => {
                let expected_insertions = [0_usize; 1];
                let expected_doc = TestType {
                    make: String::from("Honda"),
                    color: TestColor::Red,
                    id: Some(honda_id.clone()),
                    ..Default::default()
                };
                if let Ok(counter) = counter_mtx.lock() {
                    assert_eq!(*counter, 2);
                }
                let first_doc = docs[0].typed::<TestType>().unwrap();
                assert_eq!(docs.len(), 1);
                assert!(old_documents.is_empty());
                assert_eq!(insertions, expected_insertions.into());
                assert_eq!(first_doc, expected_doc);
                finished_clone.store(true, Ordering::SeqCst);
            }
        }
    };
    let lq = collection.find("color == \'Red\'").observe(handler)?; // String values in queries must be single quoted

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

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

    Ok(())

    // The core ditto ptr is freed here unless something is holding the test
    // open
}

/// 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_ditto().unwrap();
    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_v2(handler).unwrap();

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

        lq.stop();
    });
}

/// Test inserting a new document triggers the correct Update LiveQueryEvent
#[test]
fn insert_lq_with_query_args() -> Result<(), Box<dyn std::error::Error>> {
    let ditto = common::get_ditto().unwrap();
    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 honda_id = DocumentId::new(&"honda".to_string()).unwrap();
    let collection_clone = collection.clone(); // Ref is not allowed as must be 'static

    let handler = move |docs: Vec<BoxedDocument>, event| {
        let counter_mtx = &*counter; // move (copy) and reborrow
        if let Ok(mut counter) = counter_mtx.lock() {
            *counter += 1;
        }
        match event {
            LiveQueryEvent::Initial => {
                // this is a bit overkill but it keeps our Handler Fn
                if let Ok(counter) = counter_mtx.lock() {
                    assert_eq!(*counter, 1);
                }
                assert!(docs.is_empty());
                // insert a matching doc
                let doc1 = TestType {
                    make: String::from("Honda"),
                    color: TestColor::Red,
                    id: Some(honda_id.clone()),
                    ..Default::default()
                };
                // The Collection here has a reference to the underlying core
                // ditto instance which may be freed by the time this upsert is
                // called unless something holds the test open
                let _id = collection_clone.upsert(doc1).unwrap();
            }
            LiveQueryEvent::Update {
                old_documents,
                insertions,
                ..
            } => {
                let expected_insertions = [0_usize; 1];
                let expected_doc = TestType {
                    make: String::from("Honda"),
                    color: TestColor::Red,
                    id: Some(honda_id.clone()),
                    ..Default::default()
                };
                if let Ok(counter) = counter_mtx.lock() {
                    assert_eq!(*counter, 2);
                }
                let first_doc = docs[0].typed::<TestType>().unwrap();
                assert_eq!(docs.len(), 1);
                assert!(old_documents.is_empty());
                assert_eq!(insertions, expected_insertions.into());
                assert_eq!(first_doc, expected_doc);
                finished_clone.store(true, Ordering::SeqCst);
            }
        }
    };
    let lq = collection
        .find_with_args("color == $args.color", json!({"color": "Red"}))
        .observe(handler)?;

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

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

    Ok(())

    // The core ditto ptr is freed here unless something is holding the test
    // open
}

#[test]
fn single_document_lq() -> Result<(), Box<dyn std::error::Error>> {
    let ditto = common::get_ditto().unwrap();
    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_v2(handler)?;

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

    lq.stop(); // 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_ditto().unwrap();
    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(), 8298659613316817713);
                assert_eq!(
                    event.hash_mnemonic(&docs).unwrap(),
                    "antonio-romeo-king--torch-crack-transit"
                );
                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(handler)?;

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

    lq.stop(); // 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_ditto().unwrap();
    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(), 8298659613316817713);
                assert_eq!(
                    event.hash_mnemonic(&doc).unwrap(),
                    "antonio-romeo-king--torch-crack-transit"
                );
                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_v2(handler)?;

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

    lq.stop(); // 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_ditto().unwrap();
    let store = ditto.store();
    let collection = store.collection("test").unwrap();
    let mut v = vec![];
    let _lq = collection.find_all().observe(move |_, _| {
        // Should be able to use mutable state in a LQ handler
        v.push(1);
    })?;
    Ok(())
}