ft_watcher: add logic to persist token info

Signed-off-by: bingyuyap <[email protected]>

database/fast-transfer-schema.sql

@@ -5,6 +5,7 @@ DROP TABLE IF EXISTS fast_transfer_settlements;
 DROP TABLE IF EXISTS auction_logs;
 DROP TABLE IF EXISTS auction_history_mapping;
 DROP TABLE IF EXISTS redeem_swaps;
+DROP TABLE IF EXISTS token_infos;
 
 DROP TYPE IF EXISTS FastTransferStatus;
 DROP TYPE IF EXISTS FastTransferProtocol;
@@ -107,3 +108,14 @@ CREATE TABLE redeem_swaps (
   relaying_fee BIGINT NOT NULL,
   redeem_time TIMESTAMP NOT NULL
 );
+
+-- Token Infos table to store information about different tokens
+-- A normalized table for us to reference token info for analytics purposes
+CREATE TABLE token_infos (
+  name VARCHAR(255) NOT NULL,
+  chain_id INTEGER NOT NULL,
+  decimals INTEGER NOT NULL,
+  symbol VARCHAR(255) NOT NULL,
+  token_address VARCHAR(255) NOT NULL,
+  PRIMARY KEY (token_address, chain_id)
+);

watcher/src/fastTransfer/swapLayer/parser.ts

@@ -1,18 +1,31 @@
 import { ethers } from 'ethers';
 import { TransferCompletion } from '../types';
 import { parseVaa } from '@wormhole-foundation/wormhole-monitor-common';
+import { Knex } from 'knex';
+import { Chain, chainToChainId } from '@wormhole-foundation/sdk-base';
+import { TokenInfoManager } from '../../utils/TokenInfoManager';
 
 class SwapLayerParser {
   private provider: ethers.providers.JsonRpcProvider;
   private swapLayerAddress: string;
   private swapLayerInterface: ethers.utils.Interface;
-
-  constructor(provider: ethers.providers.JsonRpcProvider, swapLayerAddress: string) {
+  private tokenInfoManager: TokenInfoManager | null = null;
+
+  constructor(
+    provider: ethers.providers.JsonRpcProvider,
+    swapLayerAddress: string,
+    db: Knex | null,
+    chain: Chain
+  ) {
     this.provider = provider;
     this.swapLayerAddress = swapLayerAddress;
     this.swapLayerInterface = new ethers.utils.Interface([
       'event Redeemed(address indexed recipient, address outputToken, uint256 outputAmount, uint256 relayingFee)',
     ]);
+
+    if (db) {
+      this.tokenInfoManager = new TokenInfoManager(db, chainToChainId(chain), provider);
+    }
   }
 
   async parseSwapLayerTransaction(
@@ -59,6 +72,11 @@ class SwapLayerParser {
 
     if (!swapEvent) return null;
 
+    // if we have the tokenInfoManager inited, persist the token info
+    // this ensures we have the token decimal and name for analytics purposes
+    if (this.tokenInfoManager)
+      await this.tokenInfoManager.saveTokenInfoIfNotExist(swapEvent.args.outputToken);
+
     return {
       tx_hash: txHash,
       recipient: swapEvent.args.recipient,

watcher/src/fastTransfer/types.ts

@@ -156,3 +156,11 @@ export type TransferCompletion = {
   // on Solana Swap Layer, this acts as a link between complete_{transfer, swap}_payload and release_inbound
   staged_inbound?: string;
 };
+
+export type TokenInfo = {
+  name: string;
+  chain_id: number;
+  decimals: number;
+  symbol: string;
+  token_address: string;
+};

watcher/src/utils/TokenInfoManager.ts

@@ -0,0 +1,160 @@
+import { Knex } from 'knex';
+import { ethers, providers } from 'ethers';
+import { ChainId, chainIdToChain } from '@wormhole-foundation/sdk-base';
+import { Connection, PublicKey } from '@solana/web3.js';
+import { TokenInfo } from 'src/fastTransfer/types';
+
+const minABI = [
+  {
+    inputs: [],
+    name: 'name',
+    outputs: [{ internalType: 'string', name: '', type: 'string' }],
+    stateMutability: 'view',
+    type: 'function',
+  },
+  {
+    inputs: [],
+    name: 'symbol',
+    outputs: [{ internalType: 'string', name: '', type: 'string' }],
+    stateMutability: 'view',
+    type: 'function',
+  },
+  {
+    inputs: [],
+    name: 'decimals',
+    outputs: [{ internalType: 'uint8', name: '', type: 'uint8' }],
+    stateMutability: 'view',
+    type: 'function',
+  },
+];
+// TokenInfoManager class for managing token information across different chains
+// This class is to ensure that token info (e.g. decimal, name) for tokens that we see on Swap Layer is persisted for analytics purposes
+export class TokenInfoManager {
+  private tokenInfoMap: Map<string, TokenInfo>;
+  private db: Knex;
+  private chainId: ChainId;
+  private provider: providers.JsonRpcProvider | Connection;
+
+  constructor(db: Knex, chainId: ChainId, provider: providers.JsonRpcProvider | Connection) {
+    this.tokenInfoMap = new Map();
+    this.db = db;
+    this.chainId = chainId;
+    this.provider = provider;
+  }
+
+  // Retrieve token information from the database
+  private async getTokenInfoFromDB(tokenAddress: string): Promise<TokenInfo | null> {
+    return await this.db('token_infos')
+      .select('token_address', 'name', 'symbol', 'decimals')
+      .where('token_address', tokenAddress)
+      .andWhere('chain_id', this.chainId)
+      .first();
+  }
+
+  private async saveTokenInfo(tokenAddress: string, tokenInfo: TokenInfo): Promise<void> {
+    await this.db('token_infos')
+      .insert({
+        token_address: tokenAddress,
+        name: tokenInfo.name,
+        symbol: tokenInfo.symbol,
+        decimals: tokenInfo.decimals,
+        chain_id: this.chainId,
+      })
+      .onConflict(['token_address', 'chain_id'])
+      .merge();
+  }
+
+  // Save token information if it doesn't exist in the cache or database
+  public async saveTokenInfoIfNotExist(tokenAddress: string): Promise<TokenInfo | null> {
+    if (this.tokenInfoMap.has(tokenAddress)) {
+      return this.tokenInfoMap.get(tokenAddress) || null;
+    }
+    // Check if token info is in the database
+    const tokenInfo = await this.getTokenInfoFromDB(tokenAddress);
+    if (tokenInfo) {
+      this.tokenInfoMap.set(tokenAddress, tokenInfo);
+      return tokenInfo;
+    }
+    // If not in database, fetch from RPC
+    const fetchedTokenInfo = await this.fetchTokenInfoFromRPC(tokenAddress);
+    if (fetchedTokenInfo) {
+      await this.saveTokenInfo(tokenAddress, fetchedTokenInfo);
+      this.tokenInfoMap.set(tokenAddress, fetchedTokenInfo);
+      return fetchedTokenInfo;
+    }
+    return null;
+  }
+
+  // Fetch token information from RPC based on the chain ID
+  private async fetchTokenInfoFromRPC(tokenAddress: string): Promise<TokenInfo | null> {
+    if (chainIdToChain(this.chainId) === 'Solana') {
+      return this.fetchSolanaTokenInfo(tokenAddress);
+    }
+    return this.fetchEVMTokenInfo(tokenAddress);
+  }
+
+  // Fetch Solana token information
+  private async fetchSolanaTokenInfo(tokenAddress: string): Promise<TokenInfo | null> {
+    try {
+      const connection = this.provider as Connection;
+      const tokenPublicKey = new PublicKey(tokenAddress);
+      const accountInfo = await connection.getParsedAccountInfo(tokenPublicKey);
+
+      if (accountInfo.value && accountInfo.value.data && 'parsed' in accountInfo.value.data) {
+        const parsedData = accountInfo.value.data.parsed;
+        if (parsedData.type === 'mint' && 'info' in parsedData) {
+          const { name, symbol, decimals } = parsedData.info;
+          if (
+            typeof name === 'string' &&
+            typeof symbol === 'string' &&
+            typeof decimals === 'number'
+          ) {
+            return { name, symbol, decimals, chain_id: this.chainId, token_address: tokenAddress };
+          }
+        }
+      }
+      throw new Error('Invalid token account');
+    } catch (error) {
+      console.error('Error fetching Solana token info:', error);
+      return null;
+    }
+  }
+
+  // Fetch EVM token information
+  private async fetchEVMTokenInfo(tokenAddress: string): Promise<TokenInfo | null> {
+    // If it's null address, it's Ether or Wrapped Ether
+    if (tokenAddress.toLowerCase() === '0x0000000000000000000000000000000000000000') {
+      const { name, symbol } = this.getEtherInfo();
+      return {
+        name,
+        symbol,
+        decimals: 18,
+        chain_id: this.chainId,
+        token_address: tokenAddress,
+      };
+    }
+
+    const provider = this.provider as providers.JsonRpcProvider;
+    const tokenContract = new ethers.Contract(tokenAddress, minABI, provider);
+    try {
+      const name = await tokenContract.name();
+      const symbol = await tokenContract.symbol();
+      const decimals = await tokenContract.decimals();
+      return { name, symbol, decimals, chain_id: this.chainId, token_address: tokenAddress };
+    } catch (error) {
+      console.error('Error fetching EVM token info:', error, tokenAddress);
+      return null;
+    }
+  }
+
+  // Helper function to get Ether or Wrapped Ether info based on chain ID
+  private getEtherInfo(): { name: string; symbol: string } {
+    switch (this.chainId) {
+      case 2:
+      case 5:
+        return { name: 'Ether', symbol: 'ETH' };
+      default:
+        return { name: 'Wrapped Ether', symbol: 'WETH' };
+    }
+  }
+}

watcher/src/watchers/FTEVMWatcher.ts

@@ -42,27 +42,25 @@ export class FTEVMWatcher extends Watcher {
     this.provider = new ethers.providers.JsonRpcProvider(RPCS_BY_CHAIN[network][chain]);
     this.rpc = RPCS_BY_CHAIN[this.network][this.chain]!;
     this.tokenRouterParser = new TokenRouterParser(this.network, chain, this.provider);
+
+    // Initialize database connection before creating swap layer parser
+    if (!isTest) {
+      this.pg = knex({
+        client: 'pg',
+        connection: {
+          user: assertEnvironmentVariable('PG_FT_USER'),
+          password: assertEnvironmentVariable('PG_FT_PASSWORD'),
+          database: assertEnvironmentVariable('PG_FT_DATABASE'),
+          host: assertEnvironmentVariable('PG_FT_HOST'),
+          port: Number(assertEnvironmentVariable('PG_FT_PORT')),
+        },
+      });
+    }
+
     this.swapLayerParser = this.swapLayerAddress
-      ? new SwapLayerParser(this.provider, this.swapLayerAddress)
+      ? new SwapLayerParser(this.provider, this.swapLayerAddress, this.pg, chain)
       : null;
     this.logger.debug('FTWatcher', network, chain, finalizedBlockTag);
-    // hacky way to not connect to the db in tests
-    // this is to allow ci to run without a db
-    if (isTest) {
-      // Components needed for testing is complete
-      return;
-    }
-
-    this.pg = knex({
-      client: 'pg',
-      connection: {
-        user: assertEnvironmentVariable('PG_FT_USER'),
-        password: assertEnvironmentVariable('PG_FT_PASSWORD'),
-        database: assertEnvironmentVariable('PG_FT_DATABASE'),
-        host: assertEnvironmentVariable('PG_FT_HOST'),
-        port: Number(assertEnvironmentVariable('PG_FT_PORT')),
-      },
-    });
   }
 
   async getBlock(blockNumberOrTag: number | BlockTag): Promise<Block> {

watcher/src/watchers/__tests__/FTEVMWatcher.test.ts

@@ -74,7 +74,7 @@ describe('SwapLayerParser', () => {
 
   beforeEach(() => {
     mockProvider = new MockJsonRpcProvider();
-    parser = new SwapLayerParser(mockProvider, swapLayerAddress);
+    parser = new SwapLayerParser(mockProvider, swapLayerAddress, null, 'ArbitrumSepolia');
   });
 
   it('should parse a swap layer transaction correctly', async () => {