Postgres TRIGGER to call NOTIFY with a JSON payload

create_triggers

CREATE TRIGGER person_notify AFTER INSERT OR UPDATE OR DELETE ON income
FOR EACH ROW EXECUTE PROCEDURE notify_trigger(
  'id',
  'email',
  'username'
);

CREATE TRIGGER income_notify AFTER INSERT OR UPDATE OR DELETE ON income
FOR EACH ROW EXECUTE PROCEDURE notify_trigger(
  'id',
  'person_id',
  'amount',
  'currency_code',
  'start_date',
  'end_date',
  'data_source'
);

listener.js

import createSubscriber from "pg-listen"

const connectionString = process.env.DATABASE_URL,
      subscriber = createSubscriber({ connectionString }),
      notifyChannel = 'db_notifications'

process.on("exit", () => {
    subscriber.close()
})

subscriber.events.on("error", (error) => {
    console.error("Fatal database connection error:", error)
    process.exit(1)
})

subscriber.notifications.on(notifyChannel, (payload) => {
    console.log(payload)
})

subscriber.connect().then(() => {
    subscriber.listenTo(notifyChannel)
})

notify_trigger.sql

-- Trigger notification for messaging to PG Notify
CREATE FUNCTION notify_trigger() RETURNS trigger AS $trigger$
DECLARE
  rec RECORD;
  payload TEXT;
  column_name TEXT;
  column_value TEXT;
  payload_items TEXT[];
BEGIN
  -- Set record row depending on operation
  CASE TG_OP
  WHEN 'INSERT', 'UPDATE' THEN
     rec := NEW;
  WHEN 'DELETE' THEN
     rec := OLD;
  ELSE
     RAISE EXCEPTION 'Unknown TG_OP: "%". Should not occur!', TG_OP;
  END CASE;
  
  -- Get required fields
  FOREACH column_name IN ARRAY TG_ARGV LOOP
    EXECUTE format('SELECT $1.%I::TEXT', column_name)
    INTO column_value
    USING rec;
    payload_items := array_cat(payload_items, array[column_name, column_value]);
  END LOOP;

  -- Build the payload
  payload := ''
              || '{'
              || '"timestamp":"' || CURRENT_TIMESTAMP                    || '",'
              || '"operation":"' || TG_OP                                || '",'
              || '"schema":"'    || TG_TABLE_SCHEMA                      || '",'
              || '"table":"'     || TG_TABLE_NAME                        || '",'
              || '"data":'      || to_json(json_object(payload_items))
              || '}';

  -- Notify the channel
  PERFORM pg_notify('db_notifications', payload);
  
  RETURN rec;
END;
$trigger$ LANGUAGE plpgsql;

Don't overlook the FOR EACH ROW part of FOR EACH ROW EXECUTE PROCEDURE, or you'll forever wonder how come neither of OLD, NEW and rec contain any data...