#[frame_support::pallet]

	#[pallet::pallet]

	#[pallet::call]

		pub fn suspend_xcm_execution(origin: OriginFor<T>) -> DispatchResult {

			T::ControllerOrigin::ensure_origin(origin)?;

			QueueSuspended::<T>::try_mutate(|suspended| {

				if *suspended {

					Err(Error::<T>::AlreadySuspended.into())

					*suspended = true;

					Ok(())

			})

		pub fn resume_xcm_execution(origin: OriginFor<T>) -> DispatchResult {

			T::ControllerOrigin::ensure_origin(origin)?;

			QueueSuspended::<T>::try_mutate(|suspended| {

				if !*suspended {

					Err(Error::<T>::AlreadyResumed.into())

					*suspended = false;

					Ok(())

			})

		pub fn update_suspend_threshold(origin: OriginFor<T>, new: u32) -> DispatchResult {

			ensure_root(origin)?;

			QueueConfig::<T>::try_mutate(|data| {

				data.suspend_threshold = new;

				data.validate::<T>()

			})

		pub fn update_drop_threshold(origin: OriginFor<T>, new: u32) -> DispatchResult {

			ensure_root(origin)?;

			QueueConfig::<T>::try_mutate(|data| {

				data.drop_threshold = new;

				data.validate::<T>()

			})

		pub fn update_resume_threshold(origin: OriginFor<T>, new: u32) -> DispatchResult {

			ensure_root(origin)?;

			QueueConfig::<T>::try_mutate(|data| {

				data.resume_threshold = new;

				data.validate::<T>()

			})

	#[pallet::hooks]

		fn on_idle(_block: BlockNumberFor<T>, limit: Weight) -> Weight {

			let mut meter = WeightMeter::with_limit(limit);

			if meter.try_consume(Self::on_idle_weight()).is_err() {

				log::debug!(

					"Not enough weight for on_idle. {} < {}",

					Self::on_idle_weight(),

				return meter.consumed()

			}

			migration::v3::lazy_migrate_inbound_queue::<T>();

			meter.consumed()

		}

	#[pallet::generate_deposit(pub(super) fn deposit_event)]

		/// An HRMP message was sent to a sibling parachain.

		XcmpMessageSent { message_hash: XcmHash },

	#[pallet::error]

		/// Setting the queue config failed since one of its values was invalid.

		BadQueueConfig,

		/// The execution is already suspended.

		AlreadySuspended,

		/// The execution is already resumed.

		AlreadyResumed,

		/// There are too many active outbound channels.

		TooManyActiveOutboundChannels,

		/// The message is too big.

		TooBig,

	#[pallet::storage]

	#[pallet::storage]

	#[pallet::storage]

	#[pallet::storage]

	#[pallet::storage]

	#[pallet::storage]

	#[pallet::type_value]

	pub fn InitialFactor() -> FixedU128 {

		FixedU128::from_u32(1)

	}

	#[pallet::storage]

#[derive(Copy, Clone, Eq, PartialEq, Encode, Decode, RuntimeDebug, TypeInfo, MaxEncodedLen)]

	Ok,

	Suspended,

	pub fn new(recipient: ParaId) -> OutboundChannelDetails {

		OutboundChannelDetails {

			recipient,

			state: OutboundState::Ok,

			signals_exist: false,

			first_index: 0,

			last_index: 0,

		}

	}

	pub fn with_signals(mut self) -> OutboundChannelDetails {

		self.signals_exist = true;

		self

	}

	pub fn with_suspended_state(mut self) -> OutboundChannelDetails {

		self.state = OutboundState::Suspended;

		self

	}

	fn default() -> Self {

		// NOTE that these default values are only used on genesis. They should give a rough idea of

		// what to set these values to, but is in no way a requirement.

		Self {

			drop_threshold: 48,    // 64KiB * 48 = 3MiB

			suspend_threshold: 32, // 64KiB * 32 = 2MiB

			resume_threshold: 8,   // 64KiB * 8 = 512KiB

		}

	}

	pub fn validate<T: crate::Config>(&self) -> sp_runtime::DispatchResult {

		if self.resume_threshold < self.suspend_threshold &&

			self.suspend_threshold <= self.drop_threshold &&

			self.resume_threshold > 0

			Ok(())

			Err(Error::<T>::BadQueueConfig.into())

	}

#[derive(PartialEq, Eq, Copy, Clone, Encode, Decode, TypeInfo)]

	Suspend,

	Resume,

	fn send_fragment<Fragment: Encode>(

		recipient: ParaId,

		format: XcmpMessageFormat,

		fragment: Fragment,

	) -> Result<u32, MessageSendError> {

		let encoded_fragment = fragment.encode();

		let channel_info =

			T::ChannelInfo::get_channel_info(recipient).ok_or(MessageSendError::NoChannel)?;

		let max_message_size = channel_info.max_message_size.min(T::MaxPageSize::get()) as usize;

		let format_size = format.encoded_size();

		let size_to_check = encoded_fragment

			.len()

			.checked_add(format_size)

			.ok_or(MessageSendError::TooBig)?;

		if size_to_check > max_message_size {

			return Err(MessageSendError::TooBig)

		}

		let mut all_channels = <OutboundXcmpStatus<T>>::get();

		let channel_details = if let Some(details) =

			all_channels.iter_mut().find(|channel| channel.recipient == recipient)

			details

			all_channels.try_push(OutboundChannelDetails::new(recipient)).map_err(|e| {

				log::error!("Failed to activate HRMP channel: {:?}", e);

				MessageSendError::TooManyChannels

			})?;

			all_channels

				.last_mut()

				.expect("can't be empty; a new element was just pushed; qed")

		let have_active = channel_details.last_index > channel_details.first_index;

		// Try to append fragment to the last page, if there is enough space.

		// We return the size of the last page inside of the option, to not calculate it again.

		let appended_to_last_page = have_active

			.then(|| {

				<OutboundXcmpMessages<T>>::try_mutate(

					recipient,

					channel_details.last_index - 1,

					|page| {

						if XcmpMessageFormat::decode_with_depth_limit(

							MAX_XCM_DECODE_DEPTH,

							&mut &page[..],

						) != Ok(format)

							defensive!("Bad format in outbound queue; dropping message");

							return Err(())

						}

						if page.len() + encoded_fragment.len() > max_message_size {

							return Err(())

						}

						for frag in encoded_fragment.iter() {

							page.try_push(*frag)?;

						Ok(page.len())

					},

				)

				.ok()

			})

			.flatten();

		let (number_of_pages, last_page_size) = if let Some(size) = appended_to_last_page {

			let number_of_pages = (channel_details.last_index - channel_details.first_index) as u32;

			(number_of_pages, size)

			let page_index = channel_details.last_index;

			channel_details.last_index += 1;

			let mut new_page = format.encode();

			new_page.extend_from_slice(&encoded_fragment[..]);

			let last_page_size = new_page.len();

			let number_of_pages = (channel_details.last_index - channel_details.first_index) as u32;

			let bounded_page = BoundedVec::<u8, T::MaxPageSize>::try_from(new_page)

				.map_err(|_| MessageSendError::TooBig)?;

			let bounded_page = WeakBoundedVec::force_from(bounded_page.into_inner(), None);

			<OutboundXcmpMessages<T>>::insert(recipient, page_index, bounded_page);

			<OutboundXcmpStatus<T>>::put(all_channels);

			(number_of_pages, last_page_size)

		let total_size =

			number_of_pages.saturating_sub(1) * max_message_size as u32 + last_page_size as u32;

		let threshold = channel_info.max_total_size / delivery_fee_constants::THRESHOLD_FACTOR;

		if total_size > threshold {

			let message_size_factor = FixedU128::from((encoded_fragment.len() / 1024) as u128)

				.saturating_mul(delivery_fee_constants::MESSAGE_SIZE_FEE_BASE);

			Self::increase_fee_factor(recipient, message_size_factor);

		}

		Ok(number_of_pages)

	}

	fn send_signal(dest: ParaId, signal: ChannelSignal) -> Result<(), Error<T>> {

		let mut s = <OutboundXcmpStatus<T>>::get();

		if let Some(details) = s.iter_mut().find(|item| item.recipient == dest) {

			details.signals_exist = true;

		} else {

			s.try_push(OutboundChannelDetails::new(dest).with_signals())

				.map_err(|_| Error::<T>::TooManyActiveOutboundChannels)?;

		let page = BoundedVec::<u8, T::MaxPageSize>::try_from(

			(XcmpMessageFormat::Signals, signal).encode(),

		)

		.map_err(|_| Error::<T>::TooBig)?;

		let page = WeakBoundedVec::force_from(page.into_inner(), None);

		<SignalMessages<T>>::insert(dest, page);

		<OutboundXcmpStatus<T>>::put(s);

		Ok(())

	}

	fn suspend_channel(target: ParaId) {

		<OutboundXcmpStatus<T>>::mutate(|s| {

			if let Some(details) = s.iter_mut().find(|item| item.recipient == target) {

				let ok = details.state == OutboundState::Ok;

				defensive_assert!(ok, "WARNING: Attempt to suspend channel that was not Ok.");

				details.state = OutboundState::Suspended;

				if s.try_push(OutboundChannelDetails::new(target).with_suspended_state()).is_err() {

					defensive!("Cannot pause channel; too many outbound channels");

				}

		});

	}

	fn resume_channel(target: ParaId) {

		<OutboundXcmpStatus<T>>::mutate(|s| {

			if let Some(index) = s.iter().position(|item| item.recipient == target) {

				let suspended = s[index].state == OutboundState::Suspended;

				defensive_assert!(

					suspended,

					"WARNING: Attempt to resume channel that was not suspended."

				);

				if s[index].first_index == s[index].last_index {

					s.remove(index);

				} else {

					s[index].state = OutboundState::Ok;

				}

				defensive!("WARNING: Attempt to resume channel that was not suspended.");

		});

	}

	fn enqueue_xcmp_message(

		sender: ParaId,

		xcm: BoundedVec<u8, MaxXcmpMessageLenOf<T>>,

		meter: &mut WeightMeter,

	) -> Result<(), ()> {

		if meter.try_consume(T::WeightInfo::enqueue_xcmp_message()).is_err() {

			defensive!("Out of weight: cannot enqueue XCMP messages; dropping msg");

			return Err(())

		}

		let QueueConfigData { drop_threshold, .. } = <QueueConfig<T>>::get();

		let fp = T::XcmpQueue::footprint(sender);

		// Assume that it will not fit into the current page:

		let new_pages = fp.ready_pages.saturating_add(1);

		if new_pages > drop_threshold {

			log::error!("XCMP queue for sibling {:?} is full; dropping messages.", sender);

			return Err(())

		}

		T::XcmpQueue::enqueue_message(xcm.as_bounded_slice(), sender);

		Ok(())

	}

	pub(crate) fn take_first_concatenated_xcm(

		data: &mut &[u8],

		meter: &mut WeightMeter,

	) -> Result<BoundedVec<u8, MaxXcmpMessageLenOf<T>>, ()> {

		if data.is_empty() {

			return Err(())

		}

		if meter.try_consume(T::WeightInfo::take_first_concatenated_xcm()).is_err() {

			defensive!("Out of weight; could not decode all; dropping");

			return Err(())

		}

		let xcm = VersionedXcm::<()>::decode_with_depth_limit(MAX_XCM_DECODE_DEPTH, data)

			.map_err(|_| ())?;

		xcm.encode().try_into().map_err(|_| ())

	}

	pub fn on_idle_weight() -> Weight {

		<T as crate::Config>::WeightInfo::on_idle_good_msg()

			.max(<T as crate::Config>::WeightInfo::on_idle_large_msg())

	}

	fn on_queue_changed(para: ParaId, fp: QueueFootprint) {

		let QueueConfigData { resume_threshold, suspend_threshold, .. } = <QueueConfig<T>>::get();

		let mut suspended_channels = <InboundXcmpSuspended<T>>::get();

		let suspended = suspended_channels.contains(&para);

		if suspended && fp.ready_pages <= resume_threshold {

			if let Err(err) = Self::send_signal(para, ChannelSignal::Resume) {

				log::error!("defensive: Could not send resumption signal to inbound channel of sibling {:?}: {:?}; channel remains suspended.", para, err);

			} else {

				suspended_channels.remove(&para);

				<InboundXcmpSuspended<T>>::put(suspended_channels);

			}

		} else if !suspended && fp.ready_pages >= suspend_threshold {

			log::warn!("XCMP queue for sibling {:?} is full; suspending channel.", para);

			if let Err(err) = Self::send_signal(para, ChannelSignal::Suspend) {

				log::error!(

					"defensive: Could not send suspension signal; future messages may be dropped: {:?}", err

			} else if let Err(err) = suspended_channels.try_insert(para) {

				log::error!("Too many channels suspended; cannot suspend sibling {:?}: {:?}; further messages may be dropped.", para, err);

			} else {

				<InboundXcmpSuspended<T>>::put(suspended_channels);

			}

		}

	}

	fn is_paused(para: &ParaId) -> bool {

		if !QueueSuspended::<T>::get() {

			return false

		}

		// Make an exception for the superuser queue:

		let sender_origin = T::ControllerOriginConverter::convert_origin(

			(Parent, Parachain((*para).into())),

			OriginKind::Superuser,

		);

		let is_controller =

			sender_origin.map_or(false, |origin| T::ControllerOrigin::try_origin(origin).is_ok());

		!is_controller

	}

	fn handle_xcmp_messages<'a, I: Iterator<Item = (ParaId, RelayBlockNumber, &'a [u8])>>(

		iter: I,

		max_weight: Weight,

	) -> Weight {

		let mut meter = WeightMeter::with_limit(max_weight);

		for (sender, _sent_at, mut data) in iter {

			let format = match XcmpMessageFormat::decode(&mut data) {

				Ok(f) => f,

					defensive!("Unknown XCMP message format - dropping");

					continue

			match format {

					while !data.is_empty() {

						if meter

							.try_consume(

								T::WeightInfo::suspend_channel()

									.max(T::WeightInfo::resume_channel()),

							)

							.is_err()

							defensive!("Not enough weight to process signals - dropping");

							break

						}

						match ChannelSignal::decode(&mut data) {

							Ok(ChannelSignal::Suspend) => Self::suspend_channel(sender),

							Ok(ChannelSignal::Resume) => Self::resume_channel(sender),

								defensive!("Undecodable channel signal - dropping");

								break

					while !data.is_empty() {

						let Ok(xcm) = Self::take_first_concatenated_xcm(&mut data, &mut meter)

							defensive!("HRMP inbound decode stream broke; page will be dropped.",);

							break

						if let Err(()) = Self::enqueue_xcmp_message(sender, xcm, &mut meter) {

							defensive!(

								meter.consumed_ratio()

							break

						}

					defensive!("Blob messages are unhandled - dropping");

					continue

		meter.consumed()

	}

	fn take_outbound_messages(maximum_channels: usize) -> Vec<(ParaId, Vec<u8>)> {

		let mut statuses = <OutboundXcmpStatus<T>>::get();

		let old_statuses_len = statuses.len();

		let max_message_count = statuses.len().min(maximum_channels);

		let mut result = Vec::with_capacity(max_message_count);

		for status in statuses.iter_mut() {

				recipient: para_id,

				state: outbound_state,

				mut signals_exist,

				mut first_index,

				mut last_index,

			} = *status;

			let (max_size_now, max_size_ever) = match T::ChannelInfo::get_channel_status(para_id) {

					for i in first_index..last_index {

						<OutboundXcmpMessages<T>>::remove(para_id, i);

					}

					if signals_exist {

						<SignalMessages<T>>::remove(para_id);

					}

					*status = OutboundChannelDetails::new(para_id);

					continue

				ChannelStatus::Full => continue,

				ChannelStatus::Ready(n, e) => (n, e),

			};

			// This is a hard limit from the host config; not even signals can bypass it.

			if result.len() == max_message_count {

				break

			}

			let page = if signals_exist {

				let page = <SignalMessages<T>>::get(para_id);

				defensive_assert!(!page.is_empty(), "Signals must exist");

				if page.len() < max_size_now {

					<SignalMessages<T>>::remove(para_id);

					signals_exist = false;

					page

					defensive!("Signals should fit into a single page");

					continue

			} else if outbound_state == OutboundState::Suspended {

				continue

			} else if last_index > first_index {

				let page = <OutboundXcmpMessages<T>>::get(para_id, first_index);

				if page.len() < max_size_now {

					<OutboundXcmpMessages<T>>::remove(para_id, first_index);

					first_index += 1;

					page

					continue

				continue

			if first_index == last_index {

				first_index = 0;

				last_index = 0;

			}

			if page.len() > max_size_ever {

				defensive!("WARNING: oversize message in queue - dropping");

			} else {

				result.push((para_id, page.into_inner()));

			}

			let max_total_size = match T::ChannelInfo::get_channel_info(para_id) {

				Some(channel_info) => channel_info.max_total_size,

					log::warn!("calling `get_channel_info` with no RelevantMessagingState?!");

					MAX_POSSIBLE_ALLOCATION // We use this as a fallback in case the messaging state is not present

			let threshold = max_total_size.saturating_div(delivery_fee_constants::THRESHOLD_FACTOR);

			let remaining_total_size: usize = (first_index..last_index)

				.map(|index| OutboundXcmpMessages::<T>::decode_len(para_id, index).unwrap())

				.sum();

			if remaining_total_size <= threshold as usize {

				Self::decrease_fee_factor(para_id);

			}

			*status = OutboundChannelDetails {

				recipient: para_id,

				state: outbound_state,

				signals_exist,

				first_index,

				last_index,

			};

		debug_assert!(!statuses.iter().any(|s| s.signals_exist), "Signals should be handled");

		result.sort_by_key(|m| m.0);

		// Prune hrmp channels that became empty. Additionally, because it may so happen that we

		// only gave attention to some channels in `non_empty_hrmp_channels` it's important to

		// change the order. Otherwise, the next `on_finalize` we will again give attention

		// only to those channels that happen to be in the beginning, until they are emptied.

		// This leads to "starvation" of the channels near to the end.

		//

		// To mitigate this we shift all processed elements towards the end of the vector using

		// `rotate_left`. To get intuition how it works see the examples in its rustdoc.

		statuses.retain(|x| {

			x.state == OutboundState::Suspended || x.signals_exist || x.first_index < x.last_index

		});

		// old_status_len must be >= status.len() since we never add anything to status.

		let pruned = old_statuses_len - statuses.len();

		// removing an item from status implies a message being sent, so the result messages must

		// be no less than the pruned channels.

		let _ = statuses.try_rotate_left(result.len().saturating_sub(pruned)).defensive_proof(

			"Could not store HRMP channels config. Some HRMP channels may be broken.",

		);

		<OutboundXcmpStatus<T>>::put(statuses);

		result

	}

	fn validate(

		dest: &mut Option<Location>,

		msg: &mut Option<Xcm<()>>,

	) -> SendResult<(ParaId, VersionedXcm<()>)> {

		let d = dest.take().ok_or(SendError::MissingArgument)?;

		match d.unpack() {

			(1, [Parachain(id)]) => {

				let xcm = msg.take().ok_or(SendError::MissingArgument)?;

				let id = ParaId::from(*id);

				let price = T::PriceForSiblingDelivery::price_for_delivery(id, &xcm);

				let versioned_xcm = T::VersionWrapper::wrap_version(&d, xcm)

					.map_err(|()| SendError::DestinationUnsupported)?;

				versioned_xcm

					.validate_xcm_nesting()

					.map_err(|()| SendError::ExceedsMaxMessageSize)?;

				Ok(((id, versioned_xcm), price))

				*dest = Some(d);

				Err(SendError::NotApplicable)

	}

	fn deliver((id, xcm): (ParaId, VersionedXcm<()>)) -> Result<XcmHash, SendError> {

		let hash = xcm.using_encoded(sp_io::hashing::blake2_256);

		match Self::send_fragment(id, XcmpMessageFormat::ConcatenatedVersionedXcm, xcm) {

				Self::deposit_event(Event::XcmpMessageSent { message_hash: hash });

				Ok(hash)

			Err(e) => {

				log::error!(target: LOG_TARGET, "Deliver error: {e:?}");

				Err(SendError::Transport(e.into()))

	}

	fn get_messages() -> Vec<(VersionedLocation, Vec<VersionedXcm<()>>)> {

		OutboundXcmpMessages::<T>::iter()

			.map(|(para_id, _, messages)| {

				let mut data = &messages[..];

				let decoded_format =

					XcmpMessageFormat::decode_with_depth_limit(MAX_XCM_DECODE_DEPTH, &mut data)

						.unwrap();

				if decoded_format != XcmpMessageFormat::ConcatenatedVersionedXcm {

					panic!("Unexpected format.")

				}

				let mut decoded_messages = Vec::new();

				while !data.is_empty() {

					let decoded_message = VersionedXcm::<()>::decode_with_depth_limit(

						MAX_XCM_DECODE_DEPTH,

						&mut data,

					)

					.unwrap();

					decoded_messages.push(decoded_message);

				}

				(

					VersionedLocation::V4((Parent, Parachain(para_id.into())).into()),

					decoded_messages,

				)

			})

			.collect()

	}

	fn get_fee_factor(id: Self::Id) -> FixedU128 {

		<DeliveryFeeFactor<T>>::get(id)

	}

	fn increase_fee_factor(id: Self::Id, message_size_factor: FixedU128) -> FixedU128 {

		<DeliveryFeeFactor<T>>::mutate(id, |f| {

			*f = f.saturating_mul(

				delivery_fee_constants::EXPONENTIAL_FEE_BASE.saturating_add(message_size_factor),

			);

			*f

		})

	}

	fn decrease_fee_factor(id: Self::Id) -> FixedU128 {

		<DeliveryFeeFactor<T>>::mutate(id, |f| {

			*f = InitialFactor::get().max(*f / delivery_fee_constants::EXPONENTIAL_FEE_BASE);

			*f

		})

	}