rpc

api/src/api.ts

@@ -16,6 +16,7 @@ import type { Listing, User, Want, Prisma } from '@prisma/client';
 import { Pg } from './pg';
 import routerAuth from './routers/router.auth';
 import { Events } from './events';
+import createRpc, { RpcMessageType } from './rpc';
 
 export default function createApi(log: Log, pg: Pg, discogs, offers, events) {
     const PORT = 41337;
@@ -153,22 +154,27 @@ export default function createApi(log: Log, pg: Pg, discogs, offers, events) {
     app.ws.use(passport.session());
     app.ws.use(async (ctx, next) => {
         log.info('new ws connection', { user: ctx.state.user });
-        ctx.websocket.on('message', async function incoming(message) {
-            log.info(message.toString());
+
+        // log.child...
+
+        const ws = ctx.websocket;
+        const rpc = createRpc(log, pg, ctx.state.user, ws, discogs, offers);
+
+        if (ctx.state.user) events.emit(Events.Connection, { ws, id: ctx.state.user.id });
+        ws.on('message', async function incoming(message) {
             try {
                 const parsed = JSON.parse(message.toString());
-                log.info(parsed);
-                // const response = await rpc.receive(parsed);
-                // rpc.send(ws, response);
+                const response = await rpc.receive(parsed);
+                rpc.send(response);
             } catch (e) {
                 log.warn(e);
                 log.warn(message.toString());
-                // rpc.send(ws, [RpcMessageType.Error, e.message || e.msg])
+                rpc.send([RpcMessageType.Error, e.message || e.msg]);
             }
             return true;
         });
 
-        ctx.websocket.on('close', function onClose() {
+        ws.on('close', function onClose() {
             log.info('ws disconnected', { user: ctx.state.user });
         });
     });

api/src/events.ts

@@ -12,10 +12,12 @@ export enum Events {
     WantAdded = 'WantAdded',
     StartupSync = 'StartupSync',
     ListingsUpdated = 'ListingsUpdated',
+    Connection = 'Connection',
 }
 
 export default function createEvents(log: Log, pg: Pg, discogs, offers) {
     const events = new EventEmitter({ captureRejections: true });
+    const connections = new Map();
 
     async function populateListings(want: Want) {
         try {
@@ -82,5 +84,9 @@ export default function createEvents(log: Log, pg: Pg, discogs, offers) {
         log.info('offers up to date');
     });
 
+    events.on(Events.Connection, async function onConnection({ ws, id }) {
+        connections.set(id, ws);
+    })
+
     return events;
 }

api/src/offers.ts

@@ -1,5 +1,6 @@
 import type Log from 'bunyan';
 import type { PrismaClient, Want, User, Listing, Offer, } from '@prisma/client';
+import assert from 'assert/strict';
 import groupBy from 'lodash/groupBy';
 import sortBy from 'lodash/sortBy';
 import { Context } from 'koa';
@@ -8,11 +9,11 @@ import { logAxiosError } from './utils';
 import { Pg } from './pg';
 
 export enum Status {
-    Pending      = 'pending',
-    Accepted     = 'accepted',
-    Declined     = 'declined',
-    Ordered      = 'ordered',
-    Complete     = 'complete',
+    Pending      = 'Pending',
+    Accepted     = 'Accepted',
+    Declined     = 'Declined',
+    Ordered      = 'Ordered',
+    Complete     = 'Complete',
 }
 
 export default function createOffers(log: Log, pg: Pg) {
@@ -60,7 +61,13 @@ export default function createOffers(log: Log, pg: Pg) {
 
         // check the user actually has this offer
         const offer = await pg.userOfferFind(ctx.state.user, id);
-        return pg.offerSetStatus(offer.id, Status.Accepted);
+        return pg.offerStatusSet(offer.id, Status.Accepted);
+    }
+
+    async function statusSet(user: User, id: UUID, status: Status) {
+        // check the user actually has this offer
+        const offer = await pg.userOfferFind(user, id);
+        return pg.offerStatusSet(offer.id, status);
     }
 
     async function bundles() {
@@ -108,5 +115,6 @@ export default function createOffers(log: Log, pg: Pg) {
         generate,
         accept,
         bundles,
+        statusSet,
     };
 }

api/src/pg.ts

@@ -31,7 +31,7 @@ export interface Pg {
     bundleCandidates:   () => Promise<Listing[][]>,
 
     offersCreate:       (offers: Offer[]) => Promise<Offer[]>,
-    offerSetStatus:     (id: UUID, status: Status) => Promise<Offer>,
+    offerStatusSet:     (id: UUID, status: Status) => Promise<Offer>,
     offersAdmin:        () => Promise<Offer[]>,
 }
 
@@ -258,7 +258,7 @@ export default function createPg(log: Log, prisma: PrismaClient): Pg {
         );
     }
 
-    async function offerSetStatus(id: UUID, status: Status) {
+    async function offerStatusSet(id: UUID, status: Status) {
         return prisma.offer.update({
             where: { id: id },
             data: { status },
@@ -294,7 +294,7 @@ export default function createPg(log: Log, prisma: PrismaClient): Pg {
         bundleCandidates,
 
         offersCreate,
-        offerSetStatus,
+        offerStatusSet,
         offersAdmin,
     };
 }

api/src/rpc.ts

@@ -0,0 +1,100 @@
+import { EventEmitter } from 'node:events';
+import { setTimeout } from 'timers/promises';
+import assert from 'assert/strict';
+
+import type Log from 'bunyan';
+
+import type { Want, Offer, User } from '@prisma/client';
+
+import { Pg } from './pg';
+import { Status } from './offers';
+
+function checkMsgType(msgType: RpcMessageType): msgType is RpcMessageType {
+    if (RpcMessageType[msgType]) {
+        return true;
+    }
+    return false;
+}
+
+// rpc is created per connection from the api
+export default function createRpc(log: Log, pg: Pg, user: User, ws, discogs, offers) {
+    async function ping(): Promise<PongMessage> {
+        return [RpcMessageType.Pong, []];
+    }
+
+    async function offerStatusSet(params: OfferStatusSetParams): Promise<OffersMessage> {
+        log.info(params);
+        const { id, status } = params;
+        assert(id, 'no id supplied');
+        assert(Status[status], 'unknown status');
+
+        const offer = await offers.statusSet(user, id, status);
+        return [RpcMessageType.Offers, []]
+    }
+
+    async function receive(message: any): Promise<RpcMessage> {
+        const [msgType, params] = message;
+
+        if (msgType !== RpcMessageType.Ping) log.info({ msgType, params }, 'rpc message');
+
+        // // this sucks but can't figure out how to convince typescript to agree yet
+        if (checkMsgType(msgType)) {
+            switch (msgType) {
+            case RpcMessageType.Ping:
+                return ping();
+            case RpcMessageType.OfferStatusSet:
+                return offerStatusSet(params);
+            default:
+                throw new Error(`Message type ${msgType} is known but unhandled`);
+            }
+        } else {
+            throw new Error(`Unknown message type ${msgType}`);
+        }
+
+        throw new Error(`Invalid parameters for message type ${msgType}: ${JSON.stringify(message)}`);
+    }
+
+    function send(msg: RpcMessage) {
+        ws.send(JSON.stringify(msg, (key, value) =>
+            typeof value === 'bigint'
+                ? value.toString()
+                : value // return everything else unchanged
+        ))
+    }
+
+    return {
+        receive,
+        send,
+    }
+}
+
+export enum RpcMessageType {
+    Ping = 'Ping',
+    Pong = 'Pong',
+
+    Bundles = 'Bundles',
+    Offers = 'Offers',
+    List = 'List',
+    Error = 'Error',
+
+    OfferStatusSet = 'OfferStatusSet',
+}
+
+type ErrorMessage = [RpcMessageType.Error, string];
+type PongMessage = [RpcMessageType.Pong, []];
+type PingMessage = [RpcMessageType.Ping, []];
+type OffersMessage = [RpcMessageType.Offers, Offer[]];
+type OfferStatusSetMessage = [RpcMessageType.OfferStatusSet, Offer[]];
+
+export type RpcMessage =
+    PingMessage |
+    PongMessage |
+    OffersMessage |
+    OfferStatusSetMessage |
+    ErrorMessage;
+
+
+type OfferStatusSetParams = {
+    id: UUID,
+    status: Status,
+};