pallet_messages/

lib.rs

//! Stores messages for `IPFS` and `OnChain` Schema payload locations
//!
//! ## Quick Links
//! - [Configuration: `Config`](Config)
//! - [Extrinsics: `Call`](Call)
//! - [Runtime API: `MessagesRuntimeApi`](../pallet_messages_runtime_api/trait.MessagesRuntimeApi.html)
//! - [Custom RPC API: `MessagesApiServer`](../pallet_messages_rpc/trait.MessagesApiServer.html)
//! - [Event Enum: `Event`](Event)
//! - [Error Enum: `Error`](Error)
#![doc = include_str!("../README.md")]
// Substrate macros are tripping the clippy::expect_used lint.
#![allow(clippy::expect_used)]
// Ensure we're `no_std` when compiling for Wasm.
#![cfg_attr(not(feature = "std"), no_std)]
// Strong Documentation Lints
#![deny(
	rustdoc::broken_intra_doc_links,
	rustdoc::missing_crate_level_docs,
	rustdoc::invalid_codeblock_attributes,
	missing_docs
)]

#[cfg(feature = "runtime-benchmarks")]
mod benchmarking;
#[cfg(test)]
mod tests;

pub mod weights;

mod types;

/// Storage migrations
pub mod migration;

use core::{convert::TryInto, fmt::Debug};
use frame_support::{ensure, pallet_prelude::Weight, traits::Get, BoundedVec};
use sp_runtime::DispatchError;

extern crate alloc;
extern crate core;

use alloc::vec::Vec;
use common_primitives::{
	cid::*,
	messages::*,
	msa::{DelegatorId, GrantValidator, MessageSourceId, MsaLookup, MsaValidator, ProviderId},
	schema::*,
};
use frame_support::dispatch::DispatchResult;
use parity_scale_codec::Encode;

#[cfg(feature = "runtime-benchmarks")]
use common_primitives::benchmarks::{MsaBenchmarkHelper, SchemaBenchmarkHelper};

pub use pallet::*;
pub use types::*;
pub use weights::*;

use common_primitives::node::BlockNumber;
use frame_system::pallet_prelude::*;

#[frame_support::pallet]
pub mod pallet {
	use super::*;
	use frame_support::pallet_prelude::*;

	/// The current storage version.
	pub const STORAGE_VERSION: StorageVersion = StorageVersion::new(3);

	#[pallet::config]
	pub trait Config: frame_system::Config {
		/// The overarching event type.
		#[allow(deprecated)]
		type RuntimeEvent: From<Event<Self>> + IsType<<Self as frame_system::Config>::RuntimeEvent>;

		/// Weight information for extrinsics in this pallet.
		type WeightInfo: WeightInfo;

		/// A type that will supply MSA related information
		type MsaInfoProvider: MsaLookup + MsaValidator<AccountId = Self::AccountId>;

		/// A type that will validate schema grants
		type SchemaGrantValidator: GrantValidator<IntentId, BlockNumberFor<Self>>;

		/// A type that will supply schema related information.
		type SchemaProvider: SchemaProvider<SchemaId>;

		/// The maximum size of a message payload bytes.
		#[pallet::constant]
		type MessagesMaxPayloadSizeBytes: Get<u32> + Clone + Debug + MaxEncodedLen;

		/// How often to emit status events during a storage migration.
		/// Try to make this larger than the number of message migrations that will fit
		/// in a block by weight, as multiple of these events in a block is not really useful
		/// or desireable.
		type MigrateEmitEvery: Get<u32> + Clone + Debug;

		#[cfg(feature = "runtime-benchmarks")]
		/// A set of helper functions for benchmarking.
		type MsaBenchmarkHelper: MsaBenchmarkHelper<Self::AccountId>;

		#[cfg(feature = "runtime-benchmarks")]
		/// A set of helper functions for benchmarking.
		type SchemaBenchmarkHelper: SchemaBenchmarkHelper;
	}

	#[pallet::pallet]
	#[pallet::storage_version(STORAGE_VERSION)]
	pub struct Pallet<T>(_);

	/// A temporary storage for getting the index for messages
	/// At the start of the next block this storage is set to 0
	#[pallet::storage]
	#[pallet::whitelist_storage]
	pub(super) type BlockMessageIndex<T: Config> = StorageValue<_, MessageIndex, ValueQuery>;

	/// Storage for messages
	#[pallet::storage]
	pub(super) type MessagesV3<T: Config> = StorageNMap<
		_,
		(
			storage::Key<Twox64Concat, BlockNumberFor<T>>,
			storage::Key<Twox64Concat, IntentId>,
			storage::Key<Twox64Concat, MessageIndex>,
		),
		Message<T::MessagesMaxPayloadSizeBytes>,
		OptionQuery,
	>;

	#[pallet::error]
	pub enum Error<T> {
		/// Deprecated: Too many messages are added to existing block
		TooManyMessagesInBlock,

		/// Message payload size is too large
		ExceedsMaxMessagePayloadSizeBytes,

		/// Type Conversion Overflow
		TypeConversionOverflow,

		/// Invalid Message Source Account
		InvalidMessageSourceAccount,

		/// Invalid SchemaId or Schema not found
		InvalidSchemaId,

		/// UnAuthorizedDelegate
		UnAuthorizedDelegate,

		/// Invalid payload location
		InvalidPayloadLocation,

		/// Unsupported CID version
		UnsupportedCidVersion,

		/// Invalid CID
		InvalidCid,
	}

	#[pallet::event]
	#[pallet::generate_deposit(pub(super) fn deposit_event)]
	pub enum Event<T: Config> {
		/// Deprecated: please use [`Event::MessagesInBlock`]
		/// Messages are stored for a specified schema id and block number
		MessagesStored {
			/// The schema for these messages
			schema_id: SchemaId,
			/// The block number for these messages
			block_number: BlockNumberFor<T>,
		},
		/// Messages stored in the current block
		MessagesInBlock,
		/// Event emitted during storage migration to track progress
		MessagesMigrated {
			/// The storage version being migrated from
			from_version: u16,
			/// The storage version being migrated to
			to_version: u16,
			/// Total number of messages migrated in the current migration
			cumulative_total_migrated: u64,
		},
	}

	#[pallet::hooks]
	impl<T: Config> Hooks<BlockNumberFor<T>> for Pallet<T> {
		fn on_initialize(_current: BlockNumberFor<T>) -> Weight {
			<BlockMessageIndex<T>>::set(0u16);
			// allocates 1 read and 1 write for any access of `MessageIndex` in every block
			T::DbWeight::get().reads(1u64).saturating_add(T::DbWeight::get().writes(1u64))
			// TODO: add retention policy execution GitHub Issue: #126 and #25
		}
	}

	#[pallet::call]
	impl<T: Config> Pallet<T> {
		/// Adds a message for a resource hosted on IPFS. The input consists of
		/// a Base32-encoded [CID](https://docs.ipfs.tech/concepts/content-addressing/#version-1-v1)
		/// as well as a 32-bit content length. The stored payload will contain the
		/// CID encoded as binary, as well as the 32-bit message content length.
		/// The actual message content will be on IPFS.
		///
		/// # Events
		/// * [`Event::MessagesInBlock`] - Messages Stored in the block
		///
		/// # Errors
		/// * [`Error::ExceedsMaxMessagePayloadSizeBytes`] - Payload is too large.
		/// * [`Error::InvalidSchemaId`] - Schema not found.
		/// * [`Error::InvalidPayloadLocation`] - The schema is not an IPFS payload location.
		/// * [`Error::InvalidMessageSourceAccount`] - Origin must be from an MSA.
		/// * [`Error::TypeConversionOverflow`] - Failed to add the message to storage as it is very full.
		/// * [`Error::UnsupportedCidVersion`] - CID version is not supported (V0).
		/// * [`Error::InvalidCid`] - Unable to parse provided CID.
		///
		#[pallet::call_index(0)]
		#[pallet::weight(T::WeightInfo::add_ipfs_message())]
		pub fn add_ipfs_message(
			origin: OriginFor<T>,
			#[pallet::compact] schema_id: SchemaId,
			cid: Vec<u8>,
			#[pallet::compact] payload_length: u32,
		) -> DispatchResult {
			let provider_key = ensure_signed(origin)?;
			let cid_binary = Self::validate_cid(&cid)?;
			let payload_tuple: OffchainPayloadType = (cid_binary, payload_length);
			let bounded_payload: BoundedVec<u8, T::MessagesMaxPayloadSizeBytes> = payload_tuple
				.encode()
				.try_into()
				.map_err(|_| Error::<T>::ExceedsMaxMessagePayloadSizeBytes)?;

			if let Some(schema) = T::SchemaProvider::get_schema_info_by_id(schema_id) {
				ensure!(
					schema.payload_location == PayloadLocation::IPFS,
					Error::<T>::InvalidPayloadLocation
				);

				let provider_msa_id = Self::find_msa_id(&provider_key)?;
				let current_block = frame_system::Pallet::<T>::block_number();
				if Self::add_message(
					provider_msa_id,
					None,
					bounded_payload,
					schema.intent_id,
					schema_id,
					current_block,
				)? {
					Self::deposit_event(Event::MessagesInBlock);
				}
				Ok(())
			} else {
				Err(Error::<T>::InvalidSchemaId.into())
			}
		}

		/// Add an on-chain message for a given schema id.
		///
		/// # Events
		/// * [`Event::MessagesInBlock`] - In the next block
		///
		/// # Errors
		/// * [`Error::ExceedsMaxMessagePayloadSizeBytes`] - Payload is too large.
		/// * [`Error::InvalidSchemaId`] - Schema not found.
		/// * [`Error::InvalidPayloadLocation`] - The schema is not an IPFS payload location.
		/// * [`Error::InvalidMessageSourceAccount`] - Origin must be from an MSA.
		/// * [`Error::UnAuthorizedDelegate`] - Trying to add a message without a proper delegation between the origin and the on_behalf_of MSA.
		/// * [`Error::TypeConversionOverflow`] - Failed to add the message to storage as it is very full.
		///
		#[pallet::call_index(1)]
		#[pallet::weight(T::WeightInfo::add_onchain_message(payload.len() as u32))]
		pub fn add_onchain_message(
			origin: OriginFor<T>,
			on_behalf_of: Option<MessageSourceId>,
			#[pallet::compact] schema_id: SchemaId,
			payload: Vec<u8>,
		) -> DispatchResult {
			let provider_key = ensure_signed(origin)?;

			let bounded_payload: BoundedVec<u8, T::MessagesMaxPayloadSizeBytes> =
				payload.try_into().map_err(|_| Error::<T>::ExceedsMaxMessagePayloadSizeBytes)?;

			if let Some(schema) = T::SchemaProvider::get_schema_info_by_id(schema_id) {
				ensure!(
					schema.payload_location == PayloadLocation::OnChain,
					Error::<T>::InvalidPayloadLocation
				);

				let provider_msa_id = Self::find_msa_id(&provider_key)?;
				let provider_id = ProviderId(provider_msa_id);

				let current_block = frame_system::Pallet::<T>::block_number();
				// On-chain messages either are sent from the user themselves, or on behalf of another MSA Id
				let maybe_delegator = match on_behalf_of {
					Some(delegator_msa_id) => {
						let delegator_id = DelegatorId(delegator_msa_id);
						T::SchemaGrantValidator::ensure_valid_grant(
							provider_id,
							delegator_id,
							schema.intent_id,
							current_block,
						)
						.map_err(|_| Error::<T>::UnAuthorizedDelegate)?;
						delegator_id
					},
					None => DelegatorId(provider_msa_id), // Delegate is also the Provider
				};

				if Self::add_message(
					provider_msa_id,
					Some(maybe_delegator.into()),
					bounded_payload,
					schema.intent_id,
					schema_id,
					current_block,
				)? {
					Self::deposit_event(Event::MessagesInBlock);
				}

				Ok(())
			} else {
				Err(Error::<T>::InvalidSchemaId.into())
			}
		}
	}
}

impl<T: Config> Pallet<T> {
	/// Stores a message for a given schema id.
	/// returns true if it needs to emit an event
	/// # Errors
	/// * [`Error::TypeConversionOverflow`]
	///
	pub fn add_message(
		provider_msa_id: MessageSourceId,
		msa_id: Option<MessageSourceId>,
		payload: BoundedVec<u8, T::MessagesMaxPayloadSizeBytes>,
		intent_id: IntentId,
		schema_id: SchemaId,
		current_block: BlockNumberFor<T>,
	) -> Result<bool, DispatchError> {
		let index = BlockMessageIndex::<T>::get();
		let first = index == 0;
		let msg = Message {
			schema_id,
			payload, // size is checked on top of extrinsic
			provider_msa_id,
			msa_id,
		};

		<MessagesV3<T>>::insert((current_block, intent_id, index), msg);
		BlockMessageIndex::<T>::set(index.saturating_add(1));
		Ok(first)
	}

	/// Resolve an MSA from an account key(key)
	/// An MSA Id associated with the account key is returned, if one exists.
	///
	/// # Errors
	/// * [`Error::InvalidMessageSourceAccount`]
	///
	pub fn find_msa_id(key: &T::AccountId) -> Result<MessageSourceId, DispatchError> {
		Ok(T::MsaInfoProvider::ensure_valid_msa_key(key)
			.map_err(|_| Error::<T>::InvalidMessageSourceAccount)?)
	}

	/// Retrieve the messages for a particular intent and block range (paginated)
	pub fn get_messages_by_intent_id(
		intent_id: IntentId,
		pagination: BlockPaginationRequest,
	) -> BlockPaginationResponse<MessageResponseV2> {
		let mut response = BlockPaginationResponse::new();

		// Request Validation
		if !pagination.validate() {
			return response
		}

		// Schema Fetch and Check
		let intent = match <T>::SchemaProvider::get_intent_by_id(intent_id) {
			Some(intent) => intent,
			None => return response,
		};

		let mut from_index: u32 = pagination.from_index;

		'block_loop: for block_number in pagination.from_block..pagination.to_block {
			let list: Vec<MessageResponseV2> = Self::get_messages_by_intent_and_block(
				intent_id,
				intent.payload_location,
				block_number.into(),
			);

			// Max messages in a block are constrained to MessageIndex (u16) by the storage,
			// so this is a safe type coercion. Just to be safe, we'll trap in in debug builds
			let list_size = list.len();
			debug_assert!(
				list_size <= MessageIndex::MAX.into(),
				"unexpected number of messages in block"
			);
			let list_size: u32 = list_size as u32;

			let iter = list.into_iter().skip(from_index.saturating_sub(1) as usize);
			// all subsequent blocks in this call should start at index 0
			from_index = 0;
			'message_loop: for (i, m) in iter.enumerate() {
				response.content.push(m);

				if response.check_end_condition_and_set_next_pagination(
					block_number,
					i as u32,
					list_size,
					&pagination,
				) {
					break 'block_loop;
				}

				continue 'message_loop;
			}
		}
		response
	}

	/// Gets messages for a given IntentId and block number.
	///
	/// Payload location is included to map to correct response (To avoid fetching the Intent in this method)
	///
	/// Result is a vector of [`MessageResponseV2`].
	///
	pub fn get_messages_by_intent_and_block(
		intent_id: IntentId,
		payload_location: PayloadLocation,
		block_number: BlockNumberFor<T>,
	) -> Vec<MessageResponseV2> {
		let block_number_value: BlockNumber = block_number.try_into().unwrap_or_default();

		match payload_location {
			PayloadLocation::Itemized | PayloadLocation::Paginated => Vec::new(),
			_ => {
				let mut messages: Vec<MessageResponseV2> =
					MessagesV3::<T>::iter_prefix((block_number, intent_id))
						.filter_map(|(index, msg)| {
							msg.map_to_response((block_number_value, payload_location, index))
						})
						.collect();
				messages.sort_by(|a, b| a.index.cmp(&b.index));
				messages
			},
		}
	}

	/// Validates a CID to conform to IPFS CIDv1 (or higher) formatting (does not validate decoded CID fields)
	///
	/// # Errors
	/// * [`Error::UnsupportedCidVersion`] - CID version is not supported (V0)
	/// * [`Error::InvalidCid`] - Unable to parse provided CID
	///
	pub fn validate_cid(in_cid: &[u8]) -> Result<Vec<u8>, DispatchError> {
		Ok(validate_cid(in_cid).map_err(|e| match e {
			CidError::UnsupportedCidVersion => Error::<T>::UnsupportedCidVersion,
			_ => Error::<T>::InvalidCid,
		})?)
	}
}