Dev

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"

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/domain/factory.rs

@@ -1,5 +1,6 @@
 use crate::domain::mongodb::database::MongoDatabase;
 use crate::domain::mysql::database::MySQLDatabase;
+use crate::domain::postgres::cluster::database::PostgresClusterDatabase;
 use crate::domain::postgres::database::PostgresDatabase;
 use crate::domain::postgres::{detect_format_from_file, detect_format_from_size};
 use crate::domain::redis::database::RedisDatabase;
@@ -31,6 +32,7 @@ impl DatabaseFactory {
                 let format = detect_format_from_size(&cfg).await;
                 Arc::new(PostgresDatabase::new(cfg, format))
             }
+            DbType::PostgresqlCluster => Arc::new(PostgresClusterDatabase::new(cfg)),
             DbType::Mysql => Arc::new(MySQLDatabase::new(cfg)),
             DbType::Mariadb => Arc::new(MariaDBDatabase::new(cfg)),
             DbType::MongoDB => Arc::new(MongoDatabase::new(cfg)),
@@ -48,6 +50,7 @@ impl DatabaseFactory {
                 let format = detect_format_from_file(restore_file);
                 Arc::new(PostgresDatabase::new(cfg, format))
             }
+            DbType::PostgresqlCluster => Arc::new(PostgresClusterDatabase::new(cfg)),
             DbType::Mysql => Arc::new(MySQLDatabase::new(cfg)),
             DbType::Mariadb => Arc::new(MariaDBDatabase::new(cfg)),
             DbType::MongoDB => Arc::new(MongoDatabase::new(cfg)),

src/domain/postgres/cluster/backup.rs

@@ -0,0 +1,79 @@
+use anyhow::Result;
+use std::collections::HashMap;
+use std::path::PathBuf;
+use std::process::Command;
+use std::sync::Arc;
+use std::time::Instant;
+
+use super::super::connection::{
+    is_superuser, pg_dumpall_binary_name, select_pg_path, server_version,
+};
+use crate::services::backup::logger::JobLogger;
+use crate::services::config::DatabaseConfig;
+
+pub async fn run(
+    cfg: DatabaseConfig,
+    backup_dir: PathBuf,
+    env: HashMap<String, String>,
+    logger: Arc<JobLogger>,
+) -> Result<PathBuf> {
+    tokio::task::spawn_blocking(move || -> Result<PathBuf> {
+        logger.log("info", format!("Starting cluster backup for {}", cfg.name));
+
+        let version = match futures::executor::block_on(server_version(&cfg)) {
+            Ok(v) => v,
+            Err(e) => {
+                logger.log("error", format!("Failed to get server version for {}: {:?}", cfg.name, e));
+                return Err(e.into());
+            }
+        };
+
+        match futures::executor::block_on(is_superuser(&cfg)) {
+            Ok(true) => {}
+            Ok(false) => {
+                logger.log("error", format!("postgresql-cluster backup requires a superuser role for {}", cfg.name));
+                anyhow::bail!("postgresql-cluster backup requires a superuser role for {}", cfg.name);
+            }
+            Err(e) => {
+                logger.log("error", format!("Failed to check superuser status for {}: {:?}", cfg.name, e));
+                return Err(e.into());
+            }
+        }
+
+        let pg_dumpall = select_pg_path(&version).join(pg_dumpall_binary_name());
+        let file_path = backup_dir.join(format!("{}.sql", cfg.generated_id));
+
+        logger.log("info", format!("Running pg_dumpall for cluster {} via {:?}", cfg.name, pg_dumpall));
+
+        let start = Instant::now();
+        let output = Command::new(&pg_dumpall)
+            .arg("--host").arg(&cfg.host)
+            .arg("--port").arg(cfg.port.to_string())
+            .arg("--username").arg(&cfg.username)
+            .arg("-v")
+            .arg("-f").arg(&file_path)
+            .envs(env)
+            .output();
+        let duration_ms = start.elapsed().as_millis() as f64;
+
+        match output {
+            Ok(o) => {
+                let stderr = String::from_utf8_lossy(&o.stderr).to_string();
+                let exit_code = o.status.code().unwrap_or(-1);
+                if o.status.success() {
+                    logger.log_command("pg_dumpall", if stderr.is_empty() { None } else { Some(stderr) }, Some(0), Some(duration_ms));
+                    logger.log("info", format!("Cluster backup completed for {} at {:?}", cfg.name, file_path));
+                    Ok(file_path)
+                } else {
+                    logger.log_command("pg_dumpall", Some(stderr), Some(exit_code), Some(duration_ms));
+                    anyhow::bail!("Cluster backup (pg_dumpall) failed for {}", cfg.name);
+                }
+            }
+            Err(e) => {
+                logger.log_command("pg_dumpall", Some(e.to_string()), Some(-1), Some(duration_ms));
+                Err(e.into())
+            }
+        }
+    })
+    .await?
+}

src/domain/postgres/cluster/database.rs

@@ -0,0 +1,53 @@
+use anyhow::Result;
+use async_trait::async_trait;
+use std::collections::HashMap;
+use std::path::{Path, PathBuf};
+use std::sync::Arc;
+
+use super::super::ping;
+use super::{backup, restore};
+use crate::domain::factory::Database;
+use crate::services::backup::logger::JobLogger;
+use crate::services::config::DatabaseConfig;
+use crate::utils::locks::{DbOpLock, FileLock};
+
+pub struct PostgresClusterDatabase {
+    pub cfg: DatabaseConfig,
+}
+
+impl PostgresClusterDatabase {
+    pub fn new(cfg: DatabaseConfig) -> Self {
+        Self { cfg }
+    }
+
+    fn build_env(&self) -> HashMap<String, String> {
+        let mut envs = std::env::vars().collect::<HashMap<_, _>>();
+        envs.insert("PGPASSWORD".to_string(), self.cfg.password.to_string());
+        envs
+    }
+}
+
+#[async_trait]
+impl Database for PostgresClusterDatabase {
+    fn file_extension(&self) -> &'static str {
+        ".sql"
+    }
+
+    async fn ping(&self) -> Result<bool> {
+        ping::run(self.cfg.clone()).await
+    }
+
+    async fn backup(&self, dir: &Path, logger: Arc<JobLogger>) -> Result<PathBuf> {
+        FileLock::acquire(&self.cfg.generated_id, DbOpLock::Backup.as_str()).await?;
+        let res = backup::run(self.cfg.clone(), dir.to_path_buf(), self.build_env(), logger).await;
+        FileLock::release(&self.cfg.generated_id).await?;
+        res
+    }
+
+    async fn restore(&self, file: &Path, logger: Arc<JobLogger>) -> Result<()> {
+        FileLock::acquire(&self.cfg.generated_id, DbOpLock::Restore.as_str()).await?;
+        let res = restore::run(self.cfg.clone(), file.to_path_buf(), self.build_env(), logger).await;
+        FileLock::release(&self.cfg.generated_id).await?;
+        res
+    }
+}

src/domain/postgres/cluster/mod.rs

@@ -0,0 +1,3 @@
+pub mod backup;
+pub mod database;
+pub mod restore;

src/domain/postgres/cluster/restore.rs

@@ -0,0 +1,78 @@
+use anyhow::Result;
+use std::collections::HashMap;
+use std::path::PathBuf;
+use std::process::Command;
+use std::sync::Arc;
+use std::time::Instant;
+
+use super::super::connection::{is_superuser, psql_binary_name, select_pg_path, server_version};
+use crate::services::backup::logger::JobLogger;
+use crate::services::config::DatabaseConfig;
+
+pub async fn run(
+    cfg: DatabaseConfig,
+    restore_file: PathBuf,
+    env: HashMap<String, String>,
+    logger: Arc<JobLogger>,
+) -> Result<()> {
+    tokio::task::spawn_blocking(move || -> Result<()> {
+        logger.log("info", format!("Starting cluster restore for {}", cfg.name));
+
+        let version = match futures::executor::block_on(server_version(&cfg)) {
+            Ok(v) => v,
+            Err(e) => {
+                logger.log("error", format!("Failed to get server version for {}: {:?}", cfg.name, e));
+                return Err(e.into());
+            }
+        };
+
+        match futures::executor::block_on(is_superuser(&cfg)) {
+            Ok(true) => {}
+            Ok(false) => {
+                logger.log("error", format!("postgresql-cluster restore requires a superuser role for {}", cfg.name));
+                anyhow::bail!("postgresql-cluster restore requires a superuser role for {}", cfg.name);
+            }
+            Err(e) => {
+                logger.log("error", format!("Failed to check superuser status for {}: {:?}", cfg.name, e));
+                return Err(e.into());
+            }
+        }
+
+        let psql = select_pg_path(&version).join(psql_binary_name());
+
+        logger.log("info", format!("Replaying cluster dump for {} via {:?}", cfg.name, psql));
+
+        let start = Instant::now();
+        let output = Command::new(&psql)
+            .arg("--host").arg(&cfg.host)
+            .arg("--port").arg(cfg.port.to_string())
+            .arg("--username").arg(&cfg.username)
+            .arg("--dbname").arg("postgres")
+            .arg("-f").arg(&restore_file)
+            .envs(env)
+            .output();
+        let duration_ms = start.elapsed().as_millis() as f64;
+
+        match output {
+            Ok(o) => {
+                let stderr = String::from_utf8_lossy(&o.stderr).to_string();
+                let stdout = String::from_utf8_lossy(&o.stdout).to_string();
+                let combined = format!("{}{}", stdout, stderr);
+                let exit_code = o.status.code().unwrap_or(-1);
+                if o.status.success() {
+                    logger.log_command("psql", if combined.is_empty() { None } else { Some(combined) }, Some(0), Some(duration_ms));
+                    logger.log("info", format!("Cluster restore completed for {}", cfg.name));
+                    Ok(())
+                } else {
+                    logger.log_command("psql", if combined.is_empty() { None } else { Some(combined) }, Some(exit_code), Some(duration_ms));
+                    anyhow::bail!("Cluster restore (psql) failed for {}", cfg.name);
+                }
+            }
+            Err(e) => {
+                logger.log_command("psql", Some(e.to_string()), Some(-1), Some(duration_ms));
+                Err(e.into())
+            }
+        }
+    })
+    .await?
+}

src/domain/postgres/connection.rs

@@ -33,32 +33,21 @@ pub async fn server_version(cfg: &DatabaseConfig) -> Result<String> {
     Ok(version)
 }
 
-/// Resolves the `bin` directory of a PostgreSQL installation for the given
-/// major version, in a cross-platform way.
-///
-/// Resolution order:
-/// 1. The `PG_BIN_DIR` environment variable, if set, is used as-is. This
-///    allows users/CI to override detection for non-standard installs
-///    (e.g. portable PostgreSQL distributions, custom install locations).
-/// 2. Platform-specific default install locations (Debian/Ubuntu packages,
-///    the official Windows installer, Homebrew/Postgres.app on macOS, and
-///    common RPM-based layouts on other Linux distros).
-/// 3. A `PATH` lookup for `pg_dump` (`pg_dump.exe` on Windows), returning
-///    its parent directory.
-/// 4. The historical Debian/Ubuntu path as a last-resort fallback, so the
-///    function keeps returning a `PathBuf` (never panics) even when nothing
-///    was found, preserving the previous behavior for callers.
-///
-/// The override is sourced from `CONFIG.pg_bin_dir` (the `PG_BIN_DIR`
-/// environment variable). An empty value means "unset" and falls through to
-/// detection.
+pub async fn is_superuser(cfg: &DatabaseConfig) -> Result<bool> {
+    let client = connect(cfg).await?;
+    let is_super: bool = client
+        .query_one("SELECT current_setting('is_superuser') = 'on';", &[])
+        .await?
+        .get(0);
+
+    Ok(is_super)
+}
+
+
 pub fn select_pg_path(version: &str) -> std::path::PathBuf {
     select_pg_path_with(version, &CONFIG.pg_bin_dir)
 }
 
-/// Inner resolver behind [`select_pg_path`], parameterized over the
-/// `PG_BIN_DIR` override. Kept pure (no env / no `CONFIG` access) so it is
-/// unit-testable without mutating process-global state.
 pub(crate) fn select_pg_path_with(version: &str, pg_bin_dir: &str) -> std::path::PathBuf {
     let major = version.split('.').next().unwrap_or("17");
 
@@ -109,6 +98,22 @@ pub(crate) fn pg_dump_binary_name() -> &'static str {
     }
 }
 
+pub(crate) fn pg_dumpall_binary_name() -> &'static str {
+    if cfg!(target_os = "windows") {
+        "pg_dumpall.exe"
+    } else {
+        "pg_dumpall"
+    }
+}
+
+pub(crate) fn psql_binary_name() -> &'static str {
+    if cfg!(target_os = "windows") {
+        "psql.exe"
+    } else {
+        "psql"
+    }
+}
+
 pub(crate) fn pg_dump_exists_in(dir: &std::path::Path) -> bool {
     dir.join(pg_dump_binary_name()).is_file()
 }

src/domain/postgres/mod.rs

@@ -1,4 +1,5 @@
 pub mod backup;
+pub(crate) mod cluster;
 pub(crate) mod connection;
 pub mod database;
 mod format;