more notify

api/index.ts

@@ -26,10 +26,10 @@ async function start() {
         discogs,
         api,
         events,
-        listenNotify,
+        notify,
     } = createServices(log);
 
-    await listenNotify.connect();
+    await notify.connect();
 
     api.listen();
 }

api/prisma/migrations/20221226113955_notify/migration.sql

@@ -1,7 +1,8 @@
-CREATE OR REPLACE FUNCTION notify_uuid() RETURNS TRIGGER AS $$
+CREATE OR REPLACE FUNCTION notify_users() RETURNS TRIGGER AS $$
     DECLARE
         record RECORD;
         id UUID;
+        payload JSON;
     BEGIN
     IF (TG_OP = 'DELETE') THEN
         id = OLD.id;
@@ -15,17 +16,17 @@ CREATE OR REPLACE FUNCTION notify_uuid() RETURNS TRIGGER AS $$
         'id', id
     );
 
-    PERFORM pg_notify('events', payload::text);
+    PERFORM pg_notify('users.events', payload::text);
 
     RETURN NULL;
     END;
-
 $$ LANGUAGE plpgsql;
 
-CREATE OR REPLACE FUNCTION notify_bigint() RETURNS TRIGGER AS $$
+CREATE OR REPLACE FUNCTION notify_listings() RETURNS TRIGGER AS $$
     DECLARE
         record RECORD;
         id int8;
+        payload JSON;
     BEGIN
     IF (TG_OP = 'DELETE') THEN
         id = OLD.id;
@@ -39,27 +40,72 @@ CREATE OR REPLACE FUNCTION notify_bigint() RETURNS TRIGGER AS $$
         'id', id
     );
 
-    PERFORM pg_notify('events', payload::text);
+    PERFORM pg_notify('listings.events', payload::text);
 
     RETURN NULL;
     END;
-
 $$ LANGUAGE plpgsql;
 
+CREATE OR REPLACE FUNCTION notify_offers() RETURNS TRIGGER AS $$
+    DECLARE
+        record RECORD;
+        id UUID;
+        payload JSON;
+    BEGIN
+    IF (TG_OP = 'DELETE') THEN
+        id = OLD.id;
+    ELSE
+        id = NEW.id;
+    END IF;
 
-CREATE TRIGGER notify_listings_event
-AFTER INSERT OR UPDATE OR DELETE ON listings
-    FOR EACH ROW EXECUTE PROCEDURE notify_bigint();
+    payload = json_build_object(
+        'table', TG_TABLE_NAME,
+        'action', TG_OP,
+        'id', id
+    );
 
-CREATE TRIGGER notify_wants_event
+    PERFORM pg_notify('offers.events', payload::text);
+
+    RETURN NULL;
+    END;
+$$ LANGUAGE plpgsql;
+
+CREATE OR REPLACE FUNCTION notify_wants() RETURNS TRIGGER AS $$
+    DECLARE
+        record RECORD;
+        id UUID;
+        payload JSON;
+    BEGIN
+    IF (TG_OP = 'DELETE') THEN
+        id = OLD.id;
+    ELSE
+        id = NEW.id;
+    END IF;
+
+    payload = json_build_object(
+        'table', TG_TABLE_NAME,
+        'action', TG_OP,
+        'id', id
+    );
+
+    PERFORM pg_notify('wants.events', payload::text);
+
+    RETURN NULL;
+    END;
+$$ LANGUAGE plpgsql;
+
+CREATE OR REPLACE TRIGGER notify_wants_event
 AFTER INSERT OR UPDATE OR DELETE ON wants
-    FOR EACH ROW EXECUTE PROCEDURE notify_bigint();
+    FOR EACH ROW EXECUTE PROCEDURE notify_wants();
 
-CREATE TRIGGER notify_offers_event
+CREATE OR REPLACE TRIGGER notify_listings_event
+AFTER INSERT OR UPDATE OR DELETE ON listings
+    FOR EACH ROW EXECUTE PROCEDURE notify_listings();
+
+CREATE OR REPLACE TRIGGER notify_offers_event
 AFTER INSERT OR UPDATE OR DELETE ON offers
-    FOR EACH ROW EXECUTE PROCEDURE notify_uuid();
+    FOR EACH ROW EXECUTE PROCEDURE notify_offers();
 
-CREATE TRIGGER notify_users_event
+CREATE OR REPLACE TRIGGER notify_users_event
 AFTER INSERT OR UPDATE OR DELETE ON users
-    FOR EACH ROW EXECUTE PROCEDURE notify_uuid();
-
+    FOR EACH ROW EXECUTE PROCEDURE notify_users();

api/src/listen.notify.ts

@@ -3,7 +3,7 @@ import type Log from 'bunyan';
 
 import { Events } from './events';
 
-export default function createListenNotify(log: Log, events) {
+export default function createNotify(log: Log, events) {
     const CHANNEL = 'events';
     const subscriber = createSubscriber({ connectionString: process.env.DATABASE_URL });
 

api/src/notify.ts

@@ -0,0 +1,51 @@
+import createSubscriber from 'pg-listen'
+import type Log from 'bunyan';
+
+import { Events } from './events';
+import { Pg } from './pg';
+
+export default function createNotify(log: Log, pg: Pg, events) {
+    const OFFERS_CHANNEL = 'offers.events';
+    const USERS_CHANNEL = 'users.events';
+    const LISTINGS_CHANNEL = 'listings.events';
+    const WANTS_CHANNEL = 'wants.events';
+
+    const subscriber = createSubscriber({ connectionString: process.env.DATABASE_URL });
+
+    async function handleOffers(payload) {
+        const { action, id } = payload;
+        log.info({ channel: OFFERS_CHANNEL, action, id });
+    }
+
+    async function handleUsers(payload) {
+        const { action, id } = payload;
+        log.info({ channel: USERS_CHANNEL, action, id });
+    }
+
+    async function handleListings(payload) {
+        const { action, id } = payload;
+        log.info({ channel: LISTINGS_CHANNEL, action, id });
+    }
+
+    async function handleWants(payload) {
+        const { action, id } = payload;
+        log.info({ channel: WANTS_CHANNEL, action, id });
+    }
+
+    subscriber.notifications.on(OFFERS_CHANNEL, handleOffers);
+    subscriber.notifications.on(USERS_CHANNEL, handleUsers);
+    subscriber.notifications.on(LISTINGS_CHANNEL, handleListings);
+    subscriber.notifications.on(WANTS_CHANNEL, handleWants);
+
+    async function connect() {
+        await subscriber.connect();
+        await Promise.all([
+            subscriber.listenTo(OFFERS_CHANNEL),
+            subscriber.listenTo(USERS_CHANNEL),
+            subscriber.listenTo(LISTINGS_CHANNEL),
+            subscriber.listenTo(WANTS_CHANNEL),
+        ]);
+    }
+
+    return { connect };
+}

api/src/services.ts

@@ -6,7 +6,7 @@ import createLog from './log';
 import createApi from './api';
 import createEvents from './events';
 import createPg from './pg';
-import createListenNotify from './listen.notify';
+import createNotify from './notify';
 
 export default function createServices(log) {
     const prisma = new PrismaClient();
@@ -14,14 +14,14 @@ export default function createServices(log) {
     const discogs = createDiscogs(log, pg);
     const offers = createOffers(log, pg);
     const events = createEvents(log, pg, discogs, offers);
-    const listenNotify = createListenNotify(log, events);
+    const notify = createNotify(log, pg, events);
     const api = createApi(log, pg, discogs, offers, events);
 
     return {
         api,
         discogs,
         events,
-        listenNotify,
+        notify,
         pg,
         prisma,
     };