Data architecture with ZudoJS

import { PGlite } from "@electric-sql/pglite";

const pg = new PGlite();
await pg.exec(`
  CREATE TABLE projects (id serial PRIMARY KEY, name text NOT NULL, owner_id text NOT NULL,
                         open_tasks int NOT NULL DEFAULT 0, internal_notes text);
  CREATE TABLE tasks (id serial PRIMARY KEY, project_id int NOT NULL, title text NOT NULL,
                      status text NOT NULL DEFAULT 'open');
`);

let queries = 0;
const query = <T>(sql: string, params: unknown[] = []) => {
  queries += 1;
  return pg.query<T>(sql, params);
};

async function createTask(projectId: number, title: string) {
  await query("INSERT INTO tasks (project_id, title) VALUES ($1, $2)", [projectId, title]);
  if (title.length > 20) throw new Error("title too long"); // a rule checked too late
  await query("UPDATE projects SET open_tasks = open_tasks + 1 WHERE id = $1", [projectId]);
}

async function dashboard() {
  const projects = (await query<Record<string, unknown>>("SELECT * FROM projects ORDER BY id")).rows;
  for (const project of projects) {
    const { rows } = await query<{ n: number }>(
      "SELECT count(*)::int AS n FROM tasks WHERE project_id = $1 AND status = 'open'", [project["id"]],
    );
    project["countedOpen"] = rows[0]?.n;
  }
  return projects;
}

for (let i = 1; i <= 50; i++) {
  await pg.query("INSERT INTO projects (name, owner_id, internal_notes) VALUES ($1, 'ada', 'margin 40%')", [`Project ${i}`]);
}
await createTask(1, "Draft the brief");
await createTask(1, "Collect quotes from three printers").catch(() => {});

queries = 0;
const rows = await dashboard();
console.log("queries for one page view:", queries);
console.log(rows[0]);

queries for one page view: 51
{
  id: 1,
  name: 'Project 1',
  owner_id: 'ada',
  open_tasks: 1,
  internal_notes: 'margin 40%',
  countedOpen: 2
}

import type { SoftDeletableEntity } from "@zudojs/database";

export interface Task extends SoftDeletableEntity {
  readonly id: number;
  readonly title: string;
}

entity-id.ts:3:18 - error TS2430: Interface 'Task' incorrectly extends interface 'SoftDeletableEntity'.
  Types of property 'id' are incompatible.
    Type 'number' is not assignable to type 'string'.

3 export interface Task extends SoftDeletableEntity {
                   ~~~~


Found 1 error in entity-id.ts:3

import type { Migration } from "@zudojs/database";

export const migrations: Migration[] = [
  {
    version: 1,
    name: "create-projects-and-tasks",
    up: async (tx) => {
      await tx.$executeRawUnsafe(`
        CREATE TABLE projects (
          id          serial PRIMARY KEY,
          name        text NOT NULL CHECK (length(name) BETWEEN 1 AND 100),
          owner_id    text NOT NULL,
          open_tasks  int  NOT NULL DEFAULT 0 CHECK (open_tasks >= 0),
          created_at  timestamptz(3) NOT NULL DEFAULT now(),
          updated_at  timestamptz(3) NOT NULL DEFAULT now()
        )`);
      await tx.$executeRawUnsafe(`
        CREATE TABLE tasks (
          id          serial PRIMARY KEY,
          project_id  int  NOT NULL REFERENCES projects (id),
          title       text NOT NULL CHECK (length(title) BETWEEN 1 AND 200),
          status      text NOT NULL DEFAULT 'open' CHECK (status IN ('open', 'done')),
          priority    text NOT NULL DEFAULT 'normal' CHECK (priority IN ('low', 'normal', 'high')),
          assignee_id text,
          version     int  NOT NULL DEFAULT 1,
          created_at  timestamptz(3) NOT NULL DEFAULT now(),
          updated_at  timestamptz(3) NOT NULL DEFAULT now(),
          deleted_at  timestamptz(3)
        )`);
    },
    down: async (tx) => {
      await tx.$executeRawUnsafe("DROP TABLE tasks");
      await tx.$executeRawUnsafe("DROP TABLE projects");
    },
  },
  {
    version: 2,
    name: "index-tasks-for-listing",
    up: async (tx) => {
      await tx.$executeRawUnsafe(
        "CREATE INDEX tasks_project_created_idx ON tasks (project_id, created_at DESC, id DESC)",
      );
      await tx.$executeRawUnsafe(
        "CREATE UNIQUE INDEX tasks_project_title_key ON tasks (project_id, lower(title)) WHERE deleted_at IS NULL",
      );
    },
    down: async (tx) => {
      await tx.$executeRawUnsafe("DROP INDEX tasks_project_title_key");
      await tx.$executeRawUnsafe("DROP INDEX tasks_project_created_idx");
    },
  },
];

import { PGlite } from "@electric-sql/pglite";
import { createDatabaseClient, noopDatabaseLogger } from "@zudojs/database";
import type { DatabaseEntity } from "@zudojs/database";
import { createPglitePrisma } from "./pglite-adapter.js";
import type { Model } from "./pglite-adapter.js";

export type TaskStatus = "open" | "done";
export type Priority = "low" | "normal" | "high";

export interface Project extends DatabaseEntity<number> {
  readonly name: string;
  readonly ownerId: string;
  readonly openTasks: number;
}

export interface Task extends DatabaseEntity<number> {
  readonly projectId: number;
  readonly title: string;
  readonly status: TaskStatus;
  readonly priority: Priority;
  readonly assigneeId: string | null;
  readonly version: number;
  readonly deletedAt: Date | null;
}

export const PROJECT_MODEL: Model = {
  table: "projects",
  fields: ["id", "name", "ownerId", "openTasks", "createdAt", "updatedAt"],
};
export const TASK_MODEL: Model = {
  table: "tasks",
  fields: ["id", "projectId", "title", "status", "priority", "assigneeId", "version", "createdAt", "updatedAt", "deletedAt"],
};

export async function openDatabase() {
  const pg = new PGlite();
  const prisma = createPglitePrisma(pg, { project: PROJECT_MODEL, task: TASK_MODEL });
  const client = createDatabaseClient({ prisma, logger: noopDatabaseLogger });
  await client.connect();
  return { pg, client };
}

import { createMigrationRunner } from "@zudojs/database";
import { openDatabase } from "./database.js";
import { migrations } from "./migrations.js";

export async function setup() {
  const db = await openDatabase();
  await createMigrationRunner(db.client, migrations).migrate();
  await db.pg.exec(`
    INSERT INTO projects (name, owner_id, open_tasks) VALUES ('Website relaunch', 'ada', 3), ('Office move', 'bola', 1);
    INSERT INTO tasks (project_id, title, priority, assignee_id, status, created_at) VALUES
      (1, 'Write the copy',      'high',   'ada',   'open', '2026-09-01 09:00+00'),
      (1, 'Choose a font',       'normal', 'chidi', 'done', '2026-09-01 10:00+00'),
      (1, 'Compress the images', 'low',    NULL,    'open', '2026-09-02 09:00+00'),
      (1, 'Test on phones',      'high',   'ada',   'open', '2026-09-03 09:00+00'),
      (2, 'Book the van',        'high',   'bola',  'open', '2026-09-02 12:00+00');
  `);
  return db;
}

import { BaseRepository } from "@zudojs/database";
import type { BaseRepositoryOptions, RepositoryDelegate } from "@zudojs/database";
import type { Priority, Project, Task, TaskStatus } from "./database.js";

type Step = { readonly increment: number } | { readonly decrement: number };

export type NewProject = Pick<Project, "name" | "ownerId">;
export interface ProjectChanges {
  readonly name?: string;
  readonly openTasks?: Step;
}

export class ProjectRepository extends BaseRepository<Project, number, NewProject, ProjectChanges> {
  constructor(delegate: RepositoryDelegate<Project, number, NewProject, ProjectChanges>, options: BaseRepositoryOptions = {}) {
    super(delegate, { modelName: "Project", ...options });
  }
}

export type NewTask = Pick<Task, "projectId" | "title"> & Partial<Pick<Task, "priority" | "assigneeId">>;
export interface TaskChanges {
  readonly title?: string;
  readonly status?: TaskStatus;
  readonly priority?: Priority;
  readonly assigneeId?: string | null;
  readonly version?: Step;
}

export class TaskRepository extends BaseRepository<Task, number, NewTask, TaskChanges> {
  constructor(delegate: RepositoryDelegate<Task, number, NewTask, TaskChanges>, options: BaseRepositoryOptions = {}) {
    super(delegate, { modelName: "Task", softDelete: true, ...options });
  }

  findOpenInProject(projectId: number): Promise<readonly Task[]> {
    return this.findMany({ projectId, status: "open" });
  }

  /** Changes a task only if nobody changed it since `version` was read. */
  updateIfUnchanged(id: number, version: number, changes: TaskChanges): Promise<Task> {
    return this.execute("update", () =>
      this.delegate.update({
        where: { id, version, deletedAt: null },
        data: { ...changes, version: { increment: 1 } },
      }),
    );
  }
}

import { BaseError } from "@zudojs/errors";
import { PROJECT_MODEL, TASK_MODEL } from "./database.js";
import type { Project, Task } from "./database.js";
import { modelDelegate } from "./pglite-adapter.js";
import { ProjectRepository, TaskRepository } from "./repositories.js";
import { setup } from "./setup.js";

const { pg, client } = await setup();
const projects = new ProjectRepository(modelDelegate<Project>(pg, PROJECT_MODEL));
const tasks = new TaskRepository(modelDelegate<Task>(pg, TASK_MODEL));

const open = await tasks.findOpenInProject(1);
console.log("open in project 1:", open.map((t) => t.id).sort());

const attempts: [string, () => Promise<unknown>][] = [
  ["same title, other case", () => tasks.create({ projectId: 1, title: "write THE copy" })],
  ["project that does not exist", () => tasks.create({ projectId: 9, title: "Hire movers" })],
  ["counter below zero", () => projects.update(2, { openTasks: { decrement: 5 } })],
];
for (const [label, attempt] of attempts) {
  try {
    await attempt();
  } catch (error) {
    if (error instanceof BaseError) console.log(`${label}: ${error.statusCode} ${error.message}`);
  }
}
await client.disconnect();

open in project 1: [ 1, 3, 4 ]
same title, other case: 409 Task already exists.
project that does not exist: 409 Task create violates a foreign key constraint.
counter below zero: 500 Project update failed.

import { schema } from "@zudojs/schema";
import type { Infer } from "@zudojs/schema";
import type { Priority, Task, TaskStatus } from "./database.js";

const PriorityValue = schema.enum(["low", "normal", "high"] as const);

export const CreateTaskBody = schema.object({
  title: schema.string().trim().min(1).max(200),
  priority: schema.optional(PriorityValue),
  assigneeId: schema.optional(schema.string().min(1).max(64)),
}).strict();
export type CreateTaskBody = Infer<typeof CreateTaskBody>;

export const ListTasksQuery = schema.object({
  status: schema.optional(schema.enum(["open", "done"] as const)),
  limit: schema.coerce.number().int().min(1).max(50).default(20),
  cursor: schema.optional(schema.string().max(512)),
}).strict();
export type ListTasksQuery = Infer<typeof ListTasksQuery>;

export interface TaskResponse {
  readonly id: number;
  readonly title: string;
  readonly status: TaskStatus;
  readonly priority: Priority;
  readonly assigneeId: string | null;
  readonly version: number;
  readonly createdAt: string;
}

export function toTaskResponse(task: Task): TaskResponse {
  return {
    id: task.id,
    title: task.title,
    status: task.status,
    priority: task.priority,
    assigneeId: task.assigneeId,
    version: task.version,
    createdAt: task.createdAt.toISOString(),
  };
}

import { SchemaError } from "@zudojs/errors";
import { TASK_MODEL } from "./database.js";
import type { Task } from "./database.js";
import { modelDelegate } from "./pglite-adapter.js";
import { TaskRepository } from "./repositories.js";
import { setup } from "./setup.js";
import { CreateTaskBody, ListTasksQuery, toTaskResponse } from "./task.dto.js";

const bodies: unknown[] = [
  { title: "  Order business cards ", priority: "high" },
  { title: "Order stamps", projectId: 2, status: "done", version: 99 },
  { title: "" },
];
for (const body of bodies) {
  const result = CreateTaskBody.safeParse(body);
  console.log(result.success ? result.data : result.issues.map((i) => i.message));
}

for (const query of [{ limit: "10" }, { limit: "5000" }, {}]) {
  try {
    console.log(ListTasksQuery.parse(query));
  } catch (error) {
    if (error instanceof SchemaError) console.log(error.statusCode, error.message);
  }
}

const { pg, client } = await setup();
const tasks = new TaskRepository(modelDelegate<Task>(pg, TASK_MODEL));
console.log(toTaskResponse((await tasks.findById(1))!));
await client.disconnect();

{ title: 'Order business cards', priority: 'high' }
[
  'Unknown key: projectId',
  'Unknown key: status',
  'Unknown key: version'
]
[ 'String must be at least 1 character' ]
{ limit: 10 }
400 Validation failed
{ limit: 20 }
{
  id: 1,
  title: 'Write the copy',
  status: 'open',
  priority: 'high',
  assigneeId: 'ada',
  version: 1,
  createdAt: '2026-09-01T09:00:00.000Z'
}

import { normalizeLimit } from "@zudojs/database";
import type { DatabaseClient } from "@zudojs/database";

export interface ProjectCard {
  readonly id: number;
  readonly name: string;
  readonly openTasks: number;
  readonly highPriority: number;
}

interface CardRow {
  readonly id: number;
  readonly name: string;
  readonly open: number;
  readonly high: number;
}

export class ProjectQueries {
  constructor(private readonly client: DatabaseClient) {}

  async cardsFor(ownerId: string, limit: number): Promise<readonly ProjectCard[]> {
    const rows = await this.client.queryRawUnsafe<CardRow[]>(
      `SELECT p.id, p.name,
              count(t.id) FILTER (WHERE t.status = 'open')::int AS open,
              count(t.id) FILTER (WHERE t.status = 'open' AND t.priority = 'high')::int AS high
         FROM projects p
         LEFT JOIN tasks t ON t.project_id = p.id AND t.deleted_at IS NULL
        WHERE p.owner_id = $1
        GROUP BY p.id
        ORDER BY p.id
        LIMIT $2`,
      [ownerId, normalizeLimit(limit)],
    );
    return rows.map((row) => ({ id: row.id, name: row.name, openTasks: row.open, highPriority: row.high }));
  }
}

import type { PGlite } from "@electric-sql/pglite";
import { PROJECT_MODEL, TASK_MODEL } from "./database.js";
import type { Project, Task } from "./database.js";
import { modelDelegate } from "./pglite-adapter.js";
import { ProjectQueries } from "./project.queries.js";
import { ProjectRepository, TaskRepository } from "./repositories.js";
import { setup } from "./setup.js";

function countQueries(pg: PGlite) {
  const original = pg.query.bind(pg);
  const counter = { n: 0 };
  pg.query = ((...args: Parameters<PGlite["query"]>) => {
    counter.n += 1;
    return original(...args);
  }) as PGlite["query"];
  return counter;
}

const { pg, client } = await setup();
for (let i = 1; i <= 20; i++) {
  await pg.query("INSERT INTO projects (name, owner_id) VALUES ($1, 'ada')", [`Side project ${i}`]);
}
const projects = new ProjectRepository(modelDelegate<Project>(pg, PROJECT_MODEL));
const tasks = new TaskRepository(modelDelegate<Task>(pg, TASK_MODEL));
const counter = countQueries(pg);

counter.n = 0;
const mine = await projects.findMany({ ownerId: "ada" });
for (const project of mine) await tasks.count({ projectId: project.id, status: "open" });
console.log("loop:", counter.n, "queries");

counter.n = 0;
const ids = (await projects.findMany({ ownerId: "ada" })).map((p) => p.id);
const openTasks = await tasks.findMany({ projectId: { in: ids }, status: "open" });
console.log("batched:", counter.n, "queries,", openTasks.length, "open tasks loaded");

counter.n = 0;
const cards = await new ProjectQueries(client).cardsFor("ada", 3);
console.log("query service:", counter.n, "query");
console.log(cards);
await client.disconnect();

loop: 22 queries
batched: 2 queries, 3 open tasks loaded
query service: 1 query
[
  { id: 1, name: 'Website relaunch', openTasks: 3, highPriority: 2 },
  { id: 3, name: 'Side project 1', openTasks: 0, highPriority: 0 },
  { id: 4, name: 'Side project 2', openTasks: 0, highPriority: 0 }
]

import { createRelationRegistry, includeRelation, manyToOne, oneToMany, toPrismaInclude } from "@zudojs/database";

const relations = createRelationRegistry([
  oneToMany({ name: "tasks", parent: "Project", child: "Task", foreignKey: "projectId", referencedKey: "id" }),
  manyToOne({ name: "project", parent: "Task", child: "Project", foreignKey: "projectId", referencedKey: "id" }),
]);
const options = { registry: relations, parent: "Project", includeDeleted: false };

console.log(JSON.stringify(toPrismaInclude([includeRelation("tasks", { select: ["id", "title"] })], options)));

const requests = [
  [includeRelation("owner")],
  [includeRelation("tasks", { include: [includeRelation("project", { include: [includeRelation("tasks")] })] })],
];
for (const include of requests) {
  try {
    toPrismaInclude(include, options);
  } catch (error) {
    console.log((error as Error).name, "-", (error as Error).message);
  }
}

{"tasks":{"select":{"id":true,"title":true},"where":{"deletedAt":null}}}
TypeError - Relation "owner" is not registered for the parent model.
TypeError - Relation include "tasks" is cyclic.

import { TASK_MODEL } from "./database.js";
import type { Task } from "./database.js";
import { modelDelegate } from "./pglite-adapter.js";
import { TaskRepository } from "./repositories.js";
import { setup } from "./setup.js";

const { pg, client } = await setup();
const tasks = new TaskRepository(modelDelegate<Task>(pg, TASK_MODEL));
const sort = [{ field: "createdAt", direction: "desc" }, { field: "id", direction: "desc" }] as const;
const titles = (list: readonly Task[]) => list.map((t) => t.title);

const page1 = await tasks.findPaginated({ projectId: 1 }, { pagination: { page: 1, limit: 2 }, sort });
console.log("offset page 1:", titles(page1.data));
await tasks.create({ projectId: 1, title: "Fix the footer" });
const page2 = await tasks.findPaginated({ projectId: 1 }, { pagination: { page: 2, limit: 2 }, sort });
console.log("offset page 2:", titles(page2.data));

const first = await tasks.paginateCursor({ projectId: 1 }, { limit: 2, sort: [{ field: "createdAt", direction: "desc" }] });
console.log("keyset page 1:", titles(first.data));
await tasks.create({ projectId: 1, title: "Update the sitemap" });
const second = await tasks.paginateCursor({ projectId: 1 }, {
  limit: 2, sort: [{ field: "createdAt", direction: "desc" }], cursor: first.meta.nextCursor,
});
console.log("keyset page 2:", titles(second.data));
await client.disconnect();

offset page 1: [ 'Test on phones', 'Compress the images' ]
offset page 2: [ 'Compress the images', 'Choose a font' ]
keyset page 1: [ 'Fix the footer', 'Test on phones' ]
keyset page 2: [ 'Compress the images', 'Choose a font' ]

import { buildKeysetWhere, createKeysetCursor, decodeKeysetCursor } from "@zudojs/database";

const sort = [{ field: "createdAt", direction: "desc" }, { field: "id", direction: "desc" }] as const;
const lastRow = { id: 3, createdAt: new Date("2026-09-02T09:00:00Z"), title: "Compress the images" };

const cursor = createKeysetCursor(lastRow, sort);
const payload = decodeKeysetCursor(cursor, sort);
console.log(payload);
console.log(JSON.stringify(buildKeysetWhere(payload, sort), null, 1));

{ createdAt: '2026-09-02T09:00:00.000Z', id: 3 }
{
 "OR": [
  {
   "createdAt": {
    "lt": "2026-09-02T09:00:00.000Z"
   }
  },
  {
   "AND": [
    {
     "createdAt": {
      "equals": "2026-09-02T09:00:00.000Z"
     }
    },
    {
     "id": {
      "lt": 3
     }
    }
   ]
  }
 ]
}

import { PGlite } from "@electric-sql/pglite";
import { BaseRepository } from "@zudojs/database";
import type { DatabaseEntity } from "@zudojs/database";
import { modelDelegate } from "./pglite-adapter.js";

interface Reading extends DatabaseEntity<number> {
  readonly label: string;
}
class ReadingRepository extends BaseRepository<Reading, number> {}

const pg = new PGlite();
await pg.exec(`
  CREATE TABLE readings (id serial PRIMARY KEY, label text NOT NULL,
                         created_at timestamptz NOT NULL, updated_at timestamptz NOT NULL DEFAULT now());
  INSERT INTO readings (label, created_at) VALUES
    ('first',  '2026-09-01 10:00:00.000100+00'),
    ('second', '2026-09-01 10:00:00.000200+00'),
    ('third',  '2026-09-01 10:00:00.000300+00'),
    ('older',  '2026-09-01 09:00:00+00');
`);
const readings = new ReadingRepository(
  modelDelegate<Reading>(pg, { table: "readings", fields: ["id", "label", "createdAt", "updatedAt"] }),
  { modelName: "Reading" },
);

const seen: string[] = [];
let cursor: string | undefined;
do {
  const page = await readings.paginateCursor(undefined, { limit: 1, sort: [{ field: "createdAt", direction: "desc" }], cursor });
  seen.push(...page.data.map((r) => r.label));
  cursor = page.meta.nextCursor;
} while (cursor);
console.log("paged through:", seen);
console.log("in the table:", (await pg.query("SELECT count(*)::int AS n FROM readings")).rows[0]);
await pg.close();

paged through: [ 'third', 'older' ]
in the table: { n: 4 }

import { createMigrationRunner } from "@zudojs/database";
import { migrations } from "./migrations.js";
import { setup } from "./setup.js";

const { pg, client } = await setup();
await pg.exec(`
  INSERT INTO projects (name, owner_id) SELECT 'Project ' || n, 'ada' FROM generate_series(3, 300) AS n;
  INSERT INTO tasks (project_id, title, created_at)
    SELECT 1 + n % 300, 'Task ' || n, timestamptz '2026-01-01 00:00+00' + n * interval '1 minute'
    FROM generate_series(1, 30000) AS n;
  ANALYZE;
`);

const listSql = `SELECT * FROM tasks
  WHERE project_id = 42 AND deleted_at IS NULL
    AND (created_at < '2026-01-15 00:00+00' OR (created_at = '2026-01-15 00:00+00' AND id < 20000))
  ORDER BY created_at DESC, id DESC LIMIT 21`;

async function plan(sql: string) {
  const { rows } = await pg.query<{ "QUERY PLAN": string }>("EXPLAIN " + sql);
  return rows.map((r) => r["QUERY PLAN"].replace(/\s+\(cost=.*$/, "")).filter((line) => !line.includes("Filter") && !line.includes("Cond"));
}

console.log("with the index:", await plan(listSql));
const runner = createMigrationRunner(client, migrations);
await runner.rollback();
console.log("without it:", await plan(listSql));
await runner.migrate();
await client.disconnect();

with the index: [
  'Limit',
  '  ->  Index Scan using tasks_project_created_idx on tasks'
]
without it: [
  'Limit',
  '  ->  Sort',
  '        Sort Key: created_at DESC, id DESC',
  '        ->  Seq Scan on tasks'
]

import { setup } from "./setup.js";

const { pg, client } = await setup();
await pg.exec(`
  INSERT INTO tasks (project_id, title, created_at)
    SELECT 2, 'Box ' || n, timestamptz '2026-01-01 00:00+00' + n * interval '1 minute'
    FROM generate_series(1, 20000) AS n;
  ANALYZE;
`);

async function rowsRead(sql: string): Promise<number> {
  const { rows } = await pg.query<{ "QUERY PLAN": [{ Plan: { Plans: [{ "Actual Rows": number }] } }] }>(
    "EXPLAIN (ANALYZE, FORMAT JSON) " + sql,
  );
  return rows[0]!["QUERY PLAN"][0].Plan.Plans[0]["Actual Rows"];
}

const base = "SELECT * FROM tasks WHERE project_id = 2 AND deleted_at IS NULL ORDER BY created_at DESC, id DESC";
console.log("offset 10,000:", await rowsRead(`${base} LIMIT 20 OFFSET 10000`), "rows read");
const cursorRow = (await pg.query<{ created_at: Date; id: number }>(`${base} LIMIT 1 OFFSET 9999`)).rows[0]!;
const keyset = `SELECT * FROM tasks WHERE project_id = 2 AND deleted_at IS NULL
  AND (created_at, id) < ($1::timestamptz, $2::int) ORDER BY created_at DESC, id DESC LIMIT 20`;
const { rows } = await pg.query<{ "QUERY PLAN": [{ Plan: { Plans: [{ "Actual Rows": number }] } }] }>(
  "EXPLAIN (ANALYZE, FORMAT JSON) " + keyset, [cursorRow.created_at, cursorRow.id],
);
console.log("keyset after row 10,000:", rows[0]!["QUERY PLAN"][0].Plan.Plans[0]["Actual Rows"], "rows read");
await client.disconnect();

offset 10,000: 10020 rows read
keyset after row 10,000: 20 rows read

import { BaseError } from "@zudojs/errors";
import { TASK_MODEL } from "./database.js";
import type { Task } from "./database.js";
import { modelDelegate } from "./pglite-adapter.js";
import { TaskRepository } from "./repositories.js";
import { setup } from "./setup.js";

const { pg, client } = await setup();
const tasks = new TaskRepository(modelDelegate<Task>(pg, TASK_MODEL));

await tasks.softDelete(2);
const again = await tasks.create({ projectId: 1, title: "Choose a font" });
console.log("re-created as task", again.id);

try {
  await tasks.restore(2);
} catch (error) {
  if (error instanceof BaseError) console.log("restore:", error.statusCode, error.message);
}
console.log("titles in project 1:", (await tasks.withDeleted().findMany({ projectId: 1 })).length, "rows");
await client.disconnect();

re-created as task 6
restore: 409 Task already exists.
titles in project 1: 5 rows

import { isNotFoundError, withTransaction } from "@zudojs/database";
import type { DatabaseClient } from "@zudojs/database";
import { ConflictError, NotFoundError } from "@zudojs/errors";
import type { Task } from "./database.js";
import type { ProjectRepository, TaskRepository } from "./repositories.js";
import type { CreateTaskBody } from "./task.dto.js";

export class TaskService {
  constructor(
    private readonly client: DatabaseClient,
    private readonly projects: ProjectRepository,
    private readonly tasks: TaskRepository,
  ) {}

  createTask(projectId: number, body: CreateTaskBody): Promise<Task> {
    return withTransaction(this.client, async (tx) => {
      const projects = this.projects.withTransaction(tx);
      if (!(await projects.exists({ id: projectId }))) {
        throw new NotFoundError(`Project ${projectId} not found`);
      }
      const task = await this.tasks.withTransaction(tx).create({ projectId, ...body });
      await projects.update(projectId, { openTasks: { increment: 1 } });
      return task;
    });
  }

  completeTask(id: number, expectedVersion: number): Promise<Task> {
    return withTransaction(this.client, async (tx) => {
      const tasks = this.tasks.withTransaction(tx);
      const current = await tasks.findById(id);
      if (!current) throw new NotFoundError(`Task ${id} not found`);
      if (current.version !== expectedVersion) {
        throw new ConflictError(`Task ${id} was changed by someone else. Reload it and try again.`);
      }
      if (current.status === "done") return current; // a repeated request changes nothing
      let done: Task;
      try {
        done = await tasks.updateIfUnchanged(id, expectedVersion, { status: "done" });
      } catch (error) {
        if (isNotFoundError(error)) throw new ConflictError(`Task ${id} was changed by someone else. Reload it and try again.`);
        throw error;
      }
      await this.projects.withTransaction(tx).update(done.projectId, { openTasks: { decrement: 1 } });
      return done;
    });
  }
}

import { BaseError } from "@zudojs/errors";
import { PROJECT_MODEL, TASK_MODEL } from "./database.js";
import type { Project, Task } from "./database.js";
import { modelDelegate } from "./pglite-adapter.js";
import { ProjectRepository, TaskRepository } from "./repositories.js";
import { setup } from "./setup.js";
import { TaskService } from "./task.service.js";

const { pg, client } = await setup();
const projects = new ProjectRepository(modelDelegate<Project>(pg, PROJECT_MODEL));
const tasks = new TaskRepository(modelDelegate<Task>(pg, TASK_MODEL));
const service = new TaskService(client, projects, tasks);
const openCount = async () => (await projects.findById(1))?.openTasks;

await service.createTask(1, { title: "Add a contact form" });
console.log("after a good create:", await openCount());

for (const [projectId, title] of [[1, "Write the copy"], [7, "Order chairs"]] as const) {
  try {
    await service.createTask(projectId, { title });
  } catch (error) {
    if (error instanceof BaseError) console.log(`"${title}": ${error.statusCode} ${error.message}`);
  }
}
console.log("after two failures:", await openCount(), "with", await tasks.count({ projectId: 1, status: "open" }), "open tasks");
await client.disconnect();

after a good create: 4
"Write the copy": 409 Task already exists.
"Order chairs": 404 Project 7 not found
after two failures: 4 with 4 open tasks

import { PROJECT_MODEL } from "./database.js";
import type { Project } from "./database.js";
import { modelDelegate } from "./pglite-adapter.js";
import { ProjectRepository } from "./repositories.js";
import { setup } from "./setup.js";

const { pg, client } = await setup();
const projects = new ProjectRepository(modelDelegate<Project>(pg, PROJECT_MODEL));

async function readModifyWrite(projectId: number) {
  const project = await projects.findById(projectId);
  await pg.query("UPDATE projects SET open_tasks = $1 WHERE id = $2", [project!.openTasks - 1, projectId]);
}
await Promise.all([readModifyWrite(1), readModifyWrite(1)]);
console.log("read-modify-write, 3 minus 2 =", (await projects.findById(1))?.openTasks);

await pg.query("UPDATE projects SET open_tasks = 3 WHERE id = 1");
await Promise.all([
  projects.update(1, { openTasks: { decrement: 1 } }),
  projects.update(1, { openTasks: { decrement: 1 } }),
]);
console.log("atomic decrement, 3 minus 2 =", (await projects.findById(1))?.openTasks);
await client.disconnect();

read-modify-write, 3 minus 2 = 2
atomic decrement, 3 minus 2 = 1

import { BaseError } from "@zudojs/errors";
import { PROJECT_MODEL, TASK_MODEL } from "./database.js";
import type { Project, Task } from "./database.js";
import { modelDelegate } from "./pglite-adapter.js";
import { ProjectRepository, TaskRepository } from "./repositories.js";
import { setup } from "./setup.js";
import { toTaskResponse } from "./task.dto.js";
import { TaskService } from "./task.service.js";

const { pg, client } = await setup();
const projects = new ProjectRepository(modelDelegate<Project>(pg, PROJECT_MODEL));
const tasks = new TaskRepository(modelDelegate<Task>(pg, TASK_MODEL));
const service = new TaskService(client, projects, tasks);

const seenByAda = toTaskResponse((await tasks.findById(3))!);
await tasks.update(3, { priority: "high", version: { increment: 1 } }); // Chidi saves an edit first
console.log("Chidi's edit moved the task to version", (await tasks.findById(3))?.version);

try {
  await service.completeTask(3, seenByAda.version);
} catch (error) {
  if (error instanceof BaseError) console.log("Ada:", error.statusCode, error.message);
}

const fresh = (await tasks.findById(3))!;
const done = await service.completeTask(3, fresh.version);
const repeated = await service.completeTask(3, done.version);
console.log("Ada after reloading:", done.status, "version", done.version, "- repeated:", repeated.version);
console.log("open_tasks:", (await projects.findById(1))?.openTasks);
await client.disconnect();

Chidi's edit moved the task to version 2
Ada: 409 Task 3 was changed by someone else. Reload it and try again.
Ada after reloading: done version 3 - repeated: 3
open_tasks: 2

import { strict as assert } from "node:assert";
import { PROJECT_MODEL, TASK_MODEL } from "./database.js";
import type { Project, Task } from "./database.js";
import { modelDelegate } from "./pglite-adapter.js";
import { ProjectRepository, TaskRepository } from "./repositories.js";
import { setup } from "./setup.js";
import { TaskService } from "./task.service.js";

async function drift(pg: Awaited<ReturnType<typeof setup>>["pg"]) {
  const { rows } = await pg.query<{ id: number }>(`
    SELECT p.id FROM projects p
    WHERE p.open_tasks <> (SELECT count(*) FROM tasks t
                           WHERE t.project_id = p.id AND t.status = 'open' AND t.deleted_at IS NULL)`);
  return rows.map((r) => r.id);
}

async function test(name: string, body: (service: TaskService, tasks: TaskRepository) => Promise<void>) {
  const { pg, client } = await setup();
  const tasks = new TaskRepository(modelDelegate<Task>(pg, TASK_MODEL));
  const service = new TaskService(client, new ProjectRepository(modelDelegate<Project>(pg, PROJECT_MODEL)), tasks);
  try {
    await body(service, tasks);
    assert.deepEqual(await drift(pg), [], "open_tasks drifted");
    console.log("PASS", name);
  } catch (error) {
    console.log("FAIL", name, "-", (error as Error).message.split("\n")[0]);
  } finally {
    await client.disconnect();
  }
}

await test("seed data is consistent", async () => {});
await test("create and complete keep the counter", async (service) => {
  const task = await service.createTask(2, { title: "Label the boxes" });
  await service.completeTask(task.id, task.version);
});
await test("a failed create changes nothing", async (service) => {
  await assert.rejects(service.createTask(1, { title: "WRITE THE COPY" }));
});
await test("completing twice counts once", async (service) => {
  const done = await service.completeTask(5, 1);
  await service.completeTask(5, done.version);
});
await test("soft delete keeps the counter", async (_service, tasks) => {
  await tasks.softDelete(1);
});

PASS seed data is consistent
PASS create and complete keep the counter
PASS a failed create changes nothing
PASS completing twice counts once
FAIL soft delete keeps the counter - open_tasks drifted

import { setup } from "./setup.js";

const { pg, client } = await setup();
try {
  await client.transaction(async (tx) => {
    await tx.$executeRawUnsafe("CREATE INDEX CONCURRENTLY tasks_assignee_idx ON tasks (assignee_id)");
  });
} catch (error) {
  console.log("inside a transaction:", (error as Error).message);
}
await pg.exec("CREATE INDEX CONCURRENTLY tasks_assignee_idx ON tasks (assignee_id)");
console.log("outside:", (await pg.query("SELECT indexname FROM pg_indexes WHERE indexname = 'tasks_assignee_idx'")).rows);
await client.disconnect();

inside a transaction: CREATE INDEX CONCURRENTLY cannot run inside a transaction block
outside: [ { indexname: 'tasks_assignee_idx' } ]