Merge pull request #142 from tiagosiebler/wsnotice

feat(v3.0.6): handle reconnection notice automatically, allow authenticating for market data

Author: tiagosiebler

.gitignore

@@ -20,6 +20,7 @@ node_modules/
 .env.test
 .cache
 lib
+dist
 bundleReport.html
 .history/
 .idea
@@ -28,4 +29,4 @@ restClientRegex.ts
 testfile.ts
 localtest.sh
 
-repomix.sh
+repomix.sh

package-lock.json

@@ -1,6 +1,6 @@
 {
   "name": "okx-api",
-  "version": "3.0.5",
+  "version": "3.0.6",
   "description": "Complete Node.js SDK for OKX's REST APIs and WebSockets, with TypeScript & end-to-end tests",
   "scripts": {
     "test": "jest",

package.json

@@ -17,6 +17,7 @@ import { checkWebCryptoAPISupported } from './webCryptoAPI.js';
 import {
   getNormalisedTopicRequests,
   PRIVATE_CHANNELS,
+  PUBLIC_CHANNELS_WITH_AUTH,
   safeTerminateWs,
   WS_LOGGER_CATEGORY,
   WsTopicRequest,
@@ -216,7 +217,10 @@ export abstract class BaseWebsocketClient<
 
   protected abstract isWsPong(data: any): boolean;
 
-  protected abstract getWsAuthRequestEvent(wsKey: TWSKey): Promise<object>;
+  protected abstract getWsAuthRequestEvent(
+    wsKey: TWSKey,
+    skipPublicWsKeyCheck: boolean,
+  ): Promise<object | null>;
 
   protected abstract isPrivateTopicRequest(
     request: WsTopicRequest<string>,
@@ -339,7 +343,11 @@ export abstract class BaseWebsocketClient<
       // Queue immediate auth if so
       for (const topicRequest of normalisedTopicRequests) {
         if (PRIVATE_CHANNELS.includes(topicRequest.topic)) {
-          await this.assertIsAuthenticated(wsKey);
+          await this.assertIsAuthenticated(wsKey, false);
+          break;
+        }
+        if (PUBLIC_CHANNELS_WITH_AUTH.includes(topicRequest.topic)) {
+          await this.assertIsAuthenticated(wsKey, true);
           break;
         }
       }
@@ -449,6 +457,37 @@ export abstract class BaseWebsocketClient<
     });
   }
 
+  /**
+   * Closes a connection, if it's even open. If open, this will trigger a reconnect asynchronously.
+   * If closed, trigger a reconnect immediately
+   */
+  public executeReconnectableClose(wsKey: TWSKey, reason: string) {
+    this.logger.info(`${reason} - closing socket to reconnect`, {
+      ...WS_LOGGER_CATEGORY,
+      wsKey,
+      reason,
+    });
+
+    this.clearTimers(wsKey);
+
+    const wasOpen = this.wsStore.isWsOpen(wsKey);
+    if (wasOpen) {
+      safeTerminateWs(this.wsStore.getWs(wsKey), true);
+    }
+
+    if (!wasOpen) {
+      this.logger.info(
+        `${reason} - socket already closed - trigger immediate reconnect`,
+        {
+          ...WS_LOGGER_CATEGORY,
+          wsKey,
+          reason,
+        },
+      );
+      this.reconnectWithDelay(wsKey, this.options.reconnectTimeout);
+    }
+  }
+
   public isConnected(wsKey: TWSKey): boolean {
     return this.wsStore.isConnectionState(
       wsKey,
@@ -568,7 +607,10 @@ export abstract class BaseWebsocketClient<
   }
 
   /** Get a signature, build the auth request and send it */
-  private async sendAuthRequest(wsKey: TWSKey): Promise<unknown> {
+  private async sendAuthRequest(
+    wsKey: TWSKey,
+    skipPublicWsKeyCheck: boolean,
+  ): Promise<unknown> {
     try {
       this.logger.trace('Sending auth request...', {
         ...WS_LOGGER_CATEGORY,
@@ -577,19 +619,22 @@ export abstract class BaseWebsocketClient<
 
       await this.assertIsConnected(wsKey);
 
+      const request = await this.getWsAuthRequestEvent(
+        wsKey,
+        skipPublicWsKeyCheck,
+      );
+
       if (!this.wsStore.getAuthenticationInProgressPromise(wsKey)) {
         this.wsStore.createAuthenticationInProgressPromise(wsKey, false);
       }
 
-      const request = await this.getWsAuthRequestEvent(wsKey);
-
       // console.log('ws auth req', request);
 
       this.tryWsSend(wsKey, JSON.stringify(request));
 
       return this.wsStore.getAuthenticationInProgressPromise(wsKey)?.promise;
     } catch (e) {
-      this.logger.trace(e, { ...WS_LOGGER_CATEGORY, wsKey });
+      this.logger.error(e, { ...WS_LOGGER_CATEGORY, wsKey });
     }
   }
 
@@ -919,12 +964,22 @@ export abstract class BaseWebsocketClient<
       this.isAuthOnConnectWsKey(wsKey) &&
       this.options.authPrivateConnectionsOnConnect
     ) {
-      await this.assertIsAuthenticated(wsKey);
+      await this.assertIsAuthenticated(wsKey, false);
+    }
+
+    const topicsForWsKey = [...this.wsStore.getTopics(wsKey)];
+
+    // Guard to assert auth for some of the public topics that require it
+    for (const topicRequest of topicsForWsKey) {
+      if (PUBLIC_CHANNELS_WITH_AUTH.includes(topicRequest.topic)) {
+        await this.assertIsAuthenticated(wsKey, true);
+        break;
+      }
     }
 
     // Reconnect to topics known before it connected
     const { privateReqs, publicReqs } = this.sortTopicRequestsIntoPublicPrivate(
-      [...this.wsStore.getTopics(wsKey)],
+      topicsForWsKey,
       wsKey,
     );
 
@@ -1206,7 +1261,10 @@ export abstract class BaseWebsocketClient<
   /**
    * Promise-driven method to assert that a ws has been successfully authenticated (will await until auth is confirmed)
    */
-  public async assertIsAuthenticated(wsKey: TWSKey): Promise<unknown> {
+  public async assertIsAuthenticated(
+    wsKey: TWSKey,
+    skipPublicWsKeyCheck: boolean,
+  ): Promise<unknown> {
     const isConnected = this.getWsStore().isConnectionState(
       wsKey,
       WsConnectionStateEnum.CONNECTED,
@@ -1237,7 +1295,7 @@ export abstract class BaseWebsocketClient<
     // Start authentication, it should automatically store/return a promise.
     this.logger.trace('assertIsAuthenticated(): authenticating...');
 
-    await this.sendAuthRequest(wsKey);
+    await this.sendAuthRequest(wsKey, skipPublicWsKeyCheck);
 
     this.logger.trace('assertIsAuthenticated(): newly authenticated!');
   }

src/util/BaseWSClient.ts

@@ -234,6 +234,12 @@ export const PRIVATE_CHANNELS = [
   'grid-sub-orders',
 ];
 
+export const PUBLIC_CHANNELS_WITH_AUTH = [
+  // While these are market data topics on the PUBLIC channel, they do require the public connection to be authenticated to subscribe to these. See #140.
+  'books-l2-tbt',
+  'books50-l2-tpt',
+];
+
 /**
  * The following channels only support the new business wss endpoint:
  * https://www.okx.com/help-center/changes-to-v5-api-websocket-subscription-parameter-and-url

src/util/websocket-util.ts

@@ -470,7 +470,7 @@ export class WebsocketClientLegacy extends EventEmitter {
 
   /**
    * Closes a connection, if it's even open. If open, this will trigger a reconnect asynchronously.
-   * If closed, trigger a reconnect immediately
+   * If closed, trigger a reconnect immediately.
    */
   private executeReconnectableClose(wsKey: WsKey, reason: string) {
     this.logger.info(`${reason} - closing socket to reconnect`, {

src/websocket-client-legacy.ts

@@ -125,8 +125,8 @@ export class WebsocketClient extends BaseWebsocketClient<
   public connectWSAPI(): Promise<unknown[]> {
     /** This call automatically ensures the connection is active AND authenticated before resolving */
     return Promise.allSettled([
-      this.assertIsAuthenticated(this.getMarketWsKey('private')),
-      this.assertIsAuthenticated(this.getMarketWsKey('business')),
+      this.assertIsAuthenticated(this.getMarketWsKey('private'), false),
+      this.assertIsAuthenticated(this.getMarketWsKey('business'), false),
     ]);
   }
 
@@ -422,17 +422,36 @@ export class WebsocketClient extends BaseWebsocketClient<
 
   protected async getWsAuthRequestEvent(
     wsKey: WsKey,
-  ): Promise<WsRequestOperationOKX<WsAuthRequestArg>> {
+    skipIsPublicWsKeyCheck: boolean,
+  ): Promise<WsRequestOperationOKX<WsAuthRequestArg> | null> {
     const isPublicWsKey = PUBLIC_WS_KEYS.includes(wsKey);
     const accounts = this.options.accounts;
     const hasAccountsToAuth = !!accounts?.length;
-    if (isPublicWsKey || !accounts || !hasAccountsToAuth) {
-      this.logger.trace('Starting public only websocket client.', {
-        ...WS_LOGGER_CATEGORY,
-        wsKey,
-        isPublicWsKey,
-        hasAccountsToAuth,
-      });
+    if (isPublicWsKey && !skipIsPublicWsKeyCheck) {
+      this.logger.trace(
+        'Starting public only websocket client. No auth needed.',
+        {
+          ...WS_LOGGER_CATEGORY,
+          wsKey,
+          isPublicWsKey,
+          hasAccountsToAuth,
+          skipIsPublicWsKeyCheck,
+        },
+      );
+      return null;
+    }
+
+    if (!accounts || !hasAccountsToAuth) {
+      this.logger.trace(
+        'Starting public only websocket client - missing keys.',
+        {
+          ...WS_LOGGER_CATEGORY,
+          wsKey,
+          isPublicWsKey,
+          hasAccountsToAuth,
+          skipIsPublicWsKeyCheck,
+        },
+      );
       throw new Error('Cannot auth - missing api or secret or pass in config');
     }
 
@@ -704,6 +723,31 @@ export class WebsocketClient extends BaseWebsocketClient<
         return results;
       }
 
+      if (msg.event === 'notice') {
+        const WSNOTICE = {
+          CLOSING_FOR_UPGRADE_RECOMMEND_RECONNECT: '64008',
+        };
+        if (msg?.code === WSNOTICE.CLOSING_FOR_UPGRADE_RECOMMEND_RECONNECT) {
+          const closeReason = `Received notice (${msg.code} - "${msg?.msg}") - closing socket to reconnect`;
+          this.logger.info(closeReason, {
+            ...WS_LOGGER_CATEGORY,
+            ...msg,
+            wsKey,
+          });
+
+          // Queue immediate reconnection workflow
+          this.executeReconnectableClose(wsKey, closeReason);
+
+          // Emit notice to client for visibility
+          results.push({
+            eventType: 'update',
+            event: emittableEvent,
+          });
+
+          return results;
+        }
+      }
+
       this.logger.info('Unhandled/unrecognised ws event message', {
         ...WS_LOGGER_CATEGORY,
         message: msg || 'no message',
@@ -765,7 +809,7 @@ export class WebsocketClient extends BaseWebsocketClient<
       this.logger.trace(
         'sendWSAPIRequest(): assertIsAuthenticated(${wsKey})...',
       );
-      await this.assertIsAuthenticated(wsKey);
+      await this.assertIsAuthenticated(wsKey, false);
       this.logger.trace(
         'sendWSAPIRequest(): assertIsAuthenticated(${wsKey}) ok',
       );