syncweb_core/folder/
archive_export.rs

1use std::{
2    collections::BTreeSet,
3    path::{Path, PathBuf},
4    sync::Arc,
5};
6
7use async_compression::tokio::write::ZstdEncoder;
8use iroh_blobs::Hash;
9use semver::Version;
10use serde::{Deserialize, Serialize};
11use tokio::{
12    fs,
13    io::{AsyncReadExt, AsyncWrite, AsyncWriteExt},
14    sync::Mutex,
15};
16use uuid::Uuid;
17
18use crate::{
19    daemon::ManagedPool,
20    error::{Result, SyncwebError},
21    filter::{FilterAction, FilterEngine, FilterEntry},
22    node::blob_store::BlobStore,
23};
24
25use super::{CollectionEntry, CollectionManifest};
26
27const CAR_VERSION: u64 = 1;
28const CID_VERSION: u64 = 1;
29const RAW_CODEC: u64 = 0x55;
30const BLAKE3_MULTIHASH: u64 = 0x1e;
31const HASH_SIZE: u64 = 32;
32const COPY_BUFFER_SIZE: usize = 64 * 1024;
33
34/// Options controlling which collection version and entries are included.
35#[derive(Clone, Debug, Default)]
36#[non_exhaustive]
37pub struct DropExportOptions {
38    version: Option<String>,
39    filter: Option<FilterEngine>,
40}
41
42impl DropExportOptions {
43    #[must_use]
44    pub fn with_version(mut self, version: impl Into<String>) -> Self {
45        self.version = Some(version.into());
46        self
47    }
48
49    #[must_use]
50    pub fn with_filter(mut self, filter: FilterEngine) -> Self {
51        self.filter = Some(filter);
52        self
53    }
54
55    #[must_use]
56    pub fn version(&self) -> Option<&str> {
57        self.version.as_deref()
58    }
59
60    #[must_use]
61    pub const fn filter(&self) -> Option<&FilterEngine> {
62        self.filter.as_ref()
63    }
64}
65
66/// Information about a completed archive export.
67#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
68#[non_exhaustive]
69pub struct DropExportResult {
70    pub output: PathBuf,
71    pub collection_id: uuid::Uuid,
72    pub version: String,
73    pub manifest: Hash,
74    pub entry_count: usize,
75    pub block_count: usize,
76    pub archive_size: u64,
77}
78
79/// Streams collection manifests and content blobs into atomic `.car.zst` files.
80#[derive(Clone)]
81pub struct DropExporter {
82    blob_store: BlobStore,
83    export_lock: Arc<Mutex<()>>,
84}
85
86impl DropExporter {
87    #[must_use]
88    pub fn new(blob_store: BlobStore) -> Self {
89        Self {
90            blob_store,
91            export_lock: Arc::new(Mutex::new(())),
92        }
93    }
94
95    /// Export one manifest without filtering.
96    ///
97    /// # Errors
98    ///
99    /// Returns an error if the manifest is invalid, content is unavailable, or
100    /// the archive cannot be written.
101    pub async fn export_archive(
102        &self,
103        manifest: &CollectionManifest,
104        output: impl AsRef<Path>,
105    ) -> Result<DropExportResult> {
106        self.export_drop_with_options(
107            std::slice::from_ref(manifest),
108            output,
109            DropExportOptions::default(),
110            None,
111        )
112        .await
113    }
114
115    /// Export one or more versions, selecting the requested version or latest
116    /// semver version when no version is specified.
117    ///
118    /// # Errors
119    ///
120    /// Returns an error if manifests disagree about their collection, the
121    /// requested version is unavailable, or the archive cannot be written.
122    pub async fn export_manifests(
123        &self,
124        manifests: &[CollectionManifest],
125        output: impl AsRef<Path>,
126        options: DropExportOptions,
127    ) -> Result<DropExportResult> {
128        self.export_drop_with_options(manifests, output, options, None).await
129    }
130
131    /// Export a manifest using filtering and version-selection options.
132    ///
133    /// A filtered manifest is deliberately unsigned because changing its
134    /// entries changes the signed content. The archive remains content
135    /// addressed and each included blob is verified while it is streamed.
136    ///
137    /// # Errors
138    ///
139    /// Returns an error if the selected manifest or any referenced blob is
140    /// invalid or unavailable, or if the archive cannot be written.
141    pub async fn export_drop_with_options(
142        &self,
143        manifests: &[CollectionManifest],
144        output: impl AsRef<Path>,
145        options: DropExportOptions,
146        pool: Option<&ManagedPool>,
147    ) -> Result<DropExportResult> {
148        let _export_guard = self.export_lock.lock().await;
149        let selected = in_pool(pool, || select_manifest(manifests, options.version()))?;
150        let filtered = in_pool(pool, || filter_manifest(selected, options.filter()))?;
151        let (manifest_bytes, manifest_hash, entries) = in_pool(pool, || -> Result<_> {
152            let manifest_bytes = filtered.to_bytes()?;
153            let manifest_hash = Hash::new(&manifest_bytes);
154            let entries = unique_entries(&filtered.entries);
155            Ok((manifest_bytes, manifest_hash, entries))
156        })?;
157        for entry in &entries {
158            if !self.blob_store.has(entry.content_id).await? {
159                return Err(SyncwebError::InvalidConfig(format!(
160                    "drop content is missing from the blob store: {}",
161                    entry.logical_path.display()
162                )));
163            }
164        }
165
166        let output_path = output.as_ref().to_path_buf();
167        let parent = output_path.parent().unwrap_or_else(|| Path::new("."));
168        fs::create_dir_all(parent).await?;
169        let staging = parent.join(format!(
170            "{}{}",
171            crate::constants::DROP_EXPORT_STAGING_PREFIX,
172            Uuid::new_v4()
173        ));
174        let write_result = self
175            .write_archive(&staging, &manifest_bytes, manifest_hash, &entries, pool)
176            .await;
177        if let Err(error) = write_result {
178            remove_if_present(&staging).await?;
179            return Err(error);
180        }
181        if let Err(error) = fs::rename(&staging, &output_path).await {
182            remove_if_present(&staging).await?;
183            return Err(SyncwebError::operation("failed to finalize drop archive", error));
184        }
185        let archive_size = fs::metadata(&output_path)
186            .await
187            .map_err(|error| SyncwebError::operation("failed to inspect drop archive", error))?
188            .len();
189        Ok(DropExportResult {
190            output: output_path,
191            collection_id: filtered.collection_id,
192            version: filtered.version,
193            manifest: manifest_hash,
194            entry_count: filtered.entries.len(),
195            block_count: entries.len().saturating_add(1),
196            archive_size,
197        })
198    }
199
200    async fn write_archive(
201        &self,
202        staging: &Path,
203        manifest_bytes: &[u8],
204        manifest_hash: Hash,
205        entries: &[CollectionEntry],
206        pool: Option<&ManagedPool>,
207    ) -> Result<()> {
208        let file = fs::File::create(staging)
209            .await
210            .map_err(|error| SyncwebError::operation("failed to create drop archive", error))?;
211        let mut encoder = ZstdEncoder::new(file);
212        let header = car_header(manifest_hash)?;
213        encoder.write_all(&header).await?;
214        write_section(&mut encoder, manifest_hash, manifest_bytes).await?;
215        let temporary_dir = staging.parent().unwrap_or_else(|| Path::new("."));
216        for entry in entries {
217            write_blob_section(&self.blob_store, &mut encoder, entry, temporary_dir, pool).await?;
218        }
219        encoder
220            .shutdown()
221            .await
222            .map_err(|error| SyncwebError::operation("failed to finish drop compression", error))?;
223        Ok(())
224    }
225}
226
227/// Export one collection manifest into a compressed CAR archive.
228///
229/// # Errors
230///
231/// Returns an error if the manifest or referenced content is invalid or the
232/// archive cannot be written.
233pub async fn export_archive(
234    blob_store: BlobStore,
235    manifest: &CollectionManifest,
236    output: impl AsRef<Path>,
237    filter: Option<&FilterEngine>,
238) -> Result<DropExportResult> {
239    let options = filter.map_or_else(DropExportOptions::default, |value| {
240        DropExportOptions::default().with_filter(value.clone())
241    });
242    DropExporter::new(blob_store)
243        .export_drop_with_options(std::slice::from_ref(manifest), output, options, None)
244        .await
245}
246
247fn in_pool<F, R>(pool: Option<&ManagedPool>, operation: F) -> R
248where
249    F: FnOnce() -> R + Send,
250    R: Send,
251{
252    match pool {
253        Some(managed_pool) => managed_pool.install(operation),
254        None => operation(),
255    }
256}
257
258fn select_manifest(manifests: &[CollectionManifest], requested: Option<&str>) -> Result<CollectionManifest> {
259    let mut candidates = manifests.iter();
260    let Some(first) = candidates.next() else {
261        return Err(SyncwebError::InvalidConfig(
262            "at least one collection manifest is required".to_owned(),
263        ));
264    };
265    first.validate()?;
266    let collection_id = first.collection_id;
267    let mut selected = first.clone();
268    for manifest in candidates {
269        manifest.validate()?;
270        if manifest.collection_id != collection_id {
271            return Err(SyncwebError::InvalidConfig(
272                "drop manifests must belong to the same collection".to_owned(),
273            ));
274        }
275        if requested.is_none() && manifest.is_upgrade_from(&selected)? {
276            selected = manifest.clone();
277        }
278    }
279    if let Some(version) = requested {
280        Version::parse(version)
281            .map_err(|error| SyncwebError::InvalidConfig(format!("invalid requested collection version: {error}")))?;
282        selected = manifests
283            .iter()
284            .find(|manifest| manifest.version == version)
285            .cloned()
286            .ok_or_else(|| SyncwebError::InvalidConfig(format!("collection version {version} is unavailable")))?;
287        if selected.collection_id != collection_id {
288            return Err(SyncwebError::InvalidConfig(
289                "drop manifests must belong to the same collection".to_owned(),
290            ));
291        }
292    }
293    Ok(selected)
294}
295
296fn filter_manifest(manifest: CollectionManifest, filter: Option<&FilterEngine>) -> Result<CollectionManifest> {
297    let Some(engine) = filter else {
298        return Ok(manifest);
299    };
300    let version = Version::parse(&manifest.version)
301        .map_err(|error| SyncwebError::InvalidConfig(format!("invalid collection version: {error}")))?;
302    let mut filtered = manifest;
303    filtered.entries.retain(|entry| {
304        let filter_entry = FilterEntry::new(&entry.logical_path, entry.size).with_version(version.clone());
305        engine.evaluate(&filter_entry) == FilterAction::Accept
306    });
307    filtered.signature = None;
308    filtered.public_key = None;
309    filtered.validate()?;
310    Ok(filtered)
311}
312
313fn unique_entries(entries: &[CollectionEntry]) -> Vec<CollectionEntry> {
314    let mut hashes = BTreeSet::new();
315    entries
316        .iter()
317        .filter(|entry| hashes.insert(entry.content_id))
318        .cloned()
319        .collect()
320}
321
322async fn write_blob_section(
323    blob_store: &BlobStore,
324    encoder: &mut ZstdEncoder<fs::File>,
325    entry: &CollectionEntry,
326    temporary_dir: &Path,
327    pool: Option<&ManagedPool>,
328) -> Result<()> {
329    let temporary_path = temporary_dir.join(format!(
330        "{}{}",
331        crate::constants::DROP_EXPORT_BLOB_STAGING_PREFIX,
332        Uuid::new_v4()
333    ));
334    let result = stream_blob(blob_store, encoder, entry, &temporary_path, pool).await;
335    let cleanup = remove_if_present(&temporary_path).await;
336    match (result, cleanup) {
337        (Ok(()), Ok(())) => Ok(()),
338        (Err(error), Ok(())) | (Ok(()), Err(error)) => Err(error),
339        (Err(error), Err(cleanup_error)) => Err(SyncwebError::operation(
340            "failed to clean up temporary drop blob",
341            format!("{error}; cleanup failed: {cleanup_error}"),
342        )),
343    }
344}
345
346async fn stream_blob(
347    blob_store: &BlobStore,
348    encoder: &mut ZstdEncoder<fs::File>,
349    entry: &CollectionEntry,
350    temporary_path: &Path,
351    pool: Option<&ManagedPool>,
352) -> Result<()> {
353    blob_store.export_to_path(entry.content_id, temporary_path).await?;
354    let metadata = fs::metadata(temporary_path)
355        .await
356        .map_err(|error| SyncwebError::operation("failed to inspect exported drop blob", error))?;
357    if metadata.len() != entry.size {
358        return Err(SyncwebError::InvalidConfig(format!(
359            "drop blob size does not match manifest for {}",
360            entry.logical_path.display()
361        )));
362    }
363    write_section_prefix(encoder, entry.content_id, entry.size).await?;
364    let mut source = fs::File::open(temporary_path)
365        .await
366        .map_err(|error| SyncwebError::operation("failed to open exported drop blob", error))?;
367    let mut buffer = vec![0_u8; COPY_BUFFER_SIZE];
368    let mut hasher = blake3::Hasher::new();
369    let mut copied = 0_u64;
370    loop {
371        let read = source.read(&mut buffer).await?;
372        if read == 0 {
373            break;
374        }
375        let chunk = buffer
376            .get(..read)
377            .ok_or_else(|| SyncwebError::operation("blob reader returned an invalid chunk", read))?;
378        encoder.write_all(chunk).await?;
379        in_pool(pool, || hasher.update(chunk));
380        copied = copied
381            .checked_add(u64::try_from(read).map_err(|error| SyncwebError::operation("drop blob is too large", error))?)
382            .ok_or_else(|| SyncwebError::operation("drop blob size overflow", "u64 limit exceeded"))?;
383    }
384    let actual = in_pool(pool, || Hash::from_bytes(*hasher.finalize().as_bytes()));
385    if copied != entry.size || actual != entry.content_id {
386        return Err(SyncwebError::InvalidConfig(format!(
387            "drop blob hash does not match manifest for {}",
388            entry.logical_path.display()
389        )));
390    }
391    Ok(())
392}
393
394async fn write_section<W: AsyncWrite + Unpin>(writer: &mut W, hash: Hash, bytes: &[u8]) -> Result<()> {
395    write_section_prefix(
396        writer,
397        hash,
398        u64::try_from(bytes.len())
399            .map_err(|error| SyncwebError::operation("manifest is too large for a drop archive", error))?,
400    )
401    .await?;
402    writer.write_all(bytes).await?;
403    Ok(())
404}
405
406async fn write_section_prefix<W: AsyncWrite + Unpin>(writer: &mut W, hash: Hash, size: u64) -> Result<()> {
407    let cid = cid_for_hash(hash);
408    let section_size = u64::try_from(cid.len())
409        .map_err(|error| SyncwebError::operation("CAR section is too large", error))?
410        .checked_add(size)
411        .ok_or_else(|| SyncwebError::operation("CAR section size overflow", "u64 limit exceeded"))?;
412    let mut encoded_size = Vec::with_capacity(10);
413    encode_varint(section_size, &mut encoded_size);
414    writer.write_all(&encoded_size).await?;
415    writer.write_all(&cid).await?;
416    Ok(())
417}
418
419fn car_header(root: Hash) -> Result<Vec<u8>> {
420    let cid = cid_for_hash(root);
421    let cid_length =
422        u64::try_from(cid.len()).map_err(|error| SyncwebError::operation("CAR root CID is too large", error))?;
423    let mut header = Vec::with_capacity(cid.len().saturating_add(8));
424    header.push(0x0a);
425    encode_varint(cid_length, &mut header);
426    header.extend_from_slice(&cid);
427    header.push(0x10);
428    encode_varint(CAR_VERSION, &mut header);
429    let header_length =
430        u64::try_from(header.len()).map_err(|error| SyncwebError::operation("CAR header is too large", error))?;
431    let mut encoded = Vec::with_capacity(header.len().saturating_add(10));
432    encode_varint(header_length, &mut encoded);
433    encoded.extend_from_slice(&header);
434    Ok(encoded)
435}
436
437fn cid_for_hash(hash: Hash) -> Vec<u8> {
438    let mut cid = Vec::with_capacity(2 + 2 + 2 + 32);
439    encode_varint(CID_VERSION, &mut cid);
440    encode_varint(RAW_CODEC, &mut cid);
441    encode_varint(BLAKE3_MULTIHASH, &mut cid);
442    encode_varint(HASH_SIZE, &mut cid);
443    cid.extend_from_slice(hash.as_bytes());
444    cid
445}
446
447fn encode_varint(mut value: u64, output: &mut Vec<u8>) {
448    while value >= 0x80 {
449        output.push(value.to_le_bytes()[0] | 0x80);
450        value >>= 7;
451    }
452    output.push(value.to_le_bytes()[0]);
453}
454
455async fn remove_if_present(path: &Path) -> Result<()> {
456    match fs::remove_file(path).await {
457        Ok(()) => Ok(()),
458        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
459        Err(error) => Err(SyncwebError::operation(
460            "failed to remove temporary archive file",
461            error,
462        )),
463    }
464}
syncweb_core/folder/archive_export.rs

syncweb_core/folder/
archive_export.rs
