Building the Example Schema

Connect as app_user:

psql -U app_user -h localhost -d your_database

Tables

-- Primary records table.
-- data holds whatever JSON your application stores.
-- created_by and created_at are set automatically.
CREATE TABLE records (
    id         serial      PRIMARY KEY,
    data       jsonb       NOT NULL,
    created_by text        NOT NULL DEFAULT current_user,
    created_at timestamptz NOT NULL DEFAULT now()
);

-- JSON snapshot table — one row per record, kept current via UPSERT.
-- record_id is unique so there is always exactly one snapshot per record.
CREATE TABLE records_json (
    id         bigserial   PRIMARY KEY,
    record_id  int         NOT NULL UNIQUE REFERENCES records(id),
    data       jsonb       NOT NULL,
    updated_at timestamptz NOT NULL DEFAULT now()
);

-- Audit trail — one row per event, accumulates over time.
CREATE TABLE records_audit (
    id           bigserial   PRIMARY KEY,
    record_id    int         NOT NULL REFERENCES records(id),
    audit_action text        NOT NULL,
    actioned_at  timestamptz NOT NULL DEFAULT now(),
    actioned_by  text        NOT NULL
);

Action functions

These are the functions pg_relay calls when it processes events. They receive the payload from pgrelay.notify() as their $1 argument.

-- Called by the 'json_record' channel.
-- Receives a full records row as JSON; upserts the snapshot into records_json.
-- SECURITY DEFINER: runs as app_user (owner of records_json), so pgrelay only needs
-- EXECUTE on this function — no direct table grants on records_json required.
CREATE FUNCTION insert_json_record(p_payload text)
RETURNS void LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = public, pg_temp
AS $$
DECLARE
    v_data jsonb := p_payload::jsonb;
BEGIN
    INSERT INTO records_json (record_id, data, updated_at)
    VALUES ((v_data->>'id')::int, v_data, now())
    ON CONFLICT (record_id) DO UPDATE
        SET data       = EXCLUDED.data,
            updated_at = now();
END;
$$;

-- Called by the 'audit_action' channel.
-- Receives a JSON object with keys: record_id, action, user.
CREATE FUNCTION insert_audit_record(p_payload text)
RETURNS void LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = public, pg_temp
AS $$
DECLARE
    v_data jsonb := p_payload::jsonb;
BEGIN
    INSERT INTO records_audit (record_id, audit_action, actioned_by)
    VALUES (
        (v_data->>'record_id')::int,
        v_data->>'action',
        v_data->>'user'
    );
END;
$$;

-- Grant pgrelay the right to call both action functions.
-- Both are SECURITY DEFINER and run as app_user, so no table-level grants are needed.
GRANT EXECUTE ON FUNCTION insert_json_record(text)  TO pgrelay;
GRANT EXECUTE ON FUNCTION insert_audit_record(text) TO pgrelay;

SECURITY DEFINER means a function always runs as the role that owns it, regardless of who calls it — this is what lets the tightly-limited pgrelay role call these functions without needing direct access to your tables. See The Security Model in the Technical Guide for the full explanation.

Trigger functions

-- Fires on INSERT and UPDATE to records.
-- Notifies pg_relay to update the JSON snapshot and write an audit entry.
CREATE FUNCTION records_notify()
RETURNS trigger LANGUAGE plpgsql AS $$
BEGIN
    -- Send the full new row as JSON so insert_json_record can UPSERT the snapshot.
    PERFORM pgrelay.notify('json_record', row_to_json(NEW)::text);

    -- Send a packed audit payload (record_id, action, user).
    PERFORM pgrelay.notify('audit_action',
        json_build_object(
            'record_id', NEW.id,
            'action',    CASE TG_OP WHEN 'INSERT' THEN 'NEW_RECORD'
                                    ELSE 'RECORD_UPDATED' END,
            'user',      NEW.created_by
        )::text
    );
    RETURN NEW;
END;
$$;

-- Fires on INSERT and UPDATE to records_json.
-- Writes a secondary audit entry each time the JSON snapshot changes.
-- Looks up the original owner from the records table using record_id.
CREATE FUNCTION records_json_notify()
RETURNS trigger LANGUAGE plpgsql AS $$
DECLARE
    v_user text;
BEGIN
    SELECT created_by INTO v_user FROM records WHERE id = NEW.record_id;

    PERFORM pgrelay.notify('audit_action',
        json_build_object(
            'record_id', NEW.record_id,
            'action',    CASE TG_OP WHEN 'INSERT' THEN 'NEW_JSON_RECORD'
                                    ELSE 'UPDATED_JSON_RECORD' END,
            'user',      v_user
        )::text
    );
    RETURN NEW;
END;
$$;

Triggers

A trigger is a piece of PostgreSQL configuration that automatically runs a function whenever a chosen event (here, an insert or update) happens on a table.

CREATE TRIGGER records_after_insert
AFTER INSERT ON records
FOR EACH ROW EXECUTE FUNCTION records_notify();

CREATE TRIGGER records_after_update
AFTER UPDATE ON records
FOR EACH ROW EXECUTE FUNCTION records_notify();

CREATE TRIGGER records_json_after_insert
AFTER INSERT ON records_json
FOR EACH ROW EXECUTE FUNCTION records_json_notify();

CREATE TRIGGER records_json_after_update
AFTER UPDATE ON records_json
FOR EACH ROW EXECUTE FUNCTION records_json_notify();

Register channels

SELECT pgrelay.register(
    'json_record',
    'SELECT public.insert_json_record($1)',
    'Upserts the JSON snapshot in records_json'
);

SELECT pgrelay.register(
    'audit_action',
    'SELECT public.insert_audit_record($1)',
    'Appends a row to records_audit'
);

Both channels use the default concurrency mode, 'concurrent' — events dispatch as soon as a worker is free, with no ordering restriction. If json_record upserts had to be strictly serialised per record, you would register it with p_concurrency_mode := 'channel_payload' instead (see Controlling concurrency in the User Guide).

Confirm both channels are active:

SELECT channel, active, notes FROM pgrelay.list();

   channel    | active |                    notes
--------------+--------+----------------------------------------------
 audit_action | t      | Appends a row to records_audit
 json_record  | t      | Upserts the JSON snapshot in records_json

---

Everything is now wired up. Continue to Insert, View, Update to see it working.