horizon-sdk 6.9.0

//! Connection pool creation with RLS role switching.

use sqlx::postgres::{PgConnectOptions, PgPoolOptions, PgSslMode};
use sqlx::{ConnectOptions as _, PgPool};
use tracing::log::LevelFilter;

use crate::types::config::PostgresConfig;
use crate::types::error::{ConfigError, Result};

/// Create a connection pool configured for the Horizon database.
///
/// Each new connection executes `SET ROLE` to activate row-level
/// security policies. The pool is shared across all repository instances.
///
/// # Errors
///
/// Returns `PostgresError::Query` if the pool cannot be created, or
/// `ConfigError::InvalidRoleName` if the role name is invalid.
pub async fn create_pool(config: &PostgresConfig) -> Result<PgPool> {
    if let Some(role) = &config.role
        && (role.is_empty() || {
            let mut chars = role.chars();
            let first_valid = chars
                .next()
                .is_some_and(|c| c.is_ascii_lowercase() || c == '_');
            let rest_valid =
                chars.all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '_');
            !(first_valid && rest_valid)
        })
    {
        return Err(ConfigError::InvalidRoleName {
            role: role.clone(),
            reason: "must start with letter/underscore, then lowercase alphanumeric + underscore",
        }
        .into());
    }

    let ssl_mode = match config.sslmode.as_str() {
        "disable" => PgSslMode::Disable,
        "prefer" => PgSslMode::Prefer,
        "verify-ca" => PgSslMode::VerifyCa,
        "verify-full" => PgSslMode::VerifyFull,
        _ => PgSslMode::Require,
    };

    let connect_options = PgConnectOptions::new()
        .host(&config.host)
        .port(config.port)
        .username(&config.user)
        .password(&config.password)
        .database(&config.database)
        .ssl_mode(ssl_mode)
        .log_statements(LevelFilter::Trace);

    let role_for_pool = config.role.clone();
    let pool = PgPoolOptions::new()
        .max_connections(10)
        .after_connect(move |conn, _meta| {
            let role_for_conn = role_for_pool.clone();
            Box::pin(async move {
                if let Some(role) = &role_for_conn {
                    let query = format!("SET ROLE {role}");
                    sqlx::query(&query).execute(&mut *conn).await?;
                }
                Ok(())
            })
        })
        .connect_with(connect_options)
        .await?;

    Ok(pool)
}