feat: batch methods

daemon/src/files/batch.rs

@@ -0,0 +1,179 @@
+// SPDX-License-Identifier: AGPL-3.0-only
+use crate::config::StorageEntry;
+use crate::files::copy::copy_file_locked;
+use crate::files::delete::delete_file_locked;
+use crate::files::list::list_files_locked;
+use crate::files::mkdir::mkdir_file_locked;
+use crate::files::rename::rename_file_locked;
+use crate::files::stat::stat_file_locked;
+use crate::files::write::write_file_by_url_locked;
+use crate::storage::StorageIdentifier;
+use crate::storage::sync::{StorageLockData, StorageTokens, storage_read_timeout};
+use crate::{Daemon, DaemonExtra, StorageNotification};
+use daemon_common::service::error_storage_failure::FromStorageFailure;
+use daemon_common::service::error_storage_mutex_locked::FromStorageMutexLocked;
+use daemon_common::service::error_storage_not_found::FromStorageNotFound;
+use daemon_common::service::files::task::{FileOperationKind, StorageTask};
+use daemon_common::service::files::{
+    BatchOperationError, FileOperation, FileOperationBatch, FileOperationFailure,
+    FileOperationResult, UrlWriteBatch, UrlWriteResult,
+};
+use futures::StreamExt as _;
+use std::io;
+use tokio::sync::broadcast::Sender;
+
+const DEFAULT_PARALLEL: u8 = 8;
+const MAX_PARALLEL: u8 = 16;
+
+/// Cancel safe
+pub async fn write_files_by_url<
+    E: From<io::Error>
+        + From<sqlx::Error>
+        + FromStorageNotFound
+        + FromStorageMutexLocked
+        + FromStorageFailure,
+>(
+    daemon: &Daemon<impl DaemonExtra>,
+    id: StorageIdentifier,
+    batch: UrlWriteBatch,
+) -> Result<Vec<UrlWriteResult>, E> {
+    let StorageLockData {
+        lock,
+        storage,
+        notify,
+        tokens,
+        ..
+    } = storage_read_timeout::<E>(
+        &daemon.config,
+        &*daemon.conn,
+        &daemon.storages,
+        id,
+        StorageTask::FileOperation {
+            kind: FileOperationKind::WriteByUrlBatch,
+        },
+    )
+    .await?;
+
+    let parallel = usize::from(
+        batch
+            .parallel
+            .unwrap_or(DEFAULT_PARALLEL)
+            .clamp(1, MAX_PARALLEL),
+    );
+    let (entry, token_map, sender) = (&storage, &tokens, &notify);
+    let results = futures::stream::iter(batch.files)
+        .map(|file| async move {
+            let error = write_file_by_url_locked::<BatchOperationError>(
+                daemon,
+                id,
+                entry,
+                token_map,
+                sender,
+                file.path.clone(),
+                file.url,
+            )
+            .await
+            .err()
+            .map(FileOperationFailure::from);
+            UrlWriteResult {
+                path: file.path,
+                error,
+            }
+        })
+        .buffered(parallel)
+        .collect::<Vec<_>>()
+        .await;
+
+    drop(lock);
+    Ok(results)
+}
+
+async fn run_operation(
+    id: StorageIdentifier,
+    storage: &StorageEntry,
+    tokens: &StorageTokens,
+    notify: &Sender<StorageNotification>,
+    operation: FileOperation,
+) -> Result<FileOperationResult, BatchOperationError> {
+    let mut result = FileOperationResult::default();
+    match operation {
+        FileOperation::List { path } => {
+            result.entries =
+                Some(list_files_locked::<BatchOperationError>(id, storage, tokens, path).await?);
+        }
+        FileOperation::Stat { path } => {
+            result.entry =
+                Some(stat_file_locked::<BatchOperationError>(id, storage, tokens, path).await?);
+        }
+        FileOperation::Delete { path, recursive } => {
+            delete_file_locked::<BatchOperationError>(id, storage, tokens, notify, path, recursive)
+                .await?;
+        }
+        FileOperation::Rename { from, to } => {
+            rename_file_locked::<BatchOperationError>(id, storage, tokens, notify, from, to)
+                .await?;
+        }
+        FileOperation::Copy { from, to } => {
+            copy_file_locked::<BatchOperationError>(id, storage, tokens, notify, from, to).await?;
+        }
+        FileOperation::Mkdir { path, recursive } => {
+            mkdir_file_locked::<BatchOperationError>(id, storage, tokens, notify, path, recursive)
+                .await?;
+        }
+    }
+    Ok(result)
+}
+
+/// Cancel safe
+pub async fn file_operations<
+    E: From<io::Error>
+        + From<sqlx::Error>
+        + FromStorageNotFound
+        + FromStorageMutexLocked
+        + FromStorageFailure,
+>(
+    daemon: &Daemon<impl DaemonExtra>,
+    id: StorageIdentifier,
+    batch: FileOperationBatch,
+) -> Result<Vec<FileOperationResult>, E> {
+    let StorageLockData {
+        lock,
+        storage,
+        notify,
+        tokens,
+        ..
+    } = storage_read_timeout::<E>(
+        &daemon.config,
+        &*daemon.conn,
+        &daemon.storages,
+        id,
+        StorageTask::FileOperation {
+            kind: FileOperationKind::Batch,
+        },
+    )
+    .await?;
+
+    let mut results = Vec::with_capacity(batch.operations.len());
+    let mut failed = false;
+    for operation in batch.operations {
+        if failed && batch.stop_on_error {
+            results.push(FileOperationResult {
+                error: Some(FileOperationFailure::Skipped),
+                ..FileOperationResult::default()
+            });
+            continue;
+        }
+
+        let result = run_operation(id, &storage, &tokens, &notify, operation)
+            .await
+            .unwrap_or_else(|err| FileOperationResult {
+                error: Some(err.into()),
+                ..FileOperationResult::default()
+            });
+        failed |= result.error.is_some();
+        results.push(result);
+    }
+
+    drop(lock);
+    Ok(results)
+}

daemon/src/files/copy.rs

@@ -1,7 +1,8 @@
 // SPDX-License-Identifier: AGPL-3.0-only
+use crate::config::StorageEntry;
 use crate::files::storage_path;
 use crate::storage::StorageIdentifier;
-use crate::storage::sync::{StorageLockData, storage_read_timeout};
+use crate::storage::sync::{StorageLockData, StorageTokens, storage_read_timeout};
 use crate::utils::storage_parent_ino;
 use crate::{Daemon, DaemonExtra, StorageNotification};
 use cap_async_std::fs::MetadataExt as _;
@@ -11,6 +12,7 @@ use daemon_common::service::error_storage_not_found::FromStorageNotFound;
 use daemon_common::service::files::task::{FileOperationKind, StorageTask};
 use std::io;
 use std::io::ErrorKind;
+use tokio::sync::broadcast::Sender;
 use tracing::{debug, error, warn};
 
 /// Cancel safe
@@ -43,8 +45,22 @@ pub async fn copy_file<
         },
     )
     .await?;
-    let (storage_from, _, path_from) = storage_path(id, &storage, from, &tokens).await?;
-    let (storage_to, _, path_to) = storage_path(id, &storage, to, &tokens).await?;
+    let result = copy_file_locked::<E>(id, &storage, &tokens, &notify, from, to).await;
+    drop(lock);
+    result
+}
+
+/// Cancel safe, the caller holds the storage lock
+pub(super) async fn copy_file_locked<E: From<io::Error>>(
+    id: StorageIdentifier,
+    storage: &StorageEntry,
+    tokens: &StorageTokens,
+    notify: &Sender<StorageNotification>,
+    from: String,
+    to: String,
+) -> Result<(), E> {
+    let (storage_from, _, path_from) = storage_path(id, storage, from, tokens).await?;
+    let (storage_to, _, path_to) = storage_path(id, storage, to, tokens).await?;
 
     let to_parent_ino = storage_parent_ino(&storage_to, &path_to).await;
 
@@ -77,6 +93,5 @@ pub async fn copy_file<
             });
     }
 
-    drop(lock);
     Ok(())
 }

daemon/src/files/delete.rs

@@ -1,7 +1,8 @@
 // SPDX-License-Identifier: AGPL-3.0-only
+use crate::config::StorageEntry;
 use crate::files::storage_path;
 use crate::storage::StorageIdentifier;
-use crate::storage::sync::{StorageLockData, storage_read_timeout};
+use crate::storage::sync::{StorageLockData, StorageTokens, storage_read_timeout};
 use crate::utils::storage_parent_ino;
 use crate::{Daemon, DaemonExtra, StorageNotification};
 use cap_async_std::fs::MetadataExt as _;
@@ -11,6 +12,7 @@ use daemon_common::service::error_storage_not_found::FromStorageNotFound;
 use daemon_common::service::files::task::{FileOperationKind, StorageTask};
 use std::io;
 use std::io::ErrorKind;
+use tokio::sync::broadcast::Sender;
 use tracing::{debug, error, warn};
 
 /// Cancel safe
@@ -43,7 +45,21 @@ pub async fn delete_file<
         },
     )
     .await?;
-    let (storage, _, path) = storage_path(id, &storage, path, &tokens).await?;
+    let result = delete_file_locked::<E>(id, &storage, &tokens, &notify, path, recursive).await;
+    drop(lock);
+    result
+}
+
+/// Cancel safe, the caller holds the storage lock
+pub(super) async fn delete_file_locked<E: From<io::Error>>(
+    id: StorageIdentifier,
+    storage: &StorageEntry,
+    tokens: &StorageTokens,
+    notify: &Sender<StorageNotification>,
+    path: String,
+    recursive: bool,
+) -> Result<(), E> {
+    let (storage, _, path) = storage_path(id, storage, path, tokens).await?;
 
     let ino = storage.metadata(&path).await.map(|meta| meta.ino());
     let parent_ino = storage_parent_ino(&storage, &path).await;
@@ -80,6 +96,5 @@ pub async fn delete_file<
             });
     }
 
-    drop(lock);
     Ok(())
 }

daemon/src/files/list.rs

@@ -1,8 +1,9 @@
 // SPDX-License-Identifier: AGPL-3.0-only
+use crate::config::StorageEntry;
 use crate::files::{IntoFileEntry as _, storage_path};
 use crate::storage::StorageIdentifier;
 use crate::storage::archive::format::ArchiveFormat;
-use crate::storage::sync::{StorageLockData, storage_read_timeout};
+use crate::storage::sync::{StorageLockData, StorageTokens, storage_read_timeout};
 use crate::{Daemon, DaemonExtra};
 use daemon_common::service::error_storage_failure::FromStorageFailure;
 use daemon_common::service::error_storage_mutex_locked::FromStorageMutexLocked;
@@ -43,7 +44,19 @@ pub async fn list_files<
         },
     )
     .await?;
-    let (storage, _, path) = storage_path(id, &storage, path, &tokens).await?;
+    let result = list_files_locked::<E>(id, &storage, &tokens, path).await;
+    drop(lock);
+    result
+}
+
+/// Cancel safe, the caller holds the storage lock
+pub(super) async fn list_files_locked<E: From<io::Error>>(
+    id: StorageIdentifier,
+    storage: &StorageEntry,
+    tokens: &StorageTokens,
+    path: String,
+) -> Result<Vec<FileEntry>, E> {
+    let (storage, _, path) = storage_path(id, storage, path, tokens).await?;
     let read_dir = storage
         .read_dir(&path)
         .await
@@ -97,6 +110,5 @@ pub async fn list_files<
             archive,
         ));
     }
-    drop(lock);
     Ok(files)
 }

daemon/src/files/mkdir.rs

@@ -1,7 +1,8 @@
 // SPDX-License-Identifier: AGPL-3.0-only
+use crate::config::StorageEntry;
 use crate::files::storage_path;
 use crate::storage::StorageIdentifier;
-use crate::storage::sync::{StorageLockData, storage_read_timeout};
+use crate::storage::sync::{StorageLockData, StorageTokens, storage_read_timeout};
 use crate::utils::storage_parent_ino;
 use crate::{Daemon, DaemonExtra, StorageNotification};
 use cap_async_std::fs::MetadataExt as _;
@@ -11,6 +12,7 @@ use daemon_common::service::error_storage_not_found::FromStorageNotFound;
 use daemon_common::service::files::task::{FileOperationKind, StorageTask};
 use std::io;
 use std::io::ErrorKind;
+use tokio::sync::broadcast::Sender;
 use tracing::{debug, error, warn};
 
 /// Cancel safe
@@ -43,7 +45,21 @@ pub async fn mkdir_file<
         },
     )
     .await?;
-    let (storage, _, path) = storage_path(id, &storage, path, &tokens).await?;
+    let result = mkdir_file_locked::<E>(id, &storage, &tokens, &notify, path, recursive).await;
+    drop(lock);
+    result
+}
+
+/// Cancel safe, the caller holds the storage lock
+pub(super) async fn mkdir_file_locked<E: From<io::Error>>(
+    id: StorageIdentifier,
+    storage: &StorageEntry,
+    tokens: &StorageTokens,
+    notify: &Sender<StorageNotification>,
+    path: String,
+    recursive: bool,
+) -> Result<(), E> {
+    let (storage, _, path) = storage_path(id, storage, path, tokens).await?;
 
     let parent_ino = storage_parent_ino(&storage, &path).await;
 
@@ -85,6 +101,5 @@ pub async fn mkdir_file<
             });
     }
 
-    drop(lock);
     Ok(())
 }

daemon/src/files/mod.rs

@@ -1,5 +1,6 @@
 // SPDX-License-Identifier: AGPL-3.0-only
 mod archive;
+mod batch;
 mod chmod;
 mod chown;
 mod copy;
@@ -19,6 +20,7 @@ use crate::storage::sync::StorageTokens;
 use crate::storage::{STORAGE_TOKEN_MAGIC, StorageIdentifier};
 use crate::utils::path_relative;
 pub use archive::{create_archive, unpack_archive};
+pub use batch::{file_operations, write_files_by_url};
 use cap_async_std::ambient_authority;
 use cap_async_std::fs::{Dir, FileTypeExt as _, Metadata, MetadataExt as _};
 pub use chmod::chmod_file;