feat: exex for xpl balance update webhook

Cargo.toml

@@ -10,6 +10,7 @@ exclude = [".github/"]
 [workspace]
 members = [
     "bin/reth-bench/",
+    "bin/reth-xpl-webhook/",
     "bin/reth/",
     "crates/storage/rpc-provider/",
     "crates/chain-state/",

bin/reth-xpl-webhook/Cargo.toml

@@ -0,0 +1,54 @@
+[package]
+name = "reth-xpl-webhook"
+version.workspace = true
+edition.workspace = true
+rust-version.workspace = true
+license.workspace = true
+homepage.workspace = true
+repository.workspace = true
+exclude.workspace = true
+publish = false
+
+[lints]
+workspace = true
+
+[[bin]]
+name = "reth-xpl-webhook"
+path = "src/main.rs"
+
+[dependencies]
+# reth
+reth-ethereum = { workspace = true, features = ["full", "cli", "node-api"] }
+reth-exex.workspace = true
+reth-libmdbx.workspace = true
+reth-revm.workspace = true
+reth-storage-api.workspace = true
+
+# revm / evm
+revm-inspectors.workspace = true
+alloy-evm.workspace = true
+
+# alloy
+alloy-primitives.workspace = true
+alloy-consensus.workspace = true
+
+# crypto
+hmac.workspace = true
+sha2.workspace = true
+
+# async + http
+tokio = { workspace = true, features = ["sync", "rt-multi-thread", "macros", "time"] }
+futures.workspace = true
+reqwest = { workspace = true, features = ["json"] }
+
+# misc
+eyre.workspace = true
+tracing.workspace = true
+metrics.workspace = true
+serde = { workspace = true, features = ["derive"] }
+serde_json = { workspace = true, features = ["std"] }
+clap = { workspace = true, features = ["derive", "env"] }
+rand.workspace = true
+
+[dev-dependencies]
+tempfile = "3"

bin/reth-xpl-webhook/src/args.rs

@@ -0,0 +1,44 @@
+use alloy_primitives::Address;
+use clap::Parser;
+use std::path::PathBuf;
+
+/// CLI extension flags appended to the standard `reth` argument set. Each
+/// argument ID is namespaced with the `xpl_webhook_` prefix so it doesn't
+/// collide with the built-in node flags (e.g. era `--url`).
+#[derive(Debug, Clone, Parser)]
+pub struct XplWebhookArgs {
+    /// Webhook URL that receives `{ result: Transfer[] }` POSTs.
+    #[arg(id = "xpl_webhook_url", long = "xpl-webhook-url", env = "XPL_WEBHOOK_URL")]
+    pub url: String,
+
+    /// HMAC-SHA256 token used to sign requests. Read from `XPL_WEBHOOK_TOKEN`
+    /// when the flag is omitted.
+    #[arg(id = "xpl_webhook_token", long = "xpl-webhook-token", env = "XPL_WEBHOOK_TOKEN")]
+    pub token: String,
+
+    /// Address emitted in the `address` field of every transfer row. Defaults
+    /// to the zero address (native XPL marker).
+    #[arg(
+        id = "xpl_webhook_native_address",
+        long = "xpl-webhook-native-address",
+        default_value = "0x0000000000000000000000000000000000000000"
+    )]
+    pub native_address: Address,
+
+    /// Path to the MDBX outbox directory. Defaults to `<datadir>/xpl-webhook`.
+    #[arg(id = "xpl_webhook_outbox_path", long = "xpl-webhook-outbox-path")]
+    pub outbox_path: Option<PathBuf>,
+
+    /// Maximum number of HTTP delivery attempts per batch before backing off
+    /// to the configured ceiling.
+    #[arg(id = "xpl_webhook_max_retries", long = "xpl-webhook-max-retries", default_value_t = 8)]
+    pub max_retries: u32,
+
+    /// HTTP request timeout in milliseconds.
+    #[arg(
+        id = "xpl_webhook_http_timeout_ms",
+        long = "xpl-webhook-http-timeout-ms",
+        default_value_t = 5_000
+    )]
+    pub http_timeout_ms: u64,
+}

bin/reth-xpl-webhook/src/exex.rs

@@ -0,0 +1,92 @@
+use crate::{metrics as m, outbox::Outbox, replay::replay_block};
+use alloy_consensus::BlockHeader;
+use alloy_primitives::Address;
+use futures::TryStreamExt;
+use reth_ethereum::{
+    exex::{ExExContext, ExExEvent, ExExNotification},
+    node::api::FullNodeComponents,
+};
+use tracing::{debug, error, info, warn};
+
+/// The ExEx future. Loops over canonical chain notifications, replays each
+/// committed block, persists rows to the outbox, and signals
+/// [`ExExEvent::FinishedHeight`] only after the rows are durable.
+pub async fn xpl_webhook_exex<Node: FullNodeComponents>(
+    mut ctx: ExExContext<Node>,
+    outbox: Outbox,
+    native_address: Address,
+) -> eyre::Result<()> {
+    info!(target: "xpl-webhook", "ExEx started");
+
+    while let Some(notification) = ctx.notifications.try_next().await? {
+        match &notification {
+            ExExNotification::ChainCommitted { new } => {
+                for block in new.blocks_iter() {
+                    let block_number = block.number();
+                    match replay_block(&ctx, block, native_address) {
+                        Ok(rows) => {
+                            metrics::counter!(m::BLOCKS_PROCESSED).increment(1);
+                            if rows.is_empty() {
+                                continue;
+                            }
+                            metrics::counter!(m::ROWS_EMITTED).increment(rows.len() as u64);
+                            debug!(
+                                target: "xpl-webhook",
+                                block_number,
+                                rows = rows.len(),
+                                "persisting batch",
+                            );
+                            if let Err(err) =
+                                outbox.append_block_batch(block.hash(), block_number, rows)
+                            {
+                                error!(target: "xpl-webhook", %err, "outbox write failed");
+                            }
+                        }
+                        Err(err) => {
+                            metrics::counter!(m::REPLAY_ERRORS).increment(1);
+                            error!(target: "xpl-webhook", block_number, %err, "replay failed");
+                        }
+                    }
+                }
+            }
+            ExExNotification::ChainReorged { old, new } => {
+                metrics::counter!(m::REORGS).increment(1);
+                warn!(
+                    target: "xpl-webhook",
+                    old = ?old.range(),
+                    new = ?new.range(),
+                    "reorg: discarding pending rows for old chain segment",
+                );
+                if let Err(err) = outbox.discard_blocks(old.blocks_iter().map(|b| b.hash())) {
+                    error!(target: "xpl-webhook", %err, "failed to discard reorged blocks");
+                }
+                for block in new.blocks_iter() {
+                    let block_number = block.number();
+                    if let Ok(rows) = replay_block(&ctx, block, native_address) &&
+                        !rows.is_empty()
+                    {
+                        let _ = outbox.append_block_batch(block.hash(), block_number, rows);
+                    }
+                }
+            }
+            ExExNotification::ChainReverted { old } => {
+                metrics::counter!(m::REVERTS).increment(1);
+                warn!(
+                    target: "xpl-webhook",
+                    old = ?old.range(),
+                    "revert: discarding pending rows",
+                );
+                if let Err(err) = outbox.discard_blocks(old.blocks_iter().map(|b| b.hash())) {
+                    error!(target: "xpl-webhook", %err, "failed to discard reverted blocks");
+                }
+            }
+        }
+
+        if let Some(committed) = notification.committed_chain() {
+            ctx.events.send(ExExEvent::FinishedHeight(committed.tip().num_hash()))?;
+        }
+    }
+
+    info!(target: "xpl-webhook", "ExEx exiting");
+    Ok(())
+}