@dbx-tools/postgres

Use connection-correct PostgreSQL advisory locks and a structured LISTEN/NOTIFY topic bus from Node.js. The package also provides optional message persistence, sender identity, and connection helpers.

Both work against a plain pg.Pool or anything structurally compatible with one, including the pool AppKit's Lakebase plugin exports, so a Databricks App gets them without a second database client or connection pool.

Quick start

Run protected work on the same PostgreSQL connection that holds the lock:

import { postgresConnectionOptions, withAdvisoryLock } from "@dbx-tools/postgres";
import { Pool } from "pg";

const pool = new Pool(postgresConnectionOptions({}));
const invoiceId = 42;

await withAdvisoryLock(pool, ["invoice", invoiceId], async (client) => {
  await client.query("UPDATE invoices SET status = 'sent' WHERE id = $1", [invoiceId]);
});

Capabilities

• Converts an arbitrary structured key (string, array, object, Date, or an explicit bigint) into a stable signed 64-bit advisory-lock identifier.
• Derives a legal Postgres channel name from whatever identifies a channel, so no call site has to sanitize an app name or a tenant id into an identifier.
• Holds a session lock on one dedicated pooled connection for the whole callback, and hands that connection to the callback so protected work runs on it.
• Holds a transaction lock released atomically by COMMIT or ROLLBACK, which is what one-time schema installation needs.
• Broadcasts a type/metadata/body envelope to every listening process over pg_notify, filtered by topic in-process.
• Fills in project, machine, process, deployment, and optional AppKit sender context automatically, with caller metadata always winning.
• Rejects a payload that would not survive a JSON round trip unchanged, at the call site rather than on the wire.
• Reconnects a dropped listener with bounded backoff while subscribers remain.
• Optionally persists every published message to a package-owned table, so a subscriber that was not listening can replay history by cursor, with a default TTL and cleanup on insert.
• Two payload styles: the whole envelope in the notification, or a pointer the listener reads back. The pointer lifts the 8000-byte NOTIFY limit and keeps the contents behind a table grant.

Session roles

Set DBX_TOOLS_POSTGRES_ROLE when every repository-owned PostgreSQL connection should assume one role after authenticating. postgresConnectionOptions() adds the role to pg and AppKit Lakebase pool startup options, while postgresServerSettings() returns the equivalent asyncpg server settings.

import { postgresConnectionOptions } from "@dbx-tools/postgres";
import { Pool } from "pg";

const pool = new Pool(postgresConnectionOptions({}));

An explicit second argument overrides the environment value. The helper retains existing options, including statement timeouts, and appends the role setting. Use postgresRoleStatement() only for setup work that must change role after a connection is already open.

When these options configure an AppKit Lakebase routing pool, they apply to its service-principal and per-user OBO pools. Every login identity that can reach the pool must therefore be a member of the configured role, or otherwise have permission to SET ROLE, before the pool can connect successfully.

Use with AppKit Lakebase

Use AppKit lakebase() to obtain and refresh the PostgreSQL pool. This package does not replace that plugin. Add it when work needs connection-correct advisory locks or cross-instance LISTEN/NOTIFY delivery, capabilities the native Lakebase plugin does not provide.

Advisory locks

import { postgresConnectionOptions, withAdvisoryLock } from "@dbx-tools/postgres";
import { Pool } from "pg";

const pool = new Pool(postgresConnectionOptions({}));

await withAdvisoryLock(pool, ["invoice", invoiceId], async (client) => {
  await client.query("UPDATE invoices SET status = 'sent' WHERE id = $1", [invoiceId]);
});

Run the protected work on the client the callback receives. A pool.query() inside the callback may land on a different connection, which is not the one holding the lock.

Use the transaction-scoped helper for one-time database work:

import { withAdvisoryTransactionLock } from "@dbx-tools/postgres";

await withAdvisoryTransactionLock(pool, { schema: "my_feature", version: 1 }, async (client) => {
  await client.query("CREATE TABLE IF NOT EXISTS my_feature.events (id bigint primary key)");
});

identity.advisoryLockId(key) exposes the same reduction as a JSON-safe decimal string. A bigint key is used directly (narrowed to 64 bits) rather than hashed, so a lock can interoperate with another implementation that publishes its numeric lock id, such as pgmq's installer lock, for instance. Anything else is canonicalized with object.toStableKey from @dbx-tools/shared-core and hashed, so key order in an object does not matter while a 1 and a "1" stay different locks. An array is read as several parts, so ["invoice", 7] and "invoice_7" are different locks. A non-finite number or a cyclic key throws TypeError rather than yielding an identity two callers could disagree about.

Topic bus

import { PostgresTopicBus } from "@dbx-tools/postgres";

const bus = new PostgresTopicBus(pool, {
  metadata: async () => ({ publicIp: await resolvePublicIp() }),
  onError: (cause) => logger.error("bus", { cause }),
});

const unsubscribe = await bus.listen("orders", ({ type, metadata, body }) => {
  console.log(type, metadata.project, body);
});

await bus.broadcast("orders", {
  type: "order.updated",
  metadata: { traceId: "abc-123" },
  body: { orderId: "123", status: "ready" },
});

await unsubscribe();
await bus.close();

Every process listening on the channel receives every message. There are no competing consumers. The bus fits live fan-out such as an SSE stream per browser tab, a cache invalidation, or a presence ping. It doesn't distribute work. Delivery is live-only by default. Turn on persist when a subscriber needs replay, and use a real queue when it needs acks.

Message envelope

publishedAt and id are for display and dedupe. Neither orders messages from different instances reliably.

Metadata precedence

Weakest to strongest: automatic machine context, then AppKit sender identity, then the bus's metadata option, then the per-message metadata. A key present at a stronger layer is never overwritten, so passing project explicitly replaces the inferred one. An explicit null is also preserved and prevents an automatic value from being used for that key.

Automatic keys, each omitted when it resolves to nothing: project (from DATABRICKS_APP_NAME, DATABRICKS_BUNDLE_NAME, PROJECT_NAME, or npm_package_name), hostname, platform, environment, appName, deploymentId, and databricksHost. Public/client IP, process id, CPU architecture, runtime name, and runtime version are left out on purpose: IPs and pids are noisy (and often sensitive), the rest are constant per deployment, and every message would pay for them against the payload limit. Pass a metadata function when a process needs one of those values.

The example above attaches a discovered public IP that way. The function runs on every broadcast, so memoize anything expensive.

With the optional @databricks/appkit peer installed and a user execution context active, senderId, senderName, and senderEmail are added too. AppKit is imported lazily and never required. Outside a request, or without the package, the bus skips those keys.

Supported message values

The body and metadata must survive JSON.parse(JSON.stringify(x)) unchanged, which is stricter than JSON.stringify not throwing. A Date becomes a string, NaN and Infinity become null, a Map becomes {}, and undefined vanishes. Each reaches the listener as something other than what was sent, so all of them throw TypeError at the call instead. Cycles, functions, symbols, bigints, and class instances are rejected for the same reason.

The rule lives in @dbx-tools/shared-core as object.isSerializableValue and object.SerializableValue, so a route can apply the same check to an untrusted request body and answer 400 rather than 500, and a browser-side caller can share the type.

The encoded envelope is capped at 7900 bytes, under PostgreSQL's 8000-byte NOTIFY limit, and a larger one throws RangeError. Automatic metadata counts against that, so broadcast a reference and let the receiver fetch anything big.

Connections, lifecycle, and failure

Publishing borrows a pooled connection per call. Listening cannot: LISTEN is session state, so the bus checks out one dedicated client and keeps it for as long as it has subscribers, however many topics and listeners are registered. Size the pool with that long-lived checkout in mind.

Nothing connects until the first listen, or an explicit start() when you would rather find out about a connection problem at boot. close() releases the client and is required. Register it with the host's shutdown hook. A closed bus stays closed and throws instead of silently reconnecting.

When the notification connection drops, the bus reconnects on its own: immediately first, then doubling from 250ms to a 5s ceiling, indefinitely, for as long as subscribers remain. A Postgres restart or a rotated Lakebase credential recovers without intervention. Messages published during the gap are lost. Every failed attempt, every dropped connection, and every listener that throws is reported through onError, which defaults to swallowing them. Configure a logger in anything long-running. A failing listener never affects the publisher or the other listeners.

One channel carries many topics, so adding a topic costs nothing. Give an unrelated high-volume stream its own channel because every listening session decodes every message on the channel.

Replay with persist

NOTIFY has no replay: a process that was not listening never learns what it missed. Set persist and each published message is INSERTed in the same transaction as its notification, so history becomes readable by cursor.

const bus = new PostgresTopicBus(pool, { persist: true });

await bus.broadcast("orders", { type: "order.updated", body: { id: 7 } });

let cursor: string | undefined;
do {
  const page = await bus.history("orders", { after: cursor, limit: 100 });
  for (const { message, expiresAt } of page.messages) handle(message, expiresAt);
  cursor = page.nextCursor;
} while (cursor);

The row and the notification commit together, which is the property the whole design rests on: NOTIFY is delivered at commit rather than at statement time, so a listener can never be told about a row it cannot yet read.

Messages EXPIRE by default. A 24-hour TTL is the default precisely because an unbounded message table is the failure this feature would otherwise introduce. Expired rows are deleted in bounded batches on each persisted publish, and cleanupExpired() runs the same sweep on demand. ttl: false opts one bus (or { ttl: false } on a single broadcast) out of expiry. It is available but isn't the default.

Two tiers, open and restricted, exist so one channel can carry messages with different read audiences: they are separate tables, so a role granted the open table cannot read restricted history. messageBusGrantStatements(roles, scope, options) emits exactly the grants a tier needs, and a history cursor is bound to the tier it came from. Resuming an open page against the restricted table throws instead of returning someone else's rows. Publish to a tier per-message with broadcast(topic, { scope: "restricted", ... }).

Set provision: false when a migration owns the schema. The statements are otherwise applied once, under an advisory transaction lock, so concurrent instances installing at boot do not race.

Envelope or pointer

payload: "envelope" (the default) puts the whole message in the notification. One round trip, no read amplification, bounded by PostgreSQL's payload limit.

payload: "pointer" notifies with routing and identity only ({ v, scope, topic, id, sequence }), and the listener reads the envelope back. That lifts the size limit entirely (a body larger than 7900 bytes is only publishable this way) and means a subscriber without the tier's table grant learns that something happened without learning what. The cost is one read per notification. Listeners receive the same TopicMessage either way. The difference is invisible above listen.

A pointer whose row is already gone (TTL cleanup won the race) or unreadable delivers nothing instead of raising, which is expected on the restricted tier. Use decodePointer if a non-bus consumer needs to read the raw notification.

Name a channel

channel takes whatever identifies the channel, not a pre-sanitized identifier: a name, a tenant id, a [env, feature] pair, or a config object. One value or many: an array is read as several parts, anything else as one.

new PostgresTopicBus(pool, { channel: "billing-events" }).channelName;
// "billing_events_3x55ck"
new PostgresTopicBus(pool, { channel: ["billing", "production"] }).channelName;
// "billing_production_008jgf"

The parts are tokenized into the readable half and a short hash of their canonical form (object.toStableKey) is appended. The hash is what makes the mapping trustworthy: tokenizing alone is lossy, so my-app, my_app, and myApp would collapse onto one channel, and a name long enough to hit Postgres's 63-character limit would collide with anything sharing its leading tokens. With the suffix the readable part stays readable and distinct inputs stay distinct, including ["billing", "prod"] versus "billing_prod", which differ in structure.

Derivation is deterministic across processes and runs, so every participant that passes equivalent parts lands on the same channel without coordinating. Object key order does not matter. A different spelling does. Read bus.channelName to see what a set of parts resolved to, or to confirm two processes agree. Defaults to a shared dbx_tools_topic_bus channel.

identity.channelName(parts) exposes the derivation without constructing a bus, which is useful for configuration checks and cross-runtime contract tests.

Connection lifecycles

Advisory locks are connection-scoped, and incorrect pool use can fail invisibly: taking the lock with pool.query() and doing the protected work with another pool.query() can use two different connections, so the lock protects nothing and the code looks correct. The same applies to LISTEN, which is session state a pooled query cannot hold. These helpers keep both lifecycles on dedicated connections and depend only on pg.

AppKit supplies the Lakebase pool. These helpers add locks and cross-instance message delivery on top of it.

Module map

Both modules are also flattened onto the package root, so import { withAdvisoryLock, PostgresTopicBus } from "@dbx-tools/postgres" works without the namespace.