Observability | OKModel

Observers are read-only. They cannot change a query, a commit, or a rollback. Event shapes are Observer events.

onWrite

onWrite observes committed writes. Pass writeObserver({ keys: ["notes"] }) from okmodel/observe when a table's primary key should appear on the event. A table with no primary key is refused at connect. Row counts use the driver command tag: changed, returned when RETURNING ran, or unknown when the driver reported none. An upsert counts inserted and updated rows as changed. The handler cannot change the write. await onWrite.flush() waits for handlers. A unit is one transaction, batch, or autocommit statement, and two units that run at once on one client keep separate events. A unit that rolls back ends as rolled_back. A connection lost after COMMIT was sent ends as unknown: the commit may or may not have landed, so the handler must be idempotent. A process that dies delivers no event.

on-write.ts

import { connect } from "okmodel/pg/postgresjs";
import { schema, table, t } from "okmodel/pg";
import { writeObserver } from "okmodel/observe";

const notes = table("notes", {
  id: t.identity(),
  title: t.text(),
});
const app = schema({ tables: [notes] });

const seen: unknown[] = [];
const onWrite = writeObserver({
  keys: ["notes"],
  committed(unit) {
    seen.push({
      outcome: "committed",
      writes: unit.writes.map((write) => ({
        op: write.op,
        table: write.table,
        rowCount: write.rowCount,
        rowCountMode: write.rowCountMode,
        keys: write.keys,
      })),
    });
  },
});

export async function demo(url: string) {
  await using db = connect(url, {
    schema: app,
    onWrite,
  });
  await db.notes.insert({ title: "Hello" });
  await onWrite.flush();
  console.log(seen);
}

on-write.out

[
  {
    outcome: "committed",
    writes: [
      {
        op: "insert",
        table: "notes",
        rowCount: 1,
        rowCountMode: "returned",
        keys: [{ id: "1" }],
      },
    ],
  },
]

onQuery

onQuery observes each call after it settles, reads and writes. One QueryEvent per call. The set_config prelude, BEGIN, COMMIT, and RESET ALL are part of the call and are not a separate event. Pass queryObserver({ query }) from okmodel/observe. params: true includes bound parameters, with sensitive, hidden, and tenant values as [redacted]. stack: true includes a stack. toOtel(event) maps one event onto the same db.* names as okmodel/otel, and adds the SQLSTATE on failure. route names the replica that served a read, or the primary. The handler cannot change the result. await onQuery.flush() waits for handlers.

on-query.ts

import { connect } from "okmodel/pg/postgresjs";
import { schema, table, t } from "okmodel/pg";
import { queryObserver } from "okmodel/observe";

const notes = table("notes", {
  id: t.identity(),
  title: t.text(),
});
const app = schema({ tables: [notes] });

const seen: unknown[] = [];
const onQuery = queryObserver({
  query(event) {
    seen.push({
      op: event.op,
      table: event.table,
      summary: event.summary,
      rowCount: event.rowCount,
      rowCountMode: event.rowCountMode,
    });
  },
});

export async function demo(url: string) {
  await using db = connect(url, {
    schema: app,
    onQuery,
  });
  const row = await db.notes.insert({ title: "Hello" });
  await db.notes.find({ where: { id: row.id }, limit: 1 });
  await onQuery.flush();
  console.log(seen);
}

on-query.out

[
  {
    op: "insert",
    table: "notes",
    summary: "INSERT notes",
    rowCount: 1,
    rowCountMode: "returned",
  },
  {
    op: "find",
    table: "notes",
    summary: "SELECT notes",
    rowCount: 1,
    rowCountMode: "returned",
  },
]

hookm

hookm is a list of read-only observers. Omitted, no observer runs. Pass an array: hookm: [observer]. A single observer object is not a list — wrap it. There is no hooks connect option. A typo such as hooks: […] is ignored at runtime: TypeScript rejects it on an object literal, and a value passed through a wider type never runs. Each method is optional. They run in list order. They are events only. An error a hook throws is dropped and never reaches the call. The next observer still runs.

Method	When
`onNotice`	A server notice, such as `RAISE NOTICE`.
`onTransaction`	A transaction or batch starts, commits, or rolls back.
`onError`	An error the runtime could not give to a caller, such as one from an `afterCommit` callback.

onNotice

A statement's notices are passed to every onNotice.

Field	Value
`severity`	A string.
`message`	A string.
`code`	Optional. A string.

Bun.sql does not surface RAISE NOTICE; see Drivers. Use postgres.js, node-postgres, or PGlite when you need notices.

onTransaction

onTransaction hears that a transaction or savepoint changed phase. phase is start, commit, or rollback. depth is 0 for the transaction and 1 and up for savepoints.

on-transaction.ts

import { connect } from "okmodel/pg/postgresjs";
import { schema, table, t } from "okmodel/pg";

const notes = table("notes", {
  id: t.identity(),
  title: t.text(),
});

const app = schema({ tables: [notes] });

const seen: string[] = [];

export async function demo(url: string) {
  await using db = connect(url, {
    schema: app,
    hookm: [
      {
        onTransaction: (event) => {
          seen.push(`${event.phase}:${event.depth}`);
        },
      },
    ],
  });
  await db.tx(async (tx) => {
    await tx.notes.insert({ title: "Hello" });
  });
  console.log(seen);
}

on-transaction.out

["start:0", "commit:0"]

onError

error is unknown. An afterCommit callback is one case (Transactions). That error is reported to onError and never changes the commit result.

OpenTelemetry

okmodel/otel is experimental. It turns these observers and onRoute into OpenTelemetry spans. You pass the tracer. The package does not import @opentelemetry/*.

otel({ tracer }) returns three things.

Field	Pass it to
`hookm`	`connect({ hookm })`. Transactions and errors the caller does not receive.
`onRoute`	`connect({ onRoute })`. A topology calls it after it chooses an endpoint.
`query`	Call it with `inspect()`. That is the fingerprint.

A string or a pool never calls onRoute. Routing spans need { primary, replicas }.

Attributes

Every span sets db.system.name to postgresql. The span kind is client. Attributes are passed when the span starts.

Span	Name	Attributes
Route	`read`, `write`, `tx`, or `lock`	`db.operation.name` is that same word. The endpoint and the reason are a span event named `route`.
Transaction	`transaction`, or `savepoint` when the depth is above 0	`db.operation.name` matches the name. Commit sets status OK. Rollback sets status ERROR and `error.type` to `rollback`.
Error	`error`	`error.type` is the error's name. Status is ERROR.
`inspect()`	`find notes`, and the same shape for the other reads	`db.operation.name` is the call. `db.collection.name` is the table. `db.query.summary` is the fingerprint.

onRoute fires once, so that span starts and ends in the same call. It has no duration. The route event does not carry a fingerprint, a table, or statement text. query(inspect()) is what sets db.query.summary.

Statement text is off. statement: true sets db.query.text from that inspection. Parameters are never attributes. A tenant id is not an attribute.

onError is an error the caller does not receive, such as one thrown from afterCommit. A query that throws to the caller is not that event. The exception is the error name and the OKM code when the error has one, with no message and no stack.

A minimal in-memory tracer prints what okmodel/otel records:

otel.ts

import { connect } from "okmodel/pg/postgresjs";
import { schema, table, t } from "okmodel/pg";
import { otel, type OtelSpan, type OtelTracer } from "okmodel/otel";

const notes = table("notes", {
  id: t.identity(),
  title: t.text(),
});

const app = schema({ tables: [notes] });

const spans: Array<{
  name: string;
  attributes: Record<string, string | number | boolean>;
  status?: number;
}> = [];

const tracer: OtelTracer = {
  startSpan(name, options) {
    const seen = {
      name,
      attributes: { ...(options?.attributes ?? {}) },
      status: undefined as number | undefined,
    };
    spans.push(seen);
    const span: OtelSpan = {
      setAttribute(key, value) {
        seen.attributes[key] = value;
      },
      setStatus(status) {
        seen.status = status.code;
      },
      recordException() {},
      addEvent() {},
      end() {},
    };
    return span;
  },
};

const tracing = otel({ tracer });

export async function demo(url: string) {
  await using db = connect(url, {
    schema: app,
    hookm: [tracing.hookm],
  });
  const pending = db.notes.find({ limit: 1 });
  tracing.query(await pending.inspect());
  await pending;
  await db.tx(async (tx) => {
    await tx.notes.find({ limit: 1 });
  });
  console.log(spans);
}

otel.out

[
  {
    name: "find notes",
    attributes: {
      "db.system.name": "postgresql",
      "db.namespace": "public",
      "db.operation.name": "find",
      "db.collection.name": "notes",
      "db.query.summary": "Q_…",
    },
  },
  {
    name: "transaction",
    attributes: {
      "db.system.name": "postgresql",
      "db.operation.name": "transaction",
    },
    status: 1,
  },
]

db.query.summary is the plan fingerprint from inspect() (a value such as Q_a76ebc70adc3e9f3). Status 1 is OpenTelemetry OK. Pass statement: true to otel when the span should carry db.query.text. The parameters still stay off.

On this page