[feat] mev-share collector

bot/crates/artemis-core/Cargo.toml

@@ -1,21 +1,22 @@
 [package]
+edition = "2021"
 name = "artemis-core"
 version = "0.1.0"
-edition = "2021"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
 
 ## eth
-ethers = { version = "2", features = ["ws", "rustls"]}
-ethers-flashbots = { git = "https://github.com/onbjerg/ethers-flashbots", features = ["rustls"] }
+ethers = {version = "2", features = ["ws", "rustls"]}
+ethers-flashbots = {git = "https://github.com/onbjerg/ethers-flashbots", features = ["rustls"]}
+mev-share-sse = {git = "https://github.com/mattsse/mev-share-rs"}
 
 ## async
 async-trait = "0.1.64"
-reqwest = { version = "0.11.14", default-features = false, features = ["rustls-tls"] }
-tokio = { version = "1.18", features = ["full"] }
-tokio-stream = { version = "0.1", features = ['sync'] }
+reqwest = {version = "0.11.14", default-features = false, features = ["rustls-tls"]}
+tokio = {version = "1.18", features = ["full"]}
+tokio-stream = {version = "0.1", features = ['sync']}
 
 ## misc
 anyhow = "1.0.70"

bot/crates/artemis-core/src/collectors/mevshare_collector.rs

@@ -0,0 +1,34 @@
+use crate::types::{Collector, CollectorStream};
+use anyhow::Result;
+use async_trait::async_trait;
+pub use mev_share_sse::Event;
+use mev_share_sse::EventClient;
+use tokio_stream::StreamExt;
+
+/// A collector that streams from MEV-Share SSE endpoint
+/// and generates [events](Event), which return tx hash, logs, and bundled txs.
+#[derive(Debug, Clone)]
+pub struct MevShareCollector {
+    mevshare_sse_url: String,
+}
+
+impl MevShareCollector {
+    pub fn new(mevshare_sse_url: String) -> Self {
+        Self { mevshare_sse_url }
+    }
+}
+
+/// Implementation of the [Collector](Collector) trait for the
+/// [MevShareCollector](MevShareCollector).
+#[async_trait]
+impl Collector<Event> for MevShareCollector {
+    async fn get_event_stream(&self) -> Result<CollectorStream<'_, Event>> {
+        let client = EventClient::default();
+        let stream = client.events(&self.mevshare_sse_url).await.unwrap();
+        let stream = stream.filter_map(|event| match event {
+            Ok(evt) => Some(evt),
+            Err(_) => None,
+        });
+        Ok(Box::pin(stream))
+    }
+}

bot/crates/artemis-core/src/collectors/mod.rs

@@ -7,3 +7,6 @@ pub mod block_collector;
 
 /// This collector listens to a stream of new pending transactions.
 pub mod mempool_collector;
+
+/// This collector listens to a stream of mev-share events.
+pub mod mevshare_collector;

bot/crates/strategy/Cargo.toml

@@ -21,6 +21,7 @@ eth-encode-packed = "0.1.0"
 ethers = {version = "2.0.7", features = ["abigen", "ws"]}
 ethers-flashbots = {git = "https://github.com/onbjerg/ethers-flashbots"}
 foundry-evm = {git = "https://github.com/mouseless-eth/foundry.git", branch = "ethers-version-change"}
+mev-share-sse = {git = "https://github.com/mattsse/mev-share-rs"}
 
 # Logging
 colored = "2.0.0"

bot/crates/strategy/src/bot/mod.rs

@@ -123,11 +123,23 @@ impl<M: Middleware + 'static> Strategy<Event, Action> for SandwichBot<M> {
                 }
             },
             Event::NewTransaction(tx) => self.process_new_tx(tx).await,
+            Event::MevShareCollector(event) => match self.process_mev_share_event(event).await {
+                Ok(_) => None,
+                Err(e) => {
+                    panic!("strategy is out of sync {}", e);
+                }
+            },
         }
     }
 }
 
 impl<M: Middleware + 'static> SandwichBot<M> {
+    /// Process new mev-share event
+    async fn process_mev_share_event(&mut self, event: mev_share_sse::Event) -> Result<()> {
+        dbg!(event);
+        Ok(())
+    }
+
     /// Process new blocks as they come in
     async fn process_new_block(&mut self, event: NewBlock) -> Result<()> {
         log_new_block_info!(event);

bot/sandwich-bin/Cargo.toml

@@ -20,7 +20,7 @@ thiserror = "1.0.37"
 tokio = {version = "1.29.0", features = ["full"]}
 url = "2.3.1"
 
-# EVM based crates
+# EVM
 cfmms = "0.6.2"
 ethers = {version = "2.0.7", features = ["abigen", "ws"]}
 ethers-flashbots = {git = "https://github.com/onbjerg/ethers-flashbots"}
@@ -32,6 +32,9 @@ colored = "2.0.0"
 fern = {version = "0.6.2", features = ["colored"]}
 indoc = "2"
 
-# artemis related
+# artemis
 artemis-core = {path = "../crates/artemis-core"}
 strategy = {path = "../crates/strategy"}
+
+# mev-share
+mev-share-sse = {git = "https://github.com/mattsse/mev-share-rs"}

bot/sandwich-bin/src/initialization.rs

@@ -5,12 +5,7 @@ use indoc::indoc;
 use log::LevelFilter;
 
 pub fn print_banner() {
-    let banner = indoc! {
-r#"
-======================
-    SANDWICH MAKER
-======================
-"#};
+    let banner = indoc! {r#"BAKING A SANDWICH !!!"#};
 
     log::info!("{}", format!("{}", banner.green().bold()));
 }

bot/sandwich-bin/src/main.rs

@@ -2,7 +2,10 @@ use std::sync::Arc;
 
 use anyhow::Result;
 use artemis_core::{
-    collectors::{block_collector::BlockCollector, mempool_collector::MempoolCollector},
+    collectors::{
+        block_collector::BlockCollector, mempool_collector::MempoolCollector,
+        mevshare_collector::MevShareCollector,
+    },
     engine::Engine,
     executors::flashbots_executor::FlashbotsExecutor,
     types::{CollectorMap, ExecutorMap},
@@ -47,6 +50,12 @@ async fn main() -> Result<()> {
     let mempool_collector = CollectorMap::new(mempool_collector, Event::NewTransaction);
     engine.add_collector(Box::new(mempool_collector));
 
+    let mev_share_sse_url = "https://mev-share.flashbots.net/".to_string();
+    let mev_share_collector = Box::new(MevShareCollector::new(mev_share_sse_url));
+    let mev_share_collector =
+        CollectorMap::new(mev_share_collector, |e| Event::MevShareCollector(e));
+    engine.add_collector(Box::new(mev_share_collector));
+
     // Setup strategy
     let configs = StratConfig {
         sandwich_address: config.sandwich_address,