diff --git a/api/index.ts b/api/index.ts index 4d12a0c..f8a6823 100644 --- a/api/index.ts +++ b/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(); diff --git a/api/src/api.ts b/api/src/api.ts index 6da1d9a..ae8e163 100644 --- a/api/src/api.ts +++ b/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) { diff --git a/api/src/events.ts b/api/src/events.ts index bdc465e..ba6ab62 100644 --- a/api/src/events.ts +++ b/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); }); diff --git a/api/src/rpc.ts b/api/src/rpc.ts index b3414f2..29da67e 100644 --- a/api/src/rpc.ts +++ b/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 { 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 { + assert(isAdmin(user), 'not authorised'); + events.emit(Events.Sync); + return [RpcMessageType.AcpListingsSync, []]; + } + async function receive(message: any): Promise { 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; diff --git a/client/src/routes/acp.tsx b/client/src/routes/acp.tsx index dbfc975..f22fe62 100644 --- a/client/src/routes/acp.tsx +++ b/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 ( -
- {offers.map(SellerGroup)} +
+ +
+
+ {offers.map(SellerGroup)} +
); } diff --git a/client/src/store.tsx b/client/src/store.tsx index 053f377..b2cb56e 100644 --- a/client/src/store.tsx +++ b/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__() ); diff --git a/client/src/ws.ts b/client/src/ws.ts index da8d744..2afbe62 100644 --- a/client/src/ws.ts +++ b/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, }; }