@zudojs/database

PostgreSQL data-access layer for Zudojs applications, built on Prisma 7: a lifecycle-aware client, generic repositories with soft delete, a query builder that translates to Prisma where clauses, managed transactions, keyset pagination with signed cursors, migration and seed runners guarded by advisory locks, health checks with reconnect, and a bounded in-memory cache.

Documentation: zudojs.oyinlola.site/docs/packages-database · For AI agents: Markdown version, llms.txt

Installation

npm install @zudojs/database @prisma/client @prisma/adapter-pg

@prisma/client (>=7 <8) is a peer dependency. Prisma 7 connects through a driver adapter, so you also need an adapter package for your database (@prisma/adapter-pg for PostgreSQL).

Quick Start

import { PrismaClient } from "@prisma/client";
import { PrismaPg } from "@prisma/adapter-pg";
import {
  BaseRepository,
  createDatabaseClient,
  withTransaction,
} from "@zudojs/database";

interface User {
  id: string;
  email: string;
  name: string;
  deletedAt: Date | null;
}

// 1. Build the Prisma client with a driver adapter and hand it to the wrapper.
const prisma = new PrismaClient({
  adapter: new PrismaPg({ connectionString: process.env.DATABASE_URL }),
});
const client = createDatabaseClient({ prisma });
await client.connect();

// 2. Repositories wrap a Prisma model delegate.
class UserRepository extends BaseRepository<User> {
  constructor(delegate: ConstructorParameters<typeof BaseRepository<User>>[0]) {
    super(delegate, { modelName: "User", softDelete: true });
  }

  findByEmail(email: string) {
    return this.findOne({ email });
  }
}

const users = new UserRepository(prisma.user);
const alice = await users.create({ email: "[email protected]", name: "Alice" });
await users.softDelete(alice.id); // sets deletedAt; reads now skip the row

// 3. Transactions: rebind the repository to the transaction client.
await withTransaction(client, async (tx) => {
  const txUsers = users.withTransaction(tx);
  await txUsers.restore(alice.id);
  await txUsers.update(alice.id, { name: "Alice Doe" });
});

await client.disconnect();

A generated model delegate such as prisma.user is passed as-is, with no cast. RepositoryDelegate accepts any Prisma-style delegate whose rows match the entity type; this.delegate inside a subclass is typed with the arguments the repository passes (RepositoryDelegateOperations).

createDatabaseClient accepts either a pre-built prisma instance or an adapter (in which case it constructs the PrismaClient for you). It throws a DatabaseError if neither is supplied.

Typed transaction clients

The transaction client handed to transaction() callbacks is inferred from the prisma instance you pass, so model delegates keep the types of your generated client. This works with Prisma 7's prisma-client generator (client generated into your application, e.g. src/generated/prisma) and with the legacy prisma-client-js generator alike; the package's published types never import @prisma/client, so they compile with skipLibCheck: false.

import { PrismaClient } from "./generated/prisma/client.js";

const client = createDatabaseClient({ prisma: new PrismaClient({ adapter }) });

await client.transaction(async (tx) => {
  await tx.user.create({ data: { email: "[email protected]", name: "Bob" } }); // fully typed
});

withTransaction, withTransactionRetry, TransactionManager, DatabaseUnitOfWork, the Database facade, the lock manager and the migration and seed runners all take the type from the client they wrap.

When the concrete client type is unknown (new DatabaseClient(options), createDatabaseClient({ adapter }), getDatabase()), callbacks receive DatabaseTransactionContext: the raw-query surface ($queryRaw, $executeRaw, $queryRawUnsafe, $executeRawUnsafe) without model delegates. Pass the client type to opt in, for example createDatabaseClient<PrismaClient>({ adapter }). TransactionClientOf<typeof prisma> names the inferred type if you need to annotate it.

The adapter-only form constructs PrismaClient from @prisma/client at runtime, which only exists with the prisma-client-js generator. With the prisma-client generator, construct the client yourself and pass it as prisma.

Writing your own transaction client (a hand-written adapter, a test double, a PGlite wrapper): since 1.4.0 DatabaseTransactionContext is a real structural type rather than a type that resolved to any, so an object you return as one must implement all four raw-query methods — $queryRaw, $executeRaw, $queryRawUnsafe and $executeRawUnsafe — or TypeScript reports TS2739. Code that only receives a transaction client is unaffected.

Querying

import { createQueryBuilder, toPrismaWhere, and, equals, isAfter } from "@zudojs/database";

const query = createQueryBuilder<"email" | "createdAt" | "role">()
  .where("role", "admin")
  .whereContains("email", "@example.com")
  .orderByDesc("createdAt")
  .paginate({ page: 1, limit: 20 });

// Repositories consume the builder directly ...
const admins = await users.findByQuery(query);

// ... or translate the filter yourself.
const where = toPrismaWhere(
  and(equals("role", "admin"), isAfter("createdAt", new Date("2026-01-01"))),
);

Pagination

// Offset pagination
const page = await users.findPaginated(undefined, {
  pagination: { page: 2, limit: 25 },
  sort: [{ field: "createdAt", direction: "desc" }],
});
page.meta.totalPages;

// Keyset pagination with HMAC-signed cursors (set `cursorSecret` on the repository)
const first = await users.paginateCursor(undefined, {
  limit: 25,
  sort: [{ field: "createdAt", direction: "desc" }],
});
const next = await users.paginateCursor(undefined, {
  cursor: first.meta.nextCursor,
  limit: 25,
  sort: [{ field: "createdAt", direction: "desc" }],
});
// Page back: `previousCursor` is set on any non-empty page requested with a
// cursor. The rows come back in the requested sort order.
const back = await users.paginateCursor(undefined, {
  cursor: next.meta.previousCursor,
  limit: 25,
  sort: [{ field: "createdAt", direction: "desc" }],
});

A missing, forged, tampered or malformed cursor rejects with a ValidationError (400, exposable, issues[0].code such as "cursor_signature") before any query runs. The message never echoes the cursor or its contents. Building keyset pages by hand? Fetch with keysetFetchSort(sort, getKeysetDirection(payload)) and pass that direction to createKeysetPage.

paginateCursor appends the id field to sort as a tiebreaker, so a cursor positioned at a row you hold must include it: use users.createCursor(row, { sort }) (signed with cursorSecret), not createKeysetCursor(row, sort) with the bare sort, which paginateCursor rejects as missing the id field.

A Date sort value is encoded at millisecond precision. PostgreSQL timestamp/timestamptz columns keep microseconds unless declared (3), and Prisma returns Dates, so the cursor compares such a value as its millisecond bucket (>= C AND < C + 1ms for a tie) and lets the id tiebreaker order rows created in the same millisecond. That is exact for identical timestamps (rows inserted by one statement or transaction) and for ids that increase with insertion order (serial, ULID, UUIDv7) — with random ids, sort by a millisecond-precision column or add a monotonic tiebreaker.

Migrations and seeds

import { createMigrationRunner, createSeedRunner } from "@zudojs/database";

const migrations = createMigrationRunner(
  client,
  [
    {
      version: 1,
      name: "create-users",
      up: async (tx) => {
        await tx.$executeRawUnsafe(
          'CREATE TABLE "users" ("id" TEXT PRIMARY KEY, "email" TEXT UNIQUE NOT NULL)',
        );
      },
      down: async (tx) => {
        await tx.$executeRawUnsafe('DROP TABLE "users"');
      },
    },
  ],
  { transaction: { timeoutMs: 60_000 } },
);

await migrations.migrate(); // applies pending migrations under an advisory lock
await migrations.rollback(); // reverts the most recent one

const seeds = createSeedRunner(client, [
  {
    name: "roles",
    order: 0,
    run: async (tx) => {
      await tx.$executeRawUnsafe('INSERT INTO "roles" ("name") VALUES ($1)', "admin");
    },
  },
]);
await seeds.run();

Each migration or seed runs in its own transaction by default (perItemTransaction: true); pass transaction: { timeoutMs, maxWaitMs, isolationLevel } to size it for long-running steps.

A migration that cannot run inside a transaction (CREATE INDEX CONCURRENTLY, ALTER TYPE ... ADD VALUE) sets transaction: false. Its up/down then receive the root client and run outside any transaction; the applied check and the history record still happen in short transactions under the advisory lock, which is released while the body runs, so write such a migration idempotently (IF NOT EXISTS). It requires perItemTransaction: true; a batch runner refuses it at construction.

{
  version: 20260926120000,
  name: "orders_customer_idx",
  transaction: false,
  up: (db) => db.$executeRawUnsafe(
    'CREATE INDEX CONCURRENTLY IF NOT EXISTS "orders_customer_idx" ON "orders" ("customerId")',
  ),
  down: (db) => db.$executeRawUnsafe('DROP INDEX CONCURRENTLY IF EXISTS "orders_customer_idx"'),
}

createTransactionContext here builds the immutable status record that TransactionManager.run reports (transactionId, startedAt, status); it is unrelated to the createTransactionContext of @zudojs/transactions, which creates that package's AsyncLocalStorage propagation store.

Errors

Errors your own transaction callback throws that are already @zudojs/errors errors but not database errors (a NotFoundError, DomainError, ValidationError, ...) roll the transaction back and propagate unchanged from transaction(), withTransaction(), TransactionManager and units of work: same instance, same status, and not logged as a database error. Everything else, driver and database failures included, is normalised as below.

Every database failure surfaces as a DatabaseError from @zudojs/errors. Prisma codes are mapped to databaseCode, an ErrorCode and an HTTP status (P2002 / P2003 → 409, P2025 → 404, P2034 → retryable, P1xxx → 503 with a fixed message that never includes the host name).

import {
  isConflictError,
  isNotFoundError,
  isRetryableTransactionError,
  toDatabaseErrorInfo,
} from "@zudojs/database";

try {
  await users.create({ email: "[email protected]", name: "Alice" });
} catch (error) {
  if (isConflictError(error)) {
    // 409: unique or foreign-key violation
  } else if (isNotFoundError(error)) {
    // 404
  } else if (!isRetryableTransactionError(error)) {
    console.error(toDatabaseErrorInfo(error)); // { code: "P2002", message, operation, ... }
  }
}

Locks and health

import { assertDatabaseHealth, checkDatabaseHealth, createLockManager } from "@zudojs/database";

const locks = createLockManager(client);

// Advisory lock (FNV-1a 64 key, transaction-scoped). `timeoutMs` becomes
// `SET LOCAL lock_timeout` (at least 1 ms; PostgreSQL treats 0 as "no
// timeout", so 0 is rejected — use `noWait` to fail immediately); the
// transaction timeout is raised to cover it.
await locks.withAdvisoryLock("reports:nightly", async (tx) => {
  await tx.$executeRawUnsafe("REFRESH MATERIALIZED VIEW nightly_report");
}, { timeoutMs: 10_000 });

// Row lock: throws 404 when the row is missing, 409 when skipLocked skips it.
await locks.withRowLock("orders", orderId, async (tx) => {
  await tx.$executeRawUnsafe('UPDATE "orders" SET "status" = $1 WHERE "id" = $2', "paid", orderId);
}, { mode: "for-no-key-update", skipLocked: true });

// The connection manager reconnects with exponential back-off after failed
// health checks. The back-off wait keeps the process alive until the reconnect
// finishes; `disconnect()` / `destroy()` cancel it immediately.

const health = await checkDatabaseHealth(client, { timeoutMs: 2_000 });
health.status; // "healthy" | "degraded" | "unhealthy"
await assertDatabaseHealth(client); // throws DatabaseUnhealthyError with the real cause

Subpath exports

| Import | Contents | | --- | --- | | @zudojs/database | Everything | | @zudojs/database/client | DatabaseClient, createDatabaseClient, error helpers | | @zudojs/database/repositories | BaseRepository, mapRepositoryError | | @zudojs/database/transactions | TransactionManager, withTransaction, withTransactionRetry |

Features

• Prisma 7 client wrapper with connection lifecycle, in-flight connect de-duplication, raw queries, and typed error normalisation (Prisma P* codes mapped to conflict / not-found / timeout / connection outcomes)
• BaseRepository with CRUD, createMany / deleteMany, soft delete (softDelete, restore, findDeleted, withDeleted), transaction rebinding (withTransaction), and findByQuery
• Query builder and 40 filter helpers that translate to Prisma where / orderBy / select / include
• Managed transactions with context, error enrichment, and withTransactionRetry (serialization failures retried by default) with exponential backoff capped by maxRetryDelayMs (30 s by default, never above the 2^31-1 ms timer limit) and optional jitter: "full"
• Offset pagination and keyset pagination with HMAC-signed, shape-validated cursors
• Migration and seed runners using PostgreSQL advisory locks, per-item transactions, and BIGINT versioning
• Advisory and row locks with lock_timeout, SKIP LOCKED / NOWAIT, and namespaced keys
• Health and readiness probes with timeouts; connection manager with overlap-guarded scheduled checks and reconnect back-off
• Bounded in-memory LRU cache with TTL, periodic pruning, and loader coalescing
• Relation definitions with a registry and toPrismaInclude (depth and cycle guarded)

Limitations

• PostgreSQL only. Migration, seed, and lock helpers emit PostgreSQL SQL; other dialects throw UnsupportedDialectError.
• No savepoints or nested transactions. Prisma interactive transactions are used as-is.
• Connection pooling is handled by the driver adapter, not by this package.
• timeoutMs and signal on repository operations are client-side only; the database query is not cancelled server-side. A signal passed to transaction() / withTransaction() does roll the transaction back: the abort is raised inside the Prisma callback.
• The cache is not transaction-aware. Do not populate it from inside a transaction that may roll back.

Use Cases

• Data access layer for Zudojs services
• Transaction coordination across repositories (rebind with withTransaction)
• Database migrations and seeding with single-runner guarantees
• Health endpoints and readiness gating