smom-dbis-138/services/relay/src/RelayService.js

/**
 * 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';
import {
  isRelayShedding,
  getRelaySheddingSourcePollIntervalMs,
  getRelaySheddingQueueIdleMs
} from './config.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;
    this.destinationBridgeContracts = new Map();
    this.messageSentInterface = new ethers.Interface(MessageSentABI);
    /** @type {number} throttle for shedding warning logs */
    this._lastSheddingLogTs = 0;
  }

  /** Polling interval for source router logs (longer while shedding to cut RPC churn). */
  getSourcePollIntervalMs() {
    if (isRelayShedding()) {
      return getRelaySheddingSourcePollIntervalMs();
    }
    return this.config.monitoring.pollInterval;
  }
  
  async start() {
    this.logger.info('Initializing relay service...');
    
    // Use plain JsonRpcProvider here; explicit custom-network pinning caused inconsistent log polling
    // on the nonstandard Chain 138 RPC even though direct manual queries succeed.
    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: %s', String(this.destinationSigner.address));
    
    // Validate relay router address (bridge can be dynamic from message receiver)
    if (!this.config.destinationChain.relayRouterAddress || 
        this.config.destinationChain.relayRouterAddress === '') {
      throw new Error(`Relay router address must be configured on destination chain. Router: ${this.config.destinationChain.relayRouterAddress}`);
    }
    
    // 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
    );
    
    // Backward-compatible static destination bridge (optional)
    if (this.config.destinationChain.relayBridgeAddress) {
      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
  }

  /** Preferred chunk size for scanning (adaptive split handles stricter RPCs). Override: SOURCE_LOGS_MAX_BLOCK_RANGE */
  getSourceLogsMaxBlockRange() {
    const v = parseInt(
      process.env.SOURCE_LOGS_MAX_BLOCK_RANGE || process.env.RELAY_SOURCE_LOGS_MAX_BLOCK_RANGE || '8000',
      10
    );
    return Number.isFinite(v) && v >= 500 ? v : 8000;
  }

  static _isRpcLogRangeError(err) {
    const msg = String(err && err.message ? err.message : err);
    return (
      msg.includes('maximum RPC range') ||
      msg.includes('exceeds maximum') ||
      (msg.includes('-32000') && msg.includes('range'))
    );
  }

  async fetchSourceLogs(fromBlock, toBlock) {
    const payload = {
      jsonrpc: '2.0',
      id: 1,
      method: 'eth_getLogs',
      params: [{
        address: this.config.sourceChain.routerAddress,
        fromBlock: ethers.toQuantity(fromBlock),
        toBlock: ethers.toQuantity(toBlock),
        topics: [this.messageSentInterface.getEvent('MessageSent').topicHash]
      }]
    };

    const response = await fetch(this.config.sourceChain.rpcUrl, {
      method: 'POST',
      headers: { 'content-type': 'application/json' },
      body: JSON.stringify(payload)
    });

    if (!response.ok) {
      throw new Error(`eth_getLogs HTTP ${response.status}`);
    }

    const data = await response.json();
    if (data.error) {
      throw new Error(`eth_getLogs RPC ${data.error.code}: ${data.error.message}`);
    }

    return Array.isArray(data.result) ? data.result : [];
  }

  _sortRawLogs(all) {
    all.sort((a, b) => {
      const ba = Number(BigInt(a.blockNumber));
      const bb = Number(BigInt(b.blockNumber));
      if (ba !== bb) return ba - bb;
      return String(a.logIndex || '').localeCompare(String(b.logIndex || ''));
    });
    return all;
  }

  /**
   * Single eth_getLogs for [fromBlock,toBlock]; on RPC range-limit errors, bisect until requests fit.
   */
  async fetchSourceLogsAdaptive(fromBlock, toBlock) {
    if (toBlock < fromBlock) return [];
    try {
      return await this.fetchSourceLogs(fromBlock, toBlock);
    } catch (e) {
      if (!RelayService._isRpcLogRangeError(e) || fromBlock === toBlock) throw e;
      const mid = Math.floor((fromBlock + toBlock) / 2);
      const left = await this.fetchSourceLogsAdaptive(fromBlock, mid);
      const right = await this.fetchSourceLogsAdaptive(mid + 1, toBlock);
      return this._sortRawLogs([...left, ...right]);
    }
  }

  /**
   * Scan MessageSent logs from fromBlock..toBlock using coarse chunks, each resolved via adaptive splits.
   */
  async fetchSourceLogsChunked(fromBlock, toBlock) {
    const maxSpan = this.getSourceLogsMaxBlockRange();
    if (toBlock < fromBlock) return [];

    const all = [];
    let cursor = fromBlock;
    while (cursor <= toBlock && this.isRunning) {
      const chunkEnd = Math.min(toBlock, cursor + maxSpan - 1);
      const chunk = await this.fetchSourceLogsAdaptive(cursor, chunkEnd);
      if (chunk.length) all.push(...chunk);
      cursor = chunkEnd + 1;
    }
    return this._sortRawLogs(all);
  }
  
  async startMonitoring() {
    this.logger.info('Starting event monitoring...');
    
    let startBlock;
    if (this.config.monitoring.startBlock === 'latest' || isNaN(this.config.monitoring.startBlock)) {
      const currentBlock = await this.sourceProvider.getBlockNumber();
      startBlock = currentBlock > 0 ? currentBlock - 1 : 0;
    } else {
      startBlock = parseInt(this.config.monitoring.startBlock);
    }
    
    this.logger.info(`Monitoring from block: ${startBlock}`);
    
    // HTTP filter subscriptions are unreliable on several RPCs used here.
    // Polling via queryFilter is the default and can be overridden explicitly.
    if (process.env.RELAY_USE_EVENT_SUBSCRIPTIONS === '1') {
      this.sourceRouter.on('MessageSent', async (messageId, destinationChainSelector, sender, receiver, data, tokenAmounts, feeToken, extraArgs, event) => {
        try {
          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
          });
          
          const receipt = await event.getTransactionReceipt();
          const confirmations = this.config.monitoring.confirmationBlocks;
          
          if (confirmations > 0) {
            await receipt.confirmations(confirmations);
          }
          
          const formattedTokenAmounts = tokenAmounts.map(ta => ({
            token: ta.token,
            amount: ta.amount,
            amountType: ta.amountType
          }));
          
          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();
        const finalityDelay = this.config.monitoring.finalityDelayBlocks || 0;
        const replayWindow = this.config.monitoring.replayWindowBlocks || 0;
        const toBlock = currentBlock > finalityDelay ? currentBlock - finalityDelay : currentBlock;

        if (toBlock >= startBlock) {
          this.logger.info(`Polling events from block ${startBlock} to ${toBlock}`);

          const logs = await this.fetchSourceLogsChunked(startBlock, toBlock);

          this.logger.info(`Fetched ${logs.length} MessageSent log(s) from source router`);

          for (const log of logs) {
            let decoded;
            try {
              decoded = this.messageSentInterface.parseLog(log);
            } catch (error) {
              this.logger.error('Error decoding MessageSent log:', error);
              continue;
            }

            const { messageId, destinationChainSelector, sender, receiver, data, tokenAmounts, feeToken, extraArgs } = decoded.args;

            const destSelector = destinationChainSelector.toString();
            const expectedSelector = this.config.destinationChain.chainSelector.toString();

            if (destSelector !== expectedSelector) {
              continue;
            }

            this.logger.info('Historical MessageSent detected:', {
              messageId,
              destinationChainSelector: destSelector,
              sender,
              blockNumber: log.blockNumber,
              transactionHash: log.transactionHash
            });

            await this.messageQueue.add({
              messageId,
              destinationChainSelector,
              sender,
              receiver,
              data,
              tokenAmounts: tokenAmounts.map((ta) => ({
                token: ta.token,
                amount: ta.amount,
                amountType: ta.amountType
              })),
              feeToken,
              extraArgs,
              blockNumber: log.blockNumber,
              transactionHash: log.transactionHash
            });
          }
          
          startBlock = Math.max(0, toBlock - replayWindow + 1);
        }
        
        // Wait before next poll (longer interval while relay shedding is on)
        await new Promise(resolve => setTimeout(resolve, this.getSourcePollIntervalMs()));
        
      } catch (error) {
        this.logger.error('Error polling historical events:', error);
        await new Promise(resolve => setTimeout(resolve, this.getSourcePollIntervalMs()));
      }
    }
  }
  
  async startProcessingQueue() {
    this.logger.info('Starting message queue processor...');
    
    while (this.isRunning) {
      try {
        // Relay shedding: do not submit destination-chain txs (saves gas). Messages keep queuing
        // from source polling; when shedding is off, the queue drains normally.
        if (isRelayShedding()) {
          const stats = this.messageQueue.getStats();
          const now = Date.now();
          if (stats.queueSize > 0 && now - this._lastSheddingLogTs >= 60000) {
            this._lastSheddingLogTs = now;
            this.logger.warn(
              `Relay shedding ON: ${stats.queueSize} message(s) queued; destination delivery paused. ` +
                `Set RELAY_SHEDDING=0 and RELAY_DELIVERY_ENABLED=1 then restart (or reload env) to deliver.`
            );
          }
          await new Promise((r) => setTimeout(r, getRelaySheddingQueueIdleMs()));
          continue;
        }

        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 {
      // On-chain pause (CCIPRelayRouter Pausable): avoid broadcasting reverting txs
      if (this.config.destinationChain.deliveryMode !== 'direct' && this.destinationRelayRouter) {
        try {
          const routerPaused = await this.destinationRelayRouter.paused();
          if (routerPaused) {
            this.logger.warn(
              `Destination relay router is paused; deferring ${messageId} (enable RELAY_SHEDDING=1 to pause off-chain too)`
            );
            await this.messageQueue.retry(messageId);
            await new Promise((r) => setTimeout(r, 15000));
            return null;
          }
        } catch (pauseCheckErr) {
          this.logger.debug('Router paused() check skipped or unsupported', pauseCheckErr);
        }
      }

      // Route to bridge encoded in MessageSent.receiver (bytes). Fallback to static env bridge.
      let targetBridge = this.config.destinationChain.relayBridgeAddress;
      try {
        if (receiver) {
          const decoded = ethers.AbiCoder.defaultAbiCoder().decode(['address'], receiver);
          if (decoded && decoded[0]) targetBridge = ethers.getAddress(decoded[0]);
        }
      } catch (_) {
        // keep fallback targetBridge
      }
      if (!targetBridge) {
        throw new Error(`No destination bridge for message ${messageId}: receiver decode failed and DEST_RELAY_BRIDGE not set`);
      }

      // Optional allowlist hardening.
      const allowlist = this.config.destinationChain.relayBridgeAllowlist || [];
      if (allowlist.length > 0 && !allowlist.includes(String(targetBridge).toLowerCase())) {
        throw new Error(`Bridge ${targetBridge} not in DEST_RELAY_BRIDGE_ALLOWLIST`);
      }

      let targetBridgeContract = this.destinationBridgeContracts.get(targetBridge.toLowerCase());
      if (!targetBridgeContract) {
        targetBridgeContract = new ethers.Contract(targetBridge, RelayBridgeABI, this.destinationProvider);
        this.destinationBridgeContracts.set(targetBridge.toLowerCase(), targetBridgeContract);
      }

      // Idempotency guard: do not relay a message that destination already marked processed.
      let alreadyProcessed = false;
      try {
        if (targetBridgeContract.processed) {
          alreadyProcessed = await targetBridgeContract.processed(messageId);
        }
      } catch (_) {
        // ignore and try legacy method below
      }
      if (!alreadyProcessed) {
        try {
          if (targetBridgeContract.processedTransfers) {
            alreadyProcessed = await targetBridgeContract.processedTransfers(messageId);
          }
        } catch (_) {
          // destination bridge may not expose either helper
        }
      }
      if (alreadyProcessed) {
        this.logger.info(`Message ${messageId} already processed on destination; skipping relay tx`);
        await this.messageQueue.markProcessed(messageId);
        return null;
      }

      // 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)
        };
      });
      
      // Optional normalization for legacy bridges that decode 4-field payloads:
      // (recipient, amount, sender, nonce). TwoWayTokenBridgeL1/L2 decode 2-field payloads
      // (recipient, amount), so leave those unchanged unless explicitly enabled.
      let normalizedData = data;
      if (process.env.RELAY_EXPAND_TWO_FIELD_PAYLOAD === '1') {
        try {
          const decoded = ethers.AbiCoder.defaultAbiCoder().decode(['address', 'uint256'], data);
          const reencoded2 = ethers.AbiCoder.defaultAbiCoder().encode(['address', 'uint256'], [decoded[0], decoded[1]]);
          if ((data || '').toLowerCase() === reencoded2.toLowerCase()) {
            normalizedData = ethers.AbiCoder.defaultAbiCoder().encode(
              ['address', 'uint256', 'address', 'uint256'],
              [decoded[0], decoded[1], sender, 0]
            );
            this.logger.debug(`Normalized 2-field payload to 4-field payload for message ${messageId}`);
          }
        } catch (_) {
          // Keep original payload when it does not match (address,uint256).
        }
      }

      // Construct Any2EVMMessage struct
      // tokenAmounts is an array of TokenAmount structs: { token: address, amount: uint256, amountType: uint8 }
      const any2EVMMessage = {
        messageId: messageId,
        sourceChainSelector: BigInt(this.config.sourceChainSelector.toString()),
        sender: ethers.AbiCoder.defaultAbiCoder().encode(['address'], [sender]),
        data: normalizedData,
        tokenAmounts: mappedTokenAmounts
      };
      
      this.logger.debug('Relaying message with struct:', {
        messageId: any2EVMMessage.messageId,
        sourceChainSelector: any2EVMMessage.sourceChainSelector,
        tokenAmountsCount: any2EVMMessage.tokenAmounts.length
      });
      
      let tx;
      if (this.config.destinationChain.deliveryMode === 'direct') {
        this.logger.info(`Direct-delivery mode: calling bridge ${targetBridge} without relay router`);
        const directBridge = new ethers.Contract(
          targetBridge,
          RelayBridgeABI,
          this.destinationSigner
        );
        tx = await directBridge.ccipReceive(any2EVMMessage, { gasLimit: 1000000 });
      } else {
        // Call relay router with properly formatted struct
        tx = await this.destinationRelayRouter.relayMessage(
          targetBridge,
          any2EVMMessage,
          { gasLimit: 1000000 }
        );
      }
      
      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;
    }
  }
}