#![allow(deprecated)]
mod common;

use std::{
    sync::{
        atomic::{AtomicU8, Ordering},
        mpsc, Arc,
    },
    time::Duration,
};

use common::*;

// These tests are all ignored by default because they involve authenticating with the Ditto Cloud.
// Find and remove all of the #[ignore = "Auth tests disabled by default"] attributes to run them.
// Or run the `cargo test` command using:
// ```bash
// cargo test --test auth -- --ignored
// ```

/// App ID from the Ditto Cloud Portal. Must be a valid UUID.
static APP_ID: &str = "6eff5920-8ab1-43f4-8803-8187f0d389fe";

/// Provider to use when attempting a login.
static PROVIDER: &str = "portal";

/// Token from "Playground Mode Details" section on Settings page for the app ID above.
static PLAYGROUND_TOKEN: &str = "3f03f821-9f0a-4424-8186-3aaaaa23b05c";

/// JWT token scraped from portal request to endpoint:
/// `https://<appID>.cloud.ditto.live/_ditto/auth/login`
static LOGIN_TOKEN: &str = "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCIsImtpZCI6Ik9UZEJSVFJDUVRSRU1qTTJNakpHTlVORU1UUTVPRVU0TXpoRU1qazVRa0k0UkRRMk5EVTJNZyJ9.eyJpc3MiOiJodHRwczovL2RpdHRvLmF1dGgwLmNvbS8iLCJzdWIiOiJnb29nbGUtb2F1dGgyfDEwNDc5ODQ2OTkzNzk4NTQ5MjY5MiIsImF1ZCI6WyJodHRwczovL3BvcnRhbGFwaS5kaXR0by5saXZlIiwiaHR0cHM6Ly9kaXR0by5hdXRoMC5jb20vdXNlcmluZm8iXSwiaWF0IjoxNjgwNTEzNzE2LCJleHAiOjE2ODA2MDAxMTYsImF6cCI6ImVnTjNMRU9OZVUyR1hzemZ2RVkwc1NTbWM5RzcyZHB5Iiwic2NvcGUiOiJvcGVuaWQgcHJvZmlsZSBlbWFpbCJ9.lkvRobdxCHcjze9OvOTHfdeMEdvz4lhpi6kRYQSq3vhs6rznzarbsivefzHaPKiwoZPRgaRf-EO7BRBbpIz6fOFlKhFmMdNWOqhqNr4STRmC8K4nXx1CM1us8LRu5AZlzm8HjszHIpzHlNfneEXXV7QvDLAZnN_SoEh-Sc_Ue9_16UxSsZna6qaE7q-ZTaz7MBoDYi-B0PLn79t2VuzrsrQfsyhdj7Hoy7hXG1CzrLAlM0RXQMXHtZDELiDVftaqnYvc2EuuPbY496-UBSiGGHxRrHucHrzPPqgzyWprzcOfsUyCqWb3mbiS0GYkWOBx2fKpppQZpvpNstEUk-vvCQ";

/// An expired token for testing auth errors.
static EXPIRED_LOGIN_TOKEN: &str = "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCIsImtpZCI6Ik9UZEJSVFJDUVRSRU1qTTJNakpHTlVORU1UUTVPRVU0TXpoRU1qazVRa0k0UkRRMk5EVTJNZyJ9.eyJpc3MiOiJodHRwczovL2RpdHRvLmF1dGgwLmNvbS8iLCJzdWIiOiJhdXRoMHw2MTU1NDhkYjE4MGUxNjAwNmFmNWE3NTkiLCJhdWQiOlsiaHR0cHM6Ly9wb3J0YWxhcGkuZGl0dG8ubGl2ZSIsImh0dHBzOi8vZGl0dG8uYXV0aDAuY29tL3VzZXJpbmZvIl0sImlhdCI6MTY0MjQ0MTA3NywiZXhwIjoxNjQyNTI3NDc3LCJhenAiOiJlZ04zTEVPTmVVMkdYc3pmdkVZMHNTU21jOUc3MmRweSIsInNjb3BlIjoib3BlbmlkIHByb2ZpbGUgZW1haWwifQ.mCQC3pILToK9MhZXsoe3DCw4f1uD0KEpfj2i7yRCrt6R81-7yWUZb7F5OXDGkVApX4FzI12w13L_Bzk6Nx1NxJVFa_8Tmb40Fthy5UW7JjANL0itEIBM15QH_rb9Z1YVXfnSHXn3rTRaqhZeRyPZjoIMeP43WB1mDzg-JLPsxw_BHhmTeS1kkvcDATxyT21H2IyE4xZXwQHwxpiXRv9UaE_rtP4qL13oHT46wm3Isa6rXT2wJmPoq9eXFkdZXvvNJgE_xCeXlAz_GURfNsnJQn0BExzzI8k3Ti66ac-q-48vXbzv9zESfWiJQnsHXfFAbAn2pQacXThb9NYKR-hPPQ";

// Helper functions for creating Builder and Revamp style Ditto instances

/// Create a Builder-style Ditto instance with OnlineWithAuthentication identity
fn create_builder_ditto_with_auth(
    auth_event_handler: AuthEventHandler,
    ditto_root: Option<Arc<dittolive_ditto::fs::PersistentRoot>>,
) -> Result<Ditto, dittolive_ditto::error::DittoError> {
    let mut builder = Ditto::builder();

    if let Some(root) = ditto_root {
        builder = builder.with_root(root);
    } else {
        builder = builder.with_temp_dir();
    }

    let ditto = builder
        .with_identity(|ditto_root| {
            identity::OnlineWithAuthentication::new(
                ditto_root,
                AppId::from_uuid(uuid::Uuid::parse_str(APP_ID).unwrap()),
                auth_event_handler.clone(),
                true,
                None,
            )
        })?
        .with_minimum_log_level(LogLevel::Error)
        .build()?;
    ditto.disable_sync_with_v3().unwrap();
    Ok(ditto)
}

/// Create a Revamp-style Ditto instance with OnlineWithAuthentication identity
fn create_revamp_ditto_with_auth(
    auth_event_handler: AuthEventHandler,
    persistence_dir: Option<std::path::PathBuf>,
) -> Result<Ditto, dittolive_ditto::error::DittoError> {
    let app_id = AppId::from_uuid(uuid::Uuid::parse_str(APP_ID).unwrap());

    // For revamp style, use DittoConfig::new with Server connect type
    let mut config = DittoConfig::new(
        app_id.to_string(),
        DittoConfigConnect::Server {
            url: format!("https://{}.cloud.ditto.live", app_id)
                .parse()
                .unwrap(),
        },
    );

    if let Some(dir) = persistence_dir {
        config = config.with_persistence_directory(dir);
    } else {
        let tempdir = tempfile::tempdir().expect("should create tempdir");
        config = config.with_persistence_directory(tempdir.keep());
    }

    let ditto = Ditto::open_sync(config)?;

    // Set up expiration handler with simple struct
    let auth = ditto.auth().unwrap();
    let handler = AuthExpirationHandler {
        counter: Arc::clone(&auth_event_handler.authentication_required_times_called),
        sender: Arc::clone(&auth_event_handler.auth_required_tx),
    };
    auth.set_expiration_handler(handler);

    ditto.disable_sync_with_v3().unwrap();
    Ok(ditto)
}

/// Simple handler for auth expiration events in revamp style
struct AuthExpirationHandler {
    counter: Arc<AtomicU8>,
    sender: Arc<mpsc::SyncSender<()>>,
}

impl dittolive_ditto::ditto::init::DittoAuthExpirationHandler for AuthExpirationHandler {
    async fn on_expiration(&self, _ditto: &Ditto, _seconds_remaining: Duration) {
        self.counter.fetch_add(1, Ordering::SeqCst);
        let _ = self.sender.send(());
    }
}

/// Create a Builder-style Ditto instance with OnlinePlayground identity
fn create_builder_ditto_playground() -> Result<Ditto, dittolive_ditto::error::DittoError> {
    let ditto = Ditto::builder()
        .with_temp_dir()
        .with_identity(|ditto_root| {
            identity::OnlinePlayground::new(
                ditto_root,
                AppId::from_uuid(uuid::Uuid::parse_str(APP_ID).unwrap()),
                PLAYGROUND_TOKEN.to_string(),
                true,
                None,
            )
        })?
        .with_minimum_log_level(LogLevel::Error)
        .build()?;
    ditto.disable_sync_with_v3().unwrap();
    Ok(ditto)
}

/// Create a Revamp-style Ditto instance with OnlinePlayground identity
fn create_revamp_ditto_playground() -> Result<Ditto, dittolive_ditto::error::DittoError> {
    let app_id = AppId::from_uuid(uuid::Uuid::parse_str(APP_ID).unwrap());

    let tempdir = tempfile::tempdir().expect("should create tempdir");
    // Note: For testing purposes, create playground with shared key approach
    // since new_with_playground_token is not available in public API
    let config = DittoConfig::new(
        app_id.to_string(),
        DittoConfigConnect::Server {
            url: format!("https://{}.cloud.ditto.live", app_id)
                .parse()
                .unwrap(),
        },
    )
    .with_persistence_directory(tempdir.keep());

    let ditto = Ditto::open_sync(config)?;
    ditto
        .auth()
        .unwrap()
        .set_expiration_handler(async |ditto: &Ditto, _duration_remaining| {
            let auth = ditto.auth().unwrap();
            auth.login(LOGIN_TOKEN, PROVIDER).unwrap();
        });
    ditto.disable_sync_with_v3().unwrap();
    Ok(ditto)
}

#[ignore = "Auth tests disabled by default"]
#[tokio::test]
async fn login_with_token_logs_in_given_a_valid_token() {
    // Test with Builder-style Ditto
    let (auth_required_tx, auth_required_rx) = mpsc::sync_channel::<()>(1);
    let auth_event_handler = AuthEventHandler::new(auth_required_tx);
    let ditto = create_builder_ditto_with_auth(auth_event_handler, None).unwrap();
    login_with_token_logs_in_given_a_valid_token_impl(ditto, auth_required_rx).await;

    // Test with Revamp-style Ditto
    let (auth_required_tx, auth_required_rx) = mpsc::sync_channel::<()>(1);
    let auth_event_handler = AuthEventHandler::new(auth_required_tx);
    let ditto = create_revamp_ditto_with_auth(auth_event_handler, None).unwrap();
    login_with_token_logs_in_given_a_valid_token_impl(ditto, auth_required_rx).await;
}

async fn login_with_token_logs_in_given_a_valid_token_impl(
    ditto: Ditto,
    auth_required_rx: mpsc::Receiver<()>,
) {
    assert!(!ditto.auth().unwrap().is_authenticated());
    assert!(ditto.auth().unwrap().user_id().is_none());

    auth_required_rx.recv().unwrap();

    ditto.auth().unwrap().login(LOGIN_TOKEN, PROVIDER).unwrap();

    assert!(ditto.auth().unwrap().is_authenticated());
    assert!(ditto.auth().unwrap().user_id().is_some());
}

#[ignore = "Auth tests disabled by default"]
#[tokio::test]
async fn login_with_token_fails_if_token_is_expired() {
    // Test with Builder-style Ditto
    let (auth_required_tx, auth_required_rx) = mpsc::sync_channel::<()>(1);
    let auth_event_handler = AuthEventHandler::new(auth_required_tx);
    let ditto = create_builder_ditto_with_auth(auth_event_handler, None).unwrap();
    login_with_token_fails_if_token_is_expired_impl(ditto, auth_required_rx).await;

    // Test with Revamp-style Ditto
    let (auth_required_tx, auth_required_rx) = mpsc::sync_channel::<()>(1);
    let auth_event_handler = AuthEventHandler::new(auth_required_tx);
    let ditto = create_revamp_ditto_with_auth(auth_event_handler, None).unwrap();
    login_with_token_fails_if_token_is_expired_impl(ditto, auth_required_rx).await;
}

async fn login_with_token_fails_if_token_is_expired_impl(
    ditto: Ditto,
    auth_required_rx: mpsc::Receiver<()>,
) {
    assert!(!ditto.auth().unwrap().is_authenticated());
    assert!(ditto.auth().unwrap().user_id().is_none());

    auth_required_rx.recv().unwrap();

    let res = ditto.auth().unwrap().login(EXPIRED_LOGIN_TOKEN, PROVIDER);
    assert!(res.is_err());

    assert!(!ditto.auth().unwrap().is_authenticated());
    assert!(ditto.auth().unwrap().user_id().is_none());
}

#[ignore = "Auth tests disabled by default"]
#[tokio::test]
async fn login_with_token_fails_if_token_is_invalid() {
    // Test with Builder-style Ditto
    let (auth_required_tx, auth_required_rx) = mpsc::sync_channel::<()>(1);
    let auth_event_handler = AuthEventHandler::new(auth_required_tx);
    let ditto = create_builder_ditto_with_auth(auth_event_handler, None).unwrap();
    login_with_token_fails_if_token_is_invalid_impl(ditto, auth_required_rx).await;

    // Test with Revamp-style Ditto
    let (auth_required_tx, auth_required_rx) = mpsc::sync_channel::<()>(1);
    let auth_event_handler = AuthEventHandler::new(auth_required_tx);
    let ditto = create_revamp_ditto_with_auth(auth_event_handler, None).unwrap();
    login_with_token_fails_if_token_is_invalid_impl(ditto, auth_required_rx).await;
}

async fn login_with_token_fails_if_token_is_invalid_impl(
    ditto: Ditto,
    auth_required_rx: mpsc::Receiver<()>,
) {
    assert!(!ditto.auth().unwrap().is_authenticated());
    assert!(ditto.auth().unwrap().user_id().is_none());

    auth_required_rx.recv().unwrap();

    let res = ditto
        .auth()
        .unwrap()
        .login("invalid_token_yada_yada", PROVIDER);
    assert!(res.is_err());

    assert!(!ditto.auth().unwrap().is_authenticated());
    assert!(ditto.auth().unwrap().user_id().is_none());
}

#[ignore = "Auth tests disabled by default"]
#[tokio::test]
async fn login_with_token_fails_if_provider_is_invalid() {
    // Test with Builder-style Ditto
    let (auth_required_tx, auth_required_rx) = mpsc::sync_channel::<()>(1);
    let auth_event_handler = AuthEventHandler::new(auth_required_tx);
    let ditto = create_builder_ditto_with_auth(auth_event_handler, None).unwrap();
    login_with_token_fails_if_provider_is_invalid_impl(ditto, auth_required_rx).await;

    // Test with Revamp-style Ditto
    let (auth_required_tx, auth_required_rx) = mpsc::sync_channel::<()>(1);
    let auth_event_handler = AuthEventHandler::new(auth_required_tx);
    let ditto = create_revamp_ditto_with_auth(auth_event_handler, None).unwrap();
    login_with_token_fails_if_provider_is_invalid_impl(ditto, auth_required_rx).await;
}

async fn login_with_token_fails_if_provider_is_invalid_impl(
    ditto: Ditto,
    auth_required_rx: mpsc::Receiver<()>,
) {
    assert!(!ditto.auth().unwrap().is_authenticated());
    assert!(ditto.auth().unwrap().user_id().is_none());

    auth_required_rx.recv().unwrap();

    let res = ditto.auth().unwrap().login(LOGIN_TOKEN, "invalid_provider");
    assert!(res.is_err());

    assert!(!ditto.auth().unwrap().is_authenticated());
    assert!(ditto.auth().unwrap().user_id().is_none());
}

#[ignore = "Auth tests disabled by default"]
#[tokio::test]
async fn logout() {
    // Test with Builder-style Ditto
    let (auth_required_tx, auth_required_rx) = mpsc::sync_channel::<()>(1);
    let auth_event_handler = AuthEventHandler::new(auth_required_tx);
    let ditto = create_builder_ditto_with_auth(auth_event_handler, None).unwrap();
    logout_impl(ditto, auth_required_rx).await;

    // Test with Revamp-style Ditto
    let (auth_required_tx, auth_required_rx) = mpsc::sync_channel::<()>(1);
    let auth_event_handler = AuthEventHandler::new(auth_required_tx);
    let ditto = create_revamp_ditto_with_auth(auth_event_handler, None).unwrap();
    logout_impl(ditto, auth_required_rx).await;
}

async fn logout_impl(ditto: Ditto, auth_required_rx: mpsc::Receiver<()>) {
    assert!(!ditto.auth().unwrap().is_authenticated());
    assert!(ditto.auth().unwrap().user_id().is_none());

    auth_required_rx.recv().unwrap();

    ditto.auth().unwrap().login(LOGIN_TOKEN, PROVIDER).unwrap();

    assert!(ditto.auth().unwrap().is_authenticated());
    assert!(ditto.auth().unwrap().user_id().is_some());

    let cleanup_block_times_called = Arc::new(AtomicU8::new(0));
    let cleanup_block_times_called_1 = cleanup_block_times_called.clone();
    ditto
        .auth()
        .unwrap()
        .logout(|_| {
            cleanup_block_times_called_1.fetch_add(1, Ordering::SeqCst);
        })
        .unwrap();
    assert_eq!(cleanup_block_times_called.load(Ordering::SeqCst), 1);

    assert!(!ditto.auth().unwrap().is_authenticated());
    assert!(ditto.auth().unwrap().user_id().is_none());
}

#[ignore = "Auth tests disabled by default"]
#[tokio::test]
async fn logout_calls_cleanup_block_even_if_not_logged_in() {
    // Test with Builder-style Ditto
    let (auth_required_tx, _auth_required_rx) = mpsc::sync_channel::<()>(1);
    let auth_event_handler = AuthEventHandler::new(auth_required_tx);
    let ditto = create_builder_ditto_with_auth(auth_event_handler, None).unwrap();
    logout_calls_cleanup_block_even_if_not_logged_in_impl(ditto).await;

    // Test with Revamp-style Ditto
    let (auth_required_tx, _auth_required_rx) = mpsc::sync_channel::<()>(1);
    let auth_event_handler = AuthEventHandler::new(auth_required_tx);
    let ditto = create_revamp_ditto_with_auth(auth_event_handler, None).unwrap();
    logout_calls_cleanup_block_even_if_not_logged_in_impl(ditto).await;
}

async fn logout_calls_cleanup_block_even_if_not_logged_in_impl(ditto: Ditto) {
    assert!(!ditto.auth().unwrap().is_authenticated());
    assert!(ditto.auth().unwrap().user_id().is_none());

    let cleanup_block_times_called = Arc::new(AtomicU8::new(0));
    let cleanup_block_times_called_1 = cleanup_block_times_called.clone();
    ditto
        .auth()
        .unwrap()
        .logout(|_| {
            cleanup_block_times_called_1.fetch_add(1, Ordering::SeqCst);
        })
        .unwrap();
    assert_eq!(cleanup_block_times_called.load(Ordering::SeqCst), 1);
}

#[ignore = "Auth tests disabled by default"]
#[tokio::test]
async fn status_is_authenticated_when_starting_an_instance_reusing_an_authenticated_directory() {
    // Test with Builder-style Ditto
    let ditto_dir = tempfile::tempdir().unwrap();
    let ditto_dir_path = ditto_dir.path().to_path_buf();
    status_is_authenticated_when_starting_an_instance_reusing_an_authenticated_directory_impl(
        true,
        ditto_dir_path.clone(),
    )
    .await;

    // Test with Revamp-style Ditto
    let ditto_dir2 = tempfile::tempdir().unwrap();
    let ditto_dir_path2 = ditto_dir2.path().to_path_buf();
    status_is_authenticated_when_starting_an_instance_reusing_an_authenticated_directory_impl(
        false,
        ditto_dir_path2,
    )
    .await;
}

async fn status_is_authenticated_when_starting_an_instance_reusing_an_authenticated_directory_impl(
    use_builder: bool,
    ditto_dir_path: std::path::PathBuf,
) {
    // First scope: Create Ditto, authenticate, and persist some data
    {
        let (auth_required_tx, auth_required_rx) = mpsc::sync_channel::<()>(1);
        let auth_event_handler = AuthEventHandler::new(auth_required_tx);

        let ditto = if use_builder {
            create_builder_ditto_with_auth(
                auth_event_handler,
                Some(Arc::new(PersistentRoot::new(&ditto_dir_path).unwrap())),
            )
            .unwrap()
        } else {
            create_revamp_ditto_with_auth(auth_event_handler, Some(ditto_dir_path.clone())).unwrap()
        };

        assert!(!ditto.auth().unwrap().is_authenticated());
        assert!(ditto.auth().unwrap().user_id().is_none());

        auth_required_rx.recv().unwrap();

        ditto.auth().unwrap().login(LOGIN_TOKEN, PROVIDER).unwrap();

        assert!(ditto.auth().unwrap().is_authenticated());
        assert!(ditto.auth().unwrap().user_id().is_some());
    }

    // Second scope: Create new Ditto instance with same directory
    let (auth_required_tx, _) = mpsc::sync_channel::<()>(1);
    let auth_event_handler = AuthEventHandler::new(auth_required_tx);

    let ditto = if use_builder {
        create_builder_ditto_with_auth(
            auth_event_handler.clone(),
            Some(Arc::new(PersistentRoot::new(&ditto_dir_path).unwrap())),
        )
        .unwrap()
    } else {
        create_revamp_ditto_with_auth(auth_event_handler.clone(), Some(ditto_dir_path)).unwrap()
    };

    assert!(ditto.auth().unwrap().is_authenticated());
    assert!(ditto.auth().unwrap().user_id().is_some());
    assert_eq!(
        auth_event_handler
            .authentication_required_times_called
            .load(Ordering::SeqCst),
        0
    );
}

#[ignore = "Auth tests disabled by default"]
#[tokio::test]
async fn online_playground_authenticates_and_syncs() {
    // Test with Builder-style Ditto
    let ditto1 = create_builder_ditto_playground().unwrap();
    let ditto2 = create_builder_ditto_playground().unwrap();
    online_playground_authenticates_and_syncs_impl(ditto1, ditto2).await;

    // Test with Revamp-style Ditto
    let ditto1 = create_revamp_ditto_playground().unwrap();
    let ditto2 = create_revamp_ditto_playground().unwrap();
    online_playground_authenticates_and_syncs_impl(ditto1, ditto2).await;
}

async fn online_playground_authenticates_and_syncs_impl(ditto1: Ditto, ditto2: Ditto) {
    let unique_collection_name = uuid::Uuid::new_v4().to_string().replace('-', "_");
    let unique_collection_name = format!("collection_{}", unique_collection_name);

    ditto1.sync().start().unwrap();
    ditto2.sync().start().unwrap();

    // Wait for authentication
    let now = std::time::Instant::now();
    while now.elapsed() < std::time::Duration::from_secs(10) {
        if ditto1.auth().unwrap().is_authenticated() && ditto2.auth().unwrap().is_authenticated() {
            break;
        }
        std::thread::sleep(std::time::Duration::from_millis(100));
    }

    assert!(
        ditto1.auth().unwrap().is_authenticated(),
        "Ditto instance 1 failed to authenticate"
    );
    assert!(
        ditto2.auth().unwrap().is_authenticated(),
        "Ditto instance 2 failed to authenticate"
    );

    // Subscribe to changes on both instances
    let subscription1 = ditto1
        .sync()
        .register_subscription_v2((
            &format!("SELECT * FROM {}", unique_collection_name),
            serde_json::json!({}),
        ))
        .unwrap();
    let subscription2 = ditto2
        .sync()
        .register_subscription_v2((
            &format!("SELECT * FROM {}", unique_collection_name),
            serde_json::json!({}),
        ))
        .unwrap();

    let yellow_car_id = "test_car_123";

    // Verify initial state - no documents
    let initial_query = ditto1
        .store()
        .execute_v2((
            &format!("SELECT * FROM {} WHERE _id = :id", unique_collection_name),
            serde_json::json!({"id": yellow_car_id}),
        ))
        .await
        .unwrap();
    assert_eq!(initial_query.item_count(), 0);

    // Insert document from ditto1
    let insert_result = ditto1
        .store()
        .execute_v2((
            &format!("INSERT INTO {} DOCUMENTS (:doc)", unique_collection_name),
            serde_json::json!({"doc": {"_id": yellow_car_id, "color": "yellow", "type": "sync_test"}}),
        ))
        .await
        .unwrap();

    assert_eq!(insert_result.mutated_document_ids().len(), 1);

    // Verify document exists on ditto1
    let found_docs = ditto1
        .store()
        .execute_v2((
            &format!("SELECT * FROM {} WHERE _id = :id", unique_collection_name),
            serde_json::json!({"id": yellow_car_id}),
        ))
        .await
        .unwrap();
    assert_eq!(found_docs.item_count(), 1);
    let doc = found_docs.get_item(0).unwrap();
    assert_eq!(
        doc.value().get("color").unwrap().as_str().unwrap(),
        "yellow"
    );

    let (sync_tx, sync_rx) = mpsc::sync_channel::<bool>(1);
    let _observer = ditto2
        .store()
        .register_observer_v2(
            (
                &format!("SELECT * FROM {} WHERE _id = :id", unique_collection_name),
                serde_json::json!({"id": yellow_car_id}),
            ),
            move |result| {
                if result.item_count() > 0 {
                    let synced_doc = result.get_item(0).unwrap();
                    if synced_doc.value().get("color").unwrap().as_str().unwrap() == "yellow" {
                        let _ = sync_tx.send(true);
                    }
                }
            },
        )
        .unwrap();

    // Wait up to 10 seconds for sync
    sync_rx.recv_timeout(Duration::from_secs(10)).unwrap();

    drop(subscription1);
    drop(subscription2);
    drop(_observer);
}

#[derive(Clone)]
pub struct AuthEventHandler {
    /// Count of how many times `authentication_required` has been called
    pub authentication_required_times_called: Arc<AtomicU8>,

    /// Count of how many times `authentication_expiring_soon` has been called
    pub authentication_expiring_soon_times_called: Arc<AtomicU8>,

    /// A sync_channel sender to send a message to when there has been a call to
    /// `authentication_required`
    auth_required_tx: Arc<mpsc::SyncSender<()>>,
}

impl AuthEventHandler {
    pub fn new(auth_required_tx: mpsc::SyncSender<()>) -> Self {
        Self {
            authentication_required_times_called: Arc::new(AtomicU8::new(0)),
            authentication_expiring_soon_times_called: Arc::new(AtomicU8::new(0)),
            auth_required_tx: Arc::new(auth_required_tx),
        }
    }
}

impl DittoAuthenticationEventHandler for AuthEventHandler {
    fn authentication_required(&self, _auth: DittoAuthenticator) {
        self.authentication_required_times_called
            .fetch_add(1, Ordering::SeqCst);
        self.auth_required_tx.send(()).unwrap();
    }

    fn authentication_expiring_soon(
        &self,
        _auth: DittoAuthenticator,
        _seconds_remaining: Duration,
    ) {
        self.authentication_expiring_soon_times_called
            .fetch_add(1, Ordering::SeqCst);
    }
}
