Postgres trigger and listener mechanism (Go)

how-to-postgres-listener.md

How to listen to events in Postgres (with a bit of Go)

Postgres has built-in mechanism for notifying clients (listeners) of events such as INSERT, UPDATE, and DELETE on tables.

Note that for this to work, a listener must have a dedicated direct connection to Postgres. Connection poolers like pgbouncer would not work.

First, lets go over some key Postgres components:

• triggers - automatically executes some function when an event on a table occurs
• channels - a communication mechanism identified by a string in postgres
• LISTEN <channel> - registers a client as a listener, and gets notified whenever NOTIFY <channel> is called
• NOTIFY <channel> - notifies the channel

The implementation involves first creating a trigger that calls a function upon some postgres event, such as a new record, or an update to an existing record. While this function can do anything, here we want it to notify a channel by calling pg_notify(). Clients that previously executed LISTEN <channel> will get this notification.

Note that we don't need to explicitly create channels in Postgres. We simply notify and listen to channels, and let Postgres manage the low level operations like queuing and buffering.

Example in Go

Lets say we have a table in Postgres called users. We would like to push notifcations to clients who are interested in updates to this table.

CREATE TABLE users(
	id SERIAL,
	name varchar(256)
)

Create the trigger function. Here we use 'events' as the channel, but this is just a name.

-- `RETURNS trigger` is important because it hooks up the function with the trigger that calls it
-- `NEW` comes from the trigger, and holds the new value for `INSERT/UPDATE` operations
-- `pg_notify` is a postgres helper function that calls `NOTIFY <channel>`
CREATE OR REPLACE FUNCTION check_user_update() RETURNS trigger AS
	$$
	BEGIN
		PERFORM pg_notify('events', row_to_json(NEW)::TEXT);
	END;
	$$ LANGUAGE plpgsql;

-- triggers are the key to this whole thing
CREATE OR REPLACE TRIGGER my_trigger
	AFTER INSERT OR UPDATE ON users
	FOR EACH ROW
	EXECUTE FUNCTION check_user_update();

Write a listener in Go using the pq library.

listener := pq.NewListener(connStr, ...)  // connect to the db
listener.Listen("events")                 // execute LISTEN 'events'

for {
	select {
	case event := <-listener.Notify:
		log.Println("Received data from channel", event.Channel, "| value: ", event.Extra)
	case <-time.After(time.Minute):
		log.Println("Received no events for 1 minute, pinging server...")
		go func() {
			if err := listener.Ping(); err != nil {
				log.Println(err.Error())
			}
		}()
	}
	// the select blocks until either there is a new event, or the timeout kicks in
}

Run this program, then add some rows to the users table, and observe the program respond to your changes!

go run main.go
2023/01/10 21:45:12 Received data from channel events | value: {"id":3,"name":"alice"}
2023/01/10 21:45:27 Received data from channel events | value: {"id":4,"name":"bob"}

Why is this useful?

First, if you are already using postgres, then you can keep using postgres without introducing another tool to your stack.

The main advantage is that clients don't need to continuously poll for changes in the database. Postgres instead pushes changes to the client(s). This reduces IO overhead, allowing you to serve more clients.

Lastly, you can build some interesting systems on this idea.

Pub/Sub

Topics can be represented as tables. Subscribers subscribe by "listening" to changes to these topic tables. Publishers can publish messages by inserting into these tables.

Job Queue

Jobs can also be represented as tables. The workers just listen to changes to this table and claim jobs as they are created.