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, ERC20ABI } 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;
    this.startedAt = new Date().toISOString();
    this.startEpochMs = Date.now();
    this.relayerAddress = '';
    this.lastSourcePoll = null;
    this.lastSeenMessage = null;
    this.lastRelayAttempt = null;
    this.lastRelaySuccess = null;
    this.lastError = null;
  }

  normalizeAddress(value) {
    if (!value) return '';
    try {
      return ethers.getAddress(value);
    } catch (_) {
      return '';
    }
  }

  getConfiguredSourceBridge() {
    return this.normalizeAddress(this.config.sourceChain.bridgeAddress);
  }

  getDestinationBridgeAllowlist() {
    return (this.config.destinationChain.relayBridgeAllowlist || [])
      .map((value) => String(value || '').trim().toLowerCase())
      .filter(Boolean);
  }

  resolveTargetBridge(receiver) {
    let targetBridge = this.normalizeAddress(this.config.destinationChain.relayBridgeAddress);
    try {
      if (receiver) {
        const decoded = ethers.AbiCoder.defaultAbiCoder().decode(['address'], receiver);
        if (decoded && decoded[0]) {
          targetBridge = this.normalizeAddress(decoded[0]) || targetBridge;
        }
      }
    } catch (_) {
      // Keep the configured fallback bridge when receiver decode fails.
    }
    return targetBridge;
  }

  evaluateMessageScope(messageData) {
    const sourceBridge = this.getConfiguredSourceBridge();
    const sender = this.normalizeAddress(messageData.sender);
    const targetBridge = this.resolveTargetBridge(messageData.receiver);
    const allowlist = this.getDestinationBridgeAllowlist();

    if (sourceBridge && sender && sender.toLowerCase() !== sourceBridge.toLowerCase()) {
      return {
        inScope: false,
        reason: `sender ${sender} does not match worker source bridge ${sourceBridge}`,
        sourceBridge,
        sender,
        targetBridge
      };
    }

    if (!targetBridge) {
      return {
        inScope: false,
        reason: 'destination bridge could not be resolved from receiver or config',
        sourceBridge,
        sender,
        targetBridge: ''
      };
    }

    if (allowlist.length > 0 && !allowlist.includes(targetBridge.toLowerCase())) {
      return {
        inScope: false,
        reason: `destination bridge ${targetBridge} not in worker allowlist`,
        sourceBridge,
        sender,
        targetBridge
      };
    }

    return {
      inScope: true,
      reason: '',
      sourceBridge,
      sender,
      targetBridge
    };
  }

  static summarizeError(error) {
    if (!error) return '';
    if (typeof error === 'string') return error;
    return String(error.shortMessage || error.message || error);
  }

  recordError(scope, error, extra = {}) {
    this.lastError = {
      at: new Date().toISOString(),
      scope,
      message: RelayService.summarizeError(error),
      ...extra
    };
  }

  getHealthStatus() {
    if (!this.isRunning) {
      return 'stopped';
    }
    if (isRelayShedding()) {
      return 'paused';
    }

    if (this.lastSourcePoll && this.lastSourcePoll.ok === false) {
      return 'degraded';
    }

    if (!this.lastSourcePoll || !this.lastSourcePoll.at) {
      return 'starting';
    }

    const lastPollMs = Date.parse(this.lastSourcePoll.at);
    const staleAfterMs = Math.max(this.getSourcePollIntervalMs() * 3, 15000);
    if (Number.isFinite(lastPollMs) && Date.now() - lastPollMs > staleAfterMs) {
      return 'stale';
    }

    const lastErrorMs = this.lastError && this.lastError.at ? Date.parse(this.lastError.at) : 0;
    const lastSuccessMs = this.lastRelaySuccess && this.lastRelaySuccess.at ? Date.parse(this.lastRelaySuccess.at) : 0;
    if (
      this.lastError &&
      (this.lastError.scope === 'relay_message' || this.lastError.scope === 'bridge_inventory') &&
      Number.isFinite(lastErrorMs) &&
      lastErrorMs > 0 &&
      lastErrorMs >= lastSuccessMs
    ) {
      return 'degraded';
    }

    return 'operational';
  }

  async ensureTargetBridgeInventory(messageId, targetBridge, tokenAmounts) {
    if (process.env.RELAY_ENFORCE_BRIDGE_TOKEN_BALANCE !== '1') {
      return { ok: true };
    }

    for (const tokenAmount of tokenAmounts) {
      const tokenAddress = ethers.getAddress(tokenAmount.token);
      const requiredAmount = typeof tokenAmount.amount === 'bigint'
        ? tokenAmount.amount
        : BigInt(tokenAmount.amount.toString());

      const tokenContract = new ethers.Contract(tokenAddress, ERC20ABI, this.destinationProvider);
      const availableAmount = await tokenContract.balanceOf(targetBridge);
      if (availableAmount < requiredAmount) {
        const shortfall = requiredAmount - availableAmount;
        return {
          ok: false,
          token: tokenAddress,
          requiredAmount,
          availableAmount,
          shortfall,
          message: `Insufficient bridge inventory for ${messageId}: ${tokenAddress} available=${availableAmount.toString()} required=${requiredAmount.toString()} shortfall=${shortfall.toString()}`
        };
      }
    }

    return { ok: true };
  }

  getHealthSnapshot() {
    const queueStats = this.messageQueue.getStats();
    const status = this.getHealthStatus();
    const sourceBridge = this.getConfiguredSourceBridge();
    const allowlist = this.getDestinationBridgeAllowlist();
    const defaultTargetBridge = this.normalizeAddress(this.config.destinationChain.relayBridgeAddress);

    return {
      ok: status === 'operational' || status === 'paused' || status === 'starting',
      status,
      service: {
        name: 'ccip-relay',
        running: this.isRunning,
        pid: process.pid,
        profile: process.env.RELAY_PROFILE || 'default',
        started_at: this.startedAt,
        uptime_sec: Math.max(0, Math.floor((Date.now() - this.startEpochMs) / 1000)),
        relayer_address: this.relayerAddress || this.config.relayer.address || ''
      },
      source: {
        chain_name: this.config.sourceChain.name,
        chain_id: this.config.sourceChain.chainId,
        chain_selector: this.config.sourceChainSelector.toString(),
        router_address: this.config.sourceChain.routerAddress,
        bridge_address: this.config.sourceChain.bridgeAddress,
        bridge_filter: sourceBridge || ''
      },
      destination: {
        chain_name: this.config.destinationChain.name,
        chain_id: this.config.destinationChain.chainId,
        chain_selector: this.config.destinationChain.chainSelector.toString(),
        relay_router: this.config.destinationChain.relayRouterAddress,
        relay_bridge: this.config.destinationChain.relayBridgeAddress,
        relay_bridge_default: defaultTargetBridge || '',
        relay_bridge_allowlist: allowlist,
        delivery_mode: this.config.destinationChain.deliveryMode
      },
      scope: {
        source_bridge: sourceBridge || '',
        destination_bridge_default: defaultTargetBridge || '',
        destination_bridge_allowlist: allowlist
      },
      monitoring: {
        start_block: String(this.config.monitoring.startBlock),
        poll_interval_ms: this.config.monitoring.pollInterval,
        effective_source_poll_interval_ms: this.getSourcePollIntervalMs(),
        confirmation_blocks: this.config.monitoring.confirmationBlocks,
        finality_delay_blocks: this.config.monitoring.finalityDelayBlocks || 0,
        replay_window_blocks: this.config.monitoring.replayWindowBlocks || 0,
        shedding: isRelayShedding(),
        delivery_enabled: !isRelayShedding()
      },
      queue: {
        size: queueStats.queueSize,
        processed: queueStats.processed,
        failed: queueStats.failed
      },
      last_source_poll: this.lastSourcePoll,
      last_seen_message: this.lastSeenMessage,
      last_relay_attempt: this.lastRelayAttempt,
      last_relay_success: this.lastRelaySuccess,
      last_error: this.lastError
    };
  }

  shouldSkipMessageId(messageId) {
    const normalized = String(messageId || '').toLowerCase();
    return normalized && this.config.skipMessageIds && this.config.skipMessageIds.has(normalized);
  }

  /** Polling interval for source router logs (longer while shedding to cut RPC churn). */
  getSourcePollIntervalMs() {
    if (isRelayShedding()) {
      return getRelaySheddingSourcePollIntervalMs();
    }
    return this.config.monitoring.pollInterval;
  }

  async getDestinationTxOptions() {
    const txOptions = {
      gasLimit: BigInt(process.env.RELAY_DEST_GAS_LIMIT || '1000000')
    };

    if (process.env.RELAY_DEST_LEGACY_TX !== '1') {
      return txOptions;
    }

    let gasPrice = process.env.RELAY_DEST_GAS_PRICE_WEI ? BigInt(process.env.RELAY_DEST_GAS_PRICE_WEI) : null;
    if (gasPrice === null) {
      try {
        const feeData = await this.destinationProvider.getFeeData();
        if (feeData.gasPrice) {
          gasPrice = feeData.gasPrice;
        }
      } catch (_) {
        // Fall through to raw RPC fallback below.
      }
    }

    if (gasPrice === null) {
      const rawGasPrice = await this.destinationProvider.send('eth_gasPrice', []);
      gasPrice = BigInt(rawGasPrice);
    }

    const bumpPct = BigInt(process.env.RELAY_DEST_GAS_PRICE_BUMP_PCT || '20');
    const bumpWei = BigInt(process.env.RELAY_DEST_GAS_PRICE_BUMP_WEI || '1000000');
    gasPrice = gasPrice + (gasPrice * bumpPct) / 100n + bumpWei;

    txOptions.type = 0;
    txOptions.gasPrice = gasPrice;
    return txOptions;
  }
  
  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.relayerAddress = String(this.destinationSigner.address);
    
    this.logger.info('Relayer address: %s', this.relayerAddress);
    
    // 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('requested too many blocks') ||
      msg.includes('maximum is set to') ||
      (msg.includes('-32000') && (
        msg.includes('range') ||
        msg.includes('too many blocks') ||
        msg.includes('maximum is set to')
      ))
    );
  }

  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 (this.shouldSkipMessageId(messageId)) {
            this.logger.warn(`Skipping MessageSent ${messageId}; message id is in RELAY_SKIP_MESSAGE_IDS`);
            return;
          }
          
          if (destSelector !== expectedSelector) {
            this.logger.debug(`Ignoring message for different chain: ${destSelector}`);
            return;
          }

          const scope = this.evaluateMessageScope({ messageId, sender, receiver });
          if (!scope.inScope) {
            this.logger.debug(`Ignoring message ${messageId}; ${scope.reason}`);
            return;
          }
          
          this.logger.info('MessageSent event detected:', {
            messageId: messageId,
            destinationChainSelector: destinationChainSelector.toString(),
            sender: sender,
            targetBridge: scope.targetBridge,
            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,
            targetBridge: scope.targetBridge,
            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.lastSourcePoll = {
            at: new Date().toISOString(),
            ok: true,
            from_block: startBlock,
            to_block: toBlock,
            logs_fetched: logs.length
          };
          if (this.lastError && this.lastError.scope === 'source_poll') {
            this.lastError = null;
          }

          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 (this.shouldSkipMessageId(messageId)) {
              this.logger.warn(`Skipping historical MessageSent ${messageId}; message id is in RELAY_SKIP_MESSAGE_IDS`);
              continue;
            }

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

            const scope = this.evaluateMessageScope({ messageId, sender, receiver });
            if (!scope.inScope) {
              this.logger.debug(`Ignoring historical message ${messageId}; ${scope.reason}`);
              continue;
            }

            this.logger.info('Historical MessageSent detected:', {
              messageId,
              destinationChainSelector: destSelector,
              sender,
              targetBridge: scope.targetBridge,
              blockNumber: log.blockNumber,
              transactionHash: log.transactionHash
            });
            this.lastSeenMessage = {
              at: new Date().toISOString(),
              message_id: messageId,
              destination_chain_selector: destSelector,
              sender,
              block_number: log.blockNumber,
              transaction_hash: log.transactionHash
            };

            await this.messageQueue.add({
              messageId,
              destinationChainSelector,
              sender,
              receiver,
              targetBridge: scope.targetBridge,
              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);
        this.lastSourcePoll = {
          at: new Date().toISOString(),
          ok: false,
          from_block: startBlock,
          error: RelayService.summarizeError(error)
        };
        this.recordError('source_poll', error, {
          from_block: startBlock
        });
        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;

    if (this.shouldSkipMessageId(messageId)) {
      this.logger.warn(`Skipping queued message ${messageId}; message id is in RELAY_SKIP_MESSAGE_IDS`);
      await this.messageQueue.markProcessed(messageId);
      return null;
    }

    const scope = this.evaluateMessageScope(messageData);
    if (!scope.inScope) {
      this.logger.info(`Skipping queued message ${messageId}; ${scope.reason}`);
      await this.messageQueue.markProcessed(messageId);
      return null;
    }
    
    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.
      const targetBridge = scope.targetBridge || this.resolveTargetBridge(receiver);
      if (!targetBridge) {
        throw new Error(`No destination bridge for message ${messageId}: receiver decode failed and DEST_RELAY_BRIDGE not set`);
      }
      this.lastRelayAttempt = {
        at: new Date().toISOString(),
        message_id: messageId,
        destination_chain_selector: destinationChainSelector ? destinationChainSelector.toString() : '',
        target_bridge: targetBridge,
        token_count: tokenAmounts.length
      };

      // Optional allowlist hardening.
      const allowlist = this.getDestinationBridgeAllowlist();
      if (allowlist.length > 0 && !allowlist.includes(String(targetBridge).toLowerCase())) {
        this.logger.info(
          `Skipping message ${messageId} for bridge ${targetBridge}; not in this worker's DEST_RELAY_BRIDGE_ALLOWLIST`
        );
        await this.messageQueue.markProcessed(messageId);
        return null;
      }

      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)
        };
      });

      const inventoryCheck = await this.ensureTargetBridgeInventory(
        messageId,
        targetBridge,
        mappedTokenAmounts
      );
      if (!inventoryCheck.ok) {
        const inventoryError = new Error(inventoryCheck.message);
        this.logger.warn(inventoryCheck.message);
        this.recordError('bridge_inventory', inventoryError, {
          message_id: messageId,
          target_bridge: targetBridge,
          token: inventoryCheck.token,
          available_amount: inventoryCheck.availableAmount.toString(),
          required_amount: inventoryCheck.requiredAmount.toString(),
          shortfall: inventoryCheck.shortfall.toString()
        });
        await this.messageQueue.retry(messageId);
        await new Promise(resolve => setTimeout(resolve, this.config.retry.retryDelay));
        return null;
      }
      
      // 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
      });
      
      const txOptions = await this.getDestinationTxOptions();
      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, txOptions);
      } else {
        // Call relay router with properly formatted struct
        tx = await this.destinationRelayRouter.relayMessage(
          targetBridge,
          any2EVMMessage,
          txOptions
        );
      }
      
      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}`);
      this.lastRelaySuccess = {
        at: new Date().toISOString(),
        message_id: messageId,
        destination_chain_selector: destinationChainSelector ? destinationChainSelector.toString() : '',
        target_bridge: targetBridge,
        tx_hash: receipt.hash
      };
      if (this.lastError && (this.lastError.scope === 'relay_message' || this.lastError.scope === 'bridge_inventory')) {
        this.lastError = null;
      }
      
      // Mark message as processed
      await this.messageQueue.markProcessed(messageId);
      
      return receipt;
      
    } catch (error) {
      this.logger.error(`Error relaying message ${messageId}:`, error);
      this.recordError('relay_message', error, {
        message_id: messageId
      });
      
      // 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;
    }
  }
}