#![forbid(unsafe_code)]
#[cfg(any(target_arch = "wasm32", target_arch = "wasm64"))]
compile_error!(
"cratefield-adapter-postgres is native-only: the sqlx Postgres driver does not \
compile to wasm (issue #18, ADR 0004). A wasm target must not depend on this \
crate — check the wasm dependency graph of the venture (modules see ports, \
never adapters, and runtime-native is the only consumer)."
);
mod convert;
mod lifecycle;
mod migrate;
mod reconcile;
mod registry;
pub mod testing;
pub use cratefield_core::TenantStatus;
pub use migrate::select_set;
pub use reconcile::{ModulePlan, TenantPlan, TenantRecord, TenantReport, tenant_lock_key};
pub use registry::{PoolRegistry, TENANT_POOL_IDLE, TENANT_POOL_TOTAL, TenantDsns};
use async_trait::async_trait;
use cratefield_core::{Database, DbError, Rows, Statement};
use sqlx::postgres::PgArguments;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct PoolLimits {
pub max_connections: u32,
pub idle_timeout: Option<std::time::Duration>,
pub acquire_timeout: std::time::Duration,
}
impl Default for PoolLimits {
fn default() -> Self {
Self {
max_connections: 8,
idle_timeout: Some(std::time::Duration::from_secs(300)),
acquire_timeout: std::time::Duration::from_secs(5),
}
}
}
#[derive(Clone)]
pub struct Postgres {
pool: sqlx::PgPool,
}
impl Postgres {
pub async fn connect(url: &str) -> Result<Self, DbError> {
Self::connect_with(url, PoolLimits::default()).await
}
pub async fn connect_with(url: &str, limits: PoolLimits) -> Result<Self, DbError> {
let pool = sqlx::postgres::PgPoolOptions::new()
.max_connections(limits.max_connections)
.idle_timeout(limits.idle_timeout)
.acquire_timeout(limits.acquire_timeout)
.connect(url)
.await
.map_err(|err| DbError::Execute(err.to_string()))?;
Ok(Self { pool })
}
fn prepare(stmt: &Statement) -> Result<(String, PgArguments), DbError> {
let sql = convert::rebind_placeholders(&stmt.sql);
let mut args = PgArguments::default();
convert::bind_values(&mut args, &stmt.values.0)?;
Ok((sql, args))
}
pub async fn close(&self) -> Result<(), DbError> {
self.pool.close().await;
Ok(())
}
}
#[async_trait]
impl Database for Postgres {
async fn execute(&self, stmt: &Statement) -> Result<u64, DbError> {
let (sql, args) = Self::prepare(stmt)?;
let result = sqlx::query_with(sql.as_str(), args)
.execute(&self.pool)
.await
.map_err(|err| DbError::Execute(err.to_string()))?;
Ok(result.rows_affected())
}
async fn query(&self, stmt: &Statement) -> Result<Rows, DbError> {
let (sql, args) = Self::prepare(stmt)?;
let rows = sqlx::query_with(sql.as_str(), args)
.fetch_all(&self.pool)
.await
.map_err(|err| DbError::Query(err.to_string()))?;
Ok(Rows::new(rows.iter().map(convert::pg_row_to_row).collect()))
}
async fn batch_atomic(&self, stmts: &[Statement]) -> Result<(), DbError> {
let mut tx = self
.pool
.begin()
.await
.map_err(|err| DbError::Batch(err.to_string()))?;
for stmt in stmts {
let (sql, args) = Self::prepare(stmt)?;
sqlx::query_with(sql.as_str(), args)
.execute(&mut *tx)
.await
.map_err(|err| DbError::Batch(err.to_string()))?;
}
tx.commit()
.await
.map_err(|err| DbError::Batch(err.to_string()))?;
Ok(())
}
}