# @imqueue/pg-pubsub 3.0.5 · API reference

Source: https://imqueue.org/api/pg-pubsub/latest/
Published: 2026-08-01
Author: @imqueue maintainers (https://github.com/imqueue)
Package: @imqueue/pg-pubsub 3.0.5 — generated reference, not hand-written

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

Start from [PgPubSub](https://imqueue.org/api/pg-pubsub/latest/pg-pubsub.pgpubsub/): construct it with [PgPubSubOptions](https://imqueue.org/api/pg-pubsub/latest/pg-pubsub.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


```typescript
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](https://imqueue.org/api/pg-pubsub/latest/pg-pubsub.nolock/) | Implements no lock to be used with multi-listener approach |
| [PgIpLock](https://imqueue.org/api/pg-pubsub/latest/pg-pubsub.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](https://imqueue.org/api/pg-pubsub/latest/pg-pubsub.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](https://imqueue.org/api/pg-pubsub/latest/pg-pubsub.pgpubsuboptions/), subscribe channels once connected, then read messages from either the instance's own `'message'` event or the per-channel emitter on [PgPubSub.channels](https://imqueue.org/api/pg-pubsub/latest/pg-pubsub.pgpubsub.channels/). |


## Functions


| Function | Description |
| --- | --- |
| [channel(payload)](https://imqueue.org/api/pg-pubsub/latest/pg-pubsub.channel/) | Channel listener event, occurs whenever the listening channel gets a new payload message. |
| [close()](https://imqueue.org/api/pg-pubsub/latest/pg-pubsub.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()](https://imqueue.org/api/pg-pubsub/latest/pg-pubsub.connect/) | `'connect'` event, occurs each time database connection is established. |
| [enableGracefulShutdown()](https://imqueue.org/api/pg-pubsub/latest/pg-pubsub.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()](https://imqueue.org/api/pg-pubsub/latest/pg-pubsub.end/) | `'end'` event, occurs whenever pg connection ends, so, literally it's simply proxy to `'end'` event from `pg.Client` |
| [error(err)](https://imqueue.org/api/pg-pubsub/latest/pg-pubsub.error/) | `'error'` event occurs each time connection error is happened |
| [listen(channels)](https://imqueue.org/api/pg-pubsub/latest/pg-pubsub.listen/) | `'listen'` event occurs each time channel starts being listening |
| [message(chan, payload)](https://imqueue.org/api/pg-pubsub/latest/pg-pubsub.message/) | `'message'` event occurs each time database connection gets notification to any listening channel. Fired before channel event emitted. |
| [notify(chan, payload)](https://imqueue.org/api/pg-pubsub/latest/pg-pubsub.notify/) | `'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)](https://imqueue.org/api/pg-pubsub/latest/pg-pubsub.pack/) | Serializes given input object to JSON string. On error will return serialized null value |
| [reconnect(retries)](https://imqueue.org/api/pg-pubsub/latest/pg-pubsub.reconnect/) | `'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)](https://imqueue.org/api/pg-pubsub/latest/pg-pubsub.signature/) | Constructs and returns hash string for a given set of processId, channel and payload. |
| [unlisten(channels)](https://imqueue.org/api/pg-pubsub/latest/pg-pubsub.unlisten/) | `'unlisten'` event occurs each time channel ends being listening |
| [unpack(input, logger)](https://imqueue.org/api/pg-pubsub/latest/pg-pubsub.unpack/) | Deserializes given input JSON string to corresponding JSON value object. On error will return empty object |


## Interfaces


| Interface | Description |
| --- | --- |
| [AnyLock](https://imqueue.org/api/pg-pubsub/latest/pg-pubsub.anylock/) | Lock implementation interface to follow |
| [AnyLogger](https://imqueue.org/api/pg-pubsub/latest/pg-pubsub.anylogger/) | Represents logger interface suitable to be injected into this library objects |
| [JsonMap](https://imqueue.org/api/pg-pubsub/latest/pg-pubsub.jsonmap/) | Represents JSON serializable object |
| [PgClient](https://imqueue.org/api/pg-pubsub/latest/pg-pubsub.pgclient/) | Extends `pg.Client` with additional properties |
| [PgPubSubOptions](https://imqueue.org/api/pg-pubsub/latest/pg-pubsub.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](https://imqueue.org/api/pg-pubsub/latest/pg-pubsub.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](https://imqueue.org/api/pg-pubsub/latest/pg-pubsub.defaultoptions/) | Hard-coded pre-set of PgPubSubOptions |
| [EXECUTION\_LOCK](https://imqueue.org/api/pg-pubsub/latest/pg-pubsub.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](https://imqueue.org/api/pg-pubsub/latest/pg-pubsub.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](https://imqueue.org/api/pg-pubsub/latest/pg-pubsub.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](https://imqueue.org/api/pg-pubsub/latest/pg-pubsub.retry_delay/) | Delay in milliseconds between reconnect attempts after the PostgreSQL connection drops. |
| [RETRY\_LIMIT](https://imqueue.org/api/pg-pubsub/latest/pg-pubsub.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](https://imqueue.org/api/pg-pubsub/latest/pg-pubsub.rx_lock_channel/) | Matches the internal prefix [PgIpLock](https://imqueue.org/api/pg-pubsub/latest/pg-pubsub.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](https://imqueue.org/api/pg-pubsub/latest/pg-pubsub.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](https://imqueue.org/api/pg-pubsub/latest/pg-pubsub.shutdown_timeout/) | How long, in milliseconds, [PgPubSub.destroy()](https://imqueue.org/api/pg-pubsub/latest/pg-pubsub.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](https://imqueue.org/api/pg-pubsub/latest/pg-pubsub.acquire_interval/) instead. |
| [UNIQUE\_LOCK\_TTL](https://imqueue.org/api/pg-pubsub/latest/pg-pubsub.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](https://imqueue.org/api/pg-pubsub/latest/pg-pubsub.anyjson/) | Represents any JSON-serializable value |
| [JsonArray](https://imqueue.org/api/pg-pubsub/latest/pg-pubsub.jsonarray/) | Represents JSON-serializable array |