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    /// Drop a namespace, retrying briefly while its live session replica is
168    /// still closing.
169    ///
170    /// # Errors
171    ///
172    /// Returns an error if the folder namespace cannot be dropped.
173    pub async fn drop_when_ready(&self, namespace_id: NamespaceId) -> Result<()> {
174        const MAX_ATTEMPTS: u32 = 50;
175        const RETRY_DELAY: std::time::Duration = std::time::Duration::from_millis(50);
176        let mut attempts = 0_u32;
177        loop {
178            match self.drop(namespace_id).await {
179                Err(error) if attempts < MAX_ATTEMPTS && error.to_string().contains("replica is not closed") => {
180                    attempts = attempts.saturating_add(1);
181                    tokio::time::sleep(RETRY_DELAY).await;
182                }
183                result => return result,
184            }
185        }
186    }
187
188    /// Delete the local files of a materialized folder at `path`.
189    ///
190    /// Refuses to delete a filesystem root or the current working directory;
191    /// only the folder's registered path may be deleted.
192    ///
193    /// # Errors
194    ///
195    /// Returns an error if the path cannot be resolved, is unsafe, or cannot be
196    /// deleted.
197    pub async fn delete_folder_files(path: &std::path::Path) -> Result<()> {
198        let canonical = std::fs::canonicalize(path)
199            .map_err(|error| SyncwebError::operation("failed to resolve folder path for deletion", error))?;
200        let current = std::env::current_dir()
201            .map_err(|error| SyncwebError::operation("failed to resolve current directory", error))?;
202        if canonical.as_os_str().is_empty()
203            || canonical.parent().is_none_or(|parent| parent.as_os_str().is_empty())
204            || canonical == current
205        {
206            return Err(SyncwebError::operation(
207                "refusing to delete files at a filesystem root or the current directory",
208                "unsafe folder path",
209            ));
210        }
211        let metadata = tokio::fs::symlink_metadata(&canonical).await?;
212        if metadata.is_dir() {
213            tokio::fs::remove_dir_all(&canonical).await?;
214        } else if metadata.is_file() {
215            tokio::fs::remove_file(&canonical).await?;
216        } else {
217            return Err(SyncwebError::operation(
218                "refusing to delete a non-file, non-directory path",
219                "unsupported folder path",
220            ));
221        }
222        Ok(())
223    }
224
225    /// # Errors
226    ///
227    /// Returns an error if the folders cannot be listed.
228    pub async fn list(&self) -> Result<Vec<SyncwebFolder>> {
229        let mut documents = self
230            .docs_engine
231            .inner()
232            .list()
233            .await
234            .map_err(|error| SyncwebError::operation("failed to list documents", error))?;
235        let mut listed = Vec::new();
236        while let Some(document) = documents.next().await {
237            listed.push(document.map_err(|error| SyncwebError::operation("failed to read document list", error))?);
238        }
239        for (namespace_id, capability) in listed {
240            if self.folders.read().await.contains_key(&namespace_id) {
241                continue;
242            }
243            let doc = self
244                .docs_engine
245                .open(namespace_id)
246                .await?
247                .ok_or(SyncwebError::NamespaceNotAvailable)?;
248            let fallback_mode = match capability {
249                iroh_docs::CapabilityKind::Write => SyncMode::SendReceive,
250                iroh_docs::CapabilityKind::Read => SyncMode::ReceiveOnly,
251            };
252            let mode = self.mode_from_doc(&doc, fallback_mode).await?;
253            let folder = self.folder_from_doc(doc, mode).await?;
254            if let Entry::Vacant(entry) = self.folders.write().await.entry(namespace_id) {
255                entry.insert(folder);
256            }
257        }
258        Ok(self.folders.read().await.values().cloned().collect())
259    }
260
261    /// Return a managed folder, loading locally available namespaces first.
262    ///
263    /// # Errors
264    ///
265    /// Returns an error if the local namespace list cannot be read.
266    pub async fn get(&self, namespace_id: NamespaceId) -> Result<SyncwebFolder> {
267        let existing = self.folders.read().await.get(&namespace_id).cloned();
268        if let Some(folder) = existing {
269            return Ok(folder);
270        }
271        self.list()
272            .await?
273            .into_iter()
274            .find(|folder| folder.namespace_id() == namespace_id)
275            .ok_or_else(|| SyncwebError::FolderNotFound(namespace_id.to_string()))
276    }
277
278    /// # Errors
279    ///
280    /// Returns an error if the ticket cannot be generated.
281    pub async fn ticket(&self, namespace_id: NamespaceId, writable: bool) -> Result<DocTicket> {
282        let folder = self
283            .folders
284            .read()
285            .await
286            .get(&namespace_id)
287            .cloned()
288            .ok_or(SyncwebError::FolderNotFound(namespace_id.to_string()))?;
289        folder.ticket(self.endpoint_addr.clone(), writable).await
290    }
291
292    async fn folder_from_doc(&self, doc: Doc, mode: SyncMode) -> Result<SyncwebFolder> {
293        Ok(SyncwebFolder::new(
294            doc,
295            self.docs_engine.author().await?,
296            self.blob_store.clone(),
297            self.docs_engine.clone(),
298            mode,
299        ))
300    }
301
302    async fn mode_from_doc(&self, doc: &Doc, fallback: SyncMode) -> Result<SyncMode> {
303        let author = self.docs_engine.author().await?;
304        let Some(entry) = self.docs_engine.get(doc, author, MODE_KEY).await? else {
305            return Ok(fallback);
306        };
307        let mode_bytes = self.blob_store.get(entry.content_hash()).await?;
308        let mode_value = std::str::from_utf8(&mode_bytes)
309            .map_err(|error| SyncwebError::operation("folder mode metadata is not UTF-8", error))?;
310        mode_value.parse()
311    }
312}
syncweb_core/folder/manager.rs

syncweb_core/folder/
manager.rs
