What is the status of LISTEN / NOTIFY for production purposes in 2025

[slidenerd]

• I have an express typescript application that uses server sent events
• I am trying to get notified when a new row is inserted into a specific table feed_items on my end
• I see that you have put up a sample for a Robust Listeners example here. Would be nice if you could modify it to use async await though 😅 since we are in 2025 now
• I also notice that it seems to have some issues with AWS RDS which is where I primarily intend to run my DB like the quirks being discussed here
• Took a look at pg and it seems to have even more unresolved issues
• Found this library but the author of this other library pg-listen claims that none of these work reliably
• I noticed that pg-listen has a setInterval where it keeps sending a query every 30 seconds
• Even found this dude that claims to offer a better listen notify library but obviously looks pretty deserted there and not worth risking
• I am genuinely overwhelmed by how many issues are there with respect to running a reliable listen / notify connection for postgres on RDS. (You need to worry about keepalive, take care of abrupt connectiong breaking mechanisms, retries etc etc)
• You are super experienced in all this stuff and could totally appreciate some directions on which library I should go with for doing listen notify stuff that won't turn into a production nightmare on AWS RDS 😅

[slidenerd]

since we are on the topic i also wanted to share something funny. I asked claude about Robust Listeners and it says, it has all sorts of errors 🤣

const pgp = require('pg-promise')({
    // Initialization Options
    capSQL: true
});

class PostgresNotificationListener {
    constructor(connectionConfig) {
        this.db = pgp(connectionConfig);
        this.connection = null;
        this.isShuttingDown = false;
        this.reconnectAttempts = 0;
        this.maxReconnectAttempts = 10;
        this.reconnectDelay = 5000;
        this.channels = new Set();
    }

    // Handle incoming notifications
    onNotification(data) {
        console.log('Received Notification:', {
            channel: data.channel,
            payload: data.payload,
            processId: data.processId
        });
    }

    // Set up event listeners for a client
    async setListeners(client, channels = []) {
        client.on('notification', this.onNotification.bind(this));
        
        // Listen to specified channels
        for (const channel of channels) {
            try {
                await this.connection.none('LISTEN $1~', channel);
                this.channels.add(channel);
                console.log(`Listening to channel: ${channel}`);
            } catch (error) {
                console.error(`Failed to listen to channel ${channel}:`, error);
                throw error;
            }
        }
    }

    // Remove event listeners
    removeListeners(client) {
        if (client) {
            client.removeListener('notification', this.onNotification.bind(this));
        }
    }

    // Handle connection loss
    async onConnectionLost(err, e) {
        if (this.isShuttingDown) {
            return;
        }

        console.log('Connection lost:', err.message || err);
        
        // Clean up the lost connection
        if (this.connection) {
            this.removeListeners(e.client);
            this.connection = null;
        }

        try {
            await this.reconnect();
            console.log('Successfully reconnected to database');
        } catch (error) {
            console.error('Failed to reconnect after maximum attempts');
            await this.shutdown();
        }
    }

    // Reconnect with exponential backoff
    async reconnect() {
        if (this.isShuttingDown) {
            throw new Error('Shutting down, not reconnecting');
        }

        this.reconnectAttempts++;
        
        if (this.reconnectAttempts > this.maxReconnectAttempts) {
            throw new Error(`Maximum reconnection attempts (${this.maxReconnectAttempts}) exceeded`);
        }

        const delay = Math.min(this.reconnectDelay * Math.pow(2, this.reconnectAttempts - 1), 30000);
        console.log(`Reconnection attempt ${this.reconnectAttempts}/${this.maxReconnectAttempts} in ${delay}ms`);

        await this.sleep(delay);

        try {
            const obj = await this.db.connect({
                direct: true,
                onLost: this.onConnectionLost.bind(this)
            });

            this.connection = obj;
            
            // Re-establish listeners for previously registered channels
            await this.setListeners(obj.client, Array.from(this.channels));
            
            // Reset reconnect attempts on successful connection
            this.reconnectAttempts = 0;
            
            return obj;
        } catch (error) {
            console.error(`Reconnection attempt ${this.reconnectAttempts} failed:`, error.message);
            return this.reconnect(); // Recursive retry
        }
    }

    // Initial connection setup
    async connect(channels = ['my-channel']) {
        try {
            console.log('Establishing initial connection...');
            const obj = await this.db.connect({
                direct: true,
                onLost: this.onConnectionLost.bind(this)
            });

            this.connection = obj;
            await this.setListeners(obj.client, channels);
            
            console.log('Successfully established initial connection');
            return obj;
        } catch (error) {
            console.error('Failed to establish initial connection:', error);
            throw error;
        }
    }

    // Add a new channel to listen to
    async addChannel(channel) {
        if (!this.connection) {
            throw new Error('No active connection');
        }

        try {
            await this.connection.none('LISTEN $1~', channel);
            this.channels.add(channel);
            console.log(`Added listener for channel: ${channel}`);
        } catch (error) {
            console.error(`Failed to add channel ${channel}:`, error);
            throw error;
        }
    }

    // Remove a channel listener
    async removeChannel(channel) {
        if (!this.connection) {
            throw new Error('No active connection');
        }

        try {
            await this.connection.none('UNLISTEN $1~', channel);
            this.channels.delete(channel);
            console.log(`Removed listener for channel: ${channel}`);
        } catch (error) {
            console.error(`Failed to remove channel ${channel}:`, error);
            throw error;
        }
    }

    // Send a notification (useful for testing)
    async sendNotification(channel, payload) {
        if (!this.connection) {
            throw new Error('No active connection');
        }

        try {
            await this.connection.none('NOTIFY $1~, $2', [channel, payload]);
            console.log(`Sent notification to ${channel}:`, payload);
        } catch (error) {
            console.error('Failed to send notification:', error);
            throw error;
        }
    }

    // Graceful shutdown
    async shutdown() {
        console.log('Shutting down notification listener...');
        this.isShuttingDown = true;

        if (this.connection) {
            try {
                // Unlisten from all channels
                for (const channel of this.channels) {
                    await this.connection.none('UNLISTEN $1~', channel);
                }
                
                this.removeListeners(this.connection.client);
                this.connection.done();
                this.connection = null;
                console.log('Connection closed gracefully');
            } catch (error) {
                console.error('Error during shutdown:', error);
            }
        }

        pgp.end();
        console.log('Shutdown complete');
    }

    // Utility function for delays
    sleep(ms) {
        return new Promise(resolve => setTimeout(resolve, ms));
    }
}

// Usage example
async function main() {
    const connectionConfig = {
        host: 'localhost',
        port: 5432,
        database: 'your_database',
        user: 'your_username',
        password: 'your_password'
    };

    const listener = new PostgresNotificationListener(connectionConfig);

    // Handle process termination gracefully
    const gracefulShutdown = async () => {
        console.log('Received shutdown signal');
        await listener.shutdown();
        process.exit(0);
    };

    process.on('SIGINT', gracefulShutdown);
    process.on('SIGTERM', gracefulShutdown);
    process.on('unhandledRejection', (reason, promise) => {
        console.error('Unhandled Rejection at:', promise, 'reason:', reason);
        gracefulShutdown();
    });

    try {
        // Connect and start listening
        await listener.connect(['my-channel', 'another-channel']);
        
        // Example: Send test notifications every 10 seconds
        setInterval(async () => {
            if (listener.connection && !listener.isShuttingDown) {
                try {
                    await listener.sendNotification('my-channel', JSON.stringify({
                        timestamp: new Date().toISOString(),
                        message: 'Test notification'
                    }));
                } catch (error) {
                    console.error('Failed to send test notification:', error);
                }
            }
        }, 10000);

        console.log('Notification listener is running. Press Ctrl+C to stop.');
        
    } catch (error) {
        console.error('Failed to start notification listener:', error);
        await listener.shutdown();
        process.exit(1);
    }
}

// Start the application
if (require.main === module) {
    main().catch(console.error);
}

module.exports = PostgresNotificationListener;

This is what it generated and calls it perfect 🤣

[vitaly-t]

The AI is known to lie a lot. But it doesn't matter, that code is quite old now, and the new one I provided in the UPDATE for robust listeners is better 😉

But just for fun, I asked AI Copilot on GitHub to analyze the new code, and here's the result

[vitaly-t]

These all remain unresolved issues. I tried once before, with pg-listener, but never finished it, it's not working, though it might be worth resuming work on it.

I used my own code from the example I provided, and it worked well on Windows/Linux/Mac, but I never tested it in AWS, which has its own connection-related corks.

Maybe check out this post, which looks good, though I never tested it. I might refactor my example based on that.

[Coinhexa]

interesting i just stumbled into this issue. Glad someone opened it. Have you tried pg-listen by any chance. what is your opinion about it?

[vitaly-t]

I think that library should be fine, but there is no way of knowing, given that it hasn't been updated in 5 years, which doesn't inspire confidence.

Anyhow, following this discussion, I've added an UPDATE to the robust listeners, with some good improvements. It should do the job well, at handling any scenario with listeners.

[vitaly-t]

A basic listener in TypeScript would look like this:

async function listen(channel: string, cb: (data: string) => void): Promise<null> {
    const sco = await db.connect({direct: true, onLost: () => listen(channel, cb)});
    sco.client.on('notification', ({payload}) => cb(payload));
    return sco.none('LISTEN $1~', channel);
}

...which I appended to the Robust Listeners.

But then you can improve on it, by adding support for multiple channels on the same connection, allow un-listening, closing unused event notification handlers, etc.

Usage:

listen('my_channel', (data) => {
    console.log(data);
}).catch((e: any) => {
    console. Error('DB Connection Issue:', e.message);
});

[vitaly-t]

You now can find way more in the update for robust listeners 😉

[vitaly-t]

Based on everything I implemented today I decided the volume of code and its importance warrant placing it all into its own library.

Therefore, I have started reviving the old unfinished pg-listener, rewriting from scratch, based on the complete solution.

From now on, please follow that repo. Once it is ready, it will replace all the old references to notification listening.

[vitaly-t]

UPDATE-1

Initial draft has been published as v0.0.2.
More updates are coming soon.
Feedback is welcome!

UPDATE-2

After v0.0.3, successfully tested the library locally, it works as expected. Now focus is on tests + documentation.

UPDATE-3

The library has reached v0.0.5, it is solid, according to my local tests, though unit tests are non-existent so far. At least documentation is enough to get started. It is no longer a draft; it is quite decent now.

I am parking it here for now, having spent all day on it today, because I myself will likely be needing it also. Plus, many people were asking about it in the past. I will get back to it when I have time again.

[slidenerd]

you sir are a freaking legend 🫡🫡🫡 every single time i run into any open source code on github with a tonne of issues, i always tell them "go to pg-promise and take a look at how to resolve issues and discussions" The world of software development would be 100% better if only 20% of them put as much effort as you do into maintaining stuff. Programmers think making a library is writing code, publishing it and moving on. The real work is actually maintaining libraries and doing bug fixes!

[vitaly-t]

Once pg-listener reached v0.0.9, I replaced everything inside Robust Listeners with it. This is the only way forward now.

[slidenerd]

I ll test this library within this week (stretched the time a bit, ll be working on this tomorrow and tuesday) and let ya know how it goes

[vitaly-t]

Make sure you do use keepAlive: true when creating the Database object, this one is important with AWS 😉

[vitaly-t]

@slidenerd @Coinhexa

Maybe you guys can help me with this little issue? 😉

[vitaly-t]

You guys have gone awfully quiet here, while I continued working on this 🤭

We now have very solid release, with almost 100% test coverage (plus GitHub CI integration), and I even have put together a separate solution for RxJs, in the examples 😉

[Coinhexa]

Apologies, direct connection works spendidly! 🫡 recommended approach for unit and integration testing?

[vitaly-t]

@slidenerd Will you be able to provide feedback on pg-listener? It's been a while :)

[slidenerd]

apologies due to a personal incident, i was afk for 3 days, i got back today, and have started work on this. will keep you posted

[slidenerd]

• Apologies for the late response, I was struggling to understand how I can put this together with better-sse
• I tested it, the library definitely seems to work but I am having a bit of trouble combining this with SSE (Kinda offtopic I guess)
• My solution above kinda aims to make this a little bit more mockable and testable but I am not sure if it opens a connection for every request

[vitaly-t]

I am not sure if it opens a connection for every request

If by request you mean a call to listen - then yes, it does. Each call to listen starts a new connection, as documented:

It allocates and fully occupies one physical connection to the database, thus allowing for the flexibility of choosing how to split channels across connections.

[vitaly-t]

My solution above kinda aims to make this a little bit more mockable and testabl

The library itself mocks and tests everything, which is already a proof that it is doable :) If you have issues doing that, I'd need specifics, what those problems are.

[slidenerd]

🙏🙏🙏 pg-listener is the definitive answer thanks to vitaly-t's fantastic efforts in 2025 people. I am closing this issue.

[vitaly-t]

I had so much fun with the RxJs extension for this library, that I decided to have a go at an alternative pure-RxJs solution that can work for both pg-promise and its base driver node-postgres: pg-rx-listen. It might be not quite production-ready yet, but it is evolving fast, so if you like this one better, just follow the project, and maybe contribute to it? 😉

[slidenerd]

is the difference between both just that one uses rxjs and the other doesnt ? tbh i am not familiar with rxjs, it is some kinda functional javascript module is it not?

[vitaly-t]

pg-listener has been significantly improved since, for even easier use.

For example, the addition of createIterable allows for the following approach to handling messages:

const result = await ls.listen(['channel_1', 'channel_2']);

for await (const msg of result.createIterable()) {
    console.log(msg);
}