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::{Hash, ticket::BlobTicket};
10use iroh_docs::NamespaceId;
11use serde::{Deserialize, Serialize};
12
13use crate::{
14    constants::SNAPSHOT_PIN_PREFIX,
15    error::{Result, SyncwebError},
16    folder::SyncwebFolder,
17    fs::{ExportEntry, Exporter, ParallelScanner},
18    node::{blob_store::BlobStore, docs_engine::DocsEngine},
19};
20
21const SCHEMA_VERSION: u32 = 1;
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 = crate::pins::snapshot_manifest_pin(id);
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 = crate::pins::snapshot_manifest_pin(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
419                .unpin(crate::pins::snapshot_content_pin(id, entry.hash))
420                .await?;
421        }
422        Ok(())
423    }
424
425    /// Create a ticket for a snapshot manifest.
426    ///
427    /// # Errors
428    ///
429    /// Returns an error if the snapshot is not locally available.
430    pub async fn ticket(&self, endpoint: &Endpoint, snapshot: &Snapshot) -> Result<BlobTicket> {
431        let hash = Hash::new(snapshot.to_bytes()?);
432        if !self.blobs.has(hash).await? {
433            return Err(SyncwebError::InvalidConfig(
434                "snapshot manifest is not stored".to_owned(),
435            ));
436        }
437        Ok(self.blobs.ticket(endpoint, hash))
438    }
439
440    /// Fetch a shared snapshot manifest and all referenced content blobs.
441    ///
442    /// # Errors
443    ///
444    /// Returns an error if the ticket, manifest, or any content blob cannot be fetched.
445    pub async fn import_ticket(&self, endpoint: &Endpoint, ticket: &BlobTicket) -> Result<Snapshot> {
446        if !self.blobs.has(ticket.hash()).await? {
447            self.blobs.fetch(endpoint, ticket).await?;
448        }
449        let snapshot = self.load_manifest(ticket.hash()).await?;
450        for entry in &snapshot.entries {
451            if !self.blobs.has(entry.hash).await? {
452                let content_ticket = self.blobs.ticket_for_addr(ticket.addr().clone(), entry.hash);
453                self.blobs.fetch(endpoint, &content_ticket).await?;
454            }
455        }
456        self.store(&snapshot).await?;
457        Ok(snapshot)
458    }
459
460    async fn store(&self, snapshot: &Snapshot) -> Result<()> {
461        snapshot.validate()?;
462        for entry in &snapshot.entries {
463            if !self.blobs.has(entry.hash).await? {
464                return Err(SyncwebError::InvalidConfig(format!(
465                    "snapshot content is missing: {}",
466                    entry.path.display()
467                )));
468            }
469            self.blobs
470                .pin(crate::pins::snapshot_content_pin(snapshot.id, entry.hash), entry.hash)
471                .await?;
472        }
473        let bytes = snapshot.to_bytes()?;
474        let manifest_hash = self.blobs.add_bytes(&bytes).await?;
475        self.blobs
476            .pin(crate::pins::snapshot_manifest_pin(snapshot.id), manifest_hash)
477            .await?;
478        Ok(())
479    }
480
481    async fn load_manifest(&self, hash: Hash) -> Result<Snapshot> {
482        let snapshot = Snapshot::from_bytes(self.blobs.get(hash).await?)?;
483        if snapshot.id != snapshot.root_hash {
484            return Err(SyncwebError::InvalidConfig(
485                "snapshot ID does not match its root".to_owned(),
486            ));
487        }
488        Ok(snapshot)
489    }
490
491    async fn verify_blobs(&self, snapshot: &Snapshot) -> Result<()> {
492        for entry in &snapshot.entries {
493            if !self.blobs.has(entry.hash).await? {
494                return Err(SyncwebError::InvalidConfig(format!(
495                    "snapshot content is unavailable: {}",
496                    entry.path.display()
497                )));
498            }
499        }
500        Ok(())
501    }
502}
503
504fn entries_hash(entries: &[SnapshotEntry]) -> Hash {
505    let bytes = serde_json::to_vec(entries).unwrap_or_default();
506    Hash::new(bytes)
507}
508
509fn validate_path(path: &Path) -> Result<()> {
510    if path.as_os_str().is_empty()
511        || path.is_absolute()
512        || path.components().any(|component| {
513            matches!(
514                component,
515                Component::ParentDir | Component::RootDir | Component::Prefix(_)
516            )
517        })
518    {
519        return Err(SyncwebError::InvalidConfig(format!(
520            "snapshot path must be a non-empty relative path: {}",
521            path.display()
522        )));
523    }
524    Ok(())
525}
syncweb_core/snapshot.rs

syncweb_core/
snapshot.rs
