arcly_http/data/drivers/

sqlx.rs

//! SQLx adapter — `AnyPool` over postgres / mysql / sqlite.
//!
//! Uses SQLx's `Any` driver so one facade variant covers every backend the
//! enabled `db-sqlx-*` features compile in; the concrete backend is chosen
//! by the connection URL at boot.

use crate::data::db::DbDriver;
use crate::data::DataError;

/// Build a SQLx-backed driver for [`ArclyDbPool`](crate::data::db::ArclyDbPool).
///
/// ```ignore
/// let primary = sqlx_driver("postgres://app@db-primary/orders", 16).await?;
/// let replica = sqlx_driver("postgres://app@db-replica/orders", 16).await?;
/// ctx.provide(DataSourceRegistry::new(
///     ArclyDbPool::new("default", primary).with_replica(replica),
/// ));
/// ```
pub async fn sqlx_driver(url: &str, max_connections: u32) -> Result<DbDriver, DataError> {
    // Register the compiled-in backends with the Any driver (idempotent).
    sqlx::any::install_default_drivers();

    let pool = sqlx::any::AnyPoolOptions::new()
        .max_connections(max_connections)
        .connect(url)
        .await
        .map_err(|e| DataError::connection(format!("sqlx connect failed: {e}")))?;

    Ok(DbDriver::Sqlx(pool))
}

// ─── Query facade ─────────────────────────────────────────────────────────────
//
// A deliberately small surface: enough for services to run parameterized SQL
// through the unified handle. Apps wanting the full SQLx API can match on
// the enum variants directly — the facade never hides the native types.

impl crate::data::db::OwnedDbConn {
    /// Execute a statement (INSERT/UPDATE/DELETE/DDL); returns affected rows.
    pub async fn execute(&mut self, sql: &str) -> Result<u64, DataError> {
        match self {
            crate::data::db::OwnedDbConn::Sqlx(conn) => {
                use sqlx::Executor;
                conn.execute(sql)
                    .await
                    .map(|r| r.rows_affected())
                    .map_err(|e| DataError::query(e.to_string()))
            }
            #[allow(unreachable_patterns)]
            _ => Err(DataError::config("execute(): not a SQLx connection")),
        }
    }

    /// Execute a parameterized statement; returns affected rows.
    /// Placeholder syntax follows the backend (`?` sqlite/mysql, `$1…` pg).
    pub async fn execute_bind(&mut self, sql: &str, binds: &[&str]) -> Result<u64, DataError> {
        match self {
            crate::data::db::OwnedDbConn::Sqlx(conn) => {
                let mut q = sqlx::query(sql);
                for b in binds {
                    q = q.bind(*b);
                }
                q.execute(&mut **conn)
                    .await
                    .map(|r| r.rows_affected())
                    .map_err(|e| DataError::query(e.to_string()))
            }
            #[allow(unreachable_patterns)]
            _ => Err(DataError::config("execute_bind(): not a SQLx connection")),
        }
    }

    /// Fetch a single scalar `i64` (e.g. `SELECT COUNT(*) …`).
    pub async fn fetch_one_i64(&mut self, sql: &str) -> Result<i64, DataError> {
        self.fetch_one_i64_bind(sql, &[]).await
    }

    /// Fetch a single scalar `i64` with positional binds.
    pub async fn fetch_one_i64_bind(
        &mut self,
        sql: &str,
        binds: &[&str],
    ) -> Result<i64, DataError> {
        match self {
            crate::data::db::OwnedDbConn::Sqlx(conn) => {
                let mut q = sqlx::query_scalar::<_, i64>(sql);
                for b in binds {
                    q = q.bind(*b);
                }
                q.fetch_one(&mut **conn)
                    .await
                    .map_err(|e| DataError::query(e.to_string()))
            }
            #[allow(unreachable_patterns)]
            _ => Err(DataError::config("fetch_one_i64(): not a SQLx connection")),
        }
    }

    /// Fetch a single scalar string with positional binds.
    pub async fn fetch_one_string(
        &mut self,
        sql: &str,
        binds: &[&str],
    ) -> Result<String, DataError> {
        match self {
            crate::data::db::OwnedDbConn::Sqlx(conn) => {
                let mut q = sqlx::query_scalar::<_, String>(sql);
                for b in binds {
                    q = q.bind(*b);
                }
                q.fetch_one(&mut **conn)
                    .await
                    .map_err(|e| DataError::query(e.to_string()))
            }
            #[allow(unreachable_patterns)]
            _ => Err(DataError::config(
                "fetch_one_string(): not a SQLx connection",
            )),
        }
    }
}

impl crate::data::tx::ArclyTransaction {
    /// Execute a parameterized statement inside this transaction.
    /// Bind values positionally with the backend's placeholder syntax
    /// (`?` for sqlite/mysql, `$1…` for postgres via the Any driver).
    pub async fn execute_bind(&mut self, sql: &str, binds: &[&str]) -> Result<u64, DataError> {
        match self {
            crate::data::tx::ArclyTransaction::Sqlx(tx) => {
                let mut q = sqlx::query(sql);
                for b in binds {
                    q = q.bind(*b);
                }
                q.execute(&mut **tx)
                    .await
                    .map(|r| r.rows_affected())
                    .map_err(|e| DataError::query(e.to_string()))
            }
            #[allow(unreachable_patterns)]
            _ => Err(DataError::config("execute_bind(): not a SQLx transaction")),
        }
    }

    /// Fetch a single scalar `i64` inside this transaction.
    pub async fn fetch_one_i64(&mut self, sql: &str) -> Result<i64, DataError> {
        match self {
            crate::data::tx::ArclyTransaction::Sqlx(tx) => sqlx::query_scalar::<_, i64>(sql)
                .fetch_one(&mut **tx)
                .await
                .map_err(|e| DataError::query(e.to_string())),
            #[allow(unreachable_patterns)]
            _ => Err(DataError::config("fetch_one_i64(): not a SQLx transaction")),
        }
    }
}