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        let ticket = crate::uri::parse_folder_ticket(ticket_str.as_ref())?;
73        let doc = self.docs_engine.import_ticket(ticket).await?;
74        let folder = self.folder_from_doc(doc, mode).await?;
75        self.folders.write().await.insert(folder.namespace_id(), folder.clone());
76        self.announce_namespace(folder.namespace_id()).await;
77        Ok(folder)
78    }
79
80    /// Subscribe to a public blob ticket.
81    /// Fetches the blob and returns its hash. Does NOT create an Iroh doc
82    /// namespace.
83    ///
84    /// # Errors
85    ///
86    /// Returns an error if the blob cannot be fetched.
87    pub async fn subscribe_public(&self, ticket: &BlobTicket) -> Result<iroh_blobs::Hash> {
88        let hash = ticket.hash();
89        self.blob_store.fetch(&self.endpoint, ticket).await?;
90        self.subscriptions.write().await.insert(hash);
91        Ok(hash)
92    }
93
94    /// # Errors
95    ///
96    /// Returns an error if the folder namespace cannot be accepted.
97    pub async fn accept(&self, namespace_id: NamespaceId) -> Result<SyncwebFolder> {
98        let existing_folder = self.folders.read().await.get(&namespace_id).cloned();
99        if let Some(folder) = existing_folder {
100            return Ok(folder);
101        }
102        let (doc, _ticket) = self.docs_engine.create_or_open_namespace(Some(namespace_id)).await?;
103        let folder = self.folder_from_doc(doc, SyncMode::ReceiveOnly).await?;
104        self.folders.write().await.insert(namespace_id, folder.clone());
105        self.announce_namespace(namespace_id).await;
106        Ok(folder)
107    }
108
109    /// Announce a namespace to the topic tracker if not already announced.
110    /// Idempotent — safe to call multiple times for the same namespace.
111    pub async fn announce_namespace(&self, namespace_id: NamespaceId) {
112        if self.announced.read().await.contains(&namespace_id) {
113            return;
114        }
115        if let Err(error) = self.topic_tracker.announce(namespace_id).await {
116            tracing::warn!(%namespace_id, %error, "failed to announce folder namespace");
117        } else {
118            self.announced.write().await.insert(namespace_id);
119        }
120    }
121
122    /// Return all managed public subscription hashes.
123    #[must_use]
124    pub async fn subscriptions(&self) -> Vec<iroh_blobs::Hash> {
125        self.subscriptions.read().await.iter().copied().collect()
126    }
127
128    /// Seed the subscription set from persisted hashes (e.g. loaded from `SQLite`
129    /// on startup).
130    pub async fn seed_subscriptions(&self, hashes: impl IntoIterator<Item = iroh_blobs::Hash>) {
131        self.subscriptions.write().await.extend(hashes);
132    }
133
134    /// Drop (unsubscribe from) a public blob subscription.
135    pub async fn drop_subscription(&self, hash: &iroh_blobs::Hash) {
136        self.subscriptions.write().await.remove(hash);
137    }
138
139    /// Check whether a public subscription for the given hash exists.
140    #[must_use]
141    pub async fn has_subscription(&self, hash: &iroh_blobs::Hash) -> bool {
142        self.subscriptions.read().await.contains(hash)
143    }
144
145    /// # Errors
146    ///
147    /// Returns an error if the folder namespace cannot be dropped.
148    pub async fn drop(&self, namespace_id: NamespaceId) -> Result<()> {
149        self.docs_engine.drop_namespace(namespace_id).await?;
150        self.folders.write().await.remove(&namespace_id);
151        Ok(())
152    }
153
154    /// Drop a namespace, retrying briefly while its live session replica is
155    /// still closing.
156    ///
157    /// # Errors
158    ///
159    /// Returns an error if the folder namespace cannot be dropped.
160    pub async fn drop_when_ready(&self, namespace_id: NamespaceId) -> Result<()> {
161        const MAX_ATTEMPTS: u32 = 50;
162        const RETRY_DELAY: std::time::Duration = std::time::Duration::from_millis(50);
163        let mut attempts = 0_u32;
164        loop {
165            match self.drop(namespace_id).await {
166                Err(error) if attempts < MAX_ATTEMPTS && error.to_string().contains("replica is not closed") => {
167                    attempts = attempts.saturating_add(1);
168                    tokio::time::sleep(RETRY_DELAY).await;
169                }
170                result => return result,
171            }
172        }
173    }
174
175    /// Delete the local files of a materialized folder at `path`.
176    ///
177    /// Refuses to delete a filesystem root or the current working directory;
178    /// only the folder's registered path may be deleted.
179    ///
180    /// # Errors
181    ///
182    /// Returns an error if the path cannot be resolved, is unsafe, or cannot be
183    /// deleted.
184    pub async fn delete_folder_files(path: &std::path::Path) -> Result<()> {
185        let canonical = std::fs::canonicalize(path)
186            .map_err(|error| SyncwebError::operation("failed to resolve folder path for deletion", error))?;
187        let current = std::env::current_dir()
188            .map_err(|error| SyncwebError::operation("failed to resolve current directory", error))?;
189        if canonical.as_os_str().is_empty()
190            || canonical.parent().is_none_or(|parent| parent.as_os_str().is_empty())
191            || canonical == current
192        {
193            return Err(SyncwebError::operation(
194                "refusing to delete files at a filesystem root or the current directory",
195                "unsafe folder path",
196            ));
197        }
198        let metadata = tokio::fs::symlink_metadata(&canonical).await?;
199        if metadata.is_dir() {
200            tokio::fs::remove_dir_all(&canonical).await?;
201        } else if metadata.is_file() {
202            tokio::fs::remove_file(&canonical).await?;
203        } else {
204            return Err(SyncwebError::operation(
205                "refusing to delete a non-file, non-directory path",
206                "unsupported folder path",
207            ));
208        }
209        Ok(())
210    }
211
212    /// # Errors
213    ///
214    /// Returns an error if the folders cannot be listed.
215    pub async fn list(&self) -> Result<Vec<SyncwebFolder>> {
216        let mut documents = self
217            .docs_engine
218            .inner()
219            .list()
220            .await
221            .map_err(|error| SyncwebError::operation("failed to list documents", error))?;
222        let mut listed = Vec::new();
223        while let Some(document) = documents.next().await {
224            listed.push(document.map_err(|error| SyncwebError::operation("failed to read document list", error))?);
225        }
226        for (namespace_id, capability) in listed {
227            if self.folders.read().await.contains_key(&namespace_id) {
228                continue;
229            }
230            let doc = self
231                .docs_engine
232                .open(namespace_id)
233                .await?
234                .ok_or(SyncwebError::NamespaceNotAvailable)?;
235            if self.docs_engine.get_any(&doc, NETWORK_MEMBERS_KEY).await?.is_some() {
236                continue;
237            }
238            let fallback_mode = match capability {
239                iroh_docs::CapabilityKind::Write => SyncMode::SendReceive,
240                iroh_docs::CapabilityKind::Read => SyncMode::ReceiveOnly,
241            };
242            let mode = self.mode_from_doc(&doc, fallback_mode).await?;
243            let folder = self.folder_from_doc(doc, mode).await?;
244            if let Entry::Vacant(entry) = self.folders.write().await.entry(namespace_id) {
245                entry.insert(folder);
246            }
247        }
248        Ok(self.folders.read().await.values().cloned().collect())
249    }
250
251    /// Return a managed folder, loading locally available namespaces first.
252    ///
253    /// # Errors
254    ///
255    /// Returns an error if the local namespace list cannot be read.
256    pub async fn get(&self, namespace_id: NamespaceId) -> Result<SyncwebFolder> {
257        let existing = self.folders.read().await.get(&namespace_id).cloned();
258        if let Some(folder) = existing {
259            return Ok(folder);
260        }
261        self.list()
262            .await?
263            .into_iter()
264            .find(|folder| folder.namespace_id() == namespace_id)
265            .ok_or_else(|| SyncwebError::FolderNotFound(namespace_id.to_string()))
266    }
267
268    /// Resolve a folder selector (namespace ID, managed path, or the sole managed
269    /// folder) to a folder. Mirrors the CLI's path-or-namespace resolution so the
270    /// logic lives in one place.
271    ///
272    /// # Errors
273    ///
274    /// Returns an error when no folder matches the selector or the selector is
275    /// ambiguous.
276    pub async fn resolve(&self, selector: &std::path::Path) -> Result<SyncwebFolder> {
277        let namespace_id = self.resolve_namespace(&selector.to_string_lossy()).await?;
278        self.get(namespace_id).await
279    }
280
281    /// Resolve a string selector to a folder namespace ID. Accepts a namespace ID,
282    /// an existing managed path, or falls back to the sole managed folder.
283    ///
284    /// # Errors
285    ///
286    /// Returns an error when the selector cannot be resolved to a namespace.
287    pub async fn resolve_namespace(&self, selector: &str) -> Result<NamespaceId> {
288        if let Ok(namespace_id) = selector.parse() {
289            return Ok(namespace_id);
290        }
291        let folders = self.list().await?;
292        match folders.as_slice() {
293            [folder] => Ok(folder.namespace_id()),
294            [] => Err(SyncwebError::NoFolders),
295            _ => Err(SyncwebError::AmbiguousFolderSelector(selector.to_owned())),
296        }
297    }
298
299    /// # Errors
300    ///
301    /// Returns an error if the ticket cannot be generated.
302    pub async fn ticket(&self, namespace_id: NamespaceId, writable: bool) -> Result<DocTicket> {
303        let folder = self
304            .folders
305            .read()
306            .await
307            .get(&namespace_id)
308            .cloned()
309            .ok_or(SyncwebError::FolderNotFound(namespace_id.to_string()))?;
310        folder.ticket(writable).await
311    }
312
313    async fn folder_from_doc(&self, doc: Doc, mode: SyncMode) -> Result<SyncwebFolder> {
314        Ok(SyncwebFolder::new(
315            doc,
316            self.docs_engine.author().await?,
317            self.blob_store.clone(),
318            self.docs_engine.clone(),
319            mode,
320        ))
321    }
322
323    async fn mode_from_doc(&self, doc: &Doc, fallback: SyncMode) -> Result<SyncMode> {
324        let author = self.docs_engine.author().await?;
325        let Some(entry) = self.docs_engine.get(doc, author, MODE_KEY).await? else {
326            return Ok(fallback);
327        };
328        let mode_bytes = self.blob_store.get(entry.content_hash()).await?;
329        let mode_value = std::str::from_utf8(&mode_bytes)
330            .map_err(|error| SyncwebError::operation("folder mode metadata is not UTF-8", error))?;
331        mode_value.parse()
332    }
333}
syncweb_core/folder/manager.rs

syncweb_core/folder/
manager.rs
