dittolive_ditto/store/attachment/
mod.rs

1//! Use attachments to sync large binary files between peers.
2
3use std::{
4    collections::HashMap,
5    path::{Path, PathBuf},
6    sync::{Arc, Weak},
7};
8
9use ffi_sdk::{BoxedAttachmentHandle, BoxedDitto};
10use safer_ffi::prelude::*;
11use serde::ser::SerializeMap;
12
13use crate::{ditto::TryUpgrade, prelude::DittoError, utils::prelude::ErrorKind};
14
15// TODO(v5): Remove pub
16#[doc(hidden)]
17pub mod fetch_event;
18pub use self::fetch_event::DittoAttachmentFetchEvent;
19
20// TODO(v5): Remove pub
21#[doc(hidden)]
22pub mod fetcher;
23pub use self::fetcher::{DittoAttachmentFetcher, FetcherVersion};
24
25// TODO(v5): Remove pub
26#[doc(hidden)]
27pub mod token;
28pub use self::token::{DittoAttachmentToken, DittoAttachmentTokenLike};
29
30#[derive(Debug)]
31/// Represents an attachment and can be used to insert the associated attachment into a document at
32/// a specific key.
33pub struct DittoAttachment {
34    id: Box<[u8]>,
35    len: u64,
36    metadata: HashMap<String, String>,
37    ditto: Weak<BoxedDitto>,
38    attachment_handle: BoxedAttachmentHandle,
39}
40
41impl DittoAttachment {
42    /// Returns the `id` of this attachment, encoded so as to be compatible with the
43    /// [`crate::store::dql::Query`] API.
44    pub fn id(&self) -> String {
45        crate::utils::base64_encode_unpadded(&self.id)
46    }
47
48    /// Returns the `len`, in bytes, of this attachment's data. Compatible with the
49    /// [`crate::store::dql::Query`] API.
50    #[allow(clippy::len_without_is_empty)]
51    pub fn len(&self) -> u64 {
52        self.len
53    }
54
55    /// Returns the metadata that was associated with this attachment file when the source peer
56    /// called [`new_attachment`][crate::store::Store::new_attachment].
57    pub fn metadata(&self) -> &HashMap<String, String> {
58        &self.metadata
59    }
60}
61
62impl serde::Serialize for DittoAttachment {
63    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
64    where
65        S: serde::Serializer,
66    {
67        let mut map = serializer.serialize_map(Some(4))?;
68        map.serialize_entry(
69            // This breaks compatibility with the Rust SDK prior
70            // to versions 3.0.8 and 4.0.0. It is necessary for
71            // cross-SDK attachments compatibility.
72            "_ditto_internal_type_jkb12973t4b",
73            &(::ffi_sdk::DittoCrdtType::Attachment as u64),
74        )?;
75        map.serialize_entry("_id", ::serde_bytes::Bytes::new(&self.id[..]))?;
76        map.serialize_entry("_len", &self.len)?;
77        map.serialize_entry("_meta", &self.metadata)?;
78        map.end()
79    }
80}
81
82impl DittoAttachment {
83    /// Create a new DittoAttachment
84    pub(crate) fn new(
85        id: Box<[u8]>,
86        len: u64,
87        metadata: HashMap<String, String>,
88        ditto: Weak<BoxedDitto>,
89        attachment_handle: BoxedAttachmentHandle,
90    ) -> Self {
91        Self {
92            id,
93            len,
94            metadata,
95            ditto,
96            attachment_handle,
97        }
98    }
99
100    pub(crate) fn from_file_and_metadata(
101        filepath: &(impl ?Sized + AsRef<Path>),
102        metadata: HashMap<String, String>,
103        ditto: &Arc<ffi_sdk::BoxedDitto>,
104    ) -> Result<DittoAttachment, DittoError> {
105        let source_path = char_p::new(filepath.as_ref().to_str().unwrap());
106        let file_operation = ffi_sdk::AttachmentFileOperation::Copy;
107        let mut slot = ::core::mem::MaybeUninit::<ffi_sdk::Attachment>::uninit();
108        let status = {
109            ffi_sdk::ditto_new_attachment_from_file(
110                ditto,
111                source_path.as_ref(),
112                file_operation,
113                slot.as_out(),
114            )
115        };
116        if status != 0 {
117            Err(DittoError::from_ffi(ErrorKind::InvalidInput))
118        } else {
119            let attachment = unsafe { slot.assume_init() }; // safe assuming above ffi call was successful
120            let ret = DittoAttachment::new(
121                attachment.id.into(),
122                attachment.len,
123                metadata,
124                Arc::downgrade(ditto),
125                attachment.handle,
126            );
127            Ok(ret)
128        }
129    }
130
131    pub(crate) fn from_bytes_and_metadata(
132        bytes: &(impl ?Sized + AsRef<[u8]>),
133        metadata: HashMap<String, String>,
134        ditto: &Arc<ffi_sdk::BoxedDitto>,
135    ) -> Result<DittoAttachment, DittoError> {
136        let mut slot = ::core::mem::MaybeUninit::<ffi_sdk::Attachment>::uninit();
137        let status = {
138            ffi_sdk::ditto_new_attachment_from_bytes(ditto, bytes.as_ref().into(), slot.as_out())
139        };
140        if status != 0 {
141            Err(DittoError::from_ffi(ErrorKind::InvalidInput))
142        } else {
143            let attachment = unsafe { slot.assume_init() }; // safe assuming above ffi call was successful
144            let ret = DittoAttachment::new(
145                attachment.id.into(),
146                attachment.len,
147                metadata,
148                Arc::downgrade(ditto),
149                attachment.handle,
150            );
151            Ok(ret)
152        }
153    }
154
155    /// Create a new DittoAttachment from a Token
156    pub(crate) fn new_with_token(
157        token: DittoAttachmentToken,
158        ditto: Weak<BoxedDitto>,
159        attachment_handle: BoxedAttachmentHandle,
160    ) -> Self {
161        Self {
162            id: token.id,
163            len: token.len,
164            metadata: token.metadata,
165            ditto,
166            attachment_handle,
167        }
168    }
169
170    /// Return path to an attachment.
171    /// # Panics
172    /// Panics if Ditto has been released
173    pub fn path(&self) -> PathBuf {
174        // FIXME(Ronan) Ideally wrap this function in a Result
175        let ditto = self.ditto.try_upgrade().unwrap();
176        let p = ffi_sdk::ditto_get_complete_attachment_path(&ditto, &self.attachment_handle);
177        let p_string = p.to_string();
178        p_string.into()
179    }
180}
181
182#[cfg(test)]
183#[allow(deprecated)]
184// TODO(v5): remove query_builder tests
185mod tests {
186    use std::{
187        collections::HashMap,
188        sync::{
189            atomic::{AtomicBool, Ordering},
190            Arc, Mutex,
191        },
192    };
193
194    use serde_json::json;
195
196    use crate::{
197        prelude::*,
198        store::{
199            ditto_attachment_fetch_event::DittoAttachmentFetchEvent,
200            ditto_attachment_token::DittoAttachmentToken,
201        },
202        test_helpers::setup_ditto,
203    };
204
205    #[::tokio::test]
206    async fn attachment_serialize() {
207        let ditto = setup_ditto().unwrap();
208        let store = ditto.store();
209        let collection = store.collection("test").unwrap();
210
211        let original_test_file_path = "tests/data/attachment_file_1.txt";
212
213        let metadata = {
214            let mut m = HashMap::new();
215            m.insert("key_1".to_string(), "value_1".to_string());
216            m.insert("key_2".to_string(), "value_2".to_string());
217            m
218        };
219
220        let attachment = store
221            .new_attachment(original_test_file_path, metadata.clone())
222            .await
223            .expect("new_attachment");
224        let attachment_id = attachment.id.clone();
225        let attachment_len = attachment.len;
226        let attachment_file_path = attachment.path();
227        assert_ne!(
228            original_test_file_path,
229            attachment_file_path
230                .clone()
231                .into_os_string()
232                .into_string()
233                .unwrap()
234        );
235
236        // We want to test that you can upsert with an attachment when the document's contents is
237        // provided as JSON or as CBOR (so that the attachment's ID gets serialized as an array or a
238        // binary blob respectively)
239        let id = collection
240            .upsert(json!({ "hello": "again", "att": attachment }))
241            .unwrap();
242
243        {
244            let mut map = HashMap::new();
245            map.insert("hello", serde_cbor::value::to_value("again").unwrap());
246            map.insert("att", serde_cbor::value::to_value(&attachment).unwrap());
247            let _other = collection.upsert(map).unwrap();
248        }
249
250        let mut doc = collection.find_by_id(id).exec().unwrap();
251
252        // TODO(Ham): We should not be able to call `set` on a document returned by a call to `exec`
253        let set = doc.set("att_two", &attachment);
254        assert!(set.is_ok());
255
256        let attachment_token = doc.get::<DittoAttachmentToken>("att").unwrap();
257        assert_eq!(attachment_token.id, attachment_id);
258        assert_eq!(attachment_token.len, attachment_len);
259        assert_eq!(attachment_token.metadata, metadata);
260
261        let attachment_token_two = doc.get::<DittoAttachmentToken>("att_two").unwrap();
262        assert_eq!(attachment_token.id, attachment_token_two.id);
263        assert_eq!(attachment_token.len, attachment_token_two.len);
264        assert_eq!(attachment_token.metadata, attachment_token_two.metadata);
265
266        let test_file = std::fs::read(original_test_file_path).unwrap();
267        let attachment_file = std::fs::read(attachment_file_path).unwrap();
268
269        assert_eq!(test_file, attachment_file);
270
271        assert_eq!(test_file.len() as u64, attachment_len);
272    }
273
274    #[test]
275    fn attachment_fetch_legacy() {
276        #![allow(deprecated)]
277        let ditto = setup_ditto().unwrap();
278        let store = ditto.store();
279        let collection = store.collection("test").unwrap();
280
281        let original_test_file_path = "tests/data/attachment_file_1.txt";
282
283        let attachment = collection
284            .new_attachment(original_test_file_path, HashMap::new())
285            .expect("new_attachment");
286
287        let collection = store.collection("test").unwrap();
288        let id = collection.upsert(json!({"hello": "again"})).unwrap();
289        let mut doc = collection.find_by_id(id).exec().unwrap();
290
291        let set = doc.set("att", attachment);
292        assert!(set.is_ok());
293
294        let attachment_token = doc.get::<DittoAttachmentToken>("att").unwrap();
295
296        let finished = Arc::new(AtomicBool::new(false));
297        let finished_clone = Arc::clone(&finished);
298        let fetched_attachment_data: Arc<Mutex<Vec<u8>>> = Arc::new(Mutex::new(vec![]));
299
300        let _fetcher = collection
301            .fetch_attachment(attachment_token, |event| {
302                if let DittoAttachmentFetchEvent::Completed { attachment } = event {
303                    let att_data_mtx = &*fetched_attachment_data; // move (copy) and reborrow
304                    if let Ok(mut fetched_attachment_data) = att_data_mtx.lock() {
305                        *fetched_attachment_data = std::fs::read(attachment.path()).unwrap();
306                        finished_clone.store(true, Ordering::SeqCst);
307                    }
308                }
309            })
310            .unwrap();
311
312        while !finished.load(Ordering::SeqCst) {
313            std::thread::yield_now();
314        }
315
316        let test_file_data = std::fs::read(original_test_file_path).unwrap();
317        let fetched_att_data = fetched_attachment_data.lock().unwrap();
318        assert_eq!(test_file_data, *fetched_att_data);
319    }
320
321    #[::tokio::test]
322    async fn attachment_fetch() {
323        let ditto = setup_ditto().unwrap();
324        let store = ditto.store();
325        let collection = store.collection("test").unwrap();
326
327        let original_test_file_path = "tests/data/attachment_file_1.txt";
328
329        let attachment = store
330            .new_attachment(original_test_file_path, HashMap::new())
331            .await
332            .expect("new_attachment");
333
334        let id = collection.upsert(json!({"hello": "again"})).unwrap();
335        let mut doc = collection.find_by_id(id).exec().unwrap();
336
337        let set = doc.set("att", attachment);
338        assert!(set.is_ok());
339
340        let attachment_token = doc.get::<DittoAttachmentToken>("att").unwrap();
341
342        let finished = Arc::new(AtomicBool::new(false));
343        let finished_clone = Arc::clone(&finished);
344        let fetched_attachment_data: Arc<Mutex<Vec<u8>>> = Arc::new(Mutex::new(vec![]));
345
346        let _fetcher = store
347            .fetch_attachment(attachment_token, {
348                let fetched_attachment_data = fetched_attachment_data.clone();
349                move |event| {
350                    if let DittoAttachmentFetchEvent::Completed { attachment } = event {
351                        let att_data_mtx = &*fetched_attachment_data; // move (copy) and reborrow
352                        if let Ok(mut fetched_attachment_data) = att_data_mtx.lock() {
353                            *fetched_attachment_data = std::fs::read(attachment.path()).unwrap();
354                            finished_clone.store(true, Ordering::SeqCst);
355                        }
356                    }
357                }
358            })
359            .unwrap();
360
361        while !finished.load(Ordering::SeqCst) {
362            std::thread::yield_now();
363        }
364
365        let test_file_data = std::fs::read(original_test_file_path).unwrap();
366        let fetched_att_data = fetched_attachment_data.lock().unwrap();
367        assert_eq!(test_file_data, *fetched_att_data);
368    }
369}
dittolive_ditto/store/attachment/mod.rs

dittolive_ditto/store/attachment/
mod.rs
