add mongo db mock and infra (#633)

event-watcher/package.json

@@ -21,8 +21,9 @@
     "read-firestore": "ts-node scripts/readFirestore.ts"
   },
   "dependencies": {
-    "@certusone/wormhole-sdk": "^0.9.22",
     "@celo-tools/celo-ethers-wrapper": "^0.3.0",
+    "@certusone/wormhole-sdk": "^0.9.22",
+    "@fastify/swagger": "^8.8.0",
     "@google-cloud/bigtable": "^4.1.0",
     "@google-cloud/pubsub": "^3.4.1",
     "@mysten/sui.js": "^0.33.0",
@@ -32,8 +33,10 @@
     "axios": "^1.2.1",
     "bs58": "^5.0.0",
     "dotenv": "^16.0.3",
+    "fastify": "^4.21.0",
     "firebase-admin": "^11.4.0",
     "js-sha512": "^0.8.0",
+    "mongodb": "^5.7.0",
     "near-api-js": "^1.1.0",
     "ora": "^5.4.1",
     "winston": "^3.8.2",

event-watcher/src/builder/server.ts

@@ -0,0 +1,26 @@
+import Fastify, { FastifyInstance } from 'fastify'
+
+
+export const createServer = async (port: number): Promise<FastifyInstance> => {
+    const server = Fastify({ logger: true })
+    await server.register(require('@fastify/swagger'), {
+        swagger: {
+            info: {
+                title: 'VAA Payload Parser',
+                description: 'API allows the parsing of VAA with a custom parser depending on the application that originated the VAA',
+                version: '0.0.1'
+            },
+            externalDocs: {
+                url: 'https://swagger.io',
+            },
+            host: `localhost:${port}`,
+            schemes: ['http'],
+            consumes: ['application/json'],
+            produces: ['application/json'],
+
+        }
+    })
+
+
+    return server
+}

event-watcher/src/databases/MongoDB.ts

@@ -0,0 +1,80 @@
+import { ChainName, coalesceChainId } from '@certusone/wormhole-sdk/lib/cjs/utils/consts';
+import { readFileSync, writeFileSync } from 'fs';
+import { DB_LAST_BLOCK_FILE, JSON_DB_FILE } from '../consts';
+import { Database } from './Database';
+import { DB, LastBlockByChain, VaasByBlock } from './types';
+import * as mongoDB from "mongodb";
+import { SequenceNumber } from '@mysten/sui.js';
+
+export const collections: { wormholeTx?: mongoDB.Collection } = {}
+
+const ENCODING = 'utf8';
+export class MongoDatabase extends Database {
+  lastBlockByChain: LastBlockByChain;
+  dbFile: string;
+  dbLastBlockFile: string;
+  client: mongoDB.MongoClient;
+  db: mongoDB.Db;
+  wormholeTx: mongoDB.Collection;
+  constructor() {
+    super();
+    this.client = new mongoDB.MongoClient("mongodb://localhost:27017");
+    this.client.connect();
+    this.db  = this.client.db("wormhole");
+    this.wormholeTx = this.db.collection("wormholeTx");
+ 
+   // collections.games = gamesCollection;
+       
+    console.log(`Successfully connected to database: ${this.db.databaseName} `);
+    //this.db = client.db("wormhole");
+    this.lastBlockByChain = {};
+
+    if (!process.env.DB_LAST_BLOCK_FILE) {
+      this.logger.info(`no db file set, using default path=${DB_LAST_BLOCK_FILE}`);
+    }
+    this.dbFile = JSON_DB_FILE;
+    this.dbLastBlockFile = DB_LAST_BLOCK_FILE;
+
+    try {
+      const rawLast = readFileSync(this.dbLastBlockFile, ENCODING);
+      this.lastBlockByChain = JSON.parse(rawLast);
+    } catch (e) {
+      this.logger.warn('Failed to load DB, initiating a fresh one.');
+    }
+  }
+
+  async getLastBlockByChain(chain: ChainName): Promise<string | null> {
+    const chainId = coalesceChainId(chain);
+    const blockInfo = this.lastBlockByChain[chainId];
+    if (blockInfo) {
+      const tokens = blockInfo.split('/');
+      return chain === 'aptos' ? tokens.at(-1)! : tokens[0];
+    }
+    return null;
+  }
+  async storeVaasByBlock(chain: ChainName, vaasByBlock: VaasByBlock): Promise<void> {
+    const chainId = coalesceChainId(chain);
+    const filteredVaasByBlock = Database.filterEmptyBlocks(vaasByBlock);
+    if (Object.keys(filteredVaasByBlock).length) {
+    }
+
+    // this will always overwrite the "last" block, so take caution if manually backfilling gaps
+    const blockKeys = Object.keys(vaasByBlock).sort(
+      (bk1, bk2) => Number(bk1.split('/')[0]) - Number(bk2.split('/')[0])
+    );
+    if (blockKeys.length) {
+      this.lastBlockByChain[chainId] = blockKeys[blockKeys.length - 1];
+      this.wormholeTx.insertOne({chainId: chainId, block: this.lastBlockByChain[chainId], data: vaasByBlock});
+
+      //writeFileSync(this.dbLastBlockFile, JSON.stringify(this.lastBlockByChain), ENCODING);
+    }
+  }
+
+  async storeVaa(chain: ChainName, txHash: string, vaa_id:string, payload: string): Promise<void> {
+    const chainId = coalesceChainId(chain);
+    this.wormholeTx.insertOne({chainId: chainId, txHash: txHash, vaa_id: vaa_id, payload: payload});
+  }
+
+
+}
+

event-watcher/src/databases/utils.ts

@@ -10,6 +10,7 @@ import { BigtableDatabase } from './BigtableDatabase';
 import { Database } from './Database';
 import { JsonDatabase } from './JsonDatabase';
 import { VaasByBlock } from './types';
+import { MongoDatabase } from './MongoDB';
 
 // Bigtable Message ID format
 // chain/MAX_UINT64-block/emitter/sequence
@@ -64,6 +65,8 @@ export const initDb = (): Database => {
   if (DB_SOURCE === 'bigtable') {
     database = new BigtableDatabase();
     (database as BigtableDatabase).watchMissing();
+  } else if (DB_SOURCE === 'mongo') {
+    database = new MongoDatabase();
   } else {
     database = new JsonDatabase();
   }

event-watcher/src/index.ts

@@ -4,33 +4,55 @@ dotenv.config();
 import { ChainName } from '@certusone/wormhole-sdk/lib/cjs/utils/consts';
 import { initDb } from './databases/utils';
 import { makeFinalizedWatcher } from './watchers/utils';
+import { InfrastructureController } from "./infrastructure/infrastructure.controller";
+import { createServer } from "./builder/server";
+
 
 initDb();
 
+const infrastructureController = new InfrastructureController();
+
+const startServer = async () => {
+  const port = Number(process.env.PORT) || 3005;
+  const server =  await createServer(port);
+
+  server.get("/ready", { logLevel: "silent" }, infrastructureController.ready);
+  server.get("/health",{ logLevel: "silent" }, infrastructureController.health);
+
+  server.listen({ port, host: "0.0.0.0" }, (err: any, address: any) => {
+    if (err) {
+      process.exit(1);
+    }
+    console.log(`Server listening at ${address}`);
+  });
+}
+
+startServer();
+
 const supportedChains: ChainName[] = [
-  // 'solana',
-  // 'ethereum',
-  //'bsc',
-   'polygon',
-  // 'avalanche',
-  // 'oasis',
-  // 'algorand',
-  // 'fantom',
-  // 'karura',
-  // 'acala',
-  // 'klaytn',
-  // 'celo',
-  // 'moonbeam',
-  // 'arbitrum',
-  // 'optimism',
-  // 'aptos',
-  // 'near',
-  // 'terra2',
-  // 'terra',
-  // 'xpla',
-  // 'injective',
-  // 'sui',
-  // 'base',
+  'solana',
+  'ethereum',
+  'bsc',
+  'polygon',
+  'avalanche',
+  'oasis',
+  'algorand',
+  'fantom',
+  'karura',
+  'acala',
+  'klaytn',
+  'celo',
+  'moonbeam',
+  'arbitrum',
+  'optimism',
+  'aptos',
+  'near',
+  'terra2',
+  'terra',
+  'xpla',
+  'injective',
+  'sui',
+  'base',
 ];
 
 for (const chain of supportedChains) {

event-watcher/src/infrastructure/infrastructure.controller.ts

@@ -0,0 +1,13 @@
+import { FastifyReply, FastifyRequest } from "fastify";
+
+export class InfrastructureController {
+
+    ready = async (_: FastifyRequest, reply: FastifyReply) => {
+        return reply.code(200).send({ status: "OK" })
+    }
+
+    health = async (_: FastifyRequest, reply: FastifyReply) => {
+        return reply.code(200).send({ status: "OK" })
+    }
+
+}