smom-dbis-138/watcher/src/index.ts

/**
 * CCIP Watcher and Relayer
 * Production-grade service to watch Chain-138 and relay transactions to Ethereum via CCIP
 */

import { ethers } from "ethers";
import { Pool } from "pg";
import * as dotenv from "dotenv";
import winston from "winston";

dotenv.config();

// Configuration
const CONFIG = {
  // Chain-138 RPC
  CHAIN138_RPC_WS: process.env.CHAIN138_RPC_WS || process.env.CHAIN138_RPC_URL || "",
  CHAIN138_RPC_HTTP: process.env.CHAIN138_RPC_HTTP || process.env.CHAIN138_RPC_URL || "",
  
  // Ethereum RPC
  ETHEREUM_RPC: process.env.ETHEREUM_MAINNET_RPC || process.env.ETHEREUM_RPC || "",
  
  // Contract addresses
  CCIP_REPORTER_ADDRESS: process.env.CCIP_REPORTER_CHAIN138_ADDRESS || "",
  CCIP_LOGGER_ADDRESS: process.env.CCIP_LOGGER_ETH_ADDRESS || "",
  
  // Chain selectors (from CCIP Directory)
  CHAIN138_SELECTOR: process.env.CHAIN138_SELECTOR || "0x000000000000008a",
  ETH_MAINNET_SELECTOR: process.env.ETH_MAINNET_SELECTOR || "0x500147",
  
  // Relayer configuration
  RELAYER_PRIVATE_KEY: process.env.RELAYER_PRIVATE_KEY || "",
  BATCH_SIGNER_PRIVATE_KEY: process.env.BATCH_SIGNER_PRIVATE_KEY || "",
  
  // Batching configuration
  BATCH_SIZE: parseInt(process.env.BATCH_SIZE || "10"),
  BATCH_INTERVAL_MS: parseInt(process.env.BATCH_INTERVAL_MS || "60000"), // 1 minute
  
  // Confirmation requirements
  CONFIRMATIONS_REQUIRED: parseInt(process.env.CONFIRMATIONS_REQUIRED || "12"),
  
  // Database
  DATABASE_URL: process.env.DATABASE_URL || "postgresql://user:password@localhost:5432/ccip_relayer",
  
  // Monitoring
  METRICS_PORT: parseInt(process.env.METRICS_PORT || "9090"),
};

// Logger setup
const logger = winston.createLogger({
  level: process.env.LOG_LEVEL || "info",
  format: winston.format.combine(
    winston.format.timestamp(),
    winston.format.json()
  ),
  transports: [
    new winston.transports.Console({
      format: winston.format.simple(),
    }),
  ],
});

// Database connection
const pool = new Pool({
  connectionString: CONFIG.DATABASE_URL,
});

// Initialize database schema
async function initDatabase() {
  await pool.query(`
    CREATE TABLE IF NOT EXISTS outbox (
      id SERIAL PRIMARY KEY,
      tx_hash VARCHAR(66) UNIQUE NOT NULL,
      from_address VARCHAR(42) NOT NULL,
      to_address VARCHAR(42) NOT NULL,
      value NUMERIC NOT NULL,
      block_number BIGINT NOT NULL,
      observed_at TIMESTAMP DEFAULT NOW(),
      confirmed_at TIMESTAMP,
      relayed BOOLEAN DEFAULT FALSE,
      relayed_txhash VARCHAR(66),
      batch_id VARCHAR(66),
      attempts INT DEFAULT 0,
      last_error TEXT
    );
    
    CREATE INDEX IF NOT EXISTS idx_outbox_relayed ON outbox(relayed);
    CREATE INDEX IF NOT EXISTS idx_outbox_confirmed ON outbox(confirmed_at);
    CREATE INDEX IF NOT EXISTS idx_outbox_batch ON outbox(batch_id);
  `);
  
  logger.info("Database initialized");
}

// Watch Chain-138 for transactions
async function watchChain138() {
  if (!CONFIG.CHAIN138_RPC_WS && !CONFIG.CHAIN138_RPC_HTTP) {
    throw new Error("Chain-138 RPC URL not configured");
  }

  const provider = CONFIG.CHAIN138_RPC_WS
    ? new ethers.WebSocketProvider(CONFIG.CHAIN138_RPC_WS)
    : new ethers.JsonRpcProvider(CONFIG.CHAIN138_RPC_HTTP);

  logger.info("Watching Chain-138 for transactions...");

  // Subscribe to new blocks
  provider.on("block", async (blockNumber) => {
    try {
      const block = await provider.getBlock(blockNumber, true);
      if (!block || !block.transactions) return;

      // Process transactions in block
      for (const tx of block.transactions) {
        if (typeof tx === "string") continue; // Skip if just hash

        // Check if transaction matches criteria (you can add filters here)
        const txReceipt = await provider.getTransactionReceipt(tx.hash);
        if (!txReceipt || txReceipt.status !== 1) continue; // Skip failed txs

        // Insert into outbox (idempotent)
        await pool.query(
          `INSERT INTO outbox (tx_hash, from_address, to_address, value, block_number, confirmed_at)
           VALUES ($1, $2, $3, $4, $5, NOW())
           ON CONFLICT (tx_hash) DO NOTHING`,
          [
            tx.hash,
            tx.from || ethers.ZeroAddress,
            tx.to || ethers.ZeroAddress,
            tx.value?.toString() || "0",
            blockNumber,
          ]
        );

        logger.debug(`Observed transaction: ${tx.hash}`);
      }
    } catch (error) {
      logger.error("Error processing block:", error);
    }
  });
}

// Batch dispatcher
async function dispatchBatches() {
  if (!CONFIG.CCIP_REPORTER_ADDRESS || !CONFIG.RELAYER_PRIVATE_KEY) {
    throw new Error("CCIP Reporter address or relayer key not configured");
  }

  const provider = new ethers.JsonRpcProvider(CONFIG.CHAIN138_RPC_HTTP);
  const wallet = new ethers.Wallet(CONFIG.RELAYER_PRIVATE_KEY, provider);
  const reporter = new ethers.Contract(
    CONFIG.CCIP_REPORTER_ADDRESS,
    [
      "function reportBatch(bytes32,bytes32[],address[],address[],uint256[],bytes) payable",
      "function estimateFee(bytes32[],address[],address[],uint256[],bytes) view returns(uint256)",
    ],
    wallet
  );

  setInterval(async () => {
    try {
      // Get unrelayed, confirmed transactions
      const result = await pool.query(
        `SELECT * FROM outbox 
         WHERE relayed = FALSE 
         AND confirmed_at IS NOT NULL
         ORDER BY block_number ASC
         LIMIT $1`,
        [CONFIG.BATCH_SIZE]
      );

      if (result.rows.length === 0) return;

      // Prepare batch data
      const txHashes: string[] = [];
      const froms: string[] = [];
      const tos: string[] = [];
      const values: string[] = [];

      for (const row of result.rows) {
        txHashes.push(row.tx_hash);
        froms.push(row.from_address);
        tos.push(row.to_address);
        values.push(row.value);
      }

      // Generate batch ID
      const batchId = ethers.keccak256(
        ethers.AbiCoder.defaultAbiCoder().encode(
          ["bytes32[]", "uint256"],
          [txHashes, Date.now()]
        )
      );

      // Sign batch (if signer key configured)
      let signature = "0x";
      if (CONFIG.BATCH_SIGNER_PRIVATE_KEY) {
        const signerWallet = new ethers.Wallet(CONFIG.BATCH_SIGNER_PRIVATE_KEY);
        const digest = ethers.keccak256(
          ethers.AbiCoder.defaultAbiCoder().encode(
            ["bytes32", "bytes32", "bytes32", "bytes32", "bytes32"],
            [
              batchId,
              ethers.keccak256(ethers.AbiCoder.defaultAbiCoder().encode(["bytes32[]"], [txHashes])),
              ethers.keccak256(ethers.AbiCoder.defaultAbiCoder().encode(["address[]"], [froms])),
              ethers.keccak256(ethers.AbiCoder.defaultAbiCoder().encode(["address[]"], [tos])),
              ethers.keccak256(ethers.AbiCoder.defaultAbiCoder().encode(["uint256[]"], [values])),
            ]
          )
        );
        const messageHash = ethers.hashMessage(ethers.getBytes(digest));
        signature = await signerWallet.signMessage(ethers.getBytes(messageHash));
      }

      // Estimate fee
      const fee = await reporter.estimateFee(
        txHashes,
        froms,
        tos,
        values,
        signature
      );

      // Check balance
      const balance = await provider.getBalance(wallet.address);
      if (balance < fee) {
        logger.warn(`Insufficient balance. Required: ${ethers.formatEther(fee)} ETH, Have: ${ethers.formatEther(balance)} ETH`);
        return;
      }

      // Send batch
      logger.info(`Relaying batch ${batchId} with ${txHashes.length} transactions`);
      const tx = await reporter.reportBatch(
        batchId,
        txHashes,
        froms,
        tos,
        values,
        signature,
        { value: fee }
      );

      const receipt = await tx.wait();
      logger.info(`Batch relayed. TX: ${receipt.hash}`);

      // Update outbox
      await pool.query(
        `UPDATE outbox 
         SET relayed = TRUE, relayed_txhash = $1, batch_id = $2
         WHERE tx_hash = ANY($3::varchar[])`,
        [receipt.hash, batchId, txHashes]
      );

      logger.info(`Batch ${batchId} successfully relayed`);
    } catch (error) {
      logger.error("Error dispatching batch:", error);
      // Update attempts
      // (Implementation for retry logic)
    }
  }, CONFIG.BATCH_INTERVAL_MS);
}

// Main function
async function main() {
  logger.info("Starting CCIP Watcher and Relayer...");
  
  try {
    await initDatabase();
    await watchChain138();
    await dispatchBatches();
    
    logger.info("CCIP Watcher and Relayer started successfully");
  } catch (error) {
    logger.error("Failed to start:", error);
    process.exit(1);
  }
}

// Handle graceful shutdown
process.on("SIGINT", async () => {
  logger.info("Shutting down...");
  await pool.end();
  process.exit(0);
});

if (require.main === module) {
  main();
}

export { main, watchChain138, dispatchBatches };