syncweb_core/fs/
importer.rs

1use std::{fs, path::Path, sync::Arc};
2
3use iroh_blobs::Hash;
4use iroh_docs::{AuthorId, api::Doc};
5use n0_future::{BufferedStreamExt, TryStreamExt, stream};
6
7use crate::{
8    error::{Result, SyncwebError},
9    node::{blob_store::BlobStore, docs_engine::DocsEngine},
10};
11
12use super::{FileEntry, ParallelScanner, Scanner};
13
14/// A local file after it has been added to the blob store and document.
15#[derive(Clone, Debug)]
16#[non_exhaustive]
17pub struct ImportEntry {
18    pub path: std::path::PathBuf,
19    pub relative_path: std::path::PathBuf,
20    pub hash: Hash,
21    pub size: u64,
22}
23
24/// Sequential scanner/blob/document import pipeline.
25#[derive(Clone)]
26pub struct Importer {
27    blob_store: BlobStore,
28    docs_engine: DocsEngine,
29    doc: Doc,
30    author: AuthorId,
31    root: Option<std::path::PathBuf>,
32    ignore_patterns: Vec<String>,
33}
34
35impl Importer {
36    #[must_use]
37    pub const fn new(blob_store: BlobStore, docs_engine: DocsEngine, doc: Doc, author: AuthorId) -> Self {
38        Self {
39            blob_store,
40            docs_engine,
41            doc,
42            author,
43            root: None,
44            ignore_patterns: Vec::new(),
45        }
46    }
47
48    #[must_use]
49    pub fn with_root(mut self, root: impl Into<std::path::PathBuf>) -> Self {
50        self.root = Some(root.into());
51        self
52    }
53
54    #[must_use]
55    pub fn with_ignore_patterns(mut self, patterns: Vec<String>) -> Self {
56        self.ignore_patterns = patterns;
57        self
58    }
59
60    /// Import one file or every file below a directory.
61    /// # Errors
62    ///
63    /// Returns an error if the filesystem or database cannot be accessed.
64    pub async fn import_path(&self, input: impl AsRef<Path>) -> Result<Vec<ImportEntry>> {
65        let files = self.scan_input(input.as_ref(), None)?;
66        self.import_entries(files).await
67    }
68
69    /// Import pre-scanned entries.
70    /// # Errors
71    ///
72    /// Returns an error if the filesystem or database cannot be accessed.
73    pub async fn import_entries(&self, source_entries: Vec<FileEntry>) -> Result<Vec<ImportEntry>> {
74        let mut imported = Vec::with_capacity(source_entries.len());
75        for entry in source_entries {
76            imported.push(self.import_one(entry).await?);
77        }
78        Ok(imported)
79    }
80
81    async fn import_one(&self, entry: FileEntry) -> Result<ImportEntry> {
82        // Conflict copies (`<path>.conflict.<hash>`) are materialization
83        // artifacts, not folder content: re-importing one would echo it to
84        // peers as a brand-new key. Skip them so they stay purely local.
85        if entry
86            .relative_path
87            .to_string_lossy()
88            .contains(crate::fs::CONFLICT_COPY_MARKER)
89        {
90            return Ok(ImportEntry {
91                path: entry.path,
92                relative_path: entry.relative_path,
93                hash: Hash::from_bytes(*entry.hash.as_bytes()),
94                size: entry.size,
95            });
96        }
97        let before = fs::metadata(&entry.path)?;
98        if !metadata_matches_entry(&before, &entry) {
99            return Err(file_changed_error(&entry.path));
100        }
101        let hash = self.blob_store.add_file_ref(&entry.path).await?;
102        let after = fs::metadata(&entry.path)?;
103        if !metadata_matches_entry(&after, &entry) || hash != Hash::from_bytes(*entry.hash.as_bytes()) {
104            return Err(file_changed_error(&entry.path));
105        }
106        let key = entry.relative_path.as_os_str().as_encoded_bytes();
107        // Guard against the watcher re-stamping content that materialization
108        // (or a previous remote sync) already recorded for this key. iroh-docs
109        // stamps a fresh timestamp on every `set`, so re-importing an identical
110        // file would turn a remote winner into a newer local edit and flip-flop
111        // LWW forever. If the current latest entry already carries this hash,
112        // the block content is unchanged; leave the doc alone.
113        if self
114            .docs_engine
115            .get_any(&self.doc, key)
116            .await?
117            .is_some_and(|existing| existing.content_hash() == hash)
118        {
119            return Ok(ImportEntry {
120                path: entry.path,
121                relative_path: entry.relative_path,
122                hash,
123                size: entry.size,
124            });
125        }
126        self.docs_engine
127            .set_blob(&self.doc, self.author, key, hash, entry.size)
128            .await?;
129        Ok(ImportEntry {
130            path: entry.path,
131            relative_path: entry.relative_path,
132            hash,
133            size: entry.size,
134        })
135    }
136
137    fn scan_input(&self, input: &Path, threads: Option<usize>) -> Result<Vec<FileEntry>> {
138        if !input.exists() {
139            return Err(SyncwebError::Io(std::io::Error::new(
140                std::io::ErrorKind::NotFound,
141                format!("input path does not exist: {}", input.display()),
142            )));
143        }
144        let root = self.root.clone().unwrap_or_else(|| {
145            if input.is_dir() {
146                input.to_path_buf()
147            } else {
148                input.parent().unwrap_or_else(|| Path::new(".")).to_path_buf()
149            }
150        });
151        let mut entries = match threads {
152            Some(threads_count) => ParallelScanner::new(&root, self.ignore_patterns.clone(), threads_count).scan()?,
153            None => Scanner::new(&root, self.ignore_patterns.clone()).scan()?,
154        };
155        let target = fs::canonicalize(input)?;
156        if input.is_dir() {
157            entries.retain(|entry| fs::canonicalize(&entry.path).is_ok_and(|path| path.starts_with(&target)));
158        } else {
159            entries.retain(|entry| fs::canonicalize(&entry.path).is_ok_and(|path| path == target));
160        }
161        Ok(entries)
162    }
163}
164
165fn metadata_matches_entry(metadata: &fs::Metadata, entry: &FileEntry) -> bool {
166    metadata.len() == entry.size && metadata.modified().unwrap_or(std::time::SystemTime::UNIX_EPOCH) == entry.modified
167}
168
169fn file_changed_error(path: &Path) -> SyncwebError {
170    SyncwebError::operation("file changed during import", path.display())
171}
172
173/// Parallel import pipeline. Hashing and filesystem reads are parallelized by
174/// the scanner; blob/document writes retain their ordering and async safety.
175#[derive(Clone)]
176pub struct ParallelImporter {
177    importer: Arc<Importer>,
178    threads: usize,
179}
180
181impl ParallelImporter {
182    #[must_use]
183    pub fn new(blob_store: BlobStore, docs_engine: DocsEngine, doc: Doc, author: AuthorId) -> Self {
184        Self {
185            importer: Arc::new(Importer::new(blob_store, docs_engine, doc, author)),
186            threads: 0,
187        }
188    }
189
190    #[must_use]
191    pub const fn with_threads(mut self, threads: usize) -> Self {
192        self.threads = threads;
193        self
194    }
195
196    #[must_use]
197    pub fn with_root(mut self, root: impl Into<std::path::PathBuf>) -> Self {
198        self.importer = Arc::new((*self.importer).clone().with_root(root));
199        self
200    }
201
202    #[must_use]
203    pub fn with_ignore_patterns(mut self, patterns: Vec<String>) -> Self {
204        self.importer = Arc::new((*self.importer).clone().with_ignore_patterns(patterns));
205        self
206    }
207
208    /// Import all files under a directory.
209    /// # Errors
210    ///
211    /// Returns an error if the filesystem or database cannot be accessed.
212    pub async fn import_path(&self, input: impl AsRef<Path>) -> Result<Vec<ImportEntry>> {
213        let entries = self.importer.scan_input(input.as_ref(), Some(self.threads))?;
214        self.import_entries(entries).await
215    }
216
217    /// Import pre-scanned entries. This method can be used with a custom
218    /// scanner while preserving the document update semantics.
219    /// # Errors
220    ///
221    /// Returns an error if the filesystem or database cannot be accessed.
222    pub async fn import_entries(&self, source_entries: Vec<FileEntry>) -> Result<Vec<ImportEntry>> {
223        let concurrency = if self.threads == 0 {
224            std::thread::available_parallelism().map_or(1, std::num::NonZeroUsize::get)
225        } else {
226            self.threads
227        };
228        let shared_importer = self.importer.clone();
229        let mut imported = stream::iter(source_entries.into_iter().enumerate().map(|(index, entry)| {
230            let task_importer = shared_importer.clone();
231            async move {
232                task_importer
233                    .import_one(entry)
234                    .await
235                    .map(|imported_entry| (index, imported_entry))
236            }
237        }))
238        .buffered_unordered(concurrency)
239        .try_collect::<Vec<_>>()
240        .await?;
241        imported.sort_by_key(|(index, _)| *index);
242        Ok(imported.into_iter().map(|(_, entry)| entry).collect())
243    }
244}
syncweb_core/fs/importer.rs

syncweb_core/fs/
importer.rs
