Replace ethereum-burn-event-monitor with Corvette

bridge/src/index.ts

@@ -3,7 +3,7 @@ import { init } from "@sentry/node";
 import { KmsProvider } from "@planetarium/aws-kms-provider";
 
 import { IWrappedNCGMinter } from "./interfaces/wrapped-ncg-minter";
-import { EthereumBurnEventMonitor } from "./monitors/ethereum-burn-event-monitor";
+import { CorvetteEventMonitor } from "./monitors/corvette-monitor";
 import { NineChroniclesTransferredEventMonitor } from "./monitors/nine-chronicles-transferred-event-monitor";
 import { WrappedNCGMinter } from "./wrapped-ncg-minter";
 import { wNCGTokenAbi } from "./wrapped-ncg-token";
@@ -366,11 +366,9 @@ process.on("uncaughtException", console.error);
         ETHERSCAN_ROOT_URL,
         integration
     );
-    const ethereumBurnEventMonitor = new EthereumBurnEventMonitor(
-        web3,
+    const ethereumBurnEventMonitor = new CorvetteEventMonitor(
         wNCGToken,
-        await monitorStateStore.load("ethereum"),
-        CONFIRMATIONS
+        await monitorStateStore.load("ethereum")
     );
     ethereumBurnEventMonitor.attach(ethereumBurnEventObserver);
 

bridge/src/monitors/corvette-monitor.ts

@@ -0,0 +1,116 @@
+import { createServer } from "http";
+import axios from "axios";
+import { Monitor } from ".";
+import type { ContractDescription } from "../types/contract-description";
+import type { TransactionLocation } from "../types/transaction-location";
+
+const CORVETTE_EVENT_API_URL = "http://localhost:8000/";
+const CORVETTE_LISTEN_PORT = 4000;
+const BURN_ABI_SIGNATURE = "Burn(address,bytes32,uint256)";
+
+function delay(ms: number): Promise<void> {
+    return new Promise((resolve) => {
+        setTimeout(() => {
+            resolve();
+        }, ms);
+    });
+}
+
+export interface CorvetteEventResponse {
+    timestamp: string;
+    blockIndex: number;
+    logIndex: number;
+    blockHash: string;
+    transactionHash: string;
+    sourceAddress: string;
+    abiHash: string;
+    abiSignature: string;
+    args: {
+        named: Record<string, unknown>;
+        ordered: unknown[];
+    };
+}
+
+export class CorvetteEventMonitor extends Monitor<CorvetteEventResponse> {
+    private readonly webhookReceiver;
+
+    private eventDataQueue: Set<CorvetteEventResponse> = new Set();
+
+    constructor(
+        private readonly contractDescription: ContractDescription,
+        private readonly latestTransactionLocation: TransactionLocation | null
+    ) {
+        super();
+
+        this.webhookReceiver = this.createWebhookReceiver(CORVETTE_LISTEN_PORT);
+    }
+
+    async *loop(): AsyncIterableIterator<{
+        blockHash: string;
+        events: CorvetteEventResponse[];
+    }> {
+        if (this.latestTransactionLocation !== null) {
+            const { data } = await axios
+                .post<CorvetteEventResponse[]>(CORVETTE_EVENT_API_URL, {
+                    blockFrom: this.latestTransactionLocation.blockHash,
+                    abiSignature: BURN_ABI_SIGNATURE,
+                })
+                .catch((err) => {
+                    console.error(err);
+                    return { data: [] };
+                });
+            data.forEach((event) => this.eventDataQueue.add(event));
+        }
+
+        while (true) {
+            try {
+                const eventDataMap = new Map<string, CorvetteEventResponse[]>();
+
+                for (const eventData of this.eventDataQueue) {
+                    const { blockHash } = eventData;
+                    const blockEventData = eventDataMap.get(blockHash);
+
+                    if (!blockEventData)
+                        eventDataMap.set(blockHash, [eventData]);
+                    else blockEventData.push(eventData);
+
+                    this.eventDataQueue.delete(eventData);
+                }
+
+                for (const [blockHash, events] of eventDataMap) {
+                    yield { blockHash, events };
+                }
+
+                await delay(2000);
+            } catch (error) {
+                console.error(
+                    "Ignore and continue loop without breaking though unexpected error occurred:",
+                    error
+                );
+            }
+        }
+    }
+
+    private createWebhookReceiver(port: number) {
+        const server = createServer(async (request, response) => {
+            let body = "";
+            for await (const chunk of request) {
+                body += chunk.toString();
+            }
+
+            const eventData: CorvetteEventResponse = JSON.parse(body);
+
+            if (
+                this.contractDescription.address.toLowerCase() ===
+                    eventData.sourceAddress.toLowerCase() &&
+                eventData.abiSignature === BURN_ABI_SIGNATURE
+            ) {
+                this.eventDataQueue.add(eventData);
+            }
+
+            response.end();
+        }).listen(port);
+
+        return server;
+    }
+}

bridge/src/observers/burn-event-observer.ts

@@ -1,6 +1,6 @@
 import { IObserver } from ".";
 import { BlockHash } from "../types/block-hash";
-import { EventData } from "web3-eth-contract";
+import type { CorvetteEventResponse } from "../monitors/corvette-monitor";
 import { TransactionLocation } from "../types/transaction-location";
 import { BurnEventResult } from "../types/burn-event-result";
 import { INCGTransfer } from "../interfaces/ncg-transfer";
@@ -18,7 +18,7 @@ export class EthereumBurnEventObserver
     implements
         IObserver<{
             blockHash: BlockHash;
-            events: (EventData & TransactionLocation)[];
+            events: (CorvetteEventResponse & TransactionLocation)[];
         }>
 {
     private readonly _ncgTransfer: INCGTransfer;
@@ -58,7 +58,7 @@ export class EthereumBurnEventObserver
 
     async notify(data: {
         blockHash: BlockHash;
-        events: (EventData & TransactionLocation)[];
+        events: (CorvetteEventResponse & TransactionLocation)[];
     }): Promise<void> {
         const { blockHash, events } = data;
         if (events.length === 0) {
@@ -68,12 +68,12 @@ export class EthereumBurnEventObserver
             });
         }
 
-        for (const { returnValues, transactionHash, blockHash } of events) {
+        for (const { args, transactionHash, blockHash } of events) {
             const {
                 _sender: sender,
                 _to,
                 amount: burnedWrappedNcgAmountString,
-            } = returnValues as BurnEventResult;
+            } = args.named as unknown as BurnEventResult;
             const recipient = _to.substring(0, 42);
             const amount = new Decimal(burnedWrappedNcgAmountString).div(
                 new Decimal(10).pow(18)