Add basic integration with multiversx event notifier.

.env.example

@@ -18,3 +18,6 @@ DATADOG_API_KEY=""
 # The chain env is used to selected a set of evm chains, axelar chain and cosmos chains.
 # The default value is "testnet". The available values are "devnet", "testnet", and "mainnet".
 CHAIN_ENV="testnet"
+
+MULTIVERSX_NOTIFIER_URL=amqp://user:password@rabbitmq:5672/
+MULTIVERSX_GATEWAY_ADDRESS=

.gitignore

@@ -6,3 +6,5 @@ src/types/contracts
 *.log
 secret.json
 unused
+
+docker-compose.override.yml

package.json

@@ -20,8 +20,10 @@
     "@hapi/hapi": "^21.1.0",
     "@hapi/inert": "^7.0.0",
     "@hapi/vision": "^7.0.0",
+    "@multiversx/sdk-core": "^12.6.1",
     "@prisma/client": "4.8.1",
     "@reactivex/rxjs": "^6.6.7",
+    "amqplib": "^0.10.3",
     "dotenv": "^16.0.3",
     "ethers": "^5.7.2",
     "hapi-swagger": "^15.0.0",

src/clients/DatabaseClient.ts

@@ -14,6 +14,8 @@ import {
 } from '../types/contracts/IAxelarGateway';
 import { logger } from '../logger';
 import { ethers } from 'ethers';
+import { NotifierEvent } from '../listeners/MultiversXListener/types';
+import { Address } from '@multiversx/sdk-core/out';
 
 export class DatabaseClient {
   private prisma: PrismaClient;
@@ -255,6 +257,24 @@ export class DatabaseClient {
     logger.info(`[DBUpdate] ${JSON.stringify(executeDb)}`);
   }
 
+  createMvxCallContractEvent(event: NotifierEvent, messageId: string, sourceChain: string, destinationChain: any, payload: string, payloadHash: any, destinationContractAddress: any, sender: string) {
+    return this.prisma.relayData.create({
+      data: {
+        id: messageId,
+        from: sourceChain,
+        to: destinationChain,
+        callContract: {
+          create: {
+            payload: payload.toLowerCase(),
+            payloadHash: payloadHash.toLowerCase(),
+            contractAddress: destinationContractAddress.toLowerCase(),
+            sourceAddress: sender.toLowerCase(),
+          },
+        },
+      },
+    });
+  }
+
   connect() {
     return this.prisma.$connect();
   }

src/config/index.ts

@@ -11,7 +11,10 @@ export const env = {
   PORT: process.env.PORT || 3000,
   DD_API_KEY: process.env.DD_API_KEY || '',
   CHAIN_ENV: process.env.CHAIN_ENV || 'testnet',
-  LOG_LEVEL: process.env.LOG_LEVEL || "info"
+  LOG_LEVEL: process.env.LOG_LEVEL || "info",
+  MULTIVERSX_NOTIFIER_URL: process.env.MULTIVERSX_NOTIFIER_URL || '',
+  MULTIVERSX_GATEWAY_ADDRESS: process.env.MULTIVERSX_GATEWAY_ADDRESS || '',
+  MULTIVERSX_CHAIN_ID: process.env.MULTIVERSX_CHAIN_ID || '',
 };
 
 export * from './chains';

src/handler/eventHandler.ts

@@ -1,4 +1,4 @@
-import { AxelarClient, DatabaseClient, EvmClient } from '..';
+import { AxelarClient, cosmosChains, DatabaseClient, EvmClient } from '..';
 import { logger } from '../logger';
 import {
   ContractCallSubmitted,
@@ -15,6 +15,7 @@ import {
   ContractCallApprovedEventObject,
   ContractCallEventObject,
 } from '../types/contracts/IAxelarGateway';
+import { filterCosmosDestination } from '../relayer/rxOperators';
 
 const getBatchCommandIdFromSignTx = (signTx: any) => {
   const rawLog = JSON.parse(signTx.rawLog || '{}');
@@ -275,6 +276,79 @@ export async function handleEvmToCosmosCompleteEvent(client: AxelarClient, event
   logger.info(`[handleEvmToCosmosCompleteEvent] Memo: ${event.memo}`);
 }
 
+// TODO: This is all very much WIP and probably won't work :)
+export async function handleMvxToEvmOrCosmosEvent(
+  vxClient: AxelarClient,
+  evmClients: EvmClient[],
+  messageId: string,
+  payload: string,
+  sourceChain: string,
+  destinationChain: string,
+) {
+  const isCosmosDestination = cosmosChains.map((chain) => chain.chainId).includes(destinationChain);
+
+  if (isCosmosDestination) {
+    // TODO: This expects a EVM txHash instead of this messageId, not sure how to handle it.
+    const confirmTx = await vxClient.confirmEvmTx(sourceChain, messageId);
+    if (confirmTx) {
+      logger.info(`[handleEvmToCosmosEvent] Confirmed: ${confirmTx.transactionHash}`);
+    }
+
+    return;
+  }
+
+  // TODO: Should also support other destination chains in the future
+  const evmClient = evmClients.find(
+    (client) => client.chainId.toLowerCase() === destinationChain.toLowerCase()
+  );
+
+  // If no evm client found, return
+  if (!evmClient) return;
+
+  // TODO: This should contain the source_address, source_chain, destination_address, destination_chain, payload_hash
+  // For the Gateway contract running on Axelar CosmWASM. Are those passed through the payload?
+  const routeMessage = await vxClient.routeMessageRequest(
+    -1,
+    messageId,
+    payload
+  );
+
+  if (routeMessage) {
+    logger.info(`[handleMvxToEvmEvent] RouteMessage: ${routeMessage.transactionHash}`);
+  }
+
+  const pendingCommands = await vxClient.getPendingCommands(destinationChain);
+
+  logger.info(`[handleMvxToEvmEvent] PendingCommands: ${JSON.stringify(pendingCommands)}`);
+  if (pendingCommands.length === 0) return;
+
+  const signCommand = await vxClient.signCommands(destinationChain);
+  logger.debug(`[handleMvxToEvmEvent] SignCommand: ${JSON.stringify(signCommand)}`);
+
+  if (signCommand && signCommand.rawLog?.includes('failed')) {
+    throw new Error(signCommand.rawLog);
+  }
+  if (!signCommand) {
+    throw new Error('cannot sign command');
+  }
+
+  const batchedCommandId = getBatchCommandIdFromSignTx(signCommand);
+  logger.info(`[handleMvxToEvmEvent] BatchCommandId: ${batchedCommandId}`);
+
+  const executeData = await vxClient.getExecuteDataFromBatchCommands(
+    destinationChain,
+    batchedCommandId
+  );
+
+  logger.info(`[handleMvxToEvmEvent] BatchCommands: ${JSON.stringify(executeData)}`);
+
+  const tx = await evmClient.gatewayExecute(executeData);
+  if (!tx) return;
+  logger.info(`[handleMvxToEvmEvent] Execute: ${tx.transactionHash}`);
+
+  return tx;
+}
+
 export async function prepareHandler(event: any, db: DatabaseClient, label = '') {
   // reconnect prisma db
   await db.connect();

src/listeners/MultiversXListener/MultiversXListener.ts

@@ -0,0 +1,65 @@
+import { env } from '../../config';
+import { logger } from '../../logger';
+import { BlockEvent, NotifierEvent } from './types';
+import { Address, BinaryCodec, BytesType } from '@multiversx/sdk-core/out';
+import { AxelarClient, DatabaseClient, EvmClient } from '../../clients';
+import { handleEvmToCosmosEvent, handleMvxToEvmOrCosmosEvent } from '../../handler';
+
+const amqp = require('amqplib/channel_api.js');
+
+export class MultiversXListener {
+  constructor(private readonly db: DatabaseClient, private readonly axelarClient: AxelarClient, private readonly evmClients: EvmClient[]) {
+  }
+
+  async listenToMultiversXEvents() {
+    const connection = await amqp.connect(env.MULTIVERSX_NOTIFIER_URL, {});
+
+    const channel = await connection.createChannel();
+
+    const queue = await channel.checkQueue('events-638c4d10');
+
+    await channel.consume(queue.queue, async (msg: any) => {
+      const blockEvent: BlockEvent = JSON.parse(msg.content.toString());
+
+      for (const event of blockEvent.events) {
+        await this.handleEvent(event);
+      }
+    });
+
+    logger.info('[MultiversXListener] Listening to MultiversX Gateway events');
+  }
+
+  private async handleEvent(event: NotifierEvent) {
+    if (event.address !== env.MULTIVERSX_GATEWAY_ADDRESS) {
+      return;
+    }
+
+    if (event.identifier === 'callContract') {
+      logger.info('[MultiversXListener] Received callContract event from MultiversX Gateway contract:');
+      logger.info(JSON.stringify(event));
+
+      const sender = Address.fromBuffer(Buffer.from(event.topics[1], 'base64'));
+      const destinationChain = Buffer.from(event.topics[2], 'base64').toString();
+      const destinationContractAddress = Buffer.from(event.topics[3], 'base64').toString('hex');
+
+      const attributes = Buffer.from(event.data, 'base64');
+      const payloadHash = attributes.slice(0, 32).toString('hex');
+      const dataPayloadBuffer = attributes.slice(32);
+
+      const codec = new BinaryCodec();
+      const [decoded] = codec.decodeNested(dataPayloadBuffer, new BytesType());
+      const payload = (decoded.valueOf() as Buffer).toString('hex');
+
+      logger.info(`Event contains data: ${ sender } ${ destinationChain } ${ destinationContractAddress } ${ payloadHash } ${ payload }`);
+
+      const sourceChain = 'multiversx-' + env.MULTIVERSX_CHAIN_ID;
+
+      // TODO: Should this be hashed?
+      const messageId = event.txHash + sourceChain;
+
+      await this.db.createMvxCallContractEvent(event, messageId, sourceChain, destinationChain, payload, payloadHash, destinationContractAddress, sender.bech32());
+
+      await handleMvxToEvmOrCosmosEvent(this.axelarClient, this.evmClients, messageId, payload, sourceChain, destinationChain);
+    }
+  }
+}

src/listeners/MultiversXListener/types.ts

@@ -0,0 +1,14 @@
+export interface BlockEvent {
+  hash: string;
+  shardId: number;
+  timestamp: Number;
+  events: NotifierEvent[];
+}
+
+export interface NotifierEvent {
+  txHash: string;
+  address: string;
+  identifier: string;
+  data: string;
+  topics: string[];
+}

src/relayer/relay.ts

@@ -1,6 +1,6 @@
 import { Subject, mergeMap } from 'rxjs';
 import { AxelarClient, EvmClient, DatabaseClient } from '../clients';
-import { axelarChain, cosmosChains, evmChains } from '../config';
+import { axelarChain, cosmosChains, env, evmChains } from '../config';
 import {
   ContractCallSubmitted,
   ContractCallWithTokenSubmitted,
@@ -38,6 +38,8 @@ import {
 } from '../handler';
 import { createCosmosEventSubject, createEvmEventSubject } from './subject';
 import { filterCosmosDestination, mapEventToEvmClient } from './rxOperators';
+import { logger } from '../logger';
+import { MultiversXListener } from '../listeners/MultiversXListener/MultiversXListener';
 
 const sEvmCallContract = createEvmEventSubject<ContractCallEventObject>();
 const sEvmCallContractWithToken = createEvmEventSubject<ContractCallWithTokenEventObject>();
@@ -198,4 +200,14 @@ export async function startRelayer() {
   axelarListener.listen(AxelarCosmosContractCallWithTokenEvent, sCosmosContractCallWithToken);
   axelarListener.listen(AxelarIBCCompleteEvent, sCosmosApproveAny);
   axelarListener.listen(AxelarEVMCompletedEvent, sEvmConfirmEvent);
+
+  // Listening for MultiversX events
+  try {
+    const multiversXListener = new MultiversXListener(db, axelarClient, evmClients);
+
+    await multiversXListener.listenToMultiversXEvents();
+  } catch (e) {
+    logger.error('Could not listen to MultiversX events');
+    logger.error(e);
+  }
 }