Postgres TRIGGER to call NOTIFY with a JSON payload for Node.js

README.md

Recommendation: use together with https://github.com/andywer/pg-listen

See also: https://gist.github.com/colophonemes/9701b906c5be572a40a84b08f4d2fa4e

sqlTriggerFunctions.js

const sqlCreateTrigger = (triggerName, channelName, schemaName, tableName, functionName, fields) => `
  CREATE TRIGGER ${triggerName}
  AFTER INSERT OR UPDATE OR DELETE
  ON ${schemaName}.${tableName}
  FOR EACH ROW EXECUTE PROCEDURE ${schemaName}.${functionName}('${channelName}', '{${fields.map(f => `"${f}"`).join(',')}}');`;

const sqlCreateReplaceTriggerFunction = (schemaName, functionName) => `
  DROP FUNCTION IF EXISTS ${schemaName}.${functionName}() CASCADE;
  CREATE FUNCTION ${schemaName}.${functionName}() RETURNS trigger AS $$
  DECLARE
    notification json;
    new_or_changed_record_as_jsonb jsonb;
    channelName text;
    fields text[];
    field text;
    extracted_fields text[];
    extracted_values text[];
  BEGIN
    -- TG_ARGV[0]: 1st trigger argument must be channel name
    channelName = TG_ARGV[0];

    -- TG_ARGV[1]: 2nd trigger argument must be list of fields to include in JSON
    fields = TG_ARGV[1];

    -- create temp. jsonb version of changed or created record 1
    IF (TG_OP = 'DELETE') THEN
      new_or_changed_record_as_jsonb = to_jsonb(OLD);
    ELSE
      new_or_changed_record_as_jsonb = to_jsonb(NEW);
    END IF;

    -- build json dynamically for a given list of fields
    FOREACH field IN ARRAY fields LOOP
      extracted_fields = array_append(extracted_fields, field);
      extracted_values = array_append(extracted_values, new_or_changed_record_as_jsonb->>field);
    END LOOP;

    notification = json_build_object(
      'type', TG_OP,
      'dateTime', clock_timestamp(),
      'table', TG_TABLE_NAME::text,
      'schema', TG_TABLE_SCHEMA::text,
      'data', json_object(extracted_fields, extracted_values)
    );

    PERFORM pg_notify(channelName, notification::text);
    RETURN NULL;
  END;
  $$ LANGUAGE plpgsql;
  `;

module.exports = {
  sqlCreateTrigger,
  sqlCreateReplaceTriggerFunction
}