Copy page

Developer Guide

Repository

The Repository is the low-level engine that turns entity metadata + the Query DSL into parameterized SQL. The Dal gateway delegates to an internal Repository(pool), so dal.findById(...) and repo.findById(...) are the same call. Inside dal.transaction(txRepo => …) the callback gets a tx-scoped Repository — see Connections & transactions.

Code
import { getDal, Repository } from "@primebrick/dal-pg";

const dal = getDal({ connectionString: process.env.DATABASE_URL!, schemas: ["myapp"] });

// dal.add, dal.findAll, dal.update, ... all delegate to an internal Repository.
// Inside a transaction the callback receives a tx-scoped Repository:
await dal.transaction(async (txRepo) => {
  await txRepo.add(CustomerEntity, { name: "TX" }, { actor: "tx" });
  // COMMIT on return, ROLLBACK on throw — never manual.
});

Finders

findById(entity, id, options?)

Find a single row by its @Key() column. id is bigint | string (matches bigint PKs). Throws NotFoundError by default if no row matches; pass throwIfNotFound: false to return null instead.

Code
const found = await dal.findById(CustomerEntity, 42n);
// SELECT ... FROM customers WHERE id = $1 LIMIT 1

const maybe = await dal.findById(CustomerEntity, 42n, { throwIfNotFound: false });
// maybe is null if no row matches

findByUUID(entity, uuid, options?)

Find a single row by its UUID. Assumes the entity has a uuid column (the convention enforced by IExposableEntity). Same throw/return semantics as findById.

Code
const found = await dal.findByUUID(CustomerEntity, "550e8400-e29b-41d4-a716-446655440000");

find(entity, projections, options?)

Find a single row matching the filters. Throws NotFoundError by default if zero rows match; throws MultipleRowsError if more than one row matches. Pass throwIfNotFound: false to return null instead of throwing on zero rows.

Code
import { Filter, field } from "@primebrick/dal-pg";

const found = await dal.find(CustomerEntity, null, {
  filters: [Filter.fieldValue(field(CustomerEntity, "email"), "=", "alice@example.com")],
});
// SELECT ... FROM customers WHERE email = $1 LIMIT 2
// (LIMIT 2 so the DAL can detect >1 row and throw MultipleRowsError)

findAll(entity, projections, options?)

Find all rows matching the filters. Returns TResult[] (empty array if no matches). Pass stream: true to get an AsyncIterable instead — see Streaming below.

Code
import { Filter, Sort, field } from "@primebrick/dal-pg";

const rows = await dal.findAll(CustomerEntity, null, {
  filters: [Filter.fieldValue(field(CustomerEntity, "is_active"), "=", true)],
  sorting: [Sort.by(field(CustomerEntity, "created_at"), "DESC")],
});

findByPage(entity, projections, options?)

Paginated find. Returns PaginatedEntity<TEntity> — { entities, total_records }. total_records is bigint (PostgreSQL COUNT(*) returns bigint). Pass limit and offset on the options.

Code
const page = await dal.findByPage(CustomerEntity, null, {
  filters: [Filter.fieldValue(field(CustomerEntity, "is_active"), "=", true)],
  sorting: [Sort.by(field(CustomerEntity, "created_at"), "DESC")],
  limit: 20,
  offset: 0,
});
console.log(page.entities.length);          // ≤ 20
console.log(page.total_records);            // bigint — total matching rows

count(entity, options?)

Count rows matching the filters. Returns bigint.

Code
const total = await dal.count(CustomerEntity, {
  filters: [Filter.fieldValue(field(CustomerEntity, "is_active"), "=", true)],
});
// total is bigint — use Number(total) if you know it's < 2^53

FindOptions

Field	Type	Default	Purpose
throwIfNotFound	boolean	true (for find)	Throw NotFoundError on zero rows
deletedRecords	"EXCLUDED" | "ONLY" | "INCLUDED"	"EXCLUDED"	How to handle soft-deleted rows
filters	FilterExpr[]	[]	WHERE predicates
sorting	SortingExpr[]	[]	ORDER BY clauses
joins	JoinExpr[]	[]	JOIN clauses
stream	boolean	false	Stream results via pg-query-stream
tableName	string	entity's table	Override the table name (e.g. for audit tables)

deletedRecords modes

Mode	Behavior
"EXCLUDED" (default)	WHERE deleted_at IS NULL — soft-deleted rows invisible
"INCLUDED"	No deleted_at filter — all rows returned
"ONLY"	WHERE deleted_at IS NOT NULL — only soft-deleted rows returned

Code
// Default — exclude soft-deleted:
await dal.findAll(CustomerEntity, null, {});

// Include soft-deleted:
await dal.findAll(CustomerEntity, null, { deletedRecords: "INCLUDED" });

// Only soft-deleted (e.g. for a trash bin UI):
await dal.findAll(CustomerEntity, null, { deletedRecords: "ONLY" });

Writes

Every write returns the full hydrated row (RETURNING *). For auditable entities, pass AuditableWriteOptions (which requires actor); for non-auditable entities, pass plain WriteOptions.

add(entity, values, options?)

Insert a single row. The DAL auto-fills created_at, created_by, updated_at, updated_by, version for auditable entities — you only pass business fields.

Code
const created = await dal.add(CustomerEntity, {
  name: "Alice",
  email: "alice@example.com",
  is_active: true,
}, { actor: "system" });
// INSERT INTO customers (name, email, is_active, created_at, created_by, updated_at, updated_by, version)
// VALUES ($1, $2, $3, now(), $4, now(), $4, 1) RETURNING *
// created.version === 1

upsert(entity, values, options?)

Insert or update on conflict. conflictTarget defaults to the @Key() column; override with UpsertOptions.conflictTarget (e.g. "email" for a natural key). The version guard applies on the ON CONFLICT (update) branch only — the pure-INSERT branch starts at version = 1.

Code
const upserted = await dal.upsert(CustomerEntity, {
  uuid: "550e8400-e29b-41d4-a716-446655440000",
  name: "Alice 2",
  email: "alice@example.com",
  version: 3,
}, {
  actor: "system",
  conflictTarget: "uuid",
});
// INSERT ... ON CONFLICT (uuid) DO UPDATE SET ... WHERE customers.version = $expected

update(entity, updates, options?)

Update a single row — exactly one row, guaranteed. The row is identified by identity fields carried in updates itself; matchBy is only needed for non-standard selectors. For auditable entities, updates must include the version you read — the version guard runs atomically in the same UPDATE statement. See Optimistic locking.

Identity resolution (automatic):

• id and uuid present in updates always become AND match conditions — supplying both acts as a consistency check: if they point at different rows the write fails with ERR10 (412).
• matchBy may add alternate @Unique selectors: a prop belonging to a composite @Unique group pulls in the whole group — every group prop must be present in updates (e.g. matchBy: ["dtype", "model_id"]).
• A matchBy prop that is not @Unique/@Key, an unknown prop, an incomplete composite group, or a payload with no identity field at all → ERR09 (422) before any SQL is emitted.
• Identity fields are never SET columns — uuid/id cannot be rewritten by update(). A matched/guard CTE counts the rows the identity would touch; if it resolves to >1 row the statement aborts with ERR10 and nothing is written (atomic).

Code
// Match by uuid (auto — uuid is in the payload):
const updated = await dal.update(
  CustomerEntity,
  { uuid: "550e8400-…", name: "Alice 2", version: 3 },
  { actor: "system" },
);
// UPDATE customers SET name = $1, updated_at = …, version = version + 1
// WHERE uuid = $N AND version = $M RETURNING *

// Match by PK + uuid (AND cross-check — mismatch → ERR10):
await dal.update(CustomerEntity, { id: 42n, uuid: "550e8400-…", name: "A", version: 3 }, { actor: "s" });

// Match by a composite @Unique group:
await dal.update(
  AiModelEntity,
  { dtype: "chat", model_id: "qwen3", status: "ready", version: 7 },
  { actor: "s", matchBy: ["dtype", "model_id"] },
);

Unique conflicts. When updates touches a column belonging to a declared @Unique group, the UPDATE statement carries a conflict CTE that checks whether another live/soft-deleted row already owns the attempted values — raising ERR04/ERR05 with detail { entity, table, uuid, constraint, keys } (keys = the attempted unique values; composite groups include untouched column values read from the target row). The conflict check is gated on the target row passing the version guard, so ERR01/ERR03 always win over a unique conflict. NULL attempted values never conflict (unique indexes are NULLS DISTINCT). The check only runs when the payload touches @Unique columns — a plain name-only UPDATE emits the same SQL as before.

delete(entity, match, options?) — soft delete

Soft-delete a row: sets deleted_at and deleted_by, increments version. The row stays in the table and is excluded from default finders. Returns the deleted row. match carries the identity fields — same resolution rules as update() (auto id/uuid, optional matchBy @Unique selectors, ERR09/ ERR10).

Code
const deleted = await dal.delete(
  CustomerEntity,
  { uuid: "550e8400-…", version: 3 },
  { actor: "system" },
);
console.log(deleted.deleted_at); // Date

restore(entity, match, options?)

Restore a soft-deleted row: clears deleted_at and deleted_by, increments version. Returns the restored row.

Code
const restored = await dal.restore(
  CustomerEntity,
  { uuid: "550e8400-…", version: 3 },
  { actor: "system" },
);
console.log(restored.deleted_at); // null

hardDelete(entity, match, options?)

Permanently delete a row (DELETE FROM ...). The row is gone — no soft-delete flag, no restore. The version guard and the same identity-resolution rules as update() apply (auto id/uuid, ERR09/ERR10). Returns void.

Code
await dal.hardDelete(CustomerEntity, { uuid: "550e8400-…", version: 3 }, { actor: "system" });
// DELETE FROM customers WHERE uuid = $1 AND version = $2

WriteOptions and AuditableWriteOptions

Field	Type	Required for	Purpose
actor	string	auditable entities	Stamped into created_by/updated_by/deleted_by
audit	AuditPort	optional	If injected, write ops emit field-level audit deltas
logger	LoggerPort	optional	If injected, audit errors are logged instead of swallowed
tableName	string	optional	Override the table name (e.g. for audit tables)

MatchByOptions

Field	Type	Default	Purpose
matchBy	keyof TEntity & string or ReadonlyArray<keyof TEntity & string>	auto id/uuid from payload	Alternate @Unique selector — every prop must be unique; composite groups require all group props in the payload. Same rules for single-row and *Many ops.

Bulk operations

All *Many ops share the same contract:

• Atomic — one dedicated transaction; any failing row rolls back the whole operation. There is no partial persistence.
• TEMP TABLE strategy — rows are bulk-loaded into a temp table, then a single set-based statement joins it to the target (no per-row round-trips, no 65535-parameter limit).
• BulkResult return — { received, affected }. received is the row count submitted; affected is the count actually written. On success they are always equal (a mismatch would have rolled back).
• Same identity contract as single-row writes — id/uuid carried by EVERY item are auto-matched (AND); matchBy accepts a prop or prop array of @Unique selectors (composite groups require all group props in every item). The identity must be uniform across the batch: an identity prop present in only some items is ERR09. Illegal selectors are ERR09 before any SQL runs.
• Per-item uniqueness guard — each temp-table item must match at most one target row. A set-based aggregate (GROUP BY row_ix HAVING count>1) raises ERR10 and aborts the whole transaction before any write; the detail lists the offending row_ix values. |
• Per-row version guard (auditable entities) — every item must carry the caller-observed version; it lands in the temp table as expected_version and is enforced in the write join (target.version = tmp.expected_version). Offenders are reported in ONE pg_raise whose detail jsonb carries { entity, table, stale|missing, rows[≤10] } with per-row uuid/expected_version/actual_version/vanished/code:

Failure	Code
Row submitted without version	ERR02
Row exists but version differs (modified meanwhile)	ERR01
Row is gone (hard-deleted meanwhile)	ERR03
Item identity is incoherent (e.g. id of row A + uuid of row B)	ERR10
Item matches more than one target row	ERR10 (pre-write guard)

addMany(entity, rows, options?)

Batched multi-row INSERT through the temp-table strategy. Unique-constraint violations are collected and raised as ONE pg_raise — ERR04 when the conflicting row is live, ERR05 when every conflict hits a soft-deleted row (restore it instead of duplicating). err.detail carries { entity, table, conflicts, rows[≤10] } — each row with uuid, input_uuid, per-row code (ERR04/ERR05), deleted, constraint, and keys (the attempted unique-key values).

Code
const res = await dal.addMany(CustomerEntity, [
  { name: "Alice", email: "alice@x.com" },
  { name: "Bob",   email: "bob@x.com" },
], { actor: "system" });
// res = { received: 2, affected: 2 }

updateMany(entity, rows, options?)

Batched UPDATE: UPDATE target SET ... FROM tmp WHERE target.<id> = tmp.<id> AND target.<uuid> = tmp.<uuid> AND target.version = tmp.expected_version — the join predicate is the AND of every identity column. Atomic and SQL-injection safe.

Code
const res = await dal.updateMany(CustomerEntity, [
  { uuid: "uuid-1", name: "Alice 2", version: 2 },
  { uuid: "uuid-2", name: "Bob 2",   version: 2 },
], { actor: "system", timeoutMs: 60_000 });
// Every row MUST include its caller-observed version.
// uuid is auto-matched — matchBy:"uuid" is unnecessary.
// matchBy: ["model_id", "dtype"] selects a composite @Unique group instead.

deleteMany(entity, matches, options?)

Bulk soft-delete with the same version guard. matches is an array of match objects — { uuid, version } for auditable entities.

Code
const res = await dal.deleteMany(CustomerEntity, [
  { uuid: "uuid-1", version: 3 },
  { uuid: "uuid-2", version: 1 },
], { actor: "system", matchBy: "uuid" });

restoreMany(entity, matches, options?)

Bulk restore (clears soft-delete fields, bumps version, stamps audit fields). Same temp-table + version-guard semantics as deleteMany.

upsertMany — currently DISABLED

upsertMany is commented out pending a design decision: its ON CONFLICT DO UPDATE branch historically overwrote conflicting rows without a version check. It will return either with a guarded expected_version on the conflict branch or as an explicit sync/import-semantics operation. Do not call it.

BulkOptions

Field	Type	Default	Purpose
batchSize	number	auto-calculated	Override the temp-table batch size (rarely needed)
timeoutMs	number	DalConfig.bulkTimeoutMs (30min)	Whole-operation wall-clock budget in ms — applied as SET LOCAL statement_timeout per statement plus a JS deadline across the op; expiry throws BulkTimeoutError (ERR06) and rolls back

Clone

clone(entity, sourceUuid, options)

Copy a record by UUID. Fetches the source (including soft-deleted rows), excludes the PK and @Unique columns, stamps the @CloneField column with the source UUID, resets audit/deletable fields, and inserts the new row. See Clone for the full 7-step flow.

Code
const cloned = await dal.clone(CustomerEntity, sourceUuid, { actor: userUuid });
// cloned.uuid is a new random UUID
// cloned.cloned_from === sourceUuid
// cloned.version === 1

Streaming

findAll with stream: true returns an AsyncIterable backed by pg-query-stream. Each FETCH batch is bounded by the session statement_timeout — safe for very large result sets that would otherwise exhaust memory.

Code
const stream = await dal.findAll(CustomerEntity, null, {
  stream: true,
}) as AsyncIterable<typeof CustomerEntity.prototype>;

for await (const row of stream) {
  console.log(row.id, row.name);
  // row-by-row streaming — no buffering of the full result set
}

The stream is a Node.js async iterator; you can use for await ... of, pipe it through a transform, or collect it with Array.fromAsync (Node 22+).

Error handling

Every DAL error extends the abstract DalError class, which adds a stable code: string property. Branch on instanceof or on err.code — both work.

Code
import {
  NotFoundError, MultipleRowsError, UnknownColumnError, ValidationError,
  MissingVersionError, RecordVanishedError, OptimisticLockError,
  DalErrorCodes,
} from "@primebrick/dal-pg";

try {
  await dal.findByUUID(CustomerEntity, "nonexistent-uuid");
} catch (err) {
  if (err instanceof NotFoundError) {
    // err.code === "NOT_FOUND"
  } else if (err instanceof MissingVersionError) {
    // err.code === DalErrorCodes.ERR02 ("ERR02")
  } else if (err instanceof RecordVanishedError) {
    // err.code === DalErrorCodes.ERR03 ("ERR03")
  } else {
    throw err;
  }
}

Error codes at a glance

Code	Raised when	HTTP (via mapDalError)
ERR01	Stale version on a guarded write (single or bulk row)	409
ERR02	version missing on an auditable-entity write	400
ERR03	Target row vanished (hard-deleted) before the write	404
ERR04	Unique conflict against a live row	409
ERR05	Unique conflict(s) against soft-deleted rows only	409
ERR06	Bulk op exceeded its wall-clock timeoutMs budget (rolled back)	408
ERR07	PG 57014 — a statement was killed by statement_timeout	500
ERR08	Raw 23505 on a constraint not declared as @Unique (manual/deferred/partial index) — signals a missing @Unique in metadata	409
ERR09	Illegal match selector: matchBy prop not @Unique/@Key, incomplete composite group, or no identity field in the payload — thrown pre-SQL	422
ERR10	Identity incoherence (id+uuid pointing at different rows) or the match resolved to >1 row (guard CTE abort, nothing written)	412
NOT_FOUND / VALIDATION / UNKNOWN_COLUMN / MULTIPLE_ROWS	Finder/validation failures	404 / 400

The DAL never emits HTTP status codes — the @primebrick/sdk mapDalError function performs the pure err → { status, body } RFC 7807 mapping shared by the BE and all microservices. Bulk raises carry structured detail (≤10 offending rows) which the boundary surfaces as extra.issues.

See Architecture: error handling philosophy for the full code table, and Optimistic locking for the ERR01/ERR02/ERR03 flow.

Next steps

• Connections & transactions — pool config, dal.transaction, per-call timeouts, graceful shutdown.
• Query DSL — the expressions passed to finders.
• Audit trail — the AuditPort system that hooks into write ops.
• Optimistic locking — the version guard on auditable writes.
• Clone — the 7-step clone flow.
• API reference — the full mechanical listing of every exported symbol.