#jaredzhou/moonpg

#Overview

• 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 — FromRaw 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].
• Async concurrency — multiple connections can run queries concurrently; a slow query on one connection never blocks others.

#Install

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",
)

#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")

#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

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

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

///|
let ts : Timestamp = row.get(2) // pg timestamp → Unix µs

#Architecture

#Run the tests

# Use env var or defaults (see conn_test.mbt)
MOONPG_CONNINFO="postgres://moonpg_plain:plain_pass@localhost:5432/moonpg_test" \
  moon test --target native

#License