Add inbound queue (#4)

* modify inbound queue to support custom message processing

* passively test tuple trait implementation

* rename XCM to Xcm everywhere

* add comment explaining MessageProcessor

bridges/snowbridge/pallets/inbound-queue/src/lib.rs

@@ -23,8 +23,6 @@
 //!   parachain.
 #![cfg_attr(not(feature = "std"), no_std)]
 
-mod envelope;
-
 #[cfg(feature = "runtime-benchmarks")]
 mod benchmarking;
 
@@ -35,9 +33,9 @@ mod mock;
 
 #[cfg(test)]
 mod test;
+pub mod xcm_message_processor;
 
-use codec::{Decode, DecodeAll, Encode};
-use envelope::Envelope;
+use codec::{Decode, Encode};
 use frame_support::{
 	traits::{
 		fungible::{Inspect, Mutate},
@@ -48,7 +46,7 @@ use frame_support::{
 };
 use frame_system::ensure_signed;
 use scale_info::TypeInfo;
-use sp_core::H160;
+use sp_core::{H160, H256};
 use sp_runtime::traits::Zero;
 use sp_std::vec;
 use xcm::prelude::{
@@ -62,7 +60,7 @@ use snowbridge_core::{
 	StaticLookup,
 };
 use snowbridge_router_primitives::inbound::{
-	ConvertMessage, ConvertMessageError, VersionedMessage,
+	envelope::Envelope, ConvertMessage, ConvertMessageError, MessageProcessor, VersionedXcmMessage,
 };
 use sp_runtime::{traits::Saturating, SaturatedConversion, TokenError};
 
@@ -84,7 +82,6 @@ pub mod pallet {
 
 	use frame_support::pallet_prelude::*;
 	use frame_system::pallet_prelude::*;
-	use sp_core::H256;
 
 	#[pallet::pallet]
 	pub struct Pallet<T>(_);
@@ -139,6 +136,9 @@ pub mod pallet {
 
 		/// To withdraw and deposit an asset.
 		type AssetTransactor: TransactAsset;
+
+		/// Process the message that was submitted
+		type MessageProcessor: MessageProcessor;
 	}
 
 	#[pallet::hooks]
@@ -205,10 +205,12 @@ pub mod pallet {
 				XcmpSendError::NotApplicable => Error::<T>::Send(SendError::NotApplicable),
 				XcmpSendError::Unroutable => Error::<T>::Send(SendError::NotRoutable),
 				XcmpSendError::Transport(_) => Error::<T>::Send(SendError::Transport),
-				XcmpSendError::DestinationUnsupported =>
-					Error::<T>::Send(SendError::DestinationUnsupported),
-				XcmpSendError::ExceedsMaxMessageSize =>
-					Error::<T>::Send(SendError::ExceedsMaxMessageSize),
+				XcmpSendError::DestinationUnsupported => {
+					Error::<T>::Send(SendError::DestinationUnsupported)
+				},
+				XcmpSendError::ExceedsMaxMessageSize => {
+					Error::<T>::Send(SendError::ExceedsMaxMessageSize)
+				},
 				XcmpSendError::MissingArgument => Error::<T>::Send(SendError::MissingArgument),
 				XcmpSendError::Fees => Error::<T>::Send(SendError::Fees),
 			}
@@ -251,7 +253,7 @@ pub mod pallet {
 			// Verify message nonce
 			<Nonce<T>>::try_mutate(envelope.channel_id, |nonce| -> DispatchResult {
 				if *nonce == u64::MAX {
-					return Err(Error::<T>::MaxNonceReached.into())
+					return Err(Error::<T>::MaxNonceReached.into());
 				}
 				if envelope.nonce != nonce.saturating_add(1) {
 					Err(Error::<T>::InvalidNonce.into())
@@ -275,34 +277,7 @@ pub mod pallet {
 				T::Token::transfer(&sovereign_account, &who, amount, Preservation::Preserve)?;
 			}
 
-			// Decode payload into `VersionedMessage`
-			let message = VersionedMessage::decode_all(&mut envelope.payload.as_ref())
-				.map_err(|_| Error::<T>::InvalidPayload)?;
-
-			// Decode message into XCM
-			let (xcm, fee) = Self::do_convert(envelope.message_id, message.clone())?;
-
-			log::info!(
-				target: LOG_TARGET,
-				"💫 xcm decoded as {:?} with fee {:?}",
-				xcm,
-				fee
-			);
-
-			// Burning fees for teleport
-			Self::burn_fees(channel.para_id, fee)?;
-
-			// Attempt to send XCM to a dest parachain
-			let message_id = Self::send_xcm(xcm, channel.para_id)?;
-
-			Self::deposit_event(Event::MessageReceived {
-				channel_id: envelope.channel_id,
-				nonce: envelope.nonce,
-				message_id,
-				fee_burned: fee,
-			});
-
-			Ok(())
+			T::MessageProcessor::process_message(channel, envelope)
 		}
 
 		/// Halt or resume all pallet operations. May only be called by root.
@@ -322,7 +297,7 @@ pub mod pallet {
 	impl<T: Config> Pallet<T> {
 		pub fn do_convert(
 			message_id: H256,
-			message: VersionedMessage,
+			message: VersionedXcmMessage,
 		) -> Result<(Xcm<()>, BalanceOf<T>), Error<T>> {
 			let (xcm, fee) = T::MessageConverter::convert(message_id, message)
 				.map_err(|e| Error::<T>::ConvertMessage(e))?;

bridges/snowbridge/pallets/inbound-queue/src/mock.rs

@@ -16,12 +16,13 @@ use snowbridge_router_primitives::inbound::MessageToXcm;
 use sp_core::{H160, H256};
 use sp_runtime::{
 	traits::{IdentifyAccount, IdentityLookup, MaybeEquivalence, Verify},
-	BuildStorage, FixedU128, MultiSignature,
+	BuildStorage, FixedU128, MultiSignature, DispatchError,
 };
 use sp_std::{convert::From, default::Default};
 use xcm::{latest::SendXcm, prelude::*};
 use xcm_executor::AssetsInHolding;
 
+use crate::xcm_message_processor::XcmMessageProcessor;
 use crate::{self as inbound_queue};
 
 type Block = frame_system::mocking::MockBlock<Test>;
@@ -167,10 +168,10 @@ impl StaticLookup for MockChannelLookup {
 	type Target = Channel;
 
 	fn lookup(channel_id: Self::Source) -> Option<Self::Target> {
-		if channel_id !=
-			hex!("c173fac324158e77fb5840738a1a541f633cbec8884c6a601c567d2b376a0539").into()
+		if channel_id
+			!= hex!("c173fac324158e77fb5840738a1a541f633cbec8884c6a601c567d2b376a0539").into()
 		{
-			return None
+			return None;
 		}
 		Some(Channel { agent_id: H256::zero(), para_id: ASSET_HUB_PARAID.into() })
 	}
@@ -218,6 +219,30 @@ impl MaybeEquivalence<TokenId, Location> for MockTokenIdConvert {
 	}
 }
 
+pub struct DummyPrefix;
+
+impl MessageProcessor for DummyPrefix {
+	fn can_process_message(_channel: &Channel, _envelope: &Envelope) -> bool {
+		false
+	}
+
+	fn process_message(_channel: Channel, _envelope: Envelope) -> Result<(), DispatchError> {
+		panic!("DummyPrefix::process_message shouldn't be called");
+	}
+}
+
+pub struct DummySuffix;
+
+impl MessageProcessor for DummySuffix {
+	fn can_process_message(_channel: &Channel, _envelope: &Envelope) -> bool {
+		true
+	}
+
+	fn process_message(_channel: Channel, _envelope: Envelope) -> Result<(), DispatchError> {
+		panic!("DummySuffix::process_message shouldn't be called");
+	}
+}
+
 impl inbound_queue::Config for Test {
 	type RuntimeEvent = RuntimeEvent;
 	type Verifier = MockVerifier;
@@ -243,6 +268,7 @@ impl inbound_queue::Config for Test {
 	type LengthToFee = IdentityFee<u128>;
 	type MaxMessageSize = ConstU32<1024>;
 	type AssetTransactor = SuccessfulTransactor;
+	type MessageProcessor = (DummyPrefix, XcmMessageProcessor<Test>, DummySuffix); // We are passively testing if implementation of MessageProcessor trait works correctly for tuple
 }
 
 pub fn last_events(n: usize) -> Vec<RuntimeEvent> {

bridges/snowbridge/pallets/inbound-queue/src/xcm_message_processor.rs

@@ -0,0 +1,48 @@
+use crate::{Error, Event, LOG_TARGET};
+use codec::DecodeAll;
+use core::marker::PhantomData;
+use snowbridge_core::Channel;
+use snowbridge_router_primitives::inbound::envelope::Envelope;
+use snowbridge_router_primitives::inbound::{MessageProcessor, VersionedXcmMessage};
+use sp_runtime::DispatchError;
+
+pub struct XcmMessageProcessor<T>(PhantomData<T>);
+
+impl<T> MessageProcessor for XcmMessageProcessor<T>
+where
+	T: crate::Config,
+{
+	fn can_process_message(_channel: &Channel, envelope: &Envelope) -> bool {
+		VersionedXcmMessage::decode_all(&mut envelope.payload.as_ref()).is_ok()
+	}
+
+	fn process_message(channel: Channel, envelope: Envelope) -> Result<(), DispatchError> {
+		// Decode message into XCM
+		let (xcm, fee) = match VersionedXcmMessage::decode_all(&mut envelope.payload.as_ref()) {
+			Ok(message) => crate::Pallet::<T>::do_convert(envelope.message_id, message)?,
+			Err(_) => return Err(Error::<T>::InvalidPayload.into()),
+		};
+
+		log::info!(
+			target: LOG_TARGET,
+			"💫 xcm decoded as {:?} with fee {:?}",
+			xcm,
+			fee
+		);
+
+		// Burning fees for teleport
+		crate::Pallet::<T>::burn_fees(channel.para_id, fee)?;
+
+		// Attempt to send XCM to a dest parachain
+		let message_id = crate::Pallet::<T>::send_xcm(xcm, channel.para_id)?;
+
+		crate::Pallet::<T>::deposit_event(Event::MessageReceived {
+			channel_id: envelope.channel_id,
+			nonce: envelope.nonce,
+			message_id,
+			fee_burned: fee,
+		});
+
+		Ok(())
+	}
+}

bridges/snowbridge/primitives/ethereum/Cargo.toml

@@ -18,14 +18,14 @@ codec = { features = ["derive"], workspace = true }
 scale-info = { features = ["derive"], workspace = true }
 ethbloom = { workspace = true }
 ethereum-types = { features = ["codec", "rlp", "serialize"], workspace = true }
-hex-literal = { workspace = true }
+hex-literal = { workspace = true, default-features = false }
 parity-bytes = { workspace = true }
 rlp = { workspace = true }
 sp-io.workspace = true
 sp-std.workspace = true
 sp-runtime.workspace = true
 
-ethabi = { workspace = true }
+ethabi = { workspace = true, default-features = false }
 
 [dev-dependencies]
 wasm-bindgen-test = { workspace = true }

bridges/snowbridge/primitives/router/Cargo.toml

@@ -12,7 +12,10 @@ categories = ["cryptography::cryptocurrencies"]
 workspace = true
 
 [dependencies]
+alloy-primitives = { features = ["rlp"], workspace = true }
+alloy-sol-types = { workspace = true }
 codec = { workspace = true }
+impl-trait-for-tuples = { workspace = true, default-features = false }
 scale-info = { features = ["derive"], workspace = true }
 log = { workspace = true }
 frame-support.workspace = true

bridges/snowbridge/primitives/router/src/inbound/envelope.rs

@@ -0,0 +1,50 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: 2023 Snowfork <hello@snowfork.com>
+use snowbridge_core::{inbound::Log, ChannelId};
+
+use sp_core::{RuntimeDebug, H160, H256};
+use sp_std::prelude::*;
+
+use alloy_primitives::B256;
+use alloy_sol_types::{sol, SolEvent};
+
+sol! {
+	event OutboundMessageAccepted(bytes32 indexed channel_id, uint64 nonce, bytes32 indexed message_id, bytes payload);
+}
+
+/// An inbound message that has had its outer envelope decoded.
+#[derive(Clone, RuntimeDebug)]
+pub struct Envelope {
+    /// The address of the outbound queue on Ethereum that emitted this message as an event log
+    pub gateway: H160,
+    /// The message Channel
+    pub channel_id: ChannelId,
+    /// A nonce for enforcing replay protection and ordering.
+    pub nonce: u64,
+    /// An id for tracing the message on its route (has no role in bridge consensus)
+    pub message_id: H256,
+    /// The inner payload generated from the source application.
+    pub payload: Vec<u8>,
+}
+
+#[derive(Copy, Clone, RuntimeDebug)]
+pub struct EnvelopeDecodeError;
+
+impl TryFrom<&Log> for Envelope {
+    type Error = EnvelopeDecodeError;
+
+    fn try_from(log: &Log) -> Result<Self, Self::Error> {
+        let topics: Vec<B256> = log.topics.iter().map(|x| B256::from_slice(x.as_ref())).collect();
+
+        let event = OutboundMessageAccepted::decode_log(topics, &log.data, true)
+            .map_err(|_| EnvelopeDecodeError)?;
+
+        Ok(Self {
+            gateway: log.address,
+            channel_id: ChannelId::from(event.channel_id.as_ref()),
+            nonce: event.nonce,
+            message_id: H256::from(event.message_id.as_ref()),
+            payload: event.payload,
+        })
+    }
+}