data_sources.rs

use authority_selection_inherents::AuthoritySelectionDataSource;
use pallet_sidechain_rpc::SidechainRpcDataSource;
use partner_chains_data_source_metrics::McFollowerMetrics;
use partner_chains_demo_runtime::AccountId;
use sc_service::error::Error as ServiceError;
use sidechain_mc_hash::McHashDataSource;
use sp_block_participation::inherent_data::BlockParticipationDataSource;
use sp_governed_map::GovernedMapDataSource;
use sp_partner_chains_bridge::TokenBridgeDataSource;
use std::{error::Error, sync::Arc};

pub const DATA_SOURCE_VAR: &str = "CARDANO_DATA_SOURCE";

#[derive(Clone, Debug, PartialEq)]
pub enum DataSourceType {
	DbSync,
	Mock,
	Dolos,
}

impl DataSourceType {
	pub fn from_env() -> Result<Self, Box<dyn Error + Send + Sync + 'static>> {
		let env_value =
			std::env::var(DATA_SOURCE_VAR).map_err(|_| format!("{DATA_SOURCE_VAR} is not set"))?;

		env_value.parse().map_err(|err: String| err.into())
	}
}

impl std::str::FromStr for DataSourceType {
	type Err = String;

	fn from_str(s: &str) -> Result<Self, Self::Err> {
		match s.to_lowercase().as_str() {
			"db-sync" => Ok(DataSourceType::DbSync),
			"mock" => Ok(DataSourceType::Mock),
			"dolos" => Ok(DataSourceType::Dolos),
			_ => {
				Err(format!("Invalid data source type: {}. Valid options: db-sync, mock, dolos", s))
			},
		}
	}
}

impl std::fmt::Display for DataSourceType {
	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
		match self {
			DataSourceType::DbSync => write!(f, "db-sync"),
			DataSourceType::Mock => write!(f, "mock"),
			DataSourceType::Dolos => write!(f, "dolos"),
		}
	}
}

#[derive(Clone)]
pub struct DataSources {
	pub mc_hash: Arc<dyn McHashDataSource + Send + Sync>,
	pub authority_selection: Arc<dyn AuthoritySelectionDataSource + Send + Sync>,
	pub sidechain_rpc: Arc<dyn SidechainRpcDataSource + Send + Sync>,
	pub block_participation: Arc<dyn BlockParticipationDataSource + Send + Sync>,
	pub governed_map: Arc<dyn GovernedMapDataSource + Send + Sync>,
	pub bridge: Arc<dyn TokenBridgeDataSource<AccountId> + Send + Sync>,
}

pub(crate) async fn create_cached_data_sources(
	metrics_opt: Option<McFollowerMetrics>,
) -> std::result::Result<DataSources, ServiceError> {
	let data_source_type = DataSourceType::from_env()
		.map_err(|err| ServiceError::Application(err.to_string().into()))?;

	match data_source_type {
		DataSourceType::DbSync => {
			create_cached_db_sync_data_sources(metrics_opt).await.map_err(|err| {
				ServiceError::Application(
					format!("Failed to create db-sync data sources: {err}").into(),
				)
			})
		},

		DataSourceType::Mock => create_mock_data_sources().map_err(|err| {
			ServiceError::Application(format!("Failed to create mock data sources: {err}").into())
		}),

		DataSourceType::Dolos => create_dolos_data_sources(metrics_opt).await.map_err(|err| {
			ServiceError::Application(format!("Failed to create dolos data sources: {err}").into())
		}),
	}
}

pub fn create_mock_data_sources()
-> std::result::Result<DataSources, Box<dyn Error + Send + Sync + 'static>> {
	use partner_chains_mock_data_sources::*;
	let block = Arc::new(BlockDataSourceMock::new_from_env()?);
	Ok(DataSources {
		sidechain_rpc: Arc::new(SidechainRpcDataSourceMock::new(block.clone())),
		mc_hash: Arc::new(McHashDataSourceMock::new(block)),
		authority_selection: Arc::new(AuthoritySelectionDataSourceMock::new_from_env()?),
		block_participation: Arc::new(StakeDistributionDataSourceMock::new()),
		governed_map: Arc::new(GovernedMapDataSourceMock::default()),
		bridge: Arc::new(TokenBridgeDataSourceMock::new()),
	})
}

// TODO Currently uses db-sync for unimplemented Dolos data sources
pub async fn create_dolos_data_sources(
	_metrics_opt: Option<McFollowerMetrics>,
) -> std::result::Result<DataSources, Box<dyn Error + Send + Sync + 'static>> {
	let dolos_client = partner_chains_dolos_data_sources::get_connection_from_env()?;
	let block_dolos = Arc::new(
		partner_chains_dolos_data_sources::BlockDataSourceImpl::new_from_env(dolos_client.clone())
			.await?,
	);
	Ok(DataSources {
		sidechain_rpc: Arc::new(
			partner_chains_dolos_data_sources::SidechainRpcDataSourceImpl::new(
				dolos_client.clone(),
			),
		),
		mc_hash: Arc::new(partner_chains_dolos_data_sources::McHashDataSourceImpl::new(
			block_dolos.clone(),
		)),
		authority_selection: Arc::new(
			partner_chains_dolos_data_sources::AuthoritySelectionDataSourceImpl::new(
				dolos_client.clone(),
			),
		),
		block_participation: Arc::new(
			partner_chains_dolos_data_sources::StakeDistributionDataSourceImpl::new(
				dolos_client.clone(),
			),
		),
		governed_map: Arc::new(
			partner_chains_dolos_data_sources::GovernedMapDataSourceImpl::new(
				dolos_client.clone(),
			),
		),
		bridge: Arc::new(partner_chains_dolos_data_sources::TokenBridgeDataSourceImpl::new(
			dolos_client,
		)),
	})
}

pub const CANDIDATES_FOR_EPOCH_CACHE_SIZE: usize = 64;
pub const STAKE_CACHE_SIZE: usize = 100;
pub const GOVERNED_MAP_CACHE_SIZE: u16 = 100;
pub const BRIDGE_TRANSFER_CACHE_LOOKAHEAD: u32 = 1000;

pub async fn create_cached_db_sync_data_sources(
	metrics_opt: Option<McFollowerMetrics>,
) -> Result<DataSources, Box<dyn Error + Send + Sync + 'static>> {
	use partner_chains_db_sync_data_sources::*;
	let pool = partner_chains_db_sync_data_sources::get_connection_from_env().await?;
	// block data source is reused between mc_hash and sidechain_rpc to share cache
	let block = Arc::new(BlockDataSourceImpl::new_from_env(pool.clone()).await?);
	Ok(DataSources {
		sidechain_rpc: Arc::new(SidechainRpcDataSourceImpl::new(
			block.clone(),
			metrics_opt.clone(),
		)),
		mc_hash: Arc::new(McHashDataSourceImpl::new(block.clone(), metrics_opt.clone())),
		authority_selection: Arc::new(
			CandidatesDataSourceImpl::new(pool.clone(), metrics_opt.clone())
				.await?
				.cached(CANDIDATES_FOR_EPOCH_CACHE_SIZE)?,
		),
		block_participation: Arc::new(StakeDistributionDataSourceImpl::new(
			pool.clone(),
			metrics_opt.clone(),
			STAKE_CACHE_SIZE,
		)),
		governed_map: Arc::new(
			GovernedMapDataSourceCachedImpl::new(
				pool.clone(),
				metrics_opt.clone(),
				GOVERNED_MAP_CACHE_SIZE,
				block.clone(),
			)
			.await?,
		),
		bridge: Arc::new(CachedTokenBridgeDataSourceImpl::new(
			pool,
			metrics_opt,
			block,
			BRIDGE_TRANSFER_CACHE_LOOKAHEAD,
		)),
	})
}