# Repository


The `Repository` is the low-level engine that turns entity metadata + the
[Query DSL](./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](./connections).

```typescript
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.

```typescript
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`.

```typescript
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.

```typescript
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](#streaming) below.

```typescript
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.

```typescript
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`.

```typescript
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 |

```typescript
// 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.

```typescript
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`.

```typescript
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](./optimistic-lock).

**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).

```typescript
// 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`).

```typescript
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.

```typescript
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`.

```typescript
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).

```typescript
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.

```typescript
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.

```typescript
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](./clone) for the full 7-step flow.

```typescript
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.

```typescript
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.

```typescript
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](./architecture#error-handling-philosophy)
for the full code table, and [Optimistic locking](./optimistic-lock) for the
`ERR01`/`ERR02`/`ERR03` flow.

## Next steps

- [Connections & transactions](./connections) — pool config, `dal.transaction`, per-call timeouts, graceful shutdown.
- [Query DSL](./query-dsl) — the expressions passed to finders.
- [Audit trail](./audit-trail) — the `AuditPort` system that hooks into write ops.
- [Optimistic locking](./optimistic-lock) — the version guard on auditable writes.
- [Clone](./clone) — the 7-step clone flow.
- [API reference](./api-reference) — the full mechanical listing of every exported symbol.