bump version to v3.1.10

Author: will pankiewicz

packages/common/src/ApiHandler/ApiHandler.ts

@@ -1,9 +1,9 @@
 import { ApiPromise, WsProvider } from "@polkadot/api";
 import EventEmitter from "eventemitter3";
 
-import logger from "./logger";
-import { sleep } from "./util";
-import { POLKADOT_API_TIMEOUT } from "./constants";
+import logger from "../logger";
+import { sleep } from "../utils/util";
+import { API_PROVIDER_TIMEOUT, POLKADOT_API_TIMEOUT } from "../constants";
 
 export const apiLabel = { label: "ApiHandler" };
 
@@ -12,166 +12,219 @@ export const apiLabel = { label: "ApiHandler" };
  * to a different provider if one proves troublesome.
  */
 class ApiHandler extends EventEmitter {
-  private _api: ApiPromise;
-  private _endpoints: string[];
-  private _reconnectLock: boolean;
-  private _reconnectTries = 0;
-  static isConnected: any;
-  static _reconnect: any;
-
-  constructor(api: ApiPromise, endpoints?: string[]) {
+  private _wsProvider?: WsProvider;
+  private _api: ApiPromise | null = null;
+  private readonly _endpoints: string[] = [];
+  static isConnected = false;
+  private healthCheckInProgress = false;
+  private _currentEndpoint?: string;
+  public upSince: number = Date.now();
+  constructor(endpoints: string[]) {
     super();
-    this._api = api;
-    // this._endpoints = endpoints.sort(() => Math.random() - 0.5);
-    this._registerEventHandlers(api);
+    this._endpoints = endpoints.sort(() => Math.random() - 0.5);
+    this.upSince = Date.now();
   }
 
-  static async createApi(endpoints, reconnectTries = 0) {
-    const timeout = 12;
-    let api, wsProvider;
-    const healthCheck = async (api) => {
-      logger.info(`Performing health check for WS Provider for rpc.`, apiLabel);
+  async healthCheck(retries = 0): Promise<boolean> {
+    if (retries < 10) {
+      try {
+        this.healthCheckInProgress = true;
+        let chain;
 
-      await sleep(timeout * 1000);
-      if (api && api?.isConnected) {
-        logger.info(`All good. Connected`, apiLabel);
-        return true;
-      } else {
-        logger.info(
-          `rpc endpoint still disconnected after ${timeout} seconds. Disconnecting `,
+        const isConnected = this._wsProvider?.isConnected;
+        if (isConnected && !this._api?.isConnected) {
+          try {
+            chain = await this._api?.rpc.system.chain();
+          } catch (e) {
+            await sleep(API_PROVIDER_TIMEOUT);
+          }
+        }
+        chain = await this._api?.rpc.system.chain();
+
+        if (isConnected && chain) {
+          this.healthCheckInProgress = false;
+          return true;
+        } else {
+          await sleep(API_PROVIDER_TIMEOUT);
+          logger.info(`api still disconnected, disconnecting.`, apiLabel);
+          await this._wsProvider?.disconnect();
+          await this.getProvider(this._endpoints);
+          await this.getAPI();
+          return false;
+        }
+      } catch (e: unknown) {
+        const errorMessage =
+          e instanceof Error ? e.message : "An unknown error occurred";
+        logger.error(
+          `Error in health check for WS Provider for rpc. ${errorMessage}`,
           apiLabel,
         );
-        await api.disconnect();
-
-        throw new Error(
-          `rpc endpoint still disconnected after ${timeout} seconds.`,
-        );
+        this.healthCheckInProgress = false;
+        return false;
       }
-    };
+    }
+    return false;
+  }
 
-    try {
-      wsProvider = new WsProvider(
+  public currentEndpoint() {
+    return this._currentEndpoint;
+  }
+
+  async getProvider(endpoints: string[]): Promise<WsProvider> {
+    return await new Promise<WsProvider>((resolve, reject) => {
+      const wsProvider = new WsProvider(
         endpoints,
-        undefined,
+        5000,
         undefined,
         POLKADOT_API_TIMEOUT,
       );
 
-      api = new ApiPromise({
-        provider: new WsProvider(
-          endpoints,
-          undefined,
-          undefined,
-          POLKADOT_API_TIMEOUT,
-        ),
-        // throwOnConnect: true,
+      wsProvider.on("disconnected", async () => {
+        try {
+          const isHealthy = await this.healthCheck();
+          logger.info(
+            `[Disconnection] ${this._currentEndpoint}} Health check result: ${isHealthy}`,
+            apiLabel,
+          );
+          resolve(wsProvider);
+        } catch (error: any) {
+          logger.warn(
+            `WS provider for rpc ${endpoints[0]} disconnected!`,
+            apiLabel,
+          );
+          reject(error);
+        }
       });
+      wsProvider.on("connected", () => {
+        logger.info(`WS provider for rpc ${endpoints[0]} connected`, apiLabel);
+        this._currentEndpoint = endpoints[0];
+        resolve(wsProvider);
+      });
+      wsProvider.on("error", async () => {
+        try {
+          const isHealthy = await this.healthCheck();
+          logger.info(
+            `[Error] ${this._currentEndpoint} Health check result: ${isHealthy}`,
+            apiLabel,
+          );
+          resolve(wsProvider);
+        } catch (error: any) {
+          logger.error(`Error thrown for rpc ${this._endpoints[0]}`, apiLabel);
+          reject(error);
+        }
+      });
+    });
+  }
 
-      api
-        .on("connected", () => {
-          logger.info(`Connected to chain ${endpoints[0]}`, apiLabel);
-        })
-        .on("disconnected", async () => {
-          logger.warn(`Disconnected from chain`, apiLabel);
-          try {
-            await healthCheck(wsProvider);
-            await Promise.resolve(api);
-          } catch (error: any) {
-            await Promise.reject(error);
-          }
-        })
-        .on("ready", () => {
-          logger.info(`API connection ready ${endpoints[0]}`, apiLabel);
-        })
-        .on("error", async (error) => {
-          logger.warn("The API has an error", apiLabel);
-          logger.error(error, apiLabel);
-          logger.warn(`attempting to reconnect to ${endpoints[0]}`, apiLabel);
-          try {
-            await healthCheck(wsProvider);
-            await Promise.resolve(api);
-          } catch (error: any) {
-            await Promise.reject(error);
-          }
-        });
-
-      if (api) {
-        await api.isReadyOrError.catch(logger.error);
+  async getAPI(retries = 0): Promise<ApiPromise> {
+    if (this._wsProvider && this._api && this._api?.isConnected) {
+      return this._api;
+    }
+    const endpoints = this._endpoints.sort(() => Math.random() - 0.5);
 
-        return api;
-      }
+    try {
+      logger.info(
+        `[getAPI]: try ${retries} creating provider with endpoint ${endpoints[0]}`,
+        apiLabel,
+      );
+      const provider = await this.getProvider(endpoints);
+      this._wsProvider = provider;
+      logger.info(
+        `[getAPI]: provider created with endpoint: ${endpoints[0]}`,
+        apiLabel,
+      );
+      const api = await ApiPromise.create({
+        provider: provider,
+        noInitWarn: true,
+      });
+      await api.isReadyOrError;
+      logger.info(`[getApi] Api is ready`, apiLabel);
+      return api;
     } catch (e) {
-      logger.error(`there was an error: `, apiLabel);
-      logger.error(e, apiLabel);
-      if (reconnectTries < 10) {
-        return await this.createApi(
-          endpoints.sort(() => Math.random() - 0.5),
-          reconnectTries + 1,
-        );
+      if (retries < 15) {
+        return await this.getAPI(retries + 1);
       } else {
-        return api;
+        const provider = await this.getProvider(endpoints);
+        return await ApiPromise.create({
+          provider: provider,
+          noInitWarn: true,
+        });
       }
     }
   }
 
-  static async create(endpoints: string[]): Promise<ApiHandler> {
-    try {
-      const api = await this.createApi(
-        endpoints.sort(() => Math.random() - 0.5),
-      );
-
-      return new ApiHandler(api, endpoints);
-    } catch (e) {
-      logger.info(`there was an error: `, apiLabel);
-      logger.error(e, apiLabel);
-    }
+  async setAPI() {
+    const api = await this.getAPI(0);
+    this._api = api;
+    this._registerEventHandlers(this._api);
+    return api;
   }
 
   isConnected(): boolean {
-    return this._api.isConnected;
+    return this._wsProvider?.isConnected || false;
   }
 
-  getApi(): ApiPromise {
-    return this._api;
+  getApi(): ApiPromise | null {
+    if (!this._api) {
+      return null;
+    } else {
+      return this._api;
+    }
   }
 
   _registerEventHandlers(api: ApiPromise): void {
+    if (!api) {
+      logger.warn(`API is null, cannot register event handlers.`, apiLabel);
+      return;
+    }
+
+    logger.info(`Registering event handlers...`, apiLabel);
+
     api.query.system.events((events) => {
-      // Loop through the Vec<EventRecord>
       events.forEach((record) => {
-        // Extract the phase, event and the event types
         const { event } = record;
 
-        if (event.section == "session" && event.method == "NewSession") {
-          const [session_index] = event.data;
-
-          this.emit("newSession", {
-            sessionIndex: session_index.toString(),
-          });
+        if (event.section === "session" && event.method === "NewSession") {
+          const sessionIndex = Number(event?.data[0]?.toString()) || 0;
+          this.handleNewSessionEvent(sessionIndex);
         }
 
         if (
-          event.section == "staking" &&
-          (event.method == "Reward" || event.method == "Rewarded")
+          event.section === "staking" &&
+          (event.method === "Reward" || event.method === "Rewarded")
         ) {
           const [stash, amount] = event.data;
-
-          this.emit("reward", {
-            stash: stash.toString(),
-            amount: amount.toString(),
-          });
+          this.handleRewardEvent(stash.toString(), amount.toString());
         }
 
         if (event.section === "imOnline" && event.method === "SomeOffline") {
-          const offlineVals = event.data.toJSON()[0].map((val) => val[0]);
-
-          this.emit("someOffline", {
-            offlineVals: offlineVals,
-          });
+          const data = event.data.toJSON();
+          const offlineVals =
+            Array.isArray(data) && Array.isArray(data[0])
+              ? data[0].reduce((acc: string[], val) => {
+                  if (Array.isArray(val) && typeof val[0] === "string") {
+                    acc.push(val[0]);
+                  }
+                  return acc;
+                }, [])
+              : [];
+          this.handleSomeOfflineEvent(offlineVals as string[]);
         }
       });
     });
   }
+
+  handleNewSessionEvent(sessionIndex: number): void {
+    this.emit("newSession", { sessionIndex });
+  }
+
+  handleRewardEvent(stash: string, amount: string): void {
+    this.emit("reward", { stash, amount });
+  }
+
+  handleSomeOfflineEvent(offlineVals: string[]): void {
+    this.emit("someOffline", { offlineVals });
+  }
 }
 
 export default ApiHandler;