feat: registry event tracker

Cargo.toml

@@ -39,6 +39,7 @@ libp2p = { version = "0.53.2", features = [
 num-bigint = "0.4.4"
 serde = "1.0.197"
 serde_json = "1.0.115"
+starknet = "0.9.0"
 tempfile = "3.10.1"
 thiserror = "1.0.58"
 tokio = { version = "1.36", features = ["full"] }

crates/peer/Cargo.toml

@@ -13,3 +13,4 @@ tokio.workspace = true
 tracing-subscriber.workspace = true
 tracing.workspace = true
 sharp-p2p-common.workspace = true
+starknet.workspace = true

crates/peer/src/lib.rs

@@ -1,4 +1,5 @@
 pub mod network;
 pub mod node;
+pub mod registry;
 pub mod store;
 pub mod swarm;

crates/peer/src/node.rs

@@ -2,8 +2,8 @@ use libp2p::identity::Keypair;
 use libp2p::Multiaddr;
 use std::error::Error;
 use std::sync::Arc;
-use tokio::spawn;
 
+use crate::registry::RegistryHandler;
 use crate::store::Store;
 use crate::swarm::SwarmRunner;
 
@@ -44,9 +44,18 @@ pub struct Node {
 impl Node {
     pub async fn new(node_config: NodeConfig) -> Result<Self, Box<dyn Error>> {
         let mut swarm_runner = SwarmRunner::new(&node_config)?;
-        spawn(async move {
+        let registry_handler = RegistryHandler::new(
+            "https://starknet-sepolia.public.blastapi.io",
+            "0xcdd51fbc4e008f4ef807eaf26f5043521ef5931bbb1e04032a25bd845d286b",
+        );
+        // Node should run swarm runner and registry handler concurrently.
+        tokio::spawn(async move {
             swarm_runner.run(node_config.node_type).await;
         });
+        tokio::spawn(async move {
+            registry_handler.run().await;
+        });
+
         let store = Arc::new(node_config.store);
 
         Ok(Self { store })

crates/peer/src/registry.rs

@@ -0,0 +1,72 @@
+use std::error::Error;
+
+use starknet::{
+    core::types::{BlockId, EmittedEvent, EventFilter, FieldElement},
+    providers::{jsonrpc::HttpTransport, JsonRpcClient, Provider, Url},
+};
+
+pub struct RegistryHandler {
+    pub provider: JsonRpcClient<HttpTransport>,
+    address: FieldElement,
+}
+
+impl RegistryHandler {
+    pub fn new(url: &str, address: &str) -> Self {
+        let provider = JsonRpcClient::new(HttpTransport::new(Url::parse(url).unwrap()));
+        let address = FieldElement::from_hex_be(address).unwrap();
+        Self { provider, address }
+    }
+
+    async fn scrape_event(
+        &self,
+        event_keys: Vec<String>,
+        from_block: u64,
+    ) -> Result<Vec<EmittedEvent>, Box<dyn Error>> {
+        let keys = event_keys
+            .iter()
+            .map(|key| FieldElement::from_hex_be(key))
+            .collect::<Result<Vec<FieldElement>, _>>()?;
+
+        let latest_block_number = self.provider.block_number().await?;
+
+        let filter = EventFilter {
+            from_block: Some(BlockId::Number(from_block)),
+            to_block: Some(BlockId::Number(latest_block_number)),
+            address: Some(self.address),
+            keys: Some(vec![keys.clone()]),
+        };
+
+        let events = self.provider.get_events(filter, None, 1000).await?.events;
+        Ok(events)
+    }
+
+    pub async fn run(&self) {
+        // Create an interval of every 5 seconds
+        let mut interval = tokio::time::interval(tokio::time::Duration::from_secs(5));
+
+        loop {
+            interval.tick().await;
+
+            println!("Scraping events...");
+
+            // Scrape the event
+            let result = self
+                .scrape_event(
+                    vec!["0x17ef19eae2188756c1689ef60586c692a3aee6fecc18ee1b21f3028f75b9988"
+                        .to_string()],
+                    0,
+                )
+                .await;
+
+            // Handle the result
+            match result {
+                Ok(events) => {
+                    println!("{} Events Found", events.len());
+                }
+                Err(e) => {
+                    eprintln!("Error scraping events: {:?}", e);
+                }
+            }
+        }
+    }
+}