lmrc_postgres/

operations.rs

//! Idempotent PostgreSQL operations
//!
//! This module provides idempotent operations for PostgreSQL installation and management.
//! All operations can be safely run multiple times without causing errors.

use crate::config::PostgresConfig;
use crate::diff::{ChangeType, ConfigChange, ConfigDiff};
use crate::error::{Error, Result};
use lmrc_ssh::SshClient;
use tracing::{debug, info};

/// Check if PostgreSQL is installed on the server
///
/// # Arguments
///
/// * `ssh` - SSH client connection to the server
///
/// # Returns
///
/// * `Ok(true)` if PostgreSQL is installed
/// * `Ok(false)` if PostgreSQL is not installed
/// * `Err(_)` on SSH or command execution errors
pub async fn is_installed(ssh: &mut SshClient) -> Result<bool> {
    debug!("Checking if PostgreSQL is installed");

    match ssh.execute("which psql") {
        Ok(output) => {
            let installed = !output.stdout.trim().is_empty();
            debug!("PostgreSQL installed: {}", installed);
            Ok(installed)
        }
        Err(_) => {
            debug!("PostgreSQL not found");
            Ok(false)
        }
    }
}

/// Get installed PostgreSQL version
///
/// # Arguments
///
/// * `ssh` - SSH client connection to the server
///
/// # Returns
///
/// * `Ok(Some(version))` if PostgreSQL is installed
/// * `Ok(None)` if PostgreSQL is not installed
pub async fn get_installed_version(ssh: &mut SshClient) -> Result<Option<String>> {
    debug!("Getting installed PostgreSQL version");

    if !is_installed(ssh).await? {
        return Ok(None);
    }

    let output = ssh
        .execute("psql --version")
        .map_err(|e| Error::ssh_execution(e.to_string(), "psql --version"))?;

    // Parse version from output like "psql (PostgreSQL) 15.4"
    let version = output
        .stdout
        .split_whitespace()
        .nth(2)
        .and_then(|v| v.split('.').next())
        .map(|v| v.to_string());

    debug!("Installed version: {:?}", version);
    Ok(version)
}

/// Install PostgreSQL (idempotent)
///
/// This function will:
/// - Check if PostgreSQL is already installed
/// - If installed with the correct version, skip installation
/// - If installed with a different version, return an error
/// - If not installed, perform installation
///
/// # Arguments
///
/// * `ssh` - SSH connection to the server
/// * `config` - PostgreSQL configuration
pub async fn install(ssh: &mut SshClient, config: &PostgresConfig) -> Result<()> {
    info!("Installing PostgreSQL {}", config.version);

    // Check if already installed
    if let Some(installed_version) = get_installed_version(ssh).await? {
        if installed_version == config.version {
            info!(
                "PostgreSQL {} is already installed, skipping installation",
                config.version
            );
            return Ok(());
        } else {
            return Err(Error::AlreadyInstalled(installed_version));
        }
    }

    // Install prerequisites
    info!("Installing prerequisites");
    ssh.execute("DEBIAN_FRONTEND=noninteractive apt-get update -y")
        .map_err(|e| Error::Installation(format!("Failed to update package list: {}", e)))?;

    ssh.execute("DEBIAN_FRONTEND=noninteractive apt-get install -y gnupg2 wget lsb-release")
        .map_err(|e| Error::Installation(format!("Failed to install prerequisites: {}", e)))?;

    // Add PostgreSQL APT repository
    info!("Adding PostgreSQL APT repository");
    ssh.execute(
        "wget --quiet -O - https://www.postgresql.org/media/keys/ACCC4CF8.asc | apt-key add -",
    )
    .map_err(|e| Error::Installation(format!("Failed to add GPG key: {}", e)))?;

    ssh.execute(r#"echo "deb http://apt.postgresql.org/pub/repos/apt/ $(lsb_release -cs)-pgdg main" > /etc/apt/sources.list.d/pgdg.list"#)
        .map_err(|e| Error::Installation(format!("Failed to add repository: {}", e)))?;

    // Update package list
    info!("Updating package list");
    ssh.execute("apt-get update -y")
        .map_err(|e| Error::Installation(format!("Failed to update after adding repo: {}", e)))?;

    // Install PostgreSQL
    info!("Installing PostgreSQL {}", config.version);
    let install_cmd = format!(
        "DEBIAN_FRONTEND=noninteractive apt-get install -y postgresql-{}",
        config.version
    );
    ssh.execute(&install_cmd)
        .map_err(|e| Error::Installation(format!("Failed to install PostgreSQL: {}", e)))?;

    // Start and enable service
    info!("Starting PostgreSQL service");
    ssh.execute("systemctl start postgresql")
        .map_err(|e| Error::ServiceError(format!("Failed to start service: {}", e)))?;

    ssh.execute("systemctl enable postgresql")
        .map_err(|e| Error::ServiceError(format!("Failed to enable service: {}", e)))?;

    // Wait for service to be ready
    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;

    // Verify installation
    verify_service_running(ssh).await?;

    info!("PostgreSQL {} installed successfully", config.version);
    Ok(())
}

/// Uninstall PostgreSQL
///
/// # Arguments
///
/// * `ssh` - SSH connection to the server
/// * `config` - PostgreSQL configuration (for version)
/// * `purge` - If true, remove all data and configuration files
pub async fn uninstall(ssh: &mut SshClient, config: &PostgresConfig, purge: bool) -> Result<()> {
    info!("Uninstalling PostgreSQL {}", config.version);

    if !is_installed(ssh).await? {
        info!("PostgreSQL is not installed, nothing to uninstall");
        return Ok(());
    }

    // Stop service
    info!("Stopping PostgreSQL service");
    let _ = ssh.execute("systemctl stop postgresql");

    // Uninstall package
    let uninstall_cmd = if purge {
        format!("apt-get purge -y postgresql-{}", config.version)
    } else {
        format!("apt-get remove -y postgresql-{}", config.version)
    };

    ssh.execute(&uninstall_cmd)
        .map_err(|e| Error::Uninstallation(format!("Failed to uninstall: {}", e)))?;

    if purge {
        info!("Removing PostgreSQL data and configuration files");
        let _ = ssh.execute(&format!("rm -rf /etc/postgresql/{}", config.version));
        let _ = ssh.execute(&format!("rm -rf /var/lib/postgresql/{}", config.version));
    }

    // Clean up packages
    let _ = ssh.execute("apt-get autoremove -y");

    info!("PostgreSQL uninstalled successfully");
    Ok(())
}

/// Configure PostgreSQL database and user (idempotent)
///
/// This function will:
/// - Create database if it doesn't exist
/// - Create user if it doesn't exist
/// - Grant permissions (idempotent)
///
/// # Arguments
///
/// * `ssh` - SSH connection to the server
/// * `config` - PostgreSQL configuration
pub async fn configure_database(ssh: &mut SshClient, config: &PostgresConfig) -> Result<()> {
    info!("Configuring database and user");

    // Create database (idempotent with 'IF NOT EXISTS')
    info!("Creating database {}", config.database_name);
    let create_db_cmd = format!(
        r#"sudo -u postgres psql -c "CREATE DATABASE {} ENCODING 'UTF8';" || true"#,
        config.database_name
    );
    ssh.execute(&create_db_cmd)
        .map_err(|e| Error::Configuration(format!("Failed to create database: {}", e)))?;

    // Create user (idempotent with '|| true')
    info!("Creating user {}", config.username);
    let create_user_cmd = format!(
        r#"sudo -u postgres psql -c "CREATE USER {} WITH PASSWORD '{}';" || true"#,
        config.username, config.password
    );
    ssh.execute(&create_user_cmd)
        .map_err(|e| Error::Configuration(format!("Failed to create user: {}", e)))?;

    // Update password (in case user already exists)
    let update_password_cmd = format!(
        r#"sudo -u postgres psql -c "ALTER USER {} WITH PASSWORD '{}';" || true"#,
        config.username, config.password
    );
    ssh.execute(&update_password_cmd)
        .map_err(|e| Error::Configuration(format!("Failed to update user password: {}", e)))?;

    // Grant database privileges
    info!("Granting privileges");
    let grant_cmd = format!(
        r#"sudo -u postgres psql -c "GRANT ALL PRIVILEGES ON DATABASE {} TO {};" || true"#,
        config.database_name, config.username
    );
    ssh.execute(&grant_cmd)
        .map_err(|e| Error::Configuration(format!("Failed to grant privileges: {}", e)))?;

    // Grant schema permissions (PostgreSQL 15+)
    let grant_schema_cmd = format!(
        r#"sudo -u postgres psql -d {} -c "GRANT ALL ON SCHEMA public TO {};" || true"#,
        config.database_name, config.username
    );
    ssh.execute(&grant_schema_cmd)
        .map_err(|e| Error::Configuration(format!("Failed to grant schema privileges: {}", e)))?;

    // Grant default privileges
    let grant_default_cmd = format!(
        r#"sudo -u postgres psql -d {} -c "ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT ALL ON TABLES TO {};" || true"#,
        config.database_name, config.username
    );
    ssh.execute(&grant_default_cmd)
        .map_err(|e| Error::Configuration(format!("Failed to grant default privileges: {}", e)))?;

    info!("Database and user configured successfully");
    Ok(())
}

/// Configure PostgreSQL server settings
///
/// # Arguments
///
/// * `ssh` - SSH connection to the server
/// * `config` - PostgreSQL configuration
pub async fn configure_server(ssh: &mut SshClient, config: &PostgresConfig) -> Result<()> {
    info!("Configuring PostgreSQL server");

    let config_dir = config.config_dir();

    // Update listen_addresses
    info!("Configuring listen addresses");
    let listen_cmd = format!(
        r#"sed -i "s/#listen_addresses = 'localhost'/listen_addresses = '*'/" {}/postgresql.conf"#,
        config_dir
    );
    ssh.execute(&listen_cmd)
        .map_err(|e| Error::Configuration(format!("Failed to update listen_addresses: {}", e)))?;

    // Update port if not default
    if config.port != 5432 {
        let port_cmd = format!(
            r#"sed -i "s/#port = 5432/port = {}/" {}/postgresql.conf"#,
            config.port, config_dir
        );
        ssh.execute(&port_cmd)
            .map_err(|e| Error::Configuration(format!("Failed to update port: {}", e)))?;
    }

    // Configure max_connections
    if let Some(max_conn) = config.max_connections {
        let max_conn_cmd = format!(
            r#"sed -i "s/max_connections = [0-9]\\+/max_connections = {}/" {}/postgresql.conf"#,
            max_conn, config_dir
        );
        ssh.execute(&max_conn_cmd).map_err(|e| {
            Error::Configuration(format!("Failed to update max_connections: {}", e))
        })?;
    }

    // Configure shared_buffers
    if let Some(ref buffers) = config.shared_buffers {
        let buffers_cmd = format!(
            r#"sed -i "s/#shared_buffers = .*/shared_buffers = {}/" {}/postgresql.conf"#,
            buffers, config_dir
        );
        ssh.execute(&buffers_cmd)
            .map_err(|e| Error::Configuration(format!("Failed to update shared_buffers: {}", e)))?;
    }

    // Update pg_hba.conf for remote access
    info!("Configuring remote access");
    let hba_cmd = format!(
        r#"echo "host    all             all             {}             md5" >> {}/pg_hba.conf"#,
        config.listen_addresses, config_dir
    );
    ssh.execute(&hba_cmd)
        .map_err(|e| Error::Configuration(format!("Failed to update pg_hba.conf: {}", e)))?;

    // Restart service to apply changes
    info!("Restarting PostgreSQL to apply configuration");
    ssh.execute("systemctl restart postgresql")
        .map_err(|e| Error::ServiceError(format!("Failed to restart service: {}", e)))?;

    // Verify service is running
    tokio::time::sleep(tokio::time::Duration::from_secs(2)).await;
    verify_service_running(ssh).await?;

    info!("PostgreSQL server configured successfully");
    Ok(())
}

/// Detect configuration differences
///
/// # Arguments
///
/// * `ssh` - SSH connection to the server
/// * `config` - Desired PostgreSQL configuration
///
/// # Returns
///
/// A `ConfigDiff` containing all detected changes
pub async fn detect_diff(ssh: &mut SshClient, config: &PostgresConfig) -> Result<ConfigDiff> {
    debug!("Detecting configuration differences");

    let mut diff = ConfigDiff::new();
    let config_file = config.postgresql_conf_path();

    // Helper to get current config value
    let get_config_value = |ssh: &mut SshClient, param: &str| -> Option<String> {
        let cmd = format!(
            r#"grep "^{} = " {} | sed "s/{} = //" | tr -d "'"#,
            param, config_file, param
        );
        ssh.execute(&cmd).ok().and_then(|output| {
            let val = output.stdout.trim();
            if val.is_empty() {
                None
            } else {
                Some(val.to_string())
            }
        })
    };

    // Helper macro to check a config parameter
    macro_rules! check_param {
        ($param_name:expr, $desired_value:expr) => {
            if let Some(ref desired) = $desired_value {
                let current = get_config_value(ssh, $param_name);
                let desired_str = desired.to_string();
                if current.as_deref() != Some(desired_str.as_str()) {
                    let change_type = if current.is_some() {
                        ChangeType::Modify
                    } else {
                        ChangeType::Add
                    };
                    diff.add_change(ConfigChange {
                        parameter: $param_name.to_string(),
                        current,
                        desired: desired_str,
                        change_type,
                    });
                }
            }
        };
    }

    // Check all PostgreSQL parameters
    check_param!(
        "max_connections",
        config.max_connections.map(|v| v.to_string())
    );
    check_param!("shared_buffers", config.shared_buffers.clone());
    check_param!("effective_cache_size", config.effective_cache_size.clone());
    check_param!("work_mem", config.work_mem.clone());
    check_param!("maintenance_work_mem", config.maintenance_work_mem.clone());
    check_param!("wal_buffers", config.wal_buffers.clone());
    check_param!(
        "checkpoint_completion_target",
        config.checkpoint_completion_target.map(|v| v.to_string())
    );

    // Check port
    let current_port = get_config_value(ssh, "port");
    let desired_port = config.port.to_string();
    if current_port.as_deref() != Some(desired_port.as_str()) {
        let change_type = if current_port.is_some() {
            ChangeType::Modify
        } else {
            ChangeType::Add
        };
        diff.add_change(ConfigChange {
            parameter: "port".to_string(),
            current: current_port,
            desired: desired_port,
            change_type,
        });
    }

    // Check listen_addresses
    let current_listen = get_config_value(ssh, "listen_addresses");
    if current_listen.as_deref() != Some(&config.listen_addresses) {
        let change_type = if current_listen.is_some() {
            ChangeType::Modify
        } else {
            ChangeType::Add
        };
        diff.add_change(ConfigChange {
            parameter: "listen_addresses".to_string(),
            current: current_listen,
            desired: config.listen_addresses.clone(),
            change_type,
        });
    }

    // Check SSL setting
    let current_ssl = get_config_value(ssh, "ssl");
    let desired_ssl = if config.ssl { "on" } else { "off" }.to_string();
    if current_ssl.as_deref() != Some(desired_ssl.as_str()) {
        let change_type = if current_ssl.is_some() {
            ChangeType::Modify
        } else {
            ChangeType::Add
        };
        diff.add_change(ConfigChange {
            parameter: "ssl".to_string(),
            current: current_ssl,
            desired: desired_ssl,
            change_type,
        });
    }

    // Check extra_config parameters
    for (key, desired_value) in &config.extra_config {
        let current = get_config_value(ssh, key);
        if current.as_deref() != Some(desired_value.as_str()) {
            let change_type = if current.is_some() {
                ChangeType::Modify
            } else {
                ChangeType::Add
            };
            diff.add_change(ConfigChange {
                parameter: key.clone(),
                current,
                desired: desired_value.clone(),
                change_type,
            });
        }
    }

    debug!("Diff detection complete: {} changes", diff.len());
    Ok(diff)
}

/// Test database connection
///
/// # Arguments
///
/// * `ssh` - SSH connection to the server
/// * `config` - PostgreSQL configuration
pub async fn test_connection(ssh: &mut SshClient, config: &PostgresConfig) -> Result<()> {
    debug!("Testing database connection");

    let test_cmd = format!(
        r#"PGPASSWORD='{}' psql -U {} -h localhost -p {} -d {} -c '\l' > /dev/null 2>&1"#,
        config.password, config.username, config.port, config.database_name
    );

    ssh.execute(&test_cmd)
        .map_err(|e| Error::ConnectionTest(format!("Connection test failed: {}", e)))?;

    info!("Database connection test successful");
    Ok(())
}

/// Verify PostgreSQL service is running
async fn verify_service_running(ssh: &mut SshClient) -> Result<()> {
    let status = ssh
        .execute("systemctl is-active postgresql")
        .map_err(|e| Error::ServiceError(format!("Failed to check service status: {}", e)))?;

    if status.stdout.trim() != "active" {
        return Err(Error::ServiceError(
            "PostgreSQL service is not active".to_string(),
        ));
    }

    debug!("PostgreSQL service is active");
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;

    // Note: These are unit tests that don't require SSH connection
    // Integration tests with actual SSH connections should be in tests/

    #[test]
    fn test_config_paths() {
        let config = PostgresConfig::builder()
            .version("15")
            .database_name("test")
            .username("user")
            .password("pass")
            .build()
            .unwrap();

        assert_eq!(config.config_dir(), "/etc/postgresql/15/main");
        assert_eq!(
            config.postgresql_conf_path(),
            "/etc/postgresql/15/main/postgresql.conf"
        );
    }
}