syncweb_core/
snapshot.rs

1use std::{
2    collections::{BTreeMap, BTreeSet},
3    fs,
4    path::{Component, Path, PathBuf},
5    time::{SystemTime, UNIX_EPOCH},
6};
7
8use iroh::Endpoint;
9use iroh_blobs::{BlobFormat, Hash, ticket::BlobTicket};
10use iroh_docs::NamespaceId;
11use serde::{Deserialize, Serialize};
12
13use crate::{
14    error::{Result, SyncwebError},
15    folder::SyncwebFolder,
16    fs::{ExportEntry, Exporter, ParallelScanner},
17    node::{blob_store::BlobStore, docs_engine::DocsEngine},
18};
19
20const SCHEMA_VERSION: u32 = 1;
21const SNAPSHOT_PIN_PREFIX: &str = "syncweb/snapshot/";
22
23/// The content-addressed identifier of a snapshot.
24pub type SnapshotId = Hash;
25
26/// A file reference captured by a snapshot.
27#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
28#[non_exhaustive]
29pub struct SnapshotEntry {
30    pub path: PathBuf,
31    pub hash: Hash,
32    pub size: u64,
33}
34
35impl SnapshotEntry {
36    /// # Errors
37    ///
38    /// Returns an error if `path` is not a safe relative path.
39    pub fn new(path_value: impl Into<PathBuf>, hash: Hash, size: u64) -> Result<Self> {
40        let path = path_value.into();
41        validate_path(&path)?;
42        Ok(Self { path, hash, size })
43    }
44}
45
46/// A content-addressed snapshot of a folder or local directory.
47#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
48#[non_exhaustive]
49pub struct Snapshot {
50    pub schema_version: u32,
51    pub id: SnapshotId,
52    pub namespace_id: Option<NamespaceId>,
53    pub root_hash: Hash,
54    pub created_at: u64,
55    pub description: Option<String>,
56    pub total_size: u64,
57    pub file_count: u64,
58    pub entries: Vec<SnapshotEntry>,
59}
60
61impl Snapshot {
62    /// Build a snapshot from already-addressed content references.
63    ///
64    /// No file data is copied by this operation.
65    #[must_use]
66    pub fn new(
67        namespace_id: Option<NamespaceId>,
68        mut entries: Vec<SnapshotEntry>,
69        description: Option<String>,
70    ) -> Self {
71        entries.sort_by(|left, right| left.path.cmp(&right.path));
72        let root_hash = entries_hash(&entries);
73        let total_size = entries.iter().map(|entry| entry.size).sum();
74        let file_count = u64::try_from(entries.len()).unwrap_or(u64::MAX);
75        let created_at = SystemTime::now()
76            .duration_since(UNIX_EPOCH)
77            .map_or(0, |duration| duration.as_secs());
78        Self {
79            schema_version: SCHEMA_VERSION,
80            id: root_hash,
81            namespace_id,
82            root_hash,
83            created_at,
84            description,
85            total_size,
86            file_count,
87            entries,
88        }
89    }
90
91    /// Validate the snapshot schema, paths, counts, and content root.
92    ///
93    /// # Errors
94    ///
95    /// Returns an error if any snapshot invariant is violated.
96    pub fn validate(&self) -> Result<()> {
97        if self.schema_version != SCHEMA_VERSION {
98            return Err(SyncwebError::InvalidConfig(format!(
99                "unsupported snapshot schema version {}",
100                self.schema_version
101            )));
102        }
103        let mut paths = BTreeSet::new();
104        for entry in &self.entries {
105            validate_path(&entry.path)?;
106            if !paths.insert(&entry.path) {
107                return Err(SyncwebError::InvalidConfig(format!(
108                    "duplicate snapshot path {}",
109                    entry.path.display()
110                )));
111            }
112        }
113        let expected_count = u64::try_from(self.entries.len()).unwrap_or(u64::MAX);
114        if self.file_count != expected_count {
115            return Err(SyncwebError::InvalidConfig(
116                "snapshot file count is incorrect".to_owned(),
117            ));
118        }
119        let expected_size: u64 = self.entries.iter().map(|entry| entry.size).sum();
120        if self.total_size != expected_size {
121            return Err(SyncwebError::InvalidConfig(
122                "snapshot total size is incorrect".to_owned(),
123            ));
124        }
125        if self.root_hash != entries_hash(&self.entries) || self.id != self.root_hash {
126            return Err(SyncwebError::InvalidConfig(
127                "snapshot content root is incorrect".to_owned(),
128            ));
129        }
130        Ok(())
131    }
132
133    /// Serialize a validated snapshot manifest.
134    ///
135    /// # Errors
136    ///
137    /// Returns an error if validation or serialization fails.
138    pub fn to_bytes(&self) -> Result<Vec<u8>> {
139        self.validate()?;
140        serde_json::to_vec(self).map_err(|error| SyncwebError::operation("failed to serialize snapshot", error))
141    }
142
143    /// Decode and validate a snapshot manifest.
144    ///
145    /// # Errors
146    ///
147    /// Returns an error if decoding or validation fails.
148    pub fn from_bytes(bytes: impl AsRef<[u8]>) -> Result<Self> {
149        let snapshot: Self = serde_json::from_slice(bytes.as_ref())
150            .map_err(|error| SyncwebError::operation("failed to deserialize snapshot", error))?;
151        snapshot.validate()?;
152        Ok(snapshot)
153    }
154
155    /// Compare two snapshots by logical path and content hash.
156    ///
157    /// # Errors
158    ///
159    /// Returns an error if either snapshot is invalid.
160    pub fn diff(&self, other: &Self) -> Result<SnapshotDiff> {
161        self.validate()?;
162        other.validate()?;
163        let left = self
164            .entries
165            .iter()
166            .map(|entry| (&entry.path, entry))
167            .collect::<BTreeMap<_, _>>();
168        let right = other
169            .entries
170            .iter()
171            .map(|entry| (&entry.path, entry))
172            .collect::<BTreeMap<_, _>>();
173        let mut diff = SnapshotDiff::default();
174        for (path, entry) in &right {
175            match left.get(path) {
176                None => diff.added.push((*entry).clone()),
177                Some(previous) if *previous != *entry => diff.modified.push(((*previous).clone(), (*entry).clone())),
178                Some(_) => {}
179            }
180        }
181        for (path, entry) in &left {
182            if !right.contains_key(path) {
183                diff.removed.push((*entry).clone());
184            }
185        }
186        Ok(diff)
187    }
188}
189
190/// The path-level changes between two snapshots.
191#[derive(Clone, Debug, Default, Eq, PartialEq)]
192#[non_exhaustive]
193pub struct SnapshotDiff {
194    pub added: Vec<SnapshotEntry>,
195    pub removed: Vec<SnapshotEntry>,
196    pub modified: Vec<(SnapshotEntry, SnapshotEntry)>,
197}
198
199/// Persists snapshot manifests and pins all content referenced by them.
200#[derive(Clone)]
201pub struct SnapshotStore {
202    blobs: BlobStore,
203    docs: Option<DocsEngine>,
204}
205
206impl SnapshotStore {
207    #[must_use]
208    pub const fn new(blobs: BlobStore) -> Self {
209        Self { blobs, docs: None }
210    }
211
212    #[must_use]
213    pub const fn with_docs(blobs: BlobStore, docs: DocsEngine) -> Self {
214        Self {
215            blobs,
216            docs: Some(docs),
217        }
218    }
219
220    #[must_use]
221    pub fn from_node(node: &crate::node::iroh_node::IrohNode) -> Self {
222        Self::with_docs(node.blob_store().clone(), node.docs_engine().clone())
223    }
224
225    /// Create and persist a snapshot from existing content references.
226    ///
227    /// # Errors
228    ///
229    /// Returns an error if a referenced blob is unavailable or the manifest cannot be stored.
230    pub async fn create_snapshot(
231        &self,
232        namespace_id: Option<NamespaceId>,
233        entries: Vec<SnapshotEntry>,
234        description: Option<String>,
235    ) -> Result<Snapshot> {
236        let snapshot = Snapshot::new(namespace_id, entries, description);
237        self.store(&snapshot).await?;
238        Ok(snapshot)
239    }
240
241    /// Create a snapshot from a local directory.
242    ///
243    /// Files are imported into the content-addressed store once; the snapshot
244    /// itself only stores references to those blobs.
245    ///
246    /// # Errors
247    ///
248    /// Returns an error if scanning, importing, or manifest storage fails.
249    pub async fn create_from_path(
250        &self,
251        source_value: impl AsRef<Path>,
252        threads: usize,
253        description: Option<String>,
254    ) -> Result<Snapshot> {
255        let source_path = source_value.as_ref();
256        let entries = ParallelScanner::new(source_path, Vec::<String>::new(), threads).scan()?;
257        let mut snapshot_entries = Vec::with_capacity(entries.len());
258        for entry in entries {
259            let hash = self.blobs.add_file(&entry.path).await?;
260            let expected = Hash::from_bytes(*entry.hash.as_bytes());
261            if hash != expected {
262                return Err(SyncwebError::InvalidConfig(format!(
263                    "file changed while creating snapshot: {}",
264                    entry.path.display()
265                )));
266            }
267            snapshot_entries.push(SnapshotEntry::new(entry.relative_path, hash, entry.size)?);
268        }
269        self.create_snapshot(None, snapshot_entries, description).await
270    }
271
272    /// Create a snapshot of the latest entries in a synchronized folder.
273    ///
274    /// # Errors
275    ///
276    /// Returns an error if entries cannot be read or referenced blobs are missing.
277    pub async fn create_for_folder(&self, folder: &SyncwebFolder, description: Option<String>) -> Result<Snapshot> {
278        let docs = self
279            .docs
280            .as_ref()
281            .ok_or_else(|| SyncwebError::InvalidConfig("snapshot store has no document engine".to_owned()))?;
282        let entries = docs.list_latest(folder.doc()).await?;
283        let snapshot_entries = entries
284            .into_iter()
285            .filter(|entry| !entry.key().starts_with(b"sys/"))
286            .map(|entry| {
287                let path = String::from_utf8(entry.key().to_vec())
288                    .map_err(|error| SyncwebError::operation("folder entry path is not UTF-8", error))?;
289                SnapshotEntry::new(path, entry.content_hash(), entry.content_len())
290            })
291            .collect::<Result<Vec<_>>>()?;
292        self.create_snapshot(Some(folder.namespace_id()), snapshot_entries, description)
293            .await
294    }
295
296    /// Restore a snapshot to a local directory, removing files not present in it.
297    ///
298    /// # Errors
299    ///
300    /// Returns an error if referenced content is unavailable or materialization fails.
301    pub async fn restore_to_path(
302        &self,
303        snapshot: &Snapshot,
304        destination_value: impl AsRef<Path>,
305    ) -> Result<Vec<PathBuf>> {
306        snapshot.validate()?;
307        self.verify_blobs(snapshot).await?;
308        let destination = destination_value.as_ref();
309        fs::create_dir_all(destination)?;
310        let expected = snapshot
311            .entries
312            .iter()
313            .map(|entry| entry.path.clone())
314            .collect::<BTreeSet<_>>();
315        for entry in ParallelScanner::new(destination, Vec::<String>::new(), 1).scan()? {
316            if !expected.contains(&entry.relative_path) {
317                fs::remove_file(entry.path)?;
318            }
319        }
320        let exports = snapshot
321            .entries
322            .iter()
323            .map(|entry| ExportEntry::new(entry.path.clone(), entry.hash, entry.size))
324            .collect::<Vec<_>>();
325        Exporter::new(self.blobs.clone(), destination)
326            .export_all_verified(&exports)
327            .await
328    }
329
330    /// Restore a snapshot by updating a synchronized folder's document entries.
331    ///
332    /// # Errors
333    ///
334    /// Returns an error if the snapshot belongs to another folder, content is missing,
335    /// or document updates fail.
336    pub async fn restore_for_folder(&self, folder: &SyncwebFolder, snapshot: &Snapshot) -> Result<()> {
337        snapshot.validate()?;
338        if snapshot.namespace_id != Some(folder.namespace_id()) {
339            return Err(SyncwebError::InvalidConfig(
340                "snapshot belongs to a different folder".to_owned(),
341            ));
342        }
343        self.verify_blobs(snapshot).await?;
344        let docs = self
345            .docs
346            .as_ref()
347            .ok_or_else(|| SyncwebError::InvalidConfig("snapshot store has no document engine".to_owned()))?;
348        let expected = snapshot
349            .entries
350            .iter()
351            .map(|entry| entry.path.as_os_str().as_encoded_bytes().to_vec())
352            .collect::<BTreeSet<_>>();
353        for entry in docs.list_latest(folder.doc()).await? {
354            if !entry.key().starts_with(b"sys/") && !expected.contains(entry.key()) {
355                folder.delete_entry(entry.key()).await?;
356            }
357        }
358        for entry in &snapshot.entries {
359            folder
360                .set_blob_ref(entry.path.as_os_str().as_encoded_bytes(), entry.hash, entry.size)
361                .await?;
362        }
363        Ok(())
364    }
365
366    /// List all locally pinned snapshots.
367    ///
368    /// # Errors
369    ///
370    /// Returns an error if snapshot pins or manifests cannot be read.
371    pub async fn list(&self) -> Result<Vec<Snapshot>> {
372        let mut snapshots = Vec::new();
373        for (name, hash) in self.blobs.list_pins(SNAPSHOT_PIN_PREFIX).await? {
374            if !name.ends_with("/manifest") {
375                continue;
376            }
377            let snapshot = self.load_manifest(hash).await?;
378            snapshots.push(snapshot);
379        }
380        snapshots.sort_by(|left, right| right.created_at.cmp(&left.created_at).then(left.id.cmp(&right.id)));
381        Ok(snapshots)
382    }
383
384    /// Load a snapshot by its content-addressed identifier.
385    ///
386    /// # Errors
387    ///
388    /// Returns an error if the snapshot is not pinned or cannot be decoded.
389    pub async fn load(&self, id: SnapshotId) -> Result<Snapshot> {
390        let prefix = format!("{SNAPSHOT_PIN_PREFIX}{id}/manifest");
391        let hash = self
392            .blobs
393            .list_pins(prefix.as_bytes())
394            .await?
395            .into_iter()
396            .find_map(|(name, hash)| (name == prefix).then_some(hash))
397            .ok_or_else(|| SyncwebError::InvalidConfig(format!("snapshot not found: {id}")))?;
398        self.load_manifest(hash).await
399    }
400
401    /// Remove a snapshot and release its pins.
402    ///
403    /// # Errors
404    ///
405    /// Returns an error if the snapshot cannot be found or pins cannot be removed.
406    pub async fn delete(&self, id: SnapshotId) -> Result<()> {
407        let manifest_name = manifest_pin_name(id);
408        let manifest_hash = self
409            .blobs
410            .list_pins(manifest_name.as_bytes())
411            .await?
412            .into_iter()
413            .find_map(|(name, hash)| (name == manifest_name).then_some(hash))
414            .ok_or_else(|| SyncwebError::InvalidConfig(format!("snapshot not found: {id}")))?;
415        let snapshot = self.load_manifest(manifest_hash).await?;
416        self.blobs.unpin(manifest_name).await?;
417        for entry in &snapshot.entries {
418            self.blobs.unpin(content_pin_name(id, entry.hash)).await?;
419        }
420        Ok(())
421    }
422
423    /// Create a ticket for a snapshot manifest.
424    ///
425    /// # Errors
426    ///
427    /// Returns an error if the snapshot is not locally available.
428    pub async fn ticket(&self, endpoint: &Endpoint, snapshot: &Snapshot) -> Result<BlobTicket> {
429        let hash = Hash::new(snapshot.to_bytes()?);
430        if !self.blobs.has(hash).await? {
431            return Err(SyncwebError::InvalidConfig(
432                "snapshot manifest is not stored".to_owned(),
433            ));
434        }
435        Ok(self.blobs.ticket(endpoint, hash))
436    }
437
438    /// Fetch a shared snapshot manifest and all referenced content blobs.
439    ///
440    /// # Errors
441    ///
442    /// Returns an error if the ticket, manifest, or any content blob cannot be fetched.
443    pub async fn import_ticket(&self, endpoint: &Endpoint, ticket: &BlobTicket) -> Result<Snapshot> {
444        if !self.blobs.has(ticket.hash()).await? {
445            self.blobs.fetch(endpoint, ticket).await?;
446        }
447        let snapshot = self.load_manifest(ticket.hash()).await?;
448        for entry in &snapshot.entries {
449            if !self.blobs.has(entry.hash).await? {
450                let content_ticket = BlobTicket::new(ticket.addr().clone(), entry.hash, BlobFormat::Raw);
451                self.blobs.fetch(endpoint, &content_ticket).await?;
452            }
453        }
454        self.store(&snapshot).await?;
455        Ok(snapshot)
456    }
457
458    async fn store(&self, snapshot: &Snapshot) -> Result<()> {
459        snapshot.validate()?;
460        for entry in &snapshot.entries {
461            if !self.blobs.has(entry.hash).await? {
462                return Err(SyncwebError::InvalidConfig(format!(
463                    "snapshot content is missing: {}",
464                    entry.path.display()
465                )));
466            }
467            self.blobs
468                .pin(content_pin_name(snapshot.id, entry.hash), entry.hash)
469                .await?;
470        }
471        let bytes = snapshot.to_bytes()?;
472        let manifest_hash = self.blobs.add_bytes(&bytes).await?;
473        self.blobs.pin(manifest_pin_name(snapshot.id), manifest_hash).await?;
474        Ok(())
475    }
476
477    async fn load_manifest(&self, hash: Hash) -> Result<Snapshot> {
478        let snapshot = Snapshot::from_bytes(self.blobs.get(hash).await?)?;
479        if snapshot.id != snapshot.root_hash {
480            return Err(SyncwebError::InvalidConfig(
481                "snapshot ID does not match its root".to_owned(),
482            ));
483        }
484        Ok(snapshot)
485    }
486
487    async fn verify_blobs(&self, snapshot: &Snapshot) -> Result<()> {
488        for entry in &snapshot.entries {
489            if !self.blobs.has(entry.hash).await? {
490                return Err(SyncwebError::InvalidConfig(format!(
491                    "snapshot content is unavailable: {}",
492                    entry.path.display()
493                )));
494            }
495        }
496        Ok(())
497    }
498}
499
500fn entries_hash(entries: &[SnapshotEntry]) -> Hash {
501    let bytes = serde_json::to_vec(entries).unwrap_or_default();
502    Hash::new(bytes)
503}
504
505fn validate_path(path: &Path) -> Result<()> {
506    if path.as_os_str().is_empty()
507        || path.is_absolute()
508        || path.components().any(|component| {
509            matches!(
510                component,
511                Component::ParentDir | Component::RootDir | Component::Prefix(_)
512            )
513        })
514    {
515        return Err(SyncwebError::InvalidConfig(format!(
516            "snapshot path must be a non-empty relative path: {}",
517            path.display()
518        )));
519    }
520    Ok(())
521}
522
523fn manifest_pin_name(id: Hash) -> String {
524    format!("{SNAPSHOT_PIN_PREFIX}{id}/manifest")
525}
526
527fn content_pin_name(id: Hash, hash: Hash) -> String {
528    format!("{SNAPSHOT_PIN_PREFIX}{id}/blob/{hash}")
529}
syncweb_core/snapshot.rs

syncweb_core/
snapshot.rs
