Merge branch 'master' of ssh://git.mnml.gg:40022/home/ntr/mcrates into sample-offers

api/index.ts

@@ -28,7 +28,7 @@ async function start() {
         events,
     } = createServices(log);
 
-    events.emit(Events.StartupSync);
+    // events.emit(Events.Sync);
     // events.emit(Events.ListingsUpdated);
 
     api.listen();

api/src/api.ts

@@ -63,13 +63,6 @@ export default function createApi(log: Log, pg: Pg, discogs: Discogs, offers: Of
         ctx.status = 200;
     });
 
-    router.get('/user/offers', async function userOffersGet(ctx, next) {
-        ctx.assert(ctx.isAuthenticated(), 401, 'log in required');
-        const user = ctx.state.user;
-        const offers = await pg.userOffers(user);
-        ctx.response.body = offers;
-    });
-
     router.post('/want', async function wantPost(ctx, next) {
         ctx.assert(ctx.isAuthenticated(), 401, 'log in required');
         const { releaseId } = ctx.request.body;
@@ -89,14 +82,6 @@ export default function createApi(log: Log, pg: Pg, discogs: Discogs, offers: Of
         ctx.response.body = wants;
     });
 
-    router.get('/want/all', async function wantGet(ctx, next) {
-        ctx.assert(ctx.isAuthenticated(), 401, 'log in required');
-        const user = ctx.state.user;
-        const wants = await pg.wantList(user);
-
-        ctx.response.body = wants;
-    });
-
     router.post('/want/search', async function wantSearch(ctx, next) {
         const { query } = ctx.request.body;
         const results = await discogs.search(query);
@@ -122,7 +107,7 @@ export default function createApi(log: Log, pg: Pg, discogs: Discogs, offers: Of
             // log.child...
 
             const ws = ctx.websocket;
-            const rpc = createRpc(log, pg, ws, discogs, offers, ctx.state.user as User | null);
+            const rpc = createRpc(log, pg, ws, discogs, offers, events, ctx.state.user as User | null);
 
             if (ctx.state.user) events.emit(Events.Connection, { ws, id: ctx.state.user.id });
             ws.on('message', async function incoming(message) {

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(additions);
+        log.info('want list additions', additions);
 
         return additions;
     }

api/src/events.ts

@@ -10,7 +10,7 @@ import routerAuth from './routers/router.auth';
 
 export enum Events {
     WantsAdded = 'WantsAdded',
-    StartupSync = 'StartupSync',
+    Sync = 'Sync',
     ListingsUpdated = 'ListingsUpdated',
     Connection = 'Connection',
 }
@@ -51,11 +51,12 @@ export default function createEvents(log: Log, pg: Pg, discogs, offers) {
         log.info({ event: Events.WantsAdded, wants: wants.map(w => wantDetails(w)) }, 'starting listings population');
         await updateWantListings(wants);
         log.info({ event: Events.WantsAdded, wants: wants.map(w => wantDetails(w)) }, 'finished populating listings');
+        events.emit(Events.ListingsUpdated);
     });
 
-    events.on(Events.StartupSync, async function startupSync() {
+    events.on(Events.Sync, async function sync() {
         const wants = await pg.wantsToSync();
-        log.info({ event: Events.StartupSync, wants: wants.length });
+        log.info({ event: Events.Sync, wants: wants.length });
         await updateWantListings(wants);
         events.emit(Events.ListingsUpdated);
     });

api/src/offers.ts

@@ -51,8 +51,7 @@ export default function createOffers(log: Log, pg: Pg) {
         // TODO probably need to take one of each condition of record from a store
         // eg. they have a VG+ and a M we should make offers for each
 
-        // log.info(sorted);
-
+        log.trace('generated offers', sorted);
         const inserts = await pg.offersUpdate(sorted);
 
         return inserts;

api/src/rpc.ts

@@ -1,5 +1,3 @@
-import { EventEmitter } from 'node:events';
-import { setTimeout } from 'timers/promises';
 import assert from 'assert/strict';
 
 import type Log from 'bunyan';
@@ -12,6 +10,7 @@ import type { Offers } from './offers';
 
 import { Status } from './offers';
 import { isAdmin } from './utils';
+import { Events } from './events';
 
 function checkMsgType(msgType: RpcMessageType): msgType is RpcMessageType {
     if (RpcMessageType[msgType]) {
@@ -21,7 +20,7 @@ function checkMsgType(msgType: RpcMessageType): msgType is RpcMessageType {
 }
 
 // rpc is created per connection from the api
-export default function createRpc(log: Log, pg: Pg, ws, discogs: Discogs, offers: Offers, user: User | null) {
+export default function createRpc(log: Log, pg: Pg, ws, discogs: Discogs, offers: Offers, events, user: User | null) {
     async function ping(): Promise<PongMessage> {
         return [RpcMessageType.Pong, []];
     }
@@ -71,6 +70,12 @@ export default function createRpc(log: Log, pg: Pg, ws, discogs: Discogs, offers
         return [RpcMessageType.AcpOffers, await offers.acpOffers()];
     }
 
+    async function acpListingsSync(): Promise<AcpListingsSyncMessage> {
+        assert(isAdmin(user), 'not authorised');
+        events.emit(Events.Sync);
+        return [RpcMessageType.AcpListingsSync, []];
+    }
+
     async function receive(message: any): Promise<RpcMessage> {
         const [msgType, params] = message;
 
@@ -101,6 +106,8 @@ export default function createRpc(log: Log, pg: Pg, ws, discogs: Discogs, offers
                 return acpOfferStatusSet(params);
             case RpcMessageType.AcpOffers:
                 return acpOffers();
+            case RpcMessageType.AcpListingsSync:
+                return acpListingsSync();
 
             default:
                 throw new Error(`Message type ${msgType} is known but unhandled`);
@@ -112,18 +119,30 @@ export default function createRpc(log: Log, pg: Pg, ws, discogs: Discogs, offers
         throw new Error(`Invalid parameters for message type ${msgType}: ${JSON.stringify(message)}`);
     }
 
+    async function anonConnection() {
+
+    }
+
+    async function adminConnection() {
+        // don't need it, requested on route
+        // send(await acpOffers());
+    }
+
     async function onConnection() {
+        if (!user) return anonConnection();
+
         const userMsg: UserMessage = [RpcMessageType.User, user];
         send(userMsg);
 
+        const userOffers = await offers.userOffers(user);
+        const offersMsg: OffersMessage = [RpcMessageType.Offers, userOffers];
+        send(offersMsg);
+
         const wants = await pg.userWants(user);
         const wantsMsg: WantsMessage = [RpcMessageType.Wants, wants];
         send(wantsMsg);
 
-        const samples = await offers.samples();
-        const sampleMsg: OfferSamplesMessage = [RpcMessageType.OfferSamples, samples];
-        // console.log(samples);
-        send(sampleMsg);
+        if (isAdmin(user)) return adminConnection();
     }
 
     function send(msg: RpcMessage) {
@@ -163,6 +182,7 @@ export enum RpcMessageType {
 
     AcpOfferStatusSet = 'AcpOfferStatusSet',
     AcpOffers = 'AcpOffers',
+    AcpListingsSync = 'AcpListingsSync',
 
     WantRm = 'WantRm',
 }
@@ -179,6 +199,7 @@ type OfferSamplesMessage = [RpcMessageType.OfferSamples, Offer[]];
 
 type AcpOfferStatusSetMessage = [RpcMessageType.AcpOfferStatusSet, Offer[]];
 type AcpOffersMessage = [RpcMessageType.AcpOffers, Offer[]];
+type AcpListingsSyncMessage = [RpcMessageType.AcpListingsSync, []];
 
 export type RpcMessage =
     PingMessage |
@@ -195,6 +216,7 @@ export type RpcMessage =
 
     AcpOfferStatusSetMessage |
     AcpOffersMessage |
+    AcpListingsSyncMessage |
 
     ErrorMessage;