One thing I really like about PostgreSQL is its notifications via pg_notify. This feature is very useful when trying to get real-time notifications for certain changes in a databases.

PostgreSQL notifications

Creating notifications in PostgreSQL is very easy e.g. let's say we have a table for books:

-- User table creation
CREATE TABLE books (
  id SERIAL PRIMARY KEY,
  title TEXT NOT NULL UNIQUE
);

and we want JSON notifications in the channel new_books every time a new book is created in our database e.g:

{
  "id": 1,
  "title": "Animal Farm"
}

The trigger could be implemented as follows:

-- Trigger function creation
CREATE OR REPLACE FUNCTION trigger_new_book()
  RETURNS TRIGGER AS $$
  DECLARE
    payload JSON;
  BEGIN
    payload := json_build_object(
      'id', NEW.id,
      'title', NEW.title
    );

    PERFORM pg_notify('new_books', payload::TEXT);
    RETURN NEW;
  END;
  $$ LANGUAGE plpgsql;

-- Sets the trigger function in 'books' table
CREATE TRIGGER books_notify_new_book
  BEFORE INSERT ON books
  FOR EACH ROW
  EXECUTE PROCEDURE trigger_new_book();

Then, the following query would trigger our JSON message in the channel new_books:

INSERT INTO books (title) VALUES ('Animal Farm');

The Problem

Though subscribing to our database notifications can be done easily with Postgrex library, handling the connections to the database is a bit of a hassle. We need to ensure:

• Connection multiplexing: avoiding over consuming database resources.
• Fault-tolerant connections: supporting re-connections in case of failure or disconnection.
• Re-connection back-off time: avoiding overloading the database on multiple re-connections.

The Solution

Yggdrasil for PostgreSQL is an adapter that supports all the features mentioned above while maintaining Yggdrasil's simple API e.g:

For our example, we could subscribe to the database messages by doing the following:

iex> Yggdrasil.subscribe(name: "new_books", adapter: :postgres, transformer: :json)
iex> flush()
{:Y_CONNECTED, %Yggdrasil.Channel{...}}

Running the following query:

INSERT INTO books (title) VALUES ('1984');

We will get the following message in IEx:

iex> flush()
{:Y_EVENT, %Yggdrasil.Channel{...}, %{"id" => 2, "title" => "1984"}}

Note: Yggdrasil comes with built-in message transformers. We've used
:json transformer for this example in order to get a map from the JSON
data.

Additionally, our subscriber could also be an Yggdrasil process e.g:

defmodule Books.Subscriber do
  use Yggdrasil

  def start_link(options \\ []) do
    channel = [
      name: "new_books",
      adapter: :postgres,
      transformer: :json
    ]

    Yggdrasil.start_link(__MODULE__, [channel], options)
  end

  @impl true
  def handle_event(_channel, %{"id" => id, "title" => title}, _state) do
    ... handle event ...
    {:ok, nil}
  end
end

It's also possbible to use Yggdrasil.publish/2 with PostgreSQL:

iex> message = %{"id" => 3, "title" => "A Brave New World"}
iex> Yggdrasil.publish([name: "new_books", adapter: :postgres, transformer: :json], message)

Conclusion

Yggdrasil for PostgreSQL simplifies subscriptions to PostgreSQL notifications and let's you focus in what really matters: messages.

Cover image by Nasa