dittolive-ditto 4.11.2

use_prelude!();

use std::collections::BTreeMap;

use ffi_sdk::COrderByParam;

use crate::{
    ditto::{TryUpgrade, WeakDittoHandleWrapper},
    error::{DittoError, ErrorKind},
    store::{collection::document_id::DocumentId, live_query::LiveQuery, update::UpdateResult},
    subscription::Subscription,
};

/// Use [`.find_all()`], [`.find(...)`], or [`.find_with_args(...)`] to query documents in a
/// collection.
///
/// This object lets you take various actions on the selected documents, such as:
///
/// - [`.exec()`]: Immediately return the current state of the documents from the local store.
///
/// - [`.observe_local(...)`]: Register a callback to receive notifications whenever the documents
///   change in the local store. Note that this does not automatically begin syncing these documents
///   from remote peers, use [`.subscribe()`] for that.
///
/// - [`.subscribe()`]: Create a [`Subscription`] that syncs these documents from remote peers.
///
/// - [`.limit(...)`]: Limit the number of documents that get returned by this query.
///
/// - [`.offset(...)`]: Offsets the resulting set of matching documents. Useful for skipping the
///   first N matching documents, such as with paging.
///
/// - [`.sort(...)`]: Sort the matching documents according to chosen fields and ordering.
///
/// - [`.remove()`]: Remove the selected documents from the collection, replacing them with
///   tombstones. If another peer concurrently updates these documents, the update will win
///   ("add-wins"). This removal may be synced to other peers.
///
/// - [`.evict()`]: Evicts the selected documents from this peer's local store. This can be used to
///   reduce resource consumption on peers that have finished using documents. This will _not_ cause
///   remote peers to evict or remove these document, it only impacts the local peer's store.
///
/// - [`.update(...)`]: Update the documents via a callback that receives a mutable reference to
///   them.
///
/// Typically, an app would call [`.subscribe()`] in some part of the application which
/// is long-lived to ensure the device receives updates from the mesh. These updates will be
/// automatically received and written into the local store. Elsewhere, where you need to use this
/// data, [`.observe_local(...)`] can be used to notify you of the data, and all
/// subsequent changes to the data.
///
/// # Example
///
/// ```
/// use dittolive_ditto::prelude::*;
///
/// # fn example(ditto: &Ditto) -> anyhow::Result<()> {
/// let collection = ditto.store().collection("cars")?;
/// let pending_op = collection.find("color == 'blue'");
///
/// // Hold a `Subscription` to continuously sync from remote peers
/// let _subscription = pending_op.subscribe();
///
/// // Hold a `LiveQuery` to receive callbacks when selected documents are changed
/// let _live_query = pending_op.observe_local(|docs, _event| {
///     let typed: Vec<serde_json::Value> = docs
///         .into_iter()
///         .flat_map(|doc| doc.typed::<serde_json::Value>().ok())
///         .collect();
///     println!("Observed doc changes: {typed:?}");
/// })?;
/// # Ok(())
/// # }
/// ```
///
/// [`.find_all()`]: crate::store::query_builder::Collection::find_all
/// [`.find(...)`]: crate::store::query_builder::Collection::find
/// [`.find_with_args(...)`]: crate::store::query_builder::Collection::find_with_args
/// [`.exec()`]: Self::exec
/// [`.observe_local(...)`]: Self::observe_local
/// [`.subscribe()`]: Self::subscribe
/// [`.limit(...)`]: Self::limit
/// [`.offset(...)`]: Self::offset
/// [`.sort(...)`]: Self::sort
/// [`.remove()`]: Self::remove
/// [`.evict()`]: Self::evict
/// [`.update(...)`]: Self::update
pub struct PendingCursorOperation<'order_by> {
    pub(super) ditto: WeakDittoHandleWrapper,
    pub(super) collection_name: char_p::Box,
    pub(super) query: char_p::Box,
    pub(super) query_args: Option<Vec<u8>>,
    pub(super) offset: u32,
    pub(super) limit: i32,
    pub(super) order_by: Vec<COrderByParam<'order_by>>,
}

impl<'order_by> PendingCursorOperation<'order_by> {
    #[doc(hidden)]
    #[deprecated(
        note = "See the `store::query_builder` module docs to learn about `PendingCursorOperation`"
    )]
    pub fn new(
        ditto: WeakDittoHandleWrapper,
        collection_name: char_p::Box,
        query: &str,
        query_args: Option<Vec<u8>>,
    ) -> Self {
        let query = char_p::new(query);
        Self {
            ditto,
            collection_name,
            query,
            query_args,
            offset: 0,
            limit: -1,
            order_by: vec![],
        }
    }

    /// Execute the find operation and return the matching documents.
    ///
    /// The documents returned represent the current state of the local store,
    /// this function does not cause any syncing with remote peers.
    ///
    /// # Example
    ///
    /// ```
    /// use dittolive_ditto::prelude::*;
    ///
    /// # fn example(ditto: &Ditto) -> anyhow::Result<()> {
    /// let collection = ditto.store().collection("cars")?;
    /// let pending_op = collection.find_all();
    ///
    /// // Execute the query to receive Ditto documents
    /// let documents: Vec<_> = pending_op.exec()?;
    ///
    /// // Use `doc.typed::<T>` with a `Deserialize` type for typed access
    /// let typed_docs: Vec<serde_json::Value> = documents
    ///     .into_iter()
    ///     .flat_map(|doc| doc.typed::<serde_json::Value>().ok())
    ///     .collect();
    /// # Ok(())
    /// # }
    /// ```
    pub fn exec(&self) -> Result<Vec<BoxedDocument>, DittoError> {
        self.exec_internal(None)
    }

    fn exec_internal(
        &self,
        txn: Option<&'_ mut ffi_sdk::CWriteTransaction>,
    ) -> Result<Vec<BoxedDocument>, DittoError> {
        let ditto = self.ditto.try_upgrade()?;
        {
            ffi_sdk::ditto_collection_exec_query_str(
                &*ditto,
                self.collection_name.as_ref(),
                txn,
                self.query.as_ref(),
                self.query_args.as_ref().map(|qa| (&qa[..]).into()),
                (&self.order_by[..]).into(),
                self.limit,
                self.offset,
            )
        }
        .ok_or(ErrorKind::InvalidInput)
        .map(|it| it.into())
    }

    /// Register a callback that delivers updates when selected documents change.
    ///
    /// The callback will be called any time the selected documents are changed in this peer's local
    /// store. It will NOT cause the documents to be synced with remote peers, for that use
    /// [`.subscribe()`].
    ///
    /// Hold the returned [`LiveQuery`] for as long as you wish to continue receiving updates.
    /// Dropping the [`LiveQuery`] will cancel the callbacks.
    ///
    /// ```
    /// use dittolive_ditto::prelude::*;
    ///
    /// # fn example(ditto: &Ditto) -> anyhow::Result<()> {
    /// let collection = ditto.store().collection("cars")?;
    /// let pending_op = collection.find("color == 'blue'");
    ///
    /// // Hold the LiveQuery to continue receiving callbacks
    /// let _live_query = pending_op.observe_local(|docs, _event| {
    ///     let typed: Vec<serde_json::Value> = docs
    ///         .into_iter()
    ///         .flat_map(|doc| doc.typed::<serde_json::Value>().ok())
    ///         .collect();
    ///     println!("Received doc updates: {typed:?}");
    /// })?;
    ///
    /// // Drop the LiveQuery to cancel the callback
    /// drop(_live_query);
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// [`.subscribe()`]: Self::subscribe
    pub fn observe_local<Handler>(&self, handler: Handler) -> Result<LiveQuery, DittoError>
    where
        Handler: EventHandler,
    {
        #[allow(deprecated)] // TODO deprecation temporary until pub(crate)
        LiveQuery::with_handler(
            self.ditto.clone(),
            self.query.clone(),
            self.query_args.clone(),
            self.collection_name.clone(),
            &self.order_by,
            self.limit,
            self.offset,
            handler,
        )
    }

    /// Creates a [`Subscription`] that syncs the selected documents from remote peers.
    ///
    /// Holding the `Subscription` will cause Ditto to continue syncing the selected documents from
    /// remote peers. Dropping the `Subscription` will cancel the sync with remote peers.
    ///
    /// # Example
    ///
    /// ```
    /// use dittolive_ditto::prelude::*;
    ///
    /// # fn example(ditto: &Ditto) -> anyhow::Result<()> {
    /// let collection = ditto.store().collection("cars")?;
    /// let pending_op = collection.find_all();
    ///
    /// // Hold the Subscription to continue syncing with remote peers
    /// let _subscription = pending_op.subscribe();
    ///
    /// // Drop the Subscription to cancel sync with remote peers
    /// drop(_subscription);
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// # Panics
    ///
    /// Panics if the `Ditto` object has been closed.
    pub fn subscribe(&self) -> Subscription {
        let ditto = self.ditto.try_upgrade().unwrap();
        Subscription::new(
            ditto,
            self.collection_name.clone(),
            self.query.to_str(),
            self.query_args.clone(),
            &self.order_by,
            self.limit,
            self.offset,
        )
    }

    /// Limit the number of documents that get returned when querying a
    /// collection for matching documents.
    ///
    /// # Example
    ///
    /// ```
    /// use dittolive_ditto::prelude::*;
    ///
    /// # fn example(ditto: &Ditto) -> anyhow::Result<()> {
    /// let collection = ditto.store().collection("cars")?;
    /// let mut pending_op = collection.find_all();
    ///
    /// // The number of documents returned is limited to at most 20
    /// let documents: Vec<_> = pending_op.limit(20).exec()?;
    /// assert!(documents.len() <= 20);
    /// # Ok(())
    /// # }
    /// ```
    pub fn limit(&mut self, limit: i32) -> &mut Self {
        self.limit = limit;
        self
    }

    /// Offset the resulting set of matching documents.
    ///
    /// This is useful if you aren’t interested in the first N matching
    /// documents for one reason or another. For example, you might already
    /// have queried the collection and obtained the first 20 matching documents
    /// and so you might want to run the same query as you did previously but
    /// ignore the first 20 matching documents, and that is where you would
    /// use `offset`.
    pub fn offset(&mut self, offset: u32) -> &mut Self {
        self.offset = offset;
        self
    }

    // FIXME: To bring this in line with the other SDKs this should accept a single
    // "order_by" expression, which should then be added to the `order_by` vec
    /// Sort the documents that match the query provided in the preceding
    /// find-like function call.
    ///
    /// Documents that are missing the field to sort by will appear at
    /// the beginning of the results when sorting in ascending order.
    pub fn sort(&mut self, sort_param: Vec<COrderByParam<'order_by>>) -> &mut Self {
        self.order_by = sort_param;
        self
    }

    /// Remove the queried documents from this [`Collection`], returning the [`DocumentId`]s of all
    /// documents removed.
    ///
    /// A document removal may sync to other peers, causing them to remove the document
    /// as well. However, a concurrent update from a remote peer will win when merging the update
    /// with the removal ("add-wins").
    ///
    /// # Example
    ///
    /// ```
    /// use dittolive_ditto::prelude::*;
    ///
    /// # fn example(ditto: &Ditto) -> anyhow::Result<()> {
    /// let collection = ditto.store().collection("todos")?;
    /// let pending_op = collection.find("done == true");
    ///
    /// // Removing documents returns their IDs
    /// let removed_ids: Vec<DocumentId> = pending_op.remove()?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn remove(&self) -> Result<Vec<DocumentId>, DittoError> {
        let ditto = self
            .ditto
            .upgrade()
            .ok_or(crate::error::ErrorKind::ReleasedDittoInstance)?;
        let hint: Option<char_p::Ref<'_>> = None;
        let mut write_txn = ffi_sdk::ditto_write_transaction(&*ditto, hint).ok()?;

        let ids = {
            ffi_sdk::ditto_collection_remove_query_str(
                &*ditto,
                self.collection_name.as_ref(),
                &mut write_txn,
                self.query.as_ref(),
                self.query_args.as_ref().map(|qa| (&qa[..]).into()),
                (&self.order_by[..]).into(),
                self.limit,
                self.offset,
            )
            .ok_or(ErrorKind::InvalidInput)?
        };
        ffi_sdk::ditto_write_transaction_commit(&ditto, write_txn);
        Ok(ids
            .to::<Vec<_>>()
            .into_iter()
            .map(|s| s.to::<Box<[u8]>>().into())
            .collect())
    }

    /// Evict the selected documents from the local peer's store.
    ///
    /// Unlike a [`removal`], an eviction does not sync with other peers. When
    /// a document is evicted, it is simply removed from the local store to save
    /// disk space.
    ///
    /// This returns the [`DocumentId`]s of the documents that were evicted.
    ///
    /// # Example
    ///
    /// ```
    /// use dittolive_ditto::prelude::*;
    ///
    /// # fn example(ditto: &Ditto) -> anyhow::Result<()> {
    /// let collection = ditto.store().collection("cars")?;
    /// let pending_op = collection.find("sold == true");
    ///
    /// // Evicting documents returns their DocumentIds
    /// let evicted_ids: Vec<DocumentId> = pending_op.evict()?;
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// [`removal`]: Self::remove
    pub fn evict(&self) -> Result<Vec<DocumentId>, DittoError> {
        let ditto = self
            .ditto
            .upgrade()
            .ok_or(crate::error::ErrorKind::ReleasedDittoInstance)?;
        let hint: Option<char_p::Ref<'_>> = None;
        let mut write_txn = ffi_sdk::ditto_write_transaction(&*ditto, hint).ok()?;

        let ids = {
            ffi_sdk::ditto_collection_evict_query_str(
                &*ditto,
                self.collection_name.as_ref(),
                &mut write_txn,
                self.query.as_ref(),
                self.query_args.as_ref().map(|qa| (&qa[..]).into()),
                (&self.order_by[..]).into(),
                self.limit,
                self.offset,
            )
            .ok_or(ErrorKind::InvalidInput)?
        };
        ffi_sdk::ditto_write_transaction_commit(&ditto, write_txn);
        Ok(ids
            .to::<Vec<_>>()
            .into_iter()
            .map(|s| s.to::<Box<[u8]>>().into())
            .collect())
    }

    /// Update the selected documents by mutating them directly.
    ///
    /// This function takes a callback and provides it with a mutable slice
    /// of the documents matching the query.
    ///
    /// After performing the update, this function returns a map of each
    /// `DocumentId` affected and a list of [`UpdateResult`]s describing the
    /// updates that happened.
    ///
    /// # Example
    ///
    /// ```
    /// use dittolive_ditto::prelude::*;
    ///
    /// # fn example(ditto: &Ditto) -> anyhow::Result<()> {
    /// let collection = ditto.store().collection("cars")?;
    /// let pending_op = collection.find("model == 'pinto'");
    ///
    /// let _update_results = pending_op.update(|docs| {
    ///     for doc in docs {
    ///         doc.set("recalled", true).unwrap();
    ///     }
    /// })?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn update<Updater>(
        &self,
        updater: Updater,
    ) -> Result<BTreeMap<DocumentId, Vec<UpdateResult>>, DittoError>
    where
        Updater: Fn(&mut [BoxedDocument]),
    {
        let ditto = self
            .ditto
            .upgrade()
            .ok_or(crate::error::ErrorKind::ReleasedDittoInstance)?;
        let hint: Option<char_p::Ref<'_>> = None;
        let mut write_txn = ffi_sdk::ditto_write_transaction(&*ditto, hint).ok()?;

        let mut docs = self.exec_internal(Some(&mut write_txn))?;

        // Apply the closure to the document,
        updater(&mut docs);
        let diff = BTreeMap::<DocumentId, Vec<UpdateResult>>::new();

        let code = {
            ffi_sdk::ditto_collection_update_multiple(
                &ditto,
                self.collection_name.as_ref(),
                &mut write_txn,
                docs.into(),
            )
        };
        if code != 0 {
            ffi_sdk::ditto_write_transaction_rollback(&ditto, write_txn);
            return Err(DittoError::from_ffi(ErrorKind::InvalidInput));
        }
        ffi_sdk::ditto_write_transaction_commit(&ditto, write_txn);
        Ok(diff)
    }
}