fix(websocket): report errors as Error objects and recover from auth failures

.github/workflows/build.yml

@@ -17,3 +17,4 @@ jobs:
           node-version-file: '.nvmrc'
       - run: npm install
       - run: npm run build
+      - run: npm test

README.md

@@ -63,6 +63,44 @@ async function someBrokerizeActions() {
 someBrokerizeActions().then(console.log, console.error)
 ```
 
+# websocket error handling
+
+Every error reported by the websocket client is a `BrokerizeWebSocketError`, a subclass of
+`BrokerizeError`, so it can be handled like any other error of this library. Errors reach an
+application through two separate channels:
+
+- **Subscription callbacks** only receive _terminal_ errors: the subscription is over and will not
+  receive any further data. This maps directly onto e.g. an RxJS `subscriber.error()`.
+- **`subscribeConnectionState`** reports the state of the underlying connection, including the
+  errors that the client recovers from on its own. Use it to tell the user that live updates are
+  currently unavailable, instead of tearing down subscriptions on every connection hiccup.
+
+```js
+const client = ctx.createWebSocketClient();
+
+client.subscribeConnectionState(({ status, error }) => {
+  /* 'closed' | 'connecting' | 'open' | 'reconnecting' | 'fatal' */
+  if (status == "fatal") {
+    /* the client is unusable, all subscriptions were terminated with `error`. */
+    console.error("live updates stopped", error);
+    client.reconnect(); /* ... and subscribe again */
+  }
+});
+
+client.subscribeInvalidate({ entity: "sessions" }, (err) => {
+  if (err) {
+    /* err is a BrokerizeWebSocketError with err.terminal === true. */
+    console.error("subscription ended", err.code, err.message);
+    return;
+  }
+  reloadSessions();
+});
+```
+
+A connection that cannot be authenticated is retried once with a freshly refreshed token, because
+the usual cause is an id token that expired in flight. Only if that retry is rejected as well does
+the client become `fatal`.
+
 # development & maintenance
 
 The code in `src/swagger` is generated by the npm script `npm run generate` from the OpenAPI spec and will be regenerated whenever API changes are made. No manual changes should be made in the `src/swagger` directory. `npm run download-spec` will download the current spec from `api-preview.brokerize.com`.

etc/client.api.md

@@ -42,7 +42,7 @@ interface Amount {
 // @public (undocumented)
 export interface Auth {
     // (undocumented)
-    getToken: () => Promise<{
+    getToken: (opts?: GetTokenOptions) => Promise<{
         idToken: string;
     }>;
 }
@@ -346,6 +346,11 @@ export class Brokerize {
     refreshGuestUser(refreshToken: string): Promise<GuestAuthContextConfiguration>;
 }
 
+// @public
+export class BrokerizeAbortError extends Error {
+    constructor(msg: string);
+}
+
 // @public (undocumented)
 export interface BrokerizeConfig {
     acceptLanguage?: string | (() => string | null | undefined | Promise<string | null | undefined>);
@@ -388,15 +393,41 @@ export class BrokerizeTimeoutError extends BrokerizeError {
 
 // @public (undocumented)
 export interface BrokerizeWebSocketClient {
+    // (undocumented)
+    getConnectionState: () => WebSocketConnectionState;
+    // (undocumented)
+    reconnect: () => void;
+    // (undocumented)
+    subscribeConnectionState: (callback: ConnectionStateCallback) => Subscription;
     // (undocumented)
     subscribeDecoupledOperation: (subscribe: Pick<SubscribeDecoupledOperation, "sessionId" | "decoupledOperationId">, callback: Callback) => Subscription;
     // (undocumented)
     subscribeInvalidate: (subscribe: SubscribeInvalidateDetails, callback: Callback) => Subscription;
 }
 
-// @public (undocumented)
-type BrokerizeWebSocketError = {
+// @public
+export class BrokerizeWebSocketError extends BrokerizeError {
+    constructor(input: BrokerizeWebSocketErrorParams);
+    readonly cause?: unknown;
+    readonly terminal: boolean;
+}
+
+// @public
+type BrokerizeWebSocketError_2 = {
     message: string;
+    code?: string;
+};
+
+// @public
+export type BrokerizeWebSocketErrorCode = "AUTH" | "WEBSOCKET_CONNECTION_FAILED" | "WEBSOCKET_PROTOCOL_ERROR" | "WEBSOCKET_ERROR";
+
+// @public (undocumented)
+export type BrokerizeWebSocketErrorParams = {
+    msg: string;
+    code: BrokerizeWebSocketErrorCode;
+    httpStatusCode?: number;
+    terminal?: boolean;
+    cause?: unknown;
 };
 
 // @public
@@ -442,7 +473,7 @@ interface BrokerMetaImages {
     lightSquare: string;
 }
 
-// @public (undocumented)
+// @public
 export type Callback<T = any> = (err: any, data: T) => void;
 
 // @public
@@ -590,7 +621,7 @@ export type CognitoConfig = {
 // @public (undocumented)
 export type CognitoFacade = {
     createSession: (cognitoPoolConfig: CognitoPoolConfig, authCfg: RegisteredUserAuthContextConfiguration) => {
-        getToken: () => Promise<{
+        getToken: (opts?: GetTokenOptions) => Promise<{
             idToken: string;
         }>;
     };
@@ -616,6 +647,9 @@ interface ConfirmOAuthResponse {
     sessionId: string;
 }
 
+// @public (undocumented)
+export type ConnectionStateCallback = (state: WebSocketConnectionState) => void;
+
 // @public
 interface CostDetailsLink {
     footerText?: string;
@@ -1109,6 +1143,11 @@ interface GetRecoveryPhrasesResponse {
 // @public
 function getSizeUnitsFromConstraints(preparedTrade: PreparedTrade, orderModel: OrderModel, direction: Direction, cashAccountId: string): string[];
 
+// @public (undocumented)
+export type GetTokenOptions = {
+    forceRefresh?: boolean;
+};
+
 // @public
 interface GetUserResponse {
     createdAt: Date;
@@ -1128,6 +1167,9 @@ type InvalidateMessage = {
     subscriptionId: number;
 };
 
+// @public
+export function isBrokerizeWebSocketError(x: unknown): x is BrokerizeWebSocketError;
+
 // @public
 interface LegalTermConfirmItem {
     html: string;
@@ -2202,9 +2244,18 @@ type WebSocketCommandUnsubscribe = {
     subscriptionId: number;
 };
 
+// @public (undocumented)
+export type WebSocketConnectionState = {
+    status: WebSocketConnectionStatus;
+    error?: BrokerizeWebSocketError;
+};
+
+// @public
+export type WebSocketConnectionStatus = "closed" | "connecting" | "open" | "reconnecting" | "fatal";
+
 // @public (undocumented)
 type WebSocketError = {
-    error: BrokerizeWebSocketError;
+    error: BrokerizeWebSocketError_2;
 };
 
 // @public (undocumented)
@@ -2213,7 +2264,7 @@ type WebSocketMessage = WebSocketSubscriptionMessage | WebSocketError | WebSocke
 // @public (undocumented)
 type WebSocketMessageErrorOnSubscription = {
     subscriptionId: number;
-    error: BrokerizeWebSocketError;
+    error: BrokerizeWebSocketError_2;
 };
 
 // @public (undocumented)
@@ -2243,7 +2294,7 @@ declare namespace WebSocketTypes {
         WebSocketMessageErrorOnSubscription,
         InvalidateMessage,
         UpdateDecoupledOperationMessage,
-        BrokerizeWebSocketError,
+        BrokerizeWebSocketError_2 as BrokerizeWebSocketError,
         WebSocketError,
         DecoupledOperationState_2 as DecoupledOperationState,
         DecoupledOperationStatus_2 as DecoupledOperationStatus

package.json

@@ -52,7 +52,8 @@
     "openapi-preprocess": "node openapi-preprocess.js",
     "prepack": "node scripts/prepack-package-json.cjs strip",
     "postpack": "node scripts/prepack-package-json.cjs restore",
-    "prepare": "husky"
+    "prepare": "husky",
+    "test": "node --test \"test/**/*.test.js\""
   },
   "lint-staged": {
     "*.{js,ts}": [

src/apiCtx.ts

@@ -92,8 +92,17 @@ export interface GuestAuthContextConfiguration {
   tokens: { updatedAt: number; response: Models.CreateGuestUserResponse };
 }
 
+export type GetTokenOptions = {
+  /**
+   * Request a fresh token even if the currently known one is not expired yet. Used by the
+   * websocket client to retry once after the server rejected a token, before giving up.
+   * Implementations that cannot refresh on demand may ignore this.
+   */
+  forceRefresh?: boolean;
+};
+
 export interface Auth {
-  getToken: () => Promise<{ idToken: string }>;
+  getToken: (opts?: GetTokenOptions) => Promise<{ idToken: string }>;
 }
 
 export function createConfiguration(cfg: BrokerizeConfig) {
@@ -123,7 +132,7 @@ export function createAuth({
       JSON.stringify(authCfg),
     ) as GuestAuthContextConfiguration; // clone it
     return {
-      async getToken() {
+      async getToken(opts?: GetTokenOptions) {
         if (guestAuthCfg.tokens) {
           let expiresIn = guestAuthCfg.tokens.response.expiresIn;
           if (expiresIn == null) {
@@ -136,7 +145,8 @@ export function createAuth({
           /* modern tokens */
           const tokenExpiresAt =
             guestAuthCfg.tokens.updatedAt + expiresIn * 1000 - 10000;
-          const needsRefresh = Date.now() > tokenExpiresAt;
+          const needsRefresh =
+            opts?.forceRefresh || Date.now() > tokenExpiresAt;
           if (needsRefresh && guestAuthCfg.tokens.response.refreshToken) {
             if (!cfg.fetch) {
               throw new Error(
@@ -219,8 +229,8 @@ export function createAuth({
       authCfg,
     );
     return {
-      async getToken() {
-        return session.getToken();
+      async getToken(opts?: GetTokenOptions) {
+        return session.getToken(opts);
       },
     };
   } else {
@@ -244,7 +254,7 @@ export type CognitoFacade = {
     cognitoPoolConfig: CognitoPoolConfig,
     authCfg: RegisteredUserAuthContextConfiguration,
   ) => {
-    getToken: () => Promise<{ idToken: string }>;
+    getToken: (opts?: GetTokenOptions) => Promise<{ idToken: string }>;
   };
 };
 

src/authorizedApiContext.ts

@@ -5,7 +5,7 @@ import {
   createConfiguration,
   withAcceptLanguage,
 } from "./apiCtx";
-import { createErrorFromResponse } from "./errors";
+import { BrokerizeAbortError, createErrorFromResponse } from "./errors";
 import { createPollingSubscription } from "./pollingSubscription";
 import * as openApiClient from "./swagger";
 import {
@@ -45,6 +45,11 @@ export class AuthorizedApiContext {
   private readonly _orderApiCryptoService: openApiClient.OrderApi;
   private readonly _logoutSubject: Subject<void>;
   private readonly _wsClient: BrokerizeWebSocketClient;
+  /**
+   * `true` if this context created the websocket client itself and therefore has to destroy it.
+   * A client passed in by the application is owned by the application.
+   */
+  private readonly _ownsWsClient: boolean;
   private readonly _cache: {
     getBrokers?: Promise<openApiClient.GetBrokersResponse>;
   };
@@ -144,6 +149,7 @@ export class AuthorizedApiContext {
       );
     }
     this._abortController = cfg.createAbortController();
+    this._ownsWsClient = !wsClient;
     this._wsClient = wsClient || this._initInternalWebSocketClient();
     if (
       (this._wsClient as BrokerizeWebSocketClientImpl)._setAuthorizedApiContext
@@ -171,7 +177,7 @@ export class AuthorizedApiContext {
   }
   private async _initRequestInit() {
     if (this._isDestroyed) {
-      throw new Error("AuthorizedApiContext is destroyed");
+      throw new BrokerizeAbortError("AuthorizedApiContext is destroyed");
     }
     const tok = await this._auth.getToken();
     const headers = await withAcceptLanguage(this._cfg, {
@@ -765,12 +771,24 @@ export class AuthorizedApiContext {
       subscribeInvalidate: this._wsClient.subscribeInvalidate.bind(
         this._wsClient,
       ),
+      subscribeConnectionState: this._wsClient.subscribeConnectionState.bind(
+        this._wsClient,
+      ),
+      getConnectionState: this._wsClient.getConnectionState.bind(
+        this._wsClient,
+      ),
+      reconnect: this._wsClient.reconnect.bind(this._wsClient),
     };
     return wrappedClient;
   }
   destroy() {
     this._isDestroyed = true;
     this._abortController.abort();
+    if (this._ownsWsClient) {
+      /* otherwise its reconnect timer would keep running and keep reconnecting with the
+         tokens of a context that is gone. */
+      (this._wsClient as BrokerizeWebSocketClientImpl).destroy();
+    }
   }
   subscribeLogout(callback: Callback): Subscription {
     const s = this._logoutSubject.subscribe({