/** * CCIP Relay Service * Monitors MessageSent events and relays messages to destination chain */ import { ethers } from 'ethers'; import { MessageSentABI, RelayRouterABI, RelayBridgeABI } from './abis.js'; import { MessageQueue } from './MessageQueue.js'; export class RelayService { constructor(config, logger) { this.config = config; this.logger = logger; this.isRunning = false; this.sourceProvider = null; this.destinationProvider = null; this.sourceSigner = null; this.destinationSigner = null; this.messageQueue = new MessageQueue(this.logger); // Contract instances this.sourceRouter = null; this.destinationRelayRouter = null; this.destinationRelayBridge = null; } async start() { this.logger.info('Initializing relay service...'); // Initialize providers this.sourceProvider = new ethers.JsonRpcProvider(this.config.sourceChain.rpcUrl); this.destinationProvider = new ethers.JsonRpcProvider(this.config.destinationChain.rpcUrl); // Initialize signers if (!this.config.relayer.privateKey) { throw new Error('Relayer private key not configured'); } this.sourceSigner = new ethers.Wallet(this.config.relayer.privateKey, this.sourceProvider); this.destinationSigner = new ethers.Wallet(this.config.relayer.privateKey, this.destinationProvider); this.logger.info('Relayer address:', this.destinationSigner.address); // Validate relay router and bridge addresses if (!this.config.destinationChain.relayRouterAddress || this.config.destinationChain.relayRouterAddress === '' || !this.config.destinationChain.relayBridgeAddress || this.config.destinationChain.relayBridgeAddress === '') { throw new Error(`Relay router and bridge addresses must be configured on destination chain. Router: ${this.config.destinationChain.relayRouterAddress}, Bridge: ${this.config.destinationChain.relayBridgeAddress}`); } // Initialize contract instances this.sourceRouter = new ethers.Contract( this.config.sourceChain.routerAddress, MessageSentABI, this.sourceProvider ); this.destinationRelayRouter = new ethers.Contract( this.config.destinationChain.relayRouterAddress, RelayRouterABI, this.destinationSigner ); this.destinationRelayBridge = new ethers.Contract( this.config.destinationChain.relayBridgeAddress, RelayBridgeABI, this.destinationProvider ); // Start monitoring this.isRunning = true; await this.startMonitoring(); // Start processing queue this.startProcessingQueue(); } async stop() { this.logger.info('Stopping relay service...'); this.isRunning = false; // Additional cleanup if needed } async startMonitoring() { this.logger.info('Starting event monitoring...'); let startBlock; if (this.config.monitoring.startBlock === 'latest' || isNaN(this.config.monitoring.startBlock)) { startBlock = await this.sourceProvider.getBlockNumber(); } else { startBlock = parseInt(this.config.monitoring.startBlock); } this.logger.info(`Monitoring from block: ${startBlock}`); // Listen for MessageSent events this.sourceRouter.on('MessageSent', async (messageId, destinationChainSelector, sender, receiver, data, tokenAmounts, feeToken, extraArgs, event) => { try { // Only process messages for our destination chain const destSelector = destinationChainSelector.toString(); const expectedSelector = this.config.destinationChain.chainSelector.toString(); if (destSelector !== expectedSelector) { this.logger.debug(`Ignoring message for different chain: ${destSelector}`); return; } this.logger.info('MessageSent event detected:', { messageId: messageId, destinationChainSelector: destinationChainSelector.toString(), sender: sender, blockNumber: event.blockNumber, transactionHash: event.transactionHash }); // Wait for confirmations const receipt = await event.getTransactionReceipt(); const confirmations = this.config.monitoring.confirmationBlocks; if (confirmations > 0) { await receipt.confirmations(confirmations); } // Format tokenAmounts properly const formattedTokenAmounts = tokenAmounts.map(ta => ({ token: ta.token, amount: ta.amount, amountType: ta.amountType })); // Add message to queue await this.messageQueue.add({ messageId, destinationChainSelector, sender, receiver, data, tokenAmounts: formattedTokenAmounts, feeToken, extraArgs, blockNumber: event.blockNumber, transactionHash: event.transactionHash }); } catch (error) { this.logger.error('Error processing MessageSent event:', error); } }); // Also poll from start block to catch any missed events this.pollHistoricalEvents(startBlock); } async pollHistoricalEvents(startBlock) { while (this.isRunning) { try { const currentBlock = await this.sourceProvider.getBlockNumber(); if (currentBlock > startBlock) { this.logger.debug(`Polling events from block ${startBlock} to ${currentBlock}`); const filter = this.sourceRouter.filters.MessageSent(); const events = await this.sourceRouter.queryFilter(filter, startBlock, currentBlock); for (const event of events) { // Process event (same logic as event listener) const { messageId, destinationChainSelector, sender, receiver, data, tokenAmounts, feeToken, extraArgs } = event.args; const destSelector = destinationChainSelector.toString(); const expectedSelector = this.config.destinationChain.chainSelector.toString(); if (destSelector === expectedSelector) { await this.messageQueue.add({ messageId, destinationChainSelector, sender, receiver, data, tokenAmounts, feeToken, extraArgs, blockNumber: event.blockNumber, transactionHash: event.transactionHash }); } } startBlock = currentBlock + 1; } // Wait before next poll await new Promise(resolve => setTimeout(resolve, this.config.monitoring.pollInterval)); } catch (error) { this.logger.error('Error polling historical events:', error); await new Promise(resolve => setTimeout(resolve, this.config.monitoring.pollInterval)); } } } async startProcessingQueue() { this.logger.info('Starting message queue processor...'); while (this.isRunning) { try { const message = await this.messageQueue.getNext(); if (message) { await this.relayMessage(message); } else { // No messages, wait a bit await new Promise(resolve => setTimeout(resolve, 1000)); } } catch (error) { this.logger.error('Error processing message queue:', error); await new Promise(resolve => setTimeout(resolve, 1000)); } } } async relayMessage(messageData) { const { messageId, destinationChainSelector, sender, receiver, data, tokenAmounts } = messageData; this.logger.info(`Relaying message ${messageId} to destination chain...`); try { // Map token addresses from source chain to destination chain const mappedTokenAmounts = tokenAmounts.map(ta => { const sourceToken = ethers.getAddress(ta.token); const destinationToken = this.config.tokenMapping[sourceToken] || sourceToken; this.logger.debug(`Mapping token ${sourceToken} -> ${destinationToken}`); return { token: destinationToken, // Map to destination chain token address amount: typeof ta.amount === 'bigint' ? ta.amount : BigInt(ta.amount.toString()), amountType: Number(ta.amountType) // Ensure it's a number (uint8) }; }); // Construct Any2EVMMessage struct // tokenAmounts is an array of TokenAmount structs: { token: address, amount: uint256, amountType: uint8 } const any2EVMMessage = { messageId: messageId, sourceChainSelector: Number(this.config.sourceChainSelector), // Convert BigInt to number for uint64 sender: ethers.AbiCoder.defaultAbiCoder().encode(['address'], [sender]), data: data, tokenAmounts: mappedTokenAmounts }; this.logger.debug('Relaying message with struct:', { messageId: any2EVMMessage.messageId, sourceChainSelector: any2EVMMessage.sourceChainSelector, tokenAmountsCount: any2EVMMessage.tokenAmounts.length }); // Call relay router with properly formatted struct const tx = await this.destinationRelayRouter.relayMessage( this.config.destinationChain.relayBridgeAddress, any2EVMMessage, { gasLimit: 1000000 } // Increased gas limit ); this.logger.info(`Relay transaction sent: ${tx.hash}`); // Wait for confirmation const receipt = await tx.wait(); this.logger.info(`Message ${messageId} relayed successfully. Transaction: ${receipt.hash}`); // Mark message as processed await this.messageQueue.markProcessed(messageId); return receipt; } catch (error) { this.logger.error(`Error relaying message ${messageId}:`, error); // Retry logic const retryCount = await this.messageQueue.getRetryCount(messageId); if (retryCount < this.config.retry.maxRetries) { this.logger.info(`Retrying message ${messageId} (attempt ${retryCount + 1})`); await this.messageQueue.retry(messageId); await new Promise(resolve => setTimeout(resolve, this.config.retry.retryDelay)); } else { this.logger.error(`Message ${messageId} failed after ${this.config.retry.maxRetries} retries`); await this.messageQueue.markFailed(messageId); } throw error; } } }