events-backend: initial req/res protocol

Signed-off-by: Patrik Oldsberg <poldsberg@gmail.com>

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);

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:^",

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<BackstageServicePrincipal>;
 
+  #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<TParams>(
+    method: string,
+    schema: z.ZodType<TParams>,
+    handler: (params: TParams) => unknown,
+  ) {
+    this.#requestHandlers.set(method, { schema, handler });
+  }
+
+  async request<TReq extends JsonObject, TRes extends JsonObject>(
+    method: string,
+    params: TReq,
+  ): Promise<TRes> {
+    return new Promise<TRes>((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);
         },
       );

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