feat(): fleshout cb intx websocket handling & auth

src/WebsocketClient.ts

@@ -9,10 +9,13 @@ import {
   isCBAdvancedTradeWSEvent,
   isCBExchangeWSEvent,
   isCBExchangeWSRequestOperation,
+  isCBINTXWSRequestOperation,
 } from './lib/websocket/typeGuards.js';
 import {
   getCBExchangeWSSign,
+  getCBInternationalWSSign,
   getMergedCBExchangeWSRequestOperations,
+  getMergedCBINTXRequestOperations,
   MessageEventLike,
   WS_KEY_MAP,
   WS_URL_MAP,
@@ -25,6 +28,8 @@ import {
   WsAdvTradeRequestOperation,
   WsExchangeAuthenticatedRequestOperation,
   WsExchangeRequestOperation,
+  WsInternationalAuthenticatedRequestOperation,
+  WsInternationalRequestOperation,
   WsOperation,
 } from './types/websockets/requests.js';
 import {
@@ -455,8 +460,33 @@ export class WebsocketClient extends BaseWebsocketClient<WsKey> {
 
           return wsRequestEvent;
         }
+        case WS_KEY_MAP.internationalMarketData: {
+          if (!['subscribe', 'unsubscribe'].includes(operation)) {
+            throw new Error(
+              `Unhandled request operation type for CB International WS: "${operation}"`,
+            );
+          }
+
+          const wsRequestEvent: WsInternationalRequestOperation<WsTopic> = {
+            type: operation === 'subscribe' ? 'SUBSCRIBE' : 'UNSUBSCRIBE',
+            channels: [
+              topicRequest.payload
+                ? {
+                    name: topicRequest.topic,
+                    ...topicRequest.payload,
+                  }
+                : topicRequest.topic,
+            ],
+          };
+
+          return wsRequestEvent;
+        }
+        // case WS_KEY_MAP.primeMarketData: {
+        // }
         default: {
-          throw new Error(`Not implemented for "${wsKey}" yet`);
+          throw new Error(
+            `Not implemented for "${wsKey}" yet - if you need Prime or INTX, please get in touch.`,
+          );
         }
       }
     });
@@ -598,6 +628,105 @@ export class WebsocketClient extends BaseWebsocketClient<WsKey> {
 
         return finalOperations;
       }
+      case WS_KEY_MAP.internationalMarketData: {
+        if (
+          !operationEvents.every((evt) =>
+            isCBINTXWSRequestOperation(evt, wsKey),
+          )
+        ) {
+          // Don't expect this to ever happen, but just to please typescript...
+          throw new Error(
+            `Unexpected request schema for exchange WS request builder`,
+          );
+        }
+
+        const mergedOperationEvents =
+          getMergedCBINTXRequestOperations(operationEvents);
+
+        // We're under the max topics per request limit.
+        // Send operation requests as one merged request
+        if (
+          !maxTopicsPerEvent ||
+          mergedOperationEvents.channels.length <= maxTopicsPerEvent
+        ) {
+          if (!isPrivateChannel) {
+            return [JSON.stringify(mergedOperationEvents)];
+          }
+
+          if (!apiKey || !apiSecret || !apiPassphrase) {
+            throw new Error(
+              `One or more of apiKey, apiSecret and/or apiPassphrase are missing. These must be provided to use private channels.`,
+            );
+          }
+
+          const { sign, timestampInSeconds } =
+            await getCBExchangeWSSign(apiSecret);
+
+          const mergedOperationEventsWithSign: WsInternationalAuthenticatedRequestOperation<WsTopic> =
+            {
+              ...mergedOperationEvents,
+              time: timestampInSeconds,
+              key: apiKey,
+              passphrase: apiPassphrase,
+              signature: sign,
+            };
+
+          throw new Error(
+            'CB INTX is not fully implemented yet - awaiting test environment... if you need this, please get in touch.',
+          );
+          return [JSON.stringify(mergedOperationEventsWithSign)];
+        }
+
+        // We're over the max topics per request limit. Break into batches.
+        const signedOperations: string[] = [];
+        for (
+          let i = 0;
+          i < mergedOperationEvents.channels.length;
+          i += maxTopicsPerEvent
+        ) {
+          const batchChannels = mergedOperationEvents.channels.slice(
+            i,
+            i + maxTopicsPerEvent,
+          );
+
+          const wsRequestEvent: WsInternationalRequestOperation<WsTopic> = {
+            type: mergedOperationEvents.type,
+            channels: [...batchChannels],
+          };
+
+          if (!apiKey || !apiSecret || !apiPassphrase) {
+            throw new Error(
+              `One or more of apiKey, apiSecret and/or apiPassphrase are missing. These must be provided to use private channels.`,
+            );
+          }
+
+          const { sign, timestampInSeconds } = await getCBInternationalWSSign(
+            apiKey,
+            apiSecret,
+            apiPassphrase,
+          );
+
+          const wsRequestEventWithSign: WsInternationalAuthenticatedRequestOperation<WsTopic> =
+            {
+              ...wsRequestEvent,
+              time: timestampInSeconds,
+              key: apiKey,
+              passphrase: apiPassphrase,
+              signature: sign,
+            };
+
+          signedOperations.push(JSON.stringify(wsRequestEventWithSign));
+        }
+        throw new Error(
+          'CB INTX is not fully implemented yet - awaiting test environment... if you need this, please get in touch.',
+        );
+        return signedOperations;
+      }
+      case WS_KEY_MAP.primeMarketData: {
+        throw new Error(
+          'CB Prime is not fully implemented yet - awaiting test environment... if you need this, please get in touch.',
+        );
+      }
       default: {
         throw new Error(`Not implemented for "${wsKey}" yet`);
       }

src/lib/websocket/typeGuards.ts

@@ -4,7 +4,10 @@ import {
   CBAdvancedTradeEvent,
   CBExchangeBaseEvent,
 } from '../../types/websockets/events.js';
-import { WsExchangeRequestOperation } from '../../types/websockets/requests.js';
+import {
+  WsExchangeRequestOperation,
+  WsInternationalRequestOperation,
+} from '../../types/websockets/requests.js';
 import { WS_KEY_MAP, WsKey } from './websocket-util.js';
 
 function isDefinedObject(value: unknown): value is object {
@@ -64,11 +67,11 @@ export function isCBExchangeWSRequestOperation<
     return false;
   }
 
-  const looseTypedEvt = evt as WsExchangeRequestOperation<TWSTopic>;
+  const looselyTypedEvent = evt as WsExchangeRequestOperation<TWSTopic>;
 
   if (
-    typeof looseTypedEvt.type !== 'string' ||
-    !Array.isArray(looseTypedEvt.channels)
+    typeof looselyTypedEvent.type !== 'string' ||
+    !Array.isArray(looselyTypedEvent.channels)
   ) {
     return false;
   }
@@ -78,3 +81,27 @@ export function isCBExchangeWSRequestOperation<
     wsKey === WS_KEY_MAP.exchangeMarketData
   );
 }
+
+/**
+ * Silly type guard for the structure of events being sent to the server
+ * (e.g. when subscribing to a topic)
+ */
+export function isCBINTXWSRequestOperation<TWSTopic extends string = string>(
+  evt: unknown,
+  wsKey: WsKey,
+): evt is WsInternationalRequestOperation<TWSTopic> {
+  if (!isDefinedObject(evt)) {
+    return false;
+  }
+
+  const looselyTypedEvent = evt as WsInternationalRequestOperation<TWSTopic>;
+
+  if (
+    typeof looselyTypedEvent.type !== 'string' ||
+    !Array.isArray(looselyTypedEvent.channels)
+  ) {
+    return false;
+  }
+
+  return wsKey === WS_KEY_MAP.internationalMarketData;
+}

src/lib/websocket/websocket-util.ts

@@ -1,6 +1,9 @@
 import WebSocket from 'isomorphic-ws';
 
-import { WsExchangeRequestOperation } from '../../types/websockets/requests.js';
+import {
+  WsExchangeRequestOperation,
+  WsInternationalRequestOperation,
+} from '../../types/websockets/requests.js';
 import { signMessage } from '../webCryptoAPI.js';
 
 /**
@@ -134,6 +137,10 @@ export const WS_URL_MAP: Record<WsKey, NetworkMap<'livenet' | 'testnet'>> = {
   },
 } as const;
 
+/**
+ * Merge one or more WS Request operations (e.g. subscribe request) for
+ * CB Exchange into one, allowing them to be sent as one request
+ */
 export function getMergedCBExchangeWSRequestOperations<
   TWSTopic extends string = string,
 >(operations: WsExchangeRequestOperation<TWSTopic>[]) {
@@ -164,7 +171,44 @@ export function getMergedCBExchangeWSRequestOperations<
 
   return mergedOperationEvents;
 }
+/**
+ * Merge one or more WS Request operations (e.g. subscribe request) for
+ * CB Exchange into one, allowing them to be sent as one request
+ */
+export function getMergedCBINTXRequestOperations<
+  TWSTopic extends string = string,
+>(operations: WsInternationalRequestOperation<TWSTopic>[]) {
+  // The CB Exchange WS supports sending multiple topics in one request.
+  // Merge all requests into one
+  const mergedOperationEvents = operations.reduce(
+    (
+      acc: WsInternationalRequestOperation<TWSTopic>,
+      evt: WsInternationalRequestOperation<TWSTopic>,
+    ) => {
+      if (!acc) {
+        const wsRequestEvent: WsInternationalRequestOperation<TWSTopic> = {
+          type: evt.type,
+          channels: [...evt.channels],
+        };
+
+        return wsRequestEvent;
+      }
+
+      const wsRequestEvent: WsInternationalRequestOperation<TWSTopic> = {
+        type: evt.type,
+        channels: [...acc.channels, ...evt.channels],
+      };
+
+      return wsRequestEvent;
+    },
+  );
+
+  return mergedOperationEvents;
+}
 
+/**
+ * Return sign used to authenticate CB Exchange WS requests
+ */
 export async function getCBExchangeWSSign(apiSecret: string): Promise<{
   timestampInSeconds: string;
   sign: string;
@@ -186,3 +230,30 @@ export async function getCBExchangeWSSign(apiSecret: string): Promise<{
 
   return { sign, timestampInSeconds };
 }
+
+/**
+ * Return sign used to authenticate CB INTX WS requests
+ */
+export async function getCBInternationalWSSign(
+  apiKey: string,
+  apiSecret: string,
+  apiPassphrase: string,
+): Promise<{
+  timestampInSeconds: string;
+  sign: string;
+}> {
+  const timestampInMs = Date.now();
+  const timestampInSeconds = (timestampInMs / 1000).toFixed(0);
+
+  const signInput = `${timestampInSeconds}, ${apiKey}, CBINTLMD, ${apiPassphrase}`;
+
+  const sign = await signMessage(
+    signInput,
+    apiSecret,
+    'base64',
+    'SHA-256',
+    'base64:web',
+  );
+
+  return { sign, timestampInSeconds };
+}

src/types/websockets/requests.ts

@@ -19,13 +19,17 @@ export interface WsExchangeChannelWithParams<TWSTopic extends string = string> {
   product_ids: string[];
 }
 
+/**
+ * Public subscribe/unsubscribe requests for the Coinbase Exchange product group
+ */
 export interface WsExchangeRequestOperation<TWSTopic extends string = string> {
   type: WsOperation;
   channels: (TWSTopic | WsExchangeChannelWithParams)[];
   product_ids?: string[];
 }
 
 /**
+ * Private (authenticated) subscribe/unsubscribe requests for the Coinbase Exchange product group
  * https://docs.cdp.coinbase.com/exchange/docs/websocket-auth
  */
 export type WsExchangeAuthenticatedRequestOperation<
@@ -36,3 +40,27 @@ export type WsExchangeAuthenticatedRequestOperation<
   passphrase: string;
   timestamp: string;
 };
+
+/**
+ * Public subscribe/unsubscribe requests for the Coinbase International product group
+ */
+export interface WsInternationalRequestOperation<
+  TWSTopic extends string = string,
+> {
+  type: Uppercase<WsOperation>;
+  channels: (TWSTopic | WsExchangeChannelWithParams)[];
+  product_ids?: string[];
+}
+
+/**
+ * Private (authenticated) subscribe/unsubscribe requests for the Coinbase International product group
+ * https://docs.cdp.coinbase.com/intx/docs/websocket-auth
+ */
+export type WsInternationalAuthenticatedRequestOperation<
+  TWSTopic extends string = string,
+> = WsInternationalRequestOperation<TWSTopic> & {
+  time: string;
+  key: string;
+  passphrase: string;
+  signature: string;
+};