syncweb_core/folder/
manager.rs

1use std::{
2    collections::{HashMap, HashSet, hash_map::Entry},
3    sync::Arc,
4};
5
6use iroh::PublicKey;
7use iroh_blobs::ticket::BlobTicket;
8use iroh_docs::{DocTicket, NamespaceId, api::Doc};
9use n0_future::StreamExt;
10use tokio::sync::RwLock;
11
12use crate::constants::{MODE_KEY, NETWORK_MEMBERS_KEY};
13use crate::error::{Result, SyncwebError};
14use crate::node::discovery::TopicTracker;
15use crate::node::iroh_node::IrohNode;
16
17use super::{Capability, SyncMode, SyncwebFolder};
18
19#[derive(Clone)]
20pub struct FolderManager {
21    endpoint: iroh::Endpoint,
22    node_id: PublicKey,
23    blob_store: crate::node::blob_store::BlobStore,
24    docs_engine: crate::node::docs_engine::DocsEngine,
25    folders: Arc<RwLock<HashMap<NamespaceId, SyncwebFolder>>>,
26    subscriptions: Arc<RwLock<HashSet<iroh_blobs::Hash>>>,
27    topic_tracker: TopicTracker,
28    announced: Arc<RwLock<HashSet<NamespaceId>>>,
29}
30
31impl FolderManager {
32    #[must_use]
33    pub fn new(node: &IrohNode) -> Self {
34        Self {
35            endpoint: node.endpoint().clone(),
36            node_id: node.endpoint().id(),
37            blob_store: node.blob_store().clone(),
38            docs_engine: node.docs_engine().clone(),
39            folders: Arc::new(RwLock::new(HashMap::new())),
40            subscriptions: Arc::new(RwLock::new(HashSet::new())),
41            topic_tracker: node.topic_tracker().clone(),
42            announced: Arc::new(RwLock::new(HashSet::new())),
43        }
44    }
45
46    /// # Errors
47    ///
48    /// Returns an error if the folder namespace cannot be created or initialized.
49    pub async fn create(&self, mode: SyncMode) -> Result<SyncwebFolder> {
50        let provisioned = self
51            .docs_engine
52            .provision(crate::node::docs_engine::ProvisionKind::Folder, None, mode.to_string())
53            .await?;
54        let author = self.docs_engine.author().await?;
55        let folder = SyncwebFolder::new(
56            provisioned.doc,
57            author,
58            self.blob_store.clone(),
59            self.docs_engine.clone(),
60            mode,
61        );
62        folder.grant(self.node_id, Capability::Admin).await;
63        self.folders.write().await.insert(folder.namespace_id(), folder.clone());
64        self.announce_namespace(folder.namespace_id()).await;
65        Ok(folder)
66    }
67
68    /// # Errors
69    ///
70    /// Returns an error if the folder ticket cannot be joined or parsed.
71    pub async fn join(&self, ticket_str: impl AsRef<str>, mode: SyncMode) -> Result<SyncwebFolder> {
72        self.join_with_options(ticket_str, mode, false).await
73    }
74
75    /// Join a folder, optionally in metadata-only mode (doc entries sync but
76    /// content blobs stay out of the local store until an explicit download).
77    ///
78    /// The download policy is applied before the namespace is announced, so
79    /// peers never push content for a metadata-only folder.
80    ///
81    /// # Errors
82    ///
83    /// Returns an error if the folder ticket cannot be joined or parsed, or the
84    /// download policy cannot be set.
85    pub async fn join_with_options(
86        &self,
87        ticket_str: impl AsRef<str>,
88        mode: SyncMode,
89        metadata_only: bool,
90    ) -> Result<SyncwebFolder> {
91        let ticket = crate::uri::parse_folder_ticket(ticket_str.as_ref())?;
92        let doc = self.docs_engine.import_ticket(ticket).await?;
93        let folder = self.folder_from_doc(doc, mode).await?;
94        if metadata_only {
95            folder.set_metadata_only().await?;
96        }
97        self.folders.write().await.insert(folder.namespace_id(), folder.clone());
98        self.announce_namespace(folder.namespace_id()).await;
99        Ok(folder)
100    }
101
102    /// Subscribe to a public blob ticket.
103    /// Fetches the blob and returns its hash. Does NOT create an Iroh doc
104    /// namespace.
105    ///
106    /// # Errors
107    ///
108    /// Returns an error if the blob cannot be fetched.
109    pub async fn subscribe_public(&self, ticket: &BlobTicket) -> Result<iroh_blobs::Hash> {
110        let hash = ticket.hash();
111        self.blob_store.fetch(&self.endpoint, ticket).await?;
112        self.subscriptions.write().await.insert(hash);
113        Ok(hash)
114    }
115
116    /// # Errors
117    ///
118    /// Returns an error if the folder namespace cannot be accepted.
119    pub async fn accept(&self, namespace_id: NamespaceId) -> Result<SyncwebFolder> {
120        let existing_folder = self.folders.read().await.get(&namespace_id).cloned();
121        if let Some(folder) = existing_folder {
122            return Ok(folder);
123        }
124        let (doc, _ticket) = self.docs_engine.create_or_open_namespace(Some(namespace_id)).await?;
125        let folder = self.folder_from_doc(doc, SyncMode::ReceiveOnly).await?;
126        self.folders.write().await.insert(namespace_id, folder.clone());
127        self.announce_namespace(namespace_id).await;
128        Ok(folder)
129    }
130
131    /// Announce a namespace to the topic tracker if not already announced.
132    /// Idempotent — safe to call multiple times for the same namespace.
133    pub async fn announce_namespace(&self, namespace_id: NamespaceId) {
134        if self.announced.read().await.contains(&namespace_id) {
135            return;
136        }
137        if let Err(error) = self.topic_tracker.announce(namespace_id).await {
138            tracing::warn!(%namespace_id, %error, "failed to announce folder namespace");
139        } else {
140            self.announced.write().await.insert(namespace_id);
141        }
142    }
143
144    /// Return all managed public subscription hashes.
145    #[must_use]
146    pub async fn subscriptions(&self) -> Vec<iroh_blobs::Hash> {
147        self.subscriptions.read().await.iter().copied().collect()
148    }
149
150    /// Seed the subscription set from persisted hashes (e.g. loaded from `SQLite`
151    /// on startup).
152    pub async fn seed_subscriptions(&self, hashes: impl IntoIterator<Item = iroh_blobs::Hash>) {
153        self.subscriptions.write().await.extend(hashes);
154    }
155
156    /// Drop (unsubscribe from) a public blob subscription.
157    pub async fn drop_subscription(&self, hash: &iroh_blobs::Hash) {
158        self.subscriptions.write().await.remove(hash);
159    }
160
161    /// Check whether a public subscription for the given hash exists.
162    #[must_use]
163    pub async fn has_subscription(&self, hash: &iroh_blobs::Hash) -> bool {
164        self.subscriptions.read().await.contains(hash)
165    }
166
167    /// # Errors
168    ///
169    /// Returns an error if the folder namespace cannot be dropped.
170    pub async fn drop(&self, namespace_id: NamespaceId) -> Result<()> {
171        self.docs_engine.drop_namespace(namespace_id).await?;
172        self.folders.write().await.remove(&namespace_id);
173        Ok(())
174    }
175
176    /// Drop a namespace, retrying briefly while its live session replica is
177    /// still closing.
178    ///
179    /// # Errors
180    ///
181    /// Returns an error if the folder namespace cannot be dropped.
182    pub async fn drop_when_ready(&self, namespace_id: NamespaceId) -> Result<()> {
183        const MAX_ATTEMPTS: u32 = 50;
184        const RETRY_DELAY: std::time::Duration = std::time::Duration::from_millis(50);
185        let mut attempts = 0_u32;
186        loop {
187            match self.drop(namespace_id).await {
188                Err(error) if attempts < MAX_ATTEMPTS && error.to_string().contains("replica is not closed") => {
189                    attempts = attempts.saturating_add(1);
190                    tokio::time::sleep(RETRY_DELAY).await;
191                }
192                result => return result,
193            }
194        }
195    }
196
197    /// Delete the local files of a materialized folder at `path`.
198    ///
199    /// Refuses to delete a filesystem root or the current working directory;
200    /// only the folder's registered path may be deleted.
201    ///
202    /// # Errors
203    ///
204    /// Returns an error if the path cannot be resolved, is unsafe, or cannot be
205    /// deleted.
206    pub async fn delete_folder_files(path: &std::path::Path) -> Result<()> {
207        let canonical = std::fs::canonicalize(path)
208            .map_err(|error| SyncwebError::operation("failed to resolve folder path for deletion", error))?;
209        let current = std::env::current_dir()
210            .map_err(|error| SyncwebError::operation("failed to resolve current directory", error))?;
211        if canonical.as_os_str().is_empty()
212            || canonical.parent().is_none_or(|parent| parent.as_os_str().is_empty())
213            || canonical == current
214        {
215            return Err(SyncwebError::operation(
216                "refusing to delete files at a filesystem root or the current directory",
217                "unsafe folder path",
218            ));
219        }
220        let metadata = tokio::fs::symlink_metadata(&canonical).await?;
221        if metadata.is_dir() {
222            tokio::fs::remove_dir_all(&canonical).await?;
223        } else if metadata.is_file() {
224            tokio::fs::remove_file(&canonical).await?;
225        } else {
226            return Err(SyncwebError::operation(
227                "refusing to delete a non-file, non-directory path",
228                "unsupported folder path",
229            ));
230        }
231        Ok(())
232    }
233
234    /// # Errors
235    ///
236    /// Returns an error if the folders cannot be listed.
237    pub async fn list(&self) -> Result<Vec<SyncwebFolder>> {
238        let mut documents = self
239            .docs_engine
240            .inner()
241            .list()
242            .await
243            .map_err(|error| SyncwebError::operation("failed to list documents", error))?;
244        let mut listed = Vec::new();
245        while let Some(document) = documents.next().await {
246            listed.push(document.map_err(|error| SyncwebError::operation("failed to read document list", error))?);
247        }
248        for (namespace_id, capability) in listed {
249            if self.folders.read().await.contains_key(&namespace_id) {
250                continue;
251            }
252            let doc = self
253                .docs_engine
254                .open(namespace_id)
255                .await?
256                .ok_or(SyncwebError::NamespaceNotAvailable)?;
257            if self.docs_engine.get_any(&doc, NETWORK_MEMBERS_KEY).await?.is_some() {
258                continue;
259            }
260            let fallback_mode = match capability {
261                iroh_docs::CapabilityKind::Write => SyncMode::SendReceive,
262                iroh_docs::CapabilityKind::Read => SyncMode::ReceiveOnly,
263            };
264            let mode = self.mode_from_doc(&doc, fallback_mode).await?;
265            let folder = self.folder_from_doc(doc, mode).await?;
266            if let Entry::Vacant(entry) = self.folders.write().await.entry(namespace_id) {
267                entry.insert(folder);
268            }
269        }
270        Ok(self.folders.read().await.values().cloned().collect())
271    }
272
273    /// Return a managed folder, loading locally available namespaces first.
274    ///
275    /// # Errors
276    ///
277    /// Returns an error if the local namespace list cannot be read.
278    pub async fn get(&self, namespace_id: NamespaceId) -> Result<SyncwebFolder> {
279        let existing = self.folders.read().await.get(&namespace_id).cloned();
280        if let Some(folder) = existing {
281            return Ok(folder);
282        }
283        self.list()
284            .await?
285            .into_iter()
286            .find(|folder| folder.namespace_id() == namespace_id)
287            .ok_or_else(|| SyncwebError::FolderNotFound(namespace_id.to_string()))
288    }
289
290    /// Resolve a folder selector (namespace ID, managed path, or the sole managed
291    /// folder) to a folder. Mirrors the CLI's path-or-namespace resolution so the
292    /// logic lives in one place.
293    ///
294    /// # Errors
295    ///
296    /// Returns an error when no folder matches the selector or the selector is
297    /// ambiguous.
298    pub async fn resolve(&self, selector: &std::path::Path) -> Result<SyncwebFolder> {
299        let namespace_id = self.resolve_namespace(&selector.to_string_lossy()).await?;
300        self.get(namespace_id).await
301    }
302
303    /// Resolve a string selector to a folder namespace ID. Accepts a namespace ID,
304    /// an existing managed path, or falls back to the sole managed folder.
305    ///
306    /// # Errors
307    ///
308    /// Returns an error when the selector cannot be resolved to a namespace.
309    pub async fn resolve_namespace(&self, selector: &str) -> Result<NamespaceId> {
310        if let Ok(namespace_id) = selector.parse() {
311            return Ok(namespace_id);
312        }
313        let folders = self.list().await?;
314        match folders.as_slice() {
315            [folder] => Ok(folder.namespace_id()),
316            [] => Err(SyncwebError::NoFolders),
317            _ => Err(SyncwebError::AmbiguousFolderSelector(selector.to_owned())),
318        }
319    }
320
321    /// # Errors
322    ///
323    /// Returns an error if the ticket cannot be generated.
324    pub async fn ticket(&self, namespace_id: NamespaceId, writable: bool) -> Result<DocTicket> {
325        let folder = self
326            .folders
327            .read()
328            .await
329            .get(&namespace_id)
330            .cloned()
331            .ok_or(SyncwebError::FolderNotFound(namespace_id.to_string()))?;
332        folder.ticket(writable).await
333    }
334
335    async fn folder_from_doc(&self, doc: Doc, mode: SyncMode) -> Result<SyncwebFolder> {
336        Ok(SyncwebFolder::new(
337            doc,
338            self.docs_engine.author().await?,
339            self.blob_store.clone(),
340            self.docs_engine.clone(),
341            mode,
342        ))
343    }
344
345    async fn mode_from_doc(&self, doc: &Doc, fallback: SyncMode) -> Result<SyncMode> {
346        let author = self.docs_engine.author().await?;
347        let Some(entry) = self.docs_engine.get(doc, author, MODE_KEY).await? else {
348            return Ok(fallback);
349        };
350        let mode_bytes = self.blob_store.get(entry.content_hash()).await?;
351        let mode_value = std::str::from_utf8(&mode_bytes)
352            .map_err(|error| SyncwebError::operation("folder mode metadata is not UTF-8", error))?;
353        mode_value.parse()
354    }
355}
syncweb_core/folder/manager.rs

syncweb_core/folder/
manager.rs
