From 93b0a5e9abcef7448ec075fe6ea595f99ac5408a Mon Sep 17 00:00:00 2001 From: 0xOse <0xosepatrick@gmail.com> Date: Wed, 8 Jul 2026 12:00:05 +0100 Subject: [PATCH 1/3] feat(indexer): implement EVM indexer runtime Wires up the indexer binary with a full EVM event pipeline: - Tokio bootstrap with structured logging, DB connection, and migration runner - alloy-rs WebSocket provider per chain (generic over transport type) - RawLog conversion from alloy Log type - AdapterDispatch enum skeleton for zero-overhead protocol dispatch - ChainRunner with eth_getLogs backfill (2k-block chunks) and live subscribe_logs subscription - One Tokio task per chain (Ethereum, Arbitrum, Optimism, Base) with graceful Ctrl+C shutdown - EVM_START_BLOCK env var controls backfill depth; defaults to tip-only (no backfill) --- Cargo.lock | 2 + indexer/Cargo.toml | 2 + indexer/src/adapters/mod.rs | 30 +++++++++ indexer/src/log.rs | 15 +++++ indexer/src/main.rs | 120 +++++++++++++++++++++++++++++++++++- indexer/src/provider.rs | 14 +++++ indexer/src/runner.rs | 115 ++++++++++++++++++++++++++++++++++ 7 files changed, 296 insertions(+), 2 deletions(-) create mode 100644 indexer/src/adapters/mod.rs create mode 100644 indexer/src/log.rs create mode 100644 indexer/src/provider.rs create mode 100644 indexer/src/runner.rs diff --git a/Cargo.lock b/Cargo.lock index 70f9af8..93d0a52 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2599,10 +2599,12 @@ dependencies = [ "alloy", "anyhow", "dotenvy", + "futures", "reqwest 0.12.28", "seraph-shared", "serde", "serde_json", + "sqlx", "tokio", "tracing", "tracing-subscriber", diff --git a/indexer/Cargo.toml b/indexer/Cargo.toml index e678490..39cf5e3 100644 --- a/indexer/Cargo.toml +++ b/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 } diff --git a/indexer/src/adapters/mod.rs b/indexer/src/adapters/mod.rs new file mode 100644 index 0000000..4987148 --- /dev/null +++ b/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 { + 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(), + } + } +} diff --git a/indexer/src/log.rs b/indexer/src/log.rs new file mode 100644 index 0000000..7731439 --- /dev/null +++ b/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, + } +} diff --git a/indexer/src/main.rs b/indexer/src/main.rs index e7a11a9..3f1fcdb 100644 --- a/indexer/src/main.rs +++ b/indexer/src/main.rs @@ -1,3 +1,119 @@ -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, + }, + ]; + + 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(()) } diff --git a/indexer/src/provider.rs b/indexer/src/provider.rs new file mode 100644 index 0000000..655beee --- /dev/null +++ b/indexer/src/provider.rs @@ -0,0 +1,14 @@ +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 { + 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) +} diff --git a/indexer/src/runner.rs b/indexer/src/runner.rs new file mode 100644 index 0000000..f196597 --- /dev/null +++ b/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

{ + pub chain_id: ChainId, + pub provider: P, + pub adapters: Vec, + /// Contract addresses to watch. Empty = watch all (use only for testing). + pub watched_addresses: Vec

, + /// First block to include in the backfill pass. + pub start_block: u64, + pub pool: PgPool, +} + +impl ChainRunner

{ + /// 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) -> 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; + } + } + } +} From 1b269fa292c7825e945cae7675e0d357bbd41029 Mon Sep 17 00:00:00 2001 From: 0xOse <0xosepatrick@gmail.com> Date: Wed, 8 Jul 2026 12:11:10 +0100 Subject: [PATCH 2/3] style(provider): format connect function for improved readability --- indexer/src/provider.rs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/indexer/src/provider.rs b/indexer/src/provider.rs index 655beee..84d2ac5 100644 --- a/indexer/src/provider.rs +++ b/indexer/src/provider.rs @@ -5,7 +5,9 @@ use tracing::info; /// Connect to an EVM node via WebSocket and confirm with a chain_id call. pub async fn connect(url: &str) -> Result { - let provider = ProviderBuilder::new().connect_ws(WsConnect::new(url)).await?; + 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"); From 75d12426dab2b05b3110ca90c109a9d55037bf37 Mon Sep 17 00:00:00 2001 From: 0xOse <0xosepatrick@gmail.com> Date: Wed, 8 Jul 2026 13:44:16 +0100 Subject: [PATCH 3/3] feat(indexer): add support for additional chains in EVM indexer configuration --- indexer/src/main.rs | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/indexer/src/main.rs b/indexer/src/main.rs index 3f1fcdb..4ff0196 100644 --- a/indexer/src/main.rs +++ b/indexer/src/main.rs @@ -71,6 +71,21 @@ async fn main() -> anyhow::Result<()> { 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();