syncweb_core/folder/
manager.rs

1use std::{
2    collections::{HashMap, HashSet, hash_map::Entry},
3    str::FromStr,
4    sync::Arc,
5};
6
7use iroh::PublicKey;
8use iroh_blobs::ticket::BlobTicket;
9use iroh_docs::{DocTicket, NamespaceId, api::Doc};
10use n0_future::StreamExt;
11use tokio::sync::RwLock;
12
13use crate::error::{Result, SyncwebError};
14use crate::node::discovery::TopicTracker;
15use crate::node::iroh_node::IrohNode;
16
17use super::{Capability, SyncMode, SyncwebFolder};
18
19const MODE_KEY: &[u8] = b"sys/syncweb/mode";
20
21#[derive(Clone)]
22pub struct FolderManager {
23    endpoint: iroh::Endpoint,
24    endpoint_addr: iroh::EndpointAddr,
25    node_id: PublicKey,
26    blob_store: crate::node::blob_store::BlobStore,
27    docs_engine: crate::node::docs_engine::DocsEngine,
28    folders: Arc<RwLock<HashMap<NamespaceId, SyncwebFolder>>>,
29    subscriptions: Arc<RwLock<HashSet<iroh_blobs::Hash>>>,
30    topic_tracker: TopicTracker,
31    announced: Arc<RwLock<HashSet<NamespaceId>>>,
32}
33
34impl FolderManager {
35    #[must_use]
36    pub fn new(node: &IrohNode) -> Self {
37        Self {
38            endpoint: node.endpoint().clone(),
39            endpoint_addr: node.endpoint().addr(),
40            node_id: node.endpoint().id(),
41            blob_store: node.blob_store().clone(),
42            docs_engine: node.docs_engine().clone(),
43            folders: Arc::new(RwLock::new(HashMap::new())),
44            subscriptions: Arc::new(RwLock::new(HashSet::new())),
45            topic_tracker: node.topic_tracker().clone(),
46            announced: Arc::new(RwLock::new(HashSet::new())),
47        }
48    }
49
50    /// # Errors
51    ///
52    /// Returns an error if the folder namespace cannot be created or initialized.
53    pub async fn create(&self, mode: SyncMode) -> Result<SyncwebFolder> {
54        let doc = self.docs_engine.create_namespace().await?;
55        let author = self.docs_engine.author().await?;
56        self.docs_engine.set(&doc, author, MODE_KEY, mode.to_string()).await?;
57        let folder = SyncwebFolder::new(doc, author, self.blob_store.clone(), self.docs_engine.clone(), mode);
58        folder.grant(self.node_id, Capability::Admin).await;
59        self.folders.write().await.insert(folder.namespace_id(), folder.clone());
60        self.announce_namespace(folder.namespace_id()).await;
61        Ok(folder)
62    }
63
64    /// # Errors
65    ///
66    /// Returns an error if the folder ticket cannot be joined or parsed.
67    pub async fn join(&self, ticket_str: impl AsRef<str>, mode: SyncMode) -> Result<SyncwebFolder> {
68        let mut ticket_raw = ticket_str.as_ref();
69        if let Some(rest) = ticket_raw.strip_prefix("syncweb://") {
70            if let Some((_, query)) = rest.split_once('?') {
71                for param in query.split('&') {
72                    if let Some(val) = param.strip_prefix("ticket=") {
73                        ticket_raw = val;
74                        break;
75                    }
76                }
77            } else {
78                ticket_raw = rest;
79            }
80        }
81        let ticket = DocTicket::from_str(ticket_raw).map_err(|error| SyncwebError::InvalidTicket(error.to_string()))?;
82        let doc = self.docs_engine.import_ticket(ticket).await?;
83        let folder = self.folder_from_doc(doc, mode).await?;
84        self.folders.write().await.insert(folder.namespace_id(), folder.clone());
85        self.announce_namespace(folder.namespace_id()).await;
86        Ok(folder)
87    }
88
89    /// Subscribe to a public blob ticket.
90    /// Fetches the blob and returns its hash. Does NOT create an Iroh doc
91    /// namespace.
92    ///
93    /// # Errors
94    ///
95    /// Returns an error if the blob cannot be fetched.
96    pub async fn subscribe_public(&self, ticket: &BlobTicket) -> Result<iroh_blobs::Hash> {
97        let hash = ticket.hash();
98        self.blob_store.fetch(&self.endpoint, ticket).await?;
99        self.subscriptions.write().await.insert(hash);
100        Ok(hash)
101    }
102
103    /// # Errors
104    ///
105    /// Returns an error if the folder namespace cannot be accepted.
106    pub async fn accept(&self, namespace_id: NamespaceId) -> Result<SyncwebFolder> {
107        let existing_folder = self.folders.read().await.get(&namespace_id).cloned();
108        if let Some(folder) = existing_folder {
109            return Ok(folder);
110        }
111        let doc = self
112            .docs_engine
113            .open(namespace_id)
114            .await?
115            .ok_or(SyncwebError::NamespaceNotAvailable)?;
116        let folder = self.folder_from_doc(doc, SyncMode::ReceiveOnly).await?;
117        self.folders.write().await.insert(namespace_id, folder.clone());
118        self.announce_namespace(namespace_id).await;
119        Ok(folder)
120    }
121
122    /// Announce a namespace to the topic tracker if not already announced.
123    /// Idempotent — safe to call multiple times for the same namespace.
124    pub async fn announce_namespace(&self, namespace_id: NamespaceId) {
125        if self.announced.read().await.contains(&namespace_id) {
126            return;
127        }
128        if let Err(error) = self.topic_tracker.announce(namespace_id).await {
129            tracing::warn!(%namespace_id, %error, "failed to announce folder namespace");
130        } else {
131            self.announced.write().await.insert(namespace_id);
132        }
133    }
134
135    /// Return all managed public subscription hashes.
136    #[must_use]
137    pub async fn subscriptions(&self) -> Vec<iroh_blobs::Hash> {
138        self.subscriptions.read().await.iter().copied().collect()
139    }
140
141    /// Seed the subscription set from persisted hashes (e.g. loaded from `SQLite`
142    /// on startup).
143    pub async fn seed_subscriptions(&self, hashes: impl IntoIterator<Item = iroh_blobs::Hash>) {
144        self.subscriptions.write().await.extend(hashes);
145    }
146
147    /// Drop (unsubscribe from) a public blob subscription.
148    pub async fn drop_subscription(&self, hash: &iroh_blobs::Hash) {
149        self.subscriptions.write().await.remove(hash);
150    }
151
152    /// Check whether a public subscription for the given hash exists.
153    #[must_use]
154    pub async fn has_subscription(&self, hash: &iroh_blobs::Hash) -> bool {
155        self.subscriptions.read().await.contains(hash)
156    }
157
158    /// # Errors
159    ///
160    /// Returns an error if the folder namespace cannot be dropped.
161    pub async fn drop(&self, namespace_id: NamespaceId) -> Result<()> {
162        self.docs_engine.drop_namespace(namespace_id).await?;
163        self.folders.write().await.remove(&namespace_id);
164        Ok(())
165    }
166
167    /// # Errors
168    ///
169    /// Returns an error if the folders cannot be listed.
170    pub async fn list(&self) -> Result<Vec<SyncwebFolder>> {
171        let mut documents = self
172            .docs_engine
173            .inner()
174            .list()
175            .await
176            .map_err(|error| SyncwebError::operation("failed to list documents", error))?;
177        let mut listed = Vec::new();
178        while let Some(document) = documents.next().await {
179            listed.push(document.map_err(|error| SyncwebError::operation("failed to read document list", error))?);
180        }
181        for (namespace_id, capability) in listed {
182            if self.folders.read().await.contains_key(&namespace_id) {
183                continue;
184            }
185            let doc = self
186                .docs_engine
187                .open(namespace_id)
188                .await?
189                .ok_or(SyncwebError::NamespaceNotAvailable)?;
190            let fallback_mode = match capability {
191                iroh_docs::CapabilityKind::Write => SyncMode::SendReceive,
192                iroh_docs::CapabilityKind::Read => SyncMode::ReceiveOnly,
193            };
194            let mode = self.mode_from_doc(&doc, fallback_mode).await?;
195            let folder = self.folder_from_doc(doc, mode).await?;
196            if let Entry::Vacant(entry) = self.folders.write().await.entry(namespace_id) {
197                entry.insert(folder);
198            }
199        }
200        Ok(self.folders.read().await.values().cloned().collect())
201    }
202
203    /// Return a managed folder, loading locally available namespaces first.
204    ///
205    /// # Errors
206    ///
207    /// Returns an error if the local namespace list cannot be read.
208    pub async fn get(&self, namespace_id: NamespaceId) -> Result<SyncwebFolder> {
209        let existing = self.folders.read().await.get(&namespace_id).cloned();
210        if let Some(folder) = existing {
211            return Ok(folder);
212        }
213        self.list()
214            .await?
215            .into_iter()
216            .find(|folder| folder.namespace_id() == namespace_id)
217            .ok_or_else(|| SyncwebError::FolderNotFound(namespace_id.to_string()))
218    }
219
220    /// # Errors
221    ///
222    /// Returns an error if the ticket cannot be generated.
223    pub async fn ticket(&self, namespace_id: NamespaceId, writable: bool) -> Result<DocTicket> {
224        let folder = self
225            .folders
226            .read()
227            .await
228            .get(&namespace_id)
229            .cloned()
230            .ok_or(SyncwebError::FolderNotFound(namespace_id.to_string()))?;
231        folder.ticket(self.endpoint_addr.clone(), writable).await
232    }
233
234    async fn folder_from_doc(&self, doc: Doc, mode: SyncMode) -> Result<SyncwebFolder> {
235        Ok(SyncwebFolder::new(
236            doc,
237            self.docs_engine.author().await?,
238            self.blob_store.clone(),
239            self.docs_engine.clone(),
240            mode,
241        ))
242    }
243
244    async fn mode_from_doc(&self, doc: &Doc, fallback: SyncMode) -> Result<SyncMode> {
245        let author = self.docs_engine.author().await?;
246        let Some(entry) = self.docs_engine.get(doc, author, MODE_KEY).await? else {
247            return Ok(fallback);
248        };
249        let mode_bytes = self.blob_store.get(entry.content_hash()).await?;
250        let mode_value = std::str::from_utf8(&mode_bytes)
251            .map_err(|error| SyncwebError::operation("folder mode metadata is not UTF-8", error))?;
252        mode_value.parse()
253    }
254}
syncweb_core/folder/manager.rs

syncweb_core/folder/
manager.rs
