chore(release): 1.1.0

dev

CITATION.cff

@@ -22,5 +22,5 @@ keywords:
   - self-hosted
   - portabase
 license: Apache-2.0
-version: 1.0.5-rc.2
-date-released: "2026-02-01"
+version: 1.1.0
+date-released: "2026-02-13"

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "portabase-agent"
-version = "1.0.5-rc.2"
+version = "1.1.0"
 edition = "2024"
 
 [dependencies]
@@ -17,7 +17,7 @@ base64 = "0.22.1"
 thiserror = "2.0.17"
 log = "0.4.29"
 toml = "0.9.10"
-reqwest = { version = "0.13.1", features = ["json", "blocking", "multipart", "stream"] }
+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"] }
 async-trait = "0.1.89"
@@ -32,10 +32,25 @@ tracing-appender = "0.2.4"
 time = { version = "0.3.44", features = ["macros"] }
 mongodb = "3.5.0"
 rand = "0.9.2"
-tokio-util = "0.7.18"
 bytes = "1.11.0"
 async-stream = "0.3.6"
+uuid = { version = "1.20.0", features = ["v4"] }
+tokio-util = "0.7.18"
+aws-config = "1.8.13"
+aws-sdk-s3 = { version = "1.122.0", features = ["behavior-version-latest"] }
+async-compression = { version = "0.4.37", features = ["tokio", "gzip"] }
+tokio-tar = "0.3.1"
+oauth2 = "5.0.0"
+hyper = "1.8.1"
+async-http-client = "0.2.0"
+aes-gcm = "0.11.0-rc.3"
+generic-array = "0.14.7"
+futures-util = "0.3.31"
+tokio-stream = "0.1.18"
+aes = "0.9.0-rc.4"
+typenum = "1.19.0"
+
 
 [[bin]]
 name = "app"
-path = "src/main.rs"
+path = "src/main.rs"

docker-compose.yml

@@ -11,12 +11,13 @@ services:
       #      - ./databases.toml:/config/config.toml
       - cargo-registry:/usr/local/cargo/registry
       - cargo-git:/usr/local/cargo/git
-#      - cargo-target:/app/target
+      - cargo-target:/app/target
     environment:
       APP_ENV: development
       LOG: debug
       TZ: "Europe/Paris"
-      EDGE_KEY: "eyJzZXJ2ZXJVcmwiOiJodHRwOi8vbG9jYWxob3N0Ojg4ODciLCJhZ2VudElkIjoiYTMxNWFhZmUtNDcyNi00OWM0LThhMmEtMGM2ZWQ0N2E1NjVkIiwicHVibGljS2V5IjoiLS0tLS1CRUdJTiBSU0EgUFVCTElDIEtFWS0tLS0tXG5NSUlCQ2dLQ0FRRUE5TWV4M2pmdnVLdFB5YU1ERnh2Ulp2dmd3YkRJQ2JzQi81Wll5NDNSVVRBaXZRYjJiSDdYXG5qRHBQd1lJeCs4UFBrbHlRbDVMQzV1UWZEaCs4SVd4OG1LZ3FvMXpWMkdiZXdGbEdEWFYxVEdyU1ZEU25aSWR4XG52bWdYc29EeXhVMlJvWUFUMS9YMWxuc2YxenZKdkFMTkhXdEhRdk42SjVDZTFSMmFsendVRGFEVXlJNzRmSldQXG5tNTh0SDMrYklXL0VVTXdjaWNxM0oySWw3Vm9KNkZNUHJQL1ZSOWEvdFF1SU1qa200MXpFY2NscExPa2luRkxuXG54NmVUWkFSZUpya2UrbnRvZ2t4TGEyRWV5a1lUNzB4V3hKNWp5ZExBVnRvNkkyQlVLVVJoTkowTUFaU29NYUtvXG5iMGJRcnY1UzExZWllMnMrT2I3aTYzSFpkVUx0UmV1MVJ3SURBUUFCXG4tLS0tLUVORCBSU0EgUFVCTElDIEtFWS0tLS0tXG4ifQ"
+      EDGE_KEY: "eyJzZXJ2ZXJVcmwiOiJodHRwOi8vbG9jYWxob3N0Ojg4ODciLCJhZ2VudElkIjoiMzI3YTU4ODktYzE0MC00ODMzLTk1ZWMtNTBmMmU2NTFlZmJhIiwibWFzdGVyS2V5QjY0IjoiQlhWM1hvbEM2NTZTVjdkTmdjV1BHUWxrKytycExJNmxHRGk3Q1BCNWllbz0ifQ=="
+      #      EDGE_KEY: "eyJzZXJ2ZXJVcmwiOiJodHRwOi8vbG9jYWxob3N0Ojg4ODciLCJhZ2VudElkIjoiZWE2NTg1MDctZTA5My00NDUxLWIxZDAtMDgwZWZjMGNmNWYzIiwicHVibGljS2V5IjoiLS0tLS1CRUdJTiBSU0EgUFVCTElDIEtFWS0tLS0tXG5NSUlCQ2dLQ0FRRUE5TWV4M2pmdnVLdFB5YU1ERnh2Ulp2dmd3YkRJQ2JzQi81Wll5NDNSVVRBaXZRYjJiSDdYXG5qRHBQd1lJeCs4UFBrbHlRbDVMQzV1UWZEaCs4SVd4OG1LZ3FvMXpWMkdiZXdGbEdEWFYxVEdyU1ZEU25aSWR4XG52bWdYc29EeXhVMlJvWUFUMS9YMWxuc2YxenZKdkFMTkhXdEhRdk42SjVDZTFSMmFsendVRGFEVXlJNzRmSldQXG5tNTh0SDMrYklXL0VVTXdjaWNxM0oySWw3Vm9KNkZNUHJQL1ZSOWEvdFF1SU1qa200MXpFY2NscExPa2luRkxuXG54NmVUWkFSZUpya2UrbnRvZ2t4TGEyRWV5a1lUNzB4V3hKNWp5ZExBVnRvNkkyQlVLVVJoTkowTUFaU29NYUtvXG5iMGJRcnY1UzExZWllMnMrT2I3aTYzSFpkVUx0UmV1MVJ3SURBUUFCXG4tLS0tLUVORCBSU0EgUFVCTElDIEtFWS0tLS0tXG4ifQ"
       #POOLING: 1
       #DATABASES_CONFIG_FILE: "config.toml"
     extra_hosts:
@@ -94,7 +95,7 @@ services:
 volumes:
   cargo-registry:
   cargo-git:
-#  cargo-target:
+  cargo-target:
 
 
   postgres-data:

scripts/README.md

@@ -1,11 +1,12 @@
 # Seed instructions
 
-## MongoDB 
+## MongoDB
 
 ```bash
 make seed-mongo     
 make seed-mongo-auth  
 make seed-mysql      
+make seed-mysql-1gb    
 make seed-postgres      
 make seed-postgres-1gb  
 make seed-all      

src/core/agent.rs

@@ -52,7 +52,7 @@ impl Agent {
             if db.data.backup.action {
                 let _ = self
                     .backup_service
-                    .dispatch(&db.generated_id, &config, method.clone())
+                    .dispatch(&db.generated_id, &config, method.clone(), &db.storages, db.encrypt)
                     .await;
             } else if db.data.restore.action {
                 let _ = self

src/core/context.rs

@@ -1,3 +1,4 @@
+use crate::services::api::ApiClient;
 use crate::settings::CONFIG;
 use crate::utils::edge_key::{EdgeKey, EdgeKeyError, decode_edge_key};
 use tracing::{debug, error, info};
@@ -6,6 +7,7 @@ use tracing::{debug, error, info};
 pub struct Context {
     #[allow(dead_code)]
     pub edge_key: EdgeKey,
+    pub api: ApiClient,
 }
 
 impl Context {
@@ -33,7 +35,13 @@ impl Context {
                 panic!("Cannot initialize AgentContext due to invalid EDGE_KEY");
             }
         };
+        
+        let server_url = format!("{}/api", edge_key.server_url);
+        let api_client = ApiClient::new(server_url);
 
-        Context { edge_key }
+        Context {
+            edge_key: edge_key,
+            api: api_client,
+        }
     }
 }

src/domain/postgres/database.rs

@@ -28,7 +28,6 @@ impl Database for PostgresDatabase {
         match self.format {
             PostgresDumpFormat::Fc => ".dump",
             PostgresDumpFormat::Fd => ".gz",
-            // PostgresDumpFormat::Fd => ".tar.gz",
         }
     }
 

src/domain/postgres/restore.rs

@@ -86,9 +86,13 @@ pub async fn run(
                     }
                 };
 
+                info!("tar_gz {:?}", tar_gz);
+
                 let dec = flate2::read::GzDecoder::new(tar_gz);
                 let mut archive = tar::Archive::new(dec);
 
+
+
                 let tmp_dir = match tempfile::TempDir::new() {
                     Ok(d) => d,
                     Err(e) => {
@@ -100,12 +104,16 @@ pub async fn run(
                     }
                 };
 
+
+
                 if let Err(e) = archive.unpack(tmp_dir.path()) {
                     error!("Failed to unpack FD archive for {}: {:?}", cfg.name, e);
                     return Err(e.into());
                 }
 
                 debug!("Listing contents of temp dir: {}", tmp_dir.path().display());
+
+
                 for entry in std::fs::read_dir(tmp_dir.path())? {
                     if let Ok(entry) = entry {
                         let path = entry.path();

src/services/api/client.rs

@@ -0,0 +1,83 @@
+#![allow(dead_code)]
+
+use reqwest::{Client, Method};
+use serde::de::DeserializeOwned;
+use std::time::Duration;
+use crate::services::api::ApiError;
+
+#[derive(Clone, Debug)]
+pub struct ApiClient {
+    base_url: String,
+    http: Client,
+}
+
+impl ApiClient {
+    pub fn new(base_url: impl Into<String>) -> Self {
+        let http = Client::builder()
+            .timeout(Duration::from_secs(30))
+            .build()
+            .expect("failed to build http client");
+
+        Self {
+            base_url: base_url.into(),
+            http,
+        }
+    }
+
+    pub async fn request<T: DeserializeOwned>(
+        &self,
+        method: Method,
+        path: &str,
+    ) -> Result<Option<T>, ApiError> {
+        let url = format!("{}{}", self.base_url, path);
+
+        let res = self.http.request(method, &url).send().await?;
+        let status = res.status();
+        let body = res.text().await.unwrap_or_default();
+
+        if !status.is_success() {
+            return Err(ApiError::HttpResponse { status, body });
+        }
+
+        if body.trim().is_empty() {
+            Ok(None)
+        } else {
+            Ok(Some(serde_json::from_str::<T>(&body)?))
+        }
+    }
+    pub async fn request_with_body<T, B>(
+        &self,
+        method: Method,
+        path: &str,
+        body: &B,
+    ) -> Result<Option<T>, ApiError>
+    where
+        T: DeserializeOwned,
+        B: serde::Serialize,
+    {
+        let url = format!("{}{}", self.base_url, path);
+
+        let res = self
+            .http
+            .request(method, &url)
+            .json(body)
+            .send()
+            .await?;
+
+        let status = res.status();
+        let body_text = res.text().await.unwrap_or_default();
+
+        if !status.is_success() {
+            return Err(ApiError::HttpResponse {
+                status,
+                body: body_text,
+            });
+        }
+
+        if body_text.trim().is_empty() {
+            Ok(None)
+        } else {
+            Ok(Some(serde_json::from_str::<T>(&body_text)?))
+        }
+    }
+}

src/services/api/endpoints/agent/backup/mod.rs

@@ -0,0 +1,67 @@
+pub mod upload;
+
+use crate::services::api::models::agent::backup::BackupResponse;
+use crate::services::api::{ApiClient, ApiError};
+use anyhow::Result;
+use reqwest::Method;
+use serde::Serialize;
+
+#[derive(Serialize)]
+pub struct BackupCreateRequest {
+    pub method: String,
+    #[serde(rename = "generatedId")]
+    pub generated_id: String,
+}
+
+#[derive(Serialize)]
+pub struct BackupUpdateRequest {
+    #[serde(rename = "backupId")]
+    pub backup_id: String,
+    pub status: String,
+    pub size: Option<u64>,
+    #[serde(rename = "generatedId")]
+    pub generated_id: String,
+}
+
+
+impl ApiClient {
+    pub async fn backup_create(
+        &self,
+        method: impl Into<String>,
+        agent_id: impl Into<String>,
+        generated_id: impl Into<String>,
+    ) -> Result<Option<BackupResponse>, ApiError> {
+        let body = BackupCreateRequest {
+            method: method.into(),
+            generated_id: generated_id.into(),
+        };
+
+        let agent_id = agent_id.into();
+        let path = format!("/agent/{}/backup", agent_id);
+
+        self.request_with_body(Method::POST, path.as_str(), &body)
+            .await
+    }
+
+    pub async fn backup_update(
+        &self,
+        agent_id: impl Into<String>,
+        backup_id: impl Into<String>,
+        status: impl Into<String>,
+        file_size: impl Into<Option<u64>>,
+        generated_id: impl Into<String>,
+    ) -> Result<Option<BackupResponse>, ApiError> {
+        let body = BackupUpdateRequest {
+            backup_id: backup_id.into(),
+            status: status.into(),
+            size: file_size.into(),
+            generated_id: generated_id.into(),
+        };
+
+        let agent_id = agent_id.into();
+        let path = format!("/agent/{}/backup", agent_id);
+
+        self.request_with_body(Method::PATCH, path.as_str(), &body)
+            .await
+    }
+}

src/services/api/endpoints/agent/backup/upload/init.rs

@@ -0,0 +1,37 @@
+use crate::services::api::{ApiClient, ApiError};
+use anyhow::Result;
+use reqwest::Method;
+use serde::Serialize;
+use crate::services::api::models::agent::backup::BackupUploadResponse;
+
+#[derive(Serialize)]
+pub struct InitUploadRequest {
+    #[serde(rename = "generatedId")]
+    pub generated_id: String,
+    #[serde(rename = "storageChannelId")]
+    pub storage_channel_id: String,
+    #[serde(rename = "backupId")]
+    pub backup_id: String,
+}
+
+impl ApiClient {
+    pub async fn backup_upload_init(
+        &self,
+        agent_id: impl Into<String>,
+        generated_id: impl Into<String>,
+        storage_channel_id: impl Into<String>,
+        backup_id: impl Into<String>,
+    ) -> Result<Option<BackupUploadResponse>, ApiError> {
+        let body = InitUploadRequest {
+            generated_id: generated_id.into(),
+            storage_channel_id: storage_channel_id.into(),
+            backup_id: backup_id.into(),
+        };
+
+        let agent_id = agent_id.into();
+        let path = format!("/agent/{}/backup/upload/init", agent_id);
+
+        self.request_with_body(Method::POST, path.as_str(), &body)
+            .await
+    }
+}

src/services/api/endpoints/agent/backup/upload/mod.rs

@@ -0,0 +1,2 @@
+pub mod init;
+pub mod status;

src/services/api/endpoints/agent/backup/upload/status.rs

@@ -0,0 +1,46 @@
+use crate::services::api::models::agent::backup::BackupUploadResponse;
+use crate::services::api::{ApiClient, ApiError};
+use anyhow::Result;
+use reqwest::Method;
+use serde::Serialize;
+
+#[derive(Serialize)]
+pub struct StatusUploadRequest {
+    #[serde(rename = "generatedId")]
+    pub generated_id: String,
+    #[serde(rename = "backupStorageId")]
+    pub backup_storage_id: String,
+    pub status: String,
+    pub path: String,
+    pub size: u64,
+    #[serde(rename = "backupId")]
+    pub backup_id: String,
+}
+
+impl ApiClient {
+    pub async fn backup_upload_status(
+        &self,
+        agent_id: impl Into<String>,
+        generated_id: impl Into<String>,
+        backup_storage_id: impl Into<String>,
+        status: impl Into<String>,
+        remote_path: impl Into<String>,
+        total_size: impl Into<u64>,
+        backup_id: impl Into<String>,
+    ) -> Result<Option<BackupUploadResponse>, ApiError> {
+        let body = StatusUploadRequest {
+            generated_id: generated_id.into(),
+            backup_storage_id: backup_storage_id.into(),
+            status: status.into(),
+            path: remote_path.into(),
+            size: total_size.into(),
+            backup_id: backup_id.into(),
+        };
+
+        let agent_id = agent_id.into();
+        let path = format!("/agent/{}/backup/upload/status", agent_id);
+
+        self.request_with_body(Method::PATCH, path.as_str(), &body)
+            .await
+    }
+}

src/services/api/endpoints/agent/mod.rs

@@ -0,0 +1,2 @@
+pub mod status;
+pub mod backup;

src/services/api/endpoints/agent/status.rs

@@ -0,0 +1,36 @@
+use crate::services::api::models::agent;
+use crate::services::api::{ApiClient, ApiError};
+use agent::status::PingResult;
+use anyhow::Result;
+use reqwest::Method;
+use serde::Serialize;
+
+#[derive(Serialize)]
+pub struct DatabasePayload<'a> {
+    pub name: &'a str,
+    pub dbms: &'a str,
+    #[serde(rename = "generatedId")]
+    pub generated_id: &'a str,
+}
+
+#[derive(Serialize)]
+pub struct StatusRequest<'a> {
+    pub version: &'a str,
+    pub databases: Vec<DatabasePayload<'a>>,
+}
+
+impl ApiClient {
+    pub async fn agent_status<'a>(
+        &self,
+        agent_id: impl Into<String>,
+        version: &'a str,
+        databases: Vec<DatabasePayload<'a>>,
+    ) -> Result<Option<PingResult>, ApiError> {
+        let body = StatusRequest { version, databases };
+
+        let agent_id = agent_id.into();
+        let path = format!("/agent/{}/status", agent_id);
+
+        self.request_with_body(Method::POST, path.as_str(), &body).await
+    }
+}

src/services/api/endpoints/mod.rs

@@ -0,0 +1,3 @@
+pub mod agent;
+
+pub use agent::status;

src/services/api/error.rs

@@ -0,0 +1,21 @@
+use reqwest::StatusCode;
+use thiserror::Error;
+
+#[derive(Debug, Error)]
+#[allow(dead_code)]
+pub enum ApiError {
+    #[error("http client error: {0}")]
+    Http(#[from] reqwest::Error),
+
+    #[error("serialization error: {0}")]
+    Serialization(#[from] serde_json::Error),
+
+    #[error("api error: status={status}, body={body}")]
+    HttpResponse {
+        status: StatusCode,
+        body: String,
+    },
+
+    #[error("api returned unexpected response")]
+    UnexpectedResponse,
+}

src/services/api/mod.rs

@@ -0,0 +1,8 @@
+pub mod client;
+pub mod error;
+pub mod models;
+pub mod endpoints;
+
+pub use client::ApiClient;
+pub use error::ApiError;
+

src/services/api/models/agent/backup.rs

@@ -0,0 +1,24 @@
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Deserialize, Serialize)]
+pub struct BackupStorage {
+    pub id: String,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct BackupUploadResponse {
+    pub message: String,
+    #[serde(rename = "backupStorage")]
+    pub backup_storage: BackupStorage,
+}
+
+#[derive(Debug, Deserialize, Serialize)]
+pub struct Backup {
+    pub id: String,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct BackupResponse {
+    pub message: String,
+    pub backup: Backup,
+}

src/services/api/models/agent/mod.rs

@@ -0,0 +1,2 @@
+pub mod status;
+pub mod backup;

src/services/api/models/agent/status.rs

@@ -0,0 +1,57 @@
+#![allow(dead_code)]
+
+use serde::{Deserialize, Serialize};
+use toml::Value;
+use crate::utils::deserializer::deserialize_snake_case;
+
+#[derive(Debug, Deserialize)]
+pub struct PingResult {
+    pub agent: AgentInfo,
+    pub databases: Vec<DatabaseStatus>,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct AgentInfo {
+    pub id: String,
+    #[serde(rename = "lastContact")]
+    pub last_contact: String,
+}
+
+#[derive(Debug, Deserialize, Serialize, Clone)]
+#[serde(rename_all = "snake_case")]
+pub struct DatabaseStorage {
+    pub id: String,
+    #[serde(deserialize_with = "deserialize_snake_case")]
+    pub config: Value,
+    pub provider: String,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct DatabaseStatus {
+    pub dbms: String,
+    #[serde(rename = "generatedId")]
+    pub generated_id: String,
+    pub storages: Vec<DatabaseStorage>,
+    pub encrypt: bool,
+    pub data: DatabaseData,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct DatabaseData {
+    pub backup: BackupInfo,
+    pub restore: RestoreInfo,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct BackupInfo {
+    pub action: bool,
+    pub cron: Option<String>,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct RestoreInfo {
+    pub action: bool,
+    pub file: Option<String>,
+    #[serde(rename = "metaFile")]
+    pub meta_file: Option<String>,
+}

src/services/api/models/mod.rs

@@ -0,0 +1,2 @@
+pub mod agent;
+

src/services/backup.rs

@@ -1,21 +1,21 @@
 #![allow(dead_code)]
 
-use crate::core::context::Context;
+use crate::core::context::Context as CoreContext;
 use crate::domain::factory::DatabaseFactory;
+use crate::services::api::models::agent::status::DatabaseStorage;
 use crate::services::config::{DatabaseConfig, DatabasesConfig, DbType};
+use crate::services::storage;
 use crate::utils::common::BackupMethod;
-use crate::utils::file::{encrypt_file_stream, full_extension};
+use crate::utils::compress::compress_to_tar_gz_large;
 use anyhow::Result;
-use futures::StreamExt;
-use hex;
-use rand::RngCore;
-use reqwest::Body;
+use futures::future::join_all;
 use std::path::{Path, PathBuf};
 use std::sync::Arc;
 use tempfile::TempDir;
 use tracing::{error, info};
+use crate::utils::locks::FileLock;
 
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct BackupResult {
     pub generated_id: String,
     pub db_type: DbType,
@@ -24,12 +24,21 @@ pub struct BackupResult {
     pub code: Option<String>,
 }
 
+#[derive(Debug)]
+pub struct UploadResult {
+    pub storage_id: String,
+    pub success: bool,
+    pub error: Option<String>,
+    pub remote_file_path: Option<String>,
+    pub total_size: Option<u64>,
+}
+
 pub struct BackupService {
-    ctx: Arc<Context>,
+    ctx: Arc<CoreContext>,
 }
 
 impl BackupService {
-    pub fn new(ctx: Arc<Context>) -> Self {
+    pub fn new(ctx: Arc<CoreContext>) -> Self {
         Self { ctx }
     }
 
@@ -38,6 +47,8 @@ impl BackupService {
         generated_id: &String,
         config: &DatabasesConfig,
         method: BackupMethod,
+        storages: &Vec<DatabaseStorage>,
+        encrypt: bool,
     ) {
         if let Some(cfg) = config
             .databases
@@ -45,22 +56,100 @@ impl BackupService {
             .find(|c| c.generated_id == generated_id.as_str())
         {
             let db_cfg = cfg.clone();
-            let ctx_clone = self.ctx.clone();
+            let ctx = self.ctx.clone();
+            let storages_clone = storages.clone();
+            let generated_id_clone = generated_id.clone();
 
             tokio::spawn(async move {
                 match TempDir::new() {
                     Ok(temp_dir) => {
-                        let tmp_path = temp_dir.path().to_path_buf();
-                        info!("Created temp directory {}", tmp_path.display());
-
-                        match BackupService::run(db_cfg, &tmp_path).await {
-                            Ok(result) => {
-                                let service = BackupService { ctx: ctx_clone };
-                                service.send_result(result, method).await;
+                        match FileLock::is_locked(&generated_id_clone).await {
+                            Ok(true) => {
+                                error!("Backup already running for {}", &generated_id_clone);
+                                return;
                             }
-                            Err(e) => error!("Backup error {}", e),
+                            Ok(false) => {
+
+                                match ctx
+                                    .api
+                                    .backup_create(
+                                        method.clone().to_string(),
+                                        ctx.edge_key.agent_id.clone(),
+                                        &generated_id_clone,
+                                    )
+                                    .await
+                                {
+                                    Ok(backup_created_result) => {
+                                        info!("Backup created successfully");
+                                        let tmp_path = temp_dir.path().to_path_buf();
+                                        info!("Created temp directory {}", tmp_path.display());
+                                        match BackupService::run(db_cfg, &tmp_path).await {
+                                            Ok(mut result) => {
+                                                if let Some(backup_file) = result.backup_file.take() {
+                                                    match compress_to_tar_gz_large(&backup_file).await {
+                                                        Ok(compression_result) => {
+                                                            result.backup_file =
+                                                                Some(compression_result.compressed_path);
+                                                            let service = BackupService { ctx: ctx.clone() };
+                                                            let backup_id = backup_created_result.unwrap().backup.id;
+                                                            match service
+                                                                .upload(
+                                                                    result.clone(),
+                                                                    method,
+                                                                    storages_clone.clone(),
+                                                                    encrypt,
+                                                                    &backup_id,
+                                                                )
+                                                                .await
+                                                            {
+                                                                Ok(upload_result) => {
+                                                                    match service
+                                                                        .send_result(
+                                                                            result,
+                                                                            upload_result,
+                                                                            &backup_id
+                                                                        )
+                                                                        .await
+                                                                    {
+                                                                        Ok(_) => {
+                                                                            return;
+                                                                        }
+                                                                        Err(e) => {
+                                                                            error!(
+                                                                        "Failed to send backup result: {}",
+                                                                        e
+                                                                    );
+                                                                        }
+                                                                    }
+                                                                }
+                                                                Err(e) => {
+                                                                    error!(
+                                                                "Failed to upload backup files: {}",
+                                                                e
+                                                            );
+                                                                }
+                                                            }
+                                                        }
+                                                        Err(e) => {
+                                                            error!(
+                                                        "Failed to compress backup file : {}",
+                                                        e
+                                                    );
+                                                        }
+                                                    }
+                                                } else {
+                                                    error!("No backup file generated");
+                                                }
+                                            }
+                                            Err(e) => error!("BackupService run failed: {}", e),
+                                        }
+                                        // TempDir is automatically deleted when dropped here
+                                    }
+                                    Err(e) => error!("Backup creation failed: {}", e),
+                                }
+                            },
+                            Err(e) => error!("An error occurred while checking lock : {}", e),
                         }
-                        // TempDir is automatically deleted when dropped here
                     }
                     Err(e) => error!("Failed to create temp dir: {}", e),
                 }
@@ -112,72 +201,185 @@ impl BackupService {
         }
     }
 
-    pub async fn send_result(&self, result: BackupResult, method: BackupMethod) {
+    pub async fn upload(
+        &self,
+        result: BackupResult,
+        method: BackupMethod,
+        storages: Vec<DatabaseStorage>,
+        encrypt: bool,
+        backup_id: &String,
+    ) -> Result<Vec<UploadResult>> {
         if result.code.as_deref() == Some("backup_already_in_progress") {
             info!("Skipping send: backup already in progress");
-            return;
+            anyhow::bail!("backup_already_in_progres");
         }
 
-        let Some(file_path) = result.backup_file else {
-            return;
+        let upload_futures = storages.into_iter().map(|storage| {
+            info!(
+                "Uploading storage -> {:?} for {:?}",
+                storage.provider, storage.id
+            );
+            let provider = storage::get_provider(&storage);
+            let result_clone = result.clone();
+            let ctx_clone = self.ctx.clone();
+            let storages_clone = storage.clone();
+            let storage_id = storages_clone.id;
+            let generated_id = result_clone.generated_id.clone();
+
+            async move {
+                match self
+                    .ctx
+                    .api
+                    .backup_upload_init(
+                        self.ctx.edge_key.agent_id.clone(),
+                        generated_id.clone(),
+                        storage_id.clone(),
+                        backup_id,
+                    )
+                    .await
+                {
+                    Ok(upload_init_result) => {
+                        info!("Uploading init result: {:#?}", upload_init_result);
+                        let backup_storage_id = upload_init_result.unwrap().backup_storage.id.clone();
+                        match provider {
+                            Some(provider) => {
+                                let upload_result = provider
+                                    .upload(
+                                        ctx_clone,
+                                        result_clone,
+                                        method,
+                                        &storage,
+                                        Some(encrypt),
+                                    )
+                                    .await;
+
+                                let status = if upload_result.success {
+                                    "success"
+                                } else {
+                                    "failed"
+                                };
+                                info!("Storage {} uploaded to remote path {:?}", storage_id, upload_result.remote_file_path);
+
+
+                                let (remote_path, total_size) = match (
+                                    &upload_result.remote_file_path,
+                                    upload_result.total_size,
+                                ) {
+                                    (Some(path), Some(size)) => (path.clone(), size),
+                                    _ => {
+                                        return UploadResult {
+                                            storage_id: storage_id.clone(),
+                                            success: false,
+                                            error: Some("remote_file_path or total_size missing".to_string()),
+                                            remote_file_path: None,
+                                            total_size: None,
+                                        }
+                                    }
+                                };
+                                
+                                match self.ctx.api.backup_upload_status(
+                                    self.ctx.edge_key.agent_id.clone(),
+                                    generated_id.clone(),
+                                    backup_storage_id,
+                                    status,
+                                    remote_path,
+                                    total_size,
+                                    backup_id
+                                ).await {
+                                    Ok(_) => {
+                                        upload_result
+                                    },
+                                    Err(err)=> {
+                                        error!(
+                                        "backup_upload_status failed (generated_id={}, storage_id={}): {}",
+                                        generated_id, storage_id, err
+                                    );
+                                        UploadResult {
+                                            storage_id: storage_id.clone(),
+                                            success: false,
+                                            error: Some(err.to_string()),
+                                            remote_file_path: None,
+                                            total_size: None,
+                                        }
+                                    }
+                                }
+                            }
+                            None => {
+                                error!("Skipping storage due to missing provider");
+                                UploadResult {
+                                    storage_id: storage_id.clone(),
+                                    success: false,
+                                    error: Some(
+                                        "Skipping storage due to missing provider".to_string(),
+                                    ),
+                                    remote_file_path: None,
+                                    total_size: None,
+                                }
+                            }
+                        }
+                    }
+                    Err(e) => {
+                        error!(
+                            "Unable to create the storage backup on remote server : {}",
+                            e
+                        );
+                        UploadResult {
+                            storage_id: storage_id.clone(),
+                            success: false,
+                            error: Some(
+                                "Unable to create the storage backup on remote server".to_string(),
+                            ),
+                            remote_file_path: None,
+                            total_size: None,
+                        }
+                    }
+                }
+            }
+        });
+
+        let results: Vec<UploadResult> = join_all(upload_futures).await;
+        info!("Upload results: {:#?}", results);
+
+        Ok(results)
+    }
+
+    pub async fn send_result(
+        &self,
+        result: BackupResult,
+        upload_results: Vec<UploadResult>,
+        backup_id: &String,
+    ) -> Result<()> {
+        let status = if upload_results.iter().any(|r| r.success) {
+            "success"
+        } else {
+            "failed"
         };
 
-        let mut aes_key = [0u8; 32];
-        rand::rng().fill_bytes(&mut aes_key);
-
-        let mut iv = [0u8; 16];
-        rand::rng().fill_bytes(&mut iv);
-
-        let public_key_pem = self.ctx.edge_key.public_key.as_bytes().to_vec();
-
-        let (encrypted_stream, encrypted_key_hex) =
-            match encrypt_file_stream(file_path.clone(), aes_key, iv, public_key_pem).await {
-                Ok(v) => v,
-                Err(e) => {
-                    error!("Encryption failed: {}", e);
-                    return;
+        let file_size = upload_results
+            .iter()
+            .map(|r| r.total_size)
+            .try_fold((0u64, 0u64), |(sum, count), v| {
+                match v {
+                    Some(size) => Ok((sum + size, count + 1)),
+                    None => Err(()), // stop and return None
                 }
-            };
+            })
+            .ok()
+            .map(|(sum, count)| sum / count);
 
-        let file_size = std::fs::metadata(&file_path).map(|m| m.len()).unwrap_or(0);
-
-        let body = Body::wrap_stream(
-            encrypted_stream
-                .map(|r| r.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e))),
-        );
-
-        let url = format!(
-            "{}/services/v1/upload/{}",
-            self.ctx.edge_key.server_url, self.ctx.edge_key.agent_id
-        );
-
-        let client = reqwest::Client::new();
-
-        info!("file Size to {}", file_size);
-
-        let resp = client
-            .post(&url)
-            .header("X-Generated-Id", &result.generated_id)
-            .header("X-Status", &result.status)
-            .header("X-Method", method.to_string())
-            .header("X-AES-Key", encrypted_key_hex)
-            .header("X-IV", hex::encode(iv))
-            .header("X-Extension", full_extension(&file_path))
-            .header("Transfer-Encoding", "chunked")
-            .header("X-File-Size", file_size)
-            .body(body)
-            .send()
-            .await;
-
-        match resp {
-            Ok(r) if r.status().is_success() => {
-                info!("Backup uploaded successfully");
-            }
-            Ok(r) => {
-                error!("Upload failed: {}", r.status());
-            }
+        match self
+            .ctx
+            .api
+            .backup_update(self.ctx.edge_key.agent_id.clone(), backup_id, status, file_size, &result.generated_id)
+            .await
+        {
+            Ok(_result) => Ok(()),
             Err(e) => {
-                error!("Upload error: {}", e);
+                error!(
+                    "backup_update failed (generated_id={}, backup_id={}): {}",
+                    &result.generated_id, &backup_id, e
+                );
+                Err(e.into())
             }
         }
     }

src/services/cron.rs

@@ -1,11 +1,13 @@
 #![allow(dead_code)]
 
 use crate::core::context::Context;
-use crate::services::status::DatabaseStatus;
+use crate::utils::common::vec_to_option_json;
 use crate::utils::redis_client;
 use crate::utils::task_manager::cron::check_and_update_cron;
-use std::sync::Arc;
 use redis::aio::MultiplexedConnection;
+use serde_json::{Value, json};
+use std::sync::Arc;
+use crate::services::api::models::agent::status::DatabaseStatus;
 
 pub struct CronService {
     ctx: Arc<Context>,
@@ -23,6 +25,12 @@ impl CronService {
         let dbms = database.dbms.as_str();
         let task_name = format!("periodic.backup_{}", generated_id);
         let args = vec![generated_id.to_string(), dbms.to_string()];
+        let storages: Option<Value> = vec_to_option_json(database.storages.clone());
+        let encrypt: bool = database.encrypt;
+        let metadata = json!({
+            "storages": storages,
+            "encrypt": encrypt
+        });
 
         check_and_update_cron(
             &mut self.conn,
@@ -30,7 +38,9 @@ impl CronService {
             args,
             "tasks.database.periodic_backup",
             task_name,
-        ).await;
+            Option::from(metadata),
+        )
+        .await;
 
         Ok(true)
     }

src/services/mod.rs

@@ -1,5 +1,7 @@
 pub mod config;
-pub mod status;
 pub mod cron;
 pub mod backup;
-pub mod restore;
+pub mod restore;
+mod storage;
+pub mod api;
+pub mod status;

src/services/restore.rs

@@ -1,15 +1,18 @@
 #![allow(dead_code)]
-
+#![warn(unused_assignments)]
 use crate::core::context::Context;
 use crate::domain::factory::DatabaseFactory;
+use crate::services::api::models::agent::status::DatabaseStatus;
 use crate::services::config::{DatabaseConfig, DatabasesConfig};
-use crate::services::status::DatabaseStatus;
+use crate::utils::compress::decompress_large_tar_gz;
+use crate::utils::file::decrypt_file_stream_gcm;
 use anyhow::Result;
-use tracing::{error, info};
+use reqwest::{Client, Url};
 use serde::Serialize;
-use std::path::Path;
+use std::path::{Path, PathBuf};
 use std::sync::Arc;
 use tempfile::TempDir;
+use tracing::{error, info};
 
 #[derive(Debug, Serialize)]
 pub struct RestoreResult {
@@ -36,21 +39,30 @@ impl RestoreService {
             let db_cfg = cfg.clone();
             let ctx_clone = self.ctx.clone();
             let file_to_restore = db.data.restore.file.clone();
-
+            if file_to_restore.is_none() {
+                error!("restore file not found");
+                return;
+            }
             tokio::spawn(async move {
                 match TempDir::new() {
                     Ok(temp_dir) => {
                         let tmp_path = temp_dir.path().to_path_buf();
                         info!("Created temp directory {}", tmp_path.display());
-
-                        match RestoreService::run(db_cfg, &tmp_path, &file_to_restore).await {
+                        match RestoreService::run(
+                            &ctx_clone,
+                            db_cfg,
+                            &tmp_path,
+                            &file_to_restore.unwrap(),
+                        )
+                        .await
+                        {
                             Ok(result) => {
                                 let service = RestoreService { ctx: ctx_clone };
                                 service.send_result(result).await;
                             }
                             Err(e) => error!("Restoration error {}", e),
                         }
-                        // TempDir is automatically deleted when dropped here
+                        // TempDir is automatically deleted when dropped
                     }
                     Err(e) => error!("Failed to create temp dir: {}", e),
                 }
@@ -59,6 +71,7 @@ impl RestoreService {
     }
 
     pub async fn run(
+        ctx: &Arc<Context>,
         cfg: DatabaseConfig,
         tmp_path: &Path,
         file_url: &str,
@@ -67,8 +80,9 @@ impl RestoreService {
 
         info!("File url: {}", file_url);
 
-        let client = reqwest::Client::new();
+        let client = Client::new();
         let response = client.get(file_url).send().await?;
+
         if !response.status().is_success() {
             error!("Backup download failed with status {}", response.status());
             return Ok(RestoreResult {
@@ -77,27 +91,96 @@ impl RestoreService {
             });
         }
 
+        let filename_from_header = response
+            .headers()
+            .get(reqwest::header::CONTENT_DISPOSITION)
+            .and_then(|v| v.to_str().ok())
+            .and_then(|s| s.split("filename=").nth(1))
+            .map(|f| f.trim_matches('"').to_string());
+
+        let filename_from_url = Url::parse(file_url).ok().and_then(|u| {
+            u.path_segments()?
+                .last()
+                .filter(|s| !s.is_empty())
+                .map(|s| s.to_string())
+        });
+
+        let filename = filename_from_header
+            .or(filename_from_url)
+            .unwrap_or_else(|| "downloaded_file".to_string());
+
+        info!("File name: {}", filename);
+
         let bytes = response.bytes().await?;
 
-        let ext = if bytes.starts_with(b"PGDMP") {
-            // Postgres custom format
-            "dump"
-        } else if bytes.starts_with(&[0x1F, 0x8B]) {
-            // gzip compressed -> could be Postgres directory dump or MySQL gzipped SQL
-            "tar.gz"
-        } else if bytes.starts_with(b"--") || bytes.starts_with(b"/*") {
-            // Plain MySQL SQL dump
-            "sql"
+        let is_legacy_file = if filename.ends_with(".sql") {
+            true
+        } else if filename.ends_with(".dump") {
+            true
         } else {
-            // Fallback generic
-            "dump"
+            false
         };
+        let downloaded_file = tmp_path.join(&filename);
+        tokio::fs::write(&downloaded_file, &bytes).await?;
+        info!("Backup downloaded to {}", downloaded_file.display());
 
-        info!("Backup dump from {} to {}", tmp_path.display(), ext);
+        let backup_file_path: PathBuf = if !is_legacy_file {
+            let encrypted = if filename.ends_with(".tar.gz") {
+                false
+            } else if filename.ends_with(".tar.gz.enc") {
+                true
+            } else {
+                return Ok(RestoreResult {
+                    generated_id,
+                    status: "failed".into(),
+                });
+            };
 
-        let backup_file_path = tmp_path.join(format!("backup_file_tmp.{}", ext));
-        tokio::fs::write(&backup_file_path, &bytes).await?;
-        info!("Backup downloaded to {}", backup_file_path.display());
+            info!("Encrypted: {}", encrypted);
+
+            let mut compressed_archive = downloaded_file.clone();
+
+            if encrypted {
+                let new_name = downloaded_file
+                    .file_name()
+                    .and_then(|n| n.to_str())
+                    .and_then(|n| n.strip_suffix(".enc"))
+                    .ok_or_else(|| anyhow::anyhow!("Invalid encrypted filename"))?;
+
+                let new_compressed_archive = tmp_path.join(new_name);
+
+                decrypt_file_stream_gcm(
+                    downloaded_file,
+                    new_compressed_archive.clone(),
+                    ctx.edge_key.master_key_b64.clone(),
+                )
+                .await
+                .map_err(|e| {
+                    error!("Failed to decrypt file: {}", e);
+                    e
+                })?;
+
+                compressed_archive = new_compressed_archive;
+            }
+
+            let decompressed_files =
+                decompress_large_tar_gz(compressed_archive.as_path(), tmp_path).await?;
+
+            if decompressed_files.is_empty() {
+                return Ok(RestoreResult {
+                    generated_id,
+                    status: "failed".into(),
+                });
+            }
+
+            if decompressed_files.len() == 1 {
+                decompressed_files[0].clone()
+            } else {
+                compressed_archive
+            }
+        } else {
+            downloaded_file.clone()
+        };
 
         let db_instance = DatabaseFactory::create_for_restore(cfg.clone(), &backup_file_path).await;
         let reachable = db_instance.ping().await.unwrap_or(false);
@@ -115,7 +198,7 @@ impl RestoreService {
                 status: "success".into(),
             }),
             Err(e) => {
-                log::error!("Restore failed: {:?}", e);
+                error!("Restore failed: {:?}", e);
                 Ok(RestoreResult {
                     generated_id,
                     status: "failed".into(),
@@ -124,6 +207,7 @@ impl RestoreService {
         }
     }
 
+    // TODO : update with ctx api manager
     pub async fn send_result(&self, result: RestoreResult) {
         info!(
             "[RestoreService] DB: {} | Status: {}",
@@ -147,7 +231,7 @@ impl RestoreService {
                 if status.is_success() {
                     info!("Restoration result sent successfully");
                 } else {
-                    let text = resp.text().await.unwrap_or_default(); // consumes resp
+                    let text = resp.text().await.unwrap_or_default();
                     error!(
                         "Restoration result failed, status: {}, body: {}",
                         status, text

src/services/status.rs

@@ -4,68 +4,11 @@ use crate::core::context::Context;
 use crate::services::config::DatabaseConfig;
 use crate::settings::CONFIG;
 use reqwest::Client;
-use serde::{Deserialize, Serialize};
 use std::error::Error;
 use std::sync::Arc;
-use tracing::{error, info};
+use crate::services::api::endpoints::status::DatabasePayload;
+use crate::services::api::models::agent::status::PingResult;
 
-/// Payload for sending database info in the request
-#[derive(Serialize)]
-struct DatabasePayload<'a> {
-    name: &'a str,
-    dbms: &'a str,
-    #[serde(rename = "generatedId")]
-    generated_id: &'a str,
-}
-
-/// Body for the status API request
-#[derive(Serialize)]
-struct StatusRequestBody<'a> {
-    version: &'a str,
-    databases: Vec<DatabasePayload<'a>>,
-}
-
-/// Typed structs for the response
-#[derive(Debug, Deserialize)]
-pub struct PingResult {
-    pub agent: AgentInfo,
-    pub databases: Vec<DatabaseStatus>,
-}
-
-#[derive(Debug, Deserialize)]
-pub struct AgentInfo {
-    pub id: String,
-    #[serde(rename = "lastContact")]
-    pub last_contact: String,
-}
-
-#[derive(Debug, Deserialize)]
-pub struct DatabaseStatus {
-    pub dbms: String,
-    #[serde(rename = "generatedId")]
-    pub generated_id: String,
-    pub data: DatabaseData,
-}
-
-#[derive(Debug, Deserialize)]
-pub struct DatabaseData {
-    pub backup: BackupInfo,
-    pub restore: RestoreInfo,
-}
-
-#[derive(Debug, Deserialize)]
-pub struct BackupInfo {
-    pub action: bool,
-    pub cron: Option<String>, // can be null
-}
-
-#[derive(Debug, Deserialize)]
-pub struct RestoreInfo {
-    pub action: bool,
-    pub file: String,
-}
-
-/// Service for contacting the agent API
 pub struct StatusService {
     ctx: Arc<Context>,
     client: Client,
@@ -92,27 +35,7 @@ impl StatusService {
             .collect();
 
         let version_str = CONFIG.app_version.as_str();
-
-        let body = StatusRequestBody {
-            version: &version_str,
-            databases: databases_payload,
-        };
-
-        let url = format!(
-            "{}/api/agent/{}/status",
-            edge_key.server_url, edge_key.agent_id
-        );
-        info!("Status request | {}", url);
-
-        let resp = self.client.post(&url).json(&body).send().await?;
-        if !resp.status().is_success() {
-            let msg = format!("Request failed with status: {}", resp.status());
-            error!("{}", msg);
-            return Err(msg.into());
-        }
-
-        let result: PingResult = resp.json().await?;
-
+        let result = self.ctx.api.agent_status(&edge_key.agent_id, &version_str, databases_payload).await?.unwrap();
         Ok(result)
     }
 }

src/services/storage/mod.rs

@@ -0,0 +1,40 @@
+pub mod providers;
+
+use crate::core::context::Context;
+use crate::services::backup::{BackupResult, UploadResult};
+use crate::utils::common::BackupMethod;
+use async_trait::async_trait;
+use providers::local;
+use providers::s3;
+use providers::google_drive;
+use std::sync::Arc;
+use tracing::{error, info};
+use crate::services::api::models::agent::status::DatabaseStorage;
+
+#[async_trait]
+pub trait StorageProvider: Send + Sync {
+    async fn upload(
+        &self,
+        ctx: Arc<Context>,
+        result: BackupResult,
+        method: BackupMethod,
+        config: &DatabaseStorage,
+        encrypt: Option<bool>,
+    ) -> 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());
+
+    match storage.provider.as_str() {
+        "local" => Some(Box::new(local::LocalProvider {})),
+        "s3" => Some(Box::new(s3::S3Provider {})),
+        "google-drive" => Some(Box::new(google_drive::GoogleDriveProvider {})),
+        _ => {
+            error!("Unknown storage provider: {}", storage.provider);
+            None
+        }
+    }
+}

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

@@ -0,0 +1,272 @@
+use anyhow::{Context, Result, anyhow};
+use futures::StreamExt;
+use oauth2::{
+    AuthUrl, ClientId, ClientSecret, RefreshToken, TokenResponse, TokenUrl, basic::BasicClient,
+    reqwest::Client as OAuth2ReqwestClient,
+};
+use reqwest::{Client as ReqwestClient, StatusCode};
+use serde_json::{Value, json};
+use futures::{Stream};
+use bytes::Bytes;
+use reqwest::{Client, header};
+use crate::services::storage::providers::google_drive::models::GoogleDriveProviderConfig;
+
+pub async fn get_google_drive_token(config: &GoogleDriveProviderConfig) -> Result<String> {
+    let http_client = OAuth2ReqwestClient::new();
+
+    let oauth_client = BasicClient::new(ClientId::new(config.client_id.clone()))
+        .set_client_secret(ClientSecret::new(config.client_secret.clone()))
+        .set_auth_uri(
+            AuthUrl::new("https://accounts.google.com/o/oauth2/auth".to_string())
+                .context("invalid auth uri")?,
+        )
+        .set_token_uri(
+            TokenUrl::new("https://oauth2.googleapis.com/token".to_string())
+                .context("invalid token uri")?,
+        );
+
+    let token_result = oauth_client
+        .exchange_refresh_token(&RefreshToken::new(config.refresh_token.clone()))
+        .request_async(&http_client)
+        .await
+        .context("failed to exchange refresh token")?;
+
+    Ok(token_result.access_token().secret().clone())
+}
+
+pub async fn ensure_folder_path(config: &GoogleDriveProviderConfig, path_parts: &[&str]) -> Result<String> {
+    if path_parts.is_empty() {
+        return Ok(config.folder_id.clone());
+    }
+
+    let token = get_google_drive_token(config).await?;
+    let client = ReqwestClient::new();
+    let mut parent_id = config.folder_id.clone();
+
+    for &name in path_parts {
+        let query = format!(
+            "'{parent_id}' in parents and name='{name}' and mimeType='application/vnd.google-apps.folder' and trashed=false"
+        );
+
+        let res = client
+            .get("https://www.googleapis.com/drive/v3/files")
+            .bearer_auth(&token)
+            .query(&[
+                ("q", query),
+                ("fields", "files(id,name)".to_string()),
+                ("supportsAllDrives", "true".to_string()),
+                ("includeItemsFromAllDrives", "true".to_string()),
+                ("corpora", "allDrives".to_string()),
+            ])
+            .send()
+            .await
+            .context("list folders failed")?
+            .json::<Value>()
+            .await?;
+
+        if let Some(files) = res["files"].as_array() {
+            if let Some(folder) = files.first() {
+                if let Some(id) = folder["id"].as_str() {
+                    parent_id = id.to_string();
+                    continue;
+                }
+            }
+        }
+
+        let create_payload = json!({
+            "name": name,
+            "mimeType": "application/vnd.google-apps.folder",
+            "parents": [parent_id],
+            "supportsAllDrives": true,
+        });
+
+        let folder = client
+            .post("https://www.googleapis.com/drive/v3/files")
+            .bearer_auth(&token)
+            .json(&create_payload)
+            .send()
+            .await
+            .context("create folder failed")?
+            .json::<Value>()
+            .await?;
+
+        parent_id = folder["id"]
+            .as_str()
+            .ok_or_else(|| anyhow!("No id returned after folder creation"))?
+            .to_string();
+    }
+
+    Ok(parent_id)
+}
+
+pub async fn find_file_by_name(
+    config: &GoogleDriveProviderConfig,
+    file_name: &str,
+    folder_id: &str,
+) -> Result<Option<String>> {
+    let token = get_google_drive_token(config).await?;
+    let client = ReqwestClient::new();
+
+    let query = format!("'{folder_id}' in parents and name='{file_name}' and trashed=false");
+
+    let res = client
+        .get("https://www.googleapis.com/drive/v3/files")
+        .bearer_auth(&token)
+        .query(&[
+            ("q", query),
+            ("fields", "files(id,name)".to_string()),
+            ("supportsAllDrives", "true".to_string()),
+            ("includeItemsFromAllDrives", "true".to_string()),
+            ("corpora", "allDrives".to_string()),
+        ])
+        .send()
+        .await?
+        .json::<Value>()
+        .await?;
+
+    if let Some(files) = res["files"].as_array() {
+        if let Some(file) = files.first() {
+            if let Some(id) = file["id"].as_str() {
+                return Ok(Some(id.to_string()));
+            }
+        }
+    }
+
+    Ok(None)
+}
+
+pub async fn upload_stream_to_google_drive(
+    config: &GoogleDriveProviderConfig,
+    full_path: &str,
+    mut content_stream: impl Stream<Item = Result<Bytes, std::io::Error>> + Send + Unpin + 'static,
+    total_size: u64,
+    mime_type: Option<&str>,
+) -> Result<()> {
+    let path_parts: Vec<&str> = full_path.split('/').filter(|s| !s.is_empty()).collect();
+    if path_parts.is_empty() {
+        return Err(anyhow::anyhow!("Invalid path: empty"));
+    }
+
+    let file_name = *path_parts.last().unwrap();
+    let folder_path = &path_parts[..path_parts.len() - 1];
+
+    let folder_id = ensure_folder_path(config, folder_path).await?;
+
+    if find_file_by_name(config, file_name, &folder_id).await?.is_some() {
+        return Err(anyhow::anyhow!("File already exists: {}", full_path));
+    }
+
+    let token = get_google_drive_token(config).await?;
+    let client = Client::new();
+
+    let mime = mime_type.unwrap_or("application/octet-stream");
+
+    let metadata = json!({
+        "name": file_name,
+        "parents": [folder_id],
+        "mimeType": mime,
+        "supportsAllDrives": true,
+    });
+
+    let session_res = client
+        .post("https://www.googleapis.com/upload/drive/v3/files?uploadType=resumable")
+        .bearer_auth(&token)
+        .header("X-Upload-Content-Type", mime)
+        .header("X-Upload-Content-Length", total_size.to_string())  // Helps a lot
+        .json(&metadata)
+        .send()
+        .await
+        .context("Failed to initiate resumable upload")?;
+
+    if session_res.status() != StatusCode::OK {
+        let text = session_res.text().await.unwrap_or_default();
+        return Err(anyhow::anyhow!("Initiate failed: {}", text));
+    }
+
+    let upload_url = session_res
+        .headers()
+        .get(header::LOCATION)
+        .ok_or_else(|| anyhow::anyhow!("No Location header"))?
+        .to_str()?
+        .to_string();
+
+    const CHUNK_SIZE: u64 = 8 * 1024 * 1024;
+
+    let mut uploaded: u64 = 0;
+
+    while uploaded < total_size {
+        let chunk_size = (total_size - uploaded).min(CHUNK_SIZE);
+
+        let mut chunk_bytes = Vec::with_capacity(chunk_size as usize);
+        let mut remaining = chunk_size;
+
+        while remaining > 0 {
+            match content_stream.next().await {
+                Some(Ok(bytes)) => {
+                    let to_take = remaining.min(bytes.len() as u64) as usize;
+                    chunk_bytes.extend_from_slice(&bytes[..to_take]);
+                    remaining -= to_take as u64;
+
+                    if to_take < bytes.len() {
+                        // TODO : Put remainder back
+                    }
+                }
+                Some(Err(e)) => return Err(e).context("Stream error during chunk"),
+                None => {
+                    if uploaded + chunk_bytes.len() as u64 != total_size {
+                        return Err(anyhow::anyhow!("Stream ended early"));
+                    }
+                }
+            }
+        }
+
+        if chunk_bytes.is_empty() && uploaded < total_size {
+            return Err(anyhow::anyhow!("Unexpected end of stream"));
+        }
+
+        let range_end = uploaded + chunk_bytes.len() as u64 - 1;
+        let content_range = if uploaded + chunk_bytes.len() as u64 == total_size {
+            format!("bytes {}-{}/{}", uploaded, range_end, total_size)
+        } else {
+            format!("bytes {}-{}/*", uploaded, range_end)
+        };
+
+        let mut retries = 0;
+        loop {
+            let res = client
+                .put(&upload_url)
+                .header("Content-Range", &content_range)
+                .header("Content-Length", chunk_bytes.len().to_string())
+                .body(chunk_bytes.clone())  // clone is cheap if small; optimize later if needed
+                .send()
+                .await;
+
+            match res {
+                Ok(resp) if resp.status().is_success() || resp.status() == StatusCode::PERMANENT_REDIRECT => {
+                    // 200 or 308 = good
+                    uploaded += chunk_bytes.len() as u64;
+                    tracing::info!("Uploaded {}/{} bytes", uploaded, total_size);
+                    break;
+                }
+                Ok(resp) if resp.status() == StatusCode::TOO_MANY_REQUESTS => {
+                    // Backoff on 429
+                    tokio::time::sleep(std::time::Duration::from_secs(5 * (1 << retries))).await;
+                }
+                Ok(resp) => {
+                    let text = resp.text().await.unwrap_or_default();
+                    return Err(anyhow::anyhow!("Chunk upload failed: {}", text));
+                }
+                Err(e) if e.is_timeout() || e.is_connect() => {
+                    if retries > 5 {
+                        return Err(e).context("Too many retries");
+                    }
+                    retries += 1;
+                    tokio::time::sleep(std::time::Duration::from_secs(2u64.pow(retries))).await;
+                }
+                Err(e) => return Err(e).context("Chunk request failed"),
+            }
+        }
+    }
+
+    Ok(())
+}

src/services/storage/providers/google_drive/mod.rs

@@ -0,0 +1,129 @@
+mod helpers;
+mod models; 
+
+use crate::core::context::Context;
+use crate::services::api::models::agent::status::DatabaseStorage;
+use crate::services::backup::{BackupResult, UploadResult};
+use crate::services::storage::StorageProvider;
+use crate::utils::common::BackupMethod;
+use crate::utils::file::{full_file_name, full_file_path};
+use crate::utils::stream::build_stream;
+use async_trait::async_trait;
+use std::sync::Arc;
+use tokio::fs;
+use tracing::{error, info};
+use crate::services::storage::providers::google_drive::helpers::{upload_stream_to_google_drive};
+use crate::services::storage::providers::google_drive::models::GoogleDriveProviderConfig;
+
+pub struct GoogleDriveProvider {}
+
+#[async_trait]
+impl StorageProvider for GoogleDriveProvider {
+    async fn upload(
+        &self,
+        ctx: Arc<Context>,
+        result: BackupResult,
+        _method: BackupMethod,
+        storage: &DatabaseStorage,
+        encrypt: Option<bool>,
+    ) -> UploadResult {
+        let Some(file_path) = result.backup_file else {
+            return UploadResult {
+                storage_id: storage.id.clone(),
+                success: false,
+                error: Some("Missing backup file path".to_string()),
+                remote_file_path: None,
+                total_size: None,
+            };
+        };
+
+        let total_size = match fs::metadata(&file_path).await {
+            Ok(meta) => meta.len(),
+            Err(e) => {
+                error!("Failed to get file size: {}", e);
+                return UploadResult {
+                    storage_id: storage.id.clone(),
+                    success: false,
+                    error: Some(e.to_string()),
+                    remote_file_path: None,
+                    total_size: None,
+                };
+            }
+        };
+
+        let encrypt = encrypt.unwrap_or(false);
+
+        let upload = match build_stream(
+            &file_path,
+            encrypt,
+            &ctx.edge_key.master_key_b64
+        )
+            .await
+        {
+            Ok(u) => u,
+            Err(e) => {
+                error!("Stream build failed: {}", e);
+                return UploadResult {
+                    storage_id: storage.id.clone(),
+                    success: false,
+                    error: Some(e.to_string()),
+                    remote_file_path: None,
+                    total_size: None,
+                };
+            }
+        };
+
+        let config: GoogleDriveProviderConfig = match storage.clone().config.try_into() {
+            Ok(c) => c,
+            Err(e) => {
+                return UploadResult {
+                    storage_id: storage.id.clone(),
+                    success: false,
+                    error: Some(e.to_string()),
+                    remote_file_path: None,
+                    total_size: None,
+                };
+            }
+        };
+
+
+        let file_name = full_file_name(encrypt);
+
+        info!("Uploading file {}", file_name);
+
+        let remote_file_path = full_file_path(&file_name);
+
+        match upload_stream_to_google_drive(
+            &config,
+            &remote_file_path,
+            upload.stream,   
+            total_size,
+            Some("application/octet-stream"),
+        ).await {
+            Ok(_file_id) => {
+                
+                info!("Google Drive upload successful");
+
+                UploadResult {
+                    storage_id: storage.id.clone(),
+                    success: true,
+                    error: None,
+                    remote_file_path: Some(remote_file_path),
+                    total_size: Some(total_size),
+                }
+            }
+            Err(e) => {
+                error!("Google Drive upload failed: {:?}", e);
+
+                UploadResult {
+                    storage_id: storage.id.clone(),
+                    success: false,
+                    error: Some(e.to_string()),
+                    remote_file_path: None,
+                    total_size: Some(total_size),
+                }
+            }
+        }
+
+    }
+}

src/services/storage/providers/google_drive/models.rs

@@ -0,0 +1,11 @@
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Deserialize, Serialize)]
+pub struct GoogleDriveProviderConfig {
+    pub client_id: String,
+    pub client_secret: String,
+    pub refresh_token: String,
+    pub folder_id: String,
+}
+
+

src/services/storage/providers/local.rs

@@ -0,0 +1,116 @@
+use crate::core::context::Context;
+use crate::services::api::models::agent::status::DatabaseStorage;
+use crate::services::backup::{BackupResult, UploadResult};
+use crate::services::storage::StorageProvider;
+use crate::utils::common::BackupMethod;
+use crate::utils::file::{full_file_name, full_file_path};
+use crate::utils::stream::build_stream;
+use crate::utils::tus::upload_to_tus_stream_with_headers;
+use async_trait::async_trait;
+use reqwest::header::{HeaderMap, HeaderValue};
+use std::sync::Arc;
+use tokio::fs;
+use tracing::error;
+
+pub struct LocalProvider;
+
+#[async_trait]
+impl StorageProvider for LocalProvider {
+    async fn upload(
+        &self,
+        ctx: Arc<Context>,
+        result: BackupResult,
+        method: BackupMethod,
+        storage: &DatabaseStorage,
+        encrypt: Option<bool>,
+    ) -> UploadResult {
+        let Some(file_path) = result.backup_file else {
+            return UploadResult {
+                storage_id: storage.id.clone(),
+                success: false,
+                error: Some("File path error".to_string()),
+                remote_file_path: None,
+                total_size: None,
+            };
+        };
+
+        let encrypt = encrypt.unwrap_or(false);
+
+        let file_name = full_file_name(encrypt);
+        let remote_file_path = full_file_path(&file_name);
+
+        let total_size = match fs::metadata(&file_path).await {
+            Ok(meta) => meta.len(),
+            Err(e) => {
+                error!("Failed to get file size: {}", e);
+                return UploadResult {
+                    storage_id: storage.id.clone(),
+                    success: false,
+                    error: Some(e.to_string()),
+                    remote_file_path: None,
+                    total_size: None,
+                };
+            }
+        };
+
+        let upload = match build_stream(
+            &file_path,
+            encrypt,
+            &ctx.edge_key.master_key_b64
+        )
+        .await
+        {
+            Ok(u) => u,
+            Err(e) => {
+                error!("Stream build failed: {}", e);
+                return UploadResult {
+                    storage_id: storage.id.clone(),
+                    success: false,
+                    error: Some(e.to_string()),
+                    remote_file_path: None,
+                    total_size: None
+                };
+            }
+        };
+
+        let mut extra_headers = HeaderMap::new();
+
+        extra_headers.insert("X-File-Name", HeaderValue::from_str(&file_name).unwrap());
+        extra_headers.insert("X-File-Size", HeaderValue::from_str(&total_size.to_string()).unwrap());
+        extra_headers.insert(
+            "X-File-Path",
+            HeaderValue::from_str(&remote_file_path).unwrap(),
+        );
+        extra_headers.insert(
+            "X-Generated-Id",
+            HeaderValue::from_str(&result.generated_id).unwrap(),
+        );
+        extra_headers.insert("X-Status", HeaderValue::from_str(&result.status).unwrap());
+        extra_headers.insert(
+            "X-Method",
+            HeaderValue::from_str(&method.to_string()).unwrap(),
+        );
+        
+        let tus_endpoint = format!("{}/tus/files", ctx.edge_key.server_url);
+
+        match upload_to_tus_stream_with_headers(upload.stream, &tus_endpoint, extra_headers, total_size).await {
+            Ok(_) => UploadResult {
+                storage_id: storage.id.clone(),
+                success: true,
+                error: None,
+                remote_file_path: Some(remote_file_path),
+                total_size: Some(total_size),
+            },
+            Err(e) => {
+                error!("Local upload failed: {}", e);
+                UploadResult {
+                    storage_id: storage.id.clone(),
+                    success: false,
+                    error: Some(e.to_string()),
+                    remote_file_path: None,
+                    total_size: None,
+                }
+            }
+        }
+    }
+}

src/services/storage/providers/mod.rs

@@ -0,0 +1,3 @@
+pub mod local;
+pub mod s3;
+pub mod google_drive;

src/services/storage/providers/s3/mod.rs

@@ -0,0 +1,342 @@
+mod models;
+
+use crate::core::context::Context;
+use crate::services::api::models::agent::status::DatabaseStorage;
+use crate::services::backup::{BackupResult, UploadResult};
+use crate::services::storage::StorageProvider;
+use crate::services::storage::providers::s3::models::S3ProviderConfig;
+use crate::utils::common::BackupMethod;
+use crate::utils::file::{full_file_name, full_file_path};
+use crate::utils::stream::build_stream;
+use async_trait::async_trait;
+use aws_sdk_s3 as s3;
+use aws_sdk_s3::config::BehaviorVersion;
+use aws_sdk_s3::config::Region;
+use aws_sdk_s3::primitives::ByteStream;
+use aws_sdk_s3::types::{CompletedMultipartUpload, CompletedPart};
+use futures::StreamExt;
+use std::pin::Pin;
+use std::sync::Arc;
+use tokio::fs;
+use tracing::{error, info};
+
+pub struct S3Provider {}
+
+#[async_trait]
+impl StorageProvider for S3Provider {
+    async fn upload(
+        &self,
+        ctx: Arc<Context>,
+        result: BackupResult,
+        _method: BackupMethod,
+        storage: &DatabaseStorage,
+        encrypt: Option<bool>,
+    ) -> UploadResult {
+        let Some(file_path) = result.backup_file else {
+            return UploadResult {
+                storage_id: storage.id.clone(),
+                success: false,
+                error: Some("Missing backup file path".to_string()),
+                remote_file_path: None,
+                total_size: None,
+            };
+        };
+
+        let total_size = match fs::metadata(&file_path).await {
+            Ok(meta) => meta.len(),
+            Err(e) => {
+                error!("Failed to get file size: {}", e);
+                return UploadResult {
+                    storage_id: storage.id.clone(),
+                    success: false,
+                    error: Some(e.to_string()),
+                    remote_file_path: None,
+                    total_size: None,
+                };
+            }
+        };
+
+        let encrypt = encrypt.unwrap_or(false);
+
+        let upload = match build_stream(
+            &file_path,
+            encrypt,
+            &ctx.edge_key.master_key_b64
+            // encrypt.then(|| ctx.edge_key.public_key.as_bytes().to_vec()),
+        )
+        .await
+        {
+            Ok(u) => u,
+            Err(e) => {
+                error!("Stream build failed: {}", e);
+                return UploadResult {
+                    storage_id: storage.id.clone(),
+                    success: false,
+                    error: Some(e.to_string()),
+                    remote_file_path: None,
+                    total_size: None,
+                };
+            }
+        };
+
+        let config: S3ProviderConfig = match storage.clone().config.try_into() {
+            Ok(c) => c,
+            Err(e) => {
+                return UploadResult {
+                    storage_id: storage.id.clone(),
+                    success: false,
+                    error: Some(e.to_string()),
+                    remote_file_path: None,
+                    total_size: None,
+                };
+            }
+        };
+
+        let credentials = s3::config::Credentials::new(
+            config.access_key.clone(),
+            config.secret_key.clone(),
+            None,
+            None,
+            "static-creds",
+        );
+
+        let region = Region::new(config.region.clone().unwrap_or("eu-central-3".to_string()));
+
+        let sdk_config = s3::config::Builder::new()
+            .credentials_provider(credentials)
+            .region(region)
+            .force_path_style(true)
+            .endpoint_url(format!(
+                "{}://{}",
+                if config.ssl { "https" } else { "http" },
+                config.end_point_url
+            ))
+            .behavior_version(BehaviorVersion::latest())
+            .build();
+
+        let client = s3::Client::from_conf(sdk_config);
+
+        const PART_SIZE: usize = 100 * 1024 * 1024; // 100 MiB
+
+        let file_name = full_file_name(encrypt);
+
+        info!("Uploading file {}", file_name);
+
+        let bucket = &config.bucket_name;
+        let remote_file_path = full_file_path(&file_name);
+        info!("S3 key {:}", remote_file_path);
+        info!(
+            "Starting multipart upload to s3://{}/{}",
+            bucket, remote_file_path
+        );
+
+        let create_resp = match client
+            .create_multipart_upload()
+            .bucket(bucket)
+            .key(&remote_file_path)
+            .send()
+            .await
+        {
+            Ok(r) => r,
+            Err(e) => {
+                error!("Failed to create multipart upload: {}", e);
+                return UploadResult {
+                    storage_id: storage.id.clone(),
+                    success: false,
+                    error: Some(e.to_string()),
+                    remote_file_path: None,
+                    total_size: None,
+                };
+            }
+        };
+
+        let upload_id = match create_resp.upload_id {
+            Some(id) => id,
+            None => {
+                return UploadResult {
+                    storage_id: storage.id.clone(),
+                    success: false,
+                    error: Some("No upload ID returned".to_string()),
+                    remote_file_path: None,
+                    total_size: None,
+                };
+            }
+        };
+
+        let mut parts: Vec<CompletedPart> = Vec::new();
+        let mut part_number: i32 = 1;
+        let mut buffer: Vec<u8> = Vec::with_capacity(PART_SIZE);
+
+        let mut peekable = upload.stream.peekable();
+
+        while let Some(item) = peekable.next().await {
+            let bytes = match item {
+                Ok(b) => b,
+                Err(e) => {
+                    error!("Stream error during upload: {}", e);
+                    let _ = client
+                        .abort_multipart_upload()
+                        .bucket(bucket)
+                        .key(&remote_file_path)
+                        .upload_id(&upload_id)
+                        .send()
+                        .await;
+                    return UploadResult {
+                        storage_id: storage.id.clone(),
+                        success: false,
+                        error: Some(format!("Stream error: {}", e)),
+                        remote_file_path: None,
+                        total_size: None,
+                    };
+                }
+            };
+
+            buffer.extend_from_slice(&bytes);
+
+            let is_last = {
+                let pinned = Pin::new(&mut peekable);
+                let peek_future = pinned.peek();
+                peek_future.await.is_none()
+            };
+
+            let should_upload = buffer.len() >= PART_SIZE || is_last;
+
+            if should_upload && !buffer.is_empty() {
+                let body = ByteStream::from(buffer.clone());
+
+                match client
+                    .upload_part()
+                    .bucket(bucket)
+                    .key(&remote_file_path)
+                    .upload_id(&upload_id)
+                    .part_number(part_number)
+                    .body(body)
+                    .send()
+                    .await
+                {
+                    Ok(resp) => {
+                        if let Some(etag) = resp.e_tag {
+                            parts.push(
+                                CompletedPart::builder()
+                                    .part_number(part_number)
+                                    .e_tag(etag)
+                                    .build(),
+                            );
+                            info!("Uploaded part {} ({} bytes)", part_number, buffer.len());
+                        }
+                    }
+                    Err(e) => {
+                        error!("Failed to upload part {}: {}", part_number, e);
+                        let _ = client
+                            .abort_multipart_upload()
+                            .bucket(bucket)
+                            .key(&remote_file_path)
+                            .upload_id(&upload_id)
+                            .send()
+                            .await;
+                        return UploadResult {
+                            storage_id: storage.id.clone(),
+                            success: false,
+                            error: Some(e.to_string()),
+                            remote_file_path: None,
+                            total_size: None,
+                        };
+                    }
+                }
+                buffer.clear();
+                part_number += 1;
+            }
+        }
+
+        if !buffer.is_empty() {
+            let _ = client
+                .abort_multipart_upload()
+                .bucket(bucket)
+                .key(&remote_file_path)
+                .upload_id(&upload_id)
+                .send()
+                .await;
+            return UploadResult {
+                storage_id: storage.id.clone(),
+                success: false,
+                error: Some("No parts were uploaded".to_string()),
+                remote_file_path: None,
+                total_size: None,
+            };
+        }
+
+        let completed = CompletedMultipartUpload::builder()
+            .set_parts(Some(parts))
+            .build();
+
+        match client
+            .complete_multipart_upload()
+            .bucket(bucket)
+            .key(&remote_file_path)
+            .upload_id(&upload_id)
+            .multipart_upload(completed)
+            .send()
+            .await
+        {
+            Ok(_) => {
+                info!(
+                    "Successfully completed multipart upload: {}",
+                    remote_file_path
+                );
+
+                // if let Some(enc) = upload.encryption {
+                //     let meta = EncryptionMetadataFile {
+                //         version: 1,
+                //         cipher: "AES-256-CBC+RSA-OAEP-SHA256".to_string(),
+                //         encrypted_aes_key_b64: general_purpose::STANDARD
+                //             .encode(enc.encrypted_aes_key),
+                //         iv_b64: general_purpose::STANDARD.encode(enc.iv),
+                //     };
+                // 
+                //     let meta_toml = toml::to_string(&meta).expect("Serialization error");
+                // 
+                //     let meta_key = format!("{}.meta", remote_file_path);
+                // 
+                //     client
+                //         .put_object()
+                //         .bucket(bucket)
+                //         .key(&meta_key)
+                //         .body(ByteStream::from(meta_toml.into_bytes()))
+                //         .content_type("application/toml")
+                //         .send()
+                //         .await
+                //         .map_err(|e| {
+                //             error!("Metadata upload failed: {}", e);
+                //             e
+                //         })
+                //         .unwrap();
+                // }
+
+                UploadResult {
+                    storage_id: storage.id.clone(),
+                    success: true,
+                    error: None,
+                    remote_file_path: Some(remote_file_path),
+                    total_size: Some(total_size),
+                }
+            }
+            Err(e) => {
+                error!("Failed to complete multipart upload: {}", e);
+                let _ = client
+                    .abort_multipart_upload()
+                    .bucket(bucket)
+                    .key(&remote_file_path)
+                    .upload_id(&upload_id)
+                    .send()
+                    .await;
+                UploadResult {
+                    storage_id: storage.id.clone(),
+                    success: false,
+                    error: Some(e.to_string()),
+                    remote_file_path: None,
+                    total_size: None,
+                }
+            }
+        }
+    }
+}

src/services/storage/providers/s3/models.rs

@@ -0,0 +1,11 @@
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Deserialize, Serialize)]
+pub struct S3ProviderConfig {
+    pub access_key: String,
+    pub secret_key: String,
+    pub bucket_name: String,
+    pub end_point_url: String,
+    pub ssl: bool,
+    pub region: Option<String>,
+}

src/tasks/ping.rs

@@ -19,33 +19,6 @@ pub async fn ping_server() {
                 e
             );
         }
-
         tokio::time::sleep(Duration::from_secs(CONFIG.pooling as u64)).await;
     }
-}
-// use crate::core::agent::Agent;
-// use crate::core::context::Context;
-// use crate::utils::common::BackupMethod;
-// use std::sync::Arc;
-// use tokio::sync::Mutex;
-// use tokio::time::{sleep, Duration};
-// use tracing::{error};
-//
-// pub async fn ping_server() {
-//
-//
-//     loop {
-//         let ctx = Arc::new(Context::new());
-//         let agent = Arc::new(Mutex::new(Agent::new(ctx.clone()).await));
-//         let agent_clone = agent.clone();
-//
-//         tokio::spawn(async move {
-//             let mut agent_locked = agent_clone.lock().await;
-//             if let Err(e) = agent_locked.run(BackupMethod::Manual).await {
-//                 error!("An error occurred while executing ping_server: {:?}", e);
-//             }
-//         });
-//
-//         sleep(Duration::from_secs(5)).await;
-//     }
-// }
+}

src/utils/common.rs

@@ -1,3 +1,6 @@
+use serde::Serialize;
+use serde_json::Value;
+
 #[derive(Clone, Copy)]
 pub enum BackupMethod {
     Automatic,
@@ -12,3 +15,12 @@ impl ToString for BackupMethod {
         }
     }
 }
+
+
+pub fn vec_to_option_json<T: Serialize>(v: Vec<T>) -> Option<Value> {
+    if v.is_empty() {
+        None
+    } else {
+        Some(serde_json::to_value(v).expect("serialization failed"))
+    }
+}

src/utils/compress.rs

@@ -0,0 +1,97 @@
+use anyhow::Result;
+use async_compression::tokio::bufread::GzipDecoder;
+use async_compression::tokio::write::GzipEncoder as AsyncGzipEncoder;
+use futures::StreamExt;
+use std::path::{Path, PathBuf};
+use tokio::fs::File;
+use tokio::fs::{create_dir_all};
+use tokio::io::AsyncWriteExt;
+use tokio::io::BufReader;
+use tokio_tar::Archive;
+use tokio_tar::Builder as TokioTarBuilder;
+use tracing::info;
+
+#[allow(dead_code)]
+pub struct CompressionResult {
+    pub compressed_path: PathBuf,
+}
+
+pub async fn compress_to_tar_gz_large(file: &PathBuf) -> Result<CompressionResult> {
+    if file
+        .file_name()
+        .and_then(|n| n.to_str())
+        .map(|n| n.ends_with(".tar.gz"))
+        .unwrap_or(false)
+    {
+        info!("File {:?} is already a tar.gz, skipping compression", file);
+        return Ok(CompressionResult {
+            compressed_path: file.clone(),
+        });
+    }
+
+    let tar_gz_path = file.with_extension("").with_extension("tar.gz");
+
+    let output_file = File::create(&tar_gz_path).await?;
+    let gzip_writer = AsyncGzipEncoder::new(output_file);
+    let mut tar_builder = TokioTarBuilder::new(gzip_writer);
+
+    let file_name = file
+        .file_name()
+        .ok_or_else(|| anyhow::anyhow!("Cannot get file name for {:?}", file))?;
+
+    tar_builder
+        .append_path_with_name(file, file_name)
+        .await
+        .map_err(|e| anyhow::anyhow!("Failed to append path: {}", e))?;
+
+    tar_builder
+        .finish()
+        .await
+        .map_err(|e| anyhow::anyhow!("Failed to finish tar: {}", e))?;
+
+    let mut gzip = tar_builder
+        .into_inner()
+        .await
+        .map_err(|e| anyhow::anyhow!("Failed to extract gzip encoder: {}", e))?;
+
+    gzip.shutdown()
+        .await
+        .map_err(|e| anyhow::anyhow!("Gzip shutdown failed: {}", e))?;
+
+    info!("Compressing {:?} to {:?}", &file, &tar_gz_path);
+
+    Ok(CompressionResult {
+        compressed_path: tar_gz_path,
+    })
+}
+
+pub async fn decompress_large_tar_gz(
+    tar_gz_path: &Path,
+    output_dir: &Path,
+) -> Result<Vec<PathBuf>> {
+    let file = File::open(tar_gz_path).await?;
+    let buf_reader = BufReader::with_capacity(8 * 1024 * 1024, file);
+    let decoder = GzipDecoder::new(buf_reader);
+    let mut archive = Archive::new(decoder);
+
+    let mut extracted_files = Vec::new();
+    let mut entries = archive.entries()?;
+
+    while let Some(entry) = entries.next().await {
+        let mut entry = entry?;
+        let path = entry.path()?.to_path_buf();
+        let full_path = output_dir.join(&path);
+
+        if let Some(parent) = full_path.parent() {
+            create_dir_all(parent).await?;
+        }
+
+        entry.unpack(&full_path).await?;
+        extracted_files.push(full_path);
+    }
+
+    // remove_file(tar_gz_path).await?;
+    info!("Decompressed {:?} into {:?}", tar_gz_path, output_dir);
+
+    Ok(extracted_files)
+}

src/utils/deserializer.rs

@@ -0,0 +1,39 @@
+use serde::{Deserialize, Deserializer};
+use toml::Value;
+
+pub fn deserialize_snake_case<'de, D>(deserializer: D) -> Result<Value, D::Error>
+where
+    D: Deserializer<'de>,
+{
+    let value = Value::deserialize(deserializer)?;
+    Ok(to_snake_case(value))
+}
+fn to_snake_case(value: Value) -> Value {
+    match value {
+        Value::Table(table) => Value::Table(
+            table
+                .into_iter()
+                .map(|(k, v)| (camel_to_snake(&k), to_snake_case(v)))
+                .collect(),
+        ),
+        Value::Array(arr) => {
+            Value::Array(arr.into_iter().map(to_snake_case).collect())
+        }
+        other => other,
+    }
+}
+
+fn camel_to_snake(s: &str) -> String {
+    let mut out = String::new();
+    for (i, c) in s.chars().enumerate() {
+        if c.is_uppercase() {
+            if i > 0 {
+                out.push('_');
+            }
+            out.push(c.to_ascii_lowercase());
+        } else {
+            out.push(c);
+        }
+    }
+    out
+}

src/utils/edge_key.rs

@@ -2,7 +2,6 @@ use base64::{Engine as _, engine::general_purpose};
 use serde::Deserialize;
 use serde_json::Value;
 use thiserror::Error;
-use tracing::error;
 
 #[derive(Debug, Deserialize)]
 #[allow(dead_code)]
@@ -11,8 +10,8 @@ pub struct EdgeKey {
     pub server_url: String,
     #[serde(rename = "agentId")]
     pub agent_id: String,
-    #[serde(rename = "publicKey")]
-    pub public_key: String,
+    #[serde(rename = "masterKeyB64")]
+    pub master_key_b64: String,
 }
 
 #[derive(Debug, Error)]
@@ -35,14 +34,14 @@ pub fn decode_edge_key(edge_key: &str) -> Result<EdgeKey, EdgeKeyError> {
     let decoded_str = String::from_utf8_lossy(&decoded_bytes);
 
     let parsed: Value = serde_json::from_str(&decoded_str)?;
+    
     if parsed.get("serverUrl").is_some()
         && parsed.get("agentId").is_some()
-        && parsed.get("publicKey").is_some()
+        && parsed.get("masterKeyB64").is_some()
     {
         let edge_key: EdgeKey = serde_json::from_value(parsed)?;
         Ok(edge_key)
     } else {
-        error!("EDGE_KEY INVALID");
         Err(EdgeKeyError::InvalidKey)
     }
 }

src/utils/file.rs

@@ -1,17 +1,33 @@
-use std::path::{Path, PathBuf};
-use anyhow::Result;
-use tokio::fs::File;
-use tokio::io::{AsyncReadExt, BufReader};
-use futures::{Stream};
-use bytes::Bytes;
-use openssl::symm::{Cipher, Crypter, Mode};
-use openssl::pkey::PKey;
-use openssl::encrypt::Encrypter;
-use openssl::rsa::Padding;
-use openssl::hash::MessageDigest;
-use async_stream::try_stream;
+#![allow(dead_code)]
 
-use hex;
+use anyhow::Result;
+use chrono::Utc;
+use serde::{Deserialize, Serialize};
+use std::path::{Path, PathBuf};
+use tokio::fs::File;
+use tokio::io::{AsyncBufReadExt, AsyncReadExt, BufReader};
+use uuid::Uuid;
+
+use aes_gcm::aead::{Aead, KeyInit};
+use aes_gcm::{Aes256Gcm, Key, Nonce};
+use base64::Engine;
+use base64::engine::general_purpose;
+use bytes::Bytes;
+use futures::Stream;
+use rand::rngs::OsRng;
+use rand::TryRngCore;
+use tokio::io::{AsyncWriteExt, BufWriter};
+use tokio::sync::mpsc;
+use tokio_stream::wrappers::ReceiverStream;
+use tracing::info;
+
+#[derive(Serialize, Deserialize)]
+pub struct EncryptionMetadataFile {
+    pub version: u8,
+    pub cipher: String,
+    pub encrypted_aes_key_b64: String,
+    pub iv_b64: String,
+}
 
 pub fn full_extension(path: &Path) -> String {
     path.file_name()
@@ -21,57 +37,139 @@ pub fn full_extension(path: &Path) -> String {
         .to_string()
 }
 
-pub async fn encrypt_file_stream(
+pub fn full_file_name(encrypt: bool) -> String {
+    let uuid = Uuid::new_v4();
+    let base_name = format!("{}.{}", uuid, "tar.gz");
+    if encrypt {
+        format!("{}.enc", base_name)
+    } else {
+        base_name.to_string()
+    }
+}
+
+pub fn full_file_path(file_name: &String) -> String {
+    format!("backups/{}/{}", Utc::now().format("%Y-%m-%d"), file_name)
+}
+
+const CHUNK_SIZE: usize = 16 * 1024 * 1024;
+
+#[derive(Serialize, Deserialize, Debug)]
+struct FileHeader {
+    version: u8,
+    cipher: String,
+    chunk_size: usize,
+    base_nonce: Vec<u8>,
+}
+
+pub async fn encrypt_file_stream_gcm(
     file_path: PathBuf,
-    aes_key: [u8; 32],
-    iv: [u8; 16],
-    pub_key_pem: Vec<u8>,
-) -> Result<(impl Stream<Item = Result<Bytes>> + Send + 'static, String)> {
-    // ---------- Encrypt AES key with RSA ----------
-    let pkey = PKey::public_key_from_pem(&pub_key_pem)?;
-    let mut rsa = Encrypter::new(&pkey)?;
-    rsa.set_rsa_padding(Padding::PKCS1_OAEP)?;
-    rsa.set_rsa_oaep_md(MessageDigest::sha256())?;
-    rsa.set_rsa_mgf1_md(MessageDigest::sha256())?;
+    master_key_b64: String,
+) -> Result<impl Stream<Item = Result<Bytes>> + Send + 'static> {
+    let master_key_bytes = general_purpose::STANDARD
+        .decode(master_key_b64)
+        .map_err(|_| anyhow::anyhow!("Invalid base64"))?;
 
-    let mut encrypted_key = vec![0u8; rsa.encrypt_len(&aes_key)?];
-    let len = rsa.encrypt(&aes_key, &mut encrypted_key)?;
-    encrypted_key.truncate(len);
+    let (tx, rx) = mpsc::channel(8);
 
-    let encrypted_key_hex = hex::encode(encrypted_key);
+    tokio::spawn(async move {
+        let mut rng = OsRng;
+        let mut base_nonce = [0u8; 8];
+        rng.try_fill_bytes(&mut base_nonce).unwrap();
 
-    // ---------- Streaming AES encryption ----------
-    let stream = try_stream! {
-        let file = File::open(&file_path).await?;
+        let key = Key::<Aes256Gcm>::try_from(master_key_bytes.as_slice())
+            .map_err(|_| anyhow::anyhow!("Invalid AES-256 key length")).unwrap();
+
+        let cipher = Aes256Gcm::new(&key);
+
+        let header = FileHeader {
+            version: 1,
+            cipher: "AES-256-GCM".to_string(),
+            chunk_size: CHUNK_SIZE,
+            base_nonce: base_nonce.to_vec(),
+        };
+        let header_json = serde_json::to_string(&header).unwrap();
+        tx.send(Ok(Bytes::from(header_json + "\n"))).await.unwrap();
+
+        let file = File::open(&file_path).await.unwrap();
         let mut reader = BufReader::new(file);
-
-        let cipher = Cipher::aes_256_cbc();
-        let mut crypter = Crypter::new(cipher, Mode::Encrypt, &aes_key, Some(&iv))?;
-        crypter.pad(true);
-
-        let mut buffer = vec![0u8; 1024 * 1024]; // 1 MB chunks
+        let mut buffer = vec![0u8; CHUNK_SIZE];
+        let mut chunk_index: u32 = 0;
 
         loop {
-            let n = reader.read(&mut buffer).await?;
+            let n = reader.read(&mut buffer).await.unwrap();
             if n == 0 {
                 break;
             }
 
-            let mut out = vec![0u8; n + cipher.block_size()];
-            let count = crypter.update(&buffer[..n], &mut out)?;
-            out.truncate(count);
+            let mut nonce_bytes = [0u8; 12];
+            nonce_bytes[..8].copy_from_slice(&base_nonce);
+            nonce_bytes[8..].copy_from_slice(&chunk_index.to_be_bytes());
+            let nonce = Nonce::try_from(&nonce_bytes[..])
+                .map_err(|_| anyhow::anyhow!("Invalid nonce length")).unwrap();
 
-            yield Bytes::from(out);
+            let ciphertext = cipher.encrypt(&nonce, &buffer[..n]).unwrap();
+            let mut out = Vec::with_capacity(4 + ciphertext.len());
+            out.extend_from_slice(&(ciphertext.len() as u32).to_be_bytes());
+            out.extend_from_slice(&ciphertext);
+
+            tx.send(Ok(Bytes::from(out))).await.unwrap();
+            chunk_index += 1;
         }
+    });
 
-        let mut final_block = vec![0u8; cipher.block_size()];
-        let rest = crypter.finalize(&mut final_block)?;
-        final_block.truncate(rest);
+    Ok(ReceiverStream::new(rx))
+}
 
-        if !final_block.is_empty() {
-            yield Bytes::from(final_block);
+pub async fn decrypt_file_stream_gcm(
+    encrypted_path: PathBuf,
+    decrypted_path: PathBuf,
+    master_key_b64: String,
+) -> Result<()> {
+    info!("Decrypting {:?}", decrypted_path);
+
+    let master_key_bytes = general_purpose::STANDARD
+        .decode(master_key_b64)
+        .map_err(|_| anyhow::anyhow!("Invalid base64"))?;
+
+    let mut reader = BufReader::new(File::open(&encrypted_path).await?);
+
+    let mut header_line = Vec::new();
+    reader.read_until(b'\n', &mut header_line).await?;
+    let header: FileHeader = serde_json::from_slice(&header_line)?;
+
+    let key = Key::<Aes256Gcm>::try_from(master_key_bytes.as_slice())
+        .map_err(|_| anyhow::anyhow!("Invalid AES-256 key length"))?;
+    let cipher = Aes256Gcm::new(&key);
+
+    let mut writer = BufWriter::new(File::create(&decrypted_path).await?);
+    let mut chunk_index: u32 = 0;
+
+    loop {
+        let mut len_buf = [0u8; 4];
+        match reader.read_exact(&mut len_buf).await {
+            Ok(_) => {}
+            Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => break,
+            Err(e) => return Err(e.into()),
         }
-    };
+        let chunk_len = u32::from_be_bytes(len_buf) as usize;
 
-    Ok((stream, encrypted_key_hex))
-}
+        let mut chunk_ciphertext = vec![0u8; chunk_len];
+        reader.read_exact(&mut chunk_ciphertext).await?;
+
+        let mut nonce_bytes = [0u8; 12];
+        nonce_bytes[..8].copy_from_slice(&header.base_nonce);
+        nonce_bytes[8..].copy_from_slice(&chunk_index.to_be_bytes());
+        let nonce = Nonce::try_from(&nonce_bytes[..])
+            .map_err(|_| anyhow::anyhow!("Invalid nonce length"))?;
+
+        let plaintext = cipher
+            .decrypt(&nonce, chunk_ciphertext.as_slice())
+            .map_err(|e| anyhow::anyhow!("AES-GCM decryption failed: {:?}", e))?;
+
+        writer.write_all(&plaintext).await?;
+        chunk_index += 1;
+    }
+
+    writer.flush().await?;
+    Ok(())
+}

src/utils/locks.rs

@@ -1,10 +1,10 @@
 use anyhow::{Context, Result};
-use chrono::{Local};
-use tracing::{info, warn, error};
+use chrono::Local;
 use std::path::{Path, PathBuf};
 use std::time::{Duration, SystemTime};
-use tokio::fs::{OpenOptions, metadata, remove_file, create_dir_all, read_dir};
+use tokio::fs::{OpenOptions, create_dir_all, metadata, read_dir, remove_file};
 use tokio::io::AsyncWriteExt;
+use tracing::{error, info, warn};
 
 /// Lock type for logging purposes
 #[derive(Debug, Copy, Clone)]
@@ -59,6 +59,24 @@ impl FileLock {
         Ok(())
     }
 
+    pub async fn is_locked(id: &str) -> Result<bool> {
+        let path = Self::lock_file_path(id);
+
+        if !path.exists() {
+            return Ok(false);
+        }
+
+        let meta = metadata(&path).await?;
+
+        if let Ok(modified) = meta.modified() {
+            let age = SystemTime::now().duration_since(modified)?;
+
+            if age > Duration::from_secs(24 * 60 * 60) {
+                return Ok(false);
+            }
+        }
+        Ok(true)
+    }
 
     /// Acquire a file-based lock
     pub async fn acquire(id: &str, service_name: &str) -> Result<()> {
@@ -88,10 +106,13 @@ impl FileLock {
             .await
             .with_context(|| format!("Failed to create lock file for {}", id))?;
 
-        f.write_all(format!("Service: {}\n", service_name).as_bytes()).await?;
-        f.write_all(format!("PID: {}\n", std::process::id()).as_bytes()).await?;
+        f.write_all(format!("Service: {}\n", service_name).as_bytes())
+            .await?;
+        f.write_all(format!("PID: {}\n", std::process::id()).as_bytes())
+            .await?;
         // f.write_all(format!("Timestamp: {}\n", Utc::now()).as_bytes()).await?;
-        f.write_all(format!("Timestamp: {}\n", Local::now()).as_bytes()).await?;
+        f.write_all(format!("Timestamp: {}\n", Local::now()).as_bytes())
+            .await?;
 
         info!("Successfully acquired lock for {}", id);
         Ok(())
@@ -106,7 +127,10 @@ impl FileLock {
             remove_file(&path).await?;
             info!("Released file lock for {}", id);
         } else {
-            warn!("Attempted to release lock for {}, but file does not exist", id);
+            warn!(
+                "Attempted to release lock for {}, but file does not exist",
+                id
+            );
         }
         Ok(())
     }

src/utils/logging.rs

@@ -28,14 +28,14 @@ pub fn init_logger() {
         .with_ansi(false)
         .with_target(false);
 
-    // let env_filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));
-    let env_filter = EnvFilter::new(CONFIG.log.clone());
-
+    let env_filter = EnvFilter::new(format!(
+        "{},aws_sdk_s3=warn,aws_smithy_http=warn,aws_smithy_runtime=warn,aws_smithy_client=warn,aws_smithy_types=warn,reqwest=warn,hyper=warn,tokio=warn,h2=warn",
+        CONFIG.log
+    ));
 
     let term_layer = fmt::layer()
         .with_writer(std::io::stdout)
         .with_timer(timer.clone())
-        // .with_timer(tracing_subscriber::fmt::time::LocalTime::rfc_3339())
         .with_ansi(true)
         .with_target(false)
         .with_filter(env_filter);

src/utils/mod.rs

@@ -6,3 +6,7 @@ pub mod text;
 pub mod file;
 pub mod locks;
 pub mod logging;
+pub mod tus;
+pub mod deserializer;
+pub mod compress;
+pub mod stream;

src/utils/stream.rs

@@ -0,0 +1,37 @@
+use crate::utils::file::encrypt_file_stream_gcm;
+use anyhow::Result;
+use bytes::Bytes;
+use futures::{Stream, StreamExt};
+use std::pin::Pin;
+use tokio_util::io::ReaderStream;
+
+pub struct UploadStream {
+    pub stream: Pin<Box<dyn Stream<Item = Result<Bytes, std::io::Error>> + Send>>,
+}
+
+pub async fn build_stream(
+    file_path: &std::path::Path,
+    encrypt: bool,
+    master_key_b64: &String,
+) -> Result<UploadStream> {
+    if encrypt {
+        let encrypted_stream =
+            encrypt_file_stream_gcm(file_path.to_path_buf(), master_key_b64.to_string()).await?;
+
+        let stream = Box::pin(
+            encrypted_stream
+                .map(|r| r.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e))),
+        );
+
+        Ok(UploadStream { stream })
+    } else {
+        let file = tokio::fs::File::open(file_path).await?;
+        let reader = ReaderStream::new(file);
+
+        let stream = Box::pin(
+            reader.map(|r| r.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e))),
+        );
+
+        Ok(UploadStream { stream })
+    }
+}

src/utils/task_manager/cron.rs

@@ -1,17 +1,17 @@
 use crate::utils::task_manager::models;
 use crate::utils::task_manager::tasks::{remove_task, upsert_task};
 use crate::utils::text::normalize_cron;
-use chrono::{Local};
+use chrono::Local;
 use cron::Schedule;
-use tracing::debug;
 use redis::AsyncCommands;
 use redis::aio::MultiplexedConnection;
+use serde_json::Value;
 use std::str::FromStr;
+use tracing::debug;
 use tracing::info;
 
 pub fn next_run_timestamp(expr: &str) -> i64 {
     let schedule = Schedule::from_str(expr).unwrap();
-    // schedule.upcoming(Utc).next().unwrap().timestamp()
     schedule.upcoming(Local).next().unwrap().timestamp()
 }
 
@@ -21,6 +21,7 @@ pub async fn check_and_update_cron(
     args: Vec<String>,
     task: &str,
     task_name: String,
+    metadata: Option<Value>,
 ) {
     let redis_key = format!("redbeat:{}", task_name);
 
@@ -44,16 +45,31 @@ pub async fn check_and_update_cron(
                 let raw: String = conn.hget(&redis_key, "data").await.unwrap();
                 let stored: models::PeriodicTask = serde_json::from_str(&raw).unwrap();
 
-                if stored.cron != cron {
-                    upsert_task(conn, &task_name, task, &cron, args.clone())
-                        .await
-                        .unwrap_or_else(|e| {
-                            tracing::error!("Failed to update task {}: {:?}", task_name, e);
-                        });
-                    info!("Task {} updated", task_name);
+                let cron_changed = stored.cron != cron;
+                let args_changed = stored.args != args;
+                let metadata_changed = stored.metadata != metadata;
+
+                if cron_changed || args_changed || metadata_changed {
+                    upsert_task(
+                        conn,
+                        &task_name,
+                        task,
+                        &cron,
+                        args.clone(),
+                        metadata,
+                    )
+                    .await
+                    .unwrap_or_else(|e| {
+                        tracing::error!("Failed to update task {}: {:?}", task_name, e);
+                    });
+
+                    info!(
+                        "Task {} updated (cron: {}, args: {}, metadata: {})",
+                        task_name, cron_changed, args_changed, metadata_changed
+                    );
                 }
             } else {
-                upsert_task(conn, &task_name, task, &cron, args)
+                upsert_task(conn, &task_name, task, &cron, args, metadata)
                     .await
                     .unwrap_or_else(|e| {
                         tracing::error!("Failed to create task {}: {:?}", task_name, e);

src/utils/task_manager/models.rs

@@ -1,4 +1,5 @@
 use serde::{Deserialize, Serialize};
+use serde_json::Value;
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct PeriodicTask {
@@ -6,4 +7,5 @@ pub struct PeriodicTask {
     pub cron: String,
     pub args: Vec<String>,
     pub enabled: bool,
+    pub metadata: Option<Value>,
 }

src/utils/task_manager/scheduler.rs

@@ -5,23 +5,22 @@ use crate::utils::common::BackupMethod;
 use crate::utils::task_manager::cron::next_run_timestamp;
 use crate::utils::task_manager::models::PeriodicTask;
 use crate::utils::task_manager::tasks::SCHEDULE_KEY;
-use tracing::info;
 use redis::AsyncCommands;
 use redis::aio::MultiplexedConnection;
+use serde_json::Value;
 use std::sync::Arc;
 use tracing::error;
+use tracing::info;
+use crate::services::api::models::agent::status::DatabaseStorage;
 
 pub async fn scheduler_loop(mut conn: MultiplexedConnection) {
     loop {
-        // let now = chrono::Utc::now().timestamp();
         let now = chrono::Local::now().timestamp();
-        // info!("Scheduling task {}", chrono::Local::now());
 
         let due: Vec<String> = conn
             .zrangebyscore(SCHEDULE_KEY, 0, now)
             .await
             .unwrap_or_default();
-
         for key in due {
             let raw: String = conn.hget(&key, "data").await.unwrap();
             let task: PeriodicTask = serde_json::from_str(&raw).unwrap();
@@ -29,35 +28,39 @@ pub async fn scheduler_loop(mut conn: MultiplexedConnection) {
             if !task.enabled {
                 continue;
             }
-
             let task_clone = task.clone();
             let mut conn_clone = conn.clone();
 
             tokio::spawn(async move {
                 info!(
-                    "Executing task={} args={:?}",
-                    task_clone.task, task_clone.args
+                    "Executing task={} args={:?} metadata={:?}",
+                    task_clone.task, task_clone.args, task_clone.metadata
                 );
-
-                // let _ = execute_task(task_clone.task.as_str(), task_clone.args).await;
-
-                if let Err(e) = execute_task(task_clone.task.as_str(), task_clone.args).await {
+                if let Err(e) = execute_task(
+                    task_clone.task.as_str(),
+                    task_clone.args,
+                    task_clone.metadata,
+                )
+                .await
+                {
                     error!(
                         "An error occurred while executing task={} : {:?}",
                         task_clone.task, e
                     );
                 }
-
                 let next_ts = next_run_timestamp(&task_clone.cron);
                 let _: () = conn_clone.zadd(SCHEDULE_KEY, &key, next_ts).await.unwrap();
             });
         }
-
         tokio::time::sleep(std::time::Duration::from_secs(1)).await;
     }
 }
 
-pub async fn execute_task(task: &str, args: Vec<String>) -> Result<(), anyhow::Error> {
+pub async fn execute_task(
+    task: &str,
+    args: Vec<String>,
+    metadata: Option<Value>,
+) -> Result<(), anyhow::Error> {
     match task {
         "tasks.database.periodic_backup" => {
             let generated_id = &args[0];
@@ -69,8 +72,24 @@ pub async fn execute_task(task: &str, args: Vec<String>) -> Result<(), anyhow::E
             let backup_service = BackupService::new(ctx.clone());
             let config = config_service.load(None).unwrap();
 
+            let metadata_obj = metadata
+                .into_iter()
+                .next()
+                .ok_or_else(|| anyhow::anyhow!("Metadata missing"))?;
+
+            let storages_value: &Value = metadata_obj
+                .get("storages")
+                .ok_or_else(|| anyhow::anyhow!("storages key missing"))?;
+            
+            let encrypt_value: &Value = metadata_obj
+                .get("encrypt")
+                .ok_or_else(|| anyhow::anyhow!("encrypt key missing"))?;
+            
+            let storages: Vec<DatabaseStorage> = serde_json::from_value(storages_value.clone())?;
+            let encrypt : bool = serde_json::from_value(encrypt_value.clone())?;
+
             backup_service
-                .dispatch(generated_id, &config, BackupMethod::Automatic)
+                .dispatch(generated_id, &config, BackupMethod::Automatic, &storages, encrypt)
                 .await;
 
             Ok(())

src/utils/task_manager/tasks.rs

@@ -3,6 +3,7 @@
 use crate::utils::task_manager::cron::next_run_timestamp;
 use crate::utils::task_manager::models::PeriodicTask;
 use redis::aio::MultiplexedConnection;
+use serde_json::Value;
 
 pub const SCHEDULE_KEY: &str = "redbeat:schedule";
 
@@ -12,6 +13,7 @@ pub async fn upsert_task(
     task: &str,
     cron: &str,
     args: Vec<String>,
+    metadata: Option<Value>,
 ) -> redis::RedisResult<()> {
     let key = format!("redbeat:{}", name);
     let next_ts = next_run_timestamp(cron);
@@ -21,6 +23,7 @@ pub async fn upsert_task(
         cron: cron.to_string(),
         args,
         enabled: true,
+        metadata
     };
 
     let payload = serde_json::to_string(&entry).unwrap();

src/utils/tus.rs

@@ -0,0 +1,98 @@
+use anyhow::Result;
+use bytes::Bytes;
+use futures::{Stream, StreamExt};
+use log::info;
+use reqwest::header::{CONTENT_TYPE, HeaderMap, HeaderValue};
+
+const PATCH_CHUNK_SIZE: usize = 1 * 1024 * 1024;
+
+pub async fn upload_to_tus_stream_with_headers<S>(
+    encrypted_stream: S,
+    tus_endpoint: &str,
+    extra_headers: HeaderMap,
+    total_size: u64,
+) -> Result<()>
+where
+    S: Stream<Item = Result<Bytes, std::io::Error>> + Send + 'static,
+{
+    let client = reqwest::Client::new();
+
+    info!("File size: {}", total_size);
+
+    let mut headers = HeaderMap::new();
+    headers.insert("Tus-Resumable", HeaderValue::from_static("1.0.0"));
+    headers.insert("Upload-Defer-Length", HeaderValue::from_static("1"));
+
+    let resp = client
+        .post(tus_endpoint)
+        .headers(headers.clone())
+        .send()
+        .await?;
+    if !resp.status().is_success() {
+        anyhow::bail!("Failed to create upload: {}", resp.status());
+    }
+
+    let upload_url = resp
+        .headers()
+        .get("Location")
+        .ok_or_else(|| anyhow::anyhow!("Missing Location header"))?
+        .to_str()?
+        .to_string();
+
+    let mut stream = Box::pin(
+        encrypted_stream.map(|r| r.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e))),
+    );
+
+    let mut offset: u64 = 0;
+
+    while let Some(chunk) = stream.next().await {
+        let chunk = chunk?;
+        for sub_chunk in chunk.chunks(PATCH_CHUNK_SIZE) {
+            let mut patch_headers = extra_headers.clone();
+            patch_headers.insert("Tus-Resumable", HeaderValue::from_static("1.0.0"));
+            patch_headers.insert("Upload-Offset", HeaderValue::from_str(&offset.to_string())?);
+            patch_headers.insert(
+                CONTENT_TYPE,
+                HeaderValue::from_static("application/offset+octet-stream"),
+            );
+
+            let patch_resp = client
+                .patch(&upload_url)
+                .headers(patch_headers)
+                .body(sub_chunk.to_vec())
+                .send()
+                .await?;
+
+            if !patch_resp.status().is_success() {
+                anyhow::bail!(
+                    "Chunk upload failed at offset {}: {}",
+                    offset,
+                    patch_resp.status()
+                );
+            }
+            offset += sub_chunk.len() as u64;
+            // info!("Progress: {}/{}", offset, total_size);
+        }
+    }
+
+    let mut finalize_headers = extra_headers.clone();
+    finalize_headers.insert("Tus-Resumable", HeaderValue::from_static("1.0.0"));
+    finalize_headers.insert("Upload-Offset", HeaderValue::from_str(&offset.to_string())?);
+    finalize_headers.insert("Upload-Length", HeaderValue::from_str(&offset.to_string())?);
+    finalize_headers.insert(
+        CONTENT_TYPE,
+        HeaderValue::from_static("application/offset+octet-stream"),
+    );
+
+    let finalize_resp = client
+        .patch(&upload_url)
+        .headers(finalize_headers)
+        .send()
+        .await?;
+
+    if !finalize_resp.status().is_success() {
+        anyhow::bail!("Failed to finalize upload");
+    }
+
+    Ok(())
+}