Add server/message

src/server/message.ts

@@ -0,0 +1,66 @@
+import { upstreamSchema } from "../protocol/up";
+import { ClientID, ClientMap, Socket } from "../types/client-state";
+import { LogContext } from "../util/logger";
+import { sendError } from "../util/socket";
+import { handlePush, ProcessUntilDone } from "./push";
+import { handlePing } from "./ping";
+
+/**
+ * Handles an upstream message coming into the server by dispatching to the
+ * appropriate handler.
+ * @param handlePush handles a push message
+ * @param roomMap currently running rooms
+ * @param roomID destination room
+ * @param clientID client message came from
+ * @param data raw message data
+ * @param ws socket connection to source client
+ * @returns
+ */
+export function handleMessage(
+  lc: LogContext,
+  clientMap: ClientMap,
+  clientID: ClientID,
+  data: string,
+  ws: Socket,
+  processUntilDone: ProcessUntilDone
+) {
+  const msg = getMessage(data);
+  if (msg.error) {
+    lc.info?.("invalid message", msg.error);
+    sendError(ws, msg.error);
+    return;
+  }
+
+  // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+  const message = msg.result!;
+
+  switch (message[0]) {
+    case "ping":
+      handlePing(lc, ws);
+      break;
+    case "push":
+      handlePush(
+        lc,
+        clientMap,
+        clientID,
+        message[1],
+        ws,
+        () => Date.now(),
+        processUntilDone
+      );
+      break;
+    default:
+      throw new Error(`Unknown message type: ${message[0]}`);
+  }
+}
+
+function getMessage(data: string) {
+  let json;
+  try {
+    json = JSON.parse(data);
+    const message = upstreamSchema.parse(json);
+    return { result: message };
+  } catch (e) {
+    return { error: String(e) };
+  }
+}

test/server/message.test.ts

@@ -0,0 +1,79 @@
+import { PushBody } from "../../src/protocol/push";
+import { ClientID, ClientMap, Socket } from "../../src/types/client-state";
+import { Mocket, mutation } from "../util/test-utils";
+import { handleMessage } from "../../src/server/message";
+import { LogContext } from "../../src/util/logger";
+
+test("handleMessage", async () => {
+  type Case = {
+    name: string;
+    data: string;
+    expectedError?: string;
+    expectedPush?: PushBody;
+  };
+
+  const cases: Case[] = [
+    {
+      name: "empty",
+      data: "",
+      expectedError: "SyntaxError: Unexpected end of JSON input",
+    },
+    {
+      name: "invalid push",
+      data: "[]",
+      expectedError: "Should have at least 2 items",
+    },
+    {
+      name: "valid push",
+      data: JSON.stringify([
+        "push",
+        {
+          mutations: [mutation(1), mutation(2)],
+          pushVersion: 1,
+          schemaVersion: "",
+        },
+      ]),
+      expectedPush: {
+        mutations: [mutation(1), mutation(2)],
+        pushVersion: 1,
+        schemaVersion: "",
+        timestamp: 42,
+      },
+    },
+  ];
+
+  for (const c of cases) {
+    const clients: ClientMap = new Map();
+    const clientID = "c1";
+    const s1 = new Mocket();
+    let called = false;
+    // eslint-disable-next-line @typescript-eslint/no-unused-vars
+    const handlePush = (
+      pClients: ClientMap,
+      pClientID: ClientID,
+      pBody: PushBody,
+      pWS: Socket
+    ) => {
+      expect(pClientID).toEqual(clientID);
+      expect(pBody).toEqual(c.expectedPush);
+      expect(pWS).toEqual(s1);
+      called = true;
+    };
+    await handleMessage(
+      new LogContext("info"),
+      clients,
+      clientID,
+      c.data,
+      s1,
+      () => undefined
+    );
+    if (c.expectedError) {
+      expect(s1.log.length).toEqual(1);
+      const [type, message] = s1.log[0];
+      expect(type).toEqual("send");
+      expect(message).toContain(c.expectedError);
+    } else {
+      expect(called);
+    }
+  }
+});