Copy page

Developer Guide

Connections & transactions

The Dal gateway (getDal()) is the recommended entry point. It owns the pg.Pool, registers type parsers once per process, and sets search_path, statement_timeout, and application_name on every connection. The existing Repository class stays exported as the low-level engine — dal.transaction hands a tx-scoped instance to the callback; tests may inject a mock Queryable.

getDal(config) — the singleton gateway

getDal(config) returns a process-wide singleton. Calling it twice with the same config returns the same Dal instance; the pool is created once and reused for every request. Use resetDal() only in tests to tear down the singleton between cases.

Production consumers don't call getDal(config) inline — they wrap it in their own initDal() (called once from index.ts, before migrations) and use getDal() as the zero-arg accessor everywhere else. Calling getDal() before bootstrap throws "config is required on the first call" — the sanctioned fail-fast. See getting started.

Code
import "reflect-metadata";
import { getDal, type Dal } from "@primebrick/dal-pg";

const dal = getDal({
  connectionString: process.env.DATABASE_URL!,
  schemas: ["myapp"],
  max: 10,
  statementTimeoutMs: 30_000,
  connectionTimeoutMillis: 5_000,
  idleTimeoutMillis: 30_000,
  applicationName: "my-service",
});

DalConfig

Field	Type	Default	Purpose
connectionString	string	(required)	PostgreSQL connection string
schemas	string[]	(required, non-empty)	Declared schema domain. search_path on every connection = schemas + public (extension/type resolution only). Every @Entity schema pin must belong to this set — violations throw SchemaDomainViolationError at getDal() (boot guard) and on every entity operation (runtime guard), before SQL is built
max	number	10	Maximum pool size. Formula: max ≤ (PG max_connections − reserved) / service_instances
statementTimeoutMs	number	30000	Per-statement timeout in ms, set via SET statement_timeout. Set to 0 to disable
connectionTimeoutMillis	number	5000	Time to wait when acquiring a connection before erroring. Fail fast when pool exhausted
idleTimeoutMillis	number	30000	How long an idle connection is kept before closing
maxUses	number	undefined (off)	Recycle connections after N uses to clear per-session state
applicationName	string	"primebrick-dal"	Set as application_name for PG-side observability (pg_stat_activity)

What the constructor does

1. ensureTypeParsers() — registers INT8 → bigint, NUMERIC → string, and DATE → string parsers idempotently. Called once per process; subsequent calls are no-ops.
2. new pg.Pool(config) — creates the pool with the merged config and an onConnect hook.
3. configureSchemaDomain — registers the declared schemas as the process-wide entity domain. Entities pinned to a schema outside the set fail before any SQL is emitted.
4. onConnect — runs on every new connection: SET search_path TO <schemas…>, public, SET statement_timeout TO <ms>, SET application_name TO '<name>'. The public tail exists only so extension types (vector, …) resolve — unqualified table access outside the declared domain is rejected by the entity guard, not by the path. The application_name is single-quote-escaped to prevent injection.
5. Per-request — dal.add(...) acquires a connection from the pool, runs the parameterized SQL, hydrates the result row, and releases the connection. No per-request allocation.

Pool lifecycle

dal.getPool()

Returns the underlying pg.Pool. Exposed for migration tooling, snapshot scripts, or anything that needs raw access. Don't use it for normal queries — go through dal.add/findAll/etc. so the Repository handles metadata and hydration.

Code
const pool = dal.getPool();
await pool.query("VACUUM ANALYZE customers");

dal.close(timeoutMs = 10000)

Graceful shutdown — drains the pool with a timeout deadline. Re-entrant (concurrent calls return immediately; the first call wins). If pool.end() throws, the error is logged and swallowed. Does not install process.on() handlers — that is a consumer-side concern.

Code
process.on("SIGTERM", async () => {
  await dal.close();      // waits up to 10s for pool.end()
  process.exit(0);
});

// Custom deadline:
await dal.close(5_000);   // wait up to 5s

Transactions with dal.transaction

dal.transaction(fn) acquires a pooled client, emits BEGIN, runs fn(txRepo) with a tx-scoped Repository, then COMMITs — or ROLLBACKs if fn throws. The callback never sees the raw PoolClient: every statement inside the transaction goes through the entity/DSL surface (add, update, findAll, …), so the schema-domain guard, audit stamping, and optimistic locking all apply inside the tx exactly as outside.

Code
await dal.transaction(async (txRepo) => {
  const customer = await txRepo.add(CustomerEntity, { name: "Alice" }, { actor: "tx" });
  await txRepo.add(OrderEntity, { customer_id: customer.id, total: 100 }, { actor: "tx" });
});

Commit/rollback are never manual: the end of the callback is the commit, a thrown error is the rollback. There is no way to leak a connection or forget COMMIT.

dal.withClient — which used to expose the raw PoolClient — is now private to dal-pg internals (it powers transaction and partition maintenance). It is not part of the consumer API.

Deferred transactions with dal.beginTransaction

For straightforward dependent writes, dal.beginTransaction() is the preferred shape: it returns a DeferredTransaction whose add, update, delete, restore, hardDelete (and read ops) do NOT execute — they buffer. commit() compiles the whole buffer into one data-modifying CTE statement (WITH s0 AS (…RETURNING…), s1 AS (…), …) and runs it — atomicity is the single-statement guarantee; there is no BEGIN/COMMIT/ ROLLBACK code at all.

Code
const tx = dal.beginTransaction();
const msg = tx.add(AiMessageEntity, { conversation_uuid, body }, { actor });
tx.update(
  AiConversationEntity,
  { uuid: conversation_uuid, version },          // optimistic guard
  { last_seen_at: msg.ref("created_at") },        // flows s0→s1 in SQL
);
const { results } = await tx.commit();           // ONE round-trip
const saved = msg.value;                          // materialized row

Inter-op data flow uses deferred.ref("col") — compiled to a (SELECT col FROM sN) scalar subquery — and Deferred.value throws if read before commit. The hard limit: the buffer cannot branch on intermediate DB results (nothing exists until commit) — flows that need conditional TS logic on a read still use transaction(fn).

Per-call timeout override

transaction accepts an optional { timeoutMs } — it emits SET LOCAL statement_timeout right after BEGIN, so the override is transaction-scoped and auto-resets at COMMIT/ROLLBACK (no leakage to the session default).

Code
// A long-running transactional import that needs 2 minutes:
await dal.transaction(async (txRepo) => {
  await txRepo.addMany(ImportEntity, rows, { actor: "import" });
}, { timeoutMs: 120_000 });

The bulk operations (addMany, updateMany, deleteMany, …) accept options.timeoutMs directly — they emit SET LOCAL statement_timeout inside their own transaction (transaction-scoped, no leakage).

Type parsers

The Dal gateway registers two type parsers once per process via ensureTypeParsers():

PG type OID	JS type	Reason
INT8_OID (20)	bigint	Preserve precision for PKs and counts above 2^53
NUMERIC_OID (1700)	string	Lossless — Decimal columns hydrate via the column contract
DATE_OID (1082)	string	"YYYY-MM-DD" — a calendar date has no timezone (see ISODate)

Without these parsers, node-postgres returns int8 as a string (to avoid precision loss) and numeric as a string. The Dal gateway opts into native bigint and number because:

• bigint PKs are the convention in Primebrick (id bigint).
• numeric columns are usually numeric(18,2) money/quantity fields that fit safely in a JS number.

If you have a numeric column that can exceed Number.MAX_SAFE_INTEGER, cast it to string in your query (SELECT my_col::text) or handle it explicitly.

JSON serialization caveat

JSON.stringify throws on bigint (TypeError: Do not know how to serialize a BigInt). When returning bigint values over HTTP, use @primebrick/sdk's Ext-JSON middleware (which serializes bigint as a string with a $n tag), or cast with Number(value) when you know the value is below 2^53.

Pool sizing guidance

The pool size is the single most impactful knob for high-async REST traffic:

• Too small → requests queue waiting for a connection; connectionTimeoutMillis errors under load.
• Too large → PostgreSQL server-side connection contention; pg_stat_activity shows many idle connections holding memory.

The formula (from the DalConfig docs):

Code
max ≤ (PG max_connections − reserved) / service_instances

For a single service against a PG server with max_connections = 100 and 20 reserved for admin/migrations, max = 10 leaves headroom for 8 service instances (8 × 10 = 80 ≤ 80). Scale horizontally by adding PG connections or service instances, not by cranking max to 100.

The other anti-throttling knob is statementTimeoutMs (default 30s). A slow query holding a connection starves the pool; the per-session timeout guarantees connection release. Lower it (e.g. 5s) for latency-sensitive services; raise it (e.g. 60s) for analytical services, or use per-call timeoutMs overrides for the occasional long query.

Putting it all together

A complete bootstrap module with graceful shutdown:

Code
// src/db.ts
import "reflect-metadata";
import { getDal, resetDal, type Dal } from "@primebrick/dal-pg";

let _dal: Dal | undefined;

export function getDb(): Dal {
  if (!_dal) {
    _dal = getDal({
      connectionString: process.env.DATABASE_URL!,
      schema: "myapp",
      max: 10,
      statementTimeoutMs: 30_000,
      connectionTimeoutMillis: 5_000,
      applicationName: "my-service",
    });
  }
  return _dal;
}

export async function closeDb(): Promise<void> {
  if (_dal) {
    await _dal.close();
    resetDal();
    _dal = undefined;
  }
}

Code
// src/index.ts
import "reflect-metadata";
import { getDb, closeDb } from "./db.js";
import { startHttpServer } from "./http.js";

async function main() {
  const dal = getDb();
  process.on("SIGTERM", async () => {
    await closeDb();
    process.exit(0);
  });
  await startHttpServer(dal);
}

main().catch((err) => {
  console.error(err);
  process.exit(1);
});

Next steps

• Repository — the methods you call on dal/repo.
• Architecture — how the gateway, pool, and Repository fit together.
• Getting started — the end-to-end bootstrap walkthrough.
• Optimistic locking — how statement_timeout interacts with the version guard.