#[frame_support::pallet]

pub mod pallet {

	use super::*;

	const STORAGE_VERSION: StorageVersion = StorageVersion::new(2);

	#[pallet::without_storage_info]

	#[pallet::storage_version(STORAGE_VERSION)]

	pub struct Pallet<T>(_);

	#[pallet::config]

	pub trait Config: frame_system::Config + configuration::Config + paras::Config {

		type AssignmentProvider: AssignmentProvider<BlockNumberFor<Self>>;

	}

	/// All the validator groups. One for each core. Indices are into `ActiveValidators` - not the

	/// broader set of Polkadot validators, but instead just the subset used for parachains during

	/// this session.

	///

	/// Bound: The number of cores is the sum of the numbers of parachains and parathread

	/// multiplexers. Reasonably, 100-1000. The dominant factor is the number of validators: safe

	/// upper bound at 10k.

	pub type ValidatorGroups<T> = StorageValue<_, Vec<Vec<ValidatorIndex>>, ValueQuery>;

	/// One entry for each availability core. The i'th parachain belongs to the i'th core, with the

	/// remaining cores all being on demand parachain multiplexers.

	///

	/// Bounded by the maximum of either of these two values:

	///   * The number of parachains and parathread multiplexers

	///   * The number of validators divided by `configuration.max_validators_per_core`.

	pub type AvailabilityCores<T: Config> = StorageValue<_, Vec<CoreOccupiedType<T>>, ValueQuery>;

	/// Representation of a core in `AvailabilityCores`.

	///

	/// This is not to be confused with `CoreState` which is an enriched variant of this and exposed

	/// to the node side. It also provides information about scheduled/upcoming assignments for

	/// example and is computed on the fly in the `availability_cores` runtime call.

	#[derive(Encode, Decode, TypeInfo, RuntimeDebug, PartialEq)]

	pub enum CoreOccupied<N> {

		/// A para is currently waiting for availability/inclusion on this core.

		Paras(ParasEntry<N>),

	}

	/// Convenience type alias for `CoreOccupied`.

	pub type CoreOccupiedType<T> = CoreOccupied<BlockNumberFor<T>>;

	impl<N> CoreOccupied<N> {

		/// Is core free?

		pub fn is_free(&self) -> bool {

			matches!(self, Self::Free)

		}

	}

	/// Reasons a core might be freed.

	#[derive(Clone, Copy)]

	pub enum FreedReason {

		/// The core's work concluded and the parablock assigned to it is considered available.

		Concluded,

		/// The core's work timed out.

		TimedOut,

	}

	/// The block number where the session start occurred. Used to track how many group rotations

	/// have occurred.

	///

	/// Note that in the context of parachains modules the session change is signaled during

	/// the block and enacted at the end of the block (at the finalization stage, to be exact).

	/// Thus for all intents and purposes the effect of the session change is observed at the

	/// block following the session change, block number of which we save in this storage value.

	pub type SessionStartBlock<T: Config> = StorageValue<_, BlockNumberFor<T>, ValueQuery>;

	/// One entry for each availability core. The `VecDeque` represents the assignments to be

	/// scheduled on that core. The value contained here will not be valid after the end of

	/// a block. Runtime APIs should be used to determine scheduled cores for the upcoming block.

	pub type ClaimQueue<T: Config> =

		StorageValue<_, BTreeMap<CoreIndex, VecDeque<ParasEntryType<T>>>, ValueQuery>;

	/// Assignments as tracked in the claim queue.

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

	pub struct ParasEntry<N> {

		/// The underlying [`Assignment`].

		pub assignment: Assignment,

		/// The number of times the entry has timed out in availability already.

		pub availability_timeouts: u32,

		/// The block height until this entry needs to be backed.

		///

		/// If missed the entry will be removed from the claim queue without ever having occupied

		/// the core.

		pub ttl: N,

	}

	/// Convenience type declaration for `ParasEntry`.

	pub type ParasEntryType<T> = ParasEntry<BlockNumberFor<T>>;

	impl<N> ParasEntry<N> {

		/// Create a new `ParasEntry`.

		pub fn new(assignment: Assignment, now: N) -> Self {

			ParasEntry { assignment, availability_timeouts: 0, ttl: now }

		}

		/// Return `Id` from the underlying `Assignment`.

		pub fn para_id(&self) -> ParaId {

			self.assignment.para_id()

		}

	}

	/// Availability timeout status of a core.

	pub(crate) struct AvailabilityTimeoutStatus<BlockNumber> {

		/// Is the core already timed out?

		///

		/// If this is true the core will be freed at this block.

		pub timed_out: bool,

		/// When does this core timeout.

		///

		/// The block number the core times out. If `timed_out` is true, this will correspond to

		/// now (current block number).

		pub live_until: BlockNumber,

	}

}

		let val = if idx != &CoreIndex(self.next_idx) {

			log::trace!(target: LOG_TARGET, "idx did not match claim queue idx: {:?} vs {:?}", idx, self.next_idx);

			(CoreIndex(self.next_idx), VecDeque::new())

			let (idx, q) = self.queue.next()?;

			(idx, q)

		self.next_idx += 1;

		Some(val)

			let group_base_size = validators

				.len()

				.checked_div(n_cores as usize)

				.defensive_proof("n_cores should not be 0")

				.unwrap_or(0);

			let n_larger_groups = validators

				.len()

				.checked_rem(n_cores as usize)

				.defensive_proof("n_cores should not be 0")

				.unwrap_or(0);

			// Groups contain indices into the validators from the session change notification,

			// which are already shuffled.

			let mut groups: Vec<Vec<ValidatorIndex>> = Vec::new();

			for i in 0..n_larger_groups {

				let offset = (group_base_size + 1) * i;

				groups.push(

					(0..group_base_size + 1)

						.map(|j| offset + j)

						.map(|j| ValidatorIndex(j as _))

						.collect(),

				);

			}

			for i in 0..(n_cores as usize - n_larger_groups) {

				let offset = (n_larger_groups * (group_base_size + 1)) + (i * group_base_size);

				groups.push(

					(0..group_base_size)

						.map(|j| offset + j)

						.map(|j| ValidatorIndex(j as _))

						.collect(),

				);

			}

			ValidatorGroups::<T>::set(groups);

					match core::mem::replace(&mut cores[freed_index.0 as usize], CoreOccupied::Free)

						CoreOccupied::Free => {},

						CoreOccupied::Paras(entry) => {

							match freed_reason {

								FreedReason::Concluded => {

									concluded_paras.insert(freed_index, entry.assignment);

								},

								FreedReason::TimedOut => {

									timedout_paras.insert(freed_index, entry);

								},

				match Self::remove_from_claim_queue(*core_idx, *para_id) {

					Err(e) => {

						log::debug!(

							"[occupied] error on remove_from_claim queue {}",

						None

					Ok((pos_in_claim_queue, pe)) => {

						availability_cores[core_idx.0 as usize] = CoreOccupied::Paras(pe);

						Some((*core_idx, pos_in_claim_queue))

					let mut i = 0;

					let mut num_dropped = 0;

					while i < core_claim_queue.len() {

						let maybe_dropped = if let Some(entry) = core_claim_queue.get(i) {

							if entry.ttl < now {

								core_claim_queue.remove(i)

								None

							None

						if let Some(dropped) = maybe_dropped {

							num_dropped += 1;

							T::AssignmentProvider::report_processed(dropped.assignment);

						} else {

							i += 1;

						}

					for _ in 0..num_dropped {

						if let Some(assignment) =

							T::AssignmentProvider::pop_assignment_for_core(core_idx)

						{

							core_claim_queue.push_back(ParasEntry::new(assignment, now + ttl));

						}

	pub(crate) fn group_validators(group_index: GroupIndex) -> Option<Vec<ValidatorIndex>> {

		ValidatorGroups::<T>::get().get(group_index.0 as usize).map(|g| g.clone())

	}

	pub(crate) fn group_assigned_to_core(

		core: CoreIndex,

		at: BlockNumberFor<T>,

	) -> Option<GroupIndex> {

		let config = configuration::ActiveConfig::<T>::get();

		let session_start_block = SessionStartBlock::<T>::get();

		if at < session_start_block {

			return None

		}

		let validator_groups = ValidatorGroups::<T>::get();

		if core.0 as usize >= validator_groups.len() {

			return None

		}

		let rotations_since_session_start: BlockNumberFor<T> =

			(at - session_start_block) / config.scheduler_params.group_rotation_frequency;

		let rotations_since_session_start =

			<BlockNumberFor<T> as TryInto<u32>>::try_into(rotations_since_session_start)

				.unwrap_or(0);

		// Error case can only happen if rotations occur only once every u32::max(),

		// so functionally no difference in behavior.

		let group_idx =

			(core.0 as usize + rotations_since_session_start as usize) % validator_groups.len();

		Some(GroupIndex(group_idx as u32))

	}

		move |pending_since| {

			let time_out_at = if times_out {

				pending_since + config.scheduler_params.paras_availability_period

				next_rotation + config.scheduler_params.paras_availability_period

			AvailabilityTimeoutStatus { timed_out: time_out_at <= now, live_until: time_out_at }

		}

	pub(crate) fn next_up_on_available(core: CoreIndex) -> Option<ScheduledCore> {

		ClaimQueue::<T>::get()

			.get(&core)

			.and_then(|a| a.front().map(|pe| Self::paras_entry_to_scheduled_core(pe)))

	}

	fn paras_entry_to_scheduled_core(pe: &ParasEntryType<T>) -> ScheduledCore {

		ScheduledCore { para_id: pe.para_id(), collator: None }

	}

	pub(crate) fn next_up_on_time_out(core: CoreIndex) -> Option<ScheduledCore> {

		let max_availability_timeouts = configuration::ActiveConfig::<T>::get()

			.scheduler_params

			.max_availability_timeouts;

		Self::next_up_on_available(core).or_else(|| {

			// Or, if none, the claim currently occupying the core,

			// as it would be put back on the queue after timing out if number of retries is not at

			// the maximum.

			let cores = AvailabilityCores::<T>::get();

			cores.get(core.0 as usize).and_then(|c| match c {

				CoreOccupied::Free => None,

				CoreOccupied::Paras(pe) =>

					if pe.availability_timeouts < max_availability_timeouts {

						Some(Self::paras_entry_to_scheduled_core(pe))

						None

			})

		})

	}

					CoreOccupied::Paras(entry) => {

						Self::maybe_push_assignment(entry);

					},

			for para_entry in claim_queue.into_iter().rev() {

				Self::maybe_push_assignment(para_entry);

			}

	fn maybe_push_assignment(pe: ParasEntryType<T>) {

		if pe.availability_timeouts == 0 {

			T::AssignmentProvider::push_back_assignment(pe.assignment);

		}

	}

		}

		let n_session_cores = T::AssignmentProvider::session_core_count();

		let cq = ClaimQueue::<T>::get();

		let config = configuration::ActiveConfig::<T>::get();

		// Extra sanity, config should already never be smaller than 1:

		let n_lookahead = config.scheduler_params.lookahead.max(1);

		let max_availability_timeouts = config.scheduler_params.max_availability_timeouts;

		let ttl = config.scheduler_params.ttl;

		for core_idx in 0..n_session_cores {

			let core_idx = CoreIndex::from(core_idx);

			let n_lookahead_used = cq.get(&core_idx).map_or(0, |v| v.len() as u32);

			if let Some(mut entry) = timedout_paras.remove(&core_idx) {

				if entry.availability_timeouts < max_availability_timeouts {

					entry.availability_timeouts += 1;

					if n_lookahead_used < n_lookahead {

						entry.ttl = now + ttl;

					} else {

						// Over max capacity, we need to bump ttl (we exceeded the claim queue

						// size, so otherwise the entry might get dropped before reaching the top):

						entry.ttl = now + ttl + One::one();

					}

					Self::add_to_claim_queue(core_idx, entry);

					// The claim has been added back into the claim queue.

					// Do not pop another assignment for the core.

					continue

					let ret = concluded_paras.insert(core_idx, entry.assignment);

					debug_assert!(ret.is_none());

			}

			if let Some(concluded_para) = concluded_paras.remove(&core_idx) {

				T::AssignmentProvider::report_processed(concluded_para);

			}

			for _ in n_lookahead_used..n_lookahead {

				if let Some(assignment) = T::AssignmentProvider::pop_assignment_for_core(core_idx) {

					Self::add_to_claim_queue(core_idx, ParasEntry::new(assignment, now + ttl));

				}

		debug_assert!(timedout_paras.is_empty());

		debug_assert!(concluded_paras.is_empty());

	fn add_to_claim_queue(core_idx: CoreIndex, pe: ParasEntryType<T>) {

		ClaimQueue::<T>::mutate(|la| {

			la.entry(core_idx).or_default().push_back(pe);

		});

	}

	fn remove_from_claim_queue(

		core_idx: CoreIndex,

		para_id: ParaId,

	) -> Result<(PositionInClaimQueue, ParasEntryType<T>), &'static str> {

		ClaimQueue::<T>::mutate(|cq| {

			let core_claims = cq.get_mut(&core_idx).ok_or("core_idx not found in lookahead")?;

			let pos = core_claims

				.iter()

				.position(|pe| pe.para_id() == para_id)

				.ok_or("para id not found at core_idx lookahead")?;

			let pe = core_claims.remove(pos).ok_or("remove returned None")?;

			Ok((pos as u32, pe))

		})

	}

	pub(crate) fn scheduled_paras() -> impl Iterator<Item = (CoreIndex, ParaId)> {

		let claim_queue = ClaimQueue::<T>::get();

		claim_queue

			.into_iter()

			.filter_map(|(core_idx, v)| v.front().map(|e| (core_idx, e.assignment.para_id())))

	}

				if core != CoreOccupied::Free {

					return None

				}

				let next_scheduled = queue.front()?;

				Some((core_idx, next_scheduled.assignment.para_id()))

	fn claim_queue_len() -> usize {

		ClaimQueue::<T>::get().iter().map(|la_vec| la_vec.1.len()).sum()

	}