From 2c66c471638c75fdd13900fe8de90637681dda13 Mon Sep 17 00:00:00 2001 From: ntr Date: Sun, 11 Sep 2022 21:16:33 +1000 Subject: [PATCH] rpc --- TODO.md | 7 +-- api/index.ts | 2 +- api/src/api.ts | 20 ++++--- api/src/events.ts | 6 ++ api/src/offers.ts | 20 +++++-- api/src/pg.ts | 6 +- api/src/rpc.ts | 100 ++++++++++++++++++++++++++++++++ client/package-lock.json | 21 +------ client/src/app.tsx | 2 +- client/src/components/offer.tsx | 2 +- client/src/routes/acp.tsx | 21 +++++-- client/src/store.tsx | 5 +- client/src/style/index.less | 6 +- client/src/ws.ts | 8 +-- 14 files changed, 168 insertions(+), 58 deletions(-) create mode 100644 api/src/rpc.ts diff --git a/TODO.md b/TODO.md index 2e6a77a..b935ebe 100644 --- a/TODO.md +++ b/TODO.md @@ -1,9 +1,6 @@ # TODO push offers -redux that bitch +cover off admin routes and commands - -seller - eg min 6 items - shipping is 101aud +do we keep historical offers / wants? diff --git a/api/index.ts b/api/index.ts index 0c70153..4d12a0c 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.StartupSync); // events.emit(Events.ListingsUpdated); api.listen(); diff --git a/api/src/api.ts b/api/src/api.ts index 09cc6b8..1779092 100644 --- a/api/src/api.ts +++ b/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 }); }); }); diff --git a/api/src/events.ts b/api/src/events.ts index b5e9444..7f1650d 100644 --- a/api/src/events.ts +++ b/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; } \ No newline at end of file diff --git a/api/src/offers.ts b/api/src/offers.ts index 5892361..6d97541 100644 --- a/api/src/offers.ts +++ b/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, }; } diff --git a/api/src/pg.ts b/api/src/pg.ts index fac1935..2639ea9 100644 --- a/api/src/pg.ts +++ b/api/src/pg.ts @@ -31,7 +31,7 @@ export interface Pg { bundleCandidates: () => Promise, offersCreate: (offers: Offer[]) => Promise, - offerSetStatus: (id: UUID, status: Status) => Promise, + offerStatusSet: (id: UUID, status: Status) => Promise, offersAdmin: () => Promise, } @@ -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, }; } \ No newline at end of file diff --git a/api/src/rpc.ts b/api/src/rpc.ts new file mode 100644 index 0000000..283a50b --- /dev/null +++ b/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 { + return [RpcMessageType.Pong, []]; + } + + async function offerStatusSet(params: OfferStatusSetParams): Promise { + 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 { + 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, +}; diff --git a/client/package-lock.json b/client/package-lock.json index 5cc69d9..c09d225 100644 --- a/client/package-lock.json +++ b/client/package-lock.json @@ -16,8 +16,7 @@ "preact-router": "^3.2.1", "react-feather": "^2.0.10", "react-redux": "^8.0.2", - "redux": "^4.2.0", - "unistore": "^3.5.2" + "redux": "^4.2.0" }, "devDependencies": { "@babel/plugin-transform-react-jsx-source": "^7.18.6", @@ -19059,19 +19058,6 @@ "node": ">=8" } }, - "node_modules/unistore": { - "version": "3.5.2", - "resolved": "https://registry.npmjs.org/unistore/-/unistore-3.5.2.tgz", - "integrity": "sha512-2Aa4eX0Ua1umyiI3Eai6Li+wXYOHgaDBGOPB3Hvw7PAVuD30TAyh5kS4yNKb2fLDbQgizvPhKQRcYnOdfsm4VQ==", - "peerDependenciesMeta": { - "preact": { - "optional": true - }, - "react": { - "optional": true - } - } - }, "node_modules/universalify": { "version": "2.0.0", "resolved": "https://registry.npmjs.org/universalify/-/universalify-2.0.0.tgz", @@ -35978,11 +35964,6 @@ "crypto-random-string": "^2.0.0" } }, - "unistore": { - "version": "3.5.2", - "resolved": "https://registry.npmjs.org/unistore/-/unistore-3.5.2.tgz", - "integrity": "sha512-2Aa4eX0Ua1umyiI3Eai6Li+wXYOHgaDBGOPB3Hvw7PAVuD30TAyh5kS4yNKb2fLDbQgizvPhKQRcYnOdfsm4VQ==" - }, "universalify": { "version": "2.0.0", "resolved": "https://registry.npmjs.org/universalify/-/universalify-2.0.0.tgz", diff --git a/client/src/app.tsx b/client/src/app.tsx index 7585715..ef0ab8a 100644 --- a/client/src/app.tsx +++ b/client/src/app.tsx @@ -20,6 +20,7 @@ import store, { userState, wsState } from './store' const ws = createWs(store); ws.connect(); +// dodgy, doesn't cause the header to rerender async function init() { const [error, user] = await http.getLogin(); if (error) return route('/login'); @@ -27,7 +28,6 @@ async function init() { store.dispatch(userState.actions.set(user)); store.dispatch(wsState.actions.set(ws)); } - init(); const App = () => { diff --git a/client/src/components/offer.tsx b/client/src/components/offer.tsx index da4c424..2ab7ff0 100644 --- a/client/src/components/offer.tsx +++ b/client/src/components/offer.tsx @@ -7,7 +7,7 @@ import http from '../http'; export default function Offer(offer) { const { want: { data: details } } = offer; - const [accepted, setAccepted] = useState(offer.status !== 'pending'); + const [accepted, setAccepted] = useState(offer.status !== 'Pending'); async function onClick() { const [error] = await http.postOfferAccept(offer.id); diff --git a/client/src/routes/acp.tsx b/client/src/routes/acp.tsx index 26fd4f0..e8fb260 100644 --- a/client/src/routes/acp.tsx +++ b/client/src/routes/acp.tsx @@ -1,21 +1,30 @@ import { h } from 'preact'; import { useState, useEffect } from 'preact/hooks'; -import { connect } from 'unistore/preact' import http from './../http'; import AcpOffer from './../components/acp.offer'; -// hmmm -// bundle... -import { Status } from './../../../api/src/offers' +import { useSelector, useDispatch } from 'react-redux' +import { State } from './../store'; + +enum Status { + Pending = 'Pending', + Accepted = 'Accepted', + Declined = 'Declined', + Ordered = 'Ordered', + Complete = 'Complete', +} function Item({ item }) { + const ws = useSelector((state: State) => state.ws); + console.log(item); + const statuses = Object.entries(Status).map(([label, value]) => ( - + )); function onChange(e) { - console.log(item.id, e.target.value); + ws.sendOfferStatusSet(item.id, e.target.value); } return ( diff --git a/client/src/store.tsx b/client/src/store.tsx index 0e62733..c943ddf 100644 --- a/client/src/store.tsx +++ b/client/src/store.tsx @@ -31,7 +31,10 @@ const reducer = combineReducers({ ws: wsState.reducer, }); -const store = createStore(reducer); +const store = createStore( + reducer, + window.__REDUX_DEVTOOLS_EXTENSION__ && window.__REDUX_DEVTOOLS_EXTENSION__() +); export default store; diff --git a/client/src/style/index.less b/client/src/style/index.less index e8b2e8f..8a2779e 100644 --- a/client/src/style/index.less +++ b/client/src/style/index.less @@ -227,11 +227,11 @@ article { } .list { - display: flex; - flex-flow: row wrap; + display: grid; + grid-template-columns: + 1fr 1fr 1fr 1fr; article { - flex: 1 1 20%; padding: var(--base-spacing); margin: var(--base-spacing-half); } diff --git a/client/src/ws.ts b/client/src/ws.ts index 7afdb5c..3763ccf 100644 --- a/client/src/ws.ts +++ b/client/src/ws.ts @@ -24,13 +24,13 @@ export default function createSocket(store) { // setTimeout(sendPing, 50); } - function sendSourcesAvailable() { - send(['SourcesAvailable', {}]); + function sendOfferStatusSet(id, status) { + send(['OfferStatusSet', { id, status }]); } let pongTimeout; function onPong() { - pongTimeout = setTimeout(sendPing, 1000); + pongTimeout = setTimeout(sendPing, 5000); } // ------------- @@ -108,6 +108,6 @@ export default function createSocket(store) { return { connect, - sendSourcesAvailable, + sendOfferStatusSet, }; }