notify

api/src/discogs.ts

@@ -48,7 +48,7 @@ export default function createDiscogs(log: Log, pg: Pg) {
         // @ts-ignore
         const additions = results.filter(r => r.status === 'fulfilled').map(r => r.value);
 
-        log.info('want list additions', additions);
+        log.info('want list additions', { additions: additions.length });
 
         return additions;
     }

api/src/events.ts

@@ -15,7 +15,8 @@ export enum Events {
     ListingsUpdated = 'ListingsUpdated',
     Connection = 'Connection',
 
-    OffersUpdated = 'OffersUpdated'
+    OfferUpdated = 'OfferUpdated',
+    WantUpdated = 'WantUpdated',
 }
 
 export default function createEvents(log: Log, pg: Pg, discogs, offers) {
@@ -43,7 +44,7 @@ export default function createEvents(log: Log, pg: Pg, discogs, offers) {
     async function updateWantListings(wants: Want[]) {
         for (const want of wants) {
             const listings = await populateListings(want);
-            log.info({ listings: listings.length, ...wantDetails(want) }, 'listings added');
+            log.info({ listings: listings.length, ...wantDetails(want) }, 'listings found');
             events.emit(Events.ListingsUpdated);
         }
 
@@ -74,11 +75,18 @@ export default function createEvents(log: Log, pg: Pg, discogs, offers) {
         connections.set(id, rpc);
     })
 
-    events.on(Events.OffersUpdated, async function onOffersUpdated(offer: Offer) {
+    events.on(Events.OfferUpdated, async function onOfferUpdated(offer: Offer) {
         const rpc = connections.get(offer.userId);
         if (!rpc) return null;
-        log.debug({ event: Events.OffersUpdated, rpc, offer });
-        return rpc.sendUserOffers();
+        log.debug({ event: Events.OfferUpdated, rpc, offer });
+        return rpc.pushUserOffers();
+    });
+
+    events.on(Events.WantUpdated, async function onWantUpdated(want: Want) {
+        const rpc = connections.get(want.userId);
+        if (!rpc) return null;
+        log.debug({ event: Events.WantUpdated, rpc, want });
+        return rpc.pushUserWants();
     });
 
     return events;

api/src/listen.notify.ts

@@ -1,20 +0,0 @@
-import createSubscriber from 'pg-listen'
-import type Log from 'bunyan';
-
-import { Events } from './events';
-
-export default function createNotify(log: Log, events) {
-    const CHANNEL = 'events';
-    const subscriber = createSubscriber({ connectionString: process.env.DATABASE_URL });
-
-    subscriber.notifications.on(CHANNEL, ({ table, action, record }) => {
-        if (table === 'offers') return events.emit(Events.OffersUpdated, record);
-    })
-
-    async function connect() {
-        await subscriber.connect();
-        return subscriber.listenTo(CHANNEL);
-    }
-
-    return { connect };
-}

api/src/notify.ts

@@ -18,22 +18,26 @@ export default function createNotify(log: Log, pg: Pg, events) {
 
     async function handleOffers(payload) {
         const { action, id } = payload;
-        log.info({ channel: OFFERS_CHANNEL, action, id });
+        log.debug({ channel: OFFERS_CHANNEL, action, id });
+        if (action === 'DELETE') return null;
+
+        const offer = await pg.offerFindId(id);
+        // events.emit(Events.OfferUpdated, offer);
     }
 
     async function handleUsers(payload) {
         const { action, id } = payload;
-        log.info({ channel: USERS_CHANNEL, action, id });
+        log.debug({ channel: USERS_CHANNEL, action, id });
     }
 
     async function handleListings(payload) {
         const { action, id } = payload;
-        log.info({ channel: LISTINGS_CHANNEL, action, id });
+        log.debug({ channel: LISTINGS_CHANNEL, action, id });
     }
 
     async function handleWants(payload) {
         const { action, id } = payload;
-        log.info({ channel: WANTS_CHANNEL, action, id });
+        log.debug({ channel: WANTS_CHANNEL, action, id });
     }
 
     subscriber.notifications.on(OFFERS_CHANNEL, handleOffers);

api/src/rpc.ts

@@ -12,6 +12,76 @@ import { Status } from './offers';
 import { isAdmin } from './utils';
 import { Events } from './events';
 
+export type Rpc = ReturnType<typeof createRpc>;
+
+export enum RpcMessageType {
+    Ping = 'Ping',
+    Pong = 'Pong',
+
+    // points out user is anon
+    WhoDis = 'WhoDis',
+
+    Bundles = 'Bundles',
+    Offers = 'Offers',
+    List = 'List',
+    Wants = 'Wants',
+    User = 'User',
+
+    Error = 'Error',
+
+    OfferSamples = 'OfferSamples',
+    OfferStatusSet = 'OfferStatusSet',
+
+    AcpOfferStatusSet = 'AcpOfferStatusSet',
+    AcpOffers = 'AcpOffers',
+    AcpListingsSync = 'AcpListingsSync',
+
+    WantRm = 'WantRm',
+}
+
+type ErrorMessage = [RpcMessageType.Error, string];
+type PongMessage = [RpcMessageType.Pong, []];
+type PingMessage = [RpcMessageType.Ping, []];
+type WhoDisMessage = [RpcMessageType.WhoDis, []];
+type UserMessage = [RpcMessageType.User, User];
+type WantsMessage = [RpcMessageType.Wants, Want[]];
+type OffersMessage = [RpcMessageType.Offers, Offer[]];
+type OfferStatusSetMessage = [RpcMessageType.OfferStatusSet, Offer[]];
+type OfferSamplesMessage = [RpcMessageType.OfferSamples, Offer[]];
+
+type AcpOfferStatusSetMessage = [RpcMessageType.AcpOfferStatusSet, Offer[]];
+type AcpOffersMessage = [RpcMessageType.AcpOffers, Offer[]];
+type AcpListingsSyncMessage = [RpcMessageType.AcpListingsSync, []];
+
+export type RpcMessage =
+    PingMessage |
+    PongMessage |
+    WhoDisMessage |
+
+    UserMessage |
+
+    WantsMessage |
+
+    OffersMessage |
+    OfferSamplesMessage |
+    OfferStatusSetMessage |
+
+    AcpOfferStatusSetMessage |
+    AcpOffersMessage |
+    AcpListingsSyncMessage |
+
+    ErrorMessage;
+
+
+type OfferStatusSetParams = {
+    id: UUID,
+    status: Status,
+};
+
+type WantRmParams = {
+    id: UUID,
+};
+
 function checkMsgType(msgType: RpcMessageType): msgType is RpcMessageType {
     if (RpcMessageType[msgType]) {
         return true;
@@ -134,11 +204,8 @@ export default function createRpc(log: Log, pg: Pg, ws, discogs: Discogs, offers
         const userMsg: UserMessage = [RpcMessageType.User, user];
         send(userMsg);
 
-        await sendUserOffers();
-
-        const wants = await pg.userWants(user);
-        const wantsMsg: WantsMessage = [RpcMessageType.Wants, wants];
-        send(wantsMsg);
+        await pushUserOffers();
+        await pushUserWants();
 
         if (isAdmin(user)) return adminConnection();
     }
@@ -151,10 +218,12 @@ export default function createRpc(log: Log, pg: Pg, ws, discogs: Discogs, offers
         ))
     }
 
-    async function sendUserOffers() {
-        const userOffers = await offers.userOffers(user);
-        const offersMsg: OffersMessage = [RpcMessageType.Offers, userOffers];
-        send(offersMsg);
+    async function pushUserOffers() {
+        send(await userOffers());
+    }
+
+    async function pushUserWants() {
+        send(await userWants());
     }
 
     return {
@@ -162,76 +231,7 @@ export default function createRpc(log: Log, pg: Pg, ws, discogs: Discogs, offers
         send,
         onConnection,
 
-        sendUserOffers,
+        pushUserOffers,
+        pushUserWants,
     }
 }
-
-export type Rpc = ReturnType<typeof createRpc>;
-
-export enum RpcMessageType {
-    Ping = 'Ping',
-    Pong = 'Pong',
-
-    // points out user is anon
-    WhoDis = 'WhoDis',
-
-    Bundles = 'Bundles',
-    Offers = 'Offers',
-    List = 'List',
-    Wants = 'Wants',
-    User = 'User',
-
-    Error = 'Error',
-
-    OfferSamples = 'OfferSamples',
-    OfferStatusSet = 'OfferStatusSet',
-
-    AcpOfferStatusSet = 'AcpOfferStatusSet',
-    AcpOffers = 'AcpOffers',
-    AcpListingsSync = 'AcpListingsSync',
-
-    WantRm = 'WantRm',
-}
-
-type ErrorMessage = [RpcMessageType.Error, string];
-type PongMessage = [RpcMessageType.Pong, []];
-type PingMessage = [RpcMessageType.Ping, []];
-type WhoDisMessage = [RpcMessageType.WhoDis, []];
-type UserMessage = [RpcMessageType.User, User];
-type WantsMessage = [RpcMessageType.Wants, Want[]];
-type OffersMessage = [RpcMessageType.Offers, Offer[]];
-type OfferStatusSetMessage = [RpcMessageType.OfferStatusSet, Offer[]];
-type OfferSamplesMessage = [RpcMessageType.OfferSamples, Offer[]];
-
-type AcpOfferStatusSetMessage = [RpcMessageType.AcpOfferStatusSet, Offer[]];
-type AcpOffersMessage = [RpcMessageType.AcpOffers, Offer[]];
-type AcpListingsSyncMessage = [RpcMessageType.AcpListingsSync, []];
-
-export type RpcMessage =
-    PingMessage |
-    PongMessage |
-    WhoDisMessage |
-
-    UserMessage |
-
-    WantsMessage |
-
-    OffersMessage |
-    OfferSamplesMessage |
-    OfferStatusSetMessage |
-
-    AcpOfferStatusSetMessage |
-    AcpOffersMessage |
-    AcpListingsSyncMessage |
-
-    ErrorMessage;
-
-
-type OfferStatusSetParams = {
-    id: UUID,
-    status: Status,
-};
-
-type WantRmParams = {
-    id: UUID,
-};