feat: Add broadcast functionality from triggers

[filipecabaco]

What kind of change does this PR introduce?

Adds a new functionality to broadcast db changes. It adds:

• New function to broadcast from postgres using triggers

Example:

CREATE TABLE test_table (
    id serial PRIMARY KEY,
    name text NOT NULL,
    value integer NOT NULL,
    created_at timestamp DEFAULT CURRENT_TIMESTAMP
);

CREATE OR REPLACE FUNCTION broadcast_changes_for_test_table_trigger ()
    RETURNS TRIGGER
    AS $$
DECLARE
    topic text;
BEGIN
    topic = 'event:' || COALESCE(NEW.id::text, OLD.id::text);
    PERFORM
        realtime.broadcast_changes (topic, TG_OP, TG_OP, TG_TABLE_NAME, TG_TABLE_SCHEMA, NEW, OLD, TG_LEVEL);
    RETURN NULL;
END;
$$
LANGUAGE plpgsql;

CREATE TRIGGER broadcast_changes_for_test_table
    AFTER INSERT OR UPDATE OR DELETE ON test_table
    FOR EACH ROW
    EXECUTE FUNCTION broadcast_changes_for_test_table_trigger ();

-- Insert operation
INSERT INTO test_table (name, value) VALUES ('example_name', 42);

-- Update operation
UPDATE test_table SET value = 100 WHERE name = 'example_name';

-- Delete operation
DELETE FROM test_table WHERE name = 'example_name';

-- Send a message to the topic
SELECT realtime.send (to_jsonb ('{}'::text), 'test', 'test', FALSE);

   topic  | extension |     inserted_at     |     updated_at      |                                                                                                                                       payload                                                                                                                                       | event  | private |                  id                  
---------+-----------+---------------------+---------------------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+--------+---------+--------------------------------------
 event:1 | broadcast | 2024-09-19 15:55:51 | 2024-09-19 15:55:51 | {"table": "test_table", "record": {"id": 1, "name": "example_name", "value": 42, "created_at": "2024-09-19T15:55:50.539754"}, "schema": "public", "operation": "INSERT", "old_record": null}                                                                                        | INSERT | t       | 39d6a1d8-04a4-4099-a2a3-e3c5a9b4605b
 event:1 | broadcast | 2024-09-19 15:55:53 | 2024-09-19 15:55:53 | {"table": "test_table", "record": {"id": 1, "name": "example_name", "value": 100, "created_at": "2024-09-19T15:55:50.539754"}, "schema": "public", "operation": "UPDATE", "old_record": {"id": 1, "name": "example_name", "value": 42, "created_at": "2024-09-19T15:55:50.539754"}} | UPDATE | t       | a34a5f73-13d2-4d11-b4ca-aaf041f8ad7a
 event:1 | broadcast | 2024-09-19 15:55:56 | 2024-09-19 15:55:56 | {"table": "test_table", "record": null, "schema": "public", "operation": "DELETE", "old_record": {"id": 1, "name": "example_name", "value": 100, "created_at": "2024-09-19T15:55:50.539754"}}                                                                                       | DELETE | t       | 6f5a291c-37b4-4051-96f4-98accdc5b201
 test    | broadcast | 2024-09-19 15:55:58 | 2024-09-19 15:55:58 | "{}"                                                                                                                                                                                                                                                                                | test   | f       | 6cc07f9e-4a59-42dd-9aa1-2d1b15e7827d

[vercel]

The latest updates on your projects. Learn more about Vercel for Git ↗︎

1 Skipped Deployment

Name	Status	Preview	Comments	Updated (UTC)
realtime-demo	⬜️ Ignored (Inspect)	Visit Preview		Sep 19, 2024 4:02pm

[chasers]

Should our payload match what we have here in the Webhooks trigger?

[olirice]

It wouldn't hurt to have a check like

  IF TG_LEVEL = 'STATEMENT' THEN
    RAISE EXCEPTION 'realtime.broadcast_changes should be triggered for each row, not for each statement';
  END IF;

to make sure no one tries to "optimize" their trigger for bulk operations

[olirice]

Why is the TG_ARGV[1]::text needed vs relying on TG_TABLE_SCHEMA + TG_TABLE_NAME?

it looks like that would allow users to override the event / qualified table name
what happens if the user overrides it with nonsense instead of a valid target?

[chasers]

Our event names are arbitrary, this is just prescriptive for broadcasting changes from tables.

@filipecabaco we could just not make this an argument for now. One less thing for devs to think about.

[filipecabaco]

the idea is to have this be flexible so if a user wants to have multiple tables to be funnelled into a single topic with a single event they are able to. just giving extra flexibility (and an extra gun to potentially lose a foot to be fair)

I'll bring this into the discussion

[filipecabaco]

will keep the event overridable as it might make it easier for users to migrate to this approach

[olirice]

Our event names are arbitrary, this is just prescriptive for broadcasting changes from tables.

oh nice. In that case this looks like a great solution as long as its documented well