dittolive_ditto/store/
mod.rs

1//! Use [`ditto.store()`] to access the [`Store`] API to read, write, and remove documents.
2//!
3//! The `Store` provides two interfaces for interacting with data: Ditto Query Language
4//! (DQL), and the legacy "Query Builder" API. Where possible, we recommend developing new
5//! functionality using DQL, as we will eventually deprecate Query Builder.
6//!
7//! - [See the `dql` module docs for examples of DQL queries in action][0]
8//!
9//! [`ditto.store()`]: crate::prelude::Ditto::store
10//! [0]: crate::dql
11
12use_prelude!();
13
14use std::{
15    ops::Deref,
16    sync::{
17        atomic::{self, AtomicU64},
18        RwLock, Weak,
19    },
20};
21
22use ffi_sdk::FfiStoreObserver;
23
24use crate::{debug, error};
25
26pub mod attachment;
27mod document_id;
28mod observer;
29pub mod transactions;
30
31use self::attachment::{DittoAttachmentFetcher, FetcherVersion};
32pub use self::{
33    document_id::DocumentId,
34    observer::{ChangeHandler, ChangeHandlerWithSignalNext, SignalNext, StoreObserver},
35};
36use crate::{
37    ditto::DittoFields,
38    dql::{query::IntoQuery, *},
39    error::{DittoError, ErrorKind},
40    utils::{extension_traits::FfiResultIntoRustResult, SetArc},
41};
42
43type CancelToken = u64;
44
45/// Use [`ditto.store()`] to access the [`Store`] API to read, write, and remove documents.
46///
47/// [See the `store` module for guide-level docs and examples][0].
48///
49/// [`ditto.store()`]: crate::prelude::Ditto::store
50/// [0]: crate::store
51#[derive(Clone)]
52pub struct Store {
53    ditto: Arc<ffi_sdk::BoxedDitto>,
54    // FIXME(Daniel): unify this field with `.ditto`
55    weak_ditto_fields: Weak<DittoFields>,
56    #[allow(clippy::type_complexity)]
57    attachment_fetchers: Arc<
58        RwLock<HashMap<CancelToken, (bool, DittoAttachmentFetcher<'static, FetcherVersion::V2>)>>,
59    >,
60}
61
62impl Store {
63    pub(crate) fn new(
64        ditto: Arc<ffi_sdk::BoxedDitto>,
65        weak_ditto_fields: Weak<DittoFields>,
66    ) -> Self {
67        Self {
68            ditto,
69            weak_ditto_fields,
70            attachment_fetchers: <_>::default(),
71        }
72    }
73
74    /// Installs and returns a store observer for a query, configuring Ditto to
75    /// trigger the passed in change handler whenever documents in the local
76    /// store change such that the result of the matching query changes. The
77    /// passed in query must be a `SELECT` query, otherwise an error will be
78    /// returned.
79    ///
80    /// # Example
81    ///
82    /// ```
83    /// use dittolive_ditto::prelude::*;
84    /// # fn main() -> anyhow::Result<()> {
85    /// # let (_root, ditto) = dittolive_ditto::doctest_helpers::doctest_ditto();
86    ///
87    /// let store = ditto.store();
88    /// let _observer = store.register_observer(
89    ///     "SELECT * FROM cars WHERE color = 'blue'",
90    ///     move |query_result| {
91    ///         for item in query_result.iter() {
92    ///             // ... handle each item
93    ///         }
94    ///     },
95    /// )?;
96    /// # Ok(())
97    /// # }
98    /// ```
99    ///
100    /// The first invocation of the change handler will always happen after
101    /// this method has returned.
102    ///
103    /// The observer will remain active until:
104    ///
105    /// - the [`StoreObserver`] handle gets dropped,
106    /// - the [`observer.cancel()`] method is called, or
107    /// - the owning [`Ditto`] instance has shut down
108    ///
109    /// Observer callbacks will never be called concurrently. That is, one callback
110    /// must return before the observer will call the handler again. See
111    /// [`ditto.store().register_observer_with_signal_next(...)`] if you want
112    /// to manually signal readiness for the next callback.
113    ///
114    /// [`ditto.store().register_observer_with_signal_next(...)`]: crate::store::Store::register_observer_with_signal_next
115    /// [`observer.cancel()`]: crate::store::StoreObserver::cancel
116    /// [`Ditto`]: crate::prelude::Ditto
117    pub fn register_observer<Q, F>(
118        &self,
119        query: Q,
120        on_change: F,
121    ) -> Result<Arc<StoreObserver>, DittoError>
122    where
123        Q: IntoQuery,
124        Q::Args: Serialize,
125        F: ChangeHandler,
126    {
127        let ditto = Ditto::upgrade(&self.weak_ditto_fields)?;
128        let query = query.into_query()?;
129
130        let observer = Arc::new(StoreObserver::new(
131            &ditto,
132            &query.string,
133            query.args_cbor.as_deref(),
134            on_change,
135        )?);
136        Ok(observer)
137    }
138
139    /// Installs and returns a store observer for a query, configuring Ditto to
140    /// trigger the passed in change handler whenever documents in the local
141    /// store change such that the result of the matching query changes. The
142    /// passed in query must be a `SELECT` query, otherwise an error will be
143    /// returned.
144    ///
145    /// Here, a function is passed as an additional argument to the change
146    /// handler. Call it when the handler is ready for another callback. See
147    /// [`register_observer`] for a convenience method that automatically
148    /// signals the next invocation.
149    ///
150    /// # Example
151    ///
152    /// ```
153    /// use dittolive_ditto::prelude::*;
154    /// # fn main() -> anyhow::Result<()> {
155    /// # let (_root, ditto) = dittolive_ditto::doctest_helpers::doctest_ditto();
156    ///
157    /// let store = ditto.store();
158    /// let _observer = store.register_observer_with_signal_next(
159    ///     "SELECT * FROM cars WHERE color = 'blue'",
160    ///     move |query_result, signal_next| {
161    ///         for item in query_result.iter() {
162    ///             // ... handle each item
163    ///         }
164    ///
165    ///         // Call `signal_next` when you're ready for the next callback
166    ///         signal_next();
167    ///     },
168    /// )?;
169    /// # Ok(())
170    /// # }
171    /// ```
172    ///
173    /// The first invocation of the change handler will always happen after
174    /// this method has returned.
175    ///
176    /// The observer will remain active until:
177    ///
178    /// - the [`StoreObserver`] handle gets dropped,
179    /// - the [`StoreObserver::cancel`] method is called, or
180    /// - the owning [`Ditto`] instance has shut down
181    ///
182    /// After invoking the callback once, the observer will wait to deliver
183    /// another callback until after you've called [`signal_next`]. See [`SignalNext`] for
184    /// how Ditto handles backpressure and pending updates while callback delivery is paused.
185    ///
186    /// [`register_observer`]: Store::register_observer
187    /// [`signal_next`]: crate::store::SignalNext
188    pub fn register_observer_with_signal_next<Q, F>(
189        &self,
190        query: Q,
191        on_change: F,
192    ) -> Result<Arc<StoreObserver>, DittoError>
193    where
194        Q: IntoQuery,
195        Q::Args: Serialize,
196        F: ChangeHandlerWithSignalNext,
197    {
198        let ditto = Ditto::upgrade(&self.weak_ditto_fields)?;
199        let query = query.into_query()?;
200
201        let new_obs = Arc::new(StoreObserver::with_signal_next(
202            &ditto,
203            &query.string,
204            query.args_cbor.as_deref(),
205            on_change,
206        )?);
207        Ok(new_obs)
208    }
209
210    /// Gets temporary access to the set of currently registered observers.
211    ///
212    /// A (read) lock is held until the return value is dropped: this means
213    /// that neither [`Self::register_observer()`] nor
214    /// [`StoreObserver::cancel()`] can make progress until this read
215    /// lock is released.
216    pub fn observers(&self) -> impl '_ + Deref<Target = SetArc<StoreObserver>> {
217        let observers: repr_c::Vec<repr_c::Box<FfiStoreObserver>> =
218            ffi_sdk::dittoffi_store_observers(&self.ditto);
219        let observers: Vec<_> = observers.into();
220        let observers = observers
221            .into_iter()
222            .map(|handle: repr_c::Box<FfiStoreObserver>| Arc::new(StoreObserver { handle }))
223            .collect::<SetArc<_>>();
224
225        Box::new(observers)
226    }
227
228    /// Executes the given query in the local store and returns the result.
229    ///
230    /// # Example
231    ///
232    /// ```
233    /// use dittolive_ditto::prelude::*;
234    /// use dittolive_ditto::dql::QueryResult;
235    /// # #[tokio::main]
236    /// # async fn main() -> anyhow::Result<()> {
237    /// # let (_root, ditto) = dittolive_ditto::doctest_helpers::doctest_ditto();
238    ///
239    /// // Query a collection
240    /// let result = ditto.store().execute("SELECT * FROM cars").await?;
241    ///
242    /// // Insert a document into a collection
243    /// let insert_result: QueryResult = ditto
244    ///     .store()
245    ///     .execute((
246    ///          "INSERT INTO cars DOCUMENTS (:newCar)",
247    ///          serde_json::json!({
248    ///              "newCar": {
249    ///                  "make": "ford",
250    ///                  "color": "blue"
251    ///              }
252    ///          })
253    ///     ))
254    ///     .await?;
255    /// # Ok(())
256    /// # }
257    /// ```
258    ///
259    /// Use placeholders to incorporate values from the optional `query_args`
260    /// parameter into the query. The keys of the query arguments object must
261    /// match the placeholders used within the query. You can not use placeholders
262    /// in the `FROM` clause.
263    ///
264    /// This method only returns results from the local store without waiting for any
265    /// [`SyncSubscription`]s to have caught up with the
266    /// latest changes. Only use this method if your program must proceed with immediate results.
267    ///
268    /// Use [`ditto.store().register_observer(...)`] to receive updates to query results
269    /// as soon as they have been synced to this peer.
270    ///
271    /// ## Query parameter
272    ///
273    /// The `query` parameter must implement [`IntoQuery`], which is a trait that is implemented by
274    /// objects that can be turned into a query string along with the relevant query arguments.
275    ///
276    /// For queries with no arguments, a [`String`] is sufficient.
277    ///
278    /// [`SyncSubscription`]: crate::sync::SyncSubscription
279    /// [`ditto.store().register_observer(...)`]: crate::store::Store::register_observer
280    pub async fn execute<Q>(&self, query: Q) -> Result<QueryResult, DittoError>
281    where
282        Q: IntoQuery,
283        Q::Args: serde::Serialize,
284    {
285        let query = query.into_query()?;
286        let query_string = (&*query.string).into();
287        let query_args = query.args_cbor.as_deref().map(Into::into);
288
289        let ffi_query_result =
290            ffi_sdk::dittoffi_try_exec_statement(&self.ditto, query_string, query_args)
291                .into_rust_result()?;
292
293        Ok(QueryResult::from(ffi_query_result))
294    }
295
296    /// Creates a new attachment, which can then be inserted into a document.
297    ///
298    /// The file residing at the provided path will be copied into Ditto’s store. The
299    /// [`DittoAttachment`] object that is returned is what you can
300    /// then use to insert an attachment into a document.
301    ///
302    /// You can provide custom user data about the attachment, which will be replicated to other
303    /// peers alongside the file attachment.
304    pub async fn new_attachment(
305        &self,
306        filepath: &(impl ?Sized + AsRef<Path>),
307        user_data: HashMap<String, String>,
308    ) -> Result<DittoAttachment, DittoError> {
309        DittoAttachment::from_file_and_metadata(filepath, user_data, &self.ditto)
310    }
311
312    /// Creates a new attachment from in-memory data
313    ///
314    /// Refer to [`new_attachment`](Self::new_attachment) for additional information.
315    pub async fn new_attachment_from_bytes(
316        &self,
317        bytes: &(impl ?Sized + AsRef<[u8]>),
318        user_data: HashMap<String, String>,
319    ) -> Result<DittoAttachment, DittoError> {
320        DittoAttachment::from_bytes_and_metadata(bytes, user_data, &self.ditto)
321    }
322
323    /// Fetches the attachment corresponding to the provided attachment token.
324    /// - `attachment_token`: can be either a [`DittoAttachmentToken`], or a `&BTreeMap<CborValue,
325    ///   CborValue>`, that is, the output of a [`QueryResultItem::value()`] once casted
326    ///   [`.as_object()`][crate::prelude::CborValueGetters::as_object()].
327    ///
328    /// - `on_fetch_event`: A closure that will be called when the status of the request to fetch
329    ///   the attachment has changed. If the attachment is already available then this will be
330    ///   called almost immediately with a completed status value.
331    ///
332    /// The returned [`DittoAttachmentFetcher`] is a handle which is safe to discard, unless you
333    /// wish to be able to [`.cancel()`][DittoAttachmentFetcher::cancel] the fetching operation.
334    /// When not explicitly cancelled, the fetching operation will remain active until it either
335    /// completes, the attachment is deleted, or the owning [`Ditto`] object is dropped.
336    pub fn fetch_attachment(
337        &self,
338        attachment_token: impl attachment::DittoAttachmentTokenLike,
339        on_fetch_event: impl 'static + Send + Sync + Fn(DittoAttachmentFetchEvent),
340    ) -> Result<DittoAttachmentFetcher<'static, FetcherVersion::V2>, DittoError> {
341        let attachment_token = attachment_token.parse_attachment_token()?;
342
343        let weak_ditto = self.weak_ditto_fields.clone();
344        let ditto = weak_ditto
345            .upgrade()
346            .ok_or(ErrorKind::ReleasedDittoInstance)?;
347
348        let mut attachment_fetchers_lockguard = self.attachment_fetchers.write().unwrap();
349        let fetcher = DittoAttachmentFetcher::new(
350            attachment_token,
351            Some(&ditto),
352            &self.ditto,
353            // Shim around `on_fetch_event` to `cancel` on completion.
354            move |event, cancel_token: &AtomicU64| {
355                let has_finished = matches! {
356                    event,
357                    | DittoAttachmentFetchEvent::Completed { .. }
358                    | DittoAttachmentFetchEvent::Deleted
359                };
360                on_fetch_event(event);
361                if has_finished {
362                    if let Some(ditto) = weak_ditto.upgrade() {
363                        let mut attachment_fetchers_inner_lockguard =
364                            ditto.store.attachment_fetchers.write().unwrap();
365                        // Relaxed is fine thanks to the lock.
366                        let cancel_token = cancel_token.load(atomic::Ordering::Relaxed);
367                        ditto.store.unregister_fetcher(
368                            cancel_token,
369                            Some(&mut *attachment_fetchers_inner_lockguard),
370                        );
371                    }
372                }
373            },
374        )?;
375        let (cancel_token, was_zero) = fetcher.cancel_token_ensure_unique();
376        attachment_fetchers_lockguard.insert(cancel_token, (was_zero, fetcher.clone()));
377        Ok(fetcher)
378    }
379
380    fn unregister_fetcher(
381        &self,
382        mut fetcher_cancel_token: CancelToken,
383        fetchers: Option<
384            &mut HashMap<CancelToken, (bool, DittoAttachmentFetcher<'static, FetcherVersion::V2>)>,
385        >,
386    ) -> bool {
387        let mut lock_guard = None;
388        let fetchers = fetchers.unwrap_or_else(|| {
389            &mut **lock_guard.get_or_insert(self.attachment_fetchers.write().unwrap())
390        });
391
392        let Some((was_zero, removed_fetcher)) = fetchers.remove(&fetcher_cancel_token) else {
393            return false;
394        };
395        drop(lock_guard);
396
397        if was_zero {
398            fetcher_cancel_token = 0;
399        }
400
401        let att_token = &removed_fetcher.context.token;
402
403        #[allow(deprecated)] // Workaround for patched tracing
404        {
405            debug!(
406                token_id = %att_token.id(),
407                %fetcher_cancel_token,
408                "unregistering ditto attachment fetcher"
409            );
410        }
411
412        let status = ffi_sdk::ditto_cancel_resolve_attachment(
413            &self.ditto,
414            att_token.id.as_ref().into(),
415            fetcher_cancel_token,
416        );
417
418        if status != 0 {
419            #[allow(deprecated)] // Workaround for patched tracing
420            {
421                error!(
422                    token_id = %att_token.id(),
423                    %fetcher_cancel_token,
424                    "failed to clean up attachment fetcher"
425                );
426            }
427        }
428        status == 0
429    }
430
431    /// Gets a copy of the set of currently registered attachment fetchers.
432    ///
433    /// A (read) lock is held during the copy: this contends with [`Self::fetch_attachment()`] and
434    /// with [`DittoAttachmentFetcher::cancel()`].
435    pub fn attachment_fetchers(&self) -> Vec<DittoAttachmentFetcher<'static, FetcherVersion::V2>> {
436        self.attachment_fetchers
437            .read()
438            .unwrap()
439            .iter()
440            .map(|(_, (_, fetcher))| fetcher.clone())
441            .collect()
442    }
443}
444
445/// Specify the order of returned Documents in a query.
446#[deprecated(since = "5.1.0", note = "Use DQL `ORDER BY` clauses instead.")]
447#[non_exhaustive]
448#[derive(Clone, Copy, PartialEq, Eq, Debug)]
449pub enum SortDirection {
450    /// First result is "smallest", last result is "largest"
451    Ascending,
452
453    /// First result is "largest", last result is "smallest"
454    Descending,
455}
dittolive_ditto/store/mod.rs

dittolive_ditto/store/
mod.rs
