feat(indexer): implement EVM indexer runtime

indexer/Cargo.toml

@@ -8,6 +8,8 @@ seraph-shared       = { workspace = true }
 alloy               = { workspace = true }
 anyhow              = { workspace = true }
 dotenvy             = { workspace = true }
+futures             = "0.3"
+sqlx                = { workspace = true }
 reqwest             = { workspace = true }
 serde               = { workspace = true }
 serde_json          = { workspace = true }

indexer/src/adapters/mod.rs

@@ -0,0 +1,30 @@
+use seraph_shared::{CrossChainEvent, RawLog};
+
+/// Enum-dispatched set of active adapters.
+///
+/// Add one variant per bridge protocol. Never use `dyn ProtocolAdapter` here —
+/// the compiler monomorphizes each arm so there is zero heap allocation on the
+/// hot path.
+pub enum AdapterDispatch {
+    /// Placeholder that matches nothing. Remove once real adapters land.
+    NoOp,
+}
+
+impl AdapterDispatch {
+    /// Try to parse a raw log with this adapter. Returns `None` if the log
+    /// does not belong to this protocol.
+    pub fn parse_event(&self, _log: &RawLog) -> Option<CrossChainEvent> {
+        match self {
+            Self::NoOp => None,
+        }
+    }
+
+    /// Derive the correlation key used by the stitcher to join source and
+    /// destination events.
+    #[allow(dead_code)]
+    pub fn correlation_id(&self, event: &CrossChainEvent) -> String {
+        match self {
+            Self::NoOp => event.correlation_id.clone(),
+        }
+    }
+}

indexer/src/log.rs

@@ -0,0 +1,15 @@
+use seraph_shared::{ChainId, RawLog};
+
+/// Convert an alloy RPC log into our chain-agnostic RawLog.
+pub fn to_raw(log: &alloy::rpc::types::Log, chain_id: ChainId) -> RawLog {
+    let inner = &log.inner;
+    RawLog {
+        address: inner.address.to_string(),
+        topics: inner.topics().iter().map(|t| t.to_string()).collect(),
+        data: inner.data.data.to_vec(),
+        block_number: log.block_number,
+        tx_hash: log.transaction_hash.map(|h| h.to_string()),
+        log_index: log.log_index,
+        chain_id,
+    }
+}

indexer/src/main.rs

@@ -1,3 +1,134 @@
-fn main() {
-    println!("Hello, world!");
+mod adapters;
+mod log;
+mod provider;
+mod runner;
+
+use anyhow::Context;
+use seraph_shared::{Config, chain, db};
+use tracing::{error, info};
+
+use adapters::AdapterDispatch;
+use runner::ChainRunner;
+
+struct ChainConfig {
+    chain_id: seraph_shared::ChainId,
+    wss_url: String,
+    /// First block included in the backfill pass.
+    /// Set EVM_START_BLOCK to override; defaults to the current tip (no backfill).
+    start_block: u64,
+}
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    tracing_subscriber::fmt()
+        .with_env_filter(
+            tracing_subscriber::EnvFilter::try_from_default_env()
+                .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")),
+        )
+        .init();
+
+    dotenvy::dotenv().ok();
+
+    let config = Config::from_env().context("failed to load config")?;
+
+    info!("connecting to database");
+    let pool = db::connect(&config.database_url)
+        .await
+        .context("failed to connect to database")?;
+
+    info!("running migrations");
+    db::run_migrations(&pool)
+        .await
+        .context("failed to run migrations")?;
+
+    // EVM_START_BLOCK lets operators replay from a known block.
+    // Defaults to u64::MAX so the runner skips the backfill pass and only
+    // picks up live events — safe when there is no high-water mark yet.
+    let start_block: u64 = std::env::var("EVM_START_BLOCK")
+        .ok()
+        .and_then(|v| v.parse().ok())
+        .unwrap_or(u64::MAX);
+
+    let key = &config.alchemy_api_key;
+    let chains = vec![
+        ChainConfig {
+            chain_id: chain::ethereum(),
+            wss_url: format!("wss://eth-mainnet.g.alchemy.com/v2/{key}"),
+            start_block,
+        },
+        ChainConfig {
+            chain_id: chain::arbitrum(),
+            wss_url: format!("wss://arb-mainnet.g.alchemy.com/v2/{key}"),
+            start_block,
+        },
+        ChainConfig {
+            chain_id: chain::optimism(),
+            wss_url: format!("wss://opt-mainnet.g.alchemy.com/v2/{key}"),
+            start_block,
+        },
+        ChainConfig {
+            chain_id: chain::base(),
+            wss_url: format!("wss://base-mainnet.g.alchemy.com/v2/{key}"),
+            start_block,
+        },
+        ChainConfig {
+            chain_id: chain::polygon(),
+            wss_url: format!("wss://polygon-mainnet.g.alchemy.com/v2/{key}"),
+            start_block,
+        },
+        ChainConfig {
+            chain_id: chain::bsc(),
+            wss_url: format!("wss://bnb-mainnet.g.alchemy.com/v2/{key}"),
+            start_block,
+        },
+        ChainConfig {
+            chain_id: chain::avalanche(),
+            wss_url: format!("wss://avax-mainnet.g.alchemy.com/v2/{key}"),
+            start_block,
+        },
+    ];
+
+    let mut handles = Vec::new();
+
+    for chain in chains {
+        let pool = pool.clone();
+
+        let handle = tokio::spawn(async move {
+            let provider = match provider::connect(&chain.wss_url).await {
+                Ok(p) => p,
+                Err(e) => {
+                    error!(chain = %chain.chain_id, error = %e, "provider connection failed");
+                    return;
+                }
+            };
+
+            let runner = ChainRunner {
+                chain_id: chain.chain_id.clone(),
+                provider,
+                adapters: vec![AdapterDispatch::NoOp],
+                watched_addresses: vec![],
+                start_block: chain.start_block,
+                pool,
+            };
+
+            if let Err(e) = runner.run().await {
+                error!(chain = %chain.chain_id, error = %e, "chain runner exited with error");
+            }
+        });
+
+        handles.push(handle);
+    }
+
+    info!("indexer running — press Ctrl+C to stop");
+    tokio::signal::ctrl_c()
+        .await
+        .context("failed to listen for Ctrl+C")?;
+
+    info!("shutting down");
+    for handle in handles {
+        handle.abort();
+    }
+    pool.close().await;
+
+    Ok(())
 }

indexer/src/provider.rs

@@ -0,0 +1,16 @@
+use alloy::providers::{Provider, ProviderBuilder};
+use alloy::transports::ws::WsConnect;
+use anyhow::Result;
+use tracing::info;
+
+/// Connect to an EVM node via WebSocket and confirm with a chain_id call.
+pub async fn connect(url: &str) -> Result<impl Provider + Clone + 'static> {
+    let provider = ProviderBuilder::new()
+        .connect_ws(WsConnect::new(url))
+        .await?;
+
+    let chain_id = provider.get_chain_id().await?;
+    info!(chain_id, "connected to EVM node");
+
+    Ok(provider)
+}

indexer/src/runner.rs

@@ -0,0 +1,115 @@
+use alloy::primitives::Address;
+use alloy::providers::Provider;
+use alloy::rpc::types::{BlockNumberOrTag, Filter};
+use anyhow::Result;
+use futures::StreamExt;
+use sqlx::PgPool;
+use tracing::{debug, error, info};
+
+use seraph_shared::{ChainId, db, db::CrossChainEventRow};
+
+use crate::adapters::AdapterDispatch;
+use crate::log;
+
+/// Alchemy's eth_getLogs limit: 2 000 blocks or 10 000 events per call.
+const CHUNK_SIZE: u64 = 2_000;
+
+pub struct ChainRunner<P> {
+    pub chain_id: ChainId,
+    pub provider: P,
+    pub adapters: Vec<AdapterDispatch>,
+    /// Contract addresses to watch. Empty = watch all (use only for testing).
+    pub watched_addresses: Vec<Address>,
+    /// First block to include in the backfill pass.
+    pub start_block: u64,
+    pub pool: PgPool,
+}
+
+impl<P: Provider + Clone> ChainRunner<P> {
+    /// Run backfill then hand off to the live subscription. Blocks until the
+    /// WebSocket stream closes or an unrecoverable error occurs.
+    pub async fn run(self) -> Result<()> {
+        let tip = self.provider.get_block_number().await?;
+        info!(chain = %self.chain_id, tip, "chain runner started");
+
+        self.backfill(self.start_block, tip).await?;
+        self.subscribe(tip + 1).await
+    }
+
+    // ── backfill ──────────────────────────────────────────────────────────────
+
+    async fn backfill(&self, from: u64, to: u64) -> Result<()> {
+        if from > to {
+            return Ok(());
+        }
+        info!(chain = %self.chain_id, from, to, "backfill started");
+
+        let mut cursor = from;
+        while cursor <= to {
+            let end = (cursor + CHUNK_SIZE - 1).min(to);
+            self.fetch_chunk(cursor, end).await?;
+            cursor = end + 1;
+        }
+
+        info!(chain = %self.chain_id, "backfill complete");
+        Ok(())
+    }
+
+    async fn fetch_chunk(&self, from: u64, to: u64) -> Result<()> {
+        let filter = self.build_filter(
+            BlockNumberOrTag::Number(from),
+            Some(BlockNumberOrTag::Number(to)),
+        );
+        let logs = self.provider.get_logs(&filter).await?;
+        debug!(chain = %self.chain_id, from, to, count = logs.len(), "chunk fetched");
+
+        for l in &logs {
+            self.process(l).await;
+        }
+        Ok(())
+    }
+
+    // ── live subscription ─────────────────────────────────────────────────────
+
+    async fn subscribe(&self, from_block: u64) -> Result<()> {
+        info!(chain = %self.chain_id, from_block, "subscribing to live logs");
+
+        let filter = self.build_filter(BlockNumberOrTag::Number(from_block), None);
+        let sub = self.provider.subscribe_logs(&filter).await?;
+        let mut stream = sub.into_stream();
+
+        while let Some(log) = stream.next().await {
+            self.process(&log).await;
+        }
+
+        Ok(())
+    }
+
+    // ── shared helpers ────────────────────────────────────────────────────────
+
+    fn build_filter(&self, from: BlockNumberOrTag, to: Option<BlockNumberOrTag>) -> Filter {
+        let mut f = Filter::new().from_block(from);
+        if !self.watched_addresses.is_empty() {
+            f = f.address(self.watched_addresses.clone());
+        }
+        if let Some(to) = to {
+            f = f.to_block(to);
+        }
+        f
+    }
+
+    async fn process(&self, log: &alloy::rpc::types::Log) {
+        let raw = log::to_raw(log, self.chain_id.clone());
+
+        for adapter in &self.adapters {
+            if let Some(event) = adapter.parse_event(&raw) {
+                let row = CrossChainEventRow::from(&event);
+                match db::insert_event(&self.pool, &row).await {
+                    Ok(_) => debug!(tx = %event.source_tx_hash, "event inserted"),
+                    Err(e) => error!(error = %e, tx = %event.source_tx_hash, "insert failed"),
+                }
+                return;
+            }
+        }
+    }
+}