feat: rclone-privider

.dockerignore

@@ -1,11 +1,9 @@
-# Git
 .git
 .gitignore
-
-# MD files
 CHANGELOG.md
 README.md
 RELEASE.md
-
-#IDE configurations
 .idea
+target
+dump.rdb
+.superpowers

.gitignore

@@ -8,3 +8,4 @@
 .claude
 
 /docs
+.superpowers

Cargo.toml

@@ -19,7 +19,7 @@ log = "0.4.29"
 toml = "0.9.10"
 reqwest = { version = "0.13.1", features = ["json", "blocking", "multipart", "stream", "query"] }
 anyhow = "1.0.100"
-tokio = { version = "1.49.0", features = ["rt", "rt-multi-thread", "macros", "fs"] }
+tokio = { version = "1.49.0", features = ["rt", "rt-multi-thread", "macros", "fs", "process", "io-util"] }
 async-trait = "0.1.89"
 tempfile = "3.24.0"
 openssl = "0.10.75"

docker-compose.yml

@@ -21,7 +21,7 @@ services:
       LOG: debug
       TZ: "Europe/Paris"
 #      TMPDIR: /scratch
-      EDGE_KEY: "eyJzZXJ2ZXJVcmwiOiJodHRwOi8vbG9jYWxob3N0Ojg4ODciLCJhZ2VudElkIjoiOWMxMzM5NjItMGE5OC00MmRkLTk1NjUtOTA5ZTkyYTI5N2VkIiwibWFzdGVyS2V5QjY0IjoiMUh0djdtWCtYVkJxL0IzUEV2WDlZZjlQeUdVZW5oRHlXemo5THRqNW90WT0ifQ=="
+      EDGE_KEY: "eyJzZXJ2ZXJVcmwiOiJodHRwOi8vbG9jYWxob3N0Ojg4ODciLCJhZ2VudElkIjoiMWNhYTY2ZjEtMWJjNi00MzQzLThiMmItNGEwZDFmM2UzMWI5IiwibWFzdGVyS2V5QjY0IjoiQlhWM1hvbEM2NTZTVjdkTmdjV1BHUWxrKytycExJNmxHRGk3Q1BCNWllbz0ifQ=="
       #CHUNK_SIZE_MB: "1"
       #POOLING: 1
       #RETRY_ATTEMPTS: 3

docker/Dockerfile

@@ -44,6 +44,15 @@ RUN ARCH=$(uname -m | sed 's/x86_64/amd64/;s/aarch64/arm64/') \
        | tar -xjf - -C /usr/local/bin sqlcmd \
     && chmod +x /usr/local/bin/sqlcmd
 
+# =========================
+# rclone (all storage backends)
+# =========================
+ARG RCLONE_VERSION=1.75.1
+RUN ARCH=$(dpkg --print-architecture) \
+    && curl -fsSL -o /tmp/rclone.deb "https://downloads.rclone.org/v${RCLONE_VERSION}/rclone-v${RCLONE_VERSION}-linux-${ARCH}.deb" \
+    && dpkg -i /tmp/rclone.deb \
+    && rm /tmp/rclone.deb
+
 ARG TARGETARCH
 
 # =========================
@@ -134,6 +143,12 @@ RUN apt-get update && apt-get install -y \
     firebird3.0-utils \
     && rm -rf /var/lib/apt/lists/*
 
+ARG RCLONE_VERSION=1.75.1
+RUN ARCH=$(dpkg --print-architecture) \
+    && curl -fsSL -o /tmp/rclone.deb "https://downloads.rclone.org/v${RCLONE_VERSION}/rclone-v${RCLONE_VERSION}-linux-${ARCH}.deb" \
+    && dpkg -i /tmp/rclone.deb \
+    && rm /tmp/rclone.deb
+
 ENV DOTNET_ROOT=/usr/local/dotnet
 RUN curl -sSL https://dot.net/v1/dotnet-install.sh -o /tmp/dotnet-install.sh \
     && chmod +x /tmp/dotnet-install.sh \

src/services/restore/downloader.rs

@@ -1,14 +1,14 @@
 use super::service::RestoreService;
 
+use crate::services::backup::logger::JobLogger;
+use crate::utils::retry::{RetryPolicy, retry};
 use anyhow::Result;
 use futures::StreamExt;
 use reqwest::{Client, Url};
 use std::path::{Path, PathBuf};
 use std::sync::Arc;
 use std::time::Instant;
 use tokio::io::AsyncWriteExt;
-use crate::services::backup::logger::JobLogger;
-use crate::utils::retry::{RetryPolicy, retry};
 
 fn human_size(bytes: u64) -> String {
     if bytes >= 1024 * 1024 {
@@ -98,7 +98,9 @@ impl RestoreService {
             format!(
                 "Downloading backup '{}' ({})",
                 filename,
-                total.map(human_size).unwrap_or_else(|| "unknown size".to_string())
+                total
+                    .map(human_size)
+                    .unwrap_or_else(|| "unknown size".to_string())
             ),
         );
 
@@ -138,6 +140,16 @@ impl RestoreService {
             );
         }
 
+        if let Some(total) = total
+            && downloaded < total
+        {
+            anyhow::bail!(
+                "Downloaded {} bytes but expected at least {} - backup appears truncated",
+                downloaded,
+                total
+            );
+        }
+
         logger.log(
             "info",
             format!(

src/services/storage/mod.rs

@@ -9,6 +9,7 @@ use providers::azure_blob;
 use providers::google_cloud_storage;
 use providers::google_drive;
 use providers::local;
+use providers::rclone;
 use providers::s3;
 use std::sync::Arc;
 use tracing::{error, info};
@@ -26,7 +27,6 @@ pub trait StorageProvider: Send + Sync {
     ) -> UploadResult;
 }
 
-/// Factory to create provider instance from storage config
 pub fn get_provider(storage: &DatabaseStorage) -> Option<Box<dyn StorageProvider>> {
     info!("Getting provider");
     info!("{:#?}", storage.provider.as_str());
@@ -39,6 +39,7 @@ pub fn get_provider(storage: &DatabaseStorage) -> Option<Box<dyn StorageProvider
         "google-cloud-storage" => Some(Box::new(
             google_cloud_storage::GoogleCloudStorageProvider {},
         )),
+        "rclone" => Some(Box::new(rclone::RcloneProvider {})),
         _ => {
             error!("Unknown storage provider: {}", storage.provider);
             None

src/services/storage/providers/mod.rs

@@ -2,4 +2,5 @@ pub mod azure_blob;
 pub mod google_cloud_storage;
 pub mod google_drive;
 pub mod local;
+pub mod rclone;
 pub mod s3;

src/services/storage/providers/rclone/helpers.rs

@@ -0,0 +1,168 @@
+use anyhow::{Context, Result, bail};
+use bytes::Bytes;
+use futures::{Stream, StreamExt};
+use std::io::Write;
+use std::path::Path;
+use std::pin::Pin;
+use std::process::Stdio;
+use tempfile::NamedTempFile;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::process::Command;
+use tracing::info;
+
+const BLOCKED_BACKEND_TYPES: [&str; 13] = [
+    "local",
+    "alias",
+    "crypt",
+    "chunker",
+    "compress",
+    "union",
+    "combine",
+    "hasher",
+    "archive",
+    "cache",
+    "memory",
+    "http",
+    "googlephotos",
+];
+
+fn sections(config_text: &str) -> Vec<(String, Option<String>)> {
+    let mut out: Vec<(String, Option<String>)> = Vec::new();
+
+    for line in config_text.lines() {
+        let line = line.trim();
+
+        if line.starts_with('[') && line.ends_with(']') && line.len() > 2 {
+            out.push((line[1..line.len() - 1].trim().to_string(), None));
+            continue;
+        }
+
+        let Some((key, value)) = line.split_once('=') else {
+            continue;
+        };
+
+        if key.trim().eq_ignore_ascii_case("type")
+            && let Some(current) = out.last_mut()
+            && current.1.is_none()
+        {
+            current.1 = Some(value.trim().to_ascii_lowercase());
+        }
+    }
+
+    out
+}
+
+
+pub fn validate_config(config_text: &str, remote_name: &str) -> Result<()> {
+    let sections = sections(config_text);
+
+    if sections.is_empty() {
+        bail!("rclone config contains no remote sections");
+    }
+
+    for (name, backend) in &sections {
+        let Some(backend) = backend else { continue };
+        if BLOCKED_BACKEND_TYPES.contains(&backend.as_str()) {
+            bail!("rclone backend type '{backend}' is not allowed (remote '{name}')");
+        }
+    }
+
+    if !sections.iter().any(|(name, _)| name == remote_name) {
+        let available: Vec<&str> = sections.iter().map(|(name, _)| name.as_str()).collect();
+        bail!(
+            "remote '{remote_name}' is not defined in the rclone config (available: {})",
+            available.join(", ")
+        );
+    }
+
+    Ok(())
+}
+
+/// `<remote>:<remote_path>/<remote_file_path>`
+pub fn remote_target(remote_name: &str, remote_path: &str, remote_file_path: &str) -> String {
+    let base = remote_path.trim().trim_matches('/');
+
+    if base.is_empty() {
+        format!("{remote_name}:{remote_file_path}")
+    } else {
+        format!("{remote_name}:{base}/{remote_file_path}")
+    }
+}
+
+pub type RcloneStream = Pin<Box<dyn Stream<Item = Result<Bytes, std::io::Error>> + Send>>;
+
+pub fn write_config(config_text: &str) -> Result<NamedTempFile> {
+    let mut file = NamedTempFile::new().context("failed to create rclone config temp file")?;
+
+    #[cfg(unix)]
+    {
+        use std::os::unix::fs::PermissionsExt;
+        std::fs::set_permissions(file.path(), std::fs::Permissions::from_mode(0o600))
+            .context("failed to restrict rclone config permissions")?;
+    }
+
+    file.write_all(config_text.as_bytes())
+        .context("failed to write rclone config")?;
+    file.flush().context("failed to flush rclone config")?;
+
+    Ok(file)
+}
+
+pub async fn rcat(config_path: &Path, target: &str, mut stream: RcloneStream) -> Result<()> {
+    info!("rclone rcat -> {}", target);
+
+    let mut child = Command::new("rclone")
+        .arg("--config")
+        .arg(config_path)
+        .arg("--contimeout")
+        .arg("30s")
+        .arg("--timeout")
+        .arg("5m")
+        .arg("--retries")
+        .arg("1")
+        .arg("--low-level-retries")
+        .arg("3")
+        .arg("rcat")
+        .arg(target)
+        .stdin(Stdio::piped())
+        .stdout(Stdio::null())
+        .stderr(Stdio::piped())
+        .spawn()
+        .context("failed to spawn rclone (is the binary installed in this image?)")?;
+
+    let mut stderr_pipe = child.stderr.take().context("rclone stderr unavailable")?;
+    let stderr_task = tokio::spawn(async move {
+        let mut buf = String::new();
+        let _ = stderr_pipe.read_to_string(&mut buf).await;
+        buf
+    });
+
+    let mut stdin = child.stdin.take().context("rclone stdin unavailable")?;
+
+    while let Some(chunk) = stream.next().await {
+        let chunk = match chunk {
+            Ok(c) => c,
+            Err(e) => {
+                let _ = child.start_kill();
+                let _ = child.wait().await;
+                return Err(e).context("backup stream failed");
+            }
+        };
+
+        if stdin.write_all(&chunk).await.is_err() {
+            break;
+        }
+    }
+
+    let _ = stdin.flush().await;
+    drop(stdin);
+
+    let status = child.wait().await.context("failed to wait for rclone")?;
+    let stderr = stderr_task.await.unwrap_or_default();
+
+    if !status.success() {
+        bail!("rclone rcat failed ({status}): {}", stderr.trim());
+    }
+
+    Ok(())
+}