AbsoluteJS

@absolutejs/sync-bus-pg

@absolutejs/sync-bus-pgv0.2.3betaData & Sync

Typed Postgres LISTEN/NOTIFY channel and @absolutejs/sync cluster bus — horizontal scale without standing up Redis

#Installation

BASH
bun add @absolutejs/sync-bus-pg

#Capabilities

Overview

Postgres LISTEN/NOTIFY cluster bus for @absolutejs/sync. Run sync horizontally across several Bun processes without standing up Redis or Kafka — your existing Postgres carries the cross-instance change feed.

Why

@absolutejs/sync ships a ClusterBus seam: an in-memory bus for single-process dev + tests, and a contract you implement against your bus of choice for production. Until now that meant writing 50 lines of LISTEN/NOTIFY plumbing yourself. This package is the first-party implementation, with the 8000-byte NOTIFY payload limit handled cleanly.

Use

Every instance of your Elysia app does the same. A mutation committed on instance A now fans out to subscribers on B/C/... via Postgres pg_notify.

Options

spill strategies:

'overflow' (default) — inline JSON when small, table-backed when oversized. Best for typical workloads.

'always' — every message goes through the sync_cluster_spill table (durable, slightly slower; useful when you want every cross-instance change to survive a NOTIFY drop).

Show 1 more

'never' — throws if a message exceeds the inline budget. Useful in tests to assert payload-size discipline.

Listener health and reconnects

postgres.js automatically recreates its dedicated listener connection and reissues LISTEN after a disconnect. A query-pool health check cannot prove that recovery has completed, so this adapter also sends a private probe through the full pg_notify → dedicated listener path every 15 seconds.

Probe envelopes never reach application callbacks and do not inflate message publish/receive counters. Monitoring starts with the first subscription and stops with the last unsubscribe. Set listenerHealth: false only when another owner calls probeListener() on its own cadence.

Vacuum

Oversized messages spill to sync_cluster_spill. Rows aren't auto-deleted on consume (every listener on the channel needs to read them, including the publisher's own listener which fetches but doesn't double-apply via the engine's origin filter). Sweep periodically:

For workloads where messages stay small (the common case), the spill table never gets touched and vacuum() always returns 0.

Caveats inherited from the engine seam

Per-instance version cursors. A client that reconnects to a _different_ instance falls back to a fresh snapshot (cold-hydration cost, not catch-up diff). Use sticky sessions if cross-instance reconnect-with-since matters.

Best-effort delivery. Inline NOTIFY can be lost if a listener connection drops mid-stream — every instance also has its own change log for resume, so a missed cross-instance fan-out is recovered on the next subscribe. For at-least-once cross-instance, run with spill: 'always'.

Outcomes

What you can build

Build on the supported package contract

Use @absolutejs/sync-bus-pg through its supported public entry points.

Hardening checklist

Production guidance

Make every external boundary explicitPin the deployed @absolutejs/sync-bus-pg version, replace example or memory-backed dependencies with durable implementations, bound external calls, protect credentials, and emit enough evidence to retry or recover safely.

Follow in order

Troubleshooting path

1
Trace from the first failed boundary
Reproduce the smallest canonical @absolutejs/sync-bus-pg example, confirm the supported entry point and version in the API explorer, then inspect the first boundary that did not produce its documented result.

#Use

Partial snippet

Working example for Use.

TS
import postgres from 'postgres';
import { createSyncEngine } from '@absolutejs/sync/engine';
import { createPostgresClusterBus } from '@absolutejs/sync-bus-pg';

const sql = postgres(process.env.DATABASE_URL!);
const engine = createSyncEngine();
// ...registerReader/Writer/Reactive/Mutation as usual...

const bus = createPostgresClusterBus({ sql });
await engine.connectCluster(bus);

#Options

Partial snippet

Working example for Options.

TS
createPostgresClusterBus({
	sql, // your postgres client
	channel: 'absolutejs_sync_cluster', // override to scope multiple engines on the same PG
	spill: 'overflow', // 'overflow' (default) | 'always' | 'never'
	listenerHealth: {
		// optional; these are the defaults
		probeIntervalMs: 15_000,
		probeTimeoutMs: 5_000
	},
	onError: (e) => log.warn(e) // listener-side errors
});

#Public entry points

Supported entry points declared by this package manifest.

Package entry point declared in package.json.

@absolutejs/sync-bus-pg@absolutejs/sync-bus-pg/manifest@absolutejs/sync-bus-pg/manifest.json

#Package commands

Scripts declared by this package manifest.

bun run buildrm -rf dist && bun build src/index.ts src/manifest.ts --outdir dist --sourcemap --target=bun --external @absolutejs/sync --external postgres && tsc --project tsconfig.build.json && absolute-manifest emit
bun run formatprettier --write "./**/*.{ts,json,md}"
bun run testbun test
bun run typechecktsc --noEmit

#API reference

Search the declarations exported by the current package type files. Expand a symbol to inspect its source-backed signature.

9 symbols
PostgresClusterBusOptionstypePermalinkSource
TS
type PostgresClusterBusOptions = {
    /**
     * The `postgres` (https://github.com/porsager/postgres) client. We need
     * both a regular SQL connection (for publish + spill fetch) and the
     * ability to listen on a channel; `postgres` exposes both via the same
     * `Sql` instance.
     */
    sql: Sql;
    /**
     * Channel name passed to `LISTEN` / `pg_notify`. Defaults to
     * `'absolutejs_sync_cluster'`. Two engines on the same Postgres can scope
     * themselves to different channels by overriding this.
     */
    channel?: string;
    /**
     * Spill strategy. `'overflow'` (default): inline JSON when small, table-
     * backed when oversized. `'always'`: every message goes through the spill
     * table (durable, slower). `'never'`: throws if a message exceeds the
     * inline budget — useful in tests to assert payload-size discipline.
     */
    spill?: 'overflow' | 'always' | 'never';
    /**
     * Called when the listener encounters an error (parse failure, missing
     * spill row, etc). Defaults to `console.warn`.
     */
    onError?: (error: unknown) => void;
    /**
     * End-to-end listener monitoring. The bus periodically publishes a private
Exported from @absolutejs/sync-bus-pg