feat: implement EVM balance tracker and integrate with metrics

Author: cctdaniel

apps/price_pusher/grafana-dashboard.sample.json

@@ -396,9 +396,15 @@
               "job#C": true,
               "price_id": false,
               "price_id#B": true,
-              "price_id#C": true
+              "price_id#C": true,
+              "Value #C": false
+            },
+            "includeByName": {
+              "alias 1": true,
+              "price_id": true,
+              "Value #B": true,
+              "Value #C": true
             },
-            "includeByName": {},
             "indexByName": {
               "Time 1": 4,
               "Time 2": 10,

apps/price_pusher/src/balance-tracker/evm.ts

@@ -0,0 +1,51 @@
+import { SuperWalletClient } from "../evm/super-wallet";
+import { BaseBalanceTracker, BaseBalanceTrackerConfig } from "./interface";
+
+/**
+ * EVM-specific configuration for balance tracker
+ */
+export interface EvmBalanceTrackerConfig extends BaseBalanceTrackerConfig {
+  /** EVM wallet client */
+  client: SuperWalletClient;
+  /** EVM address with 0x prefix */
+  address: `0x${string}`;
+}
+
+/**
+ * EVM-specific implementation of the balance tracker
+ */
+export class EvmBalanceTracker extends BaseBalanceTracker {
+  private client: SuperWalletClient;
+  private evmAddress: `0x${string}`;
+
+  constructor(config: EvmBalanceTrackerConfig) {
+    super({
+      ...config,
+      logger: config.logger.child({ module: "EvmBalanceTracker" }),
+    });
+
+    this.client = config.client;
+    this.evmAddress = config.address;
+  }
+
+  /**
+   * EVM-specific implementation of balance update
+   */
+  protected async updateBalance(): Promise<void> {
+    try {
+      const balance = await this.client.getBalance({
+        address: this.evmAddress,
+      });
+
+      this.metrics.updateWalletBalance(this.address, this.network, balance);
+      this.logger.debug(
+        `Updated EVM wallet balance: ${this.address} = ${balance.toString()}`,
+      );
+    } catch (error) {
+      this.logger.error(
+        { error },
+        "Error fetching EVM wallet balance for metrics",
+      );
+    }
+  }
+}

apps/price_pusher/src/balance-tracker/index.ts

@@ -0,0 +1,46 @@
+// Export the interfaces
+export * from "./interface";
+
+// Export chain-specific implementations
+export * from "./evm";
+// export * from "./aptos";
+
+// Factory function to create the appropriate balance tracker based on the chain
+import { PricePusherMetrics } from "../metrics";
+import { Logger } from "pino";
+import { DurationInSeconds } from "../utils";
+import { IBalanceTracker } from "./interface";
+import { EvmBalanceTracker } from "./evm";
+import { SuperWalletClient } from "../evm/super-wallet";
+
+/**
+ * Parameters for creating an EVM balance tracker
+ */
+export interface CreateEvmBalanceTrackerParams {
+  client: SuperWalletClient;
+  address: `0x${string}`;
+  network: string;
+  updateInterval: DurationInSeconds;
+  metrics: PricePusherMetrics;
+  logger: Logger;
+}
+
+/**
+ * Factory function to create a balance tracker for EVM chains
+ */
+export function createEvmBalanceTracker(
+  params: CreateEvmBalanceTrackerParams,
+): IBalanceTracker {
+  return new EvmBalanceTracker({
+    client: params.client,
+    address: params.address,
+    network: params.network,
+    updateInterval: params.updateInterval,
+    metrics: params.metrics,
+    logger: params.logger,
+  });
+}
+
+// Additional factory functions for other chains would follow the same pattern:
+// export function createSuiBalanceTracker(params: CreateSuiBalanceTrackerParams): IBalanceTracker { ... }
+// export function createSolanaBalanceTracker(params: CreateSolanaBalanceTrackerParams): IBalanceTracker { ... }

apps/price_pusher/src/balance-tracker/interface.ts

@@ -0,0 +1,97 @@
+import { Logger } from "pino";
+import { PricePusherMetrics } from "../metrics";
+import { DurationInSeconds } from "../utils";
+
+/**
+ * Common configuration properties for all balance trackers
+ */
+export interface BaseBalanceTrackerConfig {
+  /** Address of the wallet to track */
+  address: string;
+  /** Name/ID of the network/chain */
+  network: string;
+  /** How often to update the balance */
+  updateInterval: DurationInSeconds;
+  /** Metrics instance to report balance updates */
+  metrics: PricePusherMetrics;
+  /** Logger instance */
+  logger: Logger;
+}
+
+/**
+ * Interface for all balance trackers to implement
+ * Each chain will have its own implementation of this interface
+ */
+export interface IBalanceTracker {
+  /**
+   * Start tracking the wallet balance
+   */
+  start(): Promise<void>;
+
+  /**
+   * Stop tracking the wallet balance
+   */
+  stop(): void;
+}
+
+/**
+ * Abstract base class that implements common functionality for all balance trackers
+ */
+export abstract class BaseBalanceTracker implements IBalanceTracker {
+  protected address: string;
+  protected network: string;
+  protected updateInterval: DurationInSeconds;
+  protected metrics: PricePusherMetrics;
+  protected logger: Logger;
+  protected isRunning: boolean = false;
+
+  constructor(config: BaseBalanceTrackerConfig) {
+    this.address = config.address;
+    this.network = config.network;
+    this.updateInterval = config.updateInterval;
+    this.metrics = config.metrics;
+    this.logger = config.logger;
+  }
+
+  public async start(): Promise<void> {
+    if (this.isRunning) {
+      return;
+    }
+
+    this.isRunning = true;
+
+    // Initial balance update
+    await this.updateBalance();
+
+    // Start the update loop
+    this.startUpdateLoop();
+  }
+
+  private async startUpdateLoop(): Promise<void> {
+    // We're using dynamic import to avoid circular dependencies
+    const { sleep } = await import("../utils");
+
+    // Run in a loop to regularly update the balance
+    for (;;) {
+      // Wait first, since we already did the initial update in start()
+      await sleep(this.updateInterval * 1000);
+
+      // Only continue if we're still running
+      if (!this.isRunning) {
+        break;
+      }
+
+      await this.updateBalance();
+    }
+  }
+
+  /**
+   * Chain-specific balance update implementation
+   * Each chain will implement this method differently
+   */
+  protected abstract updateBalance(): Promise<void>;
+
+  public stop(): void {
+    this.isRunning = false;
+  }
+}

apps/price_pusher/src/controller.ts

@@ -4,20 +4,10 @@ import { IPriceListener, IPricePusher } from "./interface";
 import { PriceConfig, shouldUpdate, UpdateCondition } from "./price-config";
 import { Logger } from "pino";
 import { PricePusherMetrics } from "./metrics";
-import { SuperWalletClient } from "./evm/super-wallet";
-
-// Define the wallet balance info interface
-interface WalletBalanceInfo {
-  client: SuperWalletClient;
-  address: `0x${string}`;
-  network: string;
-  updateInterval: DurationInSeconds;
-}
 
 export class Controller {
   private pushingFrequency: DurationInSeconds;
   private metrics?: PricePusherMetrics;
-  private walletBalanceInfo?: WalletBalanceInfo;
 
   constructor(
     private priceConfigs: PriceConfig[],
@@ -28,46 +18,20 @@ export class Controller {
     config: {
       pushingFrequency: DurationInSeconds;
       metrics?: PricePusherMetrics;
-      walletBalanceInfo?: WalletBalanceInfo;
     },
   ) {
     this.pushingFrequency = config.pushingFrequency;
     this.metrics = config.metrics;
-    this.walletBalanceInfo = config.walletBalanceInfo;
 
     // Set the number of price feeds if metrics are enabled
     this.metrics?.setPriceFeedsTotal(this.priceConfigs.length);
   }
 
-  // Get wallet balance and update metrics
-  private async updateWalletBalance(): Promise<void> {
-    if (!this.metrics || !this.walletBalanceInfo) return;
-
-    try {
-      const { client, address, network } = this.walletBalanceInfo;
-      const balance = await client.getBalance({
-        address: address,
-      });
-
-      this.metrics.updateWalletBalance(address, network, balance);
-      this.logger.debug(
-        `Updated wallet balance: ${address} = ${balance.toString()}`,
-      );
-    } catch (error) {
-      this.logger.error({ error }, "Error fetching wallet balance for metrics");
-    }
-  }
-
   async start() {
     // start the listeners
     await this.sourcePriceListener.start();
     await this.targetPriceListener.start();
 
-    // Update wallet balance initially if metrics are enabled
-    if (this.metrics && this.walletBalanceInfo) {
-      await this.updateWalletBalance();
-    }
-
     // wait for the listeners to get updated. There could be a restart
     // before this run and we need to respect the cooldown duration as
     // their might be a message sent before.
@@ -80,11 +44,6 @@ export class Controller {
       const pricesToPush: PriceConfig[] = [];
       const pubTimesToPush: UnixTimestamp[] = [];
 
-      // Update wallet balance if metrics are enabled
-      if (this.metrics && this.walletBalanceInfo) {
-        await this.updateWalletBalance();
-      }
-
       for (const priceConfig of this.priceConfigs) {
         const priceId = priceConfig.id;
         const alias = priceConfig.alias;

apps/price_pusher/src/evm/command.ts

@@ -12,6 +12,7 @@ import { createClient } from "./super-wallet";
 import { createPythContract } from "./pyth-contract";
 import { isWsEndpoint, filterInvalidPriceItems } from "../utils";
 import { PricePusherMetrics } from "../metrics";
+import { createEvmBalanceTracker } from "../balance-tracker";
 
 export default {
   command: "evm",
@@ -199,17 +200,24 @@ export default {
       {
         pushingFrequency,
         metrics,
-        walletBalanceInfo: metrics
-          ? {
-              client,
-              address: client.account.address,
-              network: await client.getChainId().then((id) => id.toString()),
-              updateInterval: pushingFrequency,
-            }
-          : undefined,
       },
     );
 
+    // Create and start the balance tracker if metrics are enabled
+    if (metrics) {
+      const balanceTracker = createEvmBalanceTracker({
+        client,
+        address: client.account.address,
+        network: await client.getChainId().then((id) => id.toString()),
+        updateInterval: pushingFrequency,
+        metrics,
+        logger,
+      });
+
+      // Start the balance tracker
+      await balanceTracker.start();
+    }
+
     await controller.start();
   },
 };