acp listings sync

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

@@ -101,7 +101,7 @@ export default function createApi(log: Log, pg: Pg, discogs, offers, events) {
         // log.child...
 
         const ws = ctx.websocket;
-        const rpc = createRpc(log, pg, ctx.state.user, ws, discogs, offers);
+        const rpc = createRpc(log, pg, ctx.state.user, ws, discogs, offers, events);
 
         if (ctx.state.user) events.emit(Events.Connection, { ws, id: ctx.state.user.id });
         ws.on('message', async function incoming(message) {

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',
 }
@@ -54,9 +54,9 @@ export default function createEvents(log: Log, pg: Pg, discogs, offers) {
         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/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';
@@ -9,6 +7,7 @@ import type { Want, Offer, User } from '@prisma/client';
 import { Pg } from './pg';
 import { Status } from './offers';
 import { isAdmin } from './utils';
+import { Events } from './events';
 
 function checkMsgType(msgType: RpcMessageType): msgType is RpcMessageType {
     if (RpcMessageType[msgType]) {
@@ -18,7 +17,7 @@ function checkMsgType(msgType: RpcMessageType): msgType is RpcMessageType {
 }
 
 // rpc is created per connection from the api
-export default function createRpc(log: Log, pg: Pg, user: User, ws, discogs, offers) {
+export default function createRpc(log: Log, pg: Pg, user: User, ws, discogs, offers, events) {
     async function ping(): Promise<PongMessage> {
         return [RpcMessageType.Pong, []];
     }
@@ -68,6 +67,12 @@ export default function createRpc(log: Log, pg: Pg, user: User, ws, discogs, off
         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;
 
@@ -98,6 +103,8 @@ export default function createRpc(log: Log, pg: Pg, user: User, ws, discogs, off
                 return acpOfferStatusSet(params);
             case RpcMessageType.AcpOffers:
                 return acpOffers();
+            case RpcMessageType.AcpListingsSync:
+                return acpListingsSync();
 
             default:
                 throw new Error(`Message type ${msgType} is known but unhandled`);
@@ -169,6 +176,7 @@ export enum RpcMessageType {
 
     AcpOfferStatusSet = 'AcpOfferStatusSet',
     AcpOffers = 'AcpOffers',
+    AcpListingsSync = 'AcpListingsSync',
 
     WantRm = 'WantRm',
 }
@@ -184,6 +192,7 @@ type OfferStatusSetMessage = [RpcMessageType.OfferStatusSet, Offer[]];
 
 type AcpOfferStatusSetMessage = [RpcMessageType.AcpOfferStatusSet, Offer[]];
 type AcpOffersMessage = [RpcMessageType.AcpOffers, Offer[]];
+type AcpListingsSyncMessage = [RpcMessageType.AcpListingsSync, []];
 
 export type RpcMessage =
     PingMessage |
@@ -199,6 +208,7 @@ export type RpcMessage =
 
     AcpOfferStatusSetMessage |
     AcpOffersMessage |
+    AcpListingsSyncMessage |
 
     ErrorMessage;
 

client/src/routes/acp.tsx

@@ -70,9 +70,9 @@ function Home() {
     const offers = useSelector((state: State) => state.acpOffers);
     const ws = useSelector((state: State) => state.ws);
 
-    // useEffect(() => {
-    //     ws.sendAcpOffers();
-    // }, [offers, ws]);
+    function sync() {
+        ws.sendAcpListingsSync();
+    }
 
     if (!offers.length) {
         return (
@@ -83,8 +83,12 @@ function Home() {
     }
 
     return (
-        <div class='bundles'>
-            {offers.map(SellerGroup)}
+        <div>
+            <button onClick={sync}> sync listings </button>
+            <hr />
+            <div class='bundles'>
+                {offers.map(SellerGroup)}
+            </div>
         </div>
     );
 }

client/src/store.tsx

@@ -52,6 +52,7 @@ const reducer = combineReducers({
 
 const store = createStore(
   reducer,
+  // @ts-ignore -- devtools window exists on dev
   window.__REDUX_DEVTOOLS_EXTENSION__ && window.__REDUX_DEVTOOLS_EXTENSION__()
 );
 

client/src/ws.ts

@@ -50,6 +50,10 @@ export default function createSocket(dispatch) {
         send(['AcpOfferStatusSet', { id, status }]);
     }
 
+    function sendAcpListingsSync() {
+        send(['AcpListingsSync', {}]);
+    }
+
     let pongTimeout;
     function onPong() {
         pongTimeout = setTimeout(sendPing, 5000);
@@ -154,5 +158,6 @@ export default function createSocket(dispatch) {
 
         sendAcpOffers,
         sendAcpOfferStatusSet,
+        sendAcpListingsSync,
     };
 }