1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147

//! Use [`ditto.sync()`] to access the [`Sync`] API of Ditto.
//!
//! The [`Sync`] API can be used to request documents to be synchronized with
//! other peers. Use [`ditto.sync().register_subscription(...)`] and provide
//! a DQL query to let Ditto know which documents you're interested in syncing.
//!
//! The returned [`SyncSubscription`] handle can be used to cancel the
//! subscription with [`.cancel()`], at which point Ditto will stop syncing
//! data for that subscription.
//!
//! # Example
//!
//! ```
//! use dittolive_ditto::prelude::*;
//! # fn example(ditto: &Ditto) -> anyhow::Result<()> {
//!
//! let sync_subscription = ditto
//!     .sync()
//!     .register_subscription("SELECT * FROM cars WHERE color = 'blue'", None)?;
//!
//! // To cancel the sync subscription, use .cancel()
//! sync_subscription.cancel();
//! # Ok(())
//! # }
//! ```
//!
//! [`ditto.sync()`]: Ditto::sync
//! [`ditto.sync().register_subscription(...)`]: Sync::register_subscription
//! [`.cancel()`]: SyncSubscription::cancel

use std::{
    ops::Deref,
    sync::{Arc, RwLock, Weak},
};

use tracing::{debug, error};

use crate::{
    ditto::{Ditto, DittoFields},
    dql::*,
    error::DittoError,
    utils::{extension_traits::FfiResultIntoRustResult, SetArc},
};

mod sync_subscription;
pub use self::sync_subscription::SyncSubscription;

/// Ditto's `Sync` API, obtained via [`ditto.sync()`].
///
/// [See the `sync` module documentation for more details][0].
///
/// [`ditto.sync()`]: crate::prelude::Ditto::sync
/// [0]: crate::sync
pub struct Sync {
    ditto: Weak<DittoFields>,
    subscriptions: RwLock<SetArc<SyncSubscription>>,
}

impl Sync {
    pub(crate) fn new(ditto: Weak<DittoFields>) -> Self {
        Self {
            ditto,
            subscriptions: RwLock::new(SetArc::default()),
        }
    }

    /// Gets temporary access to the set of currently registered subscriptions.
    ///
    /// A (read) lock is held until the return value is dropped: this means
    /// that neither [`Self::register_subscription()`] nor
    /// [`SyncSubscription::cancel()`] can make progress until this read
    /// lock is released.
    pub fn subscriptions(&self) -> impl '_ + Deref<Target = SetArc<SyncSubscription>> {
        self.subscriptions.read().unwrap()
    }

    /// Use a DQL query to subscribe to data on other Ditto peers.
    ///
    /// While the subscription is active, data matching this query on other
    /// peers will be synced to the local peer's data store.
    ///
    /// Note that dropping the `SyncSubscription` won't cancel it, to do that
    /// be sure to use [`sync_subscription.cancel()`].
    ///
    /// # Example
    ///
    /// ```
    /// use dittolive_ditto::prelude::*;
    /// # fn example(ditto: &Ditto) -> anyhow::Result<()> {
    /// let sync_subscription = ditto.sync().register_subscription(
    ///     "SELECT * FROM cars WHERE color = :color",
    ///     Some(serde_json::json!({
    ///         "color": "blue"
    ///     }).into()),
    /// )?;
    ///
    /// // Cancel the subscription with `.cancel()`
    /// sync_subscription.cancel();
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// [`sync_subscription.cancel()`]: crate::sync::SyncSubscription::cancel
    pub fn register_subscription<Q>(
        &self,
        query: Q,
        query_args: Option<QueryArguments>,
    ) -> Result<Arc<SyncSubscription>, DittoError>
    where
        Q: TryInto<Query, Error = DittoError>,
    {
        let ditto = Ditto::upgrade(&self.ditto)?;
        let new_sub = Arc::new(SyncSubscription::new(
            &ditto,
            query.try_into()?,
            query_args,
        )?);
        self.subscriptions.write().unwrap().insert(new_sub.clone());
        Ok(new_sub)
    }

    pub(crate) fn unregister_subscription(&self, subscription: &SyncSubscription) -> bool {
        let subscriptions = &mut *self.subscriptions.write().unwrap();
        let removed = subscriptions.remove(subscription);
        if removed {
            debug!(%subscription, "unregistering sync subscription");
            if let Ok(ditto) = Ditto::upgrade(&self.ditto) {
                if let Err(error) = ffi_sdk::dittoffi_try_remove_sync_subscription(
                    &ditto.ditto,
                    subscription.query.prepare_ffi(),
                    subscription.query_args.as_ref().map(|qa| qa.cbor().into()),
                )
                .into_rust_result()
                {
                    let ditto_error = DittoError::from(error);
                    error!(
                        error = %ditto_error,
                        %subscription,
                        "failed to unregister sync subscription",
                    );
                }
                return true;
            }
        }
        false
    }
}