athena_rs 2.2.0

//! Backup and restore API endpoints.
//!
//! Provides HTTP handlers for creating database backups via `pg_dump` (directory
//! format), uploading them to S3, listing available backups, and restoring via
//! `pg_restore`.
//!
//! ## Environment variables
//!
//! | Variable | Description |
//! |---|---|
//! | `ATHENA_BACKUP_S3_BUCKET` | S3 bucket name (required) |
//! | `ATHENA_BACKUP_S3_REGION` | AWS region (default: `us-east-1`) |
//! | `ATHENA_BACKUP_S3_ACCESS_KEY` | AWS access key ID |
//! | `ATHENA_BACKUP_S3_SECRET_KEY` | AWS secret access key |
//! | `ATHENA_BACKUP_S3_ENDPOINT` | Custom S3-compatible endpoint URL (optional) |
//! | `ATHENA_BACKUP_S3_PREFIX` | Key prefix for stored objects (default: `backups`) |
//! | `ATHENA_PG_DUMP_PATH` / `ATHENA_PG_RESTORE_PATH` | Optional absolute paths to pg_dump / pg_restore. When unset the server uses PATH, then auto-downloads a portable Linux x86_64 bundle. |
//!
//! ## Endpoints
//!
//! - `POST /admin/backups` – Run `pg_dump` and upload to S3.
//! - `GET /admin/backups` – List backup objects stored in S3.
//! - `GET /admin/backups/{key}/download` – Download a backup archive from S3.
//! - `POST /admin/backups/{key}/restore` – Download from S3 and run `pg_restore`.
//! - `DELETE /admin/backups/{key}` – Delete a backup from S3.

use std::env;
use std::path::PathBuf;

use actix_web::{
    HttpRequest, HttpResponse, delete, get, post,
    web::{self, Data, Json, Path},
};
use aws_config::BehaviorVersion;
use aws_sdk_s3::Client as S3Client;
use aws_sdk_s3::config::{Credentials, Region};
use chrono::Utc;
use serde::{Deserialize, Serialize};
use serde_json::json;
use sqlx::{FromRow, PgPool, QueryBuilder, Row};
use tokio::process::Command;
use uuid::Uuid;

use crate::AppState;
use crate::api::auth::authorize_static_admin_key;
use crate::api::response::{api_success, bad_request, internal_error, service_unavailable};
use crate::parser::resolve_postgres_uri;
use crate::utils::pg_tools::{PgToolsPaths, ensure_pg_tools};

// ── configuration helpers ────────────────────────────────────────────────────

/// Read a required S3 environment variable.
fn s3_env(key: &str) -> Option<String> {
    env::var(key).ok().filter(|v| !v.trim().is_empty())
}

struct S3Config {
    bucket: String,
    region: String,
    prefix: String,
    access_key: Option<String>,
    secret_key: Option<String>,
    endpoint: Option<String>,
}

impl S3Config {
    fn from_env() -> Option<Self> {
        let bucket = s3_env("ATHENA_BACKUP_S3_BUCKET")?;
        Some(Self {
            bucket,
            region: s3_env("ATHENA_BACKUP_S3_REGION").unwrap_or_else(|| "us-east-1".to_string()),
            prefix: s3_env("ATHENA_BACKUP_S3_PREFIX").unwrap_or_else(|| "backups".to_string()),
            access_key: s3_env("ATHENA_BACKUP_S3_ACCESS_KEY"),
            secret_key: s3_env("ATHENA_BACKUP_S3_SECRET_KEY"),
            endpoint: s3_env("ATHENA_BACKUP_S3_ENDPOINT"),
        })
    }
}

async fn build_s3_client(cfg: &S3Config) -> S3Client {
    tracing::info!(
        "Building S3 client with config: bucket={}, region={}, prefix={:?}, endpoint={:?}, access_key_set={}",
        cfg.bucket,
        cfg.region,
        cfg.prefix,
        cfg.endpoint,
        cfg.access_key.is_some()
    );
    let region: Region = Region::new(cfg.region.clone());

    let aws_config: aws_config::SdkConfig = if let (Some(ak), Some(sk)) =
        (&cfg.access_key, &cfg.secret_key)
    {
        tracing::info!("Using provided AWS credentials for S3 client.");
        let creds: Credentials = Credentials::new(ak, sk, None, None, "athena-env");
        let mut builder: aws_config::ConfigLoader = aws_config::defaults(BehaviorVersion::latest())
            .region(region)
            .credentials_provider(creds);
        if let Some(ep) = &cfg.endpoint {
            tracing::info!("S3 client will use custom endpoint: {ep}");
            builder = builder.endpoint_url(ep);
        }
        builder.load().await
    } else {
        tracing::info!("Using default AWS credentials/config for S3 client.");
        let mut builder: aws_config::ConfigLoader =
            aws_config::defaults(BehaviorVersion::latest()).region(region);
        if let Some(ep) = &cfg.endpoint {
            tracing::info!("S3 client will use custom endpoint: {ep}");
            builder = builder.endpoint_url(ep);
        }
        builder.load().await
    };

    let mut s3_cfg_builder: aws_sdk_s3::config::Builder =
        aws_sdk_s3::config::Builder::from(&aws_config);
    if cfg.endpoint.is_some() {
        // Required for MinIO / path-style custom S3 endpoints.
        tracing::info!("Forcing path-style for S3 client.");
        s3_cfg_builder = s3_cfg_builder.force_path_style(true);
    }

    tracing::info!("S3 client built.");
    S3Client::from_conf(s3_cfg_builder.build())
}

// ── request / response types ─────────────────────────────────────────────────

#[derive(Debug, Deserialize)]
struct CreateBackupRequest {
    /// Logical client name whose Postgres URI will be used for `pg_dump`.
    client_name: String,
    /// Optional human-readable label stored in the S3 object metadata.
    #[serde(default)]
    label: Option<String>,
}

#[derive(Debug, Deserialize)]
struct RestoreBackupRequest {
    /// Logical client name whose Postgres URI is the restore target.
    client_name: String,
}

#[derive(Debug, Serialize)]
struct BackupObject {
    id: String,
    key: String,
    client_name: String,
    label: Option<String>,
    size_bytes: i64,
    created_at: String,
}

#[derive(Debug, Serialize, FromRow)]
struct BackupJobSummary {
    id: i64,
    job_type: String,
    client_name: Option<String>,
    status: String,
    progress_pct: Option<i32>,
    progress_stage: Option<String>,
    s3_bucket: Option<String>,
    s3_key: Option<String>,
    label: Option<String>,
    size_bytes: Option<i64>,
    error_message: Option<String>,
    started_at: chrono::DateTime<Utc>,
    updated_at: chrono::DateTime<Utc>,
    completed_at: Option<chrono::DateTime<Utc>>,
}

#[derive(Debug, Serialize, FromRow)]
struct BackupJobLog {
    id: i64,
    job_id: i64,
    level: String,
    message: String,
    created_at: chrono::DateTime<Utc>,
}

// ── helpers ───────────────────────────────────────────────────────────────────

/// Retrieve the Postgres connection URI for `client_name` from the registry.
fn resolve_pg_uri(state: &AppState, client_name: &str) -> Result<String, HttpResponse> {
    tracing::info!("Resolving Postgres URI for client_name={}", client_name);
    let registered: crate::drivers::postgresql::sqlx_driver::RegisteredClient = state
        .pg_registry
        .registered_client(client_name)
        .ok_or_else(|| {
            tracing::info!("Client '{}' not found in pg_registry.", client_name);
            bad_request(
                "Unknown client",
                format!("No Postgres client named '{}' is registered.", client_name),
            )
        })?;

    // Resolve the URI: prefer config_uri_template (handles env var references),
    // then fall back to pg_uri if it was set directly.
    let uri = registered
        .config_uri_template
        .as_deref()
        .map(resolve_postgres_uri)
        .or(registered.pg_uri)
        .ok_or_else(|| {
            tracing::info!("No usable Postgres URI found for client '{}'", client_name);
            bad_request(
                "Client URI unavailable",
                format!("No Postgres URI is available for client '{}'.", client_name),
            )
        })?;

    tracing::info!("Resolved URI for {}: {}", client_name, &uri);
    Ok(uri)
}

/// Extract the password from a Postgres URI and return
/// `(sanitized_uri_without_password, Option<password>)`.
///
/// Keeps the password out of process command-line arguments (which are visible
/// in `ps` listings) by passing it through the `PGPASSWORD` environment
/// variable instead.
fn extract_pg_password(pg_uri: &str) -> (String, Option<String>) {
    tracing::info!("Extracting password from Postgres URI for command safety.");
    // URI format: postgresql://[user[:password]@]host[:port][/dbname][?...]
    let prefix = if pg_uri.starts_with("postgresql://") {
        "postgresql://"
    } else if pg_uri.starts_with("postgres://") {
        "postgres://"
    } else {
        // Not a URI format we can parse; return as-is.
        tracing::info!("Nonstandard Postgres URI, not extracting password.");
        return (pg_uri.to_string(), None);
    };

    let after_scheme = &pg_uri[prefix.len()..];

    // Find the last '@' which separates user-info from host.
    if let Some(at_pos) = after_scheme.rfind('@') {
        let userinfo = &after_scheme[..at_pos];
        let after_at = &after_scheme[at_pos..]; // includes the '@'

        if let Some(colon_pos) = userinfo.find(':') {
            let user = &userinfo[..colon_pos];
            let password = userinfo[colon_pos + 1..].to_string();
            let sanitized = format!("{}{}{}", prefix, user, after_at);
            tracing::info!("Password extracted from URI (hidden); sanitized URI prepared.");
            return (sanitized, Some(password));
        }
    }
    tracing::info!("No password found in Postgres URI.");
    (pg_uri.to_string(), None)
}

/// Create a temporary directory, run `pg_dump`, archive the result, return the
/// archive path.  All temporary intermediate directories are cleaned up on
/// error; the caller is responsible for cleaning up the returned path.
async fn run_pg_dump(pg_uri: &str) -> Result<PathBuf, String> {
    let tmp_root = env::temp_dir().join(format!("athena_backup_{}", Uuid::new_v4()));
    let dump_dir = tmp_root.join("dump");
    let archive_path = tmp_root.join("backup.tar.gz");

    tracing::info!("Ensuring PostgreSQL tools (pg_dump, etc.) are available.");
    let pg_tools: PgToolsPaths = ensure_pg_tools()
        .await
        .map_err(|e| format!("pg_dump resolution failed: {e}"))?;

    tracing::info!("Creating dump directory at {:?}", &dump_dir);
    tokio::fs::create_dir_all(&dump_dir)
        .await
        .map_err(|e| format!("Could not create temp directory: {e}"))?;

    // Strip password from URI and pass it via PGPASSWORD to avoid exposing
    // credentials in process command-line listings.
    let (pg_uri_safe, pg_password) = extract_pg_password(pg_uri);

    tracing::info!(
        "Invoking pg_dump for dump_dir={:?}, sanitized_uri={:?}",
        &dump_dir,
        &pg_uri_safe
    );

    // Run pg_dump with directory format.
    let mut cmd = Command::new(&pg_tools.pg_dump);
    if let Some(pass) = pg_password {
        cmd.env("PGPASSWORD", pass);
    }
    let status = cmd
        .args(["--format=directory", "--file"])
        .arg(&dump_dir)
        .arg(&pg_uri_safe)
        .status()
        .await
        .map_err(|e| {
            if e.kind() == std::io::ErrorKind::NotFound {
                tracing::info!("pg_dump binary could not be resolved by system.");
                "pg_dump binary could not be resolved (PATH/env/download). Set ATHENA_PG_DUMP_PATH to override."
                    .to_string()
            } else {
                format!("Failed to start pg_dump: {e}")
            }
        })?;

    if !status.success() {
        tracing::info!("pg_dump failed with status {:?}", status);
        let _ = tokio::fs::remove_dir_all(&tmp_root).await;
        return Err(format!("pg_dump exited with status {status}"));
    }

    tracing::info!("pg_dump completed successfully, archiving result.");

    // Archive the dump directory into a tar.gz.
    let dump_dir_clone = dump_dir.clone();
    let archive_path_clone = archive_path.clone();
    tokio::task::spawn_blocking(move || -> Result<(), String> {
        let file = std::fs::File::create(&archive_path_clone)
            .map_err(|e| format!("Cannot create archive: {e}"))?;
        let enc = flate2::write::GzEncoder::new(file, flate2::Compression::default());
        let mut builder = tar::Builder::new(enc);
        builder
            .append_dir_all("dump", &dump_dir_clone)
            .map_err(|e| format!("Cannot archive dump directory: {e}"))?;
        builder
            .finish()
            .map_err(|e| format!("Cannot finalize archive: {e}"))?;
        Ok(())
    })
    .await
    .map_err(|e| format!("Archive task panicked: {e}"))??;

    tracing::info!(
        "pg_dump directory archived, removing uncompressed dump dir {:?}",
        dump_dir
    );

    // Remove uncompressed dump dir, keep only the archive.
    let _ = tokio::fs::remove_dir_all(&dump_dir).await;

    tracing::info!("Backup archive produced at {:?}", archive_path);

    Ok(archive_path)
}

// ── logging helpers ──────────────────────────────────────────────────────────

#[derive(Clone)]
struct JobLogger {
    pool: PgPool,
    job_id: i64,
}

impl JobLogger {
    async fn progress(
        &self,
        status: Option<&str>,
        stage: Option<&str>,
        progress_pct: Option<i32>,
        s3_bucket: Option<&str>,
        s3_key: Option<&str>,
        size_bytes: Option<i64>,
        error_message: Option<&str>,
        completed_at: Option<chrono::DateTime<Utc>>,
    ) {
        if let Err(err) = update_job_progress(
            &self.pool,
            self.job_id,
            status,
            stage,
            progress_pct,
            s3_bucket,
            s3_key,
            size_bytes,
            error_message,
            completed_at,
        )
        .await
        {
            tracing::warn!(job_id = self.job_id, error = %err, "Failed to update backup job progress");
        }
    }

    async fn log(&self, level: &str, message: &str) {
        if let Err(err) =
            sqlx::query("INSERT INTO backup_job_logs (job_id, level, message) VALUES ($1, $2, $3)")
                .bind(self.job_id)
                .bind(level)
                .bind(message)
                .execute(&self.pool)
                .await
        {
            tracing::warn!(job_id = self.job_id, error = %err, "Failed to insert backup job log");
        }
    }
}

async fn logging_pool(state: &AppState) -> Result<PgPool, HttpResponse> {
    let Some(client_name) = state.logging_client_name.as_ref() else {
        return Err(service_unavailable(
            "Logging unavailable",
            "No athena_logging client is configured.",
        ));
    };

    state.pg_registry.get_pool(client_name).ok_or_else(|| {
        service_unavailable(
            "Logging unavailable",
            format!("Logging client '{}' is not connected.", client_name),
        )
    })
}

async fn create_backup_job(
    pool: &PgPool,
    job_type: &str,
    client_name: &str,
    initial_stage: &str,
    label: Option<&str>,
) -> Result<i64, sqlx::Error> {
    let row = sqlx::query(
        r#"
        INSERT INTO backup_jobs (job_type, client_name, status, progress_stage, label, started_at, updated_at)
        VALUES ($1, $2, 'running', $3, $4, now(), now())
        RETURNING id
        "#,
    )
    .bind(job_type)
    .bind(client_name)
    .bind(initial_stage)
    .bind(label)
    .fetch_one(pool)
    .await?;

    Ok(row.get::<i64, _>("id"))
}

async fn update_job_progress(
    pool: &PgPool,
    job_id: i64,
    status: Option<&str>,
    stage: Option<&str>,
    progress_pct: Option<i32>,
    s3_bucket: Option<&str>,
    s3_key: Option<&str>,
    size_bytes: Option<i64>,
    error_message: Option<&str>,
    completed_at: Option<chrono::DateTime<Utc>>,
) -> Result<(), sqlx::Error> {
    sqlx::query(
        r#"
        UPDATE backup_jobs
        SET status = COALESCE($2, status),
            progress_stage = COALESCE($3, progress_stage),
            progress_pct = COALESCE($4, progress_pct),
            s3_bucket = COALESCE($5, s3_bucket),
            s3_key = COALESCE($6, s3_key),
            size_bytes = COALESCE($7, size_bytes),
            error_message = COALESCE($8, error_message),
            completed_at = COALESCE($9, completed_at),
            updated_at = now()
        WHERE id = $1
        "#,
    )
    .bind(job_id)
    .bind(status)
    .bind(stage)
    .bind(progress_pct)
    .bind(s3_bucket)
    .bind(s3_key)
    .bind(size_bytes)
    .bind(error_message)
    .bind(completed_at)
    .execute(pool)
    .await?;

    Ok(())
}

/// Download `key` from S3, extract to a temp directory, run `pg_restore`.
async fn run_pg_restore(
    s3_client: &S3Client,
    bucket: &str,
    key: &str,
    pg_uri: &str,
    job: Option<JobLogger>,
) -> Result<(), String> {
    tracing::info!("Starting pg_restore from S3 bucket={} key={}", bucket, key);

    if let Some(logger) = &job {
        logger
            .progress(
                Some("running"),
                Some("downloading"),
                Some(10),
                Some(bucket),
                Some(key),
                None,
                None,
                None,
            )
            .await;
        logger
            .log(
                "info",
                "Starting restore: downloading backup from object storage",
            )
            .await;
    }

    // Download the backup archive.
    let resp: aws_sdk_s3::operation::get_object::GetObjectOutput = s3_client
        .get_object()
        .bucket(bucket)
        .key(key)
        .send()
        .await
        .map_err(|e| {
            tracing::info!("S3 download failed for key {}: {e}", key, e = &e);
            format!("S3 download failed: {e}")
        })?;

    let bytes: web::Bytes = resp
        .body
        .collect()
        .await
        .map_err(|e| {
            tracing::info!("S3 read failed for key {}: {e}", key, e = &e);
            format!("S3 read failed: {e}")
        })?
        .into_bytes();

    let size_bytes: i64 = bytes.len() as i64;

    if let Some(logger) = &job {
        logger
            .progress(
                None,
                Some("writing"),
                Some(25),
                Some(bucket),
                Some(key),
                Some(size_bytes),
                None,
                None,
            )
            .await;
    }

    let tmp_root: PathBuf = env::temp_dir().join(format!("athena_restore_{}", Uuid::new_v4()));
    tracing::info!("Creating temp directory for restore: {:?}", &tmp_root);
    tokio::fs::create_dir_all(&tmp_root)
        .await
        .map_err(|e| format!("Could not create temp dir: {e}"))?;

    let archive_path: PathBuf = tmp_root.join("backup.tar.gz");
    let restore_dir: PathBuf = tmp_root.join("dump");

    tracing::info!("Writing downloaded archive to {:?}", &archive_path);
    tokio::fs::write(&archive_path, &bytes)
        .await
        .map_err(|e| format!("Could not write archive: {e}"))?;

    if let Some(logger) = &job {
        logger
            .progress(
                None,
                Some("extracting"),
                Some(50),
                Some(bucket),
                Some(key),
                Some(size_bytes),
                None,
                None,
            )
            .await;
        logger
            .log("info", "Downloaded backup, extracting archive")
            .await;
    }

    // Extract the tar.gz.
    tracing::info!("Extracting backup archive for restore.");
    let archive_path_clone: PathBuf = archive_path.clone();
    let restore_dir_clone: PathBuf = restore_dir.clone();
    tokio::task::spawn_blocking(move || -> Result<(), String> {
        let file = std::fs::File::open(&archive_path_clone)
            .map_err(|e| format!("Cannot open archive: {e}"))?;
        let dec = flate2::read::GzDecoder::new(file);
        let mut archive: tar::Archive<flate2::read::GzDecoder<std::fs::File>> =
            tar::Archive::new(dec);
        archive
            .unpack(&restore_dir_clone)
            .map_err(|e| format!("Cannot extract archive: {e}"))?;
        Ok(())
    })
    .await
    .map_err(|e| format!("Extract task panicked: {e}"))??;

    // The pg_dump directory format stores files under the name we passed to
    // append_dir_all ("dump"), so the actual dump content is at restore_dir/dump.
    let inner_dump_dir: PathBuf = restore_dir.join("dump");
    tracing::info!("Restore will use backup contents at {:?}", inner_dump_dir);

    // Strip password from URI and pass via PGPASSWORD.
    let (pg_uri_safe, pg_password) = extract_pg_password(pg_uri);

    let pg_tools: PgToolsPaths = ensure_pg_tools().await.map_err(|e| {
        tracing::info!("pg_restore tool resolution failed: {e}");
        format!("pg_restore resolution failed: {e}")
    })?;

    tracing::info!(
        "Launching pg_restore for database restore, using pg_restore at {:?}",
        &pg_tools.pg_restore
    );

    if let Some(logger) = &job {
        logger
            .progress(
                None,
                Some("pg_restore"),
                Some(80),
                Some(bucket),
                Some(key),
                Some(size_bytes),
                None,
                None,
            )
            .await;
        logger
            .log("info", "Running pg_restore against target database")
            .await;
    }

    let mut cmd: Command = Command::new(&pg_tools.pg_restore);
    if let Some(pass) = pg_password {
        cmd.env("PGPASSWORD", pass);
    }
    let status: std::process::ExitStatus = cmd
        .args(["--format=directory", "--clean", "--if-exists", "--dbname"])
        .arg(&pg_uri_safe)
        .arg(&inner_dump_dir)
        .status()
        .await
        .map_err(|e| {
            if e.kind() == std::io::ErrorKind::NotFound {
                tracing::info!("pg_restore binary not found on system.");
                "pg_restore binary not found in PATH — ensure PostgreSQL client tools are installed"
                    .to_string()
            } else {
                format!("Failed to start pg_restore: {e}")
            }
        })?;

    let _ = tokio::fs::remove_dir_all(&tmp_root).await;
    tracing::info!("Cleanup: removed temp restore directory {:?}", &tmp_root);

    if !status.success() {
        tracing::info!("pg_restore finished with error status: {:?}", status);
        return Err(format!("pg_restore exited with status {status}"));
    }

    tracing::info!("pg_restore completed successfully!");
    Ok(())
}

/// Upload a file to S3 and return its object key.
async fn upload_to_s3(
    s3_client: &S3Client,
    cfg: &S3Config,
    local_path: &PathBuf,
    client_name: &str,
    label: Option<&str>,
) -> Result<(String, i64), String> {
    let backup_id: String = Uuid::new_v4().to_string();
    let key: String = format!("{}/{}/{}.tar.gz", cfg.prefix, client_name, backup_id);

    tracing::info!("Reading local backup archive from {:?}", local_path);
    let data: Vec<u8> = tokio::fs::read(local_path)
        .await
        .map_err(|e| format!("Cannot read archive file: {e}"))?;
    let size_bytes: i64 = data.len() as i64;

    tracing::info!(
        "Putting object to S3: bucket='{}', key='{}', client_name='{}', label={:?}",
        &cfg.bucket,
        &key,
        client_name,
        label
    );

    let mut req: aws_sdk_s3::operation::put_object::builders::PutObjectFluentBuilder = s3_client
        .put_object()
        .bucket(&cfg.bucket)
        .key(&key)
        .body(data.into())
        .content_type("application/gzip")
        .metadata("client_name", client_name)
        .metadata("backup_id", &backup_id)
        .metadata("created_at", Utc::now().to_rfc3339());

    if let Some(lbl) = label {
        req = req.metadata("label", lbl);
    }

    req.send().await.map_err(|e| {
        tracing::info!("S3 upload to key '{}' failed: {e}", key, e = &e);
        format!("S3 upload failed: {e}")
    })?;

    tracing::info!("S3 backup upload complete for key {}", key);
    Ok((key, size_bytes))
}

// ── handlers ─────────────────────────────────────────────────────────────────

/// Create a database backup via `pg_dump` and upload it to S3.
///
/// # Request body
/// ```json
/// { "client_name": "my_db", "label": "pre-deploy" }
/// ```
#[post("/admin/backups")]
pub async fn admin_create_backup(
    req: HttpRequest,
    state: Data<AppState>,
    body: Json<CreateBackupRequest>,
) -> HttpResponse {
    tracing::info!(
        "admin_create_backup called. client_name: {:?}, label: {:?}",
        body.client_name,
        body.label
    );

    if let Err(resp) = authorize_static_admin_key(&req) {
        tracing::info!("Authorization failed for admin_create_backup");
        return resp;
    }

    let logging_pool = match logging_pool(&state).await {
        Ok(pool) => pool,
        Err(resp) => return resp,
    };

    let Some(s3_cfg) = S3Config::from_env() else {
        tracing::info!("S3Config could not be constructed from env for admin_create_backup");
        return service_unavailable(
            "S3 not configured",
            "Set ATHENA_BACKUP_S3_BUCKET and related environment variables to enable backups.",
        );
    };

    let pg_uri = match resolve_pg_uri(&state, &body.client_name) {
        Ok(uri) => uri,
        Err(resp) => {
            tracing::info!(
                "Could not resolve pg_uri for client_name='{}'",
                body.client_name
            );
            return resp;
        }
    };

    let job_id = match create_backup_job(
        &logging_pool,
        "backup",
        &body.client_name,
        "pg_dump",
        body.label.as_deref(),
    )
    .await
    {
        Ok(id) => id,
        Err(err) => {
            tracing::warn!("Failed to create backup_job row: {}", err);
            return internal_error("Logging unavailable", "Could not create backup job record");
        }
    };
    let job_logger = JobLogger {
        pool: logging_pool.clone(),
        job_id,
    };
    job_logger
        .log("info", "Starting backup job and running pg_dump")
        .await;

    tracing::info!("Running pg_dump for client_name='{}'", body.client_name);

    // Run pg_dump.
    let archive_path = match run_pg_dump(&pg_uri).await {
        Ok(p) => {
            tracing::info!("pg_dump completed. Archive at {:?}", p);
            job_logger
                .progress(
                    None,
                    Some("archiving"),
                    Some(40),
                    None,
                    None,
                    None,
                    None,
                    None,
                )
                .await;
            job_logger
                .log("info", "pg_dump completed, archiving dump")
                .await;
            p
        }
        Err(err) => {
            tracing::info!("pg_dump failed: {}", err);
            job_logger
                .progress(
                    Some("failed"),
                    Some("pg_dump"),
                    Some(0),
                    None,
                    None,
                    None,
                    Some(&err),
                    Some(Utc::now()),
                )
                .await;
            job_logger
                .log("error", &format!("pg_dump failed: {}", err))
                .await;
            return internal_error("pg_dump failed", err);
        }
    };

    // Upload to S3.
    let s3_client = build_s3_client(&s3_cfg).await;
    tracing::info!("Uploading archive to S3...");
    job_logger
        .progress(
            None,
            Some("uploading"),
            Some(70),
            Some(&s3_cfg.bucket),
            None,
            None,
            None,
            None,
        )
        .await;
    let (key, size_bytes) = match upload_to_s3(
        &s3_client,
        &s3_cfg,
        &archive_path,
        &body.client_name,
        body.label.as_deref(),
    )
    .await
    {
        Ok((key, size_bytes)) => {
            tracing::info!("S3 upload succeeded. backup key: {}", key);
            job_logger
                .progress(
                    None,
                    Some("uploading"),
                    Some(90),
                    Some(&s3_cfg.bucket),
                    Some(&key),
                    Some(size_bytes),
                    None,
                    None,
                )
                .await;
            (key, size_bytes)
        }
        Err(err) => {
            tracing::info!("S3 upload failed: {}", err);
            let _ = tokio::fs::remove_file(&archive_path).await;
            job_logger
                .progress(
                    Some("failed"),
                    Some("uploading"),
                    Some(70),
                    Some(&s3_cfg.bucket),
                    None,
                    None,
                    Some(&err),
                    Some(Utc::now()),
                )
                .await;
            job_logger
                .log("error", &format!("S3 upload failed: {}", err))
                .await;
            return internal_error("S3 upload failed", err);
        }
    };

    // Clean up local archive.
    if let Some(parent) = archive_path.parent() {
        tracing::info!("Cleaning up archive directory: {:?}", parent);
        let _ = tokio::fs::remove_dir_all(parent).await;
    }

    tracing::info!("admin_create_backup successful for key {}", key);
    job_logger
        .progress(
            Some("completed"),
            Some("completed"),
            Some(100),
            Some(&s3_cfg.bucket),
            Some(&key),
            Some(size_bytes),
            None,
            Some(Utc::now()),
        )
        .await;
    job_logger
        .log("info", "Backup completed and stored in object storage")
        .await;

    api_success(
        "Backup created",
        json!({
            "job_id": job_id,
            "key": key,
            "client_name": body.client_name,
            "label": body.label,
        }),
    )
}

/// Return the configured S3 bucket/region/prefix (no credentials).
#[get("/admin/backups/config")]
pub async fn admin_backup_config(req: HttpRequest) -> HttpResponse {
    tracing::info!("admin_backup_config called");

    if let Err(resp) = authorize_static_admin_key(&req) {
        return resp;
    }

    let Some(cfg) = S3Config::from_env() else {
        return service_unavailable(
            "S3 not configured",
            "Set ATHENA_BACKUP_S3_BUCKET and related environment variables to enable backups.",
        );
    };

    api_success(
        "Backup storage configuration",
        json!({
            "bucket": cfg.bucket,
            "region": cfg.region,
            "prefix": cfg.prefix,
            "endpoint": cfg.endpoint,
        }),
    )
}

/// List backup objects stored in S3.
///
/// Optionally filter by `?client_name=…`.
#[get("/admin/backups")]
pub async fn admin_list_backups(
    req: HttpRequest,
    _state: Data<AppState>,
    query: web::Query<std::collections::HashMap<String, String>>,
) -> HttpResponse {
    tracing::info!("admin_list_backups called with query: {:?}", query);

    if let Err(resp) = authorize_static_admin_key(&req) {
        tracing::info!("Authorization failed for admin_list_backups");
        return resp;
    }

    let Some(s3_cfg) = S3Config::from_env() else {
        tracing::info!("S3Config could not be constructed from env for admin_list_backups");
        return service_unavailable(
            "S3 not configured",
            "Set ATHENA_BACKUP_S3_BUCKET and related environment variables to enable backups.",
        );
    };

    let s3_client = build_s3_client(&s3_cfg).await;

    let filter_client = query.get("client_name").cloned();
    let prefix = match &filter_client {
        Some(cn) => format!("{}/{}/", s3_cfg.prefix, cn),
        None => format!("{}/", s3_cfg.prefix),
    };

    tracing::info!("Listing S3 backups with prefix: {}", prefix);

    let resp = match s3_client
        .list_objects_v2()
        .bucket(&s3_cfg.bucket)
        .prefix(&prefix)
        .send()
        .await
    {
        Ok(r) => r,
        Err(err) => {
            tracing::info!("Failed to list S3 objects: {}", err);
            return internal_error("Failed to list S3 objects", err.to_string());
        }
    };

    let mut backups: Vec<BackupObject> = Vec::new();

    for obj in resp.contents() {
        let key = obj.key().unwrap_or_default().to_string();

        // Expected key format: {prefix}/{client_name}/{backup_id}.tar.gz
        let parts: Vec<&str> = key.split('/').collect();
        let (client_name, id) = if parts.len() >= 3 {
            let cn = parts[parts.len() - 2].to_string();
            let id = parts
                .last()
                .and_then(|s| s.strip_suffix(".tar.gz"))
                .unwrap_or_else(|| parts.last().copied().unwrap_or(""))
                .to_string();
            (cn, id)
        } else {
            // Unexpected key format; still include the object but log a warning.
            tracing::warn!(key = %key, "S3 backup key does not match expected format <prefix>/<client_name>/<id>.tar.gz");
            (
                "unknown".to_string(),
                parts
                    .last()
                    .and_then(|s| s.strip_suffix(".tar.gz"))
                    .unwrap_or_else(|| parts.last().copied().unwrap_or(&key))
                    .to_string(),
            )
        };

        tracing::info!("Fetching S3 object metadata for label for key {}", key);

        // Fetch object metadata for label.
        let label = match s3_client
            .head_object()
            .bucket(&s3_cfg.bucket)
            .key(&key)
            .send()
            .await
        {
            Ok(head) => head.metadata().and_then(|m| m.get("label")).cloned(),
            Err(_) => None,
        };

        let size_bytes = obj.size().unwrap_or(0);
        let created_at = obj
            .last_modified()
            .map(|t| t.to_string())
            .unwrap_or_default();

        backups.push(BackupObject {
            id,
            key,
            client_name,
            label,
            size_bytes,
            created_at,
        });
    }

    // Most-recent first.
    backups.sort_by(|a, b| b.created_at.cmp(&a.created_at));

    tracing::info!("Returning {} backup(s) from S3 list.", backups.len());

    api_success("Listed backups", json!({ "backups": backups }))
}

/// List backup/restore jobs recorded in athena_logging.
#[get("/admin/backups/jobs")]
pub async fn admin_list_backup_jobs(
    req: HttpRequest,
    state: Data<AppState>,
    query: web::Query<std::collections::HashMap<String, String>>,
) -> HttpResponse {
    tracing::info!("admin_list_backup_jobs called with query: {:?}", query);

    if let Err(resp) = authorize_static_admin_key(&req) {
        return resp;
    }

    let pool = match logging_pool(&state).await {
        Ok(pool) => pool,
        Err(resp) => return resp,
    };

    let limit: i64 = query
        .get("limit")
        .and_then(|v| v.parse::<i64>().ok())
        .filter(|v| *v > 0 && *v <= 500)
        .unwrap_or(50);
    let client_filter = query.get("client_name").cloned();
    let status_filter = query.get("status").cloned();
    let job_type_filter = query.get("job_type").cloned();

    let mut qb = QueryBuilder::new(
        r#"
        SELECT id, job_type, client_name, status, progress_pct, progress_stage, s3_bucket, s3_key, label, size_bytes, error_message, started_at, updated_at, completed_at
        FROM backup_jobs
        WHERE 1=1
        "#,
    );

    if let Some(c) = client_filter {
        qb.push(" AND client_name = ").push_bind(c);
    }
    if let Some(s) = status_filter {
        qb.push(" AND status = ").push_bind(s);
    }
    if let Some(j) = job_type_filter {
        qb.push(" AND job_type = ").push_bind(j);
    }

    qb.push(" ORDER BY started_at DESC LIMIT ").push_bind(limit);

    let jobs: Vec<BackupJobSummary> = match qb
        .build_query_as::<BackupJobSummary>()
        .fetch_all(&pool)
        .await
    {
        Ok(rows) => rows,
        Err(err) => {
            tracing::warn!("Failed to list backup jobs: {}", err);
            return internal_error("Failed to list backup jobs", err.to_string());
        }
    };

    api_success("Listed backup jobs", json!({ "jobs": jobs }))
}

/// Return a backup/restore job with recent logs.
#[get("/admin/backups/jobs/{id}")]
pub async fn admin_get_backup_job(
    req: HttpRequest,
    state: Data<AppState>,
    job_id: Path<i64>,
) -> HttpResponse {
    tracing::info!("admin_get_backup_job called: job_id={}", job_id);

    if let Err(resp) = authorize_static_admin_key(&req) {
        return resp;
    }

    let pool = match logging_pool(&state).await {
        Ok(pool) => pool,
        Err(resp) => return resp,
    };

    let job = match sqlx::query_as::<_, BackupJobSummary>(
        r#"
        SELECT id, job_type, client_name, status, progress_pct, progress_stage, s3_bucket, s3_key, label, size_bytes, error_message, started_at, updated_at, completed_at
        FROM backup_jobs
        WHERE id = $1
        "#,
    )
    .bind(*job_id)
    .fetch_optional(&pool)
    .await
    {
        Ok(Some(row)) => row,
        Ok(None) => {
            return HttpResponse::NotFound().json(json!({
                "status": "error",
                "message": format!("Backup job {} not found", *job_id)
            }));
        }
        Err(err) => {
            tracing::warn!("Failed to fetch backup job {}: {}", *job_id, err);
            return internal_error("Failed to fetch backup job", err.to_string());
        }
    };

    let logs: Vec<BackupJobLog> = match sqlx::query_as::<_, BackupJobLog>(
        r#"
        SELECT id, job_id, level, message, created_at
        FROM backup_job_logs
        WHERE job_id = $1
        ORDER BY created_at DESC
        LIMIT 100
        "#,
    )
    .bind(*job_id)
    .fetch_all(&pool)
    .await
    {
        Ok(rows) => rows,
        Err(err) => {
            tracing::warn!("Failed to fetch backup job logs for {}: {}", *job_id, err);
            return internal_error("Failed to fetch backup job logs", err.to_string());
        }
    };

    api_success("Backup job", json!({ "job": job, "logs": logs }))
}

/// Restore a database from a backup stored in S3 via `pg_restore`.
///
/// `{key:.*}` is the S3 object key (may contain slashes).
///
/// # Request body
/// ```json
/// { "client_name": "my_db" }
/// ```
#[post("/admin/backups/{key:.*}/restore")]
pub async fn admin_restore_backup(
    req: HttpRequest,
    state: Data<AppState>,
    key_param: Path<String>,
    body: Json<RestoreBackupRequest>,
) -> HttpResponse {
    tracing::info!(
        "admin_restore_backup called, client_name={}, key_param={}",
        body.client_name,
        key_param
    );

    if let Err(resp) = authorize_static_admin_key(&req) {
        tracing::info!("Authorization failed for admin_restore_backup");
        return resp;
    }

    let logging_pool = match logging_pool(&state).await {
        Ok(pool) => pool,
        Err(resp) => return resp,
    };

    let Some(s3_cfg) = S3Config::from_env() else {
        tracing::info!("S3Config could not be constructed from env for admin_restore_backup");
        return service_unavailable(
            "S3 not configured",
            "Set ATHENA_BACKUP_S3_BUCKET and related environment variables to enable backups.",
        );
    };

    let pg_uri = match resolve_pg_uri(&state, &body.client_name) {
        Ok(uri) => uri,
        Err(resp) => {
            tracing::info!(
                "Could not resolve pg_uri for client_name='{}'",
                body.client_name
            );
            return resp;
        }
    };

    // The key_param path segment contains the S3 object key (URL-decoded by Actix).
    let key = key_param.into_inner();
    if key.is_empty() {
        tracing::info!("No backup key provided to admin_restore_backup");
        return bad_request(
            "Missing backup key",
            "Provide the S3 object key as the path segment.",
        );
    }

    let s3_client = build_s3_client(&s3_cfg).await;

    let job_id = match create_backup_job(
        &logging_pool,
        "restore",
        &body.client_name,
        "downloading",
        None,
    )
    .await
    {
        Ok(id) => id,
        Err(err) => {
            tracing::warn!("Failed to create restore job row: {}", err);
            return internal_error("Logging unavailable", "Could not create restore job record");
        }
    };
    let job_logger = JobLogger {
        pool: logging_pool.clone(),
        job_id,
    };
    job_logger
        .log("info", "Starting restore job, downloading backup")
        .await;

    tracing::info!(
        "Calling run_pg_restore for key='{}', client_name='{}'",
        key,
        body.client_name
    );
    match run_pg_restore(
        &s3_client,
        &s3_cfg.bucket,
        &key,
        &pg_uri,
        Some(job_logger.clone()),
    )
    .await
    {
        Ok(()) => {
            tracing::info!("Restore succeeded for key='{}'", key);
            job_logger
                .progress(
                    Some("completed"),
                    Some("completed"),
                    Some(100),
                    Some(&s3_cfg.bucket),
                    Some(&key),
                    None,
                    None,
                    Some(Utc::now()),
                )
                .await;
            job_logger
                .log("info", "Restore completed successfully")
                .await;
            api_success(
                "Restore completed",
                json!({ "key": key, "client_name": body.client_name, "job_id": job_id }),
            )
        }
        Err(err) => {
            tracing::info!("Restore failed for key='{}': {}", key, err);
            job_logger
                .progress(
                    Some("failed"),
                    Some("pg_restore"),
                    Some(90),
                    Some(&s3_cfg.bucket),
                    Some(&key),
                    None,
                    Some(&err),
                    Some(Utc::now()),
                )
                .await;
            job_logger
                .log("error", &format!("Restore failed: {}", err))
                .await;
            internal_error("pg_restore failed", err)
        }
    }
}

/// Download a backup archive directly from S3.
///
/// `{key:.*}` is the S3 object key (may contain slashes).  Returns the raw
/// `application/gzip` bytes with a `Content-Disposition: attachment` header.
#[get("/admin/backups/{key:.*}/download")]
pub async fn admin_download_backup(
    req: HttpRequest,
    _state: Data<AppState>,
    key_param: Path<String>,
) -> HttpResponse {
    tracing::info!("admin_download_backup called: key_param={}", key_param);

    if let Err(resp) = authorize_static_admin_key(&req) {
        tracing::info!("Authorization failed for admin_download_backup");
        return resp;
    }

    let Some(s3_cfg) = S3Config::from_env() else {
        tracing::info!("S3Config could not be constructed from env for admin_download_backup");
        return service_unavailable(
            "S3 not configured",
            "Set ATHENA_BACKUP_S3_BUCKET and related environment variables to enable backups.",
        );
    };

    let key = key_param.into_inner();
    if key.is_empty() {
        tracing::info!("No backup key provided to admin_download_backup");
        return bad_request(
            "Missing backup key",
            "Provide the S3 object key as the path segment.",
        );
    }

    let s3_client = build_s3_client(&s3_cfg).await;

    tracing::info!("Requesting backup archive from S3 for key='{}'", key);

    let resp = match s3_client
        .get_object()
        .bucket(&s3_cfg.bucket)
        .key(&key)
        .send()
        .await
    {
        Ok(r) => r,
        Err(err) => {
            tracing::info!("S3 download failed for key '{}': {}", key, err);
            return internal_error("S3 download failed", err.to_string());
        }
    };

    let bytes = match resp.body.collect().await {
        Ok(b) => b.into_bytes(),
        Err(err) => {
            tracing::info!("S3 read failed for key '{}': {}", key, err);
            return internal_error("S3 read failed", err.to_string());
        }
    };

    let filename = key
        .rsplit('/')
        .next()
        .unwrap_or("backup.tar.gz")
        .to_string();

    tracing::info!(
        "Download backup returning S3 object for key='{}', filename='{}'",
        key,
        filename
    );

    HttpResponse::Ok()
        .content_type("application/gzip")
        .insert_header((
            "Content-Disposition",
            format!("attachment; filename=\"{}\"", filename),
        ))
        .body(bytes.to_vec())
}

/// Delete a backup from S3.
///
/// `{key:.*}` is the S3 object key (may contain slashes).
#[delete("/admin/backups/{key:.*}")]
pub async fn admin_delete_backup(
    req: HttpRequest,
    _state: Data<AppState>,
    key_param: Path<String>,
) -> HttpResponse {
    tracing::info!("admin_delete_backup called: key_param={}", key_param);

    if let Err(resp) = authorize_static_admin_key(&req) {
        tracing::info!("Authorization failed for admin_delete_backup");
        return resp;
    }

    let Some(s3_cfg) = S3Config::from_env() else {
        tracing::info!("S3Config could not be constructed from env for admin_delete_backup");
        return service_unavailable(
            "S3 not configured",
            "Set ATHENA_BACKUP_S3_BUCKET and related environment variables to enable backups.",
        );
    };

    // The key_param path segment contains the S3 object key (URL-decoded by Actix).
    let key = key_param.into_inner();
    if key.is_empty() {
        tracing::info!("No backup key provided to admin_delete_backup");
        return bad_request(
            "Missing backup key",
            "Provide the S3 object key as the path segment.",
        );
    }

    let s3_client: S3Client = build_s3_client(&s3_cfg).await;

    tracing::info!("Deleting S3 object for key='{}'", key);

    match s3_client
        .delete_object()
        .bucket(&s3_cfg.bucket)
        .key(&key)
        .send()
        .await
    {
        Ok(_) => {
            tracing::info!("Successfully deleted S3 backup for key '{}'", key);
            api_success("Backup deleted", json!({ "key": key }))
        }
        Err(err) => {
            tracing::info!("S3 delete failed for key '{}': {}", key, err);
            internal_error("S3 delete failed", err.to_string())
        }
    }
}

pub fn services(cfg: &mut web::ServiceConfig) {
    tracing::info!("Registering backup service routes.");
    cfg.service(admin_create_backup)
        .service(admin_backup_config)
        .service(admin_list_backups)
        .service(admin_list_backup_jobs)
        .service(admin_get_backup_job)
        .service(admin_download_backup)
        .service(admin_restore_backup)
        .service(admin_delete_backup);
}