Subscription example

graphile.config.ts

@@ -3,6 +3,8 @@ import "graphile-config";
 import { makePgService } from "@dataplan/pg/adaptors/pg";
 import AmberPreset from "postgraphile/presets/amber";
 import { makeV4Preset } from "postgraphile/presets/v4";
+import { context, get, lambda, listen, type Step } from "postgraphile/grafast";
+import { extendSchema } from "postgraphile/utils";
 import { makePgSmartTagsFromFilePlugin } from "postgraphile/utils";
 import { PostGraphileConnectionFilterPreset } from "postgraphile-plugin-connection-filter";
 import { PgAggregatesPreset } from "@graphile/pg-aggregates";
@@ -12,6 +14,7 @@ import PersistedPlugin from "@grafserv/persisted";
 import { PgOmitArchivedPlugin } from "@graphile-contrib/pg-omit-archived";
 import { dirname } from "path";
 import { fileURLToPath } from "url";
+import { jsonParse } from "postgraphile/@dataplan/json";
 
 const __filename = fileURLToPath(import.meta.url);
 const __dirname = dirname(__filename);
@@ -33,7 +36,38 @@ const preset: GraphileConfig.Preset = {
     PgAggregatesPreset,
     // PgSimplifyInflectionPreset
   ],
-  plugins: [PersistedPlugin.default, PgOmitArchivedPlugin, TagsFilePlugin],
+  plugins: [
+    PersistedPlugin.default,
+    PgOmitArchivedPlugin,
+    TagsFilePlugin,
+    extendSchema((build) => {
+      const { users } = build.pgResources;
+      return {
+        typeDefs: /* GraphQL */ `
+          extend type Subscription {
+            userUpdated(id: ID!): User
+          }
+        `,
+        objects: {
+          Subscription: {
+            plans: {
+              userUpdated: {
+                subscribePlan(_$root, { $id }) {
+                  const $topic = lambda($id, (id) => `users:${id}:updated`);
+                  const $pgSubscriber = context().get("pgSubscriber");
+                  return listen($pgSubscriber, $topic, jsonParse);
+                },
+                plan($event: Step) {
+                  const $id = get($event, "id");
+                  return users.get({ id: $id });
+                },
+              },
+            },
+          },
+        },
+      };
+    }),
+  ],
   pgServices: [
     makePgService({
       // Database connection string:

schema.sql

@@ -1 +1,63 @@
 -- Create your database schema here
+
+--------------------------------------------------------------------------------
+
+-- Function from https://postgraphile.org/postgraphile/next/subscriptions#triggering-subscriptions-automatically
+create function tg__graphql_subscription() returns trigger
+    language plpgsql
+    as $_$
+declare
+  v_process_new bool = (TG_OP = 'INSERT' OR TG_OP = 'UPDATE');
+  v_process_old bool = (TG_OP = 'UPDATE' OR TG_OP = 'DELETE');
+  v_event text = TG_ARGV[0];
+  v_topic_template text = TG_ARGV[1];
+  v_attribute text = TG_ARGV[2];
+  v_record record;
+  v_sub text;
+  v_topic text;
+  v_i int = 0;
+  v_last_topic text;
+begin
+  for v_i in 0..1 loop
+    if (v_i = 0) and v_process_new is true then
+      v_record = new;
+    elsif (v_i = 1) and v_process_old is true then
+      v_record = old;
+    else
+      continue;
+    end if;
+     if v_attribute is not null then
+      execute 'select $1.' || quote_ident(v_attribute)
+        using v_record
+        into v_sub;
+    end if;
+    if v_sub is not null then
+      v_topic = replace(v_topic_template, '$1', v_sub);
+    else
+      v_topic = v_topic_template;
+    end if;
+    if v_topic is distinct from v_last_topic then
+      -- This if statement prevents us from triggering the same notification twice
+      v_last_topic = v_topic;
+      perform pg_notify(v_topic, json_build_object(
+        'event', v_event,
+        'subject', v_sub,
+        'id', v_record.id
+      )::text);
+    end if;
+  end loop;
+  return v_record;
+end;
+$_$;
+
+--------------------------------------------------------------------------------
+
+create table users (
+  id int primary key generated always as identity,
+  name text not null,
+  created_at timestamptz not null default now()
+);
+create trigger _500_user_updated
+  after update on users
+  for each row
+  execute function tg__graphql_subscription('update', 'users:$1:updated', 'id');