syncweb_core/node/
docs_engine.rs

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

syncweb_core/node/
docs_engine.rs
