Reliable PostgreSQL LISTEN/NOTIFY with inter-process lock support

This library provides a clean way to use PostgreSQL LISTEN and NOTIFY commands for its asynchronous mechanism implementation. It comes as a top-level wrapper over node-postgres and provides better, cleaner way to work with database notifications engine.

To make it clear - it solves several major problems you will fall into if you're going to use LISTEN/NOTIFY in your node app:

1. Reliable connections. This library comes with handy reconnect support out-of-the box, so all you need, is, probably to tune several settings if you have special needs, like max retry limit or reconnection delay.
2. It provides clean way working with channels, so you may subscribe to an exactly required channel with no need to do additional filtering implementation on messages receive. BTW, it does not hide from you possibility to manage all messages in a single handler. You just choose what you need.
3. The most important feature here is that this library comes with the first-class implementation of inter-process locking mechanism, allowing avoiding data duplication receive problem in scalable distributed architectures. It means it allows you to define single-listener process across many similar processes (which happens on scales) which would receive notifications and with a guarantee that if it looses connection or dies - another similar process replaces it as listener.
4. It comes with support of graceful shutdown, so you may don't care about this.

As easy as:

npm i --save @imqueue/pg-pubsub

It supports passing environment variables to configure locker schema name to use and shutdown timeout.

• PG_PUBSUB_SCHEMA_NAME - string, by default is 'pgip_lock'
• PG_PUBSUB_SHUTDOWN_TIMEOUT - number, by default is 1000, in milliseconds

import{PgPubSub}from'@imqueue/pg-pubsub';constconnectionString='postgres://user:pass@localhost:5432/dbname';constpubSub=newPgPubSub({ connectionString,singleListener: false});(async()=>{awaitpubSub.connect();})();

With such instantiation options natural behavior of PgPubSub will be as follows:

See all options.

After connection established you may decide to listen for any numbers of channels your application may need to utilize:

awaitpubSub.listen('UserChanged');awaitpubSub.listen('OrderCreated');awaitpubSub.listen('ArticleUpdated');

BTW, the most reliable way is to initiate listening on 'connect' event:

pubSub.on('connect',async()=>{awaitPromise.all(['UserChanged','OrderCreated','ArticleUpdated',].map(channel=>pubSub.listen(channel)));});

Now, whenever you need to close/reopen connection, or reconnect occurred for any reason you'll be sure nothing broken.

All payloads on messages treated as JSON, so when the handler catches a message it is already parsed as JSON value, so you do not need to manage serialization/deserialization yourself.

There are 2 ways of handling channel messages - by using 'message' event handler on pubSub object, or using pubSub.channels event emitter and to listen only particular channel for its messages. On message event fires first, channels events fires afterwards, so this could be a good way if you need to inject and transform a particular message in synchronously manner before it will come to a particular channel listeners.

Also 'message' listener could be useful during implementation of handling of database side events. It is easy imagine that db can send us messages into, so called, structural channels, e.g. 'user:insert', 'company:update' or 'user_company:delete', where such names generated by some generic trigger which handles corresponding database operations and send updates to subscribers using NOTIFY calls. In such case we can treat channel on application side as self-describable database operation change, which we can easily manage with a single piece of code and keep following DRY.

// using 'message' handler:pubSub.on('message',(channel: string,payload: AnyJson)=>{// ... do the jobswitch(channel){case'UserChanged': {// ... do some staff with user change event payloadbreak;}default: {// do something with payload by defaultbreak;}}});

// handling using channelspubSub.channels.on('UserChanged',(payload: AnyJson)=>{// do something with user changed payload});pubSub.channels.on('OrderCreated',(payload: AnyJson)=>{// do something with order created payload});pubSub.channels.on('ArticleUpdated',(payload: AnyJson)=>{// do something with article updated payload});

Of course, it is better to set up listeners before calling connect() that it starts handle payloads right up on connect time.

You can send messages in many ways. For example, you may create database triggers which would notify all connected clients with some specific updates. Or you may use a database only as notifications engine and generate notifications on application level. Or you may combine both approaches - there are no limits!

Here is how you can send notification with PgPubSub API (aka application level of notifications):

pubSub.notify('UserChanged',{old: {id: 777,name: 'John Doe',phone: '555-55-55'},new: {id: 777,name: 'Sam Peters',phone: '777-77-77'},});

Now all subscribers, who listening 'UserChanged' channel will receive a given payload JSON object.

There are variety of many possible architectures to come up with when you're building scalable distributed system.

With services on scale in such systems it might be a need to make sure only single service of much similar running is listening to particular database notifications. Here why comes an idea of inter process (IP) locking mechanism, which would guarantee that only one process handles notifications and if it dies, next one which is live will immediately handle listening.

This library comes with this option turned on by default. To make it work in such manner, you would need to skip passing singleListener option to PgPubSub constructor or set it to true:

constpubSub=newPgPubSub({ connectionString });// or, equivalentlyconstpubSub=newPgPubSub({ connectionString,singleListener: true});

Locking mechanism utilizes the same connection and LISTEN/NOTIFY commands, so it won't consume any additional computing resources.

Also, if you already work with pg library in your application, and you have a need to stay for some reason with that single connection usage, you can bypass it directly as pgClient option, but that is not always a good idea. Normally, you have to understand what you are doing and why.

constpubSub=newPgPubSub({pgClient: existingClient});

NOTE: With LISTEN connections it is really hard to utilize power of connection pool as long as it will require additional implementation of some connection switching mechanism using listen/unlisten and some specific watchers which may fall into need of re-implementing pools from scratch. So, that is why most of existing listen/notify solutions based on a single connection approach. And this library as well. It is just more simple and reliable.

Also, PgPubSub supports execution lock. This means all services become listeners in single listener mode but only one listener can process a notification. To enable this feature, you can bypass executionLock as option and set it to true. By default, this lock type is turned off.

NOTE: Sometimes you might receive the notification with the same payloads in a very short period of time but execution lock will process them as the only notify message. If this important to you and your system will lave data leaks you need to ensure that payloads are unique.

• Error handling: always subscribe to the 'error' event. Connection errors are forwarded there; when no listener is attached they are routed to the configured logger instead of crashing the process.
• Automatic reconnect recreates the underlying pg client (pg clients are single-use), so construct PgPubSub with connection options (connectionString etc.) rather than a pre-made pgClient instance if you rely on reconnects. Do not cache the pgClient reference across reconnects.
• Graceful shutdown is opt-in: importing the package no longer registers process signal handlers. Construct with handleSignals: true or call enableGracefulShutdown() to get SIGINT/SIGTERM/SIGABRT handling with automatic locks release.
• Database privileges: the first run bootstraps the lock schema (CREATE SCHEMA/TABLE/FUNCTION/TRIGGER), which requires DDL rights. In locked-down environments provision it manually beforehand (see the SQL in src/PgIpLock.ts) - initialization failures are logged and locking will not work without the schema.
• Delivery semantics: LISTEN/NOTIFY is at-most-once with no backlog - messages published while a subscriber is reconnecting are lost, and NOTIFY payloads are limited to 8000 bytes (notify() throws a RangeError beyond that). Per-message execution locks keep a processed-marker row for one hour (UNIQUE_LOCK_TTL) to guarantee exactly-once handling across competing listeners.
• Integration tests: PG_TEST_DSN=... npm run test:integration runs the real-PostgreSQL flow suite (also wired into CI with a postgres service container).

You may read API docs on wiki pages , read the code of the library itself, use hints in your IDE or generate HTML docs with:

git clone git@github.com:imqueue/pg-pubsub.git
cd pg-pubsub
npm i
npm run doc

Try to run the following minimal example code of single listener scenario (do not forget to set proper database connection string):

import{PgPubSub}from'@imqueue/pg-pubsub';importTimer=NodeJS.Timer;lettimer: Timer;constNOTIFY_DELAY=2000;constCHANNEL='HelloChannel';constpubSub=newPgPubSub({connectionString: 'postgres://postgres@localhost:5432/postgres',singleListener: true,// filtered: true,});pubSub.on('listen',channel=>console.info(`Listening to ${channel}...`));pubSub.on('connect',async()=>{console.info('Database connected!');awaitpubSub.listen(CHANNEL);timer=setInterval(async()=>{awaitpubSub.notify(CHANNEL,{hello: {from: process.pid}});},NOTIFY_DELAY);});pubSub.on('notify',channel=>console.log(`${channel} notified`));pubSub.on('end',()=>console.warn('Connection closed!'));pubSub.channels.on(CHANNEL,console.log);pubSub.connect().catch(err=>console.error('Connection error:',err));

Or take a look at other minimal code examples

Play with them locally:

git clone -b examples git://github.com/imqueue/pg-pubsub.git examples
cd examples
npm i

Now you can start any of them, for example:

./node_modules/.bin/ts-node filtered.ts

Any contributions are greatly appreciated. Feel free to fork, propose PRs, open issues, do whatever you think may be helpful to this project. PRs which passes all tests and do not brake tslint rules are first-class candidates to be accepted!

This project is licensed under the GNU General Public License v3.0. See the LICENSE

Happy Coding!