syncweb_core/node/
docs_engine.rs

1use bytes::Bytes;
2use iroh_blobs::Hash;
3use iroh_docs::{
4    AuthorId, DocTicket, Entry, NamespaceId,
5    api::{
6        Doc,
7        protocol::{AddrInfoOptions, ShareMode},
8    },
9    engine::LiveEvent,
10    protocol::Docs,
11    store::Query,
12};
13use n0_future::StreamExt;
14
15use crate::{
16    constants::{CATALOG_METADATA_KEY, MODE_KEY, NETWORK_MEMBERS_KEY},
17    error::{Result, SyncwebError},
18};
19
20/// The kind of namespace being provisioned. Picks the `sys/` metadata key
21/// stamped on a freshly created namespace so folder, catalog, and membership
22/// provisioning share one code path.
23#[derive(Clone, Copy, Debug, Eq, PartialEq)]
24#[non_exhaustive]
25pub enum ProvisionKind {
26    Folder,
27    Catalog,
28    NetworkMembership,
29}
30
31impl ProvisionKind {
32    const fn metadata_key(self) -> &'static [u8] {
33        match self {
34            Self::Folder => MODE_KEY,
35            Self::Catalog => CATALOG_METADATA_KEY,
36            Self::NetworkMembership => NETWORK_MEMBERS_KEY,
37        }
38    }
39}
40
41/// A provisioned namespace and its shareable write ticket.
42#[derive(Clone, Debug)]
43#[non_exhaustive]
44pub struct ProvisionedDoc {
45    pub doc: Doc,
46    pub ticket: DocTicket,
47    pub namespace_id: NamespaceId,
48}
49
50#[derive(Clone)]
51pub struct DocsEngine {
52    docs: Docs,
53}
54
55impl DocsEngine {
56    #[must_use]
57    pub fn new(docs: &Docs) -> Self {
58        Self { docs: docs.clone() }
59    }
60
61    #[must_use]
62    pub const fn inner(&self) -> &Docs {
63        &self.docs
64    }
65
66    /// # Errors
67    ///
68    /// Returns an error if the namespace cannot be created.
69    pub async fn create_namespace(&self) -> Result<Doc> {
70        self.docs
71            .create()
72            .await
73            .map_err(|error| SyncwebError::operation("failed to create namespace", error))
74    }
75
76    /// Provision a namespace and a shareable ticket for it.
77    ///
78    /// With `existing == None` a fresh namespace is created; with
79    /// `existing == Some(id)` an already-local namespace is opened instead.
80    /// In both cases a write ticket is returned so the caller can hand the
81    /// namespace to a peer. Folder, catalog, and membership-doc provisioning
82    /// all funnel through this single helper.
83    ///
84    /// # Errors
85    ///
86    /// Returns an error if the namespace cannot be created, opened, or shared.
87    pub async fn create_or_open_namespace(&self, existing: Option<NamespaceId>) -> Result<(Doc, DocTicket)> {
88        let doc = match existing {
89            Some(namespace_id) => self
90                .open(namespace_id)
91                .await?
92                .ok_or(SyncwebError::NamespaceNotAvailable)?,
93            None => self.create_namespace().await?,
94        };
95        let ticket = self.share_ticket(&doc, true).await?;
96        Ok((doc, ticket))
97    }
98
99    /// Provision a namespace of the given kind and stamp its `sys/` metadata.
100    ///
101    /// Unifies folder, catalog, and network-membership namespace creation:
102    /// every kind funnels through [`Self::create_or_open_namespace`] and writes
103    /// its kind-specific metadata key under `sys/`.
104    ///
105    /// # Errors
106    ///
107    /// Returns an error if the namespace cannot be created, opened, shared, or
108    /// stamped.
109    pub async fn provision(
110        &self,
111        kind: ProvisionKind,
112        existing: Option<NamespaceId>,
113        metadata: impl AsRef<[u8]>,
114    ) -> Result<ProvisionedDoc> {
115        let (doc, ticket) = self.create_or_open_namespace(existing).await?;
116        let author = self.author().await?;
117        self.set(&doc, author, kind.metadata_key(), metadata).await?;
118        let namespace_id = self.namespace_id(&doc);
119        Ok(ProvisionedDoc {
120            doc,
121            ticket,
122            namespace_id,
123        })
124    }
125
126    /// # Errors
127    ///
128    /// Returns an error if the ticket cannot be imported.
129    pub async fn import_ticket(&self, ticket: DocTicket) -> Result<Doc> {
130        self.docs
131            .import(ticket)
132            .await
133            .map_err(|error| SyncwebError::operation("failed to import document ticket", error))
134    }
135
136    /// # Errors
137    ///
138    /// Returns an error if the document cannot be opened.
139    pub async fn open(&self, namespace_id: NamespaceId) -> Result<Option<Doc>> {
140        self.docs
141            .open(namespace_id)
142            .await
143            .map_err(|error| SyncwebError::operation("failed to open namespace", error))
144    }
145
146    /// # Errors
147    ///
148    /// Returns an error if the namespace cannot be dropped.
149    pub async fn drop_namespace(&self, namespace_id: NamespaceId) -> Result<()> {
150        self.docs
151            .drop_doc(namespace_id)
152            .await
153            .map_err(|error| SyncwebError::operation("failed to drop namespace", error))?;
154        Ok(())
155    }
156
157    /// Start live synchronization for a document.
158    ///
159    /// An empty peer list uses peers already learned by iroh-docs for the
160    /// namespace.
161    ///
162    /// # Errors
163    ///
164    /// Returns an error if the document cannot be opened for synchronization.
165    pub async fn start_sync(&self, doc: &Doc, peers: Vec<iroh::EndpointAddr>) -> Result<()> {
166        doc.start_sync(peers)
167            .await
168            .map_err(|error| SyncwebError::operation("failed to start document synchronization", error))
169    }
170
171    /// # Errors
172    ///
173    /// Returns an error if the document cannot be shared.
174    pub async fn share(&self, doc: &Doc, mode: ShareMode) -> Result<DocTicket> {
175        doc.share(mode, AddrInfoOptions::RelayAndAddresses)
176            .await
177            .map_err(|error| SyncwebError::operation("failed to share document", error))
178    }
179
180    /// Share a read or write ticket for a document.
181    ///
182    /// # Errors
183    ///
184    /// Returns an error if the document cannot be shared.
185    pub async fn share_ticket(&self, doc: &Doc, writable: bool) -> Result<DocTicket> {
186        let mode = if writable { ShareMode::Write } else { ShareMode::Read };
187        self.share(doc, mode).await
188    }
189
190    /// # Errors
191    ///
192    /// Returns an error if the default author cannot be retrieved.
193    pub async fn author(&self) -> Result<AuthorId> {
194        self.docs
195            .author_default()
196            .await
197            .map_err(|error| SyncwebError::operation("failed to retrieve document author", error))
198    }
199
200    /// # Errors
201    ///
202    /// Returns an error if the author cannot be exported.
203    pub async fn export_author(&self, author_id: AuthorId) -> Result<Option<iroh_docs::Author>> {
204        self.docs
205            .author_export(author_id)
206            .await
207            .map_err(|error| SyncwebError::operation("failed to export document author", error))
208    }
209
210    /// # Errors
211    ///
212    /// Returns an error if the author cannot be imported.
213    pub async fn import_author(&self, author: iroh_docs::Author) -> Result<AuthorId> {
214        self.docs
215            .author_import(author.clone())
216            .await
217            .map_err(|error| SyncwebError::operation("failed to import document author", error))?;
218        Ok(author.id())
219    }
220
221    /// # Errors
222    ///
223    /// Returns an error if setting the document entry fails.
224    pub async fn set(
225        &self,
226        doc: &Doc,
227        author: AuthorId,
228        key: impl AsRef<[u8]>,
229        value: impl AsRef<[u8]>,
230    ) -> Result<Hash> {
231        doc.set_bytes(
232            author,
233            Bytes::copy_from_slice(key.as_ref()),
234            Bytes::copy_from_slice(value.as_ref()),
235        )
236        .await
237        .map_err(|error| SyncwebError::operation("failed to set document entry", error))
238    }
239
240    /// # Errors
241    ///
242    /// Returns an error if setting the document blob fails.
243    pub async fn set_blob(
244        &self,
245        doc: &Doc,
246        author: AuthorId,
247        key: impl AsRef<[u8]>,
248        hash: Hash,
249        size: u64,
250    ) -> Result<()> {
251        doc.set_hash(author, Bytes::copy_from_slice(key.as_ref()), hash, size)
252            .await
253            .map_err(|error| SyncwebError::operation("failed to set document blob", error))?;
254        Ok(())
255    }
256
257    /// # Errors
258    ///
259    /// Returns an error if getting the document entry fails.
260    pub async fn get(&self, doc: &Doc, author: AuthorId, key: impl AsRef<[u8]>) -> Result<Option<Entry>> {
261        doc.get_exact(author, key, false)
262            .await
263            .map_err(|error| SyncwebError::operation("failed to get document entry", error))
264    }
265
266    /// Read the latest entry for a key regardless of which author wrote it.
267    ///
268    /// # Errors
269    ///
270    /// Returns an error if the document query fails.
271    pub async fn get_any(&self, doc: &Doc, key: impl AsRef<[u8]>) -> Result<Option<Entry>> {
272        let entries = doc
273            .get_many(Query::single_latest_per_key().key_exact(key))
274            .await
275            .map_err(|error| SyncwebError::operation("failed to query document entries", error))?;
276        tokio::pin!(entries);
277        n0_future::StreamExt::next(&mut entries)
278            .await
279            .transpose()
280            .map_err(|error| SyncwebError::operation("failed to read document query result", error))
281    }
282
283    /// Read the latest non-empty entry for every document key.
284    ///
285    /// # Errors
286    ///
287    /// Returns an error if the document query fails.
288    pub async fn list_latest(&self, doc: &Doc) -> Result<Vec<Entry>> {
289        let entries = doc
290            .get_many(Query::single_latest_per_key().build())
291            .await
292            .map_err(|error| SyncwebError::operation("failed to query document entries", error))?;
293        tokio::pin!(entries);
294        let mut output = Vec::new();
295        while let Some(entry_result) = n0_future::StreamExt::next(&mut entries).await {
296            let entry =
297                entry_result.map_err(|error| SyncwebError::operation("failed to read document entry", error))?;
298            output.push(entry);
299        }
300        Ok(output)
301    }
302
303    /// Delete the current value for a document key.
304    ///
305    /// # Errors
306    ///
307    /// Returns an error if the document entry cannot be deleted.
308    pub async fn delete(&self, doc: &Doc, author: AuthorId, key: impl AsRef<[u8]>) -> Result<()> {
309        doc.del(author, key.as_ref().to_vec())
310            .await
311            .map_err(|error| SyncwebError::operation("failed to delete document entry", error))?;
312        Ok(())
313    }
314
315    /// # Errors
316    ///
317    /// Returns an error if watching the document fails.
318    pub async fn watch(
319        &self,
320        doc: &Doc,
321    ) -> Result<impl n0_future::Stream<Item = Result<LiveEvent>> + Send + Unpin + 'static> {
322        let stream = doc
323            .subscribe()
324            .await
325            .map_err(|error| SyncwebError::operation("failed to subscribe to document", error))?;
326        Ok(stream.map(|event| event.map_err(|error| SyncwebError::operation("document event failed", error))))
327    }
328
329    /// Subscribe to a document's live events and start live synchronization.
330    ///
331    /// This is the shared prologue for every background doc consumer: catalog
332    /// subscribers, indexing watchers, and sync intents all pair
333    /// [`Self::watch`] with [`Self::start_sync`] before spawning a consumer
334    /// task over the returned stream.
335    ///
336    /// # Errors
337    ///
338    /// Returns an error if the document cannot be watched or synchronized.
339    pub async fn watch_and_sync(
340        &self,
341        doc: &Doc,
342    ) -> Result<impl n0_future::Stream<Item = Result<LiveEvent>> + Send + Unpin + 'static> {
343        let live_events = self.watch(doc).await?;
344        self.start_sync(doc, Vec::new()).await?;
345        Ok(live_events)
346    }
347
348    #[must_use]
349    pub fn namespace_id(&self, doc: &Doc) -> NamespaceId {
350        doc.id()
351    }
352}
syncweb_core/node/docs_engine.rs

syncweb_core/node/
docs_engine.rs
