pg-pubsub package

Reliable PostgreSQL LISTEN/NOTIFY for Node.js — with an inter-process lock so a horizontally scaled service handles each notification once.

Start from PgPubSub: construct it with PgPubSubOptions, subscribe channels inside its 'connect' handler, and read messages from the instance's 'message' event or from the per-channel emitter on channels.

Remarks

The problem this solves is that LISTEN/NOTIFY is a broadcast: every listening connection receives every notification, so a service scaled to N replicas handles each message N times. With singleListener on — the default — the replicas compete for a per-channel lock held as a row in PostgreSQL, and only the holder listens. The others stay connected as hot standbys.

That makes delivery at-most-once, and the trade is worth stating plainly: NOTIFY has no backlog, so anything published while no process holds the lock is gone. A clean shutdown releases the lock and a standby takes over at once; an unclean exit leaves the channel unhandled until the next retry, bounded by ACQUIRE_INTERVAL. Payloads are also capped at 8000 bytes by PostgreSQL. Where losing a message is unacceptable, pair this with a durable queue rather than replacing one.

Example

import { type AnyJson, PgPubSub } from '@imqueue/pg-pubsub';

const pubSub = new PgPubSub({ connectionString: process.env.DB_URL });

pubSub.on('connect', async () => {
    await pubSub.listen('UserChanged');
});
pubSub.on('message', (channel: string, payload: AnyJson) =>
    console.log(channel, payload),
);

await pubSub.connect();
await pubSub.notify('UserChanged', { id: 1 });

Classes

Class

Description

NoLock

Implements no lock to be used with multi-listener approach

PgIpLock

Implements manageable inter-process locking mechanism over existing PostgreSQL connection for a given LISTEN channel.

It uses periodic locks acquire retries and implements graceful shutdown using SIGINT, SIGTERM and SIGABRT OS signals, by which safely releases an acquired lock, which causes an event to other similar running instances on another processes (or on another hosts) to capture free lock.

By running inside Docker containers this would work flawlessly on implementation auto-scaling services, as docker destroys containers gracefully.

Currently, the only known issue could happen only if, for example, database or software (or hardware) in the middle will cause a silent disconnect. For some period of time, despite the fact that there are other live potential listeners some messages can go into void. This time period can be tuned by bypassing wanted acquireInterval argument. By the way, take into account that too short period and number of running services may cause huge flood of lock acquire requests to a database, so selecting the proper number should be a thoughtful trade-off between overall system load and reliability level.

Usually you do not need to instantiate this class directly - it will be done by a PgPubSub instances on their needs. Therefore, you may re-use this piece of code in some other implementations, so it is exported as is.

PgPubSub

Implements LISTEN/NOTIFY client for PostgreSQL connections.

It is a basic public interface of this library, so the end-user is going to work with this class directly to solve his/her tasks. Construct it with PgPubSubOptions, subscribe channels once connected, then read messages from either the instance's own 'message' event or the per-channel emitter on PgPubSub.channels.

Functions

Function	Description
channel(payload)	Channel listener event, occurs whenever the listening channel gets a new payload message.
close()	`'close'` event, occurs each time connection closed. Differs from `'end'` event, because `'end'` event may occur many times during re-connectable connection process, but `'close'` event states that connection was safely programmatically closed and further re-connections won't happen.
connect()	`'connect'` event, occurs each time database connection is established.
enableGracefulShutdown()	Registers SIGINT/SIGTERM/SIGABRT handlers performing graceful release of all instantiated locks and process exit. Idempotent. Opt-in: importing this package does not take over the process lifecycle by itself - either call this function directly or construct PgPubSub with `handleSignals: true`.
end()	`'end'` event, occurs whenever pg connection ends, so, literally it's simply proxy to `'end'` event from `pg.Client`
error(err)	`'error'` event occurs each time connection error is happened
listen(channels)	`'listen'` event occurs each time channel starts being listening
message(chan, payload)	`'message'` event occurs each time database connection gets notification to any listening channel. Fired before channel event emitted.
notify(chan, payload)	`'notify'` event occurs each time new message has been published to a particular channel. Occurs right after database NOTIFY command succeeded.
pack(input, logger, pretty)	Serializes given input object to JSON string. On error will return serialized null value
reconnect(retries)	`'reconnect'` event occurs each time, when the connection is successfully established after connection retry. It is followed by a corresponding `'connect'` event, but after all possible channel locks finished their attempts to be re-acquired.
signature(processId, channel, payload)	Constructs and returns hash string for a given set of processId, channel and payload.
unlisten(channels)	`'unlisten'` event occurs each time channel ends being listening
unpack(input, logger)	Deserializes given input JSON string to corresponding JSON value object. On error will return empty object

Interfaces

Interface	Description
AnyLock	Lock implementation interface to follow
AnyLogger	Represents logger interface suitable to be injected into this library objects
JsonMap	Represents JSON serializable object
PgClient	Extends `pg.Client` with additional properties
PgPubSubOptions	Options accepted as option argument of PgPubSub constructor. It extends `pg.ClientConfig` options, mostly because it is used to construct PostgreSQL database connection, adding more properties required to configure PgPubSub objects behavior.

Variables

Variable	Description
ACQUIRE_INTERVAL	How often, in milliseconds, a process without the lock retries acquiring it. Default 30000. This is the failover bound, and it is asymmetric. A clean shutdown releases the lock and the waiting processes are notified at once, so takeover is near-immediate. But if the lock holder dies WITHOUT releasing — SIGKILL, a lost container, a hard crash — nothing notifies anyone, and the channel stays unhandled until the next retry fires. Any notification published in that window is lost outright, because LISTEN/NOTIFY has no backlog to replay. So this value is the worst-case gap in message handling after an unclean exit. Lower it if that gap matters more than the polling cost of the extra lock acquisition attempts.
DefaultOptions	Hard-coded pre-set of PgPubSubOptions
EXECUTION_LOCK	Default for `PgPubSubOptions.executionLock` — off. Enable with `PG_PUBSUB_EXECUTION_LOCK`. When on, a message is de-duplicated by CONTENT rather than by listener: a marker row is written for each handled payload, and an identical payload arriving again within UNIQUE_LOCK_TTL is skipped. That protects against duplicate publication, but it also means two legitimately identical notifications collapse into one — which is a correctness problem if your payloads are not unique.
IS_ONE_PROCESS	Default for `PgPubSubOptions.singleListener` — on. With it on, an inter-process lock elects exactly ONE process to receive each channel's notifications, which is what makes scaling out safe: LISTEN/NOTIFY delivers to every listening connection, so without the lock every replica handles every message. Turn it off only when you genuinely want a broadcast to all replicas.
RETRY_DELAY	Delay in milliseconds between reconnect attempts after the PostgreSQL connection drops.
RETRY_LIMIT	How many reconnect attempts to make before emitting `'error'` and giving up. `Infinity` by default, so a client reconnects indefinitely and survives a database restart without supervision.
RX_LOCK_CHANNEL	Matches the internal prefix PgIpLock puts on its own coordination channels, so a listener can tell lock traffic apart from application notifications. The prefix is repeated rather than nested — hence the `+` — so this strips any number of levels in one pass.
SCHEMA_NAME	PostgreSQL schema holding the inter-process lock table, read from `PG_PUBSUB_SCHEMA_NAME` and defaulting to `pgip_lock`. Every process competing for the same channel must agree on this name — the lock is a row in that schema, so two deployments configured differently would each elect their own listener and both would handle the same notification.
SHUTDOWN_TIMEOUT	How long, in milliseconds, PgPubSub.destroy() waits for a graceful shutdown before giving up. Read from `PG_PUBSUB_SHUTDOWN_TIMEOUT`, default 1000. The wait matters because releasing a lock cleanly is what lets a standby take over immediately: the release deletes the lock row, which fires a NOTIFY the waiting processes are listening for. A shutdown that is killed before the release completes falls back to ACQUIRE_INTERVAL instead.
UNIQUE_LOCK_TTL	Time-to-live (seconds) of processed-message markers created by execution locks; expired markers are cleaned up on subsequent unique lock acquisitions

Type Aliases

Type Alias	Description
AnyJson	Represents any JSON-serializable value
JsonArray	Represents JSON-serializable array

Read this page as plain markdown — no HTML, no navigation. For pasting into an LLM, or for an agent to fetch.