import common_sql import gleam/dynamic/decode import gleam/erlang/process import gleam/int import gleam/list import gleam/string import pog fn param_to_value(param: common_sql.Param) -> pog.Value { case param { common_sql.PInt(n) -> pog.int(n) common_sql.PString(s) -> pog.text(s) common_sql.PFloat(f) -> pog.float(f) common_sql.PBool(b) -> pog.bool(b) common_sql.PNull -> pog.null() } } fn pog_error_to_db_error(err: pog.QueryError) -> common_sql.DbError { case err { pog.ConnectionUnavailable -> common_sql.ConnectionError("Connection unavailable") pog.ConstraintViolated(message, _, _) -> common_sql.QueryError("Constraint violated: " <> message) pog.PostgresqlError(_, _, message) -> common_sql.QueryError("PostgreSQL error: " <> message) pog.UnexpectedArgumentCount(expected, got) -> common_sql.QueryError( "Wrong argument count: expected " <> int.to_string(expected) <> ", got " <> int.to_string(got), ) pog.UnexpectedArgumentType(expected, got) -> common_sql.QueryError( "Wrong argument type: expected " <> expected <> ", got " <> got, ) pog.UnexpectedResultType(errors) -> common_sql.QueryError( "Unexpected result type: " <> string.inspect(errors), ) pog.QueryTimeout -> common_sql.QueryError("Query timed out") } } /// Returns a `common_sql.Driver` backed by PostgreSQL via the `pog` library. /// /// Call `driver()` once at application startup and pass it down. Each call /// creates one Erlang atom for the pool name — do not call it in a loop. /// /// Closing a pog connection pool has no explicit stop API; the pool runs as /// an OTP actor for the lifetime of the application. `common_sql.close/2` is /// a no-op for this driver. Use `pog.supervised/1` to manage the pool in an /// OTP supervision tree. pub fn driver() -> common_sql.Driver(pog.Connection) { let pool_name: process.Name(pog.Message) = process.new_name("common_sql_pg") common_sql.Driver( driver_type: "postgresql", connect: fn(url) { case pog.url_config(pool_name, url) { Error(Nil) -> Error(common_sql.ConnectionError("Invalid PostgreSQL URL: " <> url)) Ok(config) -> case pog.start(config) { Ok(started) -> Ok(started.data) Error(_) -> // Pool may already be running (e.g. called connect twice). // If a process is registered under the pool name, reuse it. case process.named(pool_name) { Ok(_) -> Ok(pog.named_connection(pool_name)) Error(Nil) -> Error(common_sql.ConnectionError( "Failed to start connection pool for: " <> url, )) } } } }, execute: fn(conn, query, params) { let sql = case query { common_sql.Sql(s) | common_sql.Portable(s) -> s } let q = list.fold(params, pog.query(sql), fn(q, param) { pog.parameter(q, param_to_value(param)) }) let q = pog.returning(q, decode.dynamic) case pog.execute(q, on: conn) { Ok(pog.Returned(_, rows)) -> Ok(rows) Error(err) -> Error(pog_error_to_db_error(err)) } }, close: fn(_conn) { // pog pools are managed as OTP actors — there is no stop/close API. Nil }, ) }