dittolive-ditto 4.14.2

use std::{ops::Not, sync::Weak};

use serde::Serialize;

use crate::{
    ditto::TryUpgrade,
    error::{DittoError, ErrorKind},
    ffi_sdk,
    store::{
        collection::document_id::DocumentId,
        live_query::{SingleDocumentEventHandler, SingleDocumentLiveQueryEvent},
        update::UpdateResult,
    },
    subscription::Subscription,
    utils::prelude::*,
};

const LMDB_ERROR_NOT_FOUND_CODE: i32 = -30798;

/// Use [`.find_by_id(...)`] to query or mutate a single Document by its ID.
///
/// This object lets you take various actions on the selected document, such as:
///
/// - [`.exec()`]: Immediately return the current state of the document from the local store.
///
/// - [`.observe_local(...)`]: Register a callback to receive notifications whenever the document
///   changes in the local store. Note that this does not automatically begin syncing this document
///   from remote peers, use [`.subscribe()`] for that.
///
/// - [`.subscribe()`]: Create a [`Subscription`] that syncs this document from remote peers.
///
/// - [`.remove()`]: Remove the document from the collection, replacing it with a tombstone. If
///   another peer concurrently updates this document, the update will win ("add-wins"). This
///   removal may be synced to other peers.
///
/// - [`.evict()`]: Evicts the document from this peer's local store. This can be used to reduce
///   resource consumption on peers that have finished using a document. This will _not_ cause
///   remote peers to evict or remove this document, it only impacts the local peer's store.
///
/// - [`.update(...)`]: Update the document via a callback that receives a mutable reference to it.
///
/// - [`.update_doc(...)`]: Update the document by upserting a new value to replace the document.
///
/// [`.find_by_id(...)`]: crate::store::query_builder::Collection::find_by_id
/// [`.exec()`]: Self::exec
/// [`.observe_local(...)`]: Self::observe_local
/// [`.subscribe()`]: Self::subscribe
/// [`.remove()`]: Self::remove
/// [`.evict()`]: Self::evict
/// [`.update(...)`]: Self::update
/// [`.update_doc(...)`]: Self::update_doc
pub struct PendingIdSpecificOperation {
    pub(super) ditto: Weak<BoxedDitto>,
    pub(super) collection_name: char_p::Box,
    pub(super) doc_id: DocumentId,
}

impl PendingIdSpecificOperation {
    fn query(&self) -> String {
        format!(
            "_id == {}",
            self.doc_id
                .to_query_compatible(ffi_sdk::StringPrimitiveFormat::WithQuotes)
        )
    }

    /// Execute the find operation and return the document with the matching ID.
    ///
    /// # Example
    ///
    /// ```
    /// use dittolive_ditto::prelude::*;
    ///
    /// # fn example(ditto: &Ditto) -> anyhow::Result<()> {
    /// let pending_op = ditto
    ///     .store()
    ///     .collection("cars")?
    ///     .find_by_id(&b"mustang"[..]);
    /// let document = pending_op.exec()?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn exec(&self) -> Result<BoxedDocument, DittoError> {
        let ditto = self
            .ditto
            .upgrade()
            .ok_or(crate::error::ErrorKind::ReleasedDittoInstance)?;
        let mut txn = ffi_sdk::ditto_read_transaction(&ditto).ok()?;
        let result = {
            ffi_sdk::ditto_collection_get(
                &ditto,
                self.collection_name.as_ref(),
                self.doc_id.bytes.as_slice().into(),
                &mut txn,
            )
        };
        if result.status_code == LMDB_ERROR_NOT_FOUND_CODE {
            Err(DittoError::from_ffi(ErrorKind::NonExtant))
        } else {
            result.ok()
        }
    }

    /// Register a callback that delivers updates when the document changes.
    ///
    /// The callback will be called any time the selected document is changed in this peer's
    /// local store. It will NOT cause the document 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.
    ///
    /// # Example
    ///
    /// ```
    /// use dittolive_ditto::prelude::*;
    ///
    /// # fn example(ditto: &Ditto) -> anyhow::Result<()> {
    /// let pending_op = ditto
    ///     .store()
    ///     .collection("cars")?
    ///     .find_by_id(&b"mustang"[..]);
    ///
    /// // Hold the LiveQuery to continue receiving callbacks
    /// let _live_query = pending_op.observe_local(|maybe_doc, event| {
    ///     if let Some(doc) = maybe_doc {
    ///         println!("Doc update: {doc:?}");
    ///     }
    /// })?;
    ///
    /// // Drop the LiveQuery to cancel the callbacks
    /// drop(_live_query);
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// [`.subscribe()`]: Self::subscribe
    pub fn observe_local<Handler>(&self, handler: Handler) -> Result<LiveQuery, DittoError>
    where
        Handler: SingleDocumentEventHandler,
    {
        let mut handler = handler;
        let mapped_handler = move |mut docs: Vec<BoxedDocument>, event: LiveQueryEvent| match event
        {
            LiveQueryEvent::Initial => {
                if docs.len() > 1 {
                    error!("single document live query returned more than 1 document");
                    debug_assert!(docs.len() <= 1);
                }
                let event = SingleDocumentLiveQueryEvent {
                    is_initial: true,
                    old_document: None,
                };
                handler(docs.pop(), event);
            }
            LiveQueryEvent::Update {
                mut old_documents,
                insertions,
                deletions,
                updates,
                moves,
            } => {
                if moves.is_empty().not() {
                    error!(
                        "single document live query received an update event saying that there \
                         was a move, which should not happen"
                    );
                    debug_assert!(moves.is_empty());
                }
                if insertions.len() > 1 || deletions.len() > 1 || updates.len() > 1 {
                    error!(
                        "single document live query received an update event with too many \
                         insertions, deletions, or updates"
                    );
                    debug_assert!(insertions.len() <= 1);
                    debug_assert!(deletions.len() <= 1);
                    debug_assert!(updates.len() <= 1);
                }
                let event = SingleDocumentLiveQueryEvent {
                    is_initial: false,
                    old_document: old_documents.pop(),
                };
                handler(docs.pop(), event);
            }
        };

        #[allow(deprecated)]
        LiveQuery::with_handler(
            self.ditto.clone(),
            char_p::new(self.query().as_str()),
            None,
            self.collection_name.clone(),
            &[],
            -1,
            0,
            mapped_handler,
        )
    }

    /// Creates a [`Subscription`] that syncs this document from remote peers.
    ///
    /// Holding the `Subscription` will cause Ditto to continue syncing this
    /// document 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 pending_op = ditto
    ///     .store()
    ///     .collection("cars")?
    ///     .find_by_id(&b"mustang"[..]);
    ///
    /// // Hold the subscription to continue syncing
    /// let _subscription = pending_op.subscribe();
    ///
    /// // Drop the subscription to cancel sync
    /// drop(_subscription);
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// # Panics
    ///
    /// Panics if the parent Ditto object has been closed.
    pub fn subscribe(&self) -> Subscription {
        Subscription::new(
            self.ditto.try_upgrade().unwrap(),
            self.collection_name.clone(),
            self.query().as_str(),
            None,
            &(vec![])[..],
            -1,
            0,
        )
    }

    /// Removes this document from its [`Collection`].
    ///
    /// 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 pending_op = ditto
    ///     .store()
    ///     .collection("cars")?
    ///     .find_by_id(&b"mustang"[..]);
    /// pending_op.remove()?;
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// [`Collection`]: crate::store::query_builder::Collection
    pub fn remove(&self) -> Result<(), DittoError> {
        let ditto = self
            .ditto
            .upgrade()
            .ok_or(crate::error::ErrorKind::ReleasedDittoInstance)?;
        let hint: Option<char_p::Ref<'_>> = None;
        let mut txn = ffi_sdk::ditto_write_transaction(&ditto, hint).ok()?;
        let removed = {
            ffi_sdk::ditto_collection_remove(
                &ditto,
                self.collection_name.as_ref(),
                &mut txn,
                self.doc_id.bytes.as_slice().into(),
            )
            .ok()?
        };
        if removed {
            let status = ffi_sdk::ditto_write_transaction_commit(&ditto, txn);
            if status != 0 {
                return Err(DittoError::from_ffi(ErrorKind::Internal));
            }
            Ok(())
        } else {
            ffi_sdk::ditto_write_transaction_rollback(&ditto, txn);
            Err(DittoError::from_ffi(ErrorKind::NonExtant))
        }
    }

    /// Evict this document from the local peer's store.
    ///
    /// Unlike a [`removal`], an eviction does not create a tombstone to sync
    /// with other peers. When a document is evicted, it is simply removed from
    /// the local store to save disk space.
    ///
    /// # Example
    ///
    /// ```
    /// use dittolive_ditto::prelude::*;
    ///
    /// # fn example(ditto: &Ditto) -> anyhow::Result<()> {
    /// let pending_op = ditto
    ///     .store()
    ///     .collection("cars")?
    ///     .find_by_id(&b"mustang"[..]);
    /// pending_op.evict()?;
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// [`removal`]: Self::remove
    pub fn evict(&self) -> Result<(), DittoError> {
        let ditto = self
            .ditto
            .upgrade()
            .ok_or(crate::error::ErrorKind::ReleasedDittoInstance)?;
        let hint: Option<char_p::Ref<'_>> = None;
        let mut txn = ffi_sdk::ditto_write_transaction(&ditto, hint).ok()?;
        let evicted = {
            ffi_sdk::ditto_collection_evict(
                &ditto,
                self.collection_name.as_ref(),
                &mut txn,
                self.doc_id.bytes.as_slice().into(),
            )
            .ok()?
        };
        if evicted {
            let status = ffi_sdk::ditto_write_transaction_commit(&ditto, txn);
            if status != 0 {
                return Err(DittoError::from_ffi(ErrorKind::Internal));
            }
            Ok(())
        } else {
            ffi_sdk::ditto_write_transaction_rollback(&ditto, txn);
            Err(DittoError::from_ffi(ErrorKind::NonExtant))
        }
    }

    /// Update the selected document by mutating it directly.
    ///
    /// # Example
    ///
    /// ```
    /// use dittolive_ditto::prelude::*;
    ///
    /// # fn example(ditto: &Ditto) -> anyhow::Result<()> {
    /// let pending_op = ditto
    ///     .store()
    ///     .collection("cars")?
    ///     .find_by_id(&b"mustang"[..]);
    ///
    /// let results = pending_op.update(|maybe_doc| {
    ///     if let Some(doc) = maybe_doc {
    ///         doc.set("details.make", "ford").unwrap();
    ///     }
    /// })?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn update<Updater>(&self, updater: Updater) -> Result<Vec<UpdateResult>, DittoError>
    where
        Updater: Fn(Option<&mut BoxedDocument>), /* Arg is a Mutable Document, which only exists
                                                  * in SDKs */
    {
        let ditto = self
            .ditto
            .upgrade()
            .ok_or(crate::error::ErrorKind::ReleasedDittoInstance)?;

        let hint: Option<char_p::Ref<'_>> = None;
        let mut write_txn = Some(ffi_sdk::ditto_write_transaction(&ditto, hint).ok()?);
        let mut document = Some(
            ffi_sdk::ditto_collection_get_with_write_transaction(
                &ditto,
                self.collection_name.as_ref(),
                self.doc_id.bytes.as_slice().into(),
                write_txn.as_mut().expect("write txn Some"),
            )
            .ok_or_else(|| {
                ffi_sdk::ditto_write_transaction_rollback(
                    &ditto,
                    write_txn.take().expect("write txn Some"),
                );
                ErrorKind::NonExtant
            })?,
        );

        // Apply the closure to the document
        updater(document.as_mut());
        let diff = Vec::with_capacity(0); // TODO Mutable doc will populate this

        if let Some(doc) = document {
            match ffi_sdk::ditto_collection_update(
                &ditto,
                self.collection_name.as_ref(),
                write_txn.as_mut().expect("write txn Some"),
                doc,
            ) {
                0 => {
                    let status = ffi_sdk::ditto_write_transaction_commit(
                        &ditto,
                        write_txn.take().expect("write txn Some"),
                    );
                    if status != 0 {
                        return Err(DittoError::from_ffi(ErrorKind::Internal));
                    }
                    Ok(diff)
                }
                i32::MIN..=-1 | 1..=i32::MAX => {
                    ffi_sdk::ditto_write_transaction_rollback(
                        &ditto,
                        write_txn.take().expect("write txn Some"),
                    );
                    Err(DittoError::from_ffi(ErrorKind::InvalidInput))
                }
            }
        } else {
            ffi_sdk::ditto_write_transaction_rollback(
                &ditto,
                write_txn.take().expect("write txn Some"),
            );
            Err(DittoError::from_ffi(ErrorKind::NonExtant))
        }
    }

    /// Replaces the matching document with the provided value.
    ///
    /// - `new_value`: A new `Serializable` which will replace the found document
    ///
    /// Note this actually follows "upsert" rules and will insert a document if no document is found
    /// with the given [`DocumentId`].
    ///
    /// # Example
    ///
    /// ```
    /// use dittolive_ditto::prelude::*;
    ///
    /// # fn example(ditto: &Ditto) -> anyhow::Result<()> {
    /// let pending_op = ditto.store().collection("cars")?.find_by_id(&b"mustang"[..]);
    /// pending_op.update_doc(&serde_json::json!({
    ///     "make": "ford",
    ///     "color": "red"
    /// }))?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn update_doc<T>(&self, new_value: &T) -> Result<(), DittoError>
    where
        T: Serialize,
    {
        let ditto = self
            .ditto
            .upgrade()
            .ok_or(crate::error::ErrorKind::ReleasedDittoInstance)?;
        // We use the doc_id to find the document (verify it exists)
        // and only if it is found, we then replace it's contents with new_value
        // then we insert this mutated document.

        let hint: Option<char_p::Ref<'_>> = None;
        let mut write_txn = Some(ffi_sdk::ditto_write_transaction(&ditto, hint).ok()?);

        // Try to find the document
        let mut document = {
            ffi_sdk::ditto_collection_get_with_write_transaction(
                &ditto,
                self.collection_name.as_ref(),
                self.doc_id.bytes.as_slice().into(),
                write_txn.as_mut().expect("write txn Some"),
            )
            .ok_or_else(|| {
                ffi_sdk::ditto_write_transaction_rollback(
                    &ditto,
                    write_txn.take().expect("write txn Some"),
                );
                ErrorKind::NonExtant
            })?
        };

        let new_content = ::serde_cbor::to_vec(new_value).unwrap();
        // REPLACE the entire document with provided value
        if ffi_sdk::ditto_document_update(&mut document, new_content.as_slice().into()) != 0 {
            ffi_sdk::ditto_write_transaction_rollback(
                &ditto,
                write_txn.take().expect("write txn Some"),
            );
            return Err(DittoError::from_ffi(ErrorKind::InvalidInput));
        }

        let code = {
            ffi_sdk::ditto_collection_update(
                &ditto,
                self.collection_name.as_ref(),
                write_txn.as_mut().expect("write txn Some"),
                document,
            )
        };
        if code != 0 {
            ffi_sdk::ditto_write_transaction_rollback(
                &ditto,
                write_txn.take().expect("write txn Some"),
            );
            Err(DittoError::from_ffi(ErrorKind::InvalidInput))
        } else {
            let status = {
                ffi_sdk::ditto_write_transaction_commit(
                    &ditto,
                    write_txn.take().expect("write txn Some"),
                )
            };
            if status != 0 {
                return Err(DittoError::from_ffi(ErrorKind::Internal));
            }
            Ok(())
        }
    }
}

// TODO(v5): remove
// Empty any() is like false, this module was giving warnings
// that couldn't be silenced and it's going to get removed anyway
#[cfg(all(test, any()))]
mod tests {
    use serde_json::json;

    use crate::{prelude::*, test_helpers::setup_ditto, utils::prelude::ErrorKind};

    #[test]
    fn test_doc_not_found() {
        let ditto = setup_ditto().unwrap();
        let store = ditto.store();
        let collection = store.collection("test").unwrap();
        let doc_id = DocumentId::new(&"test_key").unwrap();
        let doc = collection.find_by_id(doc_id).exec();

        assert!(doc.is_err());

        let e = doc.err().unwrap();
        let error_kind = e.kind();
        assert_eq!(error_kind, ErrorKind::NonExtant);
    }

    #[test]
    fn test_doc_found() {
        let ditto = setup_ditto().unwrap();
        let store = ditto.store();
        let collection = store.collection("test").unwrap();
        let doc_id = DocumentId::new(&"test_key").unwrap();
        let doc = collection.find_by_id(doc_id.clone()).exec();

        assert!(doc.is_err());

        let e = doc.err().unwrap();
        let error_kind = e.kind();
        assert_eq!(error_kind, ErrorKind::NonExtant);

        let content = json!({"hello": "again", "_id": doc_id});
        let inserted = collection.upsert(content);

        assert!(inserted.is_ok());

        let doc = collection.find_by_id(doc_id).exec();

        assert!(doc.is_ok());
    }
}