ft_watcher: worker to invoke ft

Signed-off-by: bingyuyap <bingyu.yap.21@gmail.com>

Author: bingyuyap

common/src/consts.ts

@@ -7,7 +7,7 @@ import {
   toChainId,
 } from '@wormhole-foundation/sdk-base';
 
-export type Mode = 'vaa' | 'ntt';
+export type Mode = 'vaa' | 'ntt' | 'ft';
 
 // This is defined here in an effort to keep the number and text in sync.
 // The default value is not exported because the getMissThreshold() function should be used to get the value.
@@ -17,9 +17,9 @@ export const MISS_THRESHOLD_LABEL = '40 minutes';
 export const MAX_VAA_DECIMALS = 8;
 export const VAA_VERSION = 1;
 
-export const INITIAL_DEPLOYMENT_BLOCK_BY_NETWORK_AND_CHAIN: {
-  [key in Network]: { [key in Chain]?: string };
-} = {
+type NetworkChainBlockMapping = { [key in Network]: { [key in Chain]?: string } };
+
+export const INITIAL_DEPLOYMENT_BLOCK_BY_NETWORK_AND_CHAIN: NetworkChainBlockMapping = {
   ['Mainnet']: {
     Ethereum: '12959638',
     Terra: '4810000', // not sure exactly but this should be before the first known message
@@ -84,9 +84,7 @@ export const INITIAL_DEPLOYMENT_BLOCK_BY_NETWORK_AND_CHAIN: {
   ['Devnet']: {},
 };
 
-export const INITIAL_NTT_DEPLOYMENT_BLOCK_BY_NETWORK_AND_CHAIN: {
-  [key in Network]: { [key in Chain]?: string };
-} = {
+export const INITIAL_NTT_DEPLOYMENT_BLOCK_BY_NETWORK_AND_CHAIN: NetworkChainBlockMapping = {
   ['Mainnet']: {
     Solana: '260508723',
     Ethereum: '19583505',
@@ -105,6 +103,23 @@ export const INITIAL_NTT_DEPLOYMENT_BLOCK_BY_NETWORK_AND_CHAIN: {
   ['Devnet']: {},
 };
 
+export const INITIAL_FT_DEPLOYMENT_BLOCK_BY_NETWORK_AND_CHAIN: NetworkChainBlockMapping = {
+  ['Mainnet']: {},
+  ['Testnet']: {
+    Solana: '302162456',
+    ArbitrumSepolia: '49505590',
+  },
+  ['Devnet']: {},
+};
+
+export const INITIAL_DEPLOYMENT_BLOCK_BY_MODE: {
+  [mode in Mode]: NetworkChainBlockMapping;
+} = {
+  vaa: INITIAL_DEPLOYMENT_BLOCK_BY_NETWORK_AND_CHAIN,
+  ntt: INITIAL_NTT_DEPLOYMENT_BLOCK_BY_NETWORK_AND_CHAIN,
+  ft: INITIAL_FT_DEPLOYMENT_BLOCK_BY_NETWORK_AND_CHAIN,
+};
+
 export function getMissThreshold(date: Date, chainish: number | string | Chain | ChainId): string {
   // We would like chainish to really be a ChainId.
   let missThresholdInMins: number;

common/src/utils.ts

@@ -36,7 +36,7 @@ export function getNetwork(): Network {
 
 export function getMode(): Mode {
   const mode: string = assertEnvironmentVariable('MODE').toLowerCase();
-  if (mode === 'vaa' || mode === 'ntt') {
+  if (mode === 'vaa' || mode === 'ntt' || mode === 'ft') {
     return mode;
   }
   throw new Error(`Unknown mode: ${mode}`);

database/fast-transfer-schema.sql

@@ -14,8 +14,11 @@ CREATE TYPE FastTransferStatus AS ENUM ('pending', 'no_offer', 'executed', 'sett
 -- Market Order tracks events of when fast market orders are
 -- placed in the token router
 CREATE TABLE market_orders (
-  fast_vaa_id VARCHAR(255) PRIMARY KEY,
-  fast_vaa_hash VARCHAR(255),
+  -- These two cant be primary key because on different stages they might initially be null due to the inability to find them
+  -- To accomodate this we put a unique constraint to allow nullable
+  -- It will eventually be filled up 
+  fast_vaa_id VARCHAR(255) UNIQUE,
+  fast_vaa_hash VARCHAR(255) UNIQUE,
   amount_in BIGINT,
   min_amount_out BIGINT,
   src_chain INTEGER,
@@ -89,4 +92,3 @@ CREATE TABLE auction_history_mapping (
   auction_pubkey VARCHAR(255) PRIMARY KEY,
   index INT NOT NULL
 );
-

watcher/scripts/backfill.ts

@@ -30,7 +30,7 @@ import { ChainId, toChain } from '@wormhole-foundation/sdk-base';
   const lastBlockEntries = Object.entries(localDb.lastBlockByChain);
   for (const [chain, blockKey] of lastBlockEntries) {
     console.log('backfilling last block for', chain, blockKey);
-    await remoteDb.storeLatestBlock(toChain(Number(chain) as ChainId), blockKey, false);
+    await remoteDb.storeLatestBlock(toChain(Number(chain) as ChainId), blockKey, 'vaa');
     await sleep(500);
   }
 })();

watcher/scripts/backfillNear.ts

@@ -35,7 +35,7 @@ const BATCH_SIZE = 100;
   const chain: Chain = 'Near';
   const provider = await getNearProvider(network, NEAR_ARCHIVE_RPC);
   const fromBlock = Number(
-    (await db.getLastBlockByChain(chain, false)) ??
+    (await db.getLastBlockByChain(chain, 'vaa')) ??
       INITIAL_DEPLOYMENT_BLOCK_BY_NETWORK_AND_CHAIN[network][chain] ??
       0
   );

watcher/src/databases/BigtableDatabase.ts

@@ -2,6 +2,7 @@ import { Bigtable } from '@google-cloud/bigtable';
 import { PubSub } from '@google-cloud/pubsub';
 import { Chain, chainToChainId } from '@wormhole-foundation/sdk-base';
 import {
+  Mode,
   assertEnvironmentVariable,
   chunkArray,
   sleep,
@@ -38,16 +39,25 @@ export class BigtableDatabase extends Database {
   firestoreDb: FirebaseFirestore.Firestore;
   latestCollectionName: string;
   latestNTTCollectionName: string;
+  latestFTCollectionName: string;
   pubsubSignedVAATopic: string;
+  collectionNameByMode: { [key in Mode]: string };
   pubsub: PubSub;
   constructor() {
     super();
     this.msgTableId = assertEnvironmentVariable('BIGTABLE_TABLE_ID');
     this.signedVAAsTableId = assertEnvironmentVariable('BIGTABLE_SIGNED_VAAS_TABLE_ID');
     this.vaasByTxHashTableId = assertEnvironmentVariable('BIGTABLE_VAAS_BY_TX_HASH_TABLE_ID');
     this.instanceId = assertEnvironmentVariable('BIGTABLE_INSTANCE_ID');
+    // TODO: make these const?
     this.latestCollectionName = assertEnvironmentVariable('FIRESTORE_LATEST_COLLECTION');
     this.latestNTTCollectionName = assertEnvironmentVariable('FIRESTORE_LATEST_NTT_COLLECTION');
+    this.latestFTCollectionName = assertEnvironmentVariable('FIRESTORE_LATEST_FT_COLLECTION');
+    this.collectionNameByMode = {
+      vaa: this.latestCollectionName,
+      ntt: this.latestNTTCollectionName,
+      ft: this.latestFTCollectionName,
+    };
     this.pubsubSignedVAATopic = assertEnvironmentVariable('PUBSUB_SIGNED_VAA_TOPIC');
     try {
       this.bigtable = new Bigtable();
@@ -63,11 +73,15 @@ export class BigtableDatabase extends Database {
     }
   }
 
-  async getLastBlockByChain(chain: Chain, isNTT: boolean): Promise<string | null> {
+  async getLastBlockByChain(chain: Chain, mode: Mode): Promise<string | null> {
     const chainId = chainToChainId(chain);
-    const lastObservedBlock = isNTT
-      ? this.firestoreDb.collection(this.latestNTTCollectionName).doc(chainId.toString())
-      : this.firestoreDb.collection(this.latestCollectionName).doc(chainId.toString());
+
+    const collectionName = this.collectionNameByMode[mode];
+    if (!collectionName) {
+      throw new Error(`Unknown mode: ${mode}`);
+    }
+
+    const lastObservedBlock = this.firestoreDb.collection(collectionName).doc(chainId.toString());
     const lastObservedBlockByChain = await lastObservedBlock.get();
     const blockKeyData = lastObservedBlockByChain.data();
     const lastBlockKey = blockKeyData?.lastBlockKey;
@@ -79,16 +93,19 @@ export class BigtableDatabase extends Database {
     return null;
   }
 
-  async storeLatestBlock(chain: Chain, lastBlockKey: string, isNTT: boolean): Promise<void> {
+  async storeLatestBlock(chain: Chain, lastBlockKey: string, mode: Mode): Promise<void> {
     if (this.firestoreDb === undefined) {
       this.logger.error('no firestore db set');
       return;
     }
     const chainId = chainToChainId(chain);
     this.logger.info(`storing last block=${lastBlockKey} for chain=${chainId}`);
-    const lastObservedBlock = isNTT
-      ? this.firestoreDb.collection(this.latestNTTCollectionName).doc(`${chainId.toString()}`)
-      : this.firestoreDb.collection(this.latestCollectionName).doc(`${chainId.toString()}`);
+    const collectionName = this.collectionNameByMode[mode];
+    if (!collectionName) {
+      throw new Error(`Unknown mode: ${mode}`);
+    }
+
+    const lastObservedBlock = this.firestoreDb.collection(collectionName).doc(chainId.toString());
     await lastObservedBlock.set({ lastBlockKey });
   }
 
@@ -159,7 +176,7 @@ export class BigtableDatabase extends Database {
       if (blockKeys.length) {
         const lastBlockKey = blockKeys[blockKeys.length - 1];
         this.logger.info(`for chain=${chain}, storing last bigtable block=${lastBlockKey}`);
-        await this.storeLatestBlock(chain, lastBlockKey, false);
+        await this.storeLatestBlock(chain, lastBlockKey, 'vaa');
       }
     }
   }

watcher/src/databases/Database.ts

@@ -1,3 +1,4 @@
+import { Mode } from '@wormhole-foundation/wormhole-monitor-common';
 import { getLogger, WormholeLogger } from '../utils/logger';
 import { VaasByBlock } from './types';
 import { Chain } from '@wormhole-foundation/sdk-base';
@@ -14,13 +15,13 @@ export class Database {
     }
     return filteredVaasByBlock;
   }
-  async getLastBlockByChain(chain: Chain, isNTT: boolean): Promise<string | null> {
+  async getLastBlockByChain(chain: Chain, mode: Mode): Promise<string | null> {
     throw new Error('Not Implemented');
   }
   async storeVaasByBlock(chain: Chain, vaasByBlock: VaasByBlock): Promise<void> {
     throw new Error('Not Implemented');
   }
-  async storeLatestBlock(chain: Chain, lastBlockKey: string, isNTT: boolean): Promise<void> {
+  async storeLatestBlock(chain: Chain, lastBlockKey: string, mode: Mode): Promise<void> {
     throw new Error('Not Implemented');
   }
 }

watcher/src/databases/JsonDatabase.ts

@@ -3,6 +3,7 @@ import { DB_LAST_BLOCK_FILE, JSON_DB_FILE } from '../consts';
 import { Database } from './Database';
 import { DB, LastBlockByChain, VaasByBlock } from './types';
 import { Chain, chainToChainId } from '@wormhole-foundation/sdk-base';
+import { Mode } from '@wormhole-foundation/wormhole-monitor-common';
 
 const ENCODING = 'utf8';
 export class JsonDatabase extends Database {
@@ -34,7 +35,7 @@ export class JsonDatabase extends Database {
     }
   }
 
-  async getLastBlockByChain(chain: Chain): Promise<string | null> {
+  async getLastBlockByChain(chain: Chain, mode: Mode): Promise<string | null> {
     const chainId = chainToChainId(chain);
     const blockInfo = this.lastBlockByChain[chainId];
     if (blockInfo) {

watcher/src/databases/__tests__/utils.test.ts

@@ -10,11 +10,11 @@ test('getResumeBlockByChain', async () => {
   const blockKey = makeBlockKey(fauxBlock, new Date().toISOString());
   db.lastBlockByChain = { [chainToChainId('Solana')]: blockKey };
   // if a chain is in the database, that number should be returned
-  expect(await db.getLastBlockByChain('Solana')).toEqual(fauxBlock);
-  expect(await getResumeBlockByChain('Mainnet', 'Solana', false)).toEqual(Number(fauxBlock) + 1);
+  expect(await db.getLastBlockByChain('Solana', 'vaa')).toEqual(fauxBlock);
+  expect(await getResumeBlockByChain('Mainnet', 'Solana', 'vaa')).toEqual(Number(fauxBlock) + 1);
   // if a chain is not in the database, the initial deployment block should be returned
   expect(INITIAL_DEPLOYMENT_BLOCK_BY_NETWORK_AND_CHAIN['Mainnet'].Moonbeam).toBeDefined();
-  expect(await getResumeBlockByChain('Mainnet', 'Moonbeam', false)).toEqual(
+  expect(await getResumeBlockByChain('Mainnet', 'Moonbeam', 'vaa')).toEqual(
     Number(INITIAL_DEPLOYMENT_BLOCK_BY_NETWORK_AND_CHAIN['Mainnet'].Moonbeam)
   );
 });

watcher/src/databases/utils.ts

@@ -1,8 +1,8 @@
 import { Chain, Network, chainToChainId } from '@wormhole-foundation/sdk-base';
 import {
-  INITIAL_DEPLOYMENT_BLOCK_BY_NETWORK_AND_CHAIN,
-  INITIAL_NTT_DEPLOYMENT_BLOCK_BY_NETWORK_AND_CHAIN,
+  INITIAL_DEPLOYMENT_BLOCK_BY_MODE,
   MAX_UINT_64,
+  Mode,
   padUint16,
   padUint64,
 } from '@wormhole-foundation/wormhole-monitor-common';
@@ -52,20 +52,18 @@ export const initDb = (startWatching: boolean = true): Database => {
 export const storeLatestBlock = async (
   chain: Chain,
   lastBlockKey: string,
-  isNTT: boolean
+  mode: Mode
 ): Promise<void> => {
-  return database.storeLatestBlock(chain, lastBlockKey, isNTT);
+  return database.storeLatestBlock(chain, lastBlockKey, mode);
 };
 
 export const getResumeBlockByChain = async (
   network: Network,
   chain: Chain,
-  isNTT: boolean
+  mode: Mode
 ): Promise<number | null> => {
-  const lastBlock = await database.getLastBlockByChain(chain, isNTT);
-  const initialBlock = isNTT
-    ? INITIAL_NTT_DEPLOYMENT_BLOCK_BY_NETWORK_AND_CHAIN[network][chain]
-    : INITIAL_DEPLOYMENT_BLOCK_BY_NETWORK_AND_CHAIN[network][chain];
+  const lastBlock = await database.getLastBlockByChain(chain, mode);
+  const initialBlock = INITIAL_DEPLOYMENT_BLOCK_BY_MODE[mode][network][chain];
   return lastBlock !== null
     ? Number(lastBlock) + 1
     : initialBlock !== undefined

watcher/src/index.ts

@@ -2,7 +2,6 @@ import * as dotenv from 'dotenv';
 dotenv.config();
 
 import { initDb } from './databases/utils';
-import { makeFinalizedNTTWatcher, makeFinalizedWatcher } from './watchers/utils';
 import { Mode, getNetwork, getMode } from '@wormhole-foundation/wormhole-monitor-common';
 import { startSupervisor } from './workers/supervisor';
 import { Chain, Network } from '@wormhole-foundation/sdk-base';
@@ -91,10 +90,14 @@ const supportedNTTChains: Chain[] =
     ? ['Solana', 'Sepolia', 'ArbitrumSepolia', 'BaseSepolia', 'OptimismSepolia']
     : ['Solana', 'Ethereum', 'Fantom', 'Arbitrum', 'Optimism', 'Base'];
 
+const supportedFTChains: Chain[] = network === 'Testnet' ? ['Solana', 'ArbitrumSepolia'] : [];
+
 if (mode === 'vaa') {
   startSupervisor(supportedChains);
 } else if (mode === 'ntt') {
   startSupervisor(supportedNTTChains);
+} else if (mode === 'ft') {
+  startSupervisor(supportedFTChains);
 } else {
   throw new Error(`Unknown mode: ${mode}`);
 }