change: Reduce data source boilerplate in node (#124)

mainchain-follower/db-sync-follower/src/block/mod.rs

@@ -82,7 +82,7 @@ impl DbSyncBlockDataSourceConfig {
 }
 
 impl BlockDataSourceImpl {
-	pub async fn from_env(
+	pub async fn new_from_env(
 		pool: PgPool,
 		metrics_opt: Option<McFollowerMetrics>,
 	) -> std::result::Result<Self, Box<dyn Error + Send + Sync + 'static>> {

mainchain-follower/db-sync-follower/src/candidates/cached.rs

@@ -107,6 +107,7 @@ impl CandidateDataSourceCached {
 			highest_seen_stable_epoch: Arc::new(Mutex::new(None)),
 		}
 	}
+
 	pub fn new_from_env(
 		inner: CandidatesDataSourceImpl,
 		candidates_for_epoch_cache_size: usize,

mainchain-follower/db-sync-follower/src/candidates/mod.rs

@@ -14,6 +14,7 @@ use plutus::Datum::*;
 use sidechain_domain::*;
 use sqlx::PgPool;
 use std::collections::HashMap;
+use std::error::Error;
 
 pub mod cached;
 
@@ -162,16 +163,19 @@ impl CandidateDataSource for CandidatesDataSourceImpl {
 });
 
 impl CandidatesDataSourceImpl {
-	pub(crate) fn new(pool: PgPool, metrics_opt: Option<McFollowerMetrics>) -> Self {
-		Self { pool, metrics_opt }
-	}
-
-	pub async fn from_config(
+	pub async fn new(
 		pool: PgPool,
 		metrics_opt: Option<McFollowerMetrics>,
 	) -> Result<CandidatesDataSourceImpl> {
 		db_model::create_idx_ma_tx_out_ident(&pool).await?;
-		Ok(CandidatesDataSourceImpl::new(pool, metrics_opt))
+		Ok(Self { pool, metrics_opt })
+	}
+
+	pub fn cached(
+		self,
+		candidates_for_epoch_cache_size: usize,
+	) -> std::result::Result<cached::CandidateDataSourceCached, Box<dyn Error + Send + Sync>> {
+		cached::CandidateDataSourceCached::new_from_env(self, candidates_for_epoch_cache_size)
 	}
 
 	/// Registrations state up to this block are considered as "active", after it - as "pending".

mainchain-follower/db-sync-follower/src/candidates/tests.rs

@@ -112,7 +112,7 @@ async fn test_get_ariadne_parameters_returns_the_latest_params_for_the_future_ep
 #[sqlx::test(migrations = "./testdata/migrations")]
 async fn test_make_source_creates_index(pool: PgPool) {
 	assert!(!index_exists(&pool, "idx_ma_tx_out_ident").await);
-	CandidatesDataSourceImpl::from_config(pool.clone(), None).await.unwrap();
+	CandidatesDataSourceImpl::new(pool.clone(), None).await.unwrap();
 	assert!(index_exists(&pool, "idx_ma_tx_out_ident").await);
 }
 

mainchain-follower/db-sync-follower/src/native_token/mod.rs

@@ -2,13 +2,15 @@ use crate::db_model::{Address, NativeTokenAmount, SlotNumber};
 use crate::metrics::McFollowerMetrics;
 use crate::observed_async_trait;
 use async_trait::async_trait;
+use derive_new::new;
 use main_chain_follower_api::{DataSourceError, NativeTokenManagementDataSource, Result};
 use sidechain_domain::*;
 use sqlx::PgPool;
 
 #[cfg(test)]
 mod tests;
 
+#[derive(new)]
 pub struct NativeTokenManagementDataSourceImpl {
 	pub pool: PgPool,
 	pub metrics_opt: Option<McFollowerMetrics>,

mainchain-follower/main-chain-follower-cli/src/main.rs

@@ -99,6 +99,6 @@ mod data_source {
 	}
 
 	pub async fn candidate() -> Result<CandidatesDataSourceImpl> {
-		Ok(CandidatesDataSourceImpl::from_config(pool().await?, None).await?)
+		Ok(CandidatesDataSourceImpl::new(pool().await?, None).await?)
 	}
 }

mainchain-follower/mock/src/block.rs

@@ -3,6 +3,7 @@ pub use main_chain_follower_api::block::*;
 use main_chain_follower_api::common::*;
 use main_chain_follower_api::*;
 use sidechain_domain::*;
+use std::error::Error;
 
 pub struct BlockDataSourceMock {
 	/// Duration of a mainchain epoch in milliseconds
@@ -50,6 +51,12 @@ impl BlockDataSourceMock {
 		Self { mc_epoch_duration_millis }
 	}
 
+	pub fn new_from_env() -> std::result::Result<Self, Box<dyn Error + Send + Sync + 'static>> {
+		let mc_epoch_duration_millis: u32 =
+			std::env::var("MC__EPOCH_DURATION_MILLIS")?.parse::<u32>()?;
+		Ok(Self::new(mc_epoch_duration_millis))
+	}
+
 	fn block_per_epoch(&self) -> u32 {
 		self.mc_epoch_duration_millis / 20000
 	}

mainchain-follower/mock/src/candidate.rs

@@ -177,7 +177,7 @@ impl MockCandidateDataSource {
 		self.registrations_data.epoch_rotation[rotation_no].clone()
 	}
 
-	pub fn from_env() -> std::result::Result<Self, Box<dyn Error + Send + Sync + 'static>> {
+	pub fn new_from_env() -> std::result::Result<Self, Box<dyn Error + Send + Sync + 'static>> {
 		let registrations_data = MockRegistrationsConfig::read()?;
 		let mc_epoch_config = MainchainEpochConfig::read_from_env()?;
 		Ok(MockCandidateDataSource { registrations_data, mc_epoch_config })

node/src/main_chain_follower.rs

@@ -1,9 +1,6 @@
+use db_sync_follower::candidates::CandidatesDataSourceImpl;
 use db_sync_follower::native_token::NativeTokenManagementDataSourceImpl;
-use db_sync_follower::{
-	block::{BlockDataSourceImpl, DbSyncBlockDataSourceConfig},
-	candidates::{cached::CandidateDataSourceCached, CandidatesDataSourceImpl},
-	metrics::McFollowerMetrics,
-};
+use db_sync_follower::{block::BlockDataSourceImpl, metrics::McFollowerMetrics};
 use main_chain_follower_api::{
 	BlockDataSource, CandidateDataSource, NativeTokenManagementDataSource,
 };
@@ -50,14 +47,9 @@ fn use_mock_follower() -> bool {
 
 pub fn create_mock_data_sources(
 ) -> std::result::Result<DataSources, Box<dyn Error + Send + Sync + 'static>> {
-	let mc_epoch_duration_millis: u32 = std::env::var("MC__EPOCH_DURATION_MILLIS")
-		.ok()
-		.and_then(|v| v.parse::<u32>().ok())
-		.unwrap();
-	let block_data_source_mock = BlockDataSourceMock::new(mc_epoch_duration_millis);
 	Ok(DataSources {
-		block: Arc::new(block_data_source_mock),
-		candidate: Arc::new(MockCandidateDataSource::from_env()?),
+		block: Arc::new(BlockDataSourceMock::new_from_env()?),
+		candidate: Arc::new(MockCandidateDataSource::new_from_env()?),
 		native_token: Arc::new(NativeTokenDataSourceMock::new()),
 	})
 }
@@ -68,18 +60,15 @@ pub async fn create_cached_data_sources(
 	metrics_opt: Option<McFollowerMetrics>,
 ) -> Result<DataSources, Box<dyn Error + Send + Sync + 'static>> {
 	let pool = db_sync_follower::data_sources::get_connection_from_env().await?;
-	let mc_epoch_config = &db_sync_follower::data_sources::read_mc_epoch_config()?;
 	Ok(DataSources {
-		block: Arc::new(BlockDataSourceImpl::from_config(
-			pool.clone(),
-			DbSyncBlockDataSourceConfig::from_env()?,
-			mc_epoch_config,
-			metrics_opt.clone(),
-		)),
-		candidate: Arc::new(CandidateDataSourceCached::new_from_env(
-			CandidatesDataSourceImpl::from_config(pool.clone(), metrics_opt.clone()).await?,
-			CANDIDATES_FOR_EPOCH_CACHE_SIZE,
-		)?),
-		native_token: Arc::new(NativeTokenManagementDataSourceImpl { pool, metrics_opt }),
+		block: Arc::new(
+			BlockDataSourceImpl::new_from_env(pool.clone(), metrics_opt.clone()).await?,
+		),
+		candidate: Arc::new(
+			CandidatesDataSourceImpl::new(pool.clone(), metrics_opt.clone())
+				.await?
+				.cached(CANDIDATES_FOR_EPOCH_CACHE_SIZE)?,
+		),
+		native_token: Arc::new(NativeTokenManagementDataSourceImpl::new(pool, metrics_opt)),
 	})
 }