diff --git a/api/index.ts b/api/index.ts index 36f2a22..5094a43 100644 --- a/api/index.ts +++ b/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(); } diff --git a/api/prisma/migrations/20221226113955_notify/migration.sql b/api/prisma/migrations/20221226113955_notify/migration.sql index dd672dc..4d50665 100644 --- a/api/prisma/migrations/20221226113955_notify/migration.sql +++ b/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(); diff --git a/api/src/listen.notify.ts b/api/src/listen.notify.ts index 5bb3ace..79663d0 100644 --- a/api/src/listen.notify.ts +++ b/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 }); diff --git a/api/src/notify.ts b/api/src/notify.ts new file mode 100644 index 0000000..c9d306f --- /dev/null +++ b/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 }; +} \ No newline at end of file diff --git a/api/src/services.ts b/api/src/services.ts index 8f0fb2d..0bf338c 100644 --- a/api/src/services.ts +++ b/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, };