init

apps/benchmarks_keeper/Cargo.toml

@@ -0,0 +1,16 @@
+[package]
+name = "benchmarks_keeper"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+alloy = { version = "0.3", features = ["full"] }
+tokio = { version = "1.28", features = ["full"] }
+eyre = "0.6"
+tracing = "0.1"
+tracing-subscriber = { version = "0.3", features = ["env-filter"] }
+config = "0.13"
+serde = { version = "1.0", features = ["derive"] }
+thiserror = "1.0"
+futures = "0.3"
+futures-util = "0.3"

apps/benchmarks_keeper/config/default.toml

@@ -0,0 +1 @@
+rpc_url = "wss://optimism.drpc.org"

apps/benchmarks_keeper/src/config.rs

@@ -0,0 +1,18 @@
+use config::{Config as ConfigTrait, ConfigError, Environment, File};
+use serde::Deserialize;
+
+#[derive(Debug, Deserialize)]
+pub struct Config {
+    pub rpc_url: String,
+}
+
+impl Config {
+    pub fn new() -> Result<Self, ConfigError> {
+        let config = ConfigTrait::builder()
+            .add_source(File::with_name("config/default"))
+            .add_source(Environment::with_prefix("APP"))
+            .build()?;
+
+        config.try_deserialize()
+    }
+}

apps/benchmarks_keeper/src/lib.rs

@@ -0,0 +1,50 @@
+pub mod config;
+mod price_update;
+mod utils;
+
+use alloy::providers::{Provider, ProviderBuilder, WsConnect};
+use futures_util::StreamExt;
+use tracing::info;
+
+use crate::config::Config;
+use crate::price_update::PriceUpdate;
+use crate::utils::error::BenchmarksKeeperError;
+
+pub async fn run(config: Config) -> Result<(), BenchmarksKeeperError> {
+    info!("Starting Benchmarks Keeper");
+
+    // Create the WebSocket connection
+    let ws = WsConnect::new(&config.rpc_url);
+    let provider = ProviderBuilder::new()
+        .on_ws(ws)
+        .await
+        .map_err(|e| BenchmarksKeeperError::RpcConnectionError(e.to_string()))?;
+
+    // Subscribe to PriceUpdate events
+    let filter = PriceUpdate::filter();
+    let sub = provider.subscribe_logs(&filter).await?;
+
+    println!("Awaiting PriceUpdate events...");
+
+    // Process PriceUpdate events
+    let mut stream = sub.into_stream().take(4);
+    while let Some(log) = stream.next().await {
+        match PriceUpdate::decode_log(&log) {
+            Ok(price_update) => {
+                println!("Received PriceUpdate:");
+                println!("  Publish Time: {}", price_update.publish_time);
+                println!("  Number of Price IDs: {}", price_update.price_ids.len());
+                println!("  Metadata length: {} bytes", price_update.metadata.len());
+                // TODO: Process the price update
+            }
+            Err(e) => {
+                eprintln!(
+                    "Error decoding log: {}",
+                    BenchmarksKeeperError::EventDecodeError(e.to_string())
+                );
+            }
+        }
+    }
+
+    Ok(())
+}

apps/benchmarks_keeper/src/main.rs

@@ -0,0 +1,15 @@
+use benchmarks_keeper::{config::Config, run};
+use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    tracing_subscriber::registry()
+        .with(tracing_subscriber::EnvFilter::try_from_default_env().unwrap_or_else(|_| "info".into()))
+        .with(tracing_subscriber::fmt::layer())
+        .init();
+
+    let config = Config::new()?;
+    run(config).await?;
+
+    Ok(())
+}

apps/benchmarks_keeper/src/price_update.rs

@@ -0,0 +1,55 @@
+use alloy::primitives::{Address, Bytes, U256};
+use alloy::rpc::types::Filter;
+use alloy::rpc::types::Log;
+use eyre::Result;
+
+#[derive(Debug)]
+pub struct PriceUpdate {
+    pub publish_time: U256,
+    pub price_ids: Vec<[u8; 32]>,
+    pub metadata: Bytes,
+}
+
+impl PriceUpdate {
+    pub fn filter() -> Filter {
+        Filter::new()
+            .event(&Self::event_signature_str())
+            .address(Address::ZERO) // TODO: Replace with the actual contract address
+    }
+
+    pub fn decode_log(log: &Log) -> Result<Self> {
+        let data = log.data().data.as_ref();
+
+        // Assuming the event structure: PriceUpdate(uint256 publish_time, bytes32[] price_ids, bytes metadata)
+        let publish_time = U256::from_be_bytes::<32>(data[0..32].try_into()?);
+        let price_ids_offset = U256::from_be_bytes::<32>(data[32..64].try_into()?).to::<usize>();
+        let metadata_offset = U256::from_be_bytes::<32>(data[64..96].try_into()?).to::<usize>();
+
+        let price_ids_length =
+            U256::from_be_bytes::<32>(data[price_ids_offset..price_ids_offset + 32].try_into()?)
+                .to::<usize>();
+        let price_ids: Vec<[u8; 32]> = (0..price_ids_length)
+            .map(|i| {
+                let start = price_ids_offset + 32 + i * 32;
+                data[start..start + 32].try_into().unwrap()
+            })
+            .collect();
+
+        let metadata_length =
+            U256::from_be_bytes::<32>(data[metadata_offset..metadata_offset + 32].try_into()?)
+                .to::<usize>();
+        let metadata = Bytes::copy_from_slice(
+            &data[metadata_offset + 32..metadata_offset + 32 + metadata_length],
+        );
+
+        Ok(Self {
+            publish_time,
+            price_ids,
+            metadata,
+        })
+    }
+
+    fn event_signature_str() -> &'static str {
+        "PriceUpdate(uint256,bytes32[],bytes)"
+    }
+}

apps/benchmarks_keeper/src/rpc/ethereum.rs

@@ -0,0 +1,15 @@
+use alloy::providers::Provider;
+use eyre::Result;
+
+pub struct EthereumRpc {
+    provider: Provider,
+}
+
+impl EthereumRpc {
+    pub fn new(rpc_url: String) -> Result<Self> {
+        let provider = Provider::new(rpc_url);
+        Ok(Self { provider })
+    }
+
+    // Implement methods for interacting with Ethereum RPC
+}

apps/benchmarks_keeper/src/utils/error.rs

@@ -0,0 +1,18 @@
+use alloy::transports::TransportError;
+use eyre::Report;
+use thiserror::Error;
+
+#[derive(Error, Debug)]
+pub enum BenchmarksKeeperError {
+    #[error("Failed to connect to RPC: {0}")]
+    RpcConnectionError(String),
+
+    #[error("Failed to decode event log: {0}")]
+    EventDecodeError(String),
+
+    #[error("Transport error: {0}")]
+    TransportError(#[from] TransportError),
+
+    #[error("Other error: {0}")]
+    Other(#[from] Report),
+}

apps/benchmarks_keeper/src/utils/mod.rs

@@ -0,0 +1 @@
+pub mod error;