athena_rs 3.18.0

use serde_json::Value;
use sqlx::postgres::PgPoolOptions;
use std::net::TcpListener;
use std::process::Output;
use std::time::{Duration, Instant};
use tokio::process::Command;
use tokio::time::sleep;
use uuid::Uuid;

use crate::config::Config;
use std::collections::HashMap;

use super::constants::{
    DEFAULT_INSTANCE_HOST, DEFAULT_POSTGRES_IMAGE, DEFAULT_STARTUP_TIMEOUT_SECS,
};
use super::error::ProvisioningError;
use super::types::{
    DockerContainerStatus, DockerManagedContainer, DockerRuntimeAccessStatus, ResolvedSpinUpConfig,
    SpinUpPostgresParams, SpinUpPostgresResult,
};
use crate::utils::sqlx_postgres_connect_uri::sanitize_sqlx_postgres_connect_uri;

const DOCKER_RUNTIME_PROBE_ARGS: &[&str] =
    &["ps", "-a", "-q", "--filter", "label=athena.managed=true"];

/// Spin up or reuse a Docker-backed Postgres instance and return a routable URI.
pub async fn spin_up_postgres_instance(
    params: SpinUpPostgresParams,
) -> Result<SpinUpPostgresResult, ProvisioningError> {
    let resolved: ResolvedSpinUpConfig = resolve_spin_up_config(params)?;
    let mut created_new_container: bool = false;
    let mut reused_existing_container: bool = false;

    let exists: bool = docker_container_exists(&resolved.container_name).await?;
    if exists {
        if !resolved.reuse_existing {
            return Err(ProvisioningError::Conflict(format!(
                "container '{}' already exists; set reuse_existing=true to reuse it",
                resolved.container_name
            )));
        }
        start_container(&resolved.container_name).await?;
        reused_existing_container = true;
    } else {
        run_postgres_container(&resolved).await?;
        created_new_container = true;
    }

    let status: DockerContainerStatus = inspect_container(&resolved.container_name).await?;
    let discovered_port: u16 = status.host_port.unwrap_or(resolved.host_port);
    let pg_uri: String = build_pg_uri(
        &resolved.username,
        &resolved.password,
        &resolved.host,
        discovered_port,
        &resolved.db_name,
    );
    let wait_ready_ms: u128 =
        wait_for_postgres_ready(&pg_uri, resolved.startup_timeout_secs).await?;

    Ok(SpinUpPostgresResult {
        client_name: resolved.client_name,
        container_name: resolved.container_name,
        image: resolved.image,
        host: resolved.host,
        host_port: discovered_port,
        db_name: resolved.db_name,
        username: resolved.username,
        password: resolved.password,
        pg_uri,
        created_new_container,
        reused_existing_container,
        wait_ready_ms,
    })
}

/// Inspect Docker metadata for a container.
pub async fn inspect_container(
    container_name: &str,
) -> Result<DockerContainerStatus, ProvisioningError> {
    validate_container_name(container_name)?;
    let output: Output = run_docker_capture(&["inspect", container_name]).await?;
    if !output.status.success() {
        let stderr = String::from_utf8_lossy(&output.stderr).to_string();
        if stderr.to_lowercase().contains("no such object") {
            return Ok(DockerContainerStatus {
                container_name: container_name.to_string(),
                exists: false,
                running: false,
                status: None,
                image: None,
                host_port: None,
            });
        }
        return Err(ProvisioningError::Execution(format!(
            "docker inspect failed for '{}': {}",
            container_name, stderr
        )));
    }

    let entries: Vec<Value> = serde_json::from_slice(&output.stdout).map_err(|err| {
        ProvisioningError::Execution(format!(
            "failed to parse docker inspect output for '{}': {}",
            container_name, err
        ))
    })?;

    let Some(entry) = entries.first() else {
        return Ok(DockerContainerStatus {
            container_name: container_name.to_string(),
            exists: false,
            running: false,
            status: None,
            image: None,
            host_port: None,
        });
    };

    let running: bool = entry
        .pointer("/State/Running")
        .and_then(Value::as_bool)
        .unwrap_or(false);
    let status: Option<String> = entry
        .pointer("/State/Status")
        .and_then(Value::as_str)
        .map(str::to_string);
    let image: Option<String> = entry
        .pointer("/Config/Image")
        .and_then(Value::as_str)
        .map(str::to_string);
    let host_port: Option<u16> = entry
        .pointer("/NetworkSettings/Ports/5432/tcp")
        .and_then(Value::as_array)
        .and_then(|items| items.first())
        .and_then(|item| item.get("HostPort"))
        .and_then(Value::as_str)
        .and_then(|value| value.parse::<u16>().ok());

    Ok(DockerContainerStatus {
        container_name: container_name.to_string(),
        exists: true,
        running,
        status,
        image,
        host_port,
    })
}

/// List all Docker containers managed by Athena (`athena.managed=true` label).
pub async fn list_managed_postgres_containers()
-> Result<Vec<DockerManagedContainer>, ProvisioningError> {
    let ids_output: Output = run_docker_capture(DOCKER_RUNTIME_PROBE_ARGS).await?;
    if !ids_output.status.success() {
        return Err(map_docker_failed_status(
            &["ps", "-a", "-q", "--filter", "label=athena.managed=true"],
            &ids_output,
        ));
    }

    let ids_raw: std::borrow::Cow<'_, str> = String::from_utf8_lossy(&ids_output.stdout);
    let ids: Vec<String> = ids_raw
        .lines()
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .map(str::to_string)
        .collect();
    if ids.is_empty() {
        return Ok(Vec::new());
    }

    let mut inspect_args: Vec<String> = vec!["inspect".to_string()];
    inspect_args.extend(ids);
    let inspect_output: Output = run_docker_capture_strings(&inspect_args).await?;
    if !inspect_output.status.success() {
        return Err(map_docker_failed_status(
            &["inspect", "<managed_containers>"],
            &inspect_output,
        ));
    }

    let entries: Vec<Value> = serde_json::from_slice(&inspect_output.stdout).map_err(|err| {
        ProvisioningError::Execution(format!(
            "failed to parse docker inspect output for managed containers: {}",
            err
        ))
    })?;

    let mut containers: Vec<DockerManagedContainer> = entries
        .into_iter()
        .map(|entry| {
            let container_name: String = entry
                .get("Name")
                .and_then(Value::as_str)
                .map(|value| value.trim_start_matches('/').to_string())
                .unwrap_or_default();
            let running: bool = entry
                .pointer("/State/Running")
                .and_then(Value::as_bool)
                .unwrap_or(false);
            let status: Option<String> = entry
                .pointer("/State/Status")
                .and_then(Value::as_str)
                .map(str::to_string);
            let image: Option<String> = entry
                .pointer("/Config/Image")
                .and_then(Value::as_str)
                .map(str::to_string);
            let host_port: Option<u16> = entry
                .pointer("/NetworkSettings/Ports/5432/tcp")
                .and_then(Value::as_array)
                .and_then(|items| items.first())
                .and_then(|item| item.get("HostPort"))
                .and_then(Value::as_str)
                .and_then(|value| value.parse::<u16>().ok());
            let labels: HashMap<String, String> = entry
                .pointer("/Config/Labels")
                .and_then(Value::as_object)
                .map(|value| {
                    value
                        .iter()
                        .filter_map(|(key, value)| {
                            value
                                .as_str()
                                .map(|label_value| (key.clone(), label_value.to_string()))
                        })
                        .collect()
                })
                .unwrap_or_default();

            DockerManagedContainer {
                container_name,
                running,
                status,
                image,
                host_port,
                labels,
            }
        })
        .collect();

    containers.sort_by_cached_key(|item| item.container_name.to_lowercase());
    Ok(containers)
}

/// Inspect whether the current process user can talk to the Docker daemon without sudo fallback.
pub async fn inspect_docker_runtime_access_current_user() -> DockerRuntimeAccessStatus {
    let args: Vec<String> = DOCKER_RUNTIME_PROBE_ARGS
        .iter()
        .map(|value| (*value).to_string())
        .collect();

    match run_docker_capture_strings_inner(&args, false).await {
        Ok(output) => classify_docker_runtime_access(&output, &args),
        Err(ProvisioningError::Unavailable(message)) => DockerRuntimeAccessStatus {
            available: false,
            status: if message.contains("docker binary is not installed") {
                "binary_missing".to_string()
            } else {
                "unavailable".to_string()
            },
            detail: message,
            hint: None,
        },
        Err(ProvisioningError::Execution(message)) => DockerRuntimeAccessStatus {
            available: false,
            status: "error".to_string(),
            detail: message,
            hint: None,
        },
        Err(ProvisioningError::InvalidInput(message))
        | Err(ProvisioningError::Conflict(message)) => DockerRuntimeAccessStatus {
            available: false,
            status: "error".to_string(),
            detail: message,
            hint: None,
        },
    }
}

/// Remove a Docker container by name (force remove).
pub async fn remove_container(container_name: &str) -> Result<(), ProvisioningError> {
    validate_container_name(container_name)?;
    let output: Output = run_docker_capture(&["rm", "-f", container_name]).await?;
    if output.status.success() {
        return Ok(());
    }

    let stderr: String = String::from_utf8_lossy(&output.stderr).to_string();
    if stderr.to_lowercase().contains("no such container")
        || stderr.to_lowercase().contains("no such object")
    {
        return Ok(());
    }

    Err(map_docker_failed_status(
        &["rm", "-f", container_name],
        &output,
    ))
}

fn resolve_spin_up_config(
    params: SpinUpPostgresParams,
) -> Result<ResolvedSpinUpConfig, ProvisioningError> {
    let config: Option<Config> = Config::load().ok();
    let client_name: String = normalize_simple_name("client_name", &params.client_name)?;
    let container_name: String = match params.container_name {
        Some(value) if !value.trim().is_empty() => {
            validate_container_name(value.trim())?;
            value.trim().to_string()
        }
        _ => default_container_name(&client_name),
    };
    let image: String = params
        .image
        .filter(|value| !value.trim().is_empty())
        .or_else(|| {
            config
                .as_ref()
                .map(Config::get_provisioning_default_postgres_image)
        })
        .unwrap_or_else(|| DEFAULT_POSTGRES_IMAGE.to_string());
    let host: String = params
        .host
        .filter(|value| !value.trim().is_empty())
        .or_else(|| {
            config
                .as_ref()
                .map(Config::get_provisioning_default_instance_host)
        })
        .unwrap_or_else(|| DEFAULT_INSTANCE_HOST.to_string());
    validate_host(&host)?;

    let db_name: String = normalize_simple_name(
        "db_name",
        params.db_name.as_deref().unwrap_or(client_name.as_str()),
    )?;
    let username: String =
        normalize_simple_name("username", params.username.as_deref().unwrap_or("athena"))?;
    let password: String = match params.password {
        Some(value) if !value.trim().is_empty() => normalize_password(&value)?,
        _ => format!("athena_{}", Uuid::new_v4().simple()),
    };
    let host_port: u16 = params.host_port.unwrap_or(choose_available_port()?);
    let startup_timeout_secs: u64 = params
        .startup_timeout_secs
        .or_else(|| {
            config
                .as_ref()
                .map(Config::get_provisioning_default_startup_timeout_secs)
        })
        .unwrap_or(DEFAULT_STARTUP_TIMEOUT_SECS)
        .max(5);

    Ok(ResolvedSpinUpConfig {
        client_name,
        container_name,
        image,
        host,
        host_port,
        db_name,
        username,
        password,
        startup_timeout_secs,
        reuse_existing: params.reuse_existing,
    })
}

fn normalize_simple_name(field: &str, value: &str) -> Result<String, ProvisioningError> {
    let trimmed: &str = value.trim();
    if trimmed.is_empty() {
        return Err(ProvisioningError::InvalidInput(format!(
            "'{}' must not be empty",
            field
        )));
    }
    if trimmed.len() > 63 {
        return Err(ProvisioningError::InvalidInput(format!(
            "'{}' exceeds 63 characters",
            field
        )));
    }
    if !trimmed
        .chars()
        .all(|ch| ch.is_ascii_alphanumeric() || ch == '_' || ch == '-')
    {
        return Err(ProvisioningError::InvalidInput(format!(
            "'{}' may contain only ASCII letters, numbers, '_' and '-'",
            field
        )));
    }
    Ok(trimmed.to_lowercase())
}

fn normalize_password(value: &str) -> Result<String, ProvisioningError> {
    let trimmed: &str = value.trim();
    if trimmed.len() < 8 {
        return Err(ProvisioningError::InvalidInput(
            "password must contain at least 8 characters".to_string(),
        ));
    }
    if !trimmed
        .chars()
        .all(|ch| ch.is_ascii_alphanumeric() || ch == '_' || ch == '-' || ch == '.')
    {
        return Err(ProvisioningError::InvalidInput(
            "password may contain only ASCII letters, numbers, '_', '-', and '.'".to_string(),
        ));
    }
    Ok(trimmed.to_string())
}

fn validate_container_name(name: &str) -> Result<(), ProvisioningError> {
    if name.trim().is_empty() {
        return Err(ProvisioningError::InvalidInput(
            "container_name must not be empty".to_string(),
        ));
    }
    if !name
        .chars()
        .all(|ch| ch.is_ascii_alphanumeric() || ch == '-' || ch == '_' || ch == '.')
    {
        return Err(ProvisioningError::InvalidInput(
            "container_name may contain only ASCII letters, numbers, '-', '_' and '.'".to_string(),
        ));
    }
    Ok(())
}

fn validate_host(host: &str) -> Result<(), ProvisioningError> {
    if host.trim().is_empty() {
        return Err(ProvisioningError::InvalidInput(
            "host must not be empty".to_string(),
        ));
    }
    if host.contains(char::is_whitespace) {
        return Err(ProvisioningError::InvalidInput(
            "host must not contain whitespace".to_string(),
        ));
    }
    Ok(())
}

fn default_container_name(client_name: &str) -> String {
    format!("athena-pg-{}", client_name.replace('_', "-"))
}

fn choose_available_port() -> Result<u16, ProvisioningError> {
    let listener = TcpListener::bind("127.0.0.1:0").map_err(|err| {
        ProvisioningError::Execution(format!("failed to reserve free port: {err}"))
    })?;
    let port = listener
        .local_addr()
        .map_err(|err| ProvisioningError::Execution(format!("failed to read free port: {err}")))?
        .port();
    Ok(port)
}

fn build_pg_uri(username: &str, password: &str, host: &str, port: u16, db_name: &str) -> String {
    format!("postgres://{username}:{password}@{host}:{port}/{db_name}")
}

async fn docker_container_exists(container_name: &str) -> Result<bool, ProvisioningError> {
    let output: Output = run_docker_capture(&["inspect", container_name]).await?;
    if output.status.success() {
        return Ok(true);
    }
    let stderr = String::from_utf8_lossy(&output.stderr).to_lowercase();
    if stderr.contains("no such object") || stderr.contains("no such container") {
        return Ok(false);
    }
    Err(map_docker_failed_status(
        &["inspect", container_name],
        &output,
    ))
}

async fn run_postgres_container(config: &ResolvedSpinUpConfig) -> Result<(), ProvisioningError> {
    let port_mapping: String = format!("{}:5432", config.host_port);
    let db_env: String = format!("POSTGRES_DB={}", config.db_name);
    let user_env: String = format!("POSTGRES_USER={}", config.username);
    let password_env: String = format!("POSTGRES_PASSWORD={}", config.password);
    let client_label: String = format!("athena.client={}", config.client_name);
    let args: Vec<String> = vec![
        "run".to_string(),
        "-d".to_string(),
        "--name".to_string(),
        config.container_name.clone(),
        "--restart".to_string(),
        "unless-stopped".to_string(),
        "--label".to_string(),
        "athena.managed=true".to_string(),
        "--label".to_string(),
        client_label,
        "-e".to_string(),
        db_env,
        "-e".to_string(),
        user_env,
        "-e".to_string(),
        password_env,
        "-p".to_string(),
        port_mapping,
        config.image.clone(),
    ];

    let output: Output = run_docker_capture_strings(&args).await?;
    if output.status.success() {
        return Ok(());
    }
    Err(map_docker_failed_status(
        &[
            "run",
            "-d",
            "--name",
            config.container_name.as_str(),
            "<...>",
        ],
        &output,
    ))
}

/// Start a Docker container by name.
pub async fn start_container(container_name: &str) -> Result<(), ProvisioningError> {
    validate_container_name(container_name)?;
    let output: Output = run_docker_capture(&["start", container_name]).await?;
    if output.status.success() {
        return Ok(());
    }
    Err(map_docker_failed_status(
        &["start", container_name],
        &output,
    ))
}

/// Stop a Docker container by name.
pub async fn stop_container(container_name: &str) -> Result<(), ProvisioningError> {
    validate_container_name(container_name)?;
    let output: Output = run_docker_capture(&["stop", container_name]).await?;
    if output.status.success() {
        return Ok(());
    }
    Err(map_docker_failed_status(&["stop", container_name], &output))
}

async fn wait_for_postgres_ready(
    pg_uri: &str,
    timeout_secs: u64,
) -> Result<u128, ProvisioningError> {
    let started: Instant = Instant::now();
    loop {
        let sanitized_pg_uri = sanitize_sqlx_postgres_connect_uri(pg_uri);
        let connect_result: Result<sqlx::Pool<sqlx::Postgres>, sqlx::Error> = PgPoolOptions::new()
            .max_connections(1)
            .acquire_timeout(Duration::from_secs(2))
            .connect(sanitized_pg_uri.as_ref())
            .await;

        if let Ok(pool) = connect_result {
            let ping_result: Result<i32, sqlx::Error> = sqlx::query_scalar::<_, i32>("SELECT 1")
                .fetch_one(&pool)
                .await;
            pool.close().await;
            if ping_result.is_ok() {
                return Ok(started.elapsed().as_millis());
            }
        }

        if started.elapsed() > Duration::from_secs(timeout_secs) {
            return Err(ProvisioningError::Execution(format!(
                "timed out waiting for Postgres readiness after {} seconds",
                timeout_secs
            )));
        }

        sleep(Duration::from_millis(1000)).await;
    }
}

async fn run_docker_capture(args: &[&str]) -> Result<Output, ProvisioningError> {
    let mapped: Vec<String> = args
        .iter()
        .map(|value| (*value).to_string())
        .collect::<Vec<_>>();
    run_docker_capture_strings(&mapped).await
}

async fn run_docker_capture_strings(args: &[String]) -> Result<Output, ProvisioningError> {
    run_docker_capture_strings_inner(args, true).await
}

async fn run_docker_capture_strings_inner(
    args: &[String],
    allow_sudo_retry: bool,
) -> Result<Output, ProvisioningError> {
    let _ = allow_sudo_retry;
    let output: Output = Command::new("docker")
        .args(args)
        .output()
        .await
        .map_err(|err| {
            if err.kind() == std::io::ErrorKind::NotFound {
                ProvisioningError::Unavailable(
                    "docker binary is not installed or not available in PATH".to_string(),
                )
            } else {
                ProvisioningError::Execution(format!(
                    "failed to execute docker command '{}': {}",
                    args.join(" "),
                    err
                ))
            }
        })?;

    #[cfg(unix)]
    {
        if allow_sudo_retry && should_retry_docker_with_sudo(&output) {
            // First, try to self-heal the Docker group membership. If it
            // succeeds, re-run the original command directly (no sudo needed).
            if try_self_heal_docker_group().await {
                let healed_output =
                    Command::new("docker")
                        .args(args)
                        .output()
                        .await
                        .map_err(|err| {
                            ProvisioningError::Execution(format!(
                                "failed to execute docker command '{}' after self-heal: {}",
                                args.join(" "),
                                err
                            ))
                        })?;
                if healed_output.status.success() {
                    return Ok(healed_output);
                }
            }

            // Fall back to sudo escalation with an optional password.
            if let Some(sudo_output) = run_docker_capture_with_sudo(args).await {
                if sudo_output.status.success() {
                    return Ok(sudo_output);
                }
            }
        }
    }

    Ok(output)
}

fn output_indicates_docker_daemon_permission_denied(stderr: &str, stdout: &str) -> bool {
    let combined = format!("{stderr}\n{stdout}").to_lowercase();
    combined.contains("permission denied")
        && (combined.contains("docker daemon")
            || combined.contains("error during connect")
            || combined.contains("/var/run/docker.sock"))
}

#[cfg(unix)]
fn should_retry_docker_with_sudo(output: &Output) -> bool {
    if output.status.success() {
        return false;
    }

    let stderr = String::from_utf8_lossy(&output.stderr);
    let stdout = String::from_utf8_lossy(&output.stdout);
    output_indicates_docker_daemon_permission_denied(&stderr, &stdout)
}

#[cfg(unix)]
async fn run_docker_capture_with_sudo(args: &[String]) -> Option<Output> {
    use tokio::io::AsyncWriteExt;

    // Prefer password-based sudo (`sudo -S`) when a password is configured,
    // fall back to non-interactive sudo (`sudo -n`) when not.
    let sudo_password: Option<String> = Config::load()
        .ok()
        .and_then(|cfg| cfg.get_provisioning_docker_sudo_password());

    if let Some(ref password) = sudo_password {
        // Pipe the password on stdin with a trailing newline so sudo -S can
        // read it without prompting.
        let mut child = match Command::new("sudo")
            .arg("-S")
            .arg("docker")
            .args(args)
            .stdin(std::process::Stdio::piped())
            .stdout(std::process::Stdio::piped())
            .stderr(std::process::Stdio::piped())
            .spawn()
        {
            Ok(child) => child,
            Err(_) => return None,
        };

        if let Some(mut stdin) = child.stdin.take() {
            let pw_bytes = format!("{}\n", password).into_bytes();
            let _ = stdin.write_all(&pw_bytes).await;
            // Drop stdin so sudo stops waiting for more input.
        }

        child.wait_with_output().await.ok()
    } else {
        // No password configured — attempt non-interactive sudo as before.
        Command::new("sudo")
            .arg("-n")
            .arg("docker")
            .args(args)
            .output()
            .await
            .ok()
    }
}

/// Attempt to add the current OS user to the `docker` group using sudo, then
/// re-probe Docker access. Returns `true` if Docker became accessible after the
/// fix. Requires either passwordless sudo or `ATHENA_DOCKER_SUDO_PASSWORD` /
/// `provisioning.docker_sudo_password` to be set, and only runs when
/// `provisioning.docker_self_heal_group` (or `ATHENA_DOCKER_SELF_HEAL_GROUP`)
/// is enabled.
#[cfg(unix)]
async fn try_self_heal_docker_group() -> bool {
    use tokio::io::AsyncWriteExt;

    let config = Config::load().ok();
    let self_heal = config
        .as_ref()
        .map(|cfg| cfg.get_provisioning_docker_self_heal_group())
        .unwrap_or(false);

    if !self_heal {
        return false;
    }

    // Determine the user name to add. Prefer the SUDO_USER env var (set when
    // the process was started via sudo), then fall back to the USER env var.
    let user = match std::env::var("SUDO_USER")
        .ok()
        .filter(|u| !u.trim().is_empty())
        .or_else(|| std::env::var("USER").ok().filter(|u| !u.trim().is_empty()))
    {
        Some(u) => u,
        None => return false,
    };

    let sudo_password = config.and_then(|cfg| cfg.get_provisioning_docker_sudo_password());

    tracing::warn!(
        target: "athena::provisioning",
        user = %user,
        "Docker permission denied — attempting self-heal: adding user to 'docker' group"
    );

    let run_sudo = |args: Vec<String>| {
        let pw = sudo_password.clone();
        async move {
            if let Some(ref password) = pw {
                let mut child = Command::new("sudo")
                    .arg("-S")
                    .args(&args)
                    .stdin(std::process::Stdio::piped())
                    .stdout(std::process::Stdio::piped())
                    .stderr(std::process::Stdio::piped())
                    .spawn()
                    .ok()?;

                if let Some(mut stdin) = child.stdin.take() {
                    let _ = stdin.write_all(format!("{}\n", password).as_bytes()).await;
                }
                child.wait_with_output().await.ok()
            } else {
                Command::new("sudo")
                    .arg("-n")
                    .args(&args)
                    .output()
                    .await
                    .ok()
            }
        }
    };

    let usermod_result = run_sudo(vec![
        "usermod".to_string(),
        "-aG".to_string(),
        "docker".to_string(),
        user.clone(),
    ])
    .await;
    let usermod_ok = usermod_result
        .as_ref()
        .map(|o| o.status.success())
        .unwrap_or(false);

    if !usermod_ok {
        let detail = usermod_result
            .map(|o| String::from_utf8_lossy(&o.stderr).trim().to_string())
            .unwrap_or_else(|| "command failed to execute".to_string());
        tracing::warn!(
            target: "athena::provisioning",
            user = %user,
            error = %detail,
            "Docker self-heal: usermod -aG docker failed"
        );
        return false;
    }

    // Apply the new group membership in the current process session so we
    // don't need a full re-login.
    let newgrp_result = run_sudo(vec!["newgrp".to_string(), "docker".to_string()]).await;
    let newgrp_ok = newgrp_result
        .as_ref()
        .map(|o| o.status.success())
        .unwrap_or(false);

    if newgrp_ok {
        tracing::info!(
            target: "athena::provisioning",
            user = %user,
            "Docker self-heal: usermod + newgrp succeeded — retrying Docker probe"
        );
    } else {
        // newgrp may not be available; log but continue — the usermod may have
        // taken effect via the sudo session.
        tracing::warn!(
            target: "athena::provisioning",
            user = %user,
            "Docker self-heal: usermod succeeded but newgrp failed — retrying Docker probe anyway"
        );
    }

    // Probe Docker directly (without sudo) to see if the fix worked.
    let probe_args: Vec<String> = DOCKER_RUNTIME_PROBE_ARGS
        .iter()
        .map(|s| (*s).to_string())
        .collect();
    let probe = Command::new("docker").args(&probe_args).output().await.ok();

    let healed = probe.map(|o| o.status.success()).unwrap_or(false);
    if healed {
        tracing::info!(
            target: "athena::provisioning",
            user = %user,
            "Docker self-heal succeeded — Docker is now accessible without sudo"
        );
    } else {
        tracing::warn!(
            target: "athena::provisioning",
            user = %user,
            "Docker self-heal: group membership applied but Docker probe still failing \
             (a full service restart may be required)"
        );
    }
    healed
}

fn map_docker_failed_status(args: &[&str], output: &Output) -> ProvisioningError {
    let stderr: String = String::from_utf8_lossy(&output.stderr).trim().to_string();
    let stdout: String = String::from_utf8_lossy(&output.stdout).trim().to_string();
    let combined: String = format!("{stderr}\n{stdout}").to_lowercase();
    let command: String = format!("docker {}", args.join(" "));

    if combined.contains("is already in use by container") {
        return ProvisioningError::Conflict(format!(
            "{} failed: container name is already in use. Choose another name or enable reuse_existing.",
            command
        ));
    }
    if combined.contains("port is already allocated") || combined.contains("address already in use")
    {
        return ProvisioningError::Conflict(format!(
            "{} failed: requested host port is already in use.",
            command
        ));
    }
    if combined.contains("permission denied")
        && (combined.contains("docker daemon")
            || combined.contains("error during connect")
            || combined.contains("/var/run/docker.sock"))
    {
        return ProvisioningError::Unavailable(format!(
            "{} failed: permission denied while talking to Docker daemon. {}",
            command,
            docker_permission_denied_hint()
        ));
    }
    if combined.contains("error during connect")
        || combined.contains("is the docker daemon running")
        || combined.contains("docker daemon is not running")
    {
        return ProvisioningError::Unavailable(format!(
            "{} failed: Docker daemon is unavailable. Start Docker and retry.",
            command
        ));
    }
    if combined.contains("permission denied") {
        return ProvisioningError::Unavailable(format!(
            "{} failed: permission denied while talking to Docker daemon. {}",
            command,
            docker_permission_denied_hint()
        ));
    }

    ProvisioningError::Execution(format!(
        "{} failed with status {:?}: {}",
        command,
        output.status.code(),
        if stderr.is_empty() { stdout } else { stderr }
    ))
}

fn classify_docker_runtime_access(output: &Output, args: &[String]) -> DockerRuntimeAccessStatus {
    if output.status.success() {
        return DockerRuntimeAccessStatus {
            available: true,
            status: "ready".to_string(),
            detail: "Docker daemon is reachable by the current process user.".to_string(),
            hint: None,
        };
    }

    let stderr: String = String::from_utf8_lossy(&output.stderr).trim().to_string();
    let stdout: String = String::from_utf8_lossy(&output.stdout).trim().to_string();
    let combined: String = format!("{stderr}\n{stdout}").to_lowercase();
    let detail: String = if stderr.is_empty() {
        stdout.clone()
    } else if stdout.is_empty() {
        stderr.clone()
    } else {
        format!("{stderr}\n{stdout}")
    };

    if output_indicates_docker_daemon_permission_denied(&stderr, &stdout) {
        return DockerRuntimeAccessStatus {
            available: false,
            status: "permission_denied".to_string(),
            detail:
                "Docker socket permission missing for current process user (cannot access Docker daemon)."
                    .to_string(),
            hint: Some(docker_permission_denied_hint().to_string()),
        };
    }

    if combined.contains("error during connect")
        || combined.contains("is the docker daemon running")
        || combined.contains("docker daemon is not running")
    {
        return DockerRuntimeAccessStatus {
            available: false,
            status: "daemon_unavailable".to_string(),
            detail:
                "Docker daemon is unavailable for the current process user. Start Docker and retry."
                    .to_string(),
            hint: None,
        };
    }

    DockerRuntimeAccessStatus {
        available: false,
        status: "error".to_string(),
        detail: if detail.is_empty() {
            format!(
                "docker {} failed with status {:?}",
                args.join(" "),
                output.status.code()
            )
        } else {
            detail
        },
        hint: None,
    }
}

fn docker_permission_denied_hint() -> &'static str {
    #[cfg(unix)]
    {
        "On Linux, ensure your user can access Docker (e.g. add user to 'docker' group, re-login/newgrp docker, and verify Docker service is running)."
    }
    #[cfg(not(unix))]
    {
        "Ensure Docker Desktop/daemon is running and the current user has permission to access it."
    }
}

#[cfg(test)]
mod tests {
    use super::{
        classify_docker_runtime_access, default_container_name, normalize_password,
        normalize_simple_name, output_indicates_docker_daemon_permission_denied,
    };
    use std::process::Output;

    #[cfg(unix)]
    use std::os::unix::process::ExitStatusExt;
    #[cfg(windows)]
    use std::os::windows::process::ExitStatusExt;

    #[test]
    fn default_container_name_uses_client_name() {
        assert_eq!(default_container_name("my_client"), "athena-pg-my-client");
    }

    #[test]
    fn normalize_name_rejects_invalid_chars() {
        let result = normalize_simple_name("client_name", "bad name");
        assert!(result.is_err());
    }

    #[test]
    fn normalize_password_enforces_rules() {
        assert!(normalize_password("short").is_err());
        assert!(normalize_password("with space").is_err());
        assert!(normalize_password("valid-pass.123").is_ok());
    }

    #[test]
    fn docker_daemon_permission_denied_detection_matches_expected_errors() {
        assert!(output_indicates_docker_daemon_permission_denied(
            "Got permission denied while trying to connect to the Docker daemon socket at unix:///var/run/docker.sock",
            ""
        ));
        assert!(output_indicates_docker_daemon_permission_denied(
            "",
            "error during connect: permission denied"
        ));
    }

    #[test]
    fn docker_daemon_permission_denied_detection_ignores_unrelated_permission_denied() {
        assert!(!output_indicates_docker_daemon_permission_denied(
            "permission denied while reading local config file",
            ""
        ));
    }

    #[test]
    fn classify_docker_runtime_access_reports_permission_denied_clearly() {
        let output = Output {
            status: std::process::ExitStatus::from_raw(1),
            stdout: Vec::new(),
            stderr: b"Got permission denied while trying to connect to the Docker daemon socket at unix:///var/run/docker.sock".to_vec(),
        };

        let status = classify_docker_runtime_access(&output, &["ps".to_string()]);
        assert!(!status.available);
        assert_eq!(status.status, "permission_denied");
        assert!(status.detail.contains("current process user"));
        assert!(status.hint.is_some());
    }
}