refactor:

- Modularize env and context initialization logic
- Include db manager creation logic into struct

.env.example

@@ -1,5 +1,7 @@
-MONGODB_URI=
-MONGODB_DB=
+CONNECTION_URI=
+DB_NAME=
 
+# Optional variables
 EXECUTION_NODE_URL=
 BEACON_NODE_RPC=
+LOGGER=

Cargo.toml

@@ -8,6 +8,7 @@ edition = "2021"
 [dependencies]
 async-trait = "0.1.66"
 dotenv = "0.15.0"
+envy = "0.4.2"
 ethers = "1.0.2"
 futures = "0.3.25"
 hex = "0.4.3"

src/context.rs

@@ -0,0 +1,56 @@
+use ethers::prelude::*;
+use serde::Deserialize;
+use std::error;
+
+use crate::{
+    beacon_chain::BeaconChainAPI,
+    db::{blob_db_manager::DBManager, mongodb::MongoDBManager},
+};
+
+fn default_execution_node_rpc() -> String {
+    "http://localhost:8545".to_string()
+}
+
+fn default_beacon_node_rpc() -> String {
+    "http://localhost:3500".to_string()
+}
+
+fn default_logger() -> String {
+    "indexer".to_string()
+}
+
+#[derive(Deserialize, Debug)]
+struct Environment {
+    connection_uri: String,
+    db_name: String,
+    #[serde(default = "default_execution_node_rpc")]
+    execution_node_rpc: String,
+    #[serde(default = "default_beacon_node_rpc")]
+    beacon_node_rpc: String,
+    #[serde(default = "default_logger")]
+    logger: String,
+}
+
+pub struct Context {
+    pub beacon_api: BeaconChainAPI,
+    pub db_manager: MongoDBManager,
+    pub provider: Provider<Http>,
+    pub logger: String,
+}
+
+pub async fn create_context() -> Result<Context, Box<dyn error::Error>> {
+    let Environment {
+        beacon_node_rpc,
+        connection_uri,
+        db_name,
+        execution_node_rpc,
+        logger,
+    } = envy::from_env::<Environment>()?;
+
+    Ok(Context {
+        beacon_api: BeaconChainAPI::new(beacon_node_rpc),
+        db_manager: MongoDBManager::new(&connection_uri, &db_name).await?,
+        provider: Provider::<Http>::try_from(execution_node_rpc)?,
+        logger,
+    })
+}

src/db/blob_db_manager.rs

@@ -9,6 +9,10 @@ use super::types::IndexerMetadata;
 pub trait DBManager {
     type Options;
 
+    async fn new(connection_uri: &String, db_name: &String) -> Result<Self, Box<dyn Error>>
+    where
+        Self: Sized;
+
     async fn commit_transaction(
         &mut self,
         options: Option<Self::Options>,

src/db/mongodb/mod.rs

@@ -31,24 +31,24 @@ pub struct MongoDBManagerOptions {
 
 const INDEXER_METADATA_ID: &str = "indexer_metadata";
 
-pub async fn connect() -> Result<MongoDBManager, Box<dyn Error>> {
-    let connection_url = std::env::var("MONGODB_URI").unwrap();
-    let database_name = std::env::var("MONGODB_DB").unwrap();
-
-    let mut client_options = ClientOptions::parse(connection_url).await?;
+#[async_trait]
+impl DBManager for MongoDBManager {
+    type Options = MongoDBManagerOptions;
 
-    client_options.app_name = Some("Blobscan".to_string());
+    async fn new(connection_uri: &String, db_name: &String) -> Result<Self, Box<dyn Error>>
+    where
+        Self: Sized,
+    {
+        let mut client_options = ClientOptions::parse(connection_uri).await?;
 
-    let client = Client::with_options(client_options)?;
-    let session = client.start_session(None).await?;
-    let db = client.database(&database_name);
+        client_options.app_name = Some("Blobscan".to_string());
 
-    Ok(MongoDBManager { session, db })
-}
+        let client = Client::with_options(client_options)?;
+        let session = client.start_session(None).await?;
+        let db = client.database(db_name);
 
-#[async_trait]
-impl DBManager for MongoDBManager {
-    type Options = MongoDBManagerOptions;
+        Ok(MongoDBManager { session, db })
+    }
 
     async fn commit_transaction(
         &mut self,
@@ -62,7 +62,6 @@ impl DBManager for MongoDBManager {
             let result = self.session.commit_transaction().await;
 
             if let Err(ref error) = result {
-                println!("Commit result: {:?}", error);
                 if error.contains_label(UNKNOWN_TRANSACTION_COMMIT_RESULT) {
                     continue;
                 }

src/main.rs

@@ -1,17 +1,16 @@
 use std::{
-    env,
     error::{self},
     thread,
     time::Duration,
 };
 
-use beacon_chain::BeaconChainAPI;
-use ethers::prelude::*;
-use slots::{process_slots, Config as SlotConfig};
+use context::create_context;
+use slots::process_slots;
 
-use crate::db::{blob_db_manager::DBManager, mongodb::connect};
+use crate::db::blob_db_manager::DBManager;
 
 mod beacon_chain;
+mod context;
 mod db;
 mod slots;
 mod utils;
@@ -20,35 +19,24 @@ type StdErr = Box<dyn error::Error>;
 
 #[tokio::main]
 async fn main() -> Result<(), StdErr> {
-    dotenv::dotenv()?;
-
-    let execution_node_rpc = env::var("EXECUTION_NODE_RPC")?;
-    let beacon_node_rpc = env::var("BEACON_NODE_RPC")?;
+    dotenv::dotenv().expect("Failed to read .env file");
 
     log4rs::init_file("log4rs.yml", Default::default()).unwrap();
 
-    let beacon_api = BeaconChainAPI::new(beacon_node_rpc);
-    let db_manager = connect().await?;
-    let provider = Provider::<Http>::try_from(execution_node_rpc)?;
-
-    let mut config = SlotConfig {
-        beacon_api,
-        db_manager,
-        provider,
-    };
+    let mut context = create_context().await?;
 
-    let mut current_slot = match config.db_manager.read_metadata(None).await? {
+    let mut current_slot = match context.db_manager.read_metadata(None).await? {
         Some(metadata) => metadata.last_slot + 1,
         None => 0,
     };
 
     loop {
-        match config.beacon_api.get_block(None).await? {
+        match context.beacon_api.get_block(None).await? {
             Some(latest_beacon_block) => {
                 let latest_slot: u32 = latest_beacon_block.slot.parse()?;
 
                 if current_slot < latest_slot {
-                    process_slots(current_slot, latest_slot, &mut config).await;
+                    process_slots(current_slot, latest_slot, &mut context).await;
 
                     current_slot = latest_slot;
                 }

src/slots/mod.rs

@@ -5,38 +5,25 @@ use futures::future::join_all;
 use log::{error, info};
 
 use crate::{
-    beacon_chain::BeaconChainAPI,
-    db::{
-        blob_db_manager::DBManager,
-        mongodb::{MongoDBManager, MongoDBManagerOptions},
-        types::Blob,
-    },
-    utils::{
-        logs::INDEXER_LOGGER,
-        web3::{calculate_versioned_hash, get_eip_4844_tx, get_tx_versioned_hashes},
-    },
+    context::Context,
+    db::{blob_db_manager::DBManager, mongodb::MongoDBManagerOptions, types::Blob},
+    utils::web3::{calculate_versioned_hash, get_eip_4844_tx, get_tx_versioned_hashes},
 };
 
 type StdErr = Box<dyn error::Error>;
 
-pub struct Config {
-    pub db_manager: MongoDBManager,
-    pub beacon_api: BeaconChainAPI,
-    pub provider: Provider<Http>,
-}
-
-pub async fn process_slots(start_slot: u32, end_slot: u32, config: &mut Config) {
+pub async fn process_slots(start_slot: u32, end_slot: u32, context: &mut Context) {
     let mut current_slot = start_slot;
 
     while current_slot < end_slot {
-        let result = process_slot(current_slot, config).await;
+        let result = process_slot(current_slot, context).await;
 
         // TODO: implement exponential backoff for proper error handling. If X intents have been made, then notify and stop process
         if let Err(e) = result {
-            save_slot(current_slot, config).await;
+            save_slot(current_slot - 1, context).await;
 
             error!(
-                target: INDEXER_LOGGER,
+                target: context.logger.as_str(),
                 "[Slot {}] Couldn't process slot: {}", current_slot, e
             );
 
@@ -46,20 +33,23 @@ pub async fn process_slots(start_slot: u32, end_slot: u32, config: &mut Config)
         current_slot = current_slot + 1;
     }
 
-    save_slot(current_slot, config).await
+    save_slot(current_slot, context).await
 }
 
-async fn process_slot(slot: u32, config: &mut Config) -> Result<(), StdErr> {
-    let provider = &config.provider;
-    let db_manager = &mut config.db_manager;
-    let beacon_api = &config.beacon_api;
+async fn process_slot(slot: u32, context: &mut Context) -> Result<(), StdErr> {
+    let Context {
+        beacon_api,
+        db_manager,
+        provider,
+        logger,
+    } = context;
 
     let start = Instant::now();
     let beacon_block = match beacon_api.get_block(Some(slot)).await? {
         Some(block) => block,
         None => {
             info!(
-                target: INDEXER_LOGGER,
+                target: logger,
                 "[Slot {}] Skipping as there is no beacon block", slot
             );
 
@@ -71,7 +61,7 @@ async fn process_slot(slot: u32, config: &mut Config) -> Result<(), StdErr> {
         Some(payload) => payload,
         None => {
             info!(
-                target: INDEXER_LOGGER,
+                target: logger,
                 "[Slot {}] Skipping as beacon block doesn't contain execution payload", slot
             );
 
@@ -83,7 +73,7 @@ async fn process_slot(slot: u32, config: &mut Config) -> Result<(), StdErr> {
         Some(commitments) => commitments,
         None => {
             info!(
-                target: INDEXER_LOGGER,
+                target: logger,
                 "[Slot {}] Skipping as beacon block doesn't contain blob kzg commitments", slot
             );
 
@@ -93,7 +83,7 @@ async fn process_slot(slot: u32, config: &mut Config) -> Result<(), StdErr> {
     let execution_block_hash = execution_payload.block_hash;
     let execution_block_hash = H256::from_str(execution_block_hash.as_str())?;
 
-    let execution_block = match config.provider.get_block(execution_block_hash).await? {
+    let execution_block = match provider.get_block(execution_block_hash).await? {
         Some(block) => block,
         None => {
             let error_msg = format!("Execution block {} not found", execution_block_hash);
@@ -120,7 +110,7 @@ async fn process_slot(slot: u32, config: &mut Config) -> Result<(), StdErr> {
 
     if blob_txs.len() == 0 {
         info!(
-            target: INDEXER_LOGGER,
+            target: logger,
             "[Slot {}] Skipping as execution block doesn't contain blob txs", slot
         );
 
@@ -131,7 +121,7 @@ async fn process_slot(slot: u32, config: &mut Config) -> Result<(), StdErr> {
         Some(blobs_sidecar) => {
             if blobs_sidecar.blobs.len() == 0 {
                 info!(
-                    target: INDEXER_LOGGER,
+                    target: logger,
                     "[Slot {}] Skipping as blobs sidecar is empty", slot
                 );
 
@@ -142,7 +132,7 @@ async fn process_slot(slot: u32, config: &mut Config) -> Result<(), StdErr> {
         }
         None => {
             info!(
-                target: INDEXER_LOGGER,
+                target: logger,
                 "[Slot {}] Skipping as there is no blobs sidecar", slot
             );
 
@@ -185,12 +175,12 @@ async fn process_slot(slot: u32, config: &mut Config) -> Result<(), StdErr> {
         db_manager.insert_blob(blob, blob_tx.hash, None).await?;
     }
 
-    config.db_manager.commit_transaction(None).await?;
+    db_manager.commit_transaction(None).await?;
 
     let duration = start.elapsed();
 
     info!(
-        target: INDEXER_LOGGER,
+        target: logger,
         "[Slot {}] Blobs indexed correctly (elapsed time: {:?}s)",
         slot,
         duration.as_secs()
@@ -199,14 +189,17 @@ async fn process_slot(slot: u32, config: &mut Config) -> Result<(), StdErr> {
     Ok(())
 }
 
-async fn save_slot(slot: u32, config: &mut Config) {
-    let result = config
-        .db_manager
+async fn save_slot(slot: u32, context: &mut Context) {
+    let Context {
+        db_manager, logger, ..
+    } = context;
+
+    let result = db_manager
         .update_last_slot(slot, Some(MongoDBManagerOptions { use_session: false }))
         .await;
 
     if let Err(e) = result {
-        error!(target: INDEXER_LOGGER, "Couldn't update last slot: {}", e);
+        error!(target: logger, "Couldn't update last slot: {}", e);
         panic!();
     }
 }

src/utils/mod.rs

@@ -1,2 +1 @@
-pub mod logs;
 pub mod web3;