feat: add-gcs-storage-provider

Cargo.toml

@@ -41,6 +41,8 @@ aws-config = "1.8.13"
 aws-sdk-s3 = { version = "1.122.0", features = ["behavior-version-latest"] }
 azure_core = "1.0.0"
 azure_storage_blob = "1.0.0"
+google-cloud-storage = "1.15"
+google-cloud-auth = "1.13"
 async-compression = { version = "0.4.37", features = ["tokio", "gzip"] }
 tokio-tar = "0.3.1"
 oauth2 = "5.0.0"

databases.json

@@ -111,15 +111,6 @@
       "port": 1433,
       "host": "db-mssql",
       "generated_id": "16706125-ff7e-4c97-8c83-0adeff214682"
-    },
-    {
-      "name": "Test database - PostgreSQL cluster",
-      "type": "postgresql-cluster",
-      "username": "nextclouddbuser",
-      "password": "50AL2Oh5IXajbOAxfJ",
-      "port": 5432,
-      "host": "nextcloud-db",
-      "generated_id": "16678199-ff7e-4c97-8c83-0adeff214681"
     }
   ]
 }

docker-compose.yml

@@ -19,20 +19,17 @@ services:
       APP_ENV: development
       LOG: debug
       TZ: "Europe/Paris"
-      EDGE_KEY: "eyJzZXJ2ZXJVcmwiOiJodHRwOi8vbG9jYWxob3N0Ojg4ODciLCJhZ2VudElkIjoiNWRkZTE1NTctZWQ1ZC00MjUxLThiZDMtMDE0MjkxOTg2OGZjIiwibWFzdGVyS2V5QjY0IjoiQlhWM1hvbEM2NTZTVjdkTmdjV1BHUWxrKytycExJNmxHRGk3Q1BCNWllbz0ifQ=="
+      EDGE_KEY: "eyJzZXJ2ZXJVcmwiOiJodHRwOi8vbG9jYWxob3N0Ojg4ODciLCJhZ2VudElkIjoiNzY1OGIzYjctNjg5MC00MjllLThkN2QtNjU2ZjlmYTJlMDRjIiwibWFzdGVyS2V5QjY0IjoiQlhWM1hvbEM2NTZTVjdkTmdjV1BHUWxrKytycExJNmxHRGk3Q1BCNWllbz0ifQ=="
       #CHUNK_SIZE_MB: "1"
       #POOLING: 1
       #DATABASES_CONFIG_FILE: "config.toml"
     extra_hosts:
       - "localhost:host-gateway"
     networks:
       - portabase
-
     cpus: "1.50"
-
     mem_limit: 4g
     memswap_limit: 4g
-
     pids_limit: 512
 
 

src/services/storage/mod.rs

@@ -6,6 +6,7 @@ use crate::services::backup::models::{BackupResult, UploadResult};
 use crate::utils::common::BackupMethod;
 use async_trait::async_trait;
 use providers::azure_blob;
+use providers::google_cloud_storage;
 use providers::google_drive;
 use providers::local;
 use providers::s3;
@@ -34,6 +35,9 @@ pub fn get_provider(storage: &DatabaseStorage) -> Option<Box<dyn StorageProvider
         "s3" => Some(Box::new(s3::S3Provider {})),
         "blob" => Some(Box::new(azure_blob::AzureBlobProvider {})),
         "google-drive" => Some(Box::new(google_drive::GoogleDriveProvider {})),
+        "google-cloud-storage" => Some(Box::new(
+            google_cloud_storage::GoogleCloudStorageProvider {},
+        )),
         _ => {
             error!("Unknown storage provider: {}", storage.provider);
             None

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

@@ -44,8 +44,6 @@ pub(crate) fn hmac_sha256_b64(key: &[u8], data: &str) -> Result<String> {
     Ok(STANDARD.encode(sig))
 }
 
-/// Build Service SAS query pairs (raw, un-encoded) for `canonical_resource`
-/// e.g. `/blob/{account}/{container}/{blob}`.
 pub fn build_service_sas(
     resolved: &ResolvedAzure,
     canonical_resource: &str,
@@ -81,7 +79,6 @@ pub fn build_service_sas(
     ])
 }
 
-/// Build a SAS-scoped URL for a blob (or container when `blob` is empty).
 pub fn build_sas_url(
     resolved: &ResolvedAzure,
     container: &str,
@@ -110,12 +107,10 @@ pub fn build_sas_url(
     Ok(url)
 }
 
-/// Default block size for the provider path (mirrors the S3 provider's PART_SIZE).
 pub const BLOCK_SIZE: usize = 100 * 1024 * 1024;
 
 type ByteStream = Pin<Box<dyn Stream<Item = Result<Bytes, std::io::Error>> + Send>>;
 
-/// Stage one block under a zero-padded sequential id; records the RAW id bytes.
 async fn stage_block(
     bbc: &BlockBlobClient,
     index: u32,
@@ -132,13 +127,6 @@ async fn stage_block(
     Ok(())
 }
 
-/// Stream `body` to `{container}/{blob}` using Azure block upload. Never buffers the
-/// full payload: at most one `block_size` block plus one inbound chunk is resident
-/// (mirrors the S3 provider's per-part guarantee).
-///
-/// Assumes the container already exists — S3-faithful, no container creation. Uncommitted
-/// blocks are garbage-collected by Azure if `commit_block_list` is never reached, so no
-/// explicit abort is needed on the error path (unlike S3 multipart).
 pub async fn upload_stream_to_azure(
     resolved: &ResolvedAzure,
     container: &str,

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

@@ -26,7 +26,6 @@ fn parse_connection_string(cs: &str) -> std::collections::HashMap<String, String
 }
 
 impl AzureBlobProviderConfig {
-    /// Resolve effective connection params, preferring the connection string when non-empty.
     pub fn resolve(&self) -> Result<ResolvedAzure> {
         if !self.connection_string.trim().is_empty() {
             let map = parse_connection_string(&self.connection_string);

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

@@ -0,0 +1,122 @@
+use crate::services::storage::providers::google_cloud_storage::models::GoogleCloudStorageProviderConfig;
+use anyhow::{Context, Result};
+use bytes::Bytes;
+use futures::Stream;
+use futures::StreamExt;
+use google_cloud_auth::credentials::Credentials;
+use google_cloud_storage::client::Storage;
+use google_cloud_storage::streaming_source::{SizeHint, StreamingSource};
+use std::pin::Pin;
+
+pub fn build_credentials(cfg: &GoogleCloudStorageProviderConfig) -> Result<Credentials> {
+    // Service-account JSON stores the PEM with `\n` escape sequences. When the key is
+    // carried through config as a JSON string those can arrive as literal two-char `\n`
+    // sequences rather than real newlines, so the PEM parser finds no `-----BEGIN-----`
+    // line ("no items found"). Normalize them back to real newlines. A PEM that already
+    // has real newlines contains no literal `\n` pairs, so this is a no-op for it.
+    let private_key = cfg.private_key.replace("\\n", "\n");
+
+    let key = serde_json::json!({
+        "type": "service_account",
+        "project_id": cfg.project_id,
+        "client_email": cfg.client_email,
+        "private_key": private_key,
+        "private_key_id": "",
+        "token_uri": "https://oauth2.googleapis.com/token",
+        "universe_domain": "googleapis.com",
+    });
+
+    google_cloud_auth::credentials::service_account::Builder::new(key)
+        .build()
+        .context("failed to build GCS service account credentials")
+}
+
+pub async fn build_client(cfg: &GoogleCloudStorageProviderConfig) -> Result<Storage> {
+    let endpoint = cfg.api_endpoint.as_deref().filter(|s| !s.trim().is_empty());
+
+    // A custom endpoint means a local emulator (fake-gcs-server), which does not verify
+    // credentials. Use anonymous creds so a dummy/empty `private_key` in the emulator
+    // config doesn't trip the service-account PEM parser. Real GCS still uses the
+    // service-account key built from config.
+    let builder = if let Some(ep) = endpoint {
+        let creds = google_cloud_auth::credentials::anonymous::Builder::new().build();
+        Storage::builder()
+            .with_credentials(creds)
+            .with_endpoint(ep.to_string())
+    } else {
+        Storage::builder().with_credentials(build_credentials(cfg)?)
+    };
+
+    builder.build().await.context("failed to build GCS client")
+}
+
+/// Bridges `build_stream`'s `Send`-only byte stream into the SDK's `StreamingSource`
+/// (which `send_buffered` requires to be `Send + Sync + 'static`) via a bounded mpsc
+/// channel. Also reports an exact `size_hint`: the SDK picks single-shot vs resumable
+/// upload from `size_hint().upper()` — an unknown bound forces resumable unconditionally
+/// (see `upload_with_client`).
+pub struct StreamSource {
+    rx: tokio::sync::mpsc::Receiver<Result<Bytes, std::io::Error>>,
+    total_size: u64,
+}
+
+impl StreamSource {
+    pub fn from_stream(
+        mut stream: Pin<Box<dyn Stream<Item = Result<Bytes, std::io::Error>> + Send>>,
+        total_size: u64,
+    ) -> Self {
+        let (tx, rx) = tokio::sync::mpsc::channel(8);
+        tokio::spawn(async move {
+            while let Some(item) = stream.next().await {
+                if tx.send(item).await.is_err() {
+                    break;
+                }
+            }
+        });
+        StreamSource { rx, total_size }
+    }
+}
+
+impl StreamingSource for StreamSource {
+    type Error = std::io::Error;
+    async fn next(&mut self) -> Option<Result<Bytes, Self::Error>> {
+        self.rx.recv().await
+    }
+
+    // Report the exact size so the SDK can choose single-shot uploads. The default
+    // impl returns an unknown bound, which forces the resumable path unconditionally.
+    async fn size_hint(&self) -> Result<SizeHint, Self::Error> {
+        Ok(SizeHint::with_exact(self.total_size))
+    }
+}
+
+pub async fn upload_with_client(
+    client: &Storage,
+    bucket: &str,
+    object: &str,
+    source: StreamSource,
+    force_single_shot: bool,
+) -> Result<()> {
+    // `write_object` uses gRPC-style resource names: the bucket must be passed as
+    // `projects/_/buckets/<name>`, not the bare bucket id.
+    let bucket_resource = format!("projects/_/buckets/{bucket}");
+
+    let mut builder = client.write_object(bucket_resource, object, source);
+
+    // Resumable uploads follow a server-generated `Location` URL. When pointed at a
+    // custom `apiEndpoint` on a non-443 port, the SDK's transport drops the port from
+    // the `Host` header (google-cloud-gax-internal `host.rs`), so emulators that build
+    // the `Location` from `Host` hand back a portless URL the SDK then hangs on. A
+    // single-shot upload issues one request to the configured endpoint (no `Location`
+    // to follow), sidestepping the bug. We force it only for custom endpoints; against
+    // real GCS we keep resumable (bounded memory + resume on large backups).
+    if force_single_shot {
+        builder = builder.with_resumable_upload_threshold(usize::MAX);
+    }
+
+    builder
+        .send_buffered()
+        .await
+        .context("GCS write_object failed")?;
+    Ok(())
+}

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

@@ -0,0 +1,148 @@
+pub mod helpers;
+mod models;
+
+use crate::core::context::Context;
+use crate::services::api::models::agent::status::DatabaseStorage;
+use crate::services::backup::models::{BackupResult, UploadResult};
+use crate::services::storage::StorageProvider;
+use crate::services::storage::providers::google_cloud_storage::helpers::{
+    StreamSource, build_client, upload_with_client,
+};
+use crate::services::storage::providers::google_cloud_storage::models::GoogleCloudStorageProviderConfig;
+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};
+
+pub struct GoogleCloudStorageProvider {}
+
+#[async_trait]
+impl StorageProvider for GoogleCloudStorageProvider {
+    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: GoogleCloudStorageProviderConfig = 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);
+
+        let client = match build_client(&config).await {
+            Ok(c) => c,
+            Err(e) => {
+                error!("GCS client 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 source = StreamSource::from_stream(upload.stream, total_size);
+
+        // A custom apiEndpoint (self-hosted / emulator) on a non-443 port trips an
+        // upstream SDK bug in the resumable-upload path; force single-shot for it.
+        let force_single_shot = config
+            .api_endpoint
+            .as_deref()
+            .is_some_and(|s| !s.trim().is_empty());
+
+        info!(
+            "Starting GCS upload to {}/{} (single_shot={})",
+            config.bucket_name, remote_file_path, force_single_shot
+        );
+
+        match upload_with_client(
+            &client,
+            &config.bucket_name,
+            &remote_file_path,
+            source,
+            force_single_shot,
+        )
+        .await
+        {
+            Ok(_) => {
+                info!("GCS upload successful: {}", remote_file_path);
+                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!("GCS 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),
+                }
+            }
+        }
+    }
+}