feat(mongodb): bson dumps and database backup/restore

crates/dbx-core/src/db/mongo_driver.rs

@@ -2579,17 +2579,17 @@ pub struct MongoInsertOutcome {
     pub errors: Vec<MongoBulkWriteError>,
 }
 
-pub async fn insert_bson_documents(
+pub async fn insert_bson_documents<T: serde::Serialize + Send + Sync>(
     client: &Client,
     database: &str,
     collection: &str,
-    documents: Vec<Document>,
+    documents: Vec<T>,
 ) -> Result<MongoInsertOutcome, MongoBulkWriteError> {
     if documents.is_empty() {
         return Ok(MongoInsertOutcome::default());
     }
     let total = documents.len() as u64;
-    let col = client.database(database).collection::<Document>(collection);
+    let col = client.database(database).collection::<T>(collection);
     // Unordered: one rejected document must not abandon the rest of the batch, and the server
     // then reports every rejection instead of stopping at the first.
     match col.insert_many(documents).ordered(false).await {

crates/dbx-core/src/lib.rs

@@ -58,6 +58,7 @@ pub mod models;
 pub mod mongo_oidc;
 pub mod mongo_ops;
 pub mod mongo_shell;
+pub mod mongodb_dump;
 pub mod mongodb_import_export;
 #[cfg(feature = "mq-admin")]
 pub mod mq;

crates/dbx-core/src/mongodb_dump.rs

@@ -0,0 +1,639 @@
+//! Database-scoped MongoDB dump/restore; transport adapters own dialogs and uploads.
+mod archive;
+mod directory;
+mod metadata;
+mod source;
+mod stream;
+
+pub use source::{
+    attach_mongodb_restore_upload, mongodb_restore_upload_files, prepare_mongodb_directory_catalog,
+    prepare_owned_mongodb_restore_source, MongoDumpFile,
+};
+
+use std::collections::{BTreeSet, HashMap, HashSet};
+use std::future::Future;
+use std::path::{Path, PathBuf};
+use std::pin::Pin;
+use std::sync::{
+    atomic::{AtomicBool, Ordering},
+    Arc,
+};
+use std::time::{Duration, Instant};
+
+use futures::TryStreamExt;
+use mongodb::bson::{doc, Document};
+use serde::{Deserialize, Serialize};
+
+use crate::connection::{AppState, PoolKind};
+use crate::mongodb_import_export::{export_mongodb_query_core, MongoExportFormat, MongoExportRequest};
+use metadata::CollectionMetadata;
+
+type CancelFuture = Pin<Box<dyn Future<Output = bool> + Send>>;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub enum MongoDumpFormat {
+    Directory,
+    Archive,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct MongoDumpCollection {
+    pub database: String,
+    pub name: String,
+    pub kind: String,
+    pub documents: Option<u64>,
+    pub size_bytes: u64,
+    pub indexes: usize,
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub source_files: Vec<String>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct MongoDumpCatalog {
+    pub databases: Vec<String>,
+    pub collections: Vec<MongoDumpCollection>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct MongoRestoreSourceRequest {
+    pub path: String,
+    pub format: MongoDumpFormat,
+    #[serde(default)]
+    pub gzip: bool,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct MongoRestoreSourcePreview {
+    pub source_ref: String,
+    #[serde(flatten)]
+    pub catalog: MongoDumpCatalog,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct MongoDatabaseDumpRequest {
+    pub task_id: String,
+    pub connection_id: String,
+    pub database: String,
+    pub file_path: String,
+    pub format: MongoDumpFormat,
+    #[serde(default)]
+    pub gzip: bool,
+    #[serde(default)]
+    pub collections: Option<Vec<String>>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct MongoDatabaseRestoreRequest {
+    pub task_id: String,
+    pub connection_id: String,
+    pub database: String,
+    pub source_database: String,
+    pub source_ref: String,
+    #[serde(default)]
+    pub collections: Option<Vec<String>>,
+    #[serde(default)]
+    pub drop_existing: bool,
+    #[serde(default = "yes")]
+    pub restore_options: bool,
+    #[serde(default = "yes")]
+    pub restore_indexes: bool,
+    #[serde(default = "yes")]
+    pub stop_on_error: bool,
+    #[serde(default)]
+    pub objcheck: bool,
+    #[serde(default = "batch_size")]
+    pub batch_size: usize,
+    #[serde(default)]
+    pub execution_id: Option<String>,
+}
+fn yes() -> bool {
+    true
+}
+fn batch_size() -> usize {
+    500
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct MongoDatabaseDumpProgress {
+    pub task_id: String,
+    pub status: String,
+    pub phase: String,
+    pub collection: Option<String>,
+    pub collections_done: usize,
+    pub collections_total: usize,
+    pub documents_read: u64,
+    pub documents_written: u64,
+    pub documents_failed: u64,
+    pub indexes_created: usize,
+    pub elapsed_ms: u128,
+    pub error_message: Option<String>,
+    pub file_path: Option<String>,
+    #[serde(default)]
+    pub bytes_processed: u64,
+    #[serde(default)]
+    pub documents_validated: u64,
+}
+
+impl MongoDatabaseDumpProgress {
+    pub fn initial(task_id: &str) -> Self {
+        Self {
+            task_id: task_id.into(),
+            status: "running".into(),
+            phase: "preparing".into(),
+            collection: None,
+            collections_done: 0,
+            collections_total: 0,
+            documents_read: 0,
+            documents_written: 0,
+            documents_failed: 0,
+            indexes_created: 0,
+            elapsed_ms: 0,
+            error_message: None,
+            file_path: None,
+            bytes_processed: 0,
+            documents_validated: 0,
+        }
+    }
+    fn emit(&mut self, started: Instant, callback: &mut impl FnMut(Self)) {
+        self.elapsed_ms = started.elapsed().as_millis();
+        callback(self.clone());
+    }
+}
+
+#[derive(Clone)]
+struct PreparedCollection {
+    database: String,
+    metadata: CollectionMetadata,
+    path: Option<PathBuf>,
+    documents: u64,
+    size_bytes: u64,
+    metadata_path: Option<PathBuf>,
+}
+
+pub fn release_mongodb_restore_source(source_ref: &str) -> bool {
+    source::release(source_ref)
+}
+
+pub async fn prepare_mongodb_restore_source(
+    request: MongoRestoreSourceRequest,
+) -> Result<MongoRestoreSourcePreview, String> {
+    source::prepare(request, None, None).await
+}
+
+fn catalog(entries: &[PreparedCollection], exact: bool) -> MongoDumpCatalog {
+    MongoDumpCatalog {
+        databases: entries.iter().map(|entry| entry.database.clone()).collect::<BTreeSet<_>>().into_iter().collect(),
+        collections: entries
+            .iter()
+            .map(|entry| MongoDumpCollection {
+                database: entry.database.clone(),
+                name: entry.metadata.collection_name.clone(),
+                kind: entry.metadata.kind.clone(),
+                documents: if exact { Some(entry.documents) } else { None },
+                size_bytes: entry.size_bytes,
+                indexes: entry.metadata.restore_indexes().len(),
+                source_files: Vec::new(),
+            })
+            .collect(),
+    }
+}
+
+async fn native_client(state: &AppState, id: &str, database: &str) -> Result<mongodb::Client, String> {
+    metadata::validate_database(database)?;
+    state.get_or_create_pool(id, Some(database)).await?;
+    match state.pool_handle(id).await {
+        Some(PoolKind::MongoDb(client)) => Ok(client.clone()),
+        _ => Err("Database dump/restore requires the native MongoDB driver".into()),
+    }
+}
+
+async fn database_entries(client: &mongodb::Client, database: &str) -> Result<Vec<PreparedCollection>, String> {
+    let db = client.database(database);
+    let mut cursor = db.list_collections().await.map_err(|e| e.to_string())?;
+    let mut entries = Vec::new();
+    while let Some(spec) = cursor.try_next().await.map_err(|e| e.to_string())? {
+        if spec.name.starts_with("system.") && spec.name != "system.js" {
+            continue;
+        }
+        let mut metadata = CollectionMetadata::empty(&spec.name);
+        metadata.kind =
+            crate::db::mongo_driver::MongoCollectionKind::from_driver_type(&spec.collection_type).as_str().into();
+        metadata.options = mongodb::bson::to_document(&spec.options).map_err(|e| e.to_string())?;
+        metadata.validate()?;
+        if !metadata.is_view() {
+            let mut indexes = db.collection::<Document>(&spec.name).list_indexes().await.map_err(|e| e.to_string())?;
+            while let Some(index) = indexes.try_next().await.map_err(|e| e.to_string())? {
+                metadata.indexes.push(mongodb::bson::to_document(&index).map_err(|e| e.to_string())?);
+            }
+        }
+        metadata.validate()?;
+        entries.push(PreparedCollection {
+            database: database.into(),
+            metadata,
+            path: None,
+            documents: 0,
+            size_bytes: 0,
+            metadata_path: None,
+        });
+    }
+    entries.sort_by(|a, b| a.metadata.collection_name.cmp(&b.metadata.collection_name));
+    Ok(entries)
+}
+
+pub async fn inspect_mongodb_database_dump(
+    state: &AppState,
+    id: &str,
+    database: &str,
+) -> Result<MongoDumpCatalog, String> {
+    let client = native_client(state, id, database).await?;
+    let entries = database_entries(&client, database).await?;
+    let mut catalog = catalog(&entries, false);
+    catalog.databases = vec![database.into()];
+    Ok(catalog)
+}
+
+fn select_entries(
+    entries: &[PreparedCollection],
+    selected: &Option<Vec<String>>,
+) -> Result<Vec<PreparedCollection>, String> {
+    match selected {
+        None => Ok(entries.to_vec()),
+        Some(names) => {
+            let mut requested: HashSet<_> = names.iter().collect();
+            let chosen =
+                entries.iter().filter(|entry| requested.remove(&entry.metadata.collection_name)).cloned().collect();
+            if !requested.is_empty() {
+                return Err("A selected collection is not present in the dump plan".into());
+            }
+            Ok(chosen)
+        }
+    }
+}
+
+fn ordered_entries(entries: &[PreparedCollection]) -> Result<Vec<PreparedCollection>, String> {
+    let names: HashMap<_, _> = entries
+        .iter()
+        .enumerate()
+        .map(|(i, entry)| ((entry.database.clone(), entry.metadata.collection_name.clone()), i))
+        .collect();
+    let mut ordered = Vec::with_capacity(entries.len());
+    // 0 = unseen, 1 = active dependency, 2 = emitted. Uploaded metadata may
+    // contain arbitrarily deep chains, so keep the traversal off the call stack.
+    let mut states = vec![0u8; entries.len()];
+    for (i, entry) in entries.iter().enumerate().filter(|(_, e)| !e.metadata.is_view()) {
+        states[i] = 2;
+        ordered.push(entry.clone());
+    }
+    for i in 0..entries.len() {
+        let mut pending = vec![(i, false)];
+        while let Some((index, finish)) = pending.pop() {
+            if finish {
+                states[index] = 2;
+                ordered.push(entries[index].clone());
+                continue;
+            }
+            match states[index] {
+                2 => continue,
+                1 => return Err("Cyclic view dependencies in dump metadata".into()),
+                _ => {}
+            }
+            states[index] = 1;
+            pending.push((index, true));
+            let entry = &entries[index];
+            if let Ok(base) = entry.metadata.options.get_str("viewOn") {
+                if let Some(&parent) = names.get(&(entry.database.clone(), base.into())) {
+                    pending.push((parent, false));
+                }
+            }
+        }
+    }
+    Ok(ordered)
+}
+
+fn cancelled(flag: &AtomicBool) -> Result<(), String> {
+    if flag.load(Ordering::Relaxed) {
+        Err("MongoDB dump/restore cancelled".into())
+    } else {
+        Ok(())
+    }
+}
+
+async fn blocking_work<T: Send + 'static>(
+    task_id: &str,
+    is_cancelled: &mut impl FnMut(&str) -> CancelFuture,
+    work: impl FnOnce(Arc<AtomicBool>) -> Result<T, String> + Send + 'static,
+) -> Result<T, String> {
+    if is_cancelled(task_id).await {
+        return Err("MongoDB dump/restore cancelled".into());
+    }
+    struct CancelOnDrop(Arc<AtomicBool>);
+    impl Drop for CancelOnDrop {
+        fn drop(&mut self) {
+            self.0.store(true, Ordering::Relaxed);
+        }
+    }
+    let flag = Arc::new(AtomicBool::new(false));
+    let _guard = CancelOnDrop(flag.clone());
+    let signal = flag.clone();
+    let mut task = tokio::task::spawn_blocking(move || work(signal));
+    let mut interval = tokio::time::interval(Duration::from_millis(100));
+    loop {
+        tokio::select! {
+            result = &mut task => return result.map_err(|e| e.to_string())?,
+            _ = interval.tick() => if is_cancelled(task_id).await { flag.store(true, Ordering::Relaxed); },
+        }
+    }
+}
+
+pub async fn dump_mongodb_database<C, F>(
+    state: &AppState,
+    request: &MongoDatabaseDumpRequest,
+    mut is_cancelled: C,
+    mut on_progress: F,
+) -> Result<MongoDatabaseDumpProgress, String>
+where
+    C: FnMut(&str) -> CancelFuture,
+    F: FnMut(MongoDatabaseDumpProgress),
+{
+    let started = Instant::now();
+    let mut progress = MongoDatabaseDumpProgress::initial(&request.task_id);
+    progress.emit(started, &mut on_progress);
+    let result: Result<(), String> = async {
+        if is_cancelled(&request.task_id).await {
+            return Err("MongoDB dump/restore cancelled".into());
+        }
+        let client = native_client(state, &request.connection_id, &request.database).await?;
+        let mut entries = select_entries(&database_entries(&client, &request.database).await?, &request.collections)?;
+        progress.collections_total = entries.len();
+        let target = PathBuf::from(&request.file_path);
+        if target.exists() {
+            return Err(match request.format {
+                MongoDumpFormat::Directory => "Dump destination already exists; choose a new directory",
+                MongoDumpFormat::Archive => "Dump destination already exists; choose a new archive file",
+            }
+            .into());
+        }
+        let parent = target.parent().filter(|p| !p.as_os_str().is_empty()).unwrap_or(Path::new("."));
+        std::fs::create_dir_all(parent).map_err(|e| e.to_string())?;
+        let working =
+            tempfile::Builder::new().prefix(".dbx-mongo-dump-").tempdir_in(parent).map_err(|e| e.to_string())?;
+        for (index, entry) in entries.iter_mut().enumerate() {
+            if is_cancelled(&request.task_id).await {
+                return Err("MongoDB dump/restore cancelled".into());
+            }
+            progress.collection = Some(entry.metadata.collection_name.clone());
+            progress.phase = "data".into();
+            progress.emit(started, &mut on_progress);
+            if !entry.metadata.is_view() {
+                let path = working.path().join(format!("{index}.bson"));
+                let child = MongoExportRequest {
+                    export_id: request.task_id.clone(),
+                    connection_id: request.connection_id.clone(),
+                    database: request.database.clone(),
+                    collection: entry.metadata.collection_name.clone(),
+                    filter: None,
+                    sort: None,
+                    projection: None,
+                    collation: None,
+                    format: MongoExportFormat::Bson,
+                    include_header: false,
+                    gzip: false,
+                    file_path: path.to_string_lossy().into_owned(),
+                    execution_id: None,
+                };
+                let base = progress.documents_read;
+                let summary = export_mongodb_query_core(
+                    state,
+                    &child,
+                    |id| is_cancelled(id),
+                    |child| {
+                        progress.documents_read = base + child.documents_read;
+                        progress.emit(started, &mut on_progress);
+                    },
+                )
+                .await?;
+                entry.documents = summary.documents_exported;
+                entry.size_bytes = std::fs::metadata(&path).map_err(|e| e.to_string())?.len();
+                entry.path = Some(path);
+            }
+            progress.collections_done += 1;
+        }
+        progress.phase = "archive".into();
+        progress.emit(started, &mut on_progress);
+        let server_version = client
+            .database(&request.database)
+            .run_command(doc! { "buildInfo": 1 })
+            .await
+            .map_err(|e| e.to_string())?
+            .get_str("version")
+            .unwrap_or("")
+            .to_string();
+        let database = request.database.clone();
+        let gzip = request.gzip;
+        let format = request.format;
+        blocking_work(&request.task_id, &mut is_cancelled, move |flag| {
+            let _working = working;
+            let parent = target.parent().filter(|p| !p.as_os_str().is_empty()).unwrap_or(Path::new("."));
+            match format {
+                MongoDumpFormat::Archive => {
+                    let temp = tempfile::NamedTempFile::new_in(parent).map_err(|e| e.to_string())?;
+                    directory::write_file(temp.path(), gzip, |writer| {
+                        archive::pack(writer, &entries, &server_version, || cancelled(&flag))
+                    })?;
+                    cancelled(&flag)?;
+                    if target.exists() {
+                        return Err("Dump destination already exists".into());
+                    }
+                    temp.persist(&target).map_err(|e| e.to_string())?;
+                }
+                MongoDumpFormat::Directory => {
+                    let output = tempfile::Builder::new()
+                        .prefix(".dbx-dump-output-")
+                        .tempdir_in(parent)
+                        .map_err(|e| e.to_string())?;
+                    directory::publish_files(output.path(), &database, &entries, gzip, || cancelled(&flag))?;
+                    cancelled(&flag)?;
+                    if target.exists() {
+                        return Err("Dump destination already exists".into());
+                    }
+                    std::fs::rename(output.path(), &target).map_err(|e| e.to_string())?;
+                }
+            }
+            Ok(())
+        })
+        .await?;
+        progress.file_path = Some(request.file_path.clone());
+        Ok(())
+    }
+    .await;
+    finish_progress(progress, result, started, &mut on_progress)
+}
+
+pub async fn restore_mongodb_database<C, F>(
+    state: &AppState,
+    request: &MongoDatabaseRestoreRequest,
+    mut is_cancelled: C,
+    mut on_progress: F,
+) -> Result<MongoDatabaseDumpProgress, String>
+where
+    C: FnMut(&str) -> CancelFuture,
+    F: FnMut(MongoDatabaseDumpProgress),
+{
+    let started = Instant::now();
+    let mut progress = MongoDatabaseDumpProgress::initial(&request.task_id);
+    progress.emit(started, &mut on_progress);
+    let result: Result<(), String> = async {
+        if let Some(name) = crate::query::connection_readonly_name(state, &request.connection_id).await {
+            return Err(format!("Read-only connection '{name}': restore blocked"));
+        }
+        if is_cancelled(&request.task_id).await {
+            return Err("MongoDB dump/restore cancelled".into());
+        }
+        crate::mongodb_import_export::clamp_batch_size(request.batch_size).map_err(|e| e.display_message())?;
+        let source = source::get(&request.source_ref)?;
+        let source_entries: Vec<_> =
+            source.entries.iter().filter(|entry| entry.database == request.source_database).cloned().collect();
+        if source_entries.is_empty() && !source.entries.is_empty() {
+            return Err("Source database is not present in the dump".into());
+        }
+        let entries = ordered_entries(&select_entries(&source_entries, &request.collections)?)?;
+        progress.collections_total = entries.len();
+        source.check_ready(&entries)?;
+        let client = native_client(state, &request.connection_id, &request.database).await?;
+        let db = client.database(&request.database);
+        let existing = db
+            .list_collections()
+            .await
+            .map_err(|e| e.to_string())?
+            .try_collect::<Vec<_>>()
+            .await
+            .map_err(|e| e.to_string())?;
+        for entry in &entries {
+            entry.metadata.validate()?;
+            if let Some(target) = existing.iter().find(|target| target.name == entry.metadata.collection_name) {
+                if !request.drop_existing
+                    && (entry.metadata.is_view()
+                        || !matches!(target.collection_type, mongodb::results::CollectionType::Collection))
+                {
+                    return Err(format!(
+                        "{}: replacing an existing view or non-regular collection requires dropExisting",
+                        target.name
+                    ));
+                }
+            }
+            if let Ok(base) = entry.metadata.options.get_str("viewOn") {
+                if !entries.iter().any(|e| e.metadata.collection_name == base)
+                    && !existing.iter().any(|e| e.name == base)
+                {
+                    return Err(format!(
+                        "Missing backing collection for view {}: {base}",
+                        entry.metadata.collection_name
+                    ));
+                }
+            }
+        }
+        let existing_names = existing.iter().map(|e| e.name.clone()).collect::<HashSet<_>>();
+        stream::restore_data(
+            state,
+            &client,
+            request,
+            source,
+            &entries,
+            &existing_names,
+            &mut progress,
+            started,
+            &mut is_cancelled,
+            &mut on_progress,
+        )
+        .await?;
+        for entry in &entries {
+            if is_cancelled(&request.task_id).await {
+                return Err("MongoDB dump/restore cancelled".into());
+            }
+            stream::writable(state, &request.connection_id).await?;
+            let name = &entry.metadata.collection_name;
+            progress.collection = Some(name.clone());
+            progress.phase = if entry.metadata.is_view() { "views" } else { "indexes" }.into();
+            progress.emit(started, &mut on_progress);
+            if entry.metadata.is_view() {
+                stream::initialize_collection(&client, request, entry, existing_names.contains(name)).await?;
+                progress.collections_done += 1;
+            } else if request.restore_indexes {
+                for index in entry.metadata.restore_indexes() {
+                    if is_cancelled(&request.task_id).await {
+                        return Err("MongoDB dump/restore cancelled".into());
+                    }
+                    db.run_command(doc! { "createIndexes": name, "indexes": [index] })
+                        .await
+                        .map_err(|e| e.to_string())?;
+                    progress.indexes_created += 1;
+                }
+            }
+        }
+        Ok(())
+    }
+    .await;
+    finish_progress(progress, result, started, &mut on_progress)
+}
+
+fn finish_progress(
+    mut progress: MongoDatabaseDumpProgress,
+    result: Result<(), String>,
+    started: Instant,
+    callback: &mut impl FnMut(MongoDatabaseDumpProgress),
+) -> Result<MongoDatabaseDumpProgress, String> {
+    progress.phase = "done".into();
+    progress.status = match &result {
+        Ok(()) => "done",
+        Err(error)
+            if matches!(error.as_str(), "MongoDB dump/restore cancelled" | "Import cancelled" | "Export cancelled") =>
+        {
+            "cancelled"
+        }
+        Err(_) => "error",
+    }
+    .into();
+    progress.error_message = result.as_ref().err().cloned();
+    progress.emit(started, callback);
+    result.map(|_| progress)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn view_ordering_handles_deep_dependency_chains() {
+        let count = 20_000;
+        let entries = (0..count)
+            .map(|index| {
+                let mut metadata = CollectionMetadata::empty(&format!("view_{index}"));
+                if index + 1 < count {
+                    metadata.kind = "view".into();
+                    metadata.options = doc! { "viewOn": format!("view_{}", index + 1), "pipeline": [] };
+                }
+                PreparedCollection {
+                    database: "source".into(),
+                    metadata,
+                    path: None,
+                    documents: 0,
+                    size_bytes: 0,
+                    metadata_path: None,
+                }
+            })
+            .collect::<Vec<_>>();
+        let ordered = ordered_entries(&entries).unwrap();
+        assert_eq!(ordered.len(), count);
+        for (entry, index) in ordered.iter().zip((0..count).rev()) {
+            assert_eq!(entry.metadata.collection_name, format!("view_{index}"));
+        }
+    }
+}

crates/dbx-core/src/mongodb_dump/archive.rs

@@ -0,0 +1,290 @@
+//! MongoDB Database Tools archive format 0.1, see common/archive/spec.md upstream.
+use std::collections::HashMap;
+use std::fs::File;
+use std::io::{BufReader, Read, Write};
+
+use crc::{Crc, Digest, CRC_64_XZ};
+use mongodb::bson::{doc, Document};
+
+use super::{metadata::CollectionMetadata, PreparedCollection};
+
+const MAGIC: u32 = 0x8199_e26d;
+const TERMINATOR: [u8; 4] = [255; 4];
+// Go hash/crc64 with the ECMA table reflects bits and complements the initial/final state.
+const CRC: Crc<u64> = Crc::<u64>::new(&CRC_64_XZ);
+pub(super) const MAX_DOCUMENT: usize = 16 * 1024 * 1024;
+
+pub(super) enum Frame {
+    End,
+    Terminator,
+    Document(Vec<u8>),
+}
+
+pub(super) fn read_frame(reader: &mut dyn Read) -> Result<Frame, String> {
+    let frame = read_raw_frame(reader)?;
+    if let Frame::Document(bytes) = &frame {
+        mongodb::bson::from_slice::<Document>(bytes).map_err(|e| format!("Invalid BSON document: {e}"))?;
+    }
+    Ok(frame)
+}
+
+pub(super) fn read_raw_frame(reader: &mut dyn Read) -> Result<Frame, String> {
+    let mut header = [0; 4];
+    loop {
+        match reader.read(&mut header[..1]) {
+            Ok(0) => return Ok(Frame::End),
+            Ok(_) => break,
+            Err(e) if e.kind() == std::io::ErrorKind::Interrupted => continue,
+            Err(e) => return Err(e.to_string()),
+        }
+    }
+    reader.read_exact(&mut header[1..]).map_err(|e| format!("Truncated BSON header: {e}"))?;
+    if header == TERMINATOR {
+        return Ok(Frame::Terminator);
+    }
+    let length = i32::from_le_bytes(header);
+    if !(5..=MAX_DOCUMENT as i32).contains(&length) {
+        return Err(format!("Invalid BSON frame length: {length}"));
+    }
+    let mut bytes = vec![0; length as usize];
+    bytes[..4].copy_from_slice(&header);
+    reader.read_exact(&mut bytes[4..]).map_err(|e| format!("Truncated BSON document: {e}"))?;
+    mongodb::bson::RawDocument::from_bytes(&bytes).map_err(|e| format!("Invalid BSON document: {e}"))?;
+    Ok(Frame::Document(bytes))
+}
+
+fn document(frame: Frame) -> Result<Document, String> {
+    match frame {
+        Frame::Document(bytes) => mongodb::bson::from_slice(&bytes).map_err(|e| e.to_string()),
+        _ => Err("Expected an archive BSON header".into()),
+    }
+}
+
+fn string(doc: &Document, key: &str) -> Result<String, String> {
+    doc.get_str(key).map(String::from).map_err(|_| format!("Archive header is missing {key}"))
+}
+
+fn write_document(writer: &mut dyn Write, doc: &Document) -> Result<(), String> {
+    let bytes = mongodb::bson::to_vec(doc).map_err(|e| e.to_string())?;
+    if bytes.len() > MAX_DOCUMENT {
+        return Err("Archive metadata exceeds the BSON document limit".into());
+    }
+    writer.write_all(&bytes).map_err(|e| e.to_string())
+}
+
+pub(super) fn inspect(
+    reader: &mut dyn Read,
+    mut check: impl FnMut() -> Result<(), String>,
+) -> Result<Vec<PreparedCollection>, String> {
+    let mut magic = [0; 4];
+    reader.read_exact(&mut magic).map_err(|e| format!("Invalid MongoDB archive: {e}"))?;
+    if u32::from_le_bytes(magic) != MAGIC {
+        return Err("Not a MongoDB --archive file".into());
+    }
+    let header = document(read_frame(reader)?)?;
+    if header.get_str("version") != Ok("0.1") {
+        return Err("Unsupported MongoDB archive version".into());
+    }
+    let mut entries = Vec::<PreparedCollection>::new();
+    let mut namespaces = HashMap::new();
+    let mut metadata_bytes = 0usize;
+    loop {
+        check()?;
+        let frame = read_frame(reader)?;
+        if matches!(frame, Frame::Terminator) {
+            break;
+        }
+        let header = document(frame)?;
+        let database = string(&header, "db")?;
+        super::metadata::validate_database(&database)?;
+        let name = string(&header, "collection")?;
+        let json = string(&header, "metadata")?;
+        metadata_bytes += json.len();
+        if entries.len() >= 100_000 || metadata_bytes > 128 * 1024 * 1024 {
+            return Err("Archive metadata exceeds the supported limit".into());
+        }
+        let metadata = if json.is_empty() {
+            CollectionMetadata::empty(&name)
+        } else {
+            CollectionMetadata::from_json(&json, &name)?
+        };
+        metadata.validate()?;
+        if metadata.collection_name != name {
+            return Err("Archive namespace does not match collection metadata".into());
+        }
+        if header.get_str("type").is_ok_and(|kind| kind == "timeseries") {
+            return Err("Time-series archives require server-specific restore support".into());
+        }
+        let index = entries.len();
+        if namespaces.insert((database.clone(), name), index).is_some() {
+            return Err("Duplicate archive namespace".into());
+        }
+        entries.push(PreparedCollection {
+            database,
+            metadata,
+            path: None,
+            documents: 0,
+            size_bytes: 0,
+            metadata_path: None,
+        });
+    }
+    Ok(entries)
+}
+
+pub(super) enum StreamEvent {
+    Document(usize, Vec<u8>),
+    End(usize),
+}
+
+pub(super) fn stream(
+    reader: &mut dyn Read,
+    expected: &[PreparedCollection],
+    mut check: impl FnMut() -> Result<(), String>,
+    mut emit: impl FnMut(StreamEvent) -> Result<(), String>,
+) -> Result<(), String> {
+    let entries = inspect(reader, &mut check)?;
+    let namespaces: HashMap<_, _> = entries
+        .iter()
+        .enumerate()
+        .map(|(i, e)| ((e.database.clone(), e.metadata.collection_name.clone()), i))
+        .collect();
+    if entries.len() != expected.len() {
+        return Err("Archive metadata changed after preview".into());
+    }
+    let mut mapped = Vec::with_capacity(entries.len());
+    for entry in &entries {
+        let index = expected
+            .iter()
+            .position(|e| e.database == entry.database && e.metadata.collection_name == entry.metadata.collection_name)
+            .ok_or("Archive metadata changed after preview")?;
+        if expected[index].metadata.to_json()? != entry.metadata.to_json()? {
+            return Err("Archive metadata changed after preview".into());
+        }
+        mapped.push(index);
+    }
+    let mut digests: Vec<Digest<'_, u64>> = entries.iter().map(|_| CRC.digest()).collect();
+    let mut finished = vec![false; entries.len()];
+    loop {
+        check()?;
+        let frame = read_frame(reader)?;
+        if matches!(frame, Frame::End) {
+            break;
+        }
+        let header = document(frame)?;
+        let namespace = (string(&header, "db")?, string(&header, "collection")?);
+        let index = *namespaces.get(&namespace).ok_or_else(|| "Unknown namespace in archive data".to_string())?;
+        if finished[index] {
+            return Err("Archive contains data after namespace EOF".into());
+        }
+        let eof = header.get_bool("EOF").map_err(|_| "Archive header is missing EOF".to_string())?;
+        let checksum = header.get_i64("CRC").map_err(|_| "Archive header is missing CRC".to_string())? as u64;
+        if eof {
+            if !matches!(read_frame(reader)?, Frame::Terminator) {
+                return Err("Invalid archive EOF terminator".into());
+            }
+            if digests[index].clone().finalize() != checksum {
+                return Err(format!("Archive checksum mismatch: {}.{}", namespace.0, namespace.1));
+            }
+            finished[index] = true;
+            emit(StreamEvent::End(mapped[index]))?;
+            continue;
+        }
+        if checksum != 0 {
+            return Err("Non-EOF archive header contains a checksum".into());
+        }
+        let entry = &entries[index];
+        loop {
+            check()?;
+            match read_raw_frame(reader)? {
+                Frame::Document(bytes) => {
+                    if entry.metadata.is_view() {
+                        return Err("Archive contains BSON documents for a view".into());
+                    }
+                    digests[index].update(&bytes);
+                    emit(StreamEvent::Document(mapped[index], bytes))?;
+                }
+                Frame::Terminator => break,
+                Frame::End => return Err("Truncated archive namespace segment".into()),
+            }
+        }
+    }
+    for (index, entry) in entries.iter().enumerate() {
+        if !finished[index] {
+            return Err(format!("Missing archive EOF: {}.{}", entry.database, entry.metadata.collection_name));
+        }
+    }
+    Ok(())
+}
+
+pub(super) fn pack(
+    writer: &mut dyn Write,
+    entries: &[PreparedCollection],
+    server_version: &str,
+    mut check: impl FnMut() -> Result<(), String>,
+) -> Result<(), String> {
+    writer.write_all(&MAGIC.to_le_bytes()).map_err(|e| e.to_string())?;
+    write_document(
+        writer,
+        &doc! { "concurrent_collections": 1i32, "version": "0.1", "server_version": server_version, "tool_version": concat!("DBX ", env!("CARGO_PKG_VERSION")) },
+    )?;
+    for entry in entries {
+        check()?;
+        write_document(
+            writer,
+            &doc! { "db": &entry.database, "collection": &entry.metadata.collection_name, "metadata": entry.metadata.to_json()?, "size": entry.documents as i64, "type": if entry.metadata.is_view() { "view" } else { "" } },
+        )?;
+    }
+    writer.write_all(&TERMINATOR).map_err(|e| e.to_string())?;
+    for entry in entries {
+        check()?;
+        if entry.metadata.is_view() {
+            write_document(
+                writer,
+                &doc! { "db": &entry.database, "collection": &entry.metadata.collection_name, "EOF": true, "CRC": 0i64 },
+            )?;
+            writer.write_all(&TERMINATOR).map_err(|e| e.to_string())?;
+            continue;
+        }
+        let mut digest = CRC.digest();
+        let path = entry.path.as_ref().ok_or_else(|| "Missing collection BSON file".to_string())?;
+        let mut reader = BufReader::new(File::open(path).map_err(|e| e.to_string())?);
+        let mut opened = false;
+        loop {
+            check()?;
+            match read_frame(&mut reader)? {
+                Frame::End => break,
+                Frame::Terminator => return Err("Unexpected terminator in BSON file".into()),
+                Frame::Document(bytes) => {
+                    if !opened {
+                        write_document(
+                            writer,
+                            &doc! { "db": &entry.database, "collection": &entry.metadata.collection_name, "EOF": false, "CRC": 0i64 },
+                        )?;
+                        opened = true;
+                    }
+                    writer.write_all(&bytes).map_err(|e| e.to_string())?;
+                    digest.update(&bytes);
+                }
+            }
+        }
+        if opened {
+            writer.write_all(&TERMINATOR).map_err(|e| e.to_string())?;
+        }
+        write_document(
+            writer,
+            &doc! { "db": &entry.database, "collection": &entry.metadata.collection_name, "EOF": true, "CRC": digest.finalize() as i64 },
+        )?;
+        writer.write_all(&TERMINATOR).map_err(|e| e.to_string())?;
+    }
+    writer.flush().map_err(|e| e.to_string())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn checksum_matches_go_crc64_ecma() {
+        assert_eq!(CRC.checksum(b"123456789"), 0x995d_c9bb_df19_39fa);
+    }
+}

crates/dbx-core/src/mongodb_dump/directory.rs

@@ -0,0 +1,302 @@
+use std::collections::BTreeMap;
+use std::fs::File;
+use std::io::{BufReader, Read, Write};
+use std::path::{Path, PathBuf};
+
+use flate2::{read::MultiGzDecoder, write::GzEncoder, Compression};
+
+use super::{
+    archive::MAX_DOCUMENT,
+    metadata::{self, CollectionMetadata},
+    PreparedCollection,
+};
+
+pub(super) fn reader(path: &Path, gzip: bool) -> Result<Box<dyn Read>, String> {
+    regular_file(path)?;
+    let reader = BufReader::new(File::open(path).map_err(|e| e.to_string())?);
+    if gzip {
+        Ok(Box::new(BufReader::with_capacity(64 * 1024, MultiGzDecoder::new(reader))))
+    } else {
+        Ok(Box::new(reader))
+    }
+}
+
+fn reject_link(path: &Path) -> Result<std::fs::Metadata, String> {
+    let metadata = std::fs::symlink_metadata(path).map_err(|e| e.to_string())?;
+    let is_link = metadata.file_type().is_symlink();
+    #[cfg(windows)]
+    let is_link = {
+        use std::os::windows::fs::MetadataExt;
+        is_link || metadata.file_attributes() & 0x400 != 0
+    };
+    if is_link {
+        return Err(format!("Dump sources cannot contain links: {}", path.display()));
+    }
+    Ok(metadata)
+}
+
+pub(super) fn regular_file(path: &Path) -> Result<(), String> {
+    if !reject_link(path)?.is_file() {
+        return Err(format!("Not a regular dump file: {}", path.display()));
+    }
+    Ok(())
+}
+
+pub(super) fn copy_stream(
+    reader: &mut dyn Read,
+    writer: &mut dyn Write,
+    check: &mut impl FnMut() -> Result<(), String>,
+) -> Result<(), String> {
+    let mut buffer = [0; 64 * 1024];
+    loop {
+        check()?;
+        let length = reader.read(&mut buffer).map_err(|e| e.to_string())?;
+        if length == 0 {
+            break;
+        }
+        writer.write_all(&buffer[..length]).map_err(|e| e.to_string())?;
+    }
+    Ok(())
+}
+
+pub(super) fn write_file(
+    path: &Path,
+    gzip: bool,
+    write: impl FnOnce(&mut dyn Write) -> Result<(), String>,
+) -> Result<(), String> {
+    let file = std::io::BufWriter::new(File::create(path).map_err(|e| e.to_string())?);
+    if gzip {
+        let mut writer = GzEncoder::new(file, Compression::default());
+        write(&mut writer)?;
+        writer.finish().map_err(|e| e.to_string())?.flush().map_err(|e| e.to_string())
+    } else {
+        let mut writer = file;
+        write(&mut writer)?;
+        writer.flush().map_err(|e| e.to_string())
+    }
+}
+
+#[derive(Default)]
+struct Files {
+    data: Option<PathBuf>,
+    metadata: Option<PathBuf>,
+}
+
+fn collect(
+    path: &Path,
+    depth: usize,
+    gzip: bool,
+    groups: &mut BTreeMap<(String, String), Files>,
+) -> Result<(), String> {
+    if depth > 3 || !reject_link(path)?.is_dir() {
+        return Err("Invalid dump directory layout".into());
+    }
+    let database =
+        path.file_name().and_then(|name| name.to_str()).ok_or_else(|| "Invalid database directory name".to_string())?;
+    let mut subdirectories = Vec::new();
+    let mut has_data = false;
+    for entry in std::fs::read_dir(path).map_err(|e| e.to_string())? {
+        let entry = entry.map_err(|e| e.to_string())?;
+        let path = entry.path();
+        let info = reject_link(&path)?;
+        if info.is_dir() {
+            subdirectories.push(path);
+            continue;
+        }
+        if !info.is_file() {
+            return Err("Unsupported dump directory entry".into());
+        }
+        let filename = entry.file_name().to_str().ok_or_else(|| "Invalid dump filename".to_string())?.to_string();
+        let (base, metadata_file, compressed) = if let Some(base) = filename.strip_suffix(".metadata.json.gz") {
+            (base, true, true)
+        } else if let Some(base) = filename.strip_suffix(".bson.gz") {
+            (base, false, true)
+        } else if let Some(base) = filename.strip_suffix(".metadata.json") {
+            (base, true, false)
+        } else if let Some(base) = filename.strip_suffix(".bson") {
+            (base, false, false)
+        } else {
+            continue;
+        };
+        if compressed != gzip {
+            return Err(format!("Gzip option does not match dump file {filename}"));
+        }
+        metadata::validate_database(database)?;
+        has_data = true;
+        let group = groups.entry((database.into(), base.into())).or_default();
+        let slot = if metadata_file { &mut group.metadata } else { &mut group.data };
+        if slot.replace(path).is_some() {
+            return Err("Duplicate dump filename".into());
+        }
+        if groups.len() > 100_000 {
+            return Err("Too many dump collections".into());
+        }
+    }
+    if has_data && !subdirectories.is_empty() {
+        return Err(
+            "Mixed database files and directories; select a single database directory or a dump root without oplog"
+                .into(),
+        );
+    }
+    for directory in subdirectories {
+        collect(&directory, depth + 1, gzip, groups)?;
+    }
+    Ok(())
+}
+
+pub(super) fn inspect(
+    path: &Path,
+    gzip: bool,
+    manifest: Option<&[super::MongoDumpFile]>,
+) -> Result<Vec<PreparedCollection>, String> {
+    let mut groups = BTreeMap::new();
+    if let Some(manifest) = manifest {
+        let mut seen = std::collections::HashSet::new();
+        for file in manifest {
+            let relative = super::source::relative_path(&file.path)?;
+            if !seen.insert(file.path.clone()) {
+                return Err("Duplicate dump manifest path".into());
+            }
+            let database = relative
+                .parent()
+                .and_then(Path::file_name)
+                .and_then(|s| s.to_str())
+                .ok_or("Directory uploads require database-relative paths")?;
+            metadata::validate_database(database)?;
+            let filename = relative.file_name().unwrap().to_str().ok_or("Invalid dump filename")?;
+            let plain = filename.strip_suffix(".gz").unwrap_or(filename);
+            let (base, is_metadata) = if let Some(base) = plain.strip_suffix(".metadata.json") {
+                (base, true)
+            } else if let Some(base) = plain.strip_suffix(".bson") {
+                (base, false)
+            } else {
+                continue;
+            };
+            if filename.ends_with(".gz") != gzip {
+                return Err(format!("Gzip option does not match dump file {filename}"));
+            }
+            let group = groups.entry((database.to_string(), base.to_string())).or_insert_with(Files::default);
+            let slot = if is_metadata { &mut group.metadata } else { &mut group.data };
+            if slot.replace(path.join(relative)).is_some() {
+                return Err("Duplicate dump namespace".into());
+            }
+        }
+    } else {
+        collect(path, 0, gzip, &mut groups)?;
+    }
+    let mut entries = Vec::new();
+    let root = path;
+    let mut namespaces = std::collections::HashSet::new();
+    let mut metadata_bytes = 0usize;
+    for ((database, basename), files) in groups {
+        // Truncated official filenames may end mid-escape; collectionName is authoritative.
+        let fallback = metadata::unescape_collection(&basename).unwrap_or_default();
+        let metadata = if let Some(path) = &files.metadata {
+            let mut json = String::new();
+            reader(path, gzip)?.take(MAX_DOCUMENT as u64 + 1).read_to_string(&mut json).map_err(|e| e.to_string())?;
+            if json.len() > MAX_DOCUMENT {
+                return Err("Collection metadata exceeds 16 MiB".into());
+            }
+            metadata_bytes = metadata_bytes.saturating_add(json.len());
+            if metadata_bytes > 128 * 1024 * 1024 {
+                return Err("Dump metadata exceeds 128 MiB".into());
+            }
+            if json.trim().is_empty() {
+                CollectionMetadata::empty(&fallback)
+            } else {
+                CollectionMetadata::from_json(&json, &fallback)?
+            }
+        } else {
+            CollectionMetadata::empty(&fallback)
+        };
+        metadata.validate()?;
+        if !namespaces.insert((database.clone(), metadata.collection_name.clone())) {
+            return Err("Duplicate dump namespace".into());
+        }
+        let mut entry = PreparedCollection {
+            database,
+            metadata,
+            path: None,
+            documents: 0,
+            size_bytes: 0,
+            metadata_path: files.metadata,
+        };
+        match files.data {
+            Some(path) => {
+                if entry.metadata.is_view() {
+                    return Err("View dump unexpectedly contains BSON data".into());
+                }
+                entry.size_bytes = if let Some(manifest) = manifest {
+                    manifest.iter().find(|f| path == root.join(&f.path)).map(|f| f.size_bytes).unwrap_or(0)
+                } else {
+                    regular_file(&path)?;
+                    std::fs::metadata(&path).map_err(|e| e.to_string())?.len()
+                };
+                entry.path = Some(path);
+            }
+            None if !entry.metadata.is_view() => {
+                return Err(format!("Missing BSON file for {}", entry.metadata.collection_name))
+            }
+            None => {}
+        }
+        entries.push(entry);
+    }
+    Ok(entries)
+}
+
+pub(super) fn publish_files(
+    root: &Path,
+    database: &str,
+    entries: &[PreparedCollection],
+    gzip: bool,
+    mut check: impl FnMut() -> Result<(), String>,
+) -> Result<(), String> {
+    let database_dir = root.join(database);
+    std::fs::create_dir_all(&database_dir).map_err(|e| e.to_string())?;
+    let mut names = std::collections::HashSet::new();
+    for entry in entries {
+        check()?;
+        let basename = metadata::escaped_collection(&entry.metadata.collection_name);
+        let filesystem_key = if cfg!(windows) { basename.to_lowercase() } else { basename.clone() };
+        if !names.insert(filesystem_key) {
+            return Err("Collection names collide on this filesystem; use archive format".into());
+        }
+        let suffix = if gzip { ".gz" } else { "" };
+        write_file(&database_dir.join(format!("{basename}.metadata.json{suffix}")), gzip, |writer| {
+            writer.write_all(entry.metadata.to_json()?.as_bytes()).map_err(|e| e.to_string())
+        })?;
+        if let Some(path) = &entry.path {
+            write_file(&database_dir.join(format!("{basename}.bson{suffix}")), gzip, |writer| {
+                copy_stream(&mut *reader(path, false)?, writer, &mut check)
+            })?;
+        }
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn regular_file_accepts_files_and_rejects_directories() {
+        let root = tempfile::tempdir().unwrap();
+        let file = root.path().join("records.bson");
+        std::fs::write(&file, []).unwrap();
+        regular_file(&file).unwrap();
+        assert!(regular_file(root.path()).unwrap_err().contains("Not a regular dump file"));
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn reject_link_rejects_file_and_directory_symlinks() {
+        let root = tempfile::tempdir().unwrap();
+        let file = root.path().join("records.bson");
+        std::fs::write(&file, []).unwrap();
+        for (name, target) in [("file-link", file.as_path()), ("directory-link", root.path())] {
+            let link = root.path().join(name);
+            std::os::unix::fs::symlink(target, &link).unwrap();
+            assert!(reject_link(&link).unwrap_err().contains("Dump sources cannot contain links"));
+        }
+    }
+}

crates/dbx-core/src/mongodb_dump/metadata.rs

@@ -0,0 +1,190 @@
+use base64::Engine;
+use mongodb::bson::{doc, Bson, Document};
+use serde::{Deserialize, Serialize};
+use sha1::{Digest, Sha1};
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct CollectionMetadata {
+    #[serde(default)]
+    pub options: Document,
+    #[serde(default)]
+    pub indexes: Vec<Document>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub uuid: Option<String>,
+    #[serde(default)]
+    pub collection_name: String,
+    #[serde(default, rename = "type")]
+    pub kind: String,
+}
+
+impl CollectionMetadata {
+    pub fn empty(name: &str) -> Self {
+        Self {
+            options: Document::new(),
+            indexes: Vec::new(),
+            uuid: None,
+            collection_name: name.into(),
+            kind: "collection".into(),
+        }
+    }
+
+    pub fn from_json(json: &str, name: &str) -> Result<Self, String> {
+        let value = serde_json::from_str(json).map_err(|e| format!("Invalid metadata for {name}: {e}"))?;
+        let document = crate::db::mongo_driver::json_object_to_document_extended_json(&value)?;
+        let mut metadata: Self =
+            mongodb::bson::from_document(document).map_err(|e| format!("Invalid metadata for {name}: {e}"))?;
+        if metadata.collection_name.is_empty() {
+            metadata.collection_name = name.into();
+        }
+        if metadata.kind.is_empty() {
+            metadata.kind = if metadata.options.contains_key("viewOn") { "view" } else { "collection" }.into();
+        }
+        metadata.validate()?;
+        Ok(metadata)
+    }
+
+    pub fn to_json(&self) -> Result<String, String> {
+        let document = mongodb::bson::to_document(self).map_err(|e| e.to_string())?;
+        let json = crate::db::mongo_driver::document_to_canonical_extended_json(&document).to_string();
+        if json.len() > super::archive::MAX_DOCUMENT {
+            return Err("Collection metadata exceeds 16 MiB".into());
+        }
+        Ok(json)
+    }
+
+    pub fn is_view(&self) -> bool {
+        self.kind == "view"
+    }
+
+    pub fn validate(&self) -> Result<(), String> {
+        validate_collection(&self.collection_name)?;
+        let parsed_options: mongodb::options::CreateCollectionOptions =
+            mongodb::bson::from_document(self.options.clone())
+                .map_err(|e| format!("Invalid collection options for {}: {e}", self.collection_name))?;
+        let known_options = mongodb::bson::to_document(&parsed_options).map_err(|e| e.to_string())?;
+        for key in self.options.keys() {
+            if !known_options.contains_key(key) && key != "autoIndexId" && key != "recordIdsReplicated" {
+                return Err(format!("Unsupported collection option for {}: {key}", self.collection_name));
+            }
+        }
+        if self.kind == "timeseries"
+            || self.options.contains_key("timeseries")
+            || self.options.contains_key("encryptedFields")
+        {
+            return Err(format!(
+                "{}: time-series and encrypted collection dumps require server-specific restore support",
+                self.collection_name
+            ));
+        }
+        if self.kind != "collection" && self.kind != "view" {
+            return Err(format!("Unsupported collection type: {}", self.kind));
+        }
+        if self.is_view() {
+            self.options
+                .get_str("viewOn")
+                .map_err(|_| format!("{}: view metadata is missing viewOn", self.collection_name))?;
+            if !self.indexes.is_empty() {
+                return Err("View metadata cannot contain indexes".into());
+            }
+        }
+        for index in &self.indexes {
+            index.get_document("key").map_err(|_| "Index metadata is missing key".to_string())?;
+            index.get_str("name").map_err(|_| "Index metadata is missing name".to_string())?;
+        }
+        Ok(())
+    }
+
+    pub fn create_command(&self, restore_options: bool) -> Document {
+        let mut command = doc! { "create": &self.collection_name };
+        if restore_options || self.is_view() {
+            let mut options = self.options.clone();
+            options.remove("uuid");
+            options.remove("recordIdsReplicated");
+            options.remove("autoIndexId");
+            // Keep create as the first command key and never accept it from metadata.
+            options.remove("create");
+            command.extend(options);
+        }
+        command
+    }
+
+    pub fn restore_indexes(&self) -> Vec<Document> {
+        self.indexes
+            .iter()
+            .filter(|index| {
+                let key = index.get_document("key").expect("validated index");
+                !(key.len() == 1
+                    && matches!(key.get("_id"), Some(Bson::Int32(1) | Bson::Int64(1)) | Some(Bson::Double(1.0))))
+            })
+            .map(|index| {
+                let mut index = index.clone();
+                index.remove("ns");
+                index.remove("v");
+                index.remove("background");
+                index
+            })
+            .collect()
+    }
+}
+
+pub fn validate_database(name: &str) -> Result<(), String> {
+    if name.is_empty() || name.len() > 63 || name.chars().any(|c| c.is_control() || " /\\.\"$*<>:|?".contains(c)) {
+        return Err("Invalid MongoDB database name".into());
+    }
+    if matches!(name, "admin" | "config" | "local") {
+        return Err(
+            "Internal MongoDB databases and user/role migration are not supported by database dump/restore".into()
+        );
+    }
+    Ok(())
+}
+
+pub fn validate_collection(name: &str) -> Result<(), String> {
+    if name.is_empty()
+        || name.contains('\0')
+        || name.contains('$')
+        || (name.starts_with("system.") && name != "system.js")
+    {
+        return Err(format!("Unsupported MongoDB collection name: {name}"));
+    }
+    Ok(())
+}
+
+pub fn escaped_collection(name: &str) -> String {
+    const ESCAPE: &percent_encoding::AsciiSet =
+        &percent_encoding::NON_ALPHANUMERIC.remove(b'-').remove(b'_').remove(b'.').remove(b'~');
+    let mut escaped = percent_encoding::utf8_percent_encode(name, ESCAPE).to_string().replace("%20", "+");
+    let stem = escaped.split('.').next().unwrap_or_default().to_ascii_uppercase();
+    if matches!(stem.as_str(), "CON" | "PRN" | "AUX" | "NUL")
+        || (stem.len() == 4
+            && (stem.starts_with("COM") || stem.starts_with("LPT"))
+            && matches!(stem.as_bytes()[3], b'1'..=b'9'))
+    {
+        escaped = format!("%{:02X}{}", escaped.as_bytes()[0], &escaped[1..]);
+    }
+    if escaped.len() > 238 {
+        let hash = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(Sha1::digest(name.as_bytes()));
+        escaped = format!("{}%24{hash}", &escaped[..208]);
+    }
+    escaped
+}
+
+pub fn unescape_collection(name: &str) -> Result<String, String> {
+    let bytes = name.as_bytes();
+    let mut position = 0;
+    while position < bytes.len() {
+        if bytes[position] == b'%' {
+            if position + 2 >= bytes.len() || !bytes[position + 1..position + 3].iter().all(u8::is_ascii_hexdigit) {
+                return Err("Invalid escaped collection filename".into());
+            }
+            position += 3;
+        } else {
+            position += 1;
+        }
+    }
+    percent_encoding::percent_decode_str(&name.replace('+', " "))
+        .decode_utf8()
+        .map(|v| v.into_owned())
+        .map_err(|e| e.to_string())
+}

crates/dbx-core/src/mongodb_dump/source.rs

@@ -0,0 +1,247 @@
+use std::collections::HashMap;
+use std::io::Read;
+use std::path::{Component, Path, PathBuf};
+use std::sync::{Arc, Mutex, OnceLock};
+use std::time::{Duration, Instant, SystemTime};
+
+use serde::{Deserialize, Serialize};
+
+use super::{
+    archive, catalog, directory, ordered_entries, select_entries, MongoDatabaseRestoreRequest, MongoDumpFormat,
+    MongoRestoreSourcePreview, MongoRestoreSourceRequest, PreparedCollection,
+};
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct MongoDumpFile {
+    pub path: String,
+    pub size_bytes: u64,
+}
+
+pub(super) struct CatalogSource {
+    _owner: Option<tempfile::TempDir>,
+    request: MongoRestoreSourceRequest,
+    pub entries: Vec<PreparedCollection>,
+    identities: HashMap<PathBuf, (u64, Option<SystemTime>)>,
+    manifest: Option<Vec<MongoDumpFile>>,
+    created: Instant,
+}
+
+static SOURCES: OnceLock<Mutex<HashMap<String, Arc<CatalogSource>>>> = OnceLock::new();
+fn sources() -> &'static Mutex<HashMap<String, Arc<CatalogSource>>> {
+    SOURCES.get_or_init(Default::default)
+}
+
+pub(super) fn get(reference: &str) -> Result<Arc<CatalogSource>, String> {
+    let source = sources()
+        .lock()
+        .unwrap_or_else(|e| e.into_inner())
+        .get(reference)
+        .cloned()
+        .ok_or("Restore source is no longer available")?;
+    if source.created.elapsed() >= Duration::from_secs(86400) {
+        return Err("Restore source has expired".into());
+    }
+    Ok(source)
+}
+
+pub(super) fn release(reference: &str) -> bool {
+    sources().lock().unwrap_or_else(|e| e.into_inner()).remove(reference).is_some()
+}
+
+pub(super) fn relative_path(value: &str) -> Result<&Path, String> {
+    let path = Path::new(value);
+    if value.is_empty()
+        || value.contains(['\\', ':', '\0'])
+        || path.components().count() > 4
+        || path.components().any(|c| !matches!(c, Component::Normal(_)))
+    {
+        return Err("Invalid dump upload path".into());
+    }
+    Ok(path)
+}
+
+fn identity(path: &Path) -> Result<(u64, Option<SystemTime>), String> {
+    directory::regular_file(path)?;
+    let info = std::fs::metadata(path).map_err(|e| e.to_string())?;
+    Ok((info.len(), info.modified().ok()))
+}
+
+pub(super) async fn prepare(
+    request: MongoRestoreSourceRequest,
+    owner: Option<tempfile::TempDir>,
+    manifest: Option<Vec<MongoDumpFile>>,
+) -> Result<MongoRestoreSourcePreview, String> {
+    tokio::task::spawn_blocking(move || {
+        if manifest.as_ref().is_some_and(|m| m.len() > 100_000) {
+            return Err("Too many dump files".into());
+        }
+        let path = Path::new(&request.path);
+        let entries = match request.format {
+            MongoDumpFormat::Directory => directory::inspect(path, request.gzip, manifest.as_deref())?,
+            MongoDumpFormat::Archive => archive::inspect(&mut *directory::reader(path, request.gzip)?, || Ok(()))?,
+        };
+        ordered_entries(&entries)?;
+        let mut catalog = catalog(&entries, false);
+        let mut identities = HashMap::new();
+        for (entry, preview) in entries.iter().zip(catalog.collections.iter_mut()) {
+            for file in entry.path.iter().chain(entry.metadata_path.iter()) {
+                if manifest.is_none() || entry.metadata_path.as_ref() == Some(file) {
+                    identities.insert(file.clone(), identity(file)?);
+                }
+                if let Ok(relative) = file.strip_prefix(path) {
+                    preview.source_files.push(relative.to_string_lossy().replace('\\', "/"));
+                }
+            }
+        }
+        if request.format == MongoDumpFormat::Archive {
+            identities.insert(path.to_owned(), identity(path)?);
+        }
+        let source =
+            Arc::new(CatalogSource { _owner: owner, request, entries, identities, manifest, created: Instant::now() });
+        let mut sources = sources().lock().unwrap_or_else(|e| e.into_inner());
+        sources.retain(|_, s| s.created.elapsed() < Duration::from_secs(86400));
+        if sources.len() >= 32 {
+            return Err("Too many prepared restores; close or release an existing source".into());
+        }
+        let source_ref = uuid::Uuid::new_v4().to_string();
+        sources.insert(source_ref.clone(), source);
+        Ok(MongoRestoreSourcePreview { source_ref, catalog })
+    })
+    .await
+    .map_err(|e| e.to_string())?
+}
+
+pub async fn prepare_owned_mongodb_restore_source(
+    request: MongoRestoreSourceRequest,
+    owner: tempfile::TempDir,
+) -> Result<MongoRestoreSourcePreview, String> {
+    prepare(request, Some(owner), None).await
+}
+
+pub async fn prepare_mongodb_directory_catalog(
+    owner: tempfile::TempDir,
+    manifest: Vec<MongoDumpFile>,
+    gzip: bool,
+) -> Result<MongoRestoreSourcePreview, String> {
+    let request = MongoRestoreSourceRequest {
+        path: owner.path().to_string_lossy().into_owned(),
+        format: MongoDumpFormat::Directory,
+        gzip,
+    };
+    prepare(request, Some(owner), Some(manifest)).await
+}
+
+fn same_metadata(a: &PreparedCollection, b: &PreparedCollection) -> Result<bool, String> {
+    Ok(a.database == b.database && a.metadata.to_json()? == b.metadata.to_json()? && a.size_bytes == b.size_bytes)
+}
+
+/// The server, not the upload payload, determines which files the confirmed selection requires.
+pub fn mongodb_restore_upload_files(request: &MongoDatabaseRestoreRequest) -> Result<Vec<MongoDumpFile>, String> {
+    let source = get(&request.source_ref)?;
+    let manifest = source.manifest.as_ref().ok_or("Source does not require a directory upload")?;
+    let paths = selected(&source, request)?
+        .into_iter()
+        .flat_map(|e| e.path.into_iter().chain(e.metadata_path))
+        .collect::<Vec<_>>();
+    Ok(manifest.iter().filter(|f| paths.contains(&Path::new(&source.request.path).join(&f.path))).cloned().collect())
+}
+
+/// Bind the complete selected upload to the reviewed catalog.
+pub async fn attach_mongodb_restore_upload(
+    request: &MongoDatabaseRestoreRequest,
+    owner: tempfile::TempDir,
+) -> Result<MongoRestoreSourcePreview, String> {
+    let original = get(&request.source_ref)?;
+    if original.manifest.is_none() {
+        return Err("Source does not require a directory upload".into());
+    }
+    let selected = selected(&original, request)?;
+    let next = prepare_owned_mongodb_restore_source(
+        MongoRestoreSourceRequest {
+            path: owner.path().to_string_lossy().into_owned(),
+            format: MongoDumpFormat::Directory,
+            gzip: original.request.gzip,
+        },
+        owner,
+    )
+    .await?;
+    let uploaded = get(&next.source_ref)?;
+    let matches =
+        (|| -> Result<bool, String> {
+            if selected.len() != uploaded.entries.len() {
+                return Ok(false);
+            }
+            for entry in &selected {
+                let Some(found) = uploaded.entries.iter().find(|e| {
+                    e.database == entry.database && e.metadata.collection_name == entry.metadata.collection_name
+                }) else {
+                    return Ok(false);
+                };
+                if !same_metadata(entry, found)? {
+                    return Ok(false);
+                }
+                let expected = entry
+                    .path
+                    .iter()
+                    .chain(entry.metadata_path.iter())
+                    .map(|p| p.strip_prefix(&original.request.path).unwrap().to_owned())
+                    .collect::<Vec<_>>();
+                let actual = found
+                    .path
+                    .iter()
+                    .chain(found.metadata_path.iter())
+                    .map(|p| p.strip_prefix(&uploaded.request.path).unwrap().to_owned())
+                    .collect::<Vec<_>>();
+                if expected != actual {
+                    return Ok(false);
+                }
+            }
+            Ok(true)
+        })();
+    if matches != Ok(true) {
+        release(&next.source_ref);
+        return Err("Uploaded files differ from the confirmed catalog; read the backup again".into());
+    }
+    Ok(next)
+}
+
+fn selected(source: &CatalogSource, request: &MongoDatabaseRestoreRequest) -> Result<Vec<PreparedCollection>, String> {
+    let entries: Vec<_> = source.entries.iter().filter(|e| e.database == request.source_database).cloned().collect();
+    select_entries(&entries, &request.collections)
+}
+
+fn unchanged(source: &CatalogSource, path: &Path) -> Result<(), String> {
+    if source.identities.get(path) != Some(&identity(path)?) {
+        return Err("Backup source changed; read the backup again".into());
+    }
+    Ok(())
+}
+
+impl CatalogSource {
+    pub(super) fn format(&self) -> MongoDumpFormat {
+        self.request.format
+    }
+
+    pub(super) fn check_ready(&self, entries: &[PreparedCollection]) -> Result<(), String> {
+        if self.manifest.is_some() {
+            return Err("Selected directory data has not been uploaded".into());
+        }
+        if self.request.format == MongoDumpFormat::Archive {
+            unchanged(self, Path::new(&self.request.path))?;
+        } else {
+            for entry in entries {
+                for path in entry.path.iter().chain(entry.metadata_path.iter()) {
+                    unchanged(self, path)?;
+                }
+            }
+        }
+        Ok(())
+    }
+
+    pub(super) fn reader(&self, path: Option<&Path>) -> Result<Box<dyn Read>, String> {
+        let path = path.unwrap_or_else(|| Path::new(&self.request.path));
+        unchanged(self, path)?;
+        directory::reader(path, self.request.gzip)
+    }
+}

crates/dbx-core/src/mongodb_dump/stream.rs

@@ -0,0 +1,377 @@
+use std::collections::HashSet;
+use std::sync::{
+    atomic::{AtomicBool, AtomicU64, Ordering},
+    Arc,
+};
+use std::time::{Duration, Instant};
+
+use mongodb::bson::{Document, RawDocumentBuf};
+use tokio::sync::mpsc;
+
+use super::{
+    archive, cancelled, source::CatalogSource, CancelFuture, MongoDatabaseDumpProgress, MongoDatabaseRestoreRequest,
+    MongoDumpFormat, PreparedCollection,
+};
+use crate::{connection::AppState, db::mongo_driver::insert_bson_documents};
+
+const BATCH_BYTES: usize = 8 * 1024 * 1024;
+
+enum Event {
+    Batch { index: usize, documents: Vec<RawDocumentBuf> },
+    End(usize),
+}
+
+struct Batcher {
+    tx: mpsc::Sender<Event>,
+    index: usize,
+    documents: Vec<RawDocumentBuf>,
+    bytes: usize,
+    limit: usize,
+    objcheck: bool,
+}
+
+impl Batcher {
+    fn flush(&mut self) -> Result<(), String> {
+        if self.documents.is_empty() {
+            return Ok(());
+        }
+        let documents = std::mem::take(&mut self.documents);
+        self.bytes = 0;
+        self.tx
+            .blocking_send(Event::Batch { index: self.index, documents })
+            .map_err(|_| "MongoDB dump/restore cancelled".into())
+    }
+
+    fn push(&mut self, index: usize, bytes: Vec<u8>) -> Result<(), String> {
+        if index != self.index || self.bytes.saturating_add(bytes.len()) > BATCH_BYTES {
+            self.flush()?;
+        }
+        self.index = index;
+        let document = checked_document(bytes, self.objcheck)?;
+        self.bytes += document.as_bytes().len();
+        self.documents.push(document);
+        if self.documents.len() >= self.limit {
+            self.flush()?;
+        }
+        Ok(())
+    }
+
+    fn end(&mut self, index: usize) -> Result<(), String> {
+        self.flush()?;
+        self.tx.blocking_send(Event::End(index)).map_err(|_| "MongoDB dump/restore cancelled".into())
+    }
+}
+
+// Like mongorestore's objcheck, deep parsing is optional and occurs in the stream,
+// immediately before a document is queued for insertion. No whole-backup pass.
+fn checked_document(bytes: Vec<u8>, objcheck: bool) -> Result<RawDocumentBuf, String> {
+    let raw = RawDocumentBuf::from_bytes(bytes).map_err(|e| format!("Invalid BSON document: {e}"))?;
+    if objcheck {
+        mongodb::bson::from_slice::<Document>(raw.as_bytes()).map_err(|e| format!("Invalid BSON document: {e}"))?;
+    }
+    Ok(raw)
+}
+
+fn produce(
+    source: Arc<CatalogSource>,
+    entries: Vec<PreparedCollection>,
+    flag: Arc<AtomicBool>,
+    bytes_read: Arc<AtomicU64>,
+    mut batcher: Batcher,
+) -> Result<(), String> {
+    let result = (|| {
+        source.check_ready(&entries)?;
+        if source.format() == MongoDumpFormat::Archive {
+            let indices = source
+                .entries
+                .iter()
+                .map(|original| {
+                    entries.iter().position(|e| {
+                        e.database == original.database
+                            && e.metadata.collection_name == original.metadata.collection_name
+                            && !e.metadata.is_view()
+                    })
+                })
+                .collect::<Vec<_>>();
+            archive::stream(
+                &mut *source.reader(None)?,
+                &source.entries,
+                || cancelled(&flag),
+                |event| {
+                    match event {
+                        archive::StreamEvent::Document(i, bytes) => {
+                            bytes_read.fetch_add(bytes.len() as u64, Ordering::Relaxed);
+                            if let Some(index) = indices[i] {
+                                batcher.push(index, bytes)?;
+                            }
+                        }
+                        archive::StreamEvent::End(i) => {
+                            if let Some(index) = indices[i] {
+                                batcher.end(index)?;
+                            }
+                        }
+                    }
+                    Ok(())
+                },
+            )?;
+        } else {
+            for (index, entry) in entries.iter().enumerate().filter(|(_, e)| !e.metadata.is_view()) {
+                cancelled(&flag)?;
+                let path = entry.path.as_deref().ok_or("Missing collection BSON file")?;
+                let mut reader = source.reader(Some(path))?;
+                loop {
+                    cancelled(&flag)?;
+                    match archive::read_raw_frame(&mut *reader)? {
+                        archive::Frame::End => break,
+                        archive::Frame::Terminator => return Err("Unexpected terminator in collection BSON".into()),
+                        archive::Frame::Document(bytes) => {
+                            bytes_read.fetch_add(bytes.len() as u64, Ordering::Relaxed);
+                            batcher.push(index, bytes)?;
+                        }
+                    }
+                }
+                batcher.end(index)?;
+            }
+        }
+        Ok(())
+    })();
+    // Deliver valid earlier documents even if a later frame is corrupt, matching
+    // the streaming import contract and reporting partial writes accurately.
+    let flushed = batcher.flush();
+    result.and(flushed)
+}
+
+pub(super) async fn initialize_collection(
+    client: &mongodb::Client,
+    request: &MongoDatabaseRestoreRequest,
+    entry: &PreparedCollection,
+    exists: bool,
+) -> Result<(), String> {
+    let db = client.database(&request.database);
+    if request.drop_existing && exists {
+        db.collection::<Document>(&entry.metadata.collection_name).drop().await.map_err(|e| e.to_string())?;
+    }
+    if !exists || request.drop_existing {
+        db.run_command(entry.metadata.create_command(request.restore_options)).await.map_err(|e| e.to_string())?;
+    }
+    Ok(())
+}
+
+pub(super) async fn writable(state: &AppState, id: &str) -> Result<(), String> {
+    if let Some(name) = crate::query::connection_readonly_name(state, id).await {
+        return Err(format!("Read-only connection '{name}': restore blocked"));
+    }
+    Ok(())
+}
+
+#[allow(clippy::too_many_arguments)]
+pub(super) async fn restore_data(
+    state: &AppState,
+    client: &mongodb::Client,
+    request: &MongoDatabaseRestoreRequest,
+    source: Arc<CatalogSource>,
+    entries: &[PreparedCollection],
+    existing: &HashSet<String>,
+    progress: &mut MongoDatabaseDumpProgress,
+    started: Instant,
+    is_cancelled: &mut impl FnMut(&str) -> CancelFuture,
+    on_progress: &mut impl FnMut(MongoDatabaseDumpProgress),
+) -> Result<(), String> {
+    struct StopOnDrop(Arc<AtomicBool>);
+    impl Drop for StopOnDrop {
+        fn drop(&mut self) {
+            self.0.store(true, Ordering::Relaxed);
+        }
+    }
+    let flag = Arc::new(AtomicBool::new(false));
+    let _guard = StopOnDrop(flag.clone());
+    let (tx, mut rx) = mpsc::channel(2);
+    let batcher = Batcher {
+        tx,
+        index: 0,
+        documents: Vec::new(),
+        bytes: 0,
+        limit: request.batch_size.max(1),
+        objcheck: request.objcheck,
+    };
+    let producer_entries = entries.to_vec();
+    let signal = flag.clone();
+    let bytes_read = Arc::new(AtomicU64::new(0));
+    let reader_progress = bytes_read.clone();
+    let producer =
+        tokio::task::spawn_blocking(move || produce(source, producer_entries, signal, reader_progress, batcher));
+    progress.phase = "data".into();
+    progress.emit(started, on_progress);
+    let result = async {
+        let mut initialized = HashSet::new();
+        let mut tick = tokio::time::interval(Duration::from_millis(100));
+        loop {
+            let event = tokio::select! {
+                event = rx.recv() => match event { Some(event) => event, None => break },
+                _ = tick.tick() => {
+                    if is_cancelled(&request.task_id).await { return Err("MongoDB dump/restore cancelled".into()); }
+                    progress.bytes_processed = bytes_read.load(Ordering::Relaxed);
+                    progress.emit(started, on_progress);
+                    continue;
+                },
+            };
+            if is_cancelled(&request.task_id).await {
+                return Err("MongoDB dump/restore cancelled".into());
+            }
+            writable(state, &request.connection_id).await?;
+            let index = match &event {
+                Event::Batch { index, .. } | Event::End(index) => *index,
+            };
+            let entry = &entries[index];
+            progress.phase = "data".into();
+            progress.collection = Some(entry.metadata.collection_name.clone());
+            progress.emit(started, on_progress);
+            if initialized.insert(index) {
+                initialize_collection(client, request, entry, existing.contains(&entry.metadata.collection_name))
+                    .await?;
+            }
+            match event {
+                Event::Batch { documents, .. } => {
+                    progress.documents_read += documents.len() as u64;
+                    progress.bytes_processed = bytes_read.load(Ordering::Relaxed);
+                    if request.objcheck {
+                        progress.documents_validated += documents.len() as u64;
+                    }
+                    let outcome =
+                        insert_bson_documents(client, &request.database, &entry.metadata.collection_name, documents)
+                            .await
+                            .map_err(|e| e.message)?;
+                    progress.documents_written += outcome.inserted;
+                    progress.documents_failed += outcome.errors.len() as u64;
+                    progress.emit(started, on_progress);
+                    if let Some(error) = outcome.errors.first().filter(|_| request.stop_on_error) {
+                        return Err(error.message.clone());
+                    }
+                }
+                Event::End(_) => {
+                    progress.collections_done += 1;
+                    progress.emit(started, on_progress);
+                }
+            }
+        }
+        Ok(())
+    }
+    .await;
+    flag.store(true, Ordering::Relaxed);
+    drop(rx);
+    let produced = producer.await.map_err(|e| e.to_string()).and_then(|r| r);
+    result.and(produced)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn objcheck_uses_official_bson_parser_only_when_enabled() {
+        // Valid outer framing but invalid boolean payload.
+        let bytes = vec![9, 0, 0, 0, 8, b'x', 0, 2, 0];
+        assert!(checked_document(bytes.clone(), false).is_ok());
+        assert!(checked_document(bytes, true).is_err());
+        assert!(checked_document(vec![5, 0, 0, 0, 1], false).is_err());
+    }
+
+    #[test]
+    fn raw_bson_preserves_duplicate_keys_and_exact_bytes() {
+        let bytes = vec![19, 0, 0, 0, 16, b'x', 0, 1, 0, 0, 0, 16, b'x', 0, 2, 0, 0, 0, 0];
+        for objcheck in [false, true] {
+            let raw = checked_document(bytes.clone(), objcheck).unwrap();
+            assert_eq!(raw.as_bytes(), bytes);
+            assert_eq!(mongodb::bson::to_vec(&raw).unwrap(), bytes);
+        }
+    }
+
+    #[tokio::test]
+    async fn reader_delivers_earlier_batches_before_reporting_corruption() {
+        use crate::mongodb_dump::{
+            prepare_mongodb_restore_source, release_mongodb_restore_source, MongoRestoreSourceRequest,
+        };
+        let dir = tempfile::tempdir().unwrap();
+        let db = dir.path().join("source");
+        std::fs::create_dir(&db).unwrap();
+        let mut bytes = mongodb::bson::to_vec(&mongodb::bson::doc! { "_id": 1 }).unwrap();
+        bytes.extend_from_slice(b"invalid BSON");
+        std::fs::write(db.join("records.bson"), bytes).unwrap();
+        let preview = prepare_mongodb_restore_source(MongoRestoreSourceRequest {
+            path: db.to_string_lossy().into_owned(),
+            format: MongoDumpFormat::Directory,
+            gzip: false,
+        })
+        .await
+        .unwrap();
+        let source = super::super::source::get(&preview.source_ref).unwrap();
+        let entries = source.entries.clone();
+        let (tx, mut rx) = mpsc::channel(1);
+        let batcher = Batcher { tx, index: 0, documents: Vec::new(), bytes: 0, limit: 1, objcheck: false };
+        let worker = tokio::task::spawn_blocking(move || {
+            produce(source, entries, Arc::new(AtomicBool::new(false)), Arc::new(AtomicU64::new(0)), batcher)
+        });
+        assert!(matches!(rx.recv().await, Some(Event::Batch { documents, .. }) if documents.len() == 1));
+        assert!(worker.await.unwrap().unwrap_err().contains("BSON"));
+        assert!(rx.recv().await.is_none());
+        release_mongodb_restore_source(&preview.source_ref);
+    }
+
+    #[test]
+    fn batches_are_limited_by_bytes_as_well_as_document_count() {
+        let bytes = mongodb::bson::to_vec(&mongodb::bson::doc! { "payload": "x".repeat(5 * 1024 * 1024) }).unwrap();
+        let (tx, mut rx) = mpsc::channel(2);
+        let mut batcher = Batcher { tx, index: 0, documents: Vec::new(), bytes: 0, limit: 500, objcheck: false };
+        batcher.push(0, bytes.clone()).unwrap();
+        batcher.push(0, bytes).unwrap();
+        batcher.flush().unwrap();
+        for _ in 0..2 {
+            assert!(matches!(rx.try_recv().unwrap(), Event::Batch { documents, .. } if documents.len() == 1));
+        }
+    }
+
+    #[test]
+    #[ignore = "opt-in compressed BSON parser throughput smoke test"]
+    fn compressed_bson_parser_throughput() {
+        use flate2::{read::MultiGzDecoder, write::GzEncoder, Compression};
+        use std::io::{BufReader, Cursor, Write};
+        let mut document = Document::new();
+        document.insert("_id", 1i64);
+        for i in 0..64 {
+            document.insert(format!("field_{i}"), "value".repeat(8));
+        }
+        let encoded = mongodb::bson::to_vec(&document).unwrap();
+        let documents = 8192;
+        let mut encoder = GzEncoder::new(Vec::new(), Compression::fast());
+        for _ in 0..documents {
+            encoder.write_all(&encoded).unwrap();
+        }
+        let compressed = encoder.finish().unwrap();
+        for (mode, objcheck, passes) in
+            [("raw-stream", false, 1), ("objcheck-stream", true, 1), ("old-double-parse", true, 2)]
+        {
+            let started = Instant::now();
+            let mut count = 0;
+            for _ in 0..passes {
+                let mut input = BufReader::with_capacity(64 * 1024, MultiGzDecoder::new(Cursor::new(&compressed)));
+                loop {
+                    match archive::read_raw_frame(&mut input).unwrap() {
+                        archive::Frame::Document(bytes) => {
+                            checked_document(bytes, objcheck).unwrap();
+                            count += 1;
+                        }
+                        archive::Frame::End => break,
+                        _ => panic!("unexpected terminator"),
+                    }
+                }
+            }
+            assert_eq!(count, documents * passes);
+            eprintln!(
+                "{mode}: logical_mib={:.2}, elapsed_ms={}, logical_mib_s={:.2}",
+                (documents * encoded.len()) as f64 / 1048576.0,
+                started.elapsed().as_millis(),
+                (documents * encoded.len()) as f64 / 1048576.0 / started.elapsed().as_secs_f64()
+            );
+        }
+    }
+}

crates/dbx-core/src/mongodb_import_export.rs

@@ -7,6 +7,7 @@ use std::path::{Path, PathBuf};
 use std::time::Instant;
 
 use chrono::{DateTime as ChronoDateTime, NaiveDate, Utc};
+use flate2::{read::MultiGzDecoder, write::GzEncoder, Compression};
 use mongodb::bson::{oid::ObjectId, Bson, DateTime, Decimal128, Document};
 use serde::{Deserialize, Serialize};
 
@@ -26,6 +27,7 @@ pub const MAX_BATCH_SIZE: usize = 5000;
 /// Rows sampled for CSV type inference, independent of the preview window so that the
 /// preview and the import always agree on column types.
 pub const TYPE_SAMPLE_ROWS: usize = 1000;
+const BSON_MAX_DOCUMENT_BYTES: usize = 16 * 1024 * 1024;
 
 const TYPE_STRING: u8 = 1 << 0;
 const TYPE_BOOLEAN: u8 = 1 << 1;
@@ -45,6 +47,7 @@ pub enum MongoImportFormat {
     Csv,
     Json,
     Ndjson,
+    Bson,
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
@@ -341,14 +344,16 @@ struct CsvParseConfig {
 
 pub fn format_from_path(path: &str) -> Result<MongoImportFormat, String> {
     let lower = path.to_lowercase();
-    if lower.ends_with(".csv") || lower.ends_with(".tsv") || lower.ends_with(".txt") {
+    if lower.ends_with(".bson") || lower.ends_with(".bson.gz") {
+        Ok(MongoImportFormat::Bson)
+    } else if lower.ends_with(".csv") || lower.ends_with(".tsv") || lower.ends_with(".txt") {
         Ok(MongoImportFormat::Csv)
     } else if lower.ends_with(".ndjson") || lower.ends_with(".jsonl") {
         Ok(MongoImportFormat::Ndjson)
     } else if lower.ends_with(".json") {
         Ok(MongoImportFormat::Json)
     } else {
-        Err("Unsupported MongoDB import file type; use .csv, .json, or .ndjson".to_string())
+        Err("Unsupported MongoDB import file type; use .bson, .bson.gz, .csv, .json, or .ndjson".to_string())
     }
 }
 
@@ -1348,6 +1353,72 @@ fn file_size(path: &str) -> u64 {
     std::fs::metadata(path).map(|metadata| metadata.len()).unwrap_or(0)
 }
 
+fn open_bson_dump_reader(path: &str) -> Result<Box<dyn Read>, MongoImportIssue> {
+    let file = File::open(path).map_err(|error| MongoImportIssue::new("FILE_UNREADABLE", error.to_string()))?;
+    let reader = BufReader::new(file);
+    if path.to_lowercase().ends_with(".gz") {
+        Ok(Box::new(MultiGzDecoder::new(reader)))
+    } else {
+        Ok(Box::new(reader))
+    }
+}
+
+fn read_bson_dump_document<R: Read + ?Sized>(reader: &mut R, row: u64) -> Result<Option<Document>, MongoImportIssue> {
+    let mut length_bytes = [0u8; 4];
+    loop {
+        match reader.read(&mut length_bytes[..1]) {
+            Ok(0) => return Ok(None),
+            Ok(_) => break,
+            Err(error) if error.kind() == std::io::ErrorKind::Interrupted => continue,
+            Err(error) => return Err(MongoImportIssue::new("FILE_UNREADABLE", error.to_string()).with_row(row)),
+        }
+    }
+    reader.read_exact(&mut length_bytes[1..]).map_err(|error| {
+        MongoImportIssue::new("BSON_STRUCTURE", format!("Truncated BSON document length: {error}")).with_row(row)
+    })?;
+
+    let document_length = i32::from_le_bytes(length_bytes);
+    if document_length < 5 || document_length as usize > BSON_MAX_DOCUMENT_BYTES {
+        return Err(MongoImportIssue::new(
+            "BSON_STRUCTURE",
+            format!("Invalid BSON document length {document_length}; expected 5..={BSON_MAX_DOCUMENT_BYTES} bytes"),
+        )
+        .with_row(row));
+    }
+
+    let mut bytes = vec![0u8; document_length as usize];
+    bytes[..4].copy_from_slice(&length_bytes);
+    reader.read_exact(&mut bytes[4..]).map_err(|error| {
+        MongoImportIssue::new(
+            "BSON_STRUCTURE",
+            format!("Truncated BSON document; expected {document_length} bytes: {error}"),
+        )
+        .with_row(row)
+    })?;
+    let document = mongodb::bson::from_slice::<Document>(&bytes).map_err(|error| {
+        MongoImportIssue::new("BSON_STRUCTURE", format!("Invalid BSON document: {error}")).with_row(row)
+    })?;
+    Ok(Some(document))
+}
+
+fn parse_bson_preview(
+    path: &str,
+    preview_limit: usize,
+) -> Result<(Vec<ParsedMongoDocument>, u64, bool), MongoImportIssue> {
+    let mut reader = open_bson_dump_reader(path)?;
+    let mut documents = Vec::with_capacity(preview_limit.min(DEFAULT_PREVIEW_LIMIT));
+    let mut row = 1u64;
+    while documents.len() <= preview_limit {
+        let Some(document) = read_bson_dump_document(&mut *reader, row)? else {
+            return Ok((documents, row - 1, true));
+        };
+        documents.push(parsed_document(row, document, true));
+        row += 1;
+    }
+    documents.truncate(preview_limit);
+    Ok((documents, preview_limit as u64, false))
+}
+
 pub fn preview_mongodb_import_file(
     request: &MongoImportPreviewRequest,
 ) -> Result<MongoImportPreview, MongoImportIssue> {
@@ -1406,6 +1477,25 @@ pub fn preview_mongodb_import_file(
                 estimated_rows_exact,
             })
         }
+        MongoImportFormat::Bson => {
+            let (documents, estimated_rows, estimated_rows_exact) =
+                parse_bson_preview(&request.file_path, preview_limit)?;
+            Ok(MongoImportPreview {
+                source_ref: request.source_ref.clone(),
+                format: request.format,
+                detected_encoding: None,
+                file_name: file_name(&request.file_path),
+                file_path: request.file_path.clone(),
+                size_bytes: file_size(&request.file_path),
+                columns: columns_from_documents(&documents),
+                row_numbers: documents.iter().map(|document| document.row).collect(),
+                rows: documents.into_iter().map(|document| document.extended_json).collect(),
+                warnings: Vec::new(),
+                errors: Vec::new(),
+                estimated_rows: Some(estimated_rows),
+                estimated_rows_exact,
+            })
+        }
     }
 }
 
@@ -1419,6 +1509,7 @@ pub fn preview_mongodb_import_bytes(
         MongoImportFormat::Csv => "csv",
         MongoImportFormat::Json => "json",
         MongoImportFormat::Ndjson => "ndjson",
+        MongoImportFormat::Bson => "bson",
     };
     let dir = std::env::temp_dir().join(format!("dbx-mongo-import-preview-{}", uuid::Uuid::new_v4()));
     std::fs::create_dir_all(&dir).map_err(|error| MongoImportIssue::new("FILE_UNREADABLE", error.to_string()))?;
@@ -1517,6 +1608,20 @@ where
     Ok(())
 }
 
+fn stream_bson_file<F>(path: &str, mut on_document: F) -> Result<(), MongoImportIssue>
+where
+    F: FnMut(Result<ParsedMongoDocument, MongoImportIssue>) -> Result<(), MongoImportIssue>,
+{
+    let mut reader = open_bson_dump_reader(path)?;
+    let mut row = 1u64;
+    loop {
+        // A corrupt frame has no reliable next-document boundary, even when skipping row errors.
+        let Some(document) = read_bson_dump_document(&mut *reader, row)? else { return Ok(()) };
+        on_document(Ok(parsed_document(row, document, false)))?;
+        row += 1;
+    }
+}
+
 pub fn for_each_mongodb_import_document<F>(
     path: &str,
     format: MongoImportFormat,
@@ -1531,6 +1636,7 @@ where
         MongoImportFormat::Csv => stream_csv_file(path, options, on_document),
         MongoImportFormat::Json => stream_json_file(path, options, false, on_document),
         MongoImportFormat::Ndjson => stream_json_file(path, options, true, on_document),
+        MongoImportFormat::Bson => stream_bson_file(path, on_document),
     }
 }
 
@@ -1929,6 +2035,7 @@ pub fn mongodb_export_client_session_id(export_id: &str) -> String {
 pub enum MongoExportFormat {
     Csv,
     Ndjson,
+    Bson,
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
@@ -1958,6 +2065,8 @@ pub struct MongoExportRequest {
     pub format: MongoExportFormat,
     #[serde(default = "default_true")]
     pub include_header: bool,
+    #[serde(default)]
+    pub gzip: bool,
     pub file_path: String,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub execution_id: Option<String>,
@@ -1992,7 +2101,7 @@ pub struct MongoExportSummary {
 
 fn temp_export_path(target: &Path) -> PathBuf {
     let name = target.file_name().and_then(|name| name.to_str()).unwrap_or("export");
-    target.with_file_name(format!(".{name}.dbx-export.tmp"))
+    target.with_file_name(format!(".{name}.{}.dbx-export.tmp", uuid::Uuid::new_v4()))
 }
 
 fn cleanup_temp(path: &Path) {
@@ -2214,6 +2323,9 @@ where
 {
     let started_at = Instant::now();
     on_progress(export_progress(&request.export_id, MongoExportStatus::Running, 0, 0, None, None, started_at));
+    if request.gzip && request.format != MongoExportFormat::Bson {
+        return Err("Gzip compression is only supported for BSON dump exports".to_string());
+    }
     if is_cancelled(&request.export_id).await {
         on_progress(export_progress(
             &request.export_id,
@@ -2250,6 +2362,9 @@ where
         MongoExportFormat::Csv => {
             export_csv(state, request, &temp, total_documents, started_at, &mut is_cancelled, &mut on_progress).await
         }
+        MongoExportFormat::Bson => {
+            export_bson(state, request, &temp, total_documents, started_at, &mut is_cancelled, &mut on_progress).await
+        }
     };
 
     match result {
@@ -2685,6 +2800,120 @@ where
     Ok((documents_read, bytes_written))
 }
 
+enum BsonExportWriter {
+    Plain(BufWriter<File>),
+    Gzip(Box<GzEncoder<BufWriter<File>>>),
+}
+
+impl BsonExportWriter {
+    fn finish(self) -> Result<(), String> {
+        match self {
+            Self::Plain(mut writer) => writer.flush().map_err(|error| error.to_string()),
+            Self::Gzip(writer) => {
+                let mut writer = writer.finish().map_err(|error| error.to_string())?;
+                writer.flush().map_err(|error| error.to_string())
+            }
+        }
+    }
+}
+
+impl Write for BsonExportWriter {
+    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
+        match self {
+            Self::Plain(writer) => writer.write(buf),
+            Self::Gzip(writer) => writer.write(buf),
+        }
+    }
+
+    fn flush(&mut self) -> std::io::Result<()> {
+        match self {
+            Self::Plain(writer) => writer.flush(),
+            Self::Gzip(writer) => writer.flush(),
+        }
+    }
+}
+
+async fn export_bson<C, F>(
+    state: &AppState,
+    request: &MongoExportRequest,
+    temp: &Path,
+    total_documents: Option<u64>,
+    started_at: Instant,
+    is_cancelled: &mut C,
+    on_progress: &mut F,
+) -> Result<(u64, u64), String>
+where
+    C: FnMut(&str) -> std::pin::Pin<Box<dyn std::future::Future<Output = bool> + Send>>,
+    F: FnMut(MongoExportProgress),
+{
+    if is_cancelled(&request.export_id).await {
+        return Err("Export cancelled".to_string());
+    }
+    // BSON dumps must carry native BSON types, so they are written straight from the native
+    // driver's documents; the legacy agent's canonical Extended JSON round trip does not apply.
+    let client = match state.pool_handle(&request.connection_id).await {
+        Some(PoolKind::MongoDb(client)) => client,
+        _ => return Err("BSON dump export requires the native MongoDB driver".into()),
+    };
+    let file = File::create(temp).map_err(|error| error.to_string())?;
+    let buffered = BufWriter::new(file);
+    let mut writer = if request.gzip {
+        BsonExportWriter::Gzip(Box::new(GzEncoder::new(buffered, Compression::default())))
+    } else {
+        BsonExportWriter::Plain(buffered)
+    };
+    let mut documents_read = 0u64;
+    let mut bytes_written = 0u64;
+
+    let export = for_each_find_document(
+        &client,
+        &request.database,
+        &request.collection,
+        request.filter.as_deref(),
+        request.projection.as_deref(),
+        request.sort.as_deref(),
+        request.collation.as_deref(),
+        DEFAULT_EXPORT_BATCH_SIZE,
+        |document| {
+            let bytes = mongodb::bson::to_vec(&document).map_err(|error| error.to_string())?;
+            writer.write_all(&bytes).map_err(|error| error.to_string())?;
+            bytes_written += bytes.len() as u64;
+            documents_read += 1;
+            if documents_read == 1 || documents_read.is_multiple_of(500) {
+                on_progress(export_progress(
+                    &request.export_id,
+                    MongoExportStatus::Running,
+                    documents_read,
+                    bytes_written,
+                    total_documents,
+                    None,
+                    started_at,
+                ));
+            }
+            Ok(())
+        },
+    );
+    {
+        tokio::pin!(export);
+        let mut poll_cancel = tokio::time::interval(std::time::Duration::from_millis(100));
+        loop {
+            tokio::select! {
+                result = &mut export => { result?; break; }
+                _ = poll_cancel.tick() => {
+                    if is_cancelled(&request.export_id).await {
+                        return Err("Export cancelled".to_string());
+                    }
+                }
+            }
+        }
+    }
+    writer.finish()?;
+    if is_cancelled(&request.export_id).await {
+        return Err("Export cancelled".to_string());
+    }
+    Ok((documents_read, bytes_written))
+}
+
 const CSV_FIELD_DISCOVERY_DOCS: usize = 10_000;
 
 async fn export_csv<C, F>(
@@ -2847,6 +3076,112 @@ mod tests {
         Ok(output)
     }
 
+    fn bson_dump_bytes(documents: &[Document]) -> Vec<u8> {
+        let mut output = Vec::new();
+        for document in documents {
+            output.extend_from_slice(&mongodb::bson::to_vec(document).unwrap());
+        }
+        output
+    }
+
+    #[test]
+    fn bson_export_writer_finishes_plain_and_gzip_dumps() {
+        let root = tempfile::tempdir().unwrap();
+        let documents = vec![doc! { "_id": 1, "name": "first" }, doc! { "_id": 2, "value": 42i64 }];
+        for expected in [Vec::new(), bson_dump_bytes(&documents)] {
+            for gzip in [false, true] {
+                let path = root.path().join(if gzip { "records.bson.gz" } else { "records.bson" });
+                let buffered = BufWriter::new(File::create(&path).unwrap());
+                let mut writer = if gzip {
+                    BsonExportWriter::Gzip(Box::new(GzEncoder::new(buffered, Compression::default())))
+                } else {
+                    BsonExportWriter::Plain(buffered)
+                };
+                writer.write_all(&expected).unwrap();
+                writer.finish().unwrap();
+
+                let bytes = std::fs::read(&path).unwrap();
+                let actual = if gzip {
+                    let mut decoded = Vec::new();
+                    MultiGzDecoder::new(bytes.as_slice()).read_to_end(&mut decoded).unwrap();
+                    decoded
+                } else {
+                    bytes
+                };
+                assert_eq!(actual, expected);
+            }
+        }
+    }
+
+    #[test]
+    fn bson_dump_preview_and_import_preserve_native_types() {
+        use mongodb::bson::spec::BinarySubtype;
+        use mongodb::bson::{Binary, DateTime};
+
+        let documents = vec![
+            doc! {
+                "_id": ObjectId::parse_str("507f1f77bcf86cd799439011").unwrap(),
+                "createdAt": DateTime::from_millis(1_609_459_200_000),
+                "payload": Bson::Binary(Binary { subtype: BinarySubtype::Generic, bytes: vec![1, 2, 3] }),
+                "nested": { "count": 42i64 },
+            },
+            doc! { "_id": 2i32, "name": "second" },
+        ];
+        let bytes = bson_dump_bytes(&documents);
+        let preview =
+            preview_mongodb_import_bytes(&bytes, MongoImportFormat::Bson, &MongoImportParseOptions::default(), 10)
+                .unwrap();
+        assert_eq!(preview.rows.len(), 2);
+        assert_eq!(preview.estimated_rows, Some(2));
+        assert!(preview.estimated_rows_exact);
+        assert_eq!(preview.rows[0]["_id"]["$oid"], "507f1f77bcf86cd799439011");
+        assert!(preview.rows[0]["payload"].get("$binary").is_some());
+
+        let imported = execute_source(&bytes, "bson", MongoImportFormat::Bson, &MongoImportParseOptions::default());
+        assert_eq!(imported, preview.rows);
+    }
+
+    #[test]
+    fn gzip_bson_dump_import_matches_plain_dump() {
+        let documents = vec![doc! { "_id": 1i32, "name": "Ada" }, doc! { "_id": 2i32, "name": "Bob" }];
+        let plain = bson_dump_bytes(&documents);
+        let mut encoder = GzEncoder::new(Vec::new(), Compression::default());
+        encoder.write_all(&plain).unwrap();
+        let compressed = encoder.finish().unwrap();
+
+        let imported =
+            execute_source(&compressed, "bson.gz", MongoImportFormat::Bson, &MongoImportParseOptions::default());
+        assert_eq!(imported.len(), 2);
+        assert_eq!(imported[0]["name"], "Ada");
+        assert_eq!(imported[1]["name"], "Bob");
+    }
+
+    #[test]
+    fn bson_dump_rejects_truncated_and_invalid_lengths() {
+        let mut truncated = mongodb::bson::to_vec(&doc! { "name": "Ada" }).unwrap();
+        truncated.pop();
+        let error =
+            preview_mongodb_import_bytes(&truncated, MongoImportFormat::Bson, &MongoImportParseOptions::default(), 10)
+                .unwrap_err();
+        assert_eq!(error.code, "BSON_STRUCTURE");
+        assert_eq!(error.row, Some(1));
+
+        let error = preview_mongodb_import_bytes(
+            &(BSON_MAX_DOCUMENT_BYTES as i32 + 1).to_le_bytes(),
+            MongoImportFormat::Bson,
+            &MongoImportParseOptions::default(),
+            10,
+        )
+        .unwrap_err();
+        assert_eq!(error.code, "BSON_STRUCTURE");
+    }
+
+    #[test]
+    fn bson_dump_format_is_detected_from_plain_and_gzip_paths() {
+        assert_eq!(format_from_path("users.bson").unwrap(), MongoImportFormat::Bson);
+        assert_eq!(format_from_path("users.BSON.GZ").unwrap(), MongoImportFormat::Bson);
+    }
+
     #[test]
     fn csv_quoted_comma_newline_and_escaped_quotes() {
         let csv = "name,note\n\"Ada, Lovelace\",\"line1\nline2\"\n\"quotes\",\"she said \"\"hi\"\"\"\n";
@@ -3806,6 +4141,7 @@ def handle(request):
                 collation: None,
                 format: MongoExportFormat::Ndjson,
                 include_header: true,
+                gzip: false,
                 file_path: target.to_string_lossy().to_string(),
                 execution_id: None,
             },
@@ -3857,6 +4193,7 @@ def handle(request):
                 collation: None,
                 format: MongoExportFormat::Ndjson,
                 include_header: true,
+                gzip: false,
                 file_path: target.to_string_lossy().to_string(),
                 execution_id: None,
             },
@@ -3928,6 +4265,7 @@ def handle(request):
                 collation: None,
                 format: MongoExportFormat::Ndjson,
                 include_header: true,
+                gzip: false,
                 file_path: target.to_string_lossy().to_string(),
                 execution_id: None,
             },
@@ -3979,6 +4317,7 @@ def handle(request):
                 collation: None,
                 format: MongoExportFormat::Ndjson,
                 include_header: true,
+                gzip: false,
                 file_path: target.to_string_lossy().to_string(),
                 execution_id: None,
             },
@@ -4035,6 +4374,7 @@ def handle(request):
                 collation: None,
                 format: MongoExportFormat::Ndjson,
                 include_header: true,
+                gzip: false,
                 file_path: target.to_string_lossy().to_string(),
                 execution_id: None,
             },
@@ -4085,6 +4425,7 @@ def handle(request):
                 collation: None,
                 format: MongoExportFormat::Ndjson,
                 include_header: true,
+                gzip: false,
                 file_path: target.to_string_lossy().to_string(),
                 execution_id: None,
             },