use std::path::PathBuf;
use crate::core::rclone::command::{Copy, Purge};
use crate::core::rclone::{RcloneCommand as _, check_available};
pub(crate) struct RcloneCopyPlan {
pub source: String,
pub destination: String,
pub log_path: PathBuf,
pub transfers: usize,
pub checkers: usize,
pub tpslimit: Option<usize>,
}
#[derive(Debug, Default, Clone)]
pub(crate) struct RcloneCopySummary {
pub transferred: u64,
pub errors: u64,
pub bytes: u64,
pub log_path: PathBuf,
}
pub(crate) struct RcloneDeletePlan {
pub source: String,
pub log_path: PathBuf,
pub checkers: usize,
}
#[derive(Debug, Default, Clone)]
pub(crate) struct RcloneDeleteSummary {
pub deleted: u64,
pub errors: u64,
pub log_path: PathBuf,
}
pub(crate) async fn check_rclone_available() -> Result<(), String> {
check_available()
.await
.map_err(|_| "'rclone' was not found on PATH -- required for 'pigeon job run rclone' to copy/delete data".to_string())
}
pub(crate) async fn run_copy_job(
source: &str,
destination: &str,
log_path: &std::path::Path,
transfers: usize,
checkers: usize,
tpslimit: Option<usize>,
) -> Result<RcloneCopySummary, String> {
crate::observability::metrics::set_macro_phase("rclone-copy", true);
let log_summary = Copy {
source: source.to_string(),
destination: destination.to_string(),
log_path: log_path.to_path_buf(),
transfers,
checkers,
tpslimit,
job: "rclone-copy",
step: "transfer",
upload_metrics: true,
..Default::default()
}
.run()
.await?;
Ok(RcloneCopySummary {
transferred: log_summary.transferred,
errors: log_summary.errors,
bytes: log_summary.bytes,
log_path: log_path.to_path_buf(),
})
}
pub(crate) async fn run_delete_job(
source: &str,
log_path: &std::path::Path,
checkers: usize,
) -> Result<RcloneDeleteSummary, String> {
crate::observability::metrics::set_macro_phase("rclone-delete", true);
let log_summary = Purge {
source: source.to_string(),
log_path: log_path.to_path_buf(),
checkers,
job: "rclone-delete",
step: "delete",
}
.run()
.await?;
Ok(RcloneDeleteSummary {
deleted: log_summary.deletes,
errors: log_summary.errors,
log_path: log_path.to_path_buf(),
})
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn copies_a_file_between_two_local_directories() {
if check_rclone_available().await.is_err() {
eprintln!("skipping: rclone not found on PATH");
return;
}
let source_dir = tempfile::tempdir().unwrap();
let dest_dir = tempfile::tempdir().unwrap();
let log_dir = tempfile::tempdir().unwrap();
std::fs::write(source_dir.path().join("hello.txt"), b"hello world").unwrap();
let log_path = log_dir.path().join("rclone.jsonl");
let summary = run_copy_job(
source_dir.path().to_str().unwrap(),
dest_dir.path().to_str().unwrap(),
&log_path,
2,
4,
None,
)
.await
.unwrap();
assert_eq!(summary.transferred, 1);
assert_eq!(summary.errors, 0);
assert!(dest_dir.path().join("hello.txt").exists());
}
#[tokio::test]
async fn copies_a_file_with_a_tpslimit_set() {
if check_rclone_available().await.is_err() {
eprintln!("skipping: rclone not found on PATH");
return;
}
let source_dir = tempfile::tempdir().unwrap();
let dest_dir = tempfile::tempdir().unwrap();
let log_dir = tempfile::tempdir().unwrap();
std::fs::write(source_dir.path().join("hello.txt"), b"hello world").unwrap();
let log_path = log_dir.path().join("rclone.jsonl");
let summary = run_copy_job(
source_dir.path().to_str().unwrap(),
dest_dir.path().to_str().unwrap(),
&log_path,
8,
16,
Some(10),
)
.await
.unwrap();
assert_eq!(summary.transferred, 1);
assert_eq!(summary.errors, 0);
assert!(dest_dir.path().join("hello.txt").exists());
}
#[tokio::test]
async fn purges_everything_under_a_local_directory() {
if check_rclone_available().await.is_err() {
eprintln!("skipping: rclone not found on PATH");
return;
}
let source_dir = tempfile::tempdir().unwrap();
let log_dir = tempfile::tempdir().unwrap();
std::fs::write(source_dir.path().join("a.txt"), b"a").unwrap();
std::fs::create_dir(source_dir.path().join("sub")).unwrap();
std::fs::write(source_dir.path().join("sub/b.txt"), b"b").unwrap();
let log_path = log_dir.path().join("rclone-purge.jsonl");
let summary = run_delete_job(source_dir.path().to_str().unwrap(), &log_path, 4)
.await
.unwrap();
assert_eq!(summary.errors, 0);
assert!(summary.deleted >= 1);
assert!(!source_dir.path().join("a.txt").exists());
assert!(!source_dir.path().join("sub").exists());
}
}