// Copyright 2019-2021 Parity Technologies (UK) Ltd.
// This file is part of Parity Bridges Common.

// Parity Bridges Common is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.

// Parity Bridges Common is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU General Public License for more details.

// You should have received a copy of the GNU General Public License
// along with Parity Bridges Common.  If not, see <http://www.gnu.org/licenses/>.

//! Module that adds XCM support to bridge pallets. The pallet allows to dynamically
//! open and close bridges between local (to this pallet location) and remote XCM
//! destinations.
//!
//! The `pallet_xcm_bridge_hub` pallet is used to manage (open, close) bridges between chains from
//! different consensuses. The new extrinsics `fn open_bridge` and `fn close_bridge` are introduced.
//! Other chains can manage channels with different bridged global consensuses.
//!
//! # Concept of `lane` and `LaneId`
//!
//! There is another `pallet_bridge_messages` pallet that handles inbound/outbound lanes for
//! messages. Each lane is a unique connection between two chains from different consensuses and is
//! identified by `LaneId`. `LaneId` is generated once when a new bridge is requested by `fn
//! open_bridge`. It is generated by `BridgeLocations::calculate_lane_id` based on the following
//! parameters:
//! - Source `bridge_origin_universal_location` (latest XCM)
//! - Destination `bridge_destination_universal_location` (latest XCM)
//! - XCM version (both sides of the bridge must use the same parameters to generate the same
//!   `LaneId`)
//!   - `bridge_origin_universal_location`, `bridge_destination_universal_location` is converted to
//!     the `Versioned*` structs
//!
//! `LaneId` is expected to never change because:
//! - We need the same `LaneId` on both sides of the bridge, as `LaneId` is part of the message key
//!   proofs.
//! - Runtime upgrades are entirely asynchronous.
//! - We already have a running production Polkadot/Kusama bridge that uses `LaneId([0, 0, 0, 0])`.
//!
//! `LaneId` is backward compatible, meaning it can be encoded/decoded from the older format `[u8;
//! 4]` used for static lanes, as well as the new format `H256` generated by
//! `BridgeLocations::calculate_lane_id`.
//!
//! # Concept of `bridge` and `BridgeId`
//!
//! The `pallet_xcm_bridge_hub` pallet needs to store some metadata about opened bridges. The bridge
//! (or bridge metadata) is stored under the `BridgeId` key.
//!
//! `BridgeId` is generated from `bridge_origin_relative_location` and
//! `bridge_origin_universal_location` using the `latest` XCM structs. `BridgeId` is not transferred
//! over the bridge; it is only important for local consensus. It essentially serves as an index/key
//! to bridge metadata. All the XCM infrastructure around `XcmExecutor`, `SendXcm`, `ExportXcm` use
//! the `latest` XCM, so `BridgeId` must remain compatible with the `latest` XCM. For example, we
//! have an `ExportXcm` implementation in `exporter.rs` that handles the `ExportMessage` instruction
//! with `universal_source` and `destination` (latest XCM), so we need to create `BridgeId` and the
//! corresponding `LaneId`.
//!
//! # Migrations and State
//!
//! This pallet implements `try_state`, ensuring compatibility and checking everything so we know if
//! any migration is needed. `do_try_state` checks for `BridgeId` compatibility, which is
//! recalculated on runtime upgrade. Upgrading to a new XCM version should not break anything,
//! except removing older XCM versions. In such cases, we need to add migration for `BridgeId` and
//! stored `Versioned*` structs and update `LaneToBridge` mapping, but this won't affect `LaneId`
//! over the bridge.
//!
//! # How to Open a Bridge?
//!
//! The `pallet_xcm_bridge_hub` pallet has the extrinsic `fn open_bridge` and an important
//! configuration `pallet_xcm_bridge::Config::OpenBridgeOrigin`, which translates the call's
//! origin to the XCM `Location` and converts it to the `bridge_origin_universal_location`. With the
//! current setup, this origin/location is expected to be either the relay chain or a sibling
//! parachain as one side of the bridge. Another parameter is
//! `bridge_destination_universal_location`, which is the other side of the bridge from a different
//! global consensus.
//!
//! Every bridge between two XCM locations has a dedicated lane in associated
//! messages pallet. Assuming that this pallet is deployed at the bridge hub
//! parachain and there's a similar pallet at the bridged network, the dynamic
//! bridge lifetime is as follows:
//!
//! 1) the sibling parachain opens a XCMP channel with this bridge hub;
//!
//! 2) the sibling parachain funds its sovereign parachain account at this bridge hub. It shall hold
//!    enough funds to pay for the bridge (see `BridgeDeposit`);
//!
//! 3) the sibling parachain opens the bridge by sending XCM `Transact` instruction with the
//!    `open_bridge` call. The `BridgeDeposit` amount is reserved on the sovereign account of
//!    sibling parachain;
//!
//! 4) at the other side of the bridge, the same thing (1, 2, 3) happens. Parachains that need to
//!    connect over the bridge need to coordinate the moment when they start sending messages over
//!    the bridge. Otherwise they may lose messages and/or bundled assets;
//!
//! 5) when either side wants to close the bridge, it sends the XCM `Transact` with the
//!    `close_bridge` call. The bridge is closed immediately if there are no queued messages.
//!    Otherwise, the owner must repeat the `close_bridge` call to prune all queued messages first.
//!
//! The pallet doesn't provide any mechanism for graceful closure, because it always involves
//! some contract between two connected chains and the bridge hub knows nothing about that. It
//! is the task for the connected chains to make sure that all required actions are completed
//! before the closure. In the end, the bridge hub can't even guarantee that all messages that
//! are delivered to the destination, are processed in the way their sender expects. So if we
//! can't guarantee that, we shall not care about more complex procedures and leave it to the
//! participating parties.
//!
//! # Example
//!
//! Example of opening a bridge between some random parachains from Polkadot and Kusama:
//!
//! 0. Let's have:
//! 	- BridgeHubPolkadot with `UniversalLocation` = `[GlobalConsensus(Polkadot), Parachain(1002)]`
//! 	- BridgeHubKusama with `UniversalLocation` = `[GlobalConsensus(Kusama), Parachain(1002)]`
//! 1. The Polkadot local sibling parachain `Location::new(1, Parachain(1234))` must send some DOTs
//!    to its sovereign account on BridgeHubPolkadot to cover `BridgeDeposit`, fees for `Transact`,
//!    and the existential deposit.
//! 2. Send a call to the BridgeHubPolkadot from the local sibling parachain: `Location::new(1,
//!    Parachain(1234))` ``` xcm::Transact( origin_kind: OriginKind::Xcm,
//!    XcmOverBridgeHubKusama::open_bridge( VersionedInteriorLocation::V4([GlobalConsensus(Kusama),
//!    Parachain(4567)].into()), ); ) ```
//! 3. Check the stored bridge metadata and generated `LaneId`.
//! 4. The Kusama local sibling parachain `Location::new(1, Parachain(4567))` must send some KSMs to
//!    its sovereign account
//! on BridgeHubKusama to cover `BridgeDeposit`, fees for `Transact`, and the existential deposit.
//! 5. Send a call to the BridgeHubKusama from the local sibling parachain: `Location::new(1,
//!    Parachain(4567))` ``` xcm::Transact( origin_kind: OriginKind::Xcm,
//!    XcmOverBridgeHubKusama::open_bridge(
//!    VersionedInteriorLocation::V4([GlobalConsensus(Polkadot), Parachain(1234)].into()), ); ) ```
//! 6. Check the stored bridge metadata and generated `LaneId`.
//! 7. Both `LaneId`s from steps 3 and 6 must be the same (see above _Concept of `lane` and
//!    `LaneId`_).
//! 8. Run the bridge messages relayer for `LaneId`.
//! 9. Send messages from both sides.
//!
//! The opening bridge holds the configured `BridgeDeposit` from the origin's sovereign account, but
//! this deposit is returned when the bridge is closed with `fn close_bridge`.

#![warn(missing_docs)]
#![cfg_attr(not(feature = "std"), no_std)]

use bp_messages::{LaneState, MessageNonce};
use bp_runtime::{AccountIdOf, BalanceOf, RangeInclusiveExt};
pub use bp_xcm_bridge::{Bridge, BridgeId, BridgeLocations, BridgeState, Deposit, Receiver};
use bp_xcm_bridge::{
	BridgeLocationsError, ChannelStatusProvider as DispatchChannelStatusProvider, DepositOf,
	LocalXcmChannelManager,
};
use frame_support::{traits::fungible::MutateHold, DefaultNoBound};
use frame_system::Config as SystemConfig;
use pallet_bridge_messages::{Config as BridgeMessagesConfig, LanesManagerError};
use sp_std::{boxed::Box, vec::Vec};
use xcm::prelude::*;
use xcm_builder::DispatchBlob;
use xcm_executor::traits::ConvertLocation;

pub use bp_xcm_bridge::XcmAsPlainPayload;
pub use dispatcher::XcmBlobMessageDispatchResult;
pub use exporter::PalletAsHaulBlobExporter;
pub use pallet::*;

pub mod benchmarking;
pub mod congestion;
mod dispatcher;
mod exporter;
pub mod migration;
mod mock;
#[cfg(test)]
mod tests;
pub use weights::WeightInfo;
pub mod weights;

/// The target that will be used when publishing logs related to this pallet.
pub const LOG_TARGET: &str = "runtime::bridge-xcm";

#[frame_support::pallet]
pub mod pallet {
	use super::*;
	use frame_support::{
		pallet_prelude::*,
		traits::{tokens::Precision, Contains},
	};
	use frame_system::pallet_prelude::{BlockNumberFor, *};

	/// The reason for this pallet placing a hold on funds.
	#[pallet::composite_enum]
	pub enum HoldReason<I: 'static = ()> {
		/// The funds are held as a deposit for opened bridge.
		#[codec(index = 0)]
		BridgeDeposit,
	}

	#[pallet::config]
	#[pallet::disable_frame_system_supertrait_check]
	pub trait Config<I: 'static = ()>:
		BridgeMessagesConfig<Self::BridgeMessagesPalletInstance>
	{
		/// The overarching event type.
		type RuntimeEvent: From<Event<Self, I>>
			+ IsType<<Self as frame_system::Config>::RuntimeEvent>;
		/// Benchmarks results from runtime we're plugged into.
		type WeightInfo: WeightInfo;

		/// Runtime's universal location.
		type UniversalLocation: Get<InteriorLocation>;
		// TODO: https://github.com/paritytech/parity-bridges-common/issues/1666 remove `ChainId` and
		// replace it with the `NetworkId` - then we'll be able to use
		// `T as pallet_bridge_messages::Config<T::BridgeMessagesPalletInstance>::BridgedChain::NetworkId`
		/// Bridged network as relative location of bridged `GlobalConsensus`.
		#[pallet::constant]
		type BridgedNetwork: Get<Location>;
		/// Associated messages pallet instance that bridges us with the
		/// `BridgedNetworkId` consensus.
		type BridgeMessagesPalletInstance: 'static;

		/// Price of single message export to the bridged consensus (`Self::BridgedNetwork`).
		type MessageExportPrice: Get<Assets>;
		/// Checks the XCM version for the destination.
		type DestinationVersion: GetVersion;

		/// The origin that is allowed to call privileged operations on the pallet, e.g. open/close
		/// bridge for locations.
		type ForceOrigin: EnsureOrigin<<Self as SystemConfig>::RuntimeOrigin>;
		/// A set of XCM locations within local consensus system that are allowed to open
		/// bridges with remote destinations.
		type OpenBridgeOrigin: EnsureOrigin<
			<Self as SystemConfig>::RuntimeOrigin,
			Success = Location,
		>;
		/// A converter between a location and a sovereign account.
		type BridgeOriginAccountIdConverter: ConvertLocation<AccountIdOf<ThisChainOf<Self, I>>>;

		/// Amount of this chain native tokens that is reserved on the sibling parachain account
		/// when bridge open request is registered.
		#[pallet::constant]
		type BridgeDeposit: Get<BalanceOf<ThisChainOf<Self, I>>>;
		/// Currency used to pay for bridge registration.
		type Currency: MutateHold<
			AccountIdOf<ThisChainOf<Self, I>>,
			Balance = BalanceOf<ThisChainOf<Self, I>>,
			Reason = Self::RuntimeHoldReason,
		>;
		/// The overarching runtime hold reason.
		type RuntimeHoldReason: From<HoldReason<I>>;
		/// Do not hold `Self::BridgeDeposit` for the location of `Self::OpenBridgeOrigin`.
		/// For example, it is possible to make an exception for a system parachain or relay.
		type AllowWithoutBridgeDeposit: Contains<Location>;

		/// Local XCM channel manager. Dedicated to exporting capabilities when handling congestion
		/// with the sending side.
		type LocalXcmChannelManager: LocalXcmChannelManager<BridgeId>;
		/// XCM-level dispatcher for inbound bridge messages.
		type BlobDispatcher: DispatchBlob + DispatchChannelStatusProvider;
		/// Limits for handling congestion.
		type CongestionLimits: Get<congestion::CongestionLimits>;
	}

	/// An alias for the bridge metadata.
	pub type BridgeOf<T, I> = Bridge<ThisChainOf<T, I>, LaneIdOf<T, I>>;
	/// An alias for this chain.
	pub type ThisChainOf<T, I> =
		pallet_bridge_messages::ThisChainOf<T, <T as Config<I>>::BridgeMessagesPalletInstance>;
	/// An alias for lane identifier type.
	pub type LaneIdOf<T, I> =
		<T as BridgeMessagesConfig<<T as Config<I>>::BridgeMessagesPalletInstance>>::LaneId;
	/// An alias for the associated lanes manager.
	pub type LanesManagerOf<T, I> =
		pallet_bridge_messages::LanesManager<T, <T as Config<I>>::BridgeMessagesPalletInstance>;
	/// Weight info of the given pallet.
	pub type WeightInfoOf<T, I> = <T as Config<I>>::WeightInfo;

	#[pallet::pallet]
	pub struct Pallet<T, I = ()>(PhantomData<(T, I)>);

	#[pallet::hooks]
	impl<T: Config<I>, I: 'static> Hooks<BlockNumberFor<T>> for Pallet<T, I> {
		fn integrity_test() {
			assert!(
				Self::bridged_network_id().is_ok(),
				"Configured `T::BridgedNetwork`: {:?} does not contain `GlobalConsensus` junction with `NetworkId`!",
				T::BridgedNetwork::get()
			);
			assert!(
				T::CongestionLimits::get().is_valid(),
				"Configured `T::CongestionLimits`: {:?} are not valid!",
				T::CongestionLimits::get()
			);
		}

		#[cfg(feature = "try-runtime")]
		fn try_state(_n: BlockNumberFor<T>) -> Result<(), sp_runtime::TryRuntimeError> {
			Self::do_try_state()
		}
	}

	#[pallet::call]
	impl<T: Config<I>, I: 'static> Pallet<T, I> {
		/// Open a bridge between two locations.
		///
		/// The caller must be within the `T::OpenBridgeOrigin` filter (presumably: a sibling
		/// parachain or a parent relay chain). The `bridge_destination_universal_location` must be
		/// a destination within the consensus of the `T::BridgedNetwork` network.
		///
		/// The `BridgeDeposit` amount is reserved on the caller account. This deposit
		/// is unreserved after bridge is closed.
		///
		/// The states after this call: bridge is `Opened`, outbound lane is `Opened`, inbound lane
		/// is `Opened`.
		///
		/// Optional `maybe_notify` holds data about the `bridge_origin_relative_location` where
		/// notifications can be sent to handle congestion. The notification contains an encoded
		/// tuple `(BridgeId, bool)`, where `bool` is the `is_congested` flag.
		#[pallet::call_index(0)]
		#[pallet::weight(WeightInfoOf::<T, I>::open_bridge())]
		pub fn open_bridge(
			origin: OriginFor<T>,
			bridge_destination_universal_location: Box<VersionedInteriorLocation>,
			maybe_notify: Option<Receiver>,
		) -> DispatchResult {
			// check and compute required bridge locations and laneId
			let xcm_version = bridge_destination_universal_location.identify_version();
			let locations =
				Self::bridge_locations_from_origin(origin, bridge_destination_universal_location)?;
			let lane_id = locations.calculate_lane_id(xcm_version).map_err(|e| {
				log::trace!(
					target: LOG_TARGET,
					"calculate_lane_id error: {e:?}",
				);
				Error::<T, I>::BridgeLocations(e)
			})?;

			Self::do_open_bridge(locations, lane_id, true, maybe_notify)
		}

		/// Try to close the bridge.
		///
		/// Can only be called by the "owner" of this side of the bridge, meaning that the
		/// inbound XCM channel with the local origin chain is working.
		///
		/// Closed bridge is a bridge without any traces in the runtime storage. So this method
		/// first tries to prune all queued messages at the outbound lane. When there are no
		/// outbound messages left, outbound and inbound lanes are purged. After that, funds
		/// are returned back to the owner of this side of the bridge.
		///
		/// The number of messages that we may prune in a single call is limited by the
		/// `may_prune_messages` argument. If there are more messages in the queue, the method
		/// prunes exactly `may_prune_messages` and exits early. The caller may call it again
		/// until outbound queue is depleted and get his funds back.
		///
		/// The states after this call: everything is either `Closed`, or purged from the
		/// runtime storage.
		#[pallet::call_index(1)]
		#[pallet::weight(WeightInfoOf::<T, I>::close_bridge())]
		pub fn close_bridge(
			origin: OriginFor<T>,
			bridge_destination_universal_location: Box<VersionedInteriorLocation>,
			may_prune_messages: MessageNonce,
		) -> DispatchResult {
			// compute required bridge locations
			let locations =
				Self::bridge_locations_from_origin(origin, bridge_destination_universal_location)?;

			// TODO: https://github.com/paritytech/parity-bridges-common/issues/1760 - may do refund here, if
			// bridge/lanes are already closed + for messages that are not pruned

			// update bridge metadata - this also guarantees that the bridge is in the proper state
			let bridge =
				Bridges::<T, I>::try_mutate_exists(locations.bridge_id(), |bridge| match bridge {
					Some(bridge) => {
						bridge.state = BridgeState::Closed;
						Ok(bridge.clone())
					}
					None => Err(Error::<T, I>::UnknownBridge),
				})?;

			// close inbound and outbound lanes
			let lanes_manager = LanesManagerOf::<T, I>::new();
			let mut inbound_lane = lanes_manager
				.any_state_inbound_lane(bridge.lane_id)
				.map_err(Error::<T, I>::LanesManager)?;
			let mut outbound_lane = lanes_manager
				.any_state_outbound_lane(bridge.lane_id)
				.map_err(Error::<T, I>::LanesManager)?;

			// now prune queued messages
			let mut pruned_messages = 0;
			for _ in outbound_lane.queued_messages() {
				if pruned_messages == may_prune_messages {
					break;
				}

				outbound_lane.remove_oldest_unpruned_message();
				pruned_messages += 1;
			}

			// if there are outbound messages in the queue, just update states and early exit
			if !outbound_lane.queued_messages().is_empty() {
				// update lanes state. Under normal circumstances, following calls shall never fail
				inbound_lane.set_state(LaneState::Closed);
				outbound_lane.set_state(LaneState::Closed);

				// write something to log
				let enqueued_messages = outbound_lane.queued_messages().saturating_len();
				log::trace!(
					target: LOG_TARGET,
					"Bridge {:?} between {:?} and {:?} is closing lane_id: {:?}. {} messages remaining",
					locations.bridge_id(),
					locations.bridge_origin_universal_location(),
					locations.bridge_destination_universal_location(),
					bridge.lane_id,
					enqueued_messages,
				);

				// deposit the `ClosingBridge` event
				Self::deposit_event(Event::<T, I>::ClosingBridge {
					bridge_id: *locations.bridge_id(),
					lane_id: bridge.lane_id.into(),
					pruned_messages,
					enqueued_messages,
				});

				return Ok(());
			}

			// else we have pruned all messages, so lanes and the bridge itself may gone
			inbound_lane.purge();
			outbound_lane.purge();
			Bridges::<T, I>::remove(locations.bridge_id());
			LaneToBridge::<T, I>::remove(bridge.lane_id);

			// return deposit
			let released_deposit = if let Some(deposit) = bridge.deposit {
				T::Currency::release(
					&HoldReason::BridgeDeposit.into(),
					&deposit.account,
					deposit.amount,
					Precision::BestEffort,
				)
				.inspect_err(|e| {
					// we can't do anything here - looks like funds have been (partially) unreserved
					// before by someone else. Let's not fail, though - it'll be worse for the
					// caller
					log::error!(
						target: LOG_TARGET,
						"Failed to unreserve during the bridge {:?} closure with error: {e:?}",
						locations.bridge_id(),
					);
				})
				.map(|released| Deposit::new(deposit.account, released))
				.ok()
			} else {
				None
			};

			// write something to log
			log::trace!(
				target: LOG_TARGET,
				"Bridge {:?} between {:?} and {:?} has closed lane_id: {:?}, the bridge deposit {released_deposit:?} was returned",
				locations.bridge_id(),
				bridge.lane_id,
				locations.bridge_origin_universal_location(),
				locations.bridge_destination_universal_location(),
			);

			// deposit the `BridgePruned` event
			Self::deposit_event(Event::<T, I>::BridgePruned {
				bridge_id: *locations.bridge_id(),
				lane_id: bridge.lane_id.into(),
				bridge_deposit: released_deposit,
				pruned_messages,
			});

			Ok(())
		}

		/// Attempts to update the `maybe_notify` callback for receiving congestion notifications.
		///
		/// This can only be called by the "owner" of this side of the bridge, which means that the
		/// inbound XCM channel with the local origin chain is functional.
		#[pallet::call_index(2)]
		#[pallet::weight(WeightInfoOf::<T, I>::update_notification_receiver())]
		pub fn update_notification_receiver(
			origin: OriginFor<T>,
			bridge_destination_universal_location: Box<VersionedInteriorLocation>,
			maybe_notify: Option<Receiver>,
		) -> DispatchResult {
			let locations =
				Self::bridge_locations_from_origin(origin, bridge_destination_universal_location)?;
			Bridges::<T, I>::try_mutate_exists(locations.bridge_id(), |bridge| match bridge {
				Some(b) => {
					b.maybe_notify = maybe_notify;

					Self::deposit_event(Event::<T, I>::NotificationReceiverUpdated {
						bridge_id: *locations.bridge_id(),
						maybe_notify: b.maybe_notify.clone(),
					});
					Ok(())
				}
				None => Err(Error::<T, I>::UnknownBridge.into()),
			})
		}
	}

	impl<T: Config<I>, I: 'static> Pallet<T, I> {
		/// Open bridge for lane.
		pub fn do_open_bridge(
			locations: Box<BridgeLocations>,
			lane_id: T::LaneId,
			create_lanes: bool,
			maybe_notify: Option<Receiver>,
		) -> Result<(), DispatchError> {
			// reserve balance (if needed)
			let deposit = if T::AllowWithoutBridgeDeposit::contains(
				locations.bridge_origin_relative_location(),
			) {
				None
			} else {
				// get origin's sovereign account
				let bridge_owner_account = T::BridgeOriginAccountIdConverter::convert_location(
					locations.bridge_origin_relative_location(),
				)
				.ok_or(Error::<T, I>::InvalidBridgeOriginAccount)?;

				let deposit = T::BridgeDeposit::get();
				T::Currency::hold(
					&HoldReason::BridgeDeposit.into(),
					&bridge_owner_account,
					deposit,
				)
				.map_err(|e| {
					log::error!(
						target: LOG_TARGET,
						"Failed to hold bridge deposit: {deposit:?} \
						from bridge_owner_account: {bridge_owner_account:?} derived from \
						bridge_origin_relative_location: {:?} with error: {e:?}",
						locations.bridge_origin_relative_location(),
					);
					Error::<T, I>::FailedToReserveBridgeDeposit
				})?;
				Some(Deposit::new(bridge_owner_account, deposit))
			};

			// save bridge metadata
			Bridges::<T, I>::try_mutate(locations.bridge_id(), |bridge| match bridge {
				Some(_) => Err(Error::<T, I>::BridgeAlreadyExists),
				None => {
					*bridge = Some(BridgeOf::<T, I> {
						bridge_origin_relative_location: Box::new(
							locations.bridge_origin_relative_location().clone().into(),
						),
						bridge_origin_universal_location: Box::new(
							locations.bridge_origin_universal_location().clone().into(),
						),
						bridge_destination_universal_location: Box::new(
							locations
								.bridge_destination_universal_location()
								.clone()
								.into(),
						),
						state: BridgeState::Opened,
						deposit: deposit.clone(),
						lane_id,
						maybe_notify,
					});
					Ok(())
				}
			})?;
			// save lane to bridge mapping
			LaneToBridge::<T, I>::try_mutate(lane_id, |bridge| match bridge {
				Some(_) => Err(Error::<T, I>::BridgeAlreadyExists),
				None => {
					*bridge = Some(*locations.bridge_id());
					Ok(())
				}
			})?;

			if create_lanes {
				// create new lanes. Under normal circumstances, following calls shall never fail
				let lanes_manager = LanesManagerOf::<T, I>::new();
				lanes_manager
					.create_inbound_lane(lane_id)
					.map_err(Error::<T, I>::LanesManager)?;
				lanes_manager
					.create_outbound_lane(lane_id)
					.map_err(Error::<T, I>::LanesManager)?;
			}

			// write something to log
			log::trace!(
				target: LOG_TARGET,
				"Bridge {:?} between {:?} and {:?} has been opened using lane_id: {lane_id:?}",
				locations.bridge_id(),
				locations.bridge_origin_universal_location(),
				locations.bridge_destination_universal_location(),
			);

			// deposit `BridgeOpened` event
			Self::deposit_event(Event::<T, I>::BridgeOpened {
				bridge_id: *locations.bridge_id(),
				bridge_deposit: deposit,
				local_endpoint: Box::new(locations.bridge_origin_universal_location().clone()),
				remote_endpoint: Box::new(
					locations.bridge_destination_universal_location().clone(),
				),
				lane_id: lane_id.into(),
			});

			Ok(())
		}
	}

	impl<T: Config<I>, I: 'static> Pallet<T, I> {
		/// Return bridge endpoint locations and dedicated lane identifier. This method converts
		/// runtime `origin` argument to relative `Location` using the `T::OpenBridgeOrigin`
		/// converter.
		pub fn bridge_locations_from_origin(
			origin: OriginFor<T>,
			bridge_destination_universal_location: Box<VersionedInteriorLocation>,
		) -> Result<Box<BridgeLocations>, sp_runtime::DispatchError> {
			Self::bridge_locations(
				T::OpenBridgeOrigin::ensure_origin(origin)?,
				(*bridge_destination_universal_location)
					.try_into()
					.map_err(|_| Error::<T, I>::UnsupportedXcmVersion)?,
			)
		}

		/// Return bridge endpoint locations and dedicated **bridge** identifier (`BridgeId`).
		pub fn bridge_locations(
			bridge_origin_relative_location: Location,
			bridge_destination_universal_location: InteriorLocation,
		) -> Result<Box<BridgeLocations>, sp_runtime::DispatchError> {
			BridgeLocations::bridge_locations(
				T::UniversalLocation::get(),
				bridge_origin_relative_location,
				bridge_destination_universal_location,
				Self::bridged_network_id()?,
			)
			.map_err(|e| {
				log::trace!(
					target: LOG_TARGET,
					"bridge_locations error: {e:?}",
				);
				Error::<T, I>::BridgeLocations(e).into()
			})
		}

		/// Return bridge metadata by bridge_id
		pub fn bridge(bridge_id: &BridgeId) -> Option<BridgeOf<T, I>> {
			Bridges::<T, I>::get(bridge_id)
		}

		/// Return bridge metadata by lane_id
		pub fn bridge_by_lane_id(lane_id: &T::LaneId) -> Option<(BridgeId, BridgeOf<T, I>)> {
			LaneToBridge::<T, I>::get(lane_id)
				.and_then(|bridge_id| Self::bridge(&bridge_id).map(|bridge| (bridge_id, bridge)))
		}
	}

	impl<T: Config<I>, I: 'static> Pallet<T, I> {
		/// Returns some `NetworkId` if contains `GlobalConsensus` junction.
		pub fn bridged_network_id() -> Result<NetworkId, sp_runtime::DispatchError> {
			match T::BridgedNetwork::get().take_first_interior() {
				Some(GlobalConsensus(network)) => Ok(network),
				_ => Err(Error::<T, I>::BridgeLocations(
					BridgeLocationsError::InvalidBridgeDestination,
				)
				.into()),
			}
		}
	}

	#[cfg(feature = "runtime-benchmarks")]
	impl<T: Config<I>, I: 'static> Pallet<T, I> {
		/// Open bridge for lane.
		pub fn open_bridge_for_benchmarks(
			lane_id: LaneIdOf<T, I>,
			bridge_origin_relative_location: Location,
			bridge_destination_universal_location: InteriorLocation,
			create_lanes: bool,
			maybe_notify: Option<Receiver>,
			initial_balance: impl Fn() -> BalanceOf<ThisChainOf<T, I>>,
		) -> Result<Box<BridgeLocations>, sp_runtime::DispatchError> {
			let locations = Pallet::<T, I>::bridge_locations(
				bridge_origin_relative_location.into(),
				bridge_destination_universal_location.into(),
			)?;

			if !T::AllowWithoutBridgeDeposit::contains(locations.bridge_origin_relative_location())
			{
				let account_id = T::BridgeOriginAccountIdConverter::convert_location(
					locations.bridge_origin_relative_location(),
				)
				.ok_or(Error::<T, I>::InvalidBridgeOriginAccount)?;

				use frame_support::traits::fungible::Unbalanced;
				use sp_runtime::Saturating;
				T::Currency::increase_balance(
					&account_id,
					initial_balance().saturating_add(T::BridgeDeposit::get()),
					Precision::BestEffort,
				)?;
			}

			Pallet::<T, I>::do_open_bridge(locations.clone(), lane_id, create_lanes, maybe_notify)?;

			Ok(locations)
		}
	}

	#[cfg(any(test, feature = "try-runtime", feature = "std"))]
	impl<T: Config<I>, I: 'static> Pallet<T, I> {
		/// Ensure the correctness of the state of this pallet.
		pub fn do_try_state() -> Result<(), sp_runtime::TryRuntimeError> {
			use sp_std::collections::btree_set::BTreeSet;

			let mut lanes = BTreeSet::new();

			// check all known bridge configurations
			for (bridge_id, bridge) in Bridges::<T, I>::iter() {
				lanes.insert(Self::do_try_state_for_bridge(bridge_id, bridge)?);
			}
			ensure!(
				lanes.len() == Bridges::<T, I>::iter().count(),
				"Invalid `Bridges` configuration, probably two bridges handle the same laneId!"
			);
			ensure!(
				lanes.len() == LaneToBridge::<T, I>::iter().count(),
				"Invalid `LaneToBridge` configuration, probably missing or not removed laneId!"
			);

			// check connected `pallet_bridge_messages` state.
			Self::do_try_state_for_messages()
		}

		/// Ensure the correctness of the state of the bridge.
		pub fn do_try_state_for_bridge(
			bridge_id: BridgeId,
			bridge: BridgeOf<T, I>,
		) -> Result<T::LaneId, sp_runtime::TryRuntimeError> {
			log::info!(target: LOG_TARGET, "Checking `do_try_state_for_bridge` for bridge_id: {bridge_id:?} and bridge: {bridge:?}");

			// check `BridgeId` points to the same `LaneId` and vice versa.
			ensure!(
				Some(bridge_id) == LaneToBridge::<T, I>::get(bridge.lane_id),
				"Found `LaneToBridge` inconsistency for bridge_id - missing mapping!"
			);

			// check `pallet_bridge_messages` state for that `LaneId`.
			let lanes_manager = LanesManagerOf::<T, I>::new();
			ensure!(
				lanes_manager.any_state_inbound_lane(bridge.lane_id).is_ok(),
				"Inbound lane not found!",
			);
			ensure!(
				lanes_manager
					.any_state_outbound_lane(bridge.lane_id)
					.is_ok(),
				"Outbound lane not found!",
			);

			// check that `locations` are convertible to the `latest` XCM.
			let bridge_origin_relative_location_as_latest: &Location = bridge
				.bridge_origin_relative_location
				.try_as()
				.map_err(|_| {
					"`bridge.bridge_origin_relative_location` cannot be converted to the `latest` XCM, needs migration!"
				})?;
			let bridge_origin_universal_location_as_latest: &InteriorLocation = bridge.bridge_origin_universal_location
				.try_as()
				.map_err(|_| "`bridge.bridge_origin_universal_location` cannot be converted to the `latest` XCM, needs migration!")?;
			let bridge_destination_universal_location_as_latest: &InteriorLocation = bridge.bridge_destination_universal_location
				.try_as()
				.map_err(|_| "`bridge.bridge_destination_universal_location` cannot be converted to the `latest` XCM, needs migration!")?;

			// check `BridgeId` does not change
			ensure!(
				bridge_id == BridgeId::new(bridge_origin_universal_location_as_latest, bridge_destination_universal_location_as_latest),
				"`bridge_id` is different than calculated from `bridge_origin_universal_location_as_latest` and `bridge_destination_universal_location_as_latest`, needs migration!"
			);

			// check bridge account owner
			if let Some(deposit) = bridge.deposit {
				ensure!(
					T::BridgeOriginAccountIdConverter::convert_location(bridge_origin_relative_location_as_latest) == Some(deposit.account),
					"`bridge.deposit.account` is different than calculated from `bridge.bridge_origin_relative_location`, needs migration!"
				);
			}

			Ok(bridge.lane_id)
		}

		/// Ensure the correctness of the state of the connected `pallet_bridge_messages` instance.
		pub fn do_try_state_for_messages() -> Result<(), sp_runtime::TryRuntimeError> {
			// check that all `InboundLanes` laneIds have mapping to some bridge.
			for lane_id in pallet_bridge_messages::InboundLanes::<T, T::BridgeMessagesPalletInstance>::iter_keys() {
				log::info!(target: LOG_TARGET, "Checking `do_try_state_for_messages` for `InboundLanes`'s lane_id: {lane_id:?}...");
				ensure!(
					LaneToBridge::<T, I>::get(lane_id).is_some(),
					"Found `LaneToBridge` inconsistency for `InboundLanes`'s lane_id - missing mapping!"
				);
			}

			// check that all `OutboundLanes` laneIds have mapping to some bridge.
			for lane_id in pallet_bridge_messages::OutboundLanes::<T, T::BridgeMessagesPalletInstance>::iter_keys() {
				log::info!(target: LOG_TARGET, "Checking `do_try_state_for_messages` for `OutboundLanes`'s lane_id: {lane_id:?}...");
				ensure!(
					LaneToBridge::<T, I>::get(lane_id).is_some(),
					"Found `LaneToBridge` inconsistency for `OutboundLanes`'s lane_id - missing mapping!"
				);
			}

			Ok(())
		}
	}

	/// All registered bridges.
	#[pallet::storage]
	pub type Bridges<T: Config<I>, I: 'static = ()> =
		StorageMap<_, Identity, BridgeId, BridgeOf<T, I>>;
	/// All registered `lane_id` and `bridge_id` mappings.
	#[pallet::storage]
	pub type LaneToBridge<T: Config<I>, I: 'static = ()> =
		StorageMap<_, Identity, T::LaneId, BridgeId>;

	#[pallet::genesis_config]
	#[derive(DefaultNoBound)]
	pub struct GenesisConfig<T: Config<I>, I: 'static = ()> {
		/// Opened bridges.
		///
		/// Keep in mind that we are **NOT** reserving any amount for the bridges opened at
		/// genesis. We are **NOT** opening lanes, used by this bridge. It all must be done using
		/// other pallets genesis configuration or some other means.
		pub opened_bridges: Vec<(
			Location,
			InteriorLocation,
			Option<T::LaneId>,
			Option<Receiver>,
		)>,
		/// Dummy marker.
		#[serde(skip)]
		pub _phantom: sp_std::marker::PhantomData<(T, I)>,
	}

	#[pallet::genesis_build]
	impl<T: Config<I>, I: 'static> BuildGenesisConfig for GenesisConfig<T, I>
	where
		T: frame_system::Config<AccountId = AccountIdOf<ThisChainOf<T, I>>>,
	{
		fn build(&self) {
			for (
				bridge_origin_relative_location,
				bridge_destination_universal_location,
				maybe_lane_id,
				maybe_notify,
			) in &self.opened_bridges
			{
				let locations = Pallet::<T, I>::bridge_locations(
					bridge_origin_relative_location.clone(),
					bridge_destination_universal_location.clone().into(),
				)
				.expect("Invalid genesis configuration");

				let lane_id = match maybe_lane_id {
					Some(lane_id) => *lane_id,
					None => locations
						.calculate_lane_id(xcm::latest::VERSION)
						.expect("Valid locations"),
				};

				Pallet::<T, I>::do_open_bridge(locations, lane_id, true, maybe_notify.clone())
					.expect("Valid opened bridge!");
			}
		}
	}

	#[pallet::event]
	#[pallet::generate_deposit(pub(super) fn deposit_event)]
	pub enum Event<T: Config<I>, I: 'static = ()> {
		/// The bridge between two locations has been opened.
		BridgeOpened {
			/// Bridge identifier.
			bridge_id: BridgeId,
			/// Bridge deposit held.
			bridge_deposit: Option<DepositOf<ThisChainOf<T, I>>>,

			/// Universal location of local bridge endpoint.
			local_endpoint: Box<InteriorLocation>,
			/// Universal location of remote bridge endpoint.
			remote_endpoint: Box<InteriorLocation>,
			/// Lane identifier.
			lane_id: T::LaneId,
		},
		/// Bridge is going to be closed, but not yet fully pruned from the runtime storage.
		ClosingBridge {
			/// Bridge identifier.
			bridge_id: BridgeId,
			/// Lane identifier.
			lane_id: T::LaneId,
			/// Number of pruned messages during the close call.
			pruned_messages: MessageNonce,
			/// Number of enqueued messages that need to be pruned in follow up calls.
			enqueued_messages: MessageNonce,
		},
		/// Bridge has been closed and pruned from the runtime storage. It now may be reopened
		/// again by any participant.
		BridgePruned {
			/// Bridge identifier.
			bridge_id: BridgeId,
			/// Lane identifier.
			lane_id: T::LaneId,
			/// Amount of deposit released.
			bridge_deposit: Option<DepositOf<ThisChainOf<T, I>>>,
			/// Number of pruned messages during the close call.
			pruned_messages: MessageNonce,
		},
		/// The bridge has been updated with a new notification receiver.
		NotificationReceiverUpdated {
			/// Bridge identifier.
			bridge_id: BridgeId,
			/// Updated notification receiver.
			maybe_notify: Option<Receiver>,
		},
	}

	#[pallet::error]
	pub enum Error<T, I = ()> {
		/// Bridge locations error.
		BridgeLocations(BridgeLocationsError),
		/// Invalid local bridge origin account.
		InvalidBridgeOriginAccount,
		/// The bridge is already registered in this pallet.
		BridgeAlreadyExists,
		/// The local origin already owns a maximal number of bridges.
		TooManyBridgesForLocalOrigin,
		/// Trying to close already closed bridge.
		BridgeAlreadyClosed,
		/// Lanes manager error.
		LanesManager(LanesManagerError),
		/// Trying to access unknown bridge.
		UnknownBridge,
		/// The bridge origin can't pay the required amount for opening the bridge.
		FailedToReserveBridgeDeposit,
		/// The version of XCM location argument is unsupported.
		UnsupportedXcmVersion,
	}
}