add: ETCM-12255 bridge data source (#983)

Author: AmbientTea

Cargo.lock

@@ -101,6 +101,7 @@ partner-chains-db-sync-data-sources = { workspace = true, features = [
 	"mc-hash",
 	"sidechain-rpc",
 	"block-participation",
+	"bridge",
 ] }
 partner-chains-mock-data-sources = { workspace = true, features = [
 	"block-source",

demo/node/Cargo.toml

@@ -3,7 +3,7 @@ use pallet_sidechain_rpc::SidechainRpcDataSource;
 use partner_chains_db_sync_data_sources::{
 	BlockDataSourceImpl, CandidatesDataSourceImpl, GovernedMapDataSourceCachedImpl,
 	McFollowerMetrics, McHashDataSourceImpl, NativeTokenManagementDataSourceImpl,
-	SidechainRpcDataSourceImpl, StakeDistributionDataSourceImpl,
+	SidechainRpcDataSourceImpl, StakeDistributionDataSourceImpl, TokenBridgeDataSourceImpl,
 };
 use partner_chains_demo_runtime::AccountId;
 use partner_chains_mock_data_sources::{
@@ -151,6 +151,6 @@ pub async fn create_cached_db_sync_data_sources(
 			)
 			.await?,
 		),
-		bridge: Arc::new(TokenBridgeDataSourceMock::new()),
+		bridge: Arc::new(TokenBridgeDataSourceImpl::new(pool, metrics_opt)),
 	})
 }

demo/node/src/data_sources.rs

@@ -1,9 +1,8 @@
 use super::*;
 use frame_benchmarking::v2::*;
-use frame_support::BoundedVec;
-use frame_support::assert_ok;
-use frame_support::traits::Get;
+use frame_support::{BoundedVec, assert_ok, traits::Get};
 use frame_system::RawOrigin;
+use sidechain_domain::McBlockNumber;
 use sp_partner_chains_bridge::*;
 
 /// Helper trait for injecting mock values for use in benchmarks
@@ -26,7 +25,7 @@ impl<T: crate::Config> BenchmarkHelper<T> for () {
 	}
 
 	fn data_checkpoint() -> BridgeDataCheckpoint {
-		Default::default()
+		BridgeDataCheckpoint::Block(McBlockNumber(0))
 	}
 }
 

toolkit/bridge/pallet/src/benchmarking.rs

@@ -24,12 +24,15 @@ fn main_chain_scripts() -> MainChainScripts {
 	MainChainScripts {
 		token_policy_id: PolicyId([1; 28]),
 		token_asset_name: AssetName(bounded_vec![2;8]),
-		illiquid_supply_validator_address: MainchainAddress::from_str("validator address").unwrap(),
+		illiquid_circulation_supply_validator_address: MainchainAddress::from_str(
+			"validator address",
+		)
+		.unwrap(),
 	}
 }
 
 fn data_checkpoint() -> BridgeDataCheckpoint {
-	BridgeDataCheckpoint(UtxoId::new([1; 32], 3))
+	BridgeDataCheckpoint::Utxo(UtxoId::new([1; 32], 3))
 }
 
 mod set_main_chain_scripts {

toolkit/bridge/pallet/src/tests.rs

@@ -8,7 +8,7 @@ use alloc::vec::*;
 use parity_scale_codec::{Decode, DecodeWithMemTracking, Encode, MaxEncodedLen};
 use scale_info::TypeInfo;
 use serde::{Deserialize, Serialize};
-use sidechain_domain::{AssetName, MainchainAddress, McBlockHash, PolicyId, UtxoId};
+use sidechain_domain::{AssetName, MainchainAddress, McBlockHash, McBlockNumber, PolicyId, UtxoId};
 use sp_inherents::*;
 
 #[cfg(feature = "std")]
@@ -37,7 +37,7 @@ pub struct MainChainScripts {
 	/// Address of the illiquid supply validator.
 	///
 	/// All tokens sent to that address are effectively locked and considered "sent" to the Partner Chain.
-	pub illiquid_supply_validator_address: MainchainAddress,
+	pub illiquid_circulation_supply_validator_address: MainchainAddress,
 }
 
 /// Type containing all information needed to process a single transfer incoming from
@@ -115,20 +115,21 @@ pub enum TokenBridgeInherentDataProvider<RecipientAddress> {
 	},
 }
 
-/// Pointer to last data processed
+/// Value specifying the point in time up to which bridge transfers have been processed
+///
+/// This type is an enum wrapping either a block number or a utxo to handle both a case
+/// where all transfers up to a block have been handled and a case where there were more
+/// transfers than the limit allows and observability needs to pick up after the last
+/// utxo that could be observed
 #[derive(
-	Default,
-	Clone,
-	Debug,
-	Encode,
-	Decode,
-	DecodeWithMemTracking,
-	TypeInfo,
-	PartialEq,
-	Eq,
-	MaxEncodedLen,
+	Clone, Debug, Encode, Decode, DecodeWithMemTracking, TypeInfo, PartialEq, Eq, MaxEncodedLen,
 )]
-pub struct BridgeDataCheckpoint(pub UtxoId);
+pub enum BridgeDataCheckpoint {
+	/// Last transfer utxo that has been processed
+	Utxo(UtxoId),
+	/// Cardano block up to which data has been processed
+	Block(McBlockNumber),
+}
 
 /// Interface for data sources that can be used by [TokenBridgeInherentDataProvider]
 #[cfg(feature = "std")]

toolkit/bridge/primitives/src/lib.rs

@@ -50,6 +50,7 @@ partner-chains-plutus-data = { workspace = true }
 sp-block-participation = { workspace = true, features = [
     "std",
 ], optional = true }
+sp-partner-chains-bridge = { workspace = true, optional = true, features = ["std"] }
 
 [dev-dependencies]
 tokio-test = "0.4.3"
@@ -67,3 +68,4 @@ native-token = ["sp-native-token-management"]
 mc-hash = ["sidechain-mc-hash", "block-source"]
 sidechain-rpc = ["pallet-sidechain-rpc", "block-source"]
 block-participation = ["sp-block-participation"]
+bridge = ["sp-partner-chains-bridge"]

toolkit/data-sources/db-sync/Cargo.toml

@@ -0,0 +1,144 @@
+//! Db-Sync data source used by the Partner Chain token bridge observability
+//!
+//! # Assumptions
+//!
+//! The data source implementation assumes that the utxos found at the illiquid circulating
+//! supply address conform to rules that are enforced by the Partner Chains smart contracts.
+//!
+//! Most importantly, transactions that spend any UTXOs from the ICS can only create at most
+//! one new UTXO at the ICS address. Conversely, transactions that create more than one UTXO
+//! at the illiquid supply address can only spend UTXOs from outside of it. This guarantees
+//! that the observability layer can always correctly identify the number of tokens transfered
+//! by calculating the delta of `tokens in the new UTXO` - `tokens in the old ICS UTXOs`.
+
+use crate::McFollowerMetrics;
+use crate::db_model::*;
+use crate::observed_async_trait;
+use partner_chains_plutus_data::bridge::TokenTransferDatum;
+use partner_chains_plutus_data::bridge::TokenTransferDatumV1;
+use sidechain_domain::McBlockHash;
+use sp_partner_chains_bridge::*;
+use sqlx::PgPool;
+use std::fmt::Debug;
+
+#[cfg(test)]
+mod tests;
+
+/// Db-Sync data source serving data for Partner Chains token bridge
+pub struct TokenBridgeDataSourceImpl {
+	/// Postgres connection pool
+	pool: PgPool,
+	/// Prometheus metrics client
+	metrics_opt: Option<McFollowerMetrics>,
+	/// Configuration used by Db-Sync
+	db_sync_config: DbSyncConfigurationProvider,
+}
+
+impl TokenBridgeDataSourceImpl {
+	/// Crates a new token bridge data source
+	pub fn new(pool: PgPool, metrics_opt: Option<McFollowerMetrics>) -> Self {
+		Self { db_sync_config: DbSyncConfigurationProvider::new(pool.clone()), pool, metrics_opt }
+	}
+}
+
+observed_async_trait!(
+	impl<RecipientAddress> TokenBridgeDataSource<RecipientAddress> for TokenBridgeDataSourceImpl
+	where
+		RecipientAddress: Debug,
+		RecipientAddress: (for<'a> TryFrom<&'a [u8]>),
+	{
+		async fn get_transfers(
+			&self,
+			main_chain_scripts: MainChainScripts,
+			data_checkpoint: BridgeDataCheckpoint,
+			max_transfers: u32,
+			current_mc_block_hash: McBlockHash,
+		) -> Result<
+			(Vec<BridgeTransferV1<RecipientAddress>>, BridgeDataCheckpoint),
+			Box<dyn std::error::Error + Send + Sync>,
+		> {
+			let asset = Asset {
+				policy_id: main_chain_scripts.token_policy_id.into(),
+				asset_name: main_chain_scripts.token_asset_name.into(),
+			};
+
+			let current_mc_block = get_block_by_hash(&self.pool, current_mc_block_hash.clone())
+				.await?
+				.ok_or(format!("Could not find block for hash {current_mc_block_hash:?}"))?;
+
+			let data_checkpoint = match data_checkpoint {
+				BridgeDataCheckpoint::Utxo(utxo) => {
+					let TxBlockInfo { block_number, tx_ix } =
+						get_block_info_for_utxo(&self.pool, utxo.tx_hash.into()).await?.ok_or(
+							format!(
+								"Could not find block info for data checkpoint: {data_checkpoint:?}"
+							),
+						)?;
+					BridgeCheckpoint::Utxo {
+						block_number: block_number.0,
+						tx_ix: tx_ix.0,
+						tx_out_ix: utxo.index.0,
+					}
+				},
+				BridgeDataCheckpoint::Block(checkpoint_block_number) => {
+					BridgeCheckpoint::Block { number: checkpoint_block_number.0 }
+				},
+			};
+
+			let utxos = get_bridge_utxos_tx(
+				self.db_sync_config.get_tx_in_config().await?,
+				&self.pool,
+				&main_chain_scripts.illiquid_circulation_supply_validator_address.into(),
+				asset,
+				data_checkpoint,
+				current_mc_block.block_no,
+				max_transfers,
+			)
+			.await?;
+
+			let new_checkpoint = match utxos.last() {
+				None => BridgeDataCheckpoint::Block(current_mc_block.block_no.into()),
+				Some(_) if (utxos.len() as u32) < max_transfers => {
+					BridgeDataCheckpoint::Block(current_mc_block.block_no.into())
+				},
+				Some(utxo) => BridgeDataCheckpoint::Utxo(utxo.utxo_id()),
+			};
+
+			let transfers = utxos.into_iter().flat_map(utxo_to_transfer).collect();
+
+			Ok((transfers, new_checkpoint))
+		}
+	}
+);
+
+fn utxo_to_transfer<RecipientAddress>(
+	utxo: BridgeUtxo,
+) -> Option<BridgeTransferV1<RecipientAddress>>
+where
+	RecipientAddress: for<'a> TryFrom<&'a [u8]>,
+{
+	let token_delta = (utxo.tokens_out.0 as i128) - (utxo.tokens_in.0 as i128);
+
+	if token_delta <= 0 {
+		return None;
+	}
+
+	let token_amount = token_delta as u64;
+
+	let transfer = match TokenTransferDatum::try_from(utxo.datum.0.clone()) {
+		Ok(TokenTransferDatum::V1(TokenTransferDatumV1::UserTransfer { receiver })) => {
+			match RecipientAddress::try_from(receiver.0.as_ref()) {
+				Ok(recipient) => BridgeTransferV1::UserTransfer { token_amount, recipient },
+				Err(_) => {
+					BridgeTransferV1::InvalidTransfer { token_amount, utxo_id: utxo.utxo_id() }
+				},
+			}
+		},
+		Ok(TokenTransferDatum::V1(TokenTransferDatumV1::ReserveTransfer)) => {
+			BridgeTransferV1::ReserveTransfer { token_amount }
+		},
+		Err(_) => BridgeTransferV1::InvalidTransfer { token_amount, utxo_id: utxo.utxo_id() },
+	};
+
+	Some(transfer)
+}