dittolive_ditto/store/

mod.rs

//! Use [`ditto.store()`] to access the [`Store`] API to read, write, and remove documents.
//!
//! The `Store` provides two interfaces for interacting with data: Ditto Query Language
//! (DQL), and the legacy "Query Builder" API. Where possible, we recommend developing new
//! functionality using DQL, as we will eventually deprecate Query Builder.
//!
//! - [See the `dql` module docs for examples of DQL queries in action][0]
//!
//! [`ditto.store()`]: crate::prelude::Ditto::store
//! [0]: crate::dql

use_prelude!();

use std::{
    ops::Deref,
    sync::{
        atomic::{self, AtomicU64},
        RwLock, Weak,
    },
};

use ffi_sdk::{FfiStoreObserver, FsComponent, WriteStrategyRs};

pub mod attachment;
mod document_id;
mod observer;
pub mod transactions;
pub use document_id::DocumentId;

#[deprecated(note = "Use `ditto.store().execute_v2(...)` or \
                     `ditto.store().register_observer_v2(...)` instead")]
#[allow(deprecated)] // Silence warnings within the module itself
pub mod query_builder;
pub use self::observer::{ChangeHandler, ChangeHandlerWithSignalNext, SignalNext, StoreObserver};
#[doc(hidden)]
#[allow(deprecated)]
pub use self::query_builder::{batch, collection, collections, live_query, update};
#[doc(hidden)]
pub use crate::dql;
use crate::{
    disk_usage::DiskUsage,
    ditto::DittoFields,
    error::{DittoError, ErrorKind},
    utils::{extension_traits::FfiResultIntoRustResult, SetArc},
};

// Legacy
#[doc(hidden)]
#[deprecated]
#[rustfmt::skip]
pub use self::attachment::{
    self as ditto_attachment,
    fetch_event as ditto_attachment_fetch_event,
    fetcher as ditto_attachment_fetcher,
    token as ditto_attachment_token,
};
use self::{ditto_attachment_fetcher::DittoAttachmentFetcherV2, dql::query_v2::IntoQuery};

#[doc(hidden)]
#[deprecated(note = "The experimental `timeseries` feature has been deprecated")]
#[cfg(feature = "timeseries")]
pub mod timeseries;

#[doc(hidden)]
#[allow(deprecated)]
use collections::pending_collections_operation::PendingCollectionsOperation;

use crate::dql::*;

type CancelToken = u64;

/// Use [`ditto.store()`] to access the [`Store`] API to read, write, and remove documents.
///
/// [See the `store` module for guide-level docs and examples][0].
///
/// [`ditto.store()`]: crate::prelude::Ditto::store
/// [0]: crate::store
#[derive(Clone)]
pub struct Store {
    ditto: Arc<ffi_sdk::BoxedDitto>,
    // FIXME(Daniel): unify this field with `.ditto`
    weak_ditto_fields: Weak<DittoFields>,
    disk_usage: Arc<DiskUsage>,
    attachment_fetchers: Arc<RwLock<HashMap<CancelToken, (bool, DittoAttachmentFetcherV2)>>>,
}

impl Store {
    pub(crate) fn new(
        ditto: Arc<ffi_sdk::BoxedDitto>,
        weak_ditto_fields: Weak<DittoFields>,
    ) -> Self {
        let disk_usage = Arc::new(DiskUsage::new(ditto.retain(), FsComponent::Store));
        Self {
            ditto,
            weak_ditto_fields,
            disk_usage,
            attachment_fetchers: <_>::default(),
        }
    }

    // Note this method's logic will be moved into the core ditto library
    // in the future
    fn validate_collection_name(name: &str) -> Result<(), DittoError> {
        let mut result = Ok(());

        if name.is_empty() {
            result = Err(DittoError::new(
                ErrorKind::InvalidInput,
                String::from("Collection name can not be empty"),
            ));
        }

        if name.split_whitespace().next().is_none() {
            result = Err(DittoError::new(
                ErrorKind::InvalidInput,
                String::from("Collection name can not only contain whitespace"),
            ));
        }

        result
    }

    /// Returns a [`Collection`] with the provided name.
    /// A collection name is valid if :
    /// * its length is less than 100
    /// * it is not empty
    /// * it does not contain the char '\0'
    /// * it does not begin with "$TS_"
    #[doc(hidden)]
    #[deprecated(note = "Use `ditto.store().execute_v2(...)` or \
                         `ditto.store().register_observer_v2(...)` instead")]
    #[allow(deprecated)]
    pub fn collection(&self, collection_name: &'_ str) -> Result<Collection, DittoError> {
        Self::validate_collection_name(collection_name)?;
        let c_name = char_p::new(collection_name);
        let status = { ffi_sdk::ditto_collection(&*self.ditto, c_name.as_ref()) };
        if status != 0 {
            return Err(DittoError::from_ffi(ErrorKind::InvalidInput));
        }
        Ok(Collection {
            ditto: Arc::downgrade(&self.ditto),
            collection_name: c_name,
        })
    }

    /// Returns an object that lets you fetch or observe the collections in the
    /// store.
    #[doc(hidden)]
    #[deprecated]
    #[allow(deprecated)]
    pub fn collections(&self) -> PendingCollectionsOperation<'_> {
        PendingCollectionsOperation::<'_>::new(Arc::downgrade(&self.ditto))
    }

    /// Allows you to group multiple operations together that affect multiple
    /// documents, potentially across multiple collections, without
    /// auto-committing on each operation.
    ///
    /// At the end of the batch of operations, either
    /// [`batch.commit_changes`](crate::store::batch::ScopedStore::commit_changes)
    /// or
    /// [`batch.revert_changes`](crate::store::batch::ScopedStore::revert_changes)
    /// must be called.
    ///
    /// ## Example
    ///
    /// ```rust
    /// # macro_rules! ignore {($($__:tt)*) => ()} ignore! {
    /// ditto.store().with_batched_write(|batch| {
    ///     let mut foo_coll = batch.collection("foo");
    ///     foo_coll.find...().remove();
    ///     let mut bar_coll = batch.collection("bar");
    ///     // Expensive multi-mutation op:
    ///     for _ in 0 .. 10_000 {
    ///         let doc = ...;
    ///         bar_coll.insert(doc, None, false);
    ///     }
    ///     // At this point, we must say whether we commit or revert
    ///     // these changes:
    ///     batch.commit_changes()
    /// })
    /// # }
    /// ```
    #[doc(hidden)]
    #[deprecated]
    #[allow(deprecated)]
    pub fn with_batched_write<F>(
        &self,
        f: F,
    ) -> Result<Vec<batch::WriteTransactionResult>, DittoError>
    where
        for<'batch> F: FnOnce(batch::ScopedStore<'batch>) -> batch::Action<'batch>,
    {
        query_builder::batch::with_batched_write(self, f)
    }

    /// Returns a list of the names of collections in the local store.
    pub fn collection_names(&self) -> Result<Vec<String>, DittoError> {
        let c_collections = { ffi_sdk::ditto_get_collection_names(&*self.ditto).ok()? };

        Ok(c_collections
            .iter()
            .map(|x: &char_p::Box| -> String { x.clone().into_string() })
            .collect())
    }

    /// Returns a hash representing the current version of the given queries.
    /// When a document matching such queries gets mutated, the hash will change
    /// as well.
    ///
    /// Please note that the hash depends on how queries are constructed, so you
    /// should make sure to always compare hashes generated with the same set of
    /// queries.
    #[doc(hidden)]
    #[deprecated]
    #[allow(deprecated)]
    pub fn queries_hash(&self, live_queries: &[LiveQuery]) -> Result<u64, DittoError> {
        let (coll_names, queries): (Vec<_>, Vec<_>) = live_queries
            .iter()
            .map(|lq| (lq.collection_name.as_ref(), lq.query.as_ref()))
            .unzip();

        {
            ffi_sdk::ditto_queries_hash(&self.ditto, coll_names[..].into(), queries[..].into()).ok()
        }
    }

    /// Returns a sequence of English words representing the current version of
    /// the given queries. When a document matching such queries gets mutated,
    /// the words will change as well.
    ///
    /// Please note that the resulting sequence of words depends on how queries
    /// are constructed, so you should make sure to always compare hashes
    /// generated with the same set of queries.
    #[doc(hidden)]
    #[deprecated]
    #[allow(deprecated)]
    pub fn queries_hash_mnemonic(&self, live_queries: &[LiveQuery]) -> Result<String, DittoError> {
        let (coll_names, queries): (Vec<_>, Vec<_>) = live_queries
            .iter()
            .map(|lq| (lq.collection_name.as_ref(), lq.query.as_ref()))
            .unzip();

        {
            ffi_sdk::ditto_queries_hash_mnemonic(
                &self.ditto,
                coll_names[..].into(),
                queries[..].into(),
            )
            .ok()
            .map(|c_str| c_str.into_string())
        }
    }

    #[doc(hidden)]
    #[deprecated(note = "All live query webhook related methods have been deprecated")]
    /// Start all live query webhooks.
    pub fn start_all_live_query_webhooks(&self) -> Result<(), DittoError> {
        error!("Store::start_all_live_query_webhooks has been deprecated: it is now a no-op");
        Err(DittoError::from_str(
            ErrorKind::Deprecation,
            "Live query webhooks are deprecated",
        ))
    }

    #[doc(hidden)]
    #[deprecated(note = "All live query webhook related methods have been deprecated")]
    /// Start a live query webhooks by its id.
    pub fn start_live_query_webhook_by_id(&self, _doc_id: DocumentId) -> Result<(), DittoError> {
        error!("Store::start_live_query_webhook_by_id has been deprecated: it is now a no-op");
        Err(DittoError::from_str(
            ErrorKind::Deprecation,
            "Live query webhooks are deprecated",
        ))
    }

    #[doc(hidden)]
    #[deprecated(note = "All live query webhook related methods have been deprecated")]
    /// Register a new live query webhook
    pub fn register_live_query_webhook(
        &self,
        _collection_name: &str,
        _query: &str,
        _url: &str,
    ) -> Result<DocumentId, DittoError> {
        error!("Store::register_live_query_webhook has been deprecated: it is now a no-op");
        Err(DittoError::from_str(
            ErrorKind::Deprecation,
            "Live query webhooks are deprecated",
        ))
    }

    #[doc(hidden)]
    #[deprecated(note = "All live query webhook related methods have been deprecated")]
    /// Generate a new API secret for live query webhook
    pub fn live_query_webhook_generate_new_api_secret(&self) -> Result<(), DittoError> {
        error!(
            "Store::live_query_webhook_generate_new_api_secret has been deprecated: it is now a \
             no-op"
        );
        Err(DittoError::from_str(
            ErrorKind::Deprecation,
            "Live query webhooks are deprecated",
        ))
    }

    #[doc(hidden)]
    #[deprecated(note = "The experimental `timeseries` feature has been deprecated")]
    #[allow(deprecated)]
    #[cfg(feature = "timeseries")]
    /// Returns a [`TimeSeries`] with the provided name.
    pub fn timeseries(&self, ts_name: &'_ str) -> Result<TimeSeries, DittoError> {
        Self::validate_collection_name(ts_name)?;
        let c_name = char_p::new(ts_name);
        Ok(TimeSeries {
            ditto: self.ditto.retain(),
            ts_name: c_name,
        })
    }

    /// Return a [`DiskUsage`] to monitor the disk usage of the
    /// [`Store`].
    #[doc(hidden)]
    #[deprecated(note = "Use `ditto.disk_usage()` instead")]
    pub fn disk_usage(&self) -> &DiskUsage {
        &self.disk_usage
    }

    /// Deprecated in favour of [`ditto.store().register_observer_with_signal_next_v2()`][0]
    ///
    /// [0]: Store::register_observer_with_signal_next_v2
    #[deprecated = "Use `ditto.store().register_observer_v2()` instead"]
    #[allow(deprecated)]
    #[doc(hidden)]
    pub fn register_observer<Q, F>(
        &self,
        query: Q,
        query_args: Option<QueryArguments>,
        on_change: F,
    ) -> Result<Arc<StoreObserver>, DittoError>
    where
        Q: TryInto<Query, Error = DittoError>,
        F: ChangeHandler,
    {
        let ditto = Ditto::upgrade(&self.weak_ditto_fields)?;

        let query = query.try_into()?;
        let query_args = query_args.as_ref().map(|a| a.cbor());

        let observer = Arc::new(StoreObserver::new(
            &ditto,
            &query.inner_string,
            query_args,
            on_change,
        )?);
        Ok(observer)
    }

    /// Deprecated in favour of [`ditto.store().register_observer_with_signal_next_v2()`][0]
    ///
    /// [0]: Store::register_observer_with_signal_next_v2
    #[deprecated = "Use `ditto.store().register_observer_with_signal_next_v2()` instead"]
    #[allow(deprecated)]
    #[doc(hidden)]
    pub fn register_observer_with_signal_next<Q, F>(
        &self,
        query: Q,
        query_args: Option<QueryArguments>,
        on_change: F,
    ) -> Result<Arc<StoreObserver>, DittoError>
    where
        Q: TryInto<Query, Error = DittoError>,
        F: ChangeHandlerWithSignalNext,
    {
        let ditto = Ditto::upgrade(&self.weak_ditto_fields)?;

        let query = query.try_into()?;
        let query_args = query_args.as_ref().map(|a| a.cbor());

        let new_obs = Arc::new(StoreObserver::with_signal_next(
            &ditto,
            &query.inner_string,
            query_args,
            on_change,
        )?);
        Ok(new_obs)
    }

    /// Installs and returns a store observer for a query, configuring Ditto to
    /// trigger the passed in change handler whenever documents in the local
    /// store change such that the result of the matching query changes. The
    /// passed in query must be a `SELECT` query, otherwise an error will be
    /// returned.
    ///
    /// # Example
    ///
    /// ```
    /// use dittolive_ditto::prelude::*;
    /// # fn example_with_register_observer(ditto: &Ditto) -> anyhow::Result<()> {
    ///
    /// let store = ditto.store();
    /// let _observer = store.register_observer_v2(
    ///     "SELECT * FROM cars WHERE color = 'blue'",
    ///     move |query_result| {
    ///         for item in query_result.iter() {
    ///             // ... handle each item
    ///         }
    ///     },
    /// )?;
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// The first invocation of the change handler will always happen after
    /// this method has returned.
    ///
    /// The observer will remain active until:
    ///
    /// - the [`StoreObserver`] handle gets dropped,
    /// - the [`observer.cancel()`] method is called, or
    /// - the owning [`Ditto`] instance has shut down
    ///
    /// Observer callbacks will never be called concurrently. That is, one callback
    /// must return before the observer will call the handler again. See
    /// [`ditto.store().register_observer_with_signal_next(...)`] if you want
    /// to manually signal readiness for the next callback.
    ///
    /// [`ditto.store().register_observer_with_signal_next(...)`]: crate::store::Store::register_observer_with_signal_next
    /// [`observer.cancel()`]: crate::store::StoreObserver::cancel
    /// [`Ditto`]: crate::prelude::Ditto
    pub fn register_observer_v2<Q, F>(
        &self,
        query: Q,
        on_change: F,
    ) -> Result<Arc<StoreObserver>, DittoError>
    where
        Q: IntoQuery,
        Q::Args: Serialize,
        F: ChangeHandler,
    {
        let ditto = Ditto::upgrade(&self.weak_ditto_fields)?;
        let query = query.into_query()?;

        let observer = Arc::new(StoreObserver::new(
            &ditto,
            &query.string,
            query.args_cbor.as_deref(),
            on_change,
        )?);
        Ok(observer)
    }

    /// Installs and returns a store observer for a query, configuring Ditto to
    /// trigger the passed in change handler whenever documents in the local
    /// store change such that the result of the matching query changes. The
    /// passed in query must be a `SELECT` query, otherwise an error will be
    /// returned.
    ///
    /// Here, a function is passed as an additional argument to the change
    /// handler. This allows the change handler to control how frequently
    /// it is called. See [`register_observer`] for a convenience method that
    /// automatically signals the next invocation.
    ///
    /// # Example
    ///
    /// ```
    /// use dittolive_ditto::prelude::*;
    /// # fn example_with_signal_next(ditto: &Ditto) -> anyhow::Result<()> {
    ///
    /// let store = ditto.store();
    /// let _observer = store.register_observer_with_signal_next_v2(
    ///     "SELECT * FROM cars WHERE color = 'blue'",
    ///     move |query_result, signal_next| {
    ///         for item in query_result.iter() {
    ///             // ... handle each item
    ///         }
    ///
    ///         // Call `signal_next` when you're ready for the next callback
    ///         signal_next();
    ///     },
    /// )?;
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// The first invocation of the change handler will always happen after
    /// this method has returned.
    ///
    /// The observer will remain active until:
    ///
    /// - the [`StoreObserver`] handle gets dropped,
    /// - the [`observer.cancel()`] method is called, or
    /// - the owning [`Ditto`] instance has shut down
    ///
    /// After invoking the callback once, the observer will wait to deliver
    /// another callback until after you've called [`signal_next`].
    ///
    /// [`register_observer`]: Store::register_observer
    /// [`signal_next`]: crate::store::SignalNext
    pub fn register_observer_with_signal_next_v2<Q, F>(
        &self,
        query: Q,
        on_change: F,
    ) -> Result<Arc<StoreObserver>, DittoError>
    where
        Q: IntoQuery,
        Q::Args: Serialize,
        F: ChangeHandlerWithSignalNext,
    {
        let ditto = Ditto::upgrade(&self.weak_ditto_fields)?;
        let query = query.into_query()?;

        let new_obs = Arc::new(StoreObserver::with_signal_next(
            &ditto,
            &query.string,
            query.args_cbor.as_deref(),
            on_change,
        )?);
        Ok(new_obs)
    }

    /// Gets temporary access to the set of currently registered observers.
    ///
    /// A (read) lock is held until the return value is dropped: this means
    /// that neither [`Self::register_observer()`] nor
    /// [`StoreObserver::cancel()`] can make progress until this read
    /// lock is released.
    pub fn observers(&self) -> impl '_ + Deref<Target = SetArc<StoreObserver>> {
        let observers: repr_c::Vec<repr_c::Box<FfiStoreObserver>> =
            ffi_sdk::dittoffi_store_observers(&self.ditto);
        let observers: Vec<_> = observers.into();
        let observers = observers
            .into_iter()
            .map(|handle: repr_c::Box<FfiStoreObserver>| Arc::new(StoreObserver { handle }))
            .collect::<SetArc<_>>();

        Box::new(observers)
    }

    /// Executes the given query in the local store and returns the result.
    ///
    /// Use placeholders to incorporate values from the optional `query_args`
    /// parameter into the query. The keys of the [`QueryArguments`] object must
    /// match the placeholders used within the query. You can not use placeholders
    /// in the `FROM` clause.
    ///
    /// This method only returns results from the local store without waiting for any
    /// [`SyncSubscription`]s to have caught up with the
    /// latest changes. Only use this method if your program must proceed with immediate results.
    ///
    /// Use [`ditto.store().register_observer(...)`] to receive updates to query results
    /// as soon as they have been synced to this peer.
    ///
    /// Limitations:
    ///
    /// - Supports `SELECT * FROM <collection name>` with optional `WHERE <expression>`
    /// - No transactions
    ///
    /// [`SyncSubscription`]: crate::sync::SyncSubscription
    /// [`ditto.store().register_observer(...)`]: crate::store::Store::register_observer
    #[deprecated = "Use `ditto.store().execute_v2(...)` instead"]
    #[allow(deprecated)]
    #[doc(hidden)]
    pub async fn execute<Q>(
        &self,
        query: Q,
        query_args: Option<QueryArguments>,
    ) -> Result<QueryResult, DittoError>
    where
        Q: TryInto<Query, Error = DittoError>,
    {
        // Get a DqlResult. The operation to get it is fallible, so we end up
        // with a `FfiResult<DqlResult> -> Result<DqlResult> -> DqlResult`.
        // The `Result` stutter is quite unfortunate, which _results_ from
        // having picked that name in our public SDK API (over, say, `Output`).
        let ffi_query_result = ffi_sdk::dittoffi_try_exec_statement(
            &self.ditto,
            query.try_into()?.prepare_ffi(),
            query_args.as_ref().map(|qa| qa.cbor().into()),
        )
        .into_rust_result()?;

        Ok(QueryResult::from(ffi_query_result))
    }

    /// Executes the given query in the local store and returns the result.
    ///
    /// # Example
    ///
    /// ```
    /// use dittolive_ditto::prelude::*;
    /// use dittolive_ditto::dql::QueryResult;
    /// # async fn example(ditto: &Ditto) -> anyhow::Result<()> {
    ///
    /// // Query a collection
    /// let result = ditto.store().execute_v2("SELECT * FROM cars").await?;
    ///
    /// // Insert a document into a collection
    /// let insert_result: QueryResult = ditto
    ///     .store()
    ///     .execute_v2((
    ///          "INSERT INTO cars DOCUMENTS (:newCar)",
    ///          serde_json::json!({
    ///              "newCar": {
    ///                  "make": "ford",
    ///                  "color": "blue"
    ///              }
    ///          })
    ///     ))
    ///     .await?;
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// Use placeholders to incorporate values from the optional `query_args`
    /// parameter into the query. The keys of the [`QueryArguments`] object must
    /// match the placeholders used within the query. You can not use placeholders
    /// in the `FROM` clause.
    ///
    /// This method only returns results from the local store without waiting for any
    /// [`SyncSubscription`]s to have caught up with the
    /// latest changes. Only use this method if your program must proceed with immediate results.
    ///
    /// Use [`ditto.store().register_observer(...)`] to receive updates to query results
    /// as soon as they have been synced to this peer.
    ///
    /// ## Query parameter
    ///
    /// The `query` parameter must implement [`IntoQuery`], which is a trait that is implemented by
    /// objects that can be turned into a query string along with the relevant query argumnents.
    ///
    /// For queries with no arguments, a [`String`] is sufficient
    ///
    /// [`SyncSubscription`]: crate::sync::SyncSubscription
    /// [`ditto.store().register_observer(...)`]: crate::store::Store::register_observer
    pub async fn execute_v2<Q>(&self, query: Q) -> Result<QueryResult, DittoError>
    where
        Q: IntoQuery,
        Q::Args: serde::Serialize,
    {
        let query = query.into_query()?;
        let query_string = (&*query.string).into();
        let query_args = query.args_cbor.as_deref().map(Into::into);

        let ffi_query_result =
            ffi_sdk::dittoffi_try_exec_statement(&self.ditto, query_string, query_args)
                .into_rust_result()?;

        Ok(QueryResult::from(ffi_query_result))
    }

    /// Creates a new attachment, which can then be inserted into a document.
    ///
    /// The file residing at the provided path will be copied into Ditto’s store. The
    /// [`DittoAttachment`] object that is returned is what you can
    /// then use to insert an attachment into a document.
    ///
    /// You can provide custom user data about the attachment, which will be replicated to other
    /// peers alongside the file attachment.
    pub async fn new_attachment(
        &self,
        filepath: &(impl ?Sized + AsRef<Path>),
        user_data: HashMap<String, String>,
    ) -> Result<DittoAttachment, DittoError> {
        DittoAttachment::from_file_and_metadata(filepath, user_data, &self.ditto)
    }

    /// Creates a new attachment from in-memory data
    ///
    /// Refer to [`new_attachment`](Self::new_attachment) for additional information.
    pub async fn new_attachment_from_bytes(
        &self,
        bytes: &(impl ?Sized + AsRef<[u8]>),
        user_data: HashMap<String, String>,
    ) -> Result<DittoAttachment, DittoError> {
        DittoAttachment::from_bytes_and_metadata(bytes, user_data, &self.ditto)
    }

    /// Fetches the attachment corresponding to the provided attachment token.
    /// - `attachment_token`: can be either a [`DittoAttachmentToken`], or a `&BTreeMap<CborValue,
    ///   CborValue>`, that is, the output of a [`QueryResultItem::value()`] once casted
    ///   [`.as_object()`][crate::prelude::CborValueGetters::as_object()].
    ///
    /// - `on_fetch_event`: A closure that will be called when the status of the request to fetch
    ///   the attachment has changed. If the attachment is already available then this will be
    ///   called almost immediately with a completed status value.
    ///
    /// The returned [`DittoAttachmentFetcher`] is a handle which is safe to discard, unless you
    /// wish to be able to [`.cancel()`][DittoAttachmentFetcher::cancel] the fetching operation.
    /// When not explicitly cancelled, the fetching operation will remain active until it either
    /// completes, the attachment is deleted, or the owning [`Ditto`] object is dropped.
    pub fn fetch_attachment(
        &self,
        attachment_token: impl attachment::token::DittoAttachmentTokenLike,
        on_fetch_event: impl 'static + Send + Sync + Fn(DittoAttachmentFetchEvent),
    ) -> Result<DittoAttachmentFetcherV2, DittoError> {
        let attachment_token = attachment_token.parse_attachment_token()?;

        let weak_ditto = self.weak_ditto_fields.clone();
        let ditto = weak_ditto
            .upgrade()
            .ok_or(ErrorKind::ReleasedDittoInstance)?;

        let mut attachment_fetchers_lockguard = self.attachment_fetchers.write().unwrap();
        let fetcher = DittoAttachmentFetcher::new(
            attachment_token,
            Some(&ditto),
            &self.ditto,
            // Shim around `on_fetch_event` to `cancel` on completion.
            move |event, cancel_token: &AtomicU64| {
                let has_finished = matches! {
                    event,
                    | DittoAttachmentFetchEvent::Completed { .. }
                    | DittoAttachmentFetchEvent::Deleted { .. }
                };
                on_fetch_event(event);
                if has_finished {
                    if let Some(ditto) = weak_ditto.upgrade() {
                        let mut attachment_fetchers_inner_lockguard =
                            ditto.store.attachment_fetchers.write().unwrap();
                        // Relaxed is fine thanks to the lock.
                        let cancel_token = cancel_token.load(atomic::Ordering::Relaxed);
                        ditto.store.unregister_fetcher(
                            cancel_token,
                            Some(&mut *attachment_fetchers_inner_lockguard),
                        );
                    }
                }
            },
        )?;
        let (cancel_token, was_zero) = fetcher.cancel_token_ensure_unique();
        attachment_fetchers_lockguard.insert(cancel_token, (was_zero, fetcher.clone()));
        Ok(fetcher)
    }

    fn unregister_fetcher(
        &self,
        mut fetcher_cancel_token: CancelToken,
        fetchers: Option<&mut HashMap<CancelToken, (bool, DittoAttachmentFetcherV2)>>,
    ) -> bool {
        let mut lock_guard = None;
        let fetchers = fetchers.unwrap_or_else(|| {
            &mut **lock_guard.get_or_insert(self.attachment_fetchers.write().unwrap())
        });

        let Some((was_zero, removed_fetcher)) = fetchers.remove(&fetcher_cancel_token) else {
            return false;
        };
        drop(lock_guard);

        if was_zero {
            fetcher_cancel_token = 0;
        }

        let att_token = &removed_fetcher.context.token;
        debug!(
            token_id = %att_token.id(),
            %fetcher_cancel_token,
            "unregistering ditto attachment fetcher"
        );

        let status = ffi_sdk::ditto_cancel_resolve_attachment(
            &self.ditto,
            att_token.id.as_ref().into(),
            fetcher_cancel_token,
        );

        if status != 0 {
            error!(
                token_id = %att_token.id(),
                %fetcher_cancel_token,
                "failed to clean up attachment fetcher"
            );
        }
        status == 0
    }

    /// Gets a copy of the set of currently registered attachment fetchers.
    ///
    /// A (read) lock is held during the copy: this contends with [`Self::fetch_attachment()`] and
    /// with [`DittoAttachmentFetcher::cancel()`].
    pub fn attachment_fetchers(&self) -> Vec<DittoAttachmentFetcherV2> {
        self.attachment_fetchers
            .read()
            .unwrap()
            .iter()
            .map(|(_, (_, fetcher))| fetcher.clone())
            .collect()
    }
}

/// Specify the order of returned Documents in a query.
#[non_exhaustive]
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub enum SortDirection {
    /// First result is "smallest", last result is "largest"
    Ascending,

    /// First result is "largest", last result is "smallest"
    Descending,
}

#[non_exhaustive]
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
/// Specify the write strategy when inserting documents.
pub enum WriteStrategy {
    /// An existing document will be merged with the document being inserted, if there is a
    /// pre-existing document.
    Merge,

    /// Insert the document only if there is not already a document with the same Id in the store.
    /// If there is already a document in the store with the same Id then this will be a no-op.
    InsertIfAbsent,

    /// Insert the document, with its contents treated as default data, only if there is not
    /// already a document with the same Id in the store. If there is already a document in the
    /// store with the same Id then this will be a no-op. Use this strategy if you want to
    /// insert default data for a given document Id, which you want to treat as common initial
    /// data amongst all peers and that you expect to be mutated or overwritten in due course.
    InsertDefaultIfAbsent,
}

impl WriteStrategy {
    fn as_write_strategy_rs(&self) -> WriteStrategyRs {
        match self {
            WriteStrategy::Merge => WriteStrategyRs::Merge,
            WriteStrategy::InsertIfAbsent => WriteStrategyRs::InsertIfAbsent,
            WriteStrategy::InsertDefaultIfAbsent => WriteStrategyRs::InsertDefaultIfAbsent,
        }
    }
}