From 5032c181e7387943e7693b3305a9e3cc79dc6cc6 Mon Sep 17 00:00:00 2001 From: Patrik Oldsberg Date: Tue, 21 May 2024 17:57:04 +0200 Subject: [PATCH] events-backend: initial req/res protocol Signed-off-by: Patrik Oldsberg --- plugins/events-backend/dev/index.ts | 9 +- plugins/events-backend/package.json | 6 +- .../src/service/hub/EventHub.ts | 172 +++++++++++++++++- yarn.lock | 12 +- 4 files changed, 189 insertions(+), 10 deletions(-) diff --git a/plugins/events-backend/dev/index.ts b/plugins/events-backend/dev/index.ts index f5eb40ad50..445fa4b4b9 100644 --- a/plugins/events-backend/dev/index.ts +++ b/plugins/events-backend/dev/index.ts @@ -86,7 +86,14 @@ backend.add( }); ws.onopen = () => { console.log('DEBUG: ws.onopen'); - ws.send('derp!'); + ws.send( + JSON.stringify([ + 'req', + 1, + 'subscribe', + { id: 'derp', topics: ['test'] }, + ]), + ); }; ws.onmessage = event => { console.log(`DEBUG: event=`, event.data); diff --git a/plugins/events-backend/package.json b/plugins/events-backend/package.json index 6665ff1e52..dc7c2b0110 100644 --- a/plugins/events-backend/package.json +++ b/plugins/events-backend/package.json @@ -54,12 +54,16 @@ "@backstage/backend-common": "^0.25.0", "@backstage/backend-plugin-api": "workspace:^", "@backstage/config": "workspace:^", + "@backstage/errors": "workspace:^", "@backstage/plugin-events-node": "workspace:^", + "@backstage/types": "workspace:^", "@types/express": "^4.17.6", "express": "^4.17.1", "express-promise-router": "^4.1.0", "winston": "^3.2.1", - "ws": "^8.17.0" + "ws": "^8.17.0", + "zod": "^3.22.4", + "zod-validation-error": "^3.3.0" }, "devDependencies": { "@backstage/backend-defaults": "workspace:^", diff --git a/plugins/events-backend/src/service/hub/EventHub.ts b/plugins/events-backend/src/service/hub/EventHub.ts index cc99a2b2dd..21692ca10f 100644 --- a/plugins/events-backend/src/service/hub/EventHub.ts +++ b/plugins/events-backend/src/service/hub/EventHub.ts @@ -25,6 +25,76 @@ import Router from 'express-promise-router'; import { Socket } from 'net'; import { STATUS_CODES } from 'http'; import { WebSocketServer, type WebSocket, RawData } from 'ws'; +import { z, ZodError } from 'zod'; +import { fromZodError } from 'zod-validation-error'; +import { JsonObject, JsonValue } from '@backstage/types'; +import { serializeError } from '@backstage/errors'; + +/* + +# Protocol + +## Request/Response + +General request/response format used for all communication: + +-> [type: 'req', id: number, method: string, params: JsonObject] +<- [type: 'res', id: number, status: 'resolved' | 'rejected', result: JsonObject] + +## Client -> Server + +### Subscribe + +-> method: 'subscribe', params: { id: string, topics: string[] } +<- result: void + +### Publish + +-> method: 'publish', params: { topic: string, payload: JsonObject } +<- result: void + +## Server -> Client + +### Event + +-> method: 'event', params: { topic: string, payload: JsonObject } +<- result: void + +*/ + +const messageSchema = z.union([ + z.tuple([ + z.literal('req'), + z.number().int().gt(0), + z.string().min(1), + z.any(), + ]), + z.tuple([ + z.literal('res'), + z.number().int().gt(0), + z.enum(['resolved', 'rejected']), + z.any(), + ]), +]); +const subscribeParamsSchema = z.object({ + id: z.string().min(1), + topics: z.array(z.string().min(1)), +}); +const publishParamsSchema = z.object({ + topic: z.string().min(1), + payload: z.any(), +}); +const eventParamsSchema = z.object({ + topic: z.string().min(1), + payload: z.any(), +}); + +function errorToJson(error: Error): JsonObject { + if (error.name === 'ZodError') { + return serializeError(fromZodError(error as ZodError)); + } + return serializeError(error); +} /** * Manages a single WebSocket connection. @@ -51,7 +121,6 @@ class EventClientConnection { ws.addListener('close', conn.#handleClose); ws.addListener('error', conn.#handleError); ws.addListener('message', conn.#handleMessage); - ws.addListener('ping', () => ws.pong()); logger.info( `New event client connection from '${options.socket.remoteAddress}'`, @@ -65,6 +134,19 @@ class EventClientConnection { readonly #logger: LoggerService; readonly #credentials: BackstageCredentials; + #seq = 1; + readonly #pendingRequests = new Map< + number, + { resolve(result: unknown): void; reject(error: unknown): void } + >(); + readonly #requestHandlers = new Map< + string, + { + schema: z.ZodType; + handler: (params: any) => unknown; + } + >(); + constructor( id: string, ws: WebSocket, @@ -91,10 +173,78 @@ class EventClientConnection { this.#logger.error(`WebSocket error`, error); }; - #handleMessage = (data: RawData, isBinary: boolean) => { - console.log(`DEBUG: isBinary=${isBinary} data=${data}`); + #handleMessage = (rawData: RawData, isBinary: boolean) => { + if (isBinary) { + return; + } + try { + const data = Array.isArray(rawData) + ? Buffer.concat(rawData) + : Buffer.from(rawData); + const message = messageSchema.parse(JSON.parse(data.toString('utf8'))); + + if (message[0] === 'req') { + const [, seq, method, params] = message; + const handler = this.#requestHandlers.get(method); + if (!handler) { + throw new Error(`Unknown method '${method}'`); + } + + try { + const parsedParams = handler.schema.parse(params); + + Promise.resolve(handler.handler(parsedParams)).then( + result => { + this.#sendMessage('res', seq, 'resolved', result ?? null); + }, + error => { + this.#sendMessage('res', seq, 'rejected', errorToJson(error)); + }, + ); + } catch (error) { + this.#sendMessage('res', seq, false, errorToJson(error)); + } + } else if (message[0] === 'res') { + const [, seq, success, result] = message; + const pendingRequest = this.#pendingRequests.get(seq); + if (!pendingRequest) { + throw new Error(`Received response for unknown request seq=${seq}`); + } + this.#pendingRequests.delete(seq); + if (success) { + pendingRequest.resolve(result); + } else { + pendingRequest.reject(result); + } + } + } catch (error) { + this.#logger.error('Invalid message received', error); + } }; + addRequestHandler( + method: string, + schema: z.ZodType, + handler: (params: TParams) => unknown, + ) { + this.#requestHandlers.set(method, { schema, handler }); + } + + async request( + method: string, + params: TReq, + ): Promise { + return new Promise((resolve, reject) => { + const seq = this.#seq++; + this.#pendingRequests.set(seq, { resolve, reject }); + this.#sendMessage('req', seq, method, params); + }); + } + + #sendMessage(...message: JsonValue[]) { + this.#ws.send(JSON.stringify(message)); + } + close() { this.#removeListeners(); this.#ws.close(); @@ -162,7 +312,7 @@ export class EventHub { return this.#handler; } - #handleGetConnect: Handler = async (req, _res) => { + #handleGetConnect: Handler = async (req, _res, next) => { try { const credentials = await this.#httpAuth.credentials(req, { allow: ['service'], @@ -179,6 +329,20 @@ export class EventHub { logger: this.#logger, credentials, }); + conn.addRequestHandler( + 'subscribe', + subscribeParamsSchema, + async params => { + console.log(`DEBUG: subscribe req`, params); + }, + ); + conn.addRequestHandler( + 'publish', + publishParamsSchema, + async params => { + console.log(`DEBUG: publish req`, params); + }, + ); this.#connections.set(conn.id, conn); }, ); diff --git a/yarn.lock b/yarn.lock index 18e926a14e..8f93a53290 100644 --- a/yarn.lock +++ b/yarn.lock @@ -6143,14 +6143,18 @@ __metadata: "@backstage/backend-test-utils": "workspace:^" "@backstage/cli": "workspace:^" "@backstage/config": "workspace:^" + "@backstage/errors": "workspace:^" "@backstage/plugin-events-backend-test-utils": "workspace:^" "@backstage/plugin-events-node": "workspace:^" + "@backstage/types": "workspace:^" "@types/express": ^4.17.6 express: ^4.17.1 express-promise-router: ^4.1.0 supertest: ^7.0.0 winston: ^3.2.1 ws: ^8.17.0 + zod: ^3.22.4 + zod-validation-error: ^3.3.0 languageName: unknown linkType: soft @@ -44662,12 +44666,12 @@ __metadata: languageName: node linkType: hard -"zod-validation-error@npm:^3.0.3": - version: 3.1.0 - resolution: "zod-validation-error@npm:3.1.0" +"zod-validation-error@npm:^3.0.3, zod-validation-error@npm:^3.3.0": + version: 3.3.0 + resolution: "zod-validation-error@npm:3.3.0" peerDependencies: zod: ^3.18.0 - checksum: 84df01c91d594701eaf7f5f007be881e47f7adef2e3f3765f7be031cb78033f9be0924273106cb81b586d8020da9885dbb81b3da363f00a51df00f26274f2b23 + checksum: cbf81ecd27df675d72883b69833565af787302e70ad970ae4a5dab84e1cb8739cedf094b35f7f4b78307adaadb7cab0c0a8f7debeb6516e3fee998a3d4e13422 languageName: node linkType: hard