watcher: add NTT watcher

common/src/consts.ts

@@ -101,6 +101,20 @@ export const INITIAL_DEPLOYMENT_BLOCK_BY_NETWORK_AND_CHAIN: {
   ['devnet']: {},
 };
 
+export const INITIAL_NTT_DEPLOYMENT_BLOCK_BY_NETWORK_AND_CHAIN: {
+  [key in Environment]: { [key in ChainName]?: string };
+} = {
+  ['mainnet']: {},
+  ['testnet']: {
+    solana: '284788472',
+    sepolia: '5472203',
+    arbitrum_sepolia: '22501243',
+    base_sepolia: '7249669',
+    optimism_sepolia: '9232548',
+  },
+  ['devnet']: {},
+};
+
 export const TOKEN_BRIDGE_EMITTERS: { [key in ChainName]?: string } = {
   solana: 'ec7372995d5cc8732397fb0ad35c0121e0eaa90d26f828a534cab54391b3a4f5',
   ethereum: '0000000000000000000000003ee18b2214aff97000d974cf647e7c347e8fa585',
@@ -177,6 +191,15 @@ export const CIRCLE_DOMAIN_TO_CHAIN_ID: { [key: number]: ChainId } = {
   6: CHAIN_ID_BASE,
   7: CHAIN_ID_POLYGON,
 };
+
+// TODO: This should be needed by processVaa.ts, if we go down that path
+export const NTT_EMITTERS: { [key in ChainName]?: string } = {
+  // TODO: add NTT emitters
+};
+
+export const isNTTEmitter = (chain: ChainId | ChainName, emitter: string) =>
+  NTT_EMITTERS[coalesceChainName(chain)]?.toLowerCase() === emitter.toLowerCase();
+
 export type CHAIN_INFO = {
   name: string;
   evm: boolean;

database/ntt-lifecycle-schema.sql

@@ -0,0 +1,18 @@
+
+CREATE TABLE life_cycle (
+  from_chain INTEGER,
+  to_chain INTEGER,
+  from_token VARCHAR(96),
+  token_amount DECIMAL(78, 0),
+  transfer_sent_txhash VARCHAR(96),
+  redeemed_txhash VARCHAR(96),
+  ntt_transfer_key VARCHAR(256),
+  vaa_id VARCHAR(128),
+  digest VARCHAR(96) NOT NULL,
+  transfer_time TIMESTAMP,
+  redeem_time TIMESTAMP,
+  inbound_transfer_queued_time TIMESTAMP,
+  outbound_transfer_queued_time TIMESTAMP,
+  outbound_transfer_rate_limited_time TIMESTAMP,
+	PRIMARY KEY (digest)
+);

watcher/package.json

@@ -32,6 +32,7 @@
     "@google-cloud/pubsub": "^3.4.1",
     "@mysten/sui.js": "^0.33.0",
     "@solana/web3.js": "^1.73.0",
+    "@types/bn.js": "^5.1.5",
     "@wormhole-foundation/wormhole-monitor-common": "^0.0.1",
     "algosdk": "^2.4.0",
     "aptos": "^1.4.0",

watcher/scripts/backfill.ts

@@ -33,7 +33,7 @@ import { VaasByBlock } from '../src/databases/types';
   const lastBlockEntries = Object.entries(localDb.lastBlockByChain);
   for (const [chain, blockKey] of lastBlockEntries) {
     console.log('backfilling last block for', chain, blockKey);
-    await remoteDb.storeLatestBlock(coalesceChainName(Number(chain) as ChainId), blockKey);
+    await remoteDb.storeLatestBlock(coalesceChainName(Number(chain) as ChainId), blockKey, false);
     await sleep(500);
   }
 })();

watcher/scripts/backfillNear.ts

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

watcher/src/NTTConsts.ts

@@ -0,0 +1,172 @@
+import { ChainName, keccak256 } from '@certusone/wormhole-sdk';
+import { Environment } from '@wormhole-foundation/wormhole-monitor-common';
+import { NativeTokenTransfer, NttManagerMessage } from './watchers/NTTPayloads';
+
+//
+// The following are from IRateLimiterEvents.sol
+//
+
+/// @notice Emitted when an inbound transfer is queued
+/// @dev Topic0
+///      0x7f63c9251d82a933210c2b6d0b0f116252c3c116788120e64e8e8215df6f3162.
+/// @param digest The digest of the message.
+/// event InboundTransferQueued(bytes32 digest);
+export const InboundTransferQueuedTopic =
+  '0x7f63c9251d82a933210c2b6d0b0f116252c3c116788120e64e8e8215df6f3162';
+
+/// @notice Emitted when an outbound transfer is queued.
+/// @dev Topic0
+///      0x69add1952a6a6b9cb86f04d05f0cb605cbb469a50ae916139d34495a9991481f.
+/// @param queueSequence The location of the transfer in the queue.
+/// event OutboundTransferQueued(uint64 queueSequence);
+export const OutboundTransferQueuedTopic =
+  '0x69add1952a6a6b9cb86f04d05f0cb605cbb469a50ae916139d34495a9991481f';
+
+/// @notice Emitted when an outbound transfer is rate limited.
+/// @dev Topic0
+///      0x754d657d1363ee47d967b415652b739bfe96d5729ccf2f26625dcdbc147db68b.
+/// @param sender The initial sender of the transfer.
+/// @param amount The amount to be transferred.
+/// @param currentCapacity The capacity left for transfers within the 24-hour window.:w
+/// event OutboundTransferRateLimited( address indexed sender, uint64 sequence, uint256 amount, uint256 currentCapacity);
+export const OutboundTransferRateLimitedTopic =
+  '0x754d657d1363ee47d967b415652b739bfe96d5729ccf2f26625dcdbc147db68b';
+
+//
+// The following are from INttManagerEvents.sol
+//
+
+/// @notice Emitted when a message is sent from the nttManager.
+/// @dev Topic0
+///      0x9716fe52fe4e02cf924ae28f19f5748ef59877c6496041b986fbad3dae6a8ecf
+/// @param recipient The recipient of the message.
+/// @param amount The amount transferred.
+/// @param fee The amount of ether sent along with the tx to cover the delivery fee.
+/// @param recipientChain The chain ID of the recipient.
+/// @param msgSequence The unique sequence ID of the message.
+/// event TransferSent( bytes32 recipient, uint256 amount, uint256 fee, uint16 recipientChain, uint64 msgSequence);
+export const TransferSentTopic =
+  '0x9716fe52fe4e02cf924ae28f19f5748ef59877c6496041b986fbad3dae6a8ecf';
+
+/// @notice Emitted when the peer contract is updated.
+/// @dev Topic0
+///      0x51b8437a7e22240c473f4cbdb4ed3a4f4bf5a9e7b3c511d7cfe0197325735700.
+/// @param chainId_ The chain ID of the peer contract.
+/// @param oldPeerContract The old peer contract address.
+/// @param peerContract The new peer contract address.
+/// event PeerUpdated(uint16 indexed chainId_, bytes32 oldPeerContract, bytes32 peerContract);
+export const PeerUpdatedTopic =
+  '0x51b8437a7e22240c473f4cbdb4ed3a4f4bf5a9e7b3c511d7cfe0197325735700';
+
+/// @notice Emitted when a message has been attested to.
+/// @dev Topic0
+///      0x35a2101eaac94b493e0dfca061f9a7f087913fde8678e7cde0aca9897edba0e5.
+/// @param digest The digest of the message.
+/// @param transceiver The address of the transceiver.
+/// @param index The index of the transceiver in the bitmap.
+/// event MessageAttestedTo(bytes32 digest, address transceiver, uint8 index);
+export const MessageAttestedToTopic =
+  '0x35a2101eaac94b493e0dfca061f9a7f087913fde8678e7cde0aca9897edba0e5';
+
+/// @notice Emmitted when the threshold required transceivers is changed.
+/// @dev Topic0
+///      0x2a855b929b9a53c6fb5b5ed248b27e502b709c088e036a5aa17620c8fc5085a9.
+/// @param oldThreshold The old threshold.
+/// @param threshold The new threshold.
+/// event ThresholdChanged(uint8 oldThreshold, uint8 threshold);
+export const ThresholdChangedTopic =
+  '0x2a855b929b9a53c6fb5b5ed248b27e502b709c088e036a5aa17620c8fc5085a9';
+
+/// @notice Emitted when an transceiver is removed from the nttManager.
+/// @dev Topic0
+///      0xc6289e62021fd0421276d06677862d6b328d9764cdd4490ca5ac78b173f25883.
+/// @param transceiver The address of the transceiver.
+/// @param transceiversNum The current number of transceivers.
+/// @param threshold The current threshold of transceivers.
+/// event TransceiverAdded(address transceiver, uint256 transceiversNum, uint8 threshold);
+export const TransceiverAddedTopic =
+  '0xc6289e62021fd0421276d06677862d6b328d9764cdd4490ca5ac78b173f25883';
+
+/// @notice Emitted when an transceiver is removed from the nttManager.
+/// @dev Topic0
+///     0x638e631f34d9501a3ff0295873b29f50d0207b5400bf0e48b9b34719e6b1a39e.
+/// @param transceiver The address of the transceiver.
+/// @param threshold The current threshold of transceivers.
+/// event TransceiverRemoved(address transceiver, uint8 threshold);
+export const TransceiverRemovedTopic =
+  '0x638e631f34d9501a3ff0295873b29f50d0207b5400bf0e48b9b34719e6b1a39e';
+
+/// @notice Emitted when a message has already been executed to
+///         notify client of against retries.
+/// @dev Topic0
+///      0x4069dff8c9df7e38d2867c0910bd96fd61787695e5380281148c04932d02bef2.
+/// @param sourceNttManager The address of the source nttManager.
+/// @param msgHash The keccak-256 hash of the message.
+/// event MessageAlreadyExecuted(bytes32 indexed sourceNttManager, bytes32 indexed msgHash);
+export const MessageAlreadyExecutedTopic =
+  '0x4069dff8c9df7e38d2867c0910bd96fd61787695e5380281148c04932d02bef2';
+
+/// @notice Emitted when a transfer has been redeemed
+///         (either minted or unlocked on the recipient chain).
+/// @dev Topic0
+///      0x504e6efe18ab9eed10dc6501a417f5b12a2f7f2b1593aed9b89f9bce3cf29a91.
+/// @param Topic1
+///      digest The digest of the message.
+/// event TransferRedeemed(bytes32 indexed digest);
+export const TransferRedeemedTopic =
+  '0x504e6efe18ab9eed10dc6501a417f5b12a2f7f2b1593aed9b89f9bce3cf29a91';
+
+// All topics:
+export const NTT_TOPICS = [
+  InboundTransferQueuedTopic,
+  OutboundTransferQueuedTopic,
+  OutboundTransferRateLimitedTopic,
+  TransferSentTopic,
+  PeerUpdatedTopic,
+  MessageAttestedToTopic,
+  ThresholdChangedTopic,
+  TransceiverAddedTopic,
+  TransceiverRemovedTopic,
+  MessageAlreadyExecutedTopic,
+  TransferRedeemedTopic,
+];
+
+export const NTT_CONTRACT: { [key in Environment]: { [key in ChainName]?: string } } = {
+  ['mainnet']: {},
+  ['testnet']: {
+    solana: 'nTTh3bZ5Aer6xboWZe39RDEft4MeVxSQ8D1EYAVLZw9',
+    sepolia: '0xB231aD95f2301bc82eA44c515001F0F746D637e0',
+    arbitrum_sepolia: '0xEec94CD3083e067398256a79CcA7e740C5c8ef81',
+    base_sepolia: '0xB03b030b2f5B40819Df76467d67eD1C85Ff66fAD',
+    optimism_sepolia: '0x7f430D4e7939D994C0955A01FC75D9DE33F12D11',
+  },
+  ['devnet']: {},
+};
+
+export const getNttManagerMessageDigest = (
+  emitterChain: number,
+  message: NttManagerMessage<NativeTokenTransfer>
+): string => {
+  const chainIdBuffer = Buffer.alloc(2);
+  chainIdBuffer.writeUInt16BE(emitterChain);
+  const serialized = NttManagerMessage.serialize(message, NativeTokenTransfer.serialize);
+  const digest = keccak256(Buffer.concat([chainIdBuffer, serialized]));
+  return digest.toString('hex');
+};
+
+export type LifeCycle = {
+  srcChainId: number;
+  destChainId: number;
+  sourceToken: string;
+  tokenAmount: bigint;
+  transferSentTxhash: string;
+  redeemedTxhash: string;
+  nttTransferKey: string;
+  vaaId: string;
+  digest: string;
+  transferTime: string;
+  redeemTime: string;
+  inboundTransferQueuedTime: string;
+  outboundTransferQueuedTime: string;
+  outboundTransferRateLimitedTime: string;
+};

watcher/src/databases/BigtableDatabase.ts

@@ -36,6 +36,7 @@ export class BigtableDatabase extends Database {
   bigtable: Bigtable;
   firestoreDb: FirebaseFirestore.Firestore;
   latestCollectionName: string;
+  latestNTTCollectionName: string;
   pubsubSignedVAATopic: string;
   pubsub: PubSub;
   constructor() {
@@ -45,6 +46,7 @@ export class BigtableDatabase extends Database {
     this.vaasByTxHashTableId = assertEnvironmentVariable('BIGTABLE_VAAS_BY_TX_HASH_TABLE_ID');
     this.instanceId = assertEnvironmentVariable('BIGTABLE_INSTANCE_ID');
     this.latestCollectionName = assertEnvironmentVariable('FIRESTORE_LATEST_COLLECTION');
+    this.latestNTTCollectionName = assertEnvironmentVariable('FIRESTORE_LATEST_NTT_COLLECTION');
     this.pubsubSignedVAATopic = assertEnvironmentVariable('PUBSUB_SIGNED_VAA_TOPIC');
     try {
       this.bigtable = new Bigtable();
@@ -60,11 +62,11 @@ export class BigtableDatabase extends Database {
     }
   }
 
-  async getLastBlockByChain(chain: ChainName): Promise<string | null> {
+  async getLastBlockByChain(chain: ChainName, isNTT: boolean): Promise<string | null> {
     const chainId = coalesceChainId(chain);
-    const lastObservedBlock = this.firestoreDb
-      .collection(this.latestCollectionName)
-      .doc(chainId.toString());
+    const lastObservedBlock = isNTT
+      ? this.firestoreDb.collection(this.latestNTTCollectionName).doc(chainId.toString())
+      : this.firestoreDb.collection(this.latestCollectionName).doc(chainId.toString());
     const lastObservedBlockByChain = await lastObservedBlock.get();
     const blockKeyData = lastObservedBlockByChain.data();
     const lastBlockKey = blockKeyData?.lastBlockKey;
@@ -76,16 +78,16 @@ export class BigtableDatabase extends Database {
     return null;
   }
 
-  async storeLatestBlock(chain: ChainName, lastBlockKey: string): Promise<void> {
+  async storeLatestBlock(chain: ChainName, lastBlockKey: string, isNTT: boolean): Promise<void> {
     if (this.firestoreDb === undefined) {
       this.logger.error('no firestore db set');
       return;
     }
     const chainId = coalesceChainId(chain);
     this.logger.info(`storing last block=${lastBlockKey} for chain=${chainId}`);
-    const lastObservedBlock = this.firestoreDb
-      .collection(this.latestCollectionName)
-      .doc(`${chainId.toString()}`);
+    const lastObservedBlock = isNTT
+      ? this.firestoreDb.collection(this.latestNTTCollectionName).doc(`${chainId.toString()}`)
+      : this.firestoreDb.collection(this.latestCollectionName).doc(`${chainId.toString()}`);
     await lastObservedBlock.set({ lastBlockKey });
   }
 
@@ -156,7 +158,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);
+        await this.storeLatestBlock(chain, lastBlockKey, false);
       }
     }
   }

watcher/src/databases/Database.ts

@@ -14,10 +14,13 @@ export class Database {
     }
     return filteredVaasByBlock;
   }
-  async getLastBlockByChain(chain: ChainName): Promise<string | null> {
+  async getLastBlockByChain(chain: ChainName, isNTT: boolean): Promise<string | null> {
     throw new Error('Not Implemented');
   }
   async storeVaasByBlock(chain: ChainName, vaasByBlock: VaasByBlock): Promise<void> {
     throw new Error('Not Implemented');
   }
+  async storeLatestBlock(chain: ChainName, lastBlockKey: string, isNTT: boolean): Promise<void> {
+    throw new Error('Not Implemented');
+  }
 }

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

@@ -11,13 +11,13 @@ test('getResumeBlockByChain', async () => {
   db.lastBlockByChain = { [CHAIN_ID_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')).toEqual(Number(fauxBlock) + 1);
+  expect(await getResumeBlockByChain('mainnet', 'solana', false)).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')).toEqual(
+  expect(await getResumeBlockByChain('mainnet', 'moonbeam', false)).toEqual(
     Number(INITIAL_DEPLOYMENT_BLOCK_BY_NETWORK_AND_CHAIN['mainnet'].moonbeam)
   );
   // if neither, null should be returned
   expect(INITIAL_DEPLOYMENT_BLOCK_BY_NETWORK_AND_CHAIN['mainnet'].unset).toBeUndefined();
-  expect(await getResumeBlockByChain('mainnet', 'unset')).toEqual(null);
+  expect(await getResumeBlockByChain('mainnet', 'unset', false)).toEqual(null);
 });