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

const pg = require('pg')

var pgConString = process.env.DATABASE_URL

// Connect to the DB
pg.connect(pgConString, function (err, client) {
  if (err) {
    console.error(err)
  }
  // Handle notifications
  client.on('notification', function (msg) {
    const payload = msg.payload
    console.log(payload)
    // Send payload into a queue etc...
  })
  // Listen for NOTIFY calls
  var query = client.query('LISTEN db_notifications')
})

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_append(payload_items, '"' || replace(column_name, '"', '\"') || '":"' || replace(column_value, '"', '\"') || '"');
  END LOOP;

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

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

Thanks for this! I was looking for a design, where I can listen to changes in the database and act on them from a Python application I'm writing. Only after some research I noticed that the NOTIFY/LISTEN design is not great for this purpose. Because when there are no listeners, all payloads from NOTIFY are instantly lost, and I need to make sure I act on all relevant changes. This scenario is also discussed here: https://stackoverflow.com/questions/23087347/what-happens-with-a-notify-when-no-session-has-issued-listen-in-postgresql

I repurposed most of the code to serve my needs and write trigger data to a new database. Also I'm persisting both the OLD and NEW and an extra string to identify the trigger.

persist_trigger.sql:

-- Create table to store trigger data in
CREATE TABLE invopro.trigger_data(
  id SERIAL PRIMARY KEY,
  timestamp TIMESTAMP,
  operation TEXT,
  schema_name TEXT,
  table_name TEXT,
  trigger TEXT,
  data_old JSONB,
  data_new JSONB,
  processed BOOLEAN DEFAULT FALSE
);

-- Create function to store data from trigger in table
DROP FUNCTION IF EXISTS invopro.persist_trigger() CASCADE;
CREATE FUNCTION invopro.persist_trigger() RETURNS trigger AS $trigger$
DECLARE
  rec RECORD;
  payload TEXT;
  column_name TEXT;
  column_value JSONB;
  payload_items_new JSONB;
  payload_items_old JSONB;
  event_name TEXT;
  fields 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;

  event_name = TG_ARGV[0];
  fields = TG_ARGV[1];

  -- Get old values of required fields
  FOREACH column_name IN ARRAY fields LOOP
    EXECUTE format('SELECT json_build_object(''%I'', $1.%I)', column_name, column_name)
    INTO column_value
    USING OLD;
    payload_items_old := coalesce(payload_items_old,'{}')::jsonb || column_value;
  END LOOP;

  -- Get new values of required fields
  FOREACH column_name IN ARRAY fields LOOP
    EXECUTE format('SELECT json_build_object(''%I'', $1.%I)', column_name, column_name)
    INTO column_value
    USING NEW;
    payload_items_new := coalesce(payload_items_new,'{}')::jsonb || column_value;
  END LOOP;

  INSERT INTO invopro.trigger_data (
    timestamp,
    operation,
    schema_name,
    table_name,
    trigger,
    data_old,
    data_new
  )
  VALUES (
    CURRENT_TIMESTAMP,
    TG_OP,
    TG_TABLE_SCHEMA,
    TG_TABLE_NAME,
    event_name,
    payload_items_old,
    payload_items_new
  );

  RETURN rec;
END;
$trigger$ LANGUAGE plpgsql;

create_triggers.sql:

CREATE TRIGGER move_notify_state_change
  AFTER UPDATE OF state
  ON public.account_move
  FOR EACH ROW
  EXECUTE PROCEDURE invopro.persist_trigger(
    'state_change',
    '{id, move_type, state}'
  );

CREATE TRIGGER move_notify_edi_state_change
  AFTER UPDATE OF edi_state
  ON public.account_move
  FOR EACH ROW
  EXECUTE PROCEDURE invopro.persist_trigger(
    'edi_state_change',
    '{id, move_type, edi_state}'
  );

CREATE TRIGGER move_notify_payment_state_change
  AFTER UPDATE OF payment_state
  ON public.account_move
  FOR EACH ROW
  EXECUTE PROCEDURE invopro.persist_trigger(
    'payment_state_change',
    '{id, move_type, payment_state}'
  );

Hey, first of all, thanks so much for writing this! I noticed that this function doesn't work if the column is itself a json value using PG's built-in JSON/JSONB data types. So I modified your function to support it, hope it is helpful.

CREATE FUNCTION notify_trigger() RETURNS trigger AS $trigger$
DECLARE
  rec RECORD;
  payload TEXT;
  column_name TEXT;
  column_value TEXT;
  payload_items JSONB;
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 := coalesce(payload_items,'{}')::jsonb || json_build_object(column_name,column_value)::jsonb;
  END LOOP;

  -- Build the payload
  payload := json_build_object(
    'timestamp',CURRENT_TIMESTAMP,
    'operation',TG_OP,
    'schema',TG_TABLE_SCHEMA,
    'table',TG_TABLE_NAME,
    'data',payload_items
  );

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

thanks !