fix(lazer-sdk): implemented function to convert all non-string WS frames to an array buffer, then leverage an isomorphic buffer implementation

Author: benduran

lazer/sdk/js/package.json

@@ -59,6 +59,7 @@
   "license": "Apache-2.0",
   "dependencies": {
     "@isaacs/ttlcache": "^1.4.1",
+    "buffer": "^6.0.3",
     "isomorphic-ws": "^5.0.0",
     "ts-log": "^2.2.7",
     "ws": "^8.18.0"

lazer/sdk/js/src/buffer/index.ts

@@ -0,0 +1,8 @@
+// need to disable this as this is actually a polyfill for the browser,
+// which we use, conditionally, depending on environment
+// eslint-disable-next-line unicorn/prefer-node-protocol
+import { Buffer as BrowserBuffer } from 'buffer';
+
+import { envIsBrowser, envIsWorker } from '../util.js';
+
+export const IsomorphicBuffer = envIsBrowser() || envIsWorker() ? BrowserBuffer : Buffer;

lazer/sdk/js/src/client.ts

@@ -1,9 +1,11 @@
 import WebSocket from "isomorphic-ws";
 
+import { IsomorphicBuffer } from "./buffer/index.js";
 import type { ParsedPayload, Request, Response } from "./protocol.js";
 import { BINARY_UPDATE_FORMAT_MAGIC_LE, FORMAT_MAGICS_LE } from "./protocol.js";
 import type { WebSocketPoolConfig } from "./socket/websocket-pool.js";
 import { WebSocketPool } from "./socket/websocket-pool.js";
+import { convertWebsocketDataToArrayBuffer } from "./util.js";
 
 export type BinaryResponse = {
   subscriptionId: number;
@@ -15,17 +17,17 @@ export type BinaryResponse = {
 };
 export type JsonOrBinaryResponse =
   | {
-      type: "json";
-      value: Response;
-    }
+    type: "json";
+    value: Response;
+  }
   | { type: "binary"; value: BinaryResponse };
 
 const UINT16_NUM_BYTES = 2;
 const UINT32_NUM_BYTES = 4;
 const UINT64_NUM_BYTES = 8;
 
 export class PythLazerClient {
-  private constructor(private readonly wsp: WebSocketPool) {}
+  private constructor(private readonly wsp: WebSocketPool) { }
 
   /**
    * Creates a new PythLazerClient instance.
@@ -46,53 +48,55 @@ export class PythLazerClient {
    * or a binary response containing EVM, Solana, or parsed payload data.
    */
   addMessageListener(handler: (event: JsonOrBinaryResponse) => void) {
-    this.wsp.addMessageListener((data: WebSocket.Data) => {
-      if (typeof data == "string") {
+    this.wsp.addMessageListener(async (data: WebSocket.Data) => {
+      if (typeof data === "string") {
         handler({
           type: "json",
           value: JSON.parse(data) as Response,
         });
-      } else if (Buffer.isBuffer(data)) {
-        let pos = 0;
-        const magic = data.subarray(pos, pos + UINT32_NUM_BYTES).readUint32LE();
-        pos += UINT32_NUM_BYTES;
-        if (magic != BINARY_UPDATE_FORMAT_MAGIC_LE) {
-          throw new Error("binary update format magic mismatch");
-        }
-        // TODO: some uint64 values may not be representable as Number.
-        const subscriptionId = Number(
-          data.subarray(pos, pos + UINT64_NUM_BYTES).readBigInt64BE(),
-        );
-        pos += UINT64_NUM_BYTES;
+        return;
+      }
+
+      const buff = IsomorphicBuffer.from(await convertWebsocketDataToArrayBuffer(data));
+
+      let pos = 0;
+      const magic = buff.subarray(pos, pos + UINT32_NUM_BYTES).readUint32LE();
+      pos += UINT32_NUM_BYTES;
+      if (magic != BINARY_UPDATE_FORMAT_MAGIC_LE) {
+        throw new Error("binary update format magic mismatch");
+      }
+      // TODO: some uint64 values may not be representable as Number.
+      const subscriptionId = Number(
+        buff.subarray(pos, pos + UINT64_NUM_BYTES).readBigInt64BE(),
+      );
+      pos += UINT64_NUM_BYTES;
 
-        const value: BinaryResponse = { subscriptionId };
-        while (pos < data.length) {
-          const len = data.subarray(pos, pos + UINT16_NUM_BYTES).readUint16BE();
-          pos += UINT16_NUM_BYTES;
-          const magic = data
-            .subarray(pos, pos + UINT32_NUM_BYTES)
-            .readUint32LE();
-          if (magic == FORMAT_MAGICS_LE.EVM) {
-            value.evm = data.subarray(pos, pos + len);
-          } else if (magic == FORMAT_MAGICS_LE.SOLANA) {
-            value.solana = data.subarray(pos, pos + len);
-          } else if (magic == FORMAT_MAGICS_LE.LE_ECDSA) {
-            value.leEcdsa = data.subarray(pos, pos + len);
-          } else if (magic == FORMAT_MAGICS_LE.LE_UNSIGNED) {
-            value.leUnsigned = data.subarray(pos, pos + len);
-          } else if (magic == FORMAT_MAGICS_LE.JSON) {
-            value.parsed = JSON.parse(
-              data.subarray(pos + UINT32_NUM_BYTES, pos + len).toString(),
-            ) as ParsedPayload;
-          } else {
-            throw new Error("unknown magic: " + magic.toString());
-          }
-          pos += len;
+      const value: BinaryResponse = { subscriptionId };
+      while (pos < buff.length) {
+        const len = buff.subarray(pos, pos + UINT16_NUM_BYTES).readUint16BE();
+        pos += UINT16_NUM_BYTES;
+        const magic = buff
+          .subarray(pos, pos + UINT32_NUM_BYTES)
+          .readUint32LE();
+        if (magic == FORMAT_MAGICS_LE.EVM) {
+          value.evm = buff.subarray(pos, pos + len);
+        } else if (magic == FORMAT_MAGICS_LE.SOLANA) {
+          value.solana = buff.subarray(pos, pos + len);
+        } else if (magic == FORMAT_MAGICS_LE.LE_ECDSA) {
+          value.leEcdsa = buff.subarray(pos, pos + len);
+        } else if (magic == FORMAT_MAGICS_LE.LE_UNSIGNED) {
+          value.leUnsigned = buff.subarray(pos, pos + len);
+        } else if (magic == FORMAT_MAGICS_LE.JSON) {
+          value.parsed = JSON.parse(
+            buff.subarray(pos + UINT32_NUM_BYTES, pos + len).toString(),
+          ) as ParsedPayload;
+        } else {
+          throw new Error("unknown magic: " + magic.toString());
         }
-        handler({ type: "binary", value });
-      } else {
-        throw new TypeError("unexpected event data type");
+        pos += len;
       }
+      handler({ type: "binary", value });
+      return;
     });
   }
 

lazer/sdk/js/src/socket/websocket-pool.ts

@@ -9,7 +9,8 @@ import { dummyLogger } from "ts-log";
 import type { Request, Response } from "../protocol.js";
 import type { ResilientWebSocketConfig } from "./resilient-websocket.js";
 import { ResilientWebSocket } from "./resilient-websocket.js";
-import { envIsBrowser, envIsWorker } from "../util.js";
+import { IsomorphicBuffer } from "../buffer/index.js";
+import { convertWebsocketDataToArrayBuffer, envIsBrowser, envIsWorker } from "../util.js";
 
 type AddAuthenticationReturnType = {
   endpoint: string;
@@ -43,17 +44,17 @@ function addAuthentication(
       wsOptions: undefined,
     };
   }
-    // Node.js: Add Authorization header
-    return {
-      endpoint: url,
-      wsOptions: {
-        ...wsOptions,
-        headers: {
-          ...(wsOptions.headers as Record<string, string> | undefined),
-          Authorization: `Bearer ${token}`,
-        },
+  // Node.js: Add Authorization header
+  return {
+    endpoint: url,
+    wsOptions: {
+      ...wsOptions,
+      headers: {
+        ...(wsOptions.headers as Record<string, string> | undefined),
+        Authorization: `Bearer ${token}`,
       },
-    };
+    },
+  };
 }
 
 const DEFAULT_NUM_CONNECTIONS = 4;
@@ -67,11 +68,13 @@ export type WebSocketPoolConfig = {
   onError?: (error: ErrorEvent) => void;
 };
 
+type WebSocketOnMessageCallback = (data: WebSocket.Data) => Promise<void>;
+
 export class WebSocketPool {
   rwsPool: ResilientWebSocket[];
   private cache: TTLCache<string, boolean>;
   private subscriptions: Map<number, Request>; // id -> subscription Request
-  private messageListeners: ((event: WebSocket.Data) => void)[];
+  private messageListeners: WebSocketOnMessageCallback[];
   private allConnectionsDownListeners: (() => void)[];
   private wasAllDown = true;
   private checkConnectionStatesInterval: NodeJS.Timeout;
@@ -148,6 +151,7 @@ export class WebSocketPool {
         rws.onError = config.onError;
       }
       // Handle all client messages ourselves. Dedupe before sending to registered message handlers.
+      // eslint-disable-next-line @typescript-eslint/no-misused-promises
       rws.onMessage = pool.dedupeHandler;
       pool.rwsPool.push(rws);
       rws.startWebSocket();
@@ -184,15 +188,24 @@ export class WebSocketPool {
     }
   }
 
+  /**
+   * internal utility function that will choose the best
+   * parsing mechanism, based on the environment and the
+   * data type received, and return a key, computed from the data,
+   * which can be used as a cache lookup.
+   * this function helps maintain isomorphism for this socket library
+   */
+  private async parseSocketDataToKey(data: WebSocket.Data) {
+    const buff = IsomorphicBuffer.from(await convertWebsocketDataToArrayBuffer(data));
+    return buff.toString('hex');
+  }
+
   /**
    * Handles incoming websocket messages by deduplicating identical messages received across
    * multiple connections before forwarding to registered handlers
    */
-  dedupeHandler = (data: WebSocket.Data): void => {
-    const cacheKey =
-      typeof data === "string"
-        ? data
-        : Buffer.from(data as Buffer).toString("hex");
+  dedupeHandler = async (data: WebSocket.Data): Promise<void> => {
+    const cacheKey = await this.parseSocketDataToKey(data);
 
     if (this.cache.has(cacheKey)) {
       this.logger.debug("Dropping duplicate message");
@@ -205,9 +218,7 @@ export class WebSocketPool {
       this.handleErrorMessages(data);
     }
 
-    for (const handler of this.messageListeners) {
-      handler(data);
-    }
+    await Promise.all(this.messageListeners.map(handler => handler(data)));
   };
 
   sendRequest(request: Request) {
@@ -233,7 +244,7 @@ export class WebSocketPool {
     this.sendRequest(request);
   }
 
-  addMessageListener(handler: (data: WebSocket.Data) => void): void {
+  addMessageListener(handler: WebSocketOnMessageCallback): void {
     this.messageListeners.push(handler);
   }
 

lazer/sdk/js/src/util.ts

@@ -5,12 +5,14 @@
 // is specific to node, but also exists in various browser contexts: 
 // https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/globalThis
 
+import type { Data } from 'isomorphic-ws';
+
 /**
  * Utility function for detecting whether the current JS execution
  * context is that of a background WebWorker or a ServiceWorker (not the main UI thread)
  */
 export function envIsWorker() {
-  const possiblyInAWorker = typeof WorkerGlobalScope === 'undefined' || typeof self === 'undefined';
+  const possiblyInAWorker = typeof WorkerGlobalScope !== 'undefined' && typeof self !== 'undefined';
   return possiblyInAWorker && self instanceof WorkerGlobalScope;
 }
 
@@ -33,3 +35,27 @@ export function envIsBrowser() {
 export function envIsNode() {
   return typeof globalThis !== 'undefined' && typeof globalThis.process === 'object';
 }
+
+/**
+ * given a relatively unknown websocket frame data object,
+ * returns a consistent ArrayBuffer to use in any environment
+ */
+export async function convertWebsocketDataToArrayBuffer(data: Data): Promise<ArrayBuffer> {
+  if (typeof data === 'string') {
+    return new TextEncoder().encode(data).buffer;
+  }
+  if (data instanceof Blob) {
+    // let the uncaught promise exception bubble up if there's an issue
+    return data.arrayBuffer();
+  }
+  if (data instanceof ArrayBuffer) return data;
+  if (Buffer.isBuffer(data)) {
+    const arrBuffer = new ArrayBuffer(data.length);
+    const v = new Uint8Array(arrBuffer);
+    for (const [i, item] of data.entries()) {
+      v[i] = item;
+    }
+    return arrBuffer;
+  }
+  throw new TypeError("unexpected event data type found when convertWebsocketDataToArrayBuffer() to called");
+}