Sign in

    moonpg

    A pure MoonBit PostgreSQL client, wire protocol from scratch.

    postgres
    postgresql
    wire-protocol
    database
    Download zip
    Author
    Version
    0.7.0
    License
    Apache-2.0
    Last updated
    3 months ago
    Downloads
    154

    #jaredzhou/moonpg

    A pure MoonBit PostgreSQL client — wire protocol from scratch, zero C dependencies.

    #Overview

    moonpg speaks the PostgreSQL wire protocol (v3) directly over TCP. It provides:

    • Connection — connect, authenticate (cleartext / MD5 / SCRAM-SHA-256), query, close.
    • Pool — bounded connection pool with acquire / release, min-idle, and idle-timeout expiry.
    • Transactions — begin_tx / commit / rollback on bare connections and pooled connections, plus begin_func (auto-commit/rollback).
    • Rows trait — polymorphic row iteration (ConnRows, PoolRows); pull-based with has_next() / get_row().
    • Typed decoding — FromValue trait for Int, Int64, Double, Bool, String, Bytes, Json, Decimal, UUID, Timestamp, plus Option[T] for nullable columns.
    • Parameterised queries — $1-style placeholders via the extended query protocol; ToValue trait with impls for all basic types and Option[T].
    • Codec-based encoding — per-PG-type codecs with OID dispatch; &Codec trait objects handle text and binary format transparently. Supports variant compatibility (e.g. String → JSON, Int → Int64).
    • Async concurrency — multiple connections can run queries concurrently; a slow query on one connection never blocks others.

    #Install

    This package targets native builds and has zero C dependencies — the wire protocol is implemented in pure MoonBit.

    moon add jaredzhou/moonpg

    #Quick start

    let conn = @moonpg.connect("postgres://user:pw@localhost:5432/db")

    // Simple query
    let rows = conn.query("SELECT id, name FROM users")
    while rows.has_next() {
    let row = rows.get_row()
    let id : Int = row.get(0)
    let name : String = row.get_by_name("name")
    println("\{id}: \{name}")
    }
    rows.close()

    // Parameterised — extended query protocol
    conn.execute(
    "UPDATE users SET email = $1 WHERE id = $2",
    params=[None, 42], // Option[T] encodes NULL
    ) |> ignore

    conn.close()

    #Connection pool

    let pool = Pool::new(PoolConfig::new(
    "postgres://user:pw@localhost:5432/db",
    max_conns=10,
    min_idle=2,
    ))

    // Implicit acquire — rows.close() returns conn to pool
    let rows = pool.query("SELECT 1")
    while rows.has_next() {
    let row = rows.get_row()
    // ...
    }
    rows.close()

    let row = pool.query_one("SELECT 42")
    pool.execute("INSERT INTO t (x) VALUES ($1)", params=[1]) |> ignore

    // Explicit acquire — caller controls release
    let pc = pool.acquire()
    let r = pc.execute("INSERT INTO t (x) VALUES ($1)", params=[99])
    r.close()
    pc.release()

    #TLS / SSL

    // System CA (default)

    ///|
    let conn = @moonpg.connect("postgres://user:pw@host/db?sslmode=require")

    // Custom CA certificate

    ///|
    let conn = @moonpg.connect(
    "postgres://user:pw@host/db?sslmode=verify-ca&sslrootcert=/etc/certs/ca.pem",
    )

    // Client certificate

    ///|
    let conn = @moonpg.connect(
    "postgres://user:pw@host/db?sslmode=require&sslcert=/etc/certs/client.pem&sslkey=/etc/certs/client.key",
    )

    Supported sslmodes: disable, allow, prefer, require, verify-ca, verify-full.

    #Connection timeouts

    // TCP connect timeout (seconds)
    @moonpg.connect("postgres://host/db?connect_timeout=5")

    // Server-side statement timeout (milliseconds)
    @moonpg.connect("postgres://host/db?statement_timeout=30000")

    #Target session attributes

    // Ensure we're talking to a read-write primary
    @moonpg.connect("postgres://host/db?target_session_attrs=read-write")

    // Read-only standby is fine
    @moonpg.connect("postgres://host/db?target_session_attrs=read-only")

    Values: any (default), read-write, read-only, primary, standby, prefer-standby.

    #Connection properties

    let pid = conn.backend_pid() // server process ID
    let ver = conn.param("server_version") // e.g. Some("16.4")
    let tz = conn.param("TimeZone") // e.g. Some("UTC")
    if conn.is_closed() { ... }

    #LISTEN / NOTIFY

    @async.with_task_group() <| group => {
    let listener = conn.listen("events", group)

    // Notify from another connection
    group.spawn_bg() <| () => {
    let c2 = @moonpg.connect(conninfo)
    c2.notify("events", payload="hello")
    }

    // Block until a notification arrives
    let notif = listener.recv()
    println("\{notif.channel}: \{notif.payload}")
    }

    #COPY protocol

    // Bulk insert from an iterator — one row in memory at a time
    conn.copy_in("COPY users (name, age) FROM STDIN", ["alice\t30", "bob\t25"].iter())

    // Streaming COPY writer
    let w = conn.begin_copy("users", ["name", "age"])
    w.write_row(["alice", 30])
    w.write_row(["bob", 25])
    let result = w.finish()

    #Pool with health check and stats

    let pool = Pool::new(PoolConfig::new(
    "postgres://user:pw@localhost:5432/db",
    max_conns=10,
    min_idle=2,
    max_idle_sec=300,
    max_lifetime_sec=3600,
    health_check=true, // ping idle conns before checkout
    maintenance_interval_sec=60, // background idle-pool sweeper
    ))

    // Start background maintenance
    pool.start_maintenance()

    // Inspect pool
    let stats = pool.stats()
    println("active=\{stats.active_connections} idle=\{stats.idle_connections}")

    pool.close()

    #Transactions

    // Auto-commit / rollback

    ///|
    let result = begin_func(conn, async fn(tx) {
    tx.execute("INSERT INTO users (name) VALUES ($1)", params=["alice"]) |> ignore
    tx.query_one("SELECT id FROM users WHERE name = $1", params=["alice"]).get(0)
    }) // exception → rollback; success → commit

    #Typed decoding

    FromValue decodes PostgreSQL cells into MoonBit types:

    ///|
    let id : Int = row.get(0) // raises on NULL

    ///|
    let email : String? = row.get_by_name("email") // NULL → None

    ///|
    let js : Json = row.get(2) // jsonb → Json

    ///|
    let ts : @value.Timestamp = row.get(3) // pg timestamptz → Unix µs

    #Architecture

    See arch.md for a detailed walkthrough of the codebase.

    #Run the tests

    Requires a running PostgreSQL instance. Set PGCONN to point at your database (defaults to postgres://postgres:111111@localhost:5432/postgres):

    # Uses default connection moon test --target native # Or with a custom connection PGCONN="postgres://user:pass@localhost:5432/mydb" moon test --target native

    #Authentication tests (optional)

    Three authentication tests (plain password, MD5, SCRAM-SHA-256) require separate users. Set the corresponding environment variable for each; tests are skipped when the variable is missing, so you can run the full suite without them.

    env varexample connstrauth method
    PG_PLAIN_CONNpostgres://moonpg_plain:plain_pass@localhost:5432/moonpg_testpassword
    PG_MD5_CONNpostgres://moonpg_md5:md5_pass@localhost:5432/moonpg_testmd5
    PG_SCRAM_CONNpostgres://moonpg_scram:scram_pass@localhost:5432/moonpg_testscram-sha-256

    #License

    Apache-2.0

    QueryExecutor

    pub(open) trait QueryExecutor {
    async fn query(Self, String, params? : Array[&
    ToValue
    ]) -> &Rows raise PgError
    async fn query_one(Self, String, params? : Array[&
    ToValue
    ]) -> Row raise PgError
    async fn execute(Self, String, params? : Array[&
    ToValue
    ]) -> ExecResult raise PgError
    }

    Types that can execute SQL queries.

    Rows

    pub(open) trait Rows {
    async fn has_next(Self) -> Bool raise PgError
    fn get_row(Self) -> Row raise PgError
    fn columns(Self) -> Array[
    FieldDescription
    ]
    async fn close(Self) -> Unit raise PgError
    }

    A pull-based row reader. Both ConnRows (bare connection) and PoolRows (pooled connection) implement this trait.

    while rows.has_next() {
    let row = rows.get_row()
    let v : Int = row.get(0)
    }
    rows.close()
    pub(open) trait Tx : QueryExecutor {
    async fn commit(Self) -> Unit raise PgError
    async fn rollback(Self) -> Unit raise PgError
    }

    A database transaction. Extends QueryExecutor with commit/rollback. Concrete impl: DbTx. Users can impl this trait on mock types for testing.

    TxBeginner

    pub(open) trait TxBeginner {
    async fn begin_tx(Self, opts? : TxOptions) -> &Tx raise PgError
    }

    Types that can start a transaction.

    PgError

    pub(all) suberror PgError {
    ConnectionError(String)
    QueryError(String)
    NoRows
    } derive(
    Debug
    )

    ConnRows

    pub(all) struct ConnRows {
    reader :
    ResultReader

    }

    Rows from a bare Connection. Wraps a wire ResultReader.
    impl Rows for ConnRows

    ConnStatus

    pub enum ConnStatus {
    OK
    Bad
    } derive(Eq,
    Debug
    )

    Connection status.

    Connection

    pub struct Connection {
    conn :
    RawConn

    }

    Wraps a wire-protocol RawConn.

    Connection::backend_pid

    fn Connection::backend_pid(self : Connection) -> Int

    Return the backend process ID assigned by PostgreSQL.

    Connection::begin_copy

    async fn Connection::begin_copy(self : Connection, table : String, columns : Array[String]) -> CopyWriter raise PgError

    Begin a COPY ... FROM STDIN operation on the given table and columns.

    Generates COPY table ("col1", "col2") FROM STDIN, sends the query, and returns a CopyWriter that accepts rows one at a time.

    Connection::close

    fn Connection::close(self : Connection) -> Unit

    Close the database connection.

    Connection::copy_in

    async fn Connection::copy_in(self : Connection, sql : String, rows : Iter[String]) -> ExecResult raise PgError

    Execute COPY ... FROM STDIN with rows from an iterator.

    Each element is one line of COPY data (tab-separated, newline-terminated). Rows are pulled one at a time — only a single row is held in memory.

    conn.copy_in("COPY t FROM STDIN", my_rows.iter())

    Connection::deallocate

    async fn Connection::deallocate(self : Connection, name : String) -> Unit raise PgError

    Deallocate a prepared statement.

    Sends Close(S) + Flush and waits for CloseComplete. Returns silently on success — raises PgError if the statement doesn't exist or the connection is broken.

    Passing an empty string closes the unnamed statement.

    Connection::is_closed

    fn Connection::is_closed(self : Connection) -> Bool

    Return true if the connection has been closed.

    Connection::listen

    async fn[X] Connection::listen(self : Connection, channel : String, group :
    TaskGroup
    [X]) -> Listener raise PgError

    Start listening for notifications on the given channel.

    Executes LISTEN channel, spawns a background receive loop on group, and returns a Listener. The connection is dedicated to listening.

    @async.with_task_group() <| group => {
    let listener = conn.listen("events", group)
    for ;; {
    let notif = listener.recv()
    group.spawn_bg() <| () => { handle(notif) }
    }
    }

    Connection::notify

    async fn Connection::notify(self : Connection, channel : String, payload? : String) -> Unit raise PgError

    Send a notification on the given channel.

    An optional payload string can be included.

    Connection::param

    fn Connection::param(self : Connection, key : String) -> String?

    Return a server parameter value (e.g. "server_version", "TimeZone").

    Returns None if the parameter was not reported by the server during startup.

    Connection::server_version

    fn Connection::server_version(self : Connection) -> Int

    Return the server version as an integer (e.g. 180004 for 18.0.4).

    Connection::status

    fn Connection::status(self : Connection) -> ConnStatus

    Return the current connection status.

    Connection::unlisten

    async fn Connection::unlisten(self : Connection, channel : String) -> Unit raise PgError

    Stop listening for notifications on the given channel.

    Passing "*" unlistens from all channels.

    CopyWriter

    pub(all) struct CopyWriter {
    conn :
    RawConn

    }

    Streaming writer for COPY ... FROM STDIN.

    Created via Connection::begin_copy. Rows are encoded and sent one at a time — only a single row is held in memory.

    Example

    let w = conn.begin_copy("users", ["name", "age"]) w.write_row(["Alice", 30]) w.write_row(["Bob", 25]) let result = w.finish()

    CopyWriter::finish

    async fn CopyWriter::finish(self : CopyWriter) -> ExecResult raise PgError

    Finish the COPY operation.

    Sends CopyDone, reads the server completion, and returns the ExecResult.

    CopyWriter::write_row

    async fn CopyWriter::write_row(self : CopyWriter, row : Array[&
    ToValue
    ]) -> Unit raise PgError

    Write a single row to the COPY stream.

    Each value is encoded to COPY text format (tab-separated, \N for NULL, backslash-escaped strings) and sent immediately — no batching in memory.

    DbTx

    pub(all) struct DbTx {
    conn : Connection
    }

    Concrete transaction wrapping a Connection.
    impl Tx for DbTx

    ExecResult

    pub struct ExecResult {
    tag :
    CommandTag

    }

    Execution result (for INSERT/UPDATE/DELETE/DDL).

    ExecResult::affected_rows

    fn ExecResult::affected_rows(self : ExecResult) -> Int

    Return the number of rows affected by an INSERT/UPDATE/DELETE.

    ExecResult::close

    fn ExecResult::close(_self : ExecResult) -> Unit

    Close the execution result. No-op.

    IdleConn

    pub(all) struct IdleConn {
    conn : Connection
    created_at : Int64
    idle_since : Int64
    }

    IsolationLevel

    pub(all) enum IsolationLevel {
    ReadCommitted
    RepeatableRead
    Serializable
    }

    Transaction isolation level.

    Listener

    Receiver for PostgreSQL notifications.

    Created by Connection::listen. Call recv() to block until a notification arrives.

    Listener::recv

    Block until a notification arrives.

    Pool

    pub(all) struct Pool {
    queue :
    Queue
    [IdleConn]
    conninfo : String
    max_conns : Int
    min_idle : Int
    max_idle_sec : Int
    max_lifetime_sec : Int
    health_check : Bool
    maintenance_interval_sec : Int
    count :
    Ref
    [Int]
    idle_count :
    Ref
    [Int]
    running :
    Ref
    [Bool]
    acquire_count :
    Ref
    [Int64]
    acquire_wait_count :
    Ref
    [Int64]
    acquire_wait_duration :
    Ref
    [Int64]
    }

    impl TxBeginner for Pool

    Pool::acquire

    async fn Pool::acquire(self : Pool) -> PoolConn raise PgError

    Pool::close

    fn Pool::close(self : Pool) -> Unit

    Pool::idle

    fn Pool::idle(self : Pool) -> Int

    Number of idle connections currently in the pool.

    Pool::maintain

    async fn Pool::maintain(self : Pool) -> Unit

    Run one round of maintenance:

    1. Drain all idle connections from the queue, closing expired ones.
    2. Put the good ones back.
    3. Create new connections if idle count is below min_idle.

    Called automatically by the background maintenance loop; may also be called manually.

    Pool::new

    async fn Pool::new(config : PoolConfig) -> Pool raise PgError

    Pool::start_maintenance

    async fn Pool::start_maintenance(self : Pool) -> Unit

    Start background maintenance.

    Periodically cleans expired idle connections and refills up to min_idle. Runs until Pool::close() is called. Call this once after creating the pool.

    When maintenance_interval_sec is 0 (the default) this is a no-op — maintenance happens lazily during acquire() / release() instead.

    Pool::stats

    fn Pool::stats(self : Pool) -> PoolStats

    Return a snapshot of pool statistics.

    Pool::total

    fn Pool::total(self : Pool) -> Int

    PoolConfig

    pub(all) struct PoolConfig {
    conninfo : String
    max_conns : Int
    min_idle : Int
    max_idle_sec : Int
    max_lifetime_sec : Int
    health_check : Bool
    maintenance_interval_sec : Int
    }

    Connection pool, inspired by pgxpool.

    Concurrency model

    MoonBit's async runtime is single-threaded + cooperative. The pool uses @aqueue.Queue(kind=Unbounded) for idle connections (sync try_get / try_put) and get() (async, blocks) when exhausted. A Ref[Int] counter tracks total connections — it is safe because increments happen before the first yield point (connect()).

    Basic usage

    let pool = Pool::new(PoolConfig::new("postgres://...", max_conns=10)) // Option A: implicit acquire — Pool auto-manages lifecycle let rows = pool.query("SELECT 1") while rows.has_next() { let row = rows.get_row() } rows.close() // PoolRows.close() returns conn to pool let row = pool.query_one("SELECT 42") pool.execute("INSERT ...") |> ignore let tx = pool.begin_tx() tx.commit() // PoolDbTx.commit() returns conn to pool // Option B: explicit acquire — caller manages lifecycle let pc = pool.acquire() defer pc.release() let rows = pc.query("SELECT 1") rows.close() // PoolRows.close() returns conn to pool let row = pc.query_one("SELECT 42") pc.release() // caller releases

    PoolConfig::new

    fn PoolConfig::new(conninfo : String, max_conns? : Int, min_idle? : Int, max_idle_sec? : Int, max_lifetime_sec? : Int, health_check? : Bool, maintenance_interval_sec? : Int) -> PoolConfig

    PoolConn

    pub(all) struct PoolConn {
    conn : Connection
    pool : Pool
    created_at : Int64
    released :
    Ref
    [Bool]
    }

    PoolConn::close

    fn PoolConn::close(self : PoolConn) -> Unit

    PoolConn::execute

    async fn PoolConn::execute(self : PoolConn, sql : String, params? : Array[&
    ToValue
    ]) -> ExecResult raise PgError

    PoolConn::query

    async fn PoolConn::query(self : PoolConn, sql : String, params? : Array[&
    ToValue
    ]) -> &Rows raise PgError

    PoolConn::query_one

    async fn PoolConn::query_one(self : PoolConn, sql : String, params? : Array[&
    ToValue
    ]) -> Row raise PgError

    PoolConn::release

    fn PoolConn::release(self : PoolConn) -> Unit

    PoolDbTx

    pub(all) struct PoolDbTx {
    inner : DbTx
    pc : PoolConn
    }

    Transaction from a pooled connection. Proxies DbTx for queries and releases the connection on commit() / rollback().
    impl Tx for PoolDbTx

    PoolRows

    pub(all) struct PoolRows {
    inner : &Rows
    pc : PoolConn
    }

    Rows from a pooled connection. Implements Rows — identical API to ConnRows, but close() returns the connection to the pool.
    impl Rows for PoolRows

    PoolStats

    pub(all) struct PoolStats {
    total_connections : Int
    idle_connections : Int
    active_connections : Int
    acquire_count : Int64
    acquire_wait_count : Int64
    acquire_wait_duration_ms : Int64
    }

    Snapshot of pool statistics.

    Row

    pub struct Row {
    values : Array[Bytes?]
    col_descs : Array[
    FieldDescription
    ]
    }

    A single row from a query result.

    Row::columns

    Return the column descriptions of this row's result set.

    Row::get

    fn[T :
    FromValue
    ] Row::get(self : Row, col : Int) -> T raise PgError

    Read the value at column col (zero-based index) as T.

    Row::get_by_name

    fn[T :
    FromValue
    ] Row::get_by_name(self : Row, name : String) -> T raise PgError

    Read the value of column name as T.

    TxOptions

    pub(all) struct TxOptions {
    isolation_level : IsolationLevel?
    read_only : Bool?
    deferrable : Bool?
    }

    Options for BEGIN — mirrors PostgreSQL's BEGIN ... parameters.

    TxOptions::default

    fn TxOptions::default() -> TxOptions

    begin_func

    async fn[T, B : TxBeginner] begin_func(beginner : B, f : async (&Tx) -> T) -> T raise PgError

    Execute a callback inside a transaction.

    1. Call beginner.begin_tx() to start a transaction.
    2. Execute f(tx).
    3. Success → COMMIT, error → ROLLBACK + re-raise.

    Works with any TxBeginner: Connection, Pool, PoolConn.

    Example

    let new_id = begin_func(conn, fn(tx) { let row = tx.query_one("INSERT INTO users (name) VALUES ($1) RETURNING id", params=["alice"]) row.get(0) })

    build_begin_sql

    fn build_begin_sql(opts : TxOptions) -> String

    Build a BEGIN SQL string from options.

    connect

    async fn connect(conninfo : String) -> Connection raise PgError

    Open a new connection to a PostgreSQL server.

    get_env

    fn get_env(name : String) -> String

    Read an environment variable.

    row_from_raw

    fn row_from_raw(raw_row : Array[Bytes?], col_descs : Array[
    FieldDescription
    ]) -> Row

    Build a Row from raw wire data and column descriptions, respecting the format code in each column. Used by integration tests that request binary result format.