// Ensure we're `no_std` when compiling for Wasm.
#![cfg_attr(not(feature = "std"), no_std)]

use codec::{Decode, Encode, MaxEncodedLen};
use scale_info::TypeInfo;
use serde::{Deserialize, Deserializer};

use frame_support::{
    pallet_prelude::*,
    traits::{
        Currency, DisabledValidators, Get, OneSessionHandler, ValidatorSet,
        ValidatorSetWithIdentification,
    },
    WeakBoundedVec,
};
use frame_system::{
    offchain::{SendTransactionTypes, SubmitTransaction},
    pallet_prelude::*,
};

pub use pallet::*;

use sp_core::H256;
use sp_runtime::{
    offchain::{
        self as rt_offchain,
        http::PendingRequest,
        storage::StorageValueRef,
        storage_lock::{StorageLock, Time},
    },
    traits::{AtLeast32BitUnsigned, BlockNumberProvider, Convert, Saturating, UniqueSaturatedInto},
    Perbill, RuntimeAppPublic, RuntimeDebug,
};
use sp_staking::{
    offence::{Kind, Offence, ReportOffence},
    SessionIndex,
};
use sp_std::{collections::btree_map::BTreeMap, prelude::*, vec::Vec};

use ghost_networks::{
    NetworkData, NetworkDataBasicHandler, NetworkDataInspectHandler, NetworkDataMutateHandler,
    NetworkType,
};
use ghost_traits::exposure::ExposureListener;

pub mod migrations;
pub mod weights;
pub use crate::weights::WeightInfo;
mod benchmarking;
mod mock;
mod tests;

mod bitmap;
mod deserialisations;
mod evm_types;

use bitmap::{BitMap, Bucket};
use evm_types::{EvmResponse, EvmResponseType};

pub mod sr25519 {
    mod app_sr25519 {
        use sp_application_crypto::{app_crypto, sr25519, KeyTypeId};
        const SLOW_CLAP: KeyTypeId = KeyTypeId(*b"slow");
        app_crypto!(sr25519, SLOW_CLAP);
    }

    sp_application_crypto::with_pair! {
        pub type AuthorityPair = app_sr25519::Pair;
    }

    pub type AuthoritySignature = app_sr25519::Signature;
    pub type AuthorityId = app_sr25519::Public;
}

const LOG_TARGET: &str = "runtime::ghost-slow-clap";
const DB_PREFIX: &[u8] = b"slow_clap::";

const MIN_LOCK_GUARD_PERIOD: u64 = 15_000;
const FETCH_TIMEOUT_PERIOD: u64 = 3_000;
const LOCK_BLOCK_EXPIRATION: u64 = 20;

const ONE_HOUR_MILLIS: u64 = 3_600_000;
const BLOCK_CHECK_CYCLES: u64 = 8;
const BLOCK_COMMITMENT_DELAY: u64 = 100;

pub type AuthIndex = u32;
pub type ExternalBlockNumber = u64;

#[derive(
    RuntimeDebug,
    Default,
    Copy,
    Clone,
    Eq,
    PartialEq,
    Ord,
    PartialOrd,
    Encode,
    Decode,
    TypeInfo,
    MaxEncodedLen,
)]
pub struct CommitmentDetails<BlockNumberFor> {
    pub last_stored_block: ExternalBlockNumber,
    pub last_updated: BlockNumberFor,
    pub commits: u8,
}

#[derive(
    RuntimeDebug, Clone, Eq, PartialEq, Ord, PartialOrd, Encode, Decode, TypeInfo, MaxEncodedLen,
)]
pub struct BlockCommitment<NetworkId> {
    pub session_index: SessionIndex,
    pub authority_index: AuthIndex,
    pub network_id: NetworkId,
    pub last_stored_block: ExternalBlockNumber,
}

#[derive(
    RuntimeDebug, Clone, Eq, PartialEq, Ord, PartialOrd, Encode, Decode, TypeInfo, MaxEncodedLen,
)]
pub struct Clap<AccountId, NetworkId, Balance> {
    pub session_index: SessionIndex,
    pub authority_index: AuthIndex,
    pub transaction_hash: H256,
    pub block_number: ExternalBlockNumber,
    pub removed: bool,
    pub network_id: NetworkId,
    pub receiver: AccountId,
    pub amount: Balance,
}

#[derive(Clone, PartialEq, Encode, Decode, RuntimeDebug, TypeInfo)]
pub struct ApplauseDetail<NetworkId, Balance> {
    pub network_id: NetworkId,
    pub authorities: BitMap,
    pub clapped_amount: Balance,
    pub block_number: ExternalBlockNumber,
    pub finalized: bool,
}

impl<NetworkId, Balance: Default> ApplauseDetail<NetworkId, Balance> {
    pub fn new(
        network_id: NetworkId,
        block_number: ExternalBlockNumber,
        max_authorities: usize,
    ) -> Self {
        ApplauseDetail {
            network_id,
            block_number,
            finalized: false,
            clapped_amount: Default::default(),
            authorities: BitMap::new(max_authorities as AuthIndex),
        }
    }
}

#[derive(Default, Clone, Copy, PartialEq, Encode, Decode, RuntimeDebug, TypeInfo)]
pub struct PreparedApplause<AccountId, NetworkId, Balance> {
    pub network_id: NetworkId,
    pub receiver: AccountId,
    pub amount: Balance,
    pub block_number: ExternalBlockNumber,
}

#[cfg_attr(test, derive(PartialEq))]
enum OffchainErr<NetworkId> {
    HttpJsonParsingError,
    HttpBytesParsingError,
    ErrorInEvmResponse,
    NoStoredNetworks,
    NoRequestsSent,
    EmptyResponses,
    NotValidator,
    DifferentEvmResponseTypes,
    MissingBlockNumber(u32, u32),
    ContradictoryTransactionLogs(u32, u32),
    ContradictoryBlockMedian(
        ExternalBlockNumber,
        ExternalBlockNumber,
        ExternalBlockNumber,
    ),
    UnparsableRequestBody(Vec<u8>),
    NoEndpointAvailable(NetworkId),
    StorageRetrievalError(NetworkId),
    UtxoNotImplemented(NetworkId),
    UnknownNetworkType(NetworkId),
    OffchainTimeoutPeriod(NetworkId),
}

impl<NetworkId: core::fmt::Debug> core::fmt::Debug for OffchainErr<NetworkId> {
    fn fmt(&self, fmt: &mut core::fmt::Formatter) -> core::fmt::Result {
        match *self {
            OffchainErr::HttpJsonParsingError => {
                write!(fmt, "Failed to parse evm response as JSON.")
            }
            OffchainErr::HttpBytesParsingError => {
                write!(fmt, "Failed to parse evm response as bytes.")
            }
            OffchainErr::StorageRetrievalError(ref network_id) => write!(
                fmt,
                "Storage value found for network #{:?} but it's undecodable.",
                network_id
            ),
            OffchainErr::ErrorInEvmResponse => write!(fmt, "Error in evm reponse."),
            OffchainErr::NoStoredNetworks => {
                write!(fmt, "No networks stored for the offchain slow claps.")
            }
            OffchainErr::NoRequestsSent => write!(
                fmt,
                "Could not send a request to any available RPC ednpoint."
            ),
            OffchainErr::EmptyResponses => {
                write!(fmt, "No responses to be used by the offchain worker.")
            }
            OffchainErr::NotValidator => write!(fmt, "Not a validator to broadcast slow claps"),
            OffchainErr::DifferentEvmResponseTypes => write!(
                fmt,
                "Different endpoints returned conflicting response types."
            ),
            OffchainErr::MissingBlockNumber(ref index, ref length) => write!(
                fmt,
                "Could not get block response at index {index} where total length is {length}.",
            ),
            OffchainErr::ContradictoryBlockMedian(ref mid, ref prev, ref distance) => write!(
                fmt,
                "Contradictory block median: values are {prev} {mid} while max distance is {distance}.",
            ),
            OffchainErr::ContradictoryTransactionLogs(ref count, ref number) => write!(
                fmt,
                "Contradictory tx logs: {number} event sequences from {count} endpoints.",
            ),
            OffchainErr::UnparsableRequestBody(ref bytes) => write!(
                fmt,
                "Could not get valid utf8 request body from bytes: {:?}.",
                bytes
            ),
            OffchainErr::NoEndpointAvailable(ref network_id) => write!(
                fmt,
                "No RPC endpoint available for network #{:?}.",
                network_id
            ),
            OffchainErr::UtxoNotImplemented(ref network_id) => write!(
                fmt,
                "Network #{:?} is marked as UTXO, which is not implemented yet.",
                network_id
            ),
            OffchainErr::UnknownNetworkType(ref network_id) => {
                write!(fmt, "Unknown type for network #{:?}.", network_id)
            }
            OffchainErr::OffchainTimeoutPeriod(ref network_id) => write!(
                fmt,
                "Offchain request should be in-flight for network #{:?}.",
                network_id
            ),
        }
    }
}

pub type NetworkIdOf<T> = <<T as Config>::NetworkDataHandler as NetworkDataBasicHandler>::NetworkId;

pub type BalanceOf<T> =
    <<T as Config>::Currency as Currency<<T as frame_system::Config>::AccountId>>::Balance;

pub type ValidatorId<T> = <<T as Config>::ValidatorSet as ValidatorSet<
    <T as frame_system::Config>::AccountId,
>>::ValidatorId;

pub type IdentificationTuple<T> = (
    ValidatorId<T>,
    <<T as Config>::ValidatorSet as ValidatorSetWithIdentification<
        <T as frame_system::Config>::AccountId,
    >>::Identification,
);

type OffchainResult<T, A> = Result<A, OffchainErr<NetworkIdOf<T>>>;

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

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

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

    #[pallet::config]
    pub trait Config: SendTransactionTypes<Call<Self>> + frame_system::Config {
        type RuntimeEvent: From<Event<Self>> + IsType<<Self as frame_system::Config>::RuntimeEvent>;
        type AuthorityId: Member
            + Parameter
            + RuntimeAppPublic
            + Ord
            + MaybeSerializeDeserialize
            + MaxEncodedLen;

        type ValidatorSet: ValidatorSetWithIdentification<Self::AccountId>;
        type Currency: Currency<Self::AccountId>;
        type NetworkDataHandler: NetworkDataBasicHandler
            + NetworkDataInspectHandler<NetworkData>
            + NetworkDataMutateHandler<NetworkData, BalanceOf<Self>>;
        type BlockNumberProvider: BlockNumberProvider<BlockNumber = BlockNumberFor<Self>>;
        type ReportUnresponsiveness: ReportOffence<
            Self::AccountId,
            IdentificationTuple<Self>,
            SlowClapOffence<IdentificationTuple<Self>>,
        >;
        type DisabledValidators: DisabledValidators;
        type ExposureListener: ExposureListener<BalanceOf<Self>, Self::AccountId>;

        #[pallet::constant]
        type MaxAuthorities: Get<u32>;

        #[pallet::constant]
        type ApplauseThreshold: Get<u32>;

        #[pallet::constant]
        type UnsignedPriority: Get<TransactionPriority>;

        #[pallet::constant]
        type HistoryDepth: Get<SessionIndex>;

        #[pallet::constant]
        type MinAuthoritiesNumber: Get<u32>;

        #[pallet::constant]
        type EpochDuration: Get<u64>;

        type WeightInfo: WeightInfo;
    }

    #[pallet::event]
    #[pallet::generate_deposit(pub(super) fn deposit_event)]
    pub enum Event<T: Config> {
        BlackSwan,
        AuthoritiesCommitmentEquilibrium,
        AuthoritiesApplauseEquilibrium,
        SomeAuthoritiesDelayed {
            delayed: Vec<IdentificationTuple<T>>,
        },
        SomeAuthoritiesTrottling {
            throttling: Vec<IdentificationTuple<T>>,
        },
        Clapped {
            authority_id: AuthIndex,
            network_id: NetworkIdOf<T>,
            clap_unique_hash: H256,
            receiver: T::AccountId,
            amount: BalanceOf<T>,
            removed: bool,
        },
        Applaused {
            network_id: NetworkIdOf<T>,
            receiver: T::AccountId,
            received_amount: BalanceOf<T>,
            block_number: ExternalBlockNumber,
        },
        BlockCommited {
            authority_id: AuthIndex,
            network_id: NetworkIdOf<T>,
        },
        BlockCommitmentsCheck {
            network_id: NetworkIdOf<T>,
            block_number: BlockNumberFor<T>,
        },
    }

    #[pallet::error]
    pub enum Error<T> {
        NotEnoughClaps,
        AlreadyClapped,
        UnregistedNetwork,
        UnregisteredClapRemove,
        TooMuchAuthorities,
        CouldNotAccumulateCommission,
        CouldNotAccumulateIncomingImbalance,
        CouldNotIncreaseGatekeeperAmount,
        NonExistentAuthorityIndex,
        TimeWentBackwards,
        InnerTimeWentBackwards,
        DisabledAuthority,
        ExecutedBlockIsHigher,
        CommitInWrongSession,
    }

    #[pallet::storage]
    #[pallet::getter(fn total_exposure)]
    pub(super) type TotalExposure<T: Config> = StorageValue<_, BalanceOf<T>, ValueQuery>;

    #[pallet::storage]
    #[pallet::getter(fn latest_executed_block)]
    pub(super) type LatestExecutedBlock<T: Config> =
        StorageMap<_, Twox64Concat, NetworkIdOf<T>, ExternalBlockNumber, ValueQuery>;

    #[pallet::storage]
    #[pallet::getter(fn block_commitments)]
    pub(super) type BlockCommitments<T: Config> = StorageMap<
        _,
        Twox64Concat,
        NetworkIdOf<T>,
        BTreeMap<AuthIndex, CommitmentDetails<BlockNumberFor<T>>>,
        ValueQuery,
    >;

    #[pallet::storage]
    #[pallet::getter(fn applause_details)]
    pub(super) type ApplauseDetails<T: Config> = StorageDoubleMap<
        _,
        Twox64Concat,
        SessionIndex,
        Twox64Concat,
        H256,
        ApplauseDetail<NetworkIdOf<T>, BalanceOf<T>>,
        OptionQuery,
    >;

    #[pallet::storage]
    #[pallet::getter(fn disabled_authority_indexes)]
    pub(super) type DisabledAuthorityIndexes<T: Config> =
        StorageMap<_, Twox64Concat, SessionIndex, BitMap, OptionQuery>;

    #[pallet::storage]
    #[pallet::getter(fn authorities)]
    pub(super) type Authorities<T: Config> = StorageMap<
        _,
        Twox64Concat,
        SessionIndex,
        WeakBoundedVec<T::AuthorityId, T::MaxAuthorities>,
        ValueQuery,
    >;

    #[pallet::storage]
    #[pallet::getter(fn validators)]
    pub(super) type Validators<T: Config> = StorageMap<
        _,
        Twox64Concat,
        SessionIndex,
        WeakBoundedVec<ValidatorId<T>, T::MaxAuthorities>,
        ValueQuery,
    >;

    #[pallet::genesis_config]
    #[derive(frame_support::DefaultNoBound)]
    pub struct GenesisConfig<T: Config> {
        pub authorities: Vec<T::AuthorityId>,
    }

    #[pallet::genesis_build]
    impl<T: Config> BuildGenesisConfig for GenesisConfig<T> {
        fn build(&self) {
            if !self.authorities.is_empty() {
                Pallet::<T>::initialize_authorities(self.authorities.clone());
            }
        }
    }

    #[pallet::call]
    impl<T: Config> Pallet<T> {
        #[pallet::call_index(0)]
        #[pallet::weight((T::WeightInfo::slow_clap(), DispatchClass::Normal, Pays::No))]
        pub fn slow_clap(
            origin: OriginFor<T>,
            clap: Clap<T::AccountId, NetworkIdOf<T>, BalanceOf<T>>,
            // since signature verification is done in `validate_unsigned`
            // we can skip doing it here again.
            _signature: <T::AuthorityId as RuntimeAppPublic>::Signature,
        ) -> DispatchResult {
            ensure_none(origin)?;
            Self::try_slow_clap(&clap)?;
            Ok(())
        }

        #[pallet::call_index(1)]
        #[pallet::weight((T::WeightInfo::commit_block(), DispatchClass::Normal, Pays::No))]
        pub fn commit_block(
            origin: OriginFor<T>,
            block_commitment: BlockCommitment<NetworkIdOf<T>>,
            // since signature verification is done in `validate_unsigned`
            // we can skip doing it here again.
            _signature: <T::AuthorityId as RuntimeAppPublic>::Signature,
        ) -> DispatchResult {
            ensure_none(origin)?;
            Self::try_commit_block(&block_commitment)?;
            Ok(())
        }
    }

    #[pallet::hooks]
    impl<T: Config> Hooks<BlockNumberFor<T>> for Pallet<T> {
        fn on_initialize(current_block: BlockNumberFor<T>) -> Weight {
            let mut weight = T::DbWeight::get().reads(1);
            let networks_count = T::NetworkDataHandler::count();
            let current_block_number: ExternalBlockNumber = current_block.unique_saturated_into();

            let check_block_interval = T::EpochDuration::get().saturating_div(BLOCK_CHECK_CYCLES);
            if check_block_interval == 0 {
                return weight;
            }

            let cycle_offset = current_block_number % check_block_interval;
            let block_in_epoch = current_block_number % T::EpochDuration::get();

            if cycle_offset >= networks_count.into() || block_in_epoch < check_block_interval {
                return weight;
            }

            let converted_block: usize = current_block_number.unique_saturated_into();

            if let Some((network_id, network_data)) =
                T::NetworkDataHandler::network_for_block(converted_block)
            {
                weight.saturating_accrue(T::DbWeight::get().reads_writes(3, 1));

                let session_index = T::ValidatorSet::session_index();
                let block_commitments = BlockCommitments::<T>::get(&network_id);
                let validators = Validators::<T>::get(&session_index);

                let validators_len = validators.len() as u32;

                if validators_len == 0 {
                    return weight;
                }

                let disabled_bitmap = DisabledAuthorityIndexes::<T>::get(&session_index)
                    .unwrap_or(BitMap::new(validators.len() as u32));

                let max_external_block_deviation = ONE_HOUR_MILLIS
                    .saturating_mul(6)
                    .saturating_div(network_data.avg_block_speed);

                let max_internal_block_deviation = check_block_interval.unique_saturated_into();

                let mut stored_blocks: Vec<(AuthIndex, ExternalBlockNumber)> = Vec::new();
                let mut block_updates: Vec<(AuthIndex, BlockNumberFor<T>)> = Vec::new();

                for authority_index in 0..validators_len {
                    let data = block_commitments
                        .get(&authority_index)
                        .copied()
                        .unwrap_or_default();
                    stored_blocks.push((authority_index, data.last_stored_block));
                    block_updates.push((authority_index, data.last_updated));
                }

                let stored_blocks_offence_bitmap = Self::capture_deviation_in_commitments(
                    &disabled_bitmap,
                    &mut stored_blocks,
                    validators_len,
                    max_external_block_deviation,
                );

                let block_updates_offence_bitmap = Self::capture_deviation_in_commitments(
                    &disabled_bitmap,
                    &mut block_updates,
                    validators_len,
                    max_internal_block_deviation,
                );

                let offence_bitmap =
                    stored_blocks_offence_bitmap.bitor(block_updates_offence_bitmap);

                let offence_type = OffenceType::CommitmentOffence;
                let extra_weight = Self::try_offend_validators(
                    &session_index,
                    &validators,
                    offence_bitmap,
                    disabled_bitmap,
                    offence_type,
                );

                Self::deposit_event(Event::<T>::BlockCommitmentsCheck {
                    block_number: current_block,
                    network_id,
                });

                weight.saturating_accrue(extra_weight);
            }

            weight
        }

        fn offchain_worker(now: BlockNumberFor<T>) {
            match Self::start_slow_clapping(now) {
                Ok(_) => {
                    log::info!(target: LOG_TARGET, "👻 Slow Clap #{:?} finished gracefully", now)
                }
                Err(e) => log::info!(target: LOG_TARGET, "👻 Slow Clap #{:?} failed: {:?}", now, e),
            }
        }
    }

    #[pallet::validate_unsigned]
    impl<T: Config> ValidateUnsigned for Pallet<T> {
        type Call = Call<T>;
        fn validate_unsigned(_source: TransactionSource, call: &Self::Call) -> TransactionValidity {
            match call {
                Call::commit_block {
                    block_commitment,
                    signature,
                } => {
                    let session_index = block_commitment.session_index;
                    let authority_index = block_commitment.authority_index;

                    match Authorities::<T>::get(&session_index).get(authority_index as usize) {
                        Some(authority) => {
                            if !block_commitment
                                .using_encoded(|encoded| authority.verify(&encoded, signature))
                            {
                                return InvalidTransaction::BadProof.into();
                            }
                        }
                        None => return InvalidTransaction::BadSigner.into(),
                    }

                    ValidTransaction::with_tag_prefix("SlowClap")
                        .priority(T::UnsignedPriority::get())
                        .and_provides(block_commitment.encode())
                        .longevity(LOCK_BLOCK_EXPIRATION)
                        .propagate(true)
                        .build()
                }
                Call::slow_clap { clap, signature } => {
                    let (session_index, _) = Self::mended_session_index(&clap);

                    match Authorities::<T>::get(&session_index).get(clap.authority_index as usize) {
                        Some(authority) => {
                            if !clap.using_encoded(|encoded| authority.verify(&encoded, signature))
                            {
                                return InvalidTransaction::BadProof.into();
                            }
                        }
                        None => return InvalidTransaction::BadSigner.into(),
                    }

                    ValidTransaction::with_tag_prefix("SlowClap")
                        .priority(T::UnsignedPriority::get())
                        .and_provides(signature)
                        .longevity(LOCK_BLOCK_EXPIRATION)
                        .propagate(true)
                        .build()
                }
                _ => InvalidTransaction::Call.into(),
            }
        }
    }
}

impl<T: Config> Pallet<T> {
    fn create_storage_key(first: &[u8], second: &[u8]) -> Vec<u8> {
        let mut key = DB_PREFIX.to_vec();
        key.extend(first);
        key.extend(second);
        key
    }

    fn read_persistent_offchain_storage<R: codec::Decode>(
        storage_key: &[u8],
        default_value: R,
    ) -> R {
        StorageValueRef::persistent(&storage_key)
            .get::<R>()
            .ok()
            .flatten()
            .unwrap_or(default_value)
    }

    fn generate_unique_hash(clap: &Clap<T::AccountId, NetworkIdOf<T>, BalanceOf<T>>) -> H256 {
        let mut clap_args_str = clap.receiver.encode();
        clap_args_str.extend(&clap.amount.encode());
        clap_args_str.extend(&clap.block_number.encode());
        clap_args_str.extend(&clap.network_id.encode());

        H256::from_slice(&sp_io::hashing::keccak_256(&clap_args_str)[..])
    }

    fn block_number_to_hexadecimal_bytes(value: ExternalBlockNumber) -> Vec<u8> {
        let mut hex_str = Vec::new();
        hex_str.push(b'0');
        hex_str.push(b'x');

        if value == 0 {
            hex_str.push(b'0');
            return hex_str;
        }

        for i in (0..16).rev() {
            let nibble = (value >> (i * 4)) & 0xF;
            if nibble != 0 || hex_str.len() > 2 {
                hex_str.push(match nibble {
                    0..=9 => b'0' + nibble as u8,
                    10..=15 => b'a' + (nibble - 10) as u8,
                    _ => unreachable!(),
                });
            }
        }

        hex_str
    }

    fn mended_session_index(
        clap: &Clap<T::AccountId, NetworkIdOf<T>, BalanceOf<T>>,
    ) -> (SessionIndex, H256) {
        let prev_session_index = clap.session_index.saturating_sub(1);
        let clap_unique_hash = Self::generate_unique_hash(&clap);

        let session_index =
            if ApplauseDetails::<T>::contains_key(&prev_session_index, &clap_unique_hash) {
                prev_session_index
            } else {
                clap.session_index
            };

        (session_index, clap_unique_hash)
    }

    fn try_slow_clap(clap: &Clap<T::AccountId, NetworkIdOf<T>, BalanceOf<T>>) -> DispatchResult {
        let network_id = clap.network_id;
        ensure!(
            T::NetworkDataHandler::contains_key(&network_id),
            Error::<T>::UnregistedNetwork
        );
        ensure!(
            LatestExecutedBlock::<T>::get(&network_id) <= clap.block_number,
            Error::<T>::ExecutedBlockIsHigher,
        );

        let (session_index, clap_unique_hash) = Self::mended_session_index(&clap);
        let authorities = Authorities::<T>::get(&session_index);
        let authorities_length = authorities.len();

        let not_disabled = DisabledAuthorityIndexes::<T>::get(&session_index)
            .map(|bitmap| !bitmap.exists(&clap.authority_index))
            .unwrap_or_default();

        ensure!(not_disabled, Error::<T>::DisabledAuthority);

        let applause_threshold = Perbill::from_parts(T::ApplauseThreshold::get());
        let threshold_amount = applause_threshold.mul_floor(TotalExposure::<T>::get());

        let account_id = T::ExposureListener::get_account_by_index(clap.authority_index as usize)
            .ok_or(Error::<T>::NonExistentAuthorityIndex)?;
        let new_clapped_amount = T::ExposureListener::get_validator_exposure(&account_id);

        let mut applause_details =
            ApplauseDetails::<T>::try_get(&session_index, &clap_unique_hash).unwrap_or(
                ApplauseDetail::new(network_id, clap.block_number, authorities_length),
            );

        let total_clapped = if clap.removed {
            ensure!(
                applause_details.authorities.exists(&clap.authority_index),
                Error::<T>::UnregisteredClapRemove,
            );
            applause_details.authorities.unset(clap.authority_index);
            applause_details
                .clapped_amount
                .saturating_sub(new_clapped_amount)
        } else {
            ensure!(
                !applause_details.authorities.exists(&clap.authority_index),
                Error::<T>::AlreadyClapped,
            );
            applause_details.authorities.set(clap.authority_index);
            applause_details
                .clapped_amount
                .saturating_add(new_clapped_amount)
        };

        applause_details.clapped_amount = total_clapped;

        Self::deposit_event(Event::<T>::Clapped {
            network_id,
            authority_id: clap.authority_index,
            clap_unique_hash,
            receiver: clap.receiver.clone(),
            amount: clap.amount,
            removed: clap.removed,
        });

        let is_enough = applause_details
            .authorities
            .count_ones()
            .gt(&(authorities_length as u32 / 2));

        if total_clapped > threshold_amount && is_enough && !applause_details.finalized {
            match Self::try_applause(&clap) {
                Ok(_) => applause_details.finalized = true,
                Err(e) => sp_runtime::print(e),
            }
        }

        ApplauseDetails::<T>::insert(&session_index, &clap_unique_hash, applause_details);

        Ok(())
    }

    fn try_applause(clap: &Clap<T::AccountId, NetworkIdOf<T>, BalanceOf<T>>) -> DispatchResult {
        let commission = T::NetworkDataHandler::get(&clap.network_id)
            .map(|network_data| Perbill::from_parts(network_data.incoming_fee))
            .unwrap_or_default()
            .mul_ceil(clap.amount);
        let final_amount = clap.amount.saturating_sub(commission);

        let _ = T::NetworkDataHandler::increase_gatekeeper_amount(&clap.network_id, &clap.amount)
            .map_err(|_| Error::<T>::CouldNotIncreaseGatekeeperAmount)?;
        let _ = T::NetworkDataHandler::accumulate_incoming_imbalance(&final_amount)
            .map_err(|_| Error::<T>::CouldNotAccumulateIncomingImbalance)?;
        let _ = T::NetworkDataHandler::accumulate_commission(&commission)
            .map_err(|_| Error::<T>::CouldNotAccumulateCommission)?;
        let _ = T::Currency::deposit_creating(&clap.receiver, final_amount);

        LatestExecutedBlock::<T>::set(clap.network_id, clap.block_number);

        Self::deposit_event(Event::<T>::Applaused {
            network_id: clap.network_id,
            receiver: clap.receiver.clone(),
            received_amount: final_amount,
            block_number: clap.block_number,
        });

        Ok(())
    }

    fn try_commit_block(new_commitment: &BlockCommitment<NetworkIdOf<T>>) -> DispatchResult {
        let authority_index = new_commitment.authority_index;
        let network_id = new_commitment.network_id;

        let session_index = T::ValidatorSet::session_index();
        let latest_executed_block = LatestExecutedBlock::<T>::get(&network_id);

        ensure!(
            T::NetworkDataHandler::contains_key(&network_id),
            Error::<T>::UnregistedNetwork
        );

        ensure!(
            new_commitment.session_index == session_index,
            Error::<T>::CommitInWrongSession,
        );

        BlockCommitments::<T>::try_mutate(&network_id, |current_commitments| -> DispatchResult {
            let current_block_number = Self::current_block_number();

            let (current_commits, current_last_updated) = current_commitments
                .get(&authority_index)
                .map(|details| {
                    (
                        details.commits.saturating_add(1),
                        details
                            .last_updated
                            .saturating_add(BLOCK_COMMITMENT_DELAY.unique_saturated_into()),
                    )
                })
                .unwrap_or((1, Default::default()));

            ensure!(
                current_block_number >= current_last_updated,
                Error::<T>::TimeWentBackwards,
            );

            ensure!(
                new_commitment.last_stored_block > latest_executed_block,
                Error::<T>::InnerTimeWentBackwards,
            );

            let new_commitment_details = CommitmentDetails {
                last_stored_block: new_commitment.last_stored_block,
                last_updated: Self::current_block_number(),
                commits: current_commits,
            };

            current_commitments.insert(authority_index, new_commitment_details);

            Ok(())
        })?;

        Self::deposit_event(Event::<T>::BlockCommited {
            network_id,
            authority_id: authority_index,
        });

        Ok(())
    }

    fn start_slow_clapping(block_number: BlockNumberFor<T>) -> OffchainResult<T, ()> {
        let converted_block: usize = block_number.unique_saturated_into();
        let session_index = T::ValidatorSet::session_index();
        let network_in_use = T::NetworkDataHandler::network_for_block(converted_block)
            .ok_or(OffchainErr::NoStoredNetworks)?;

        log::info!(
            target: LOG_TARGET,
            "👻 Slow Clap #{:?} started for network #{:?}",
            block_number,
            network_in_use.0,
        );

        let network_id_encoded = network_in_use.0.encode();
        let network_lock_key = Self::create_storage_key(b"network-lock-", &network_id_encoded);
        let lock_until = rt_offchain::Duration::from_millis(MIN_LOCK_GUARD_PERIOD);
        let mut network_lock = StorageLock::<Time>::with_deadline(&network_lock_key, lock_until);

        let _lock_guard = network_lock
            .try_lock()
            .map_err(|_| OffchainErr::OffchainTimeoutPeriod(network_in_use.0))?;

        Self::do_evm_claps_or_save_block(
            session_index,
            block_number,
            network_in_use.0,
            &network_in_use.1,
        )
    }

    fn do_evm_claps_or_save_block(
        session_index: SessionIndex,
        block_number: BlockNumberFor<T>,
        network_id: NetworkIdOf<T>,
        network_data: &NetworkData,
    ) -> OffchainResult<T, ()> {
        let network_id_encoded = network_id.encode();

        let block_number_key = Self::create_storage_key(b"block-", &network_id_encoded);
        let block_distance_key = Self::create_storage_key(b"block-distance-", &network_id_encoded);
        let endpoint_key = Self::create_storage_key(b"endpoint-", &network_id_encoded);

        let max_block_distance = Self::read_persistent_offchain_storage(
            &block_distance_key,
            network_data.block_distance,
        );
        let stored_endpoints = Self::read_persistent_offchain_storage(
            &endpoint_key,
            network_data.default_endpoints.clone(),
        );

        let rpc_endpoints = if !stored_endpoints.is_empty() {
            &stored_endpoints
        } else {
            &network_data.default_endpoints
        };

        if rpc_endpoints.len() == 0 {
            return Err(OffchainErr::NoEndpointAvailable(network_id));
        }

        let (from_block, to_block): (ExternalBlockNumber, ExternalBlockNumber) =
            StorageValueRef::persistent(&block_number_key)
                .get()
                .map_err(|_| OffchainErr::StorageRetrievalError(network_id))?
                .unwrap_or_default();

        match network_data.network_type {
            NetworkType::Evm => {
                let request_body = if from_block < to_block.saturating_sub(1) {
                    Self::prepare_evm_request_body_for_latest_transfers(
                        from_block,
                        to_block.saturating_sub(1),
                        network_data,
                    )
                } else {
                    Self::prepare_evm_request_body_for_latest_block(network_data)
                };

                let pending_requests =
                    Self::prepare_pending_evm_requests(&rpc_endpoints, &request_body)?;
                let parsed_evm_responses = Self::fetch_multiple_evm_from_remote(pending_requests)
                    .iter()
                    .filter_map(|response_bytes| {
                        let parsed_evm_response = Self::parse_evm_response(response_bytes).ok()?;
                        Some(parsed_evm_response)
                    })
                    .collect::<Vec<EvmResponseType>>();

                Self::check_evm_responses_correctness(&parsed_evm_responses)?;
                let parsed_evm_response =
                    Self::get_balanced_evm_response(&parsed_evm_responses, max_block_distance)?;

                let new_block_range = match parsed_evm_response {
                    EvmResponseType::BlockNumber(new_evm_block) if from_block.le(&to_block) => {
                        // stay in the range of block distance
                        let estimated_block =
                            new_evm_block.saturating_sub(network_data.finality_delay);
                        match from_block {
                            0 => (estimated_block, estimated_block),
                            start_block => {
                                let block_deviation =
                                    start_block.saturating_add(max_block_distance);
                                (start_block, estimated_block.min(block_deviation))
                            }
                        }
                    }
                    _ => (to_block, to_block),
                };

                StorageValueRef::persistent(&block_number_key).set(&new_block_range);

                log::info!(
                    target: LOG_TARGET,
                    "👻 Offchain worker #{:?} stored block #{:?} for network {:?}",
                    block_number,
                    new_block_range.0,
                    network_id,
                );

                if !sp_io::offchain::is_validator() {
                    return Err(OffchainErr::NotValidator);
                }

                for (authority_index, authority_key) in Self::local_authorities(&session_index) {
                    parsed_evm_response.sign_and_submit::<T>(
                        block_number,
                        new_block_range.0,
                        authority_index,
                        authority_key,
                        session_index,
                        network_id,
                    );
                }

                Ok(())
            }
            NetworkType::Utxo => Err(OffchainErr::UtxoNotImplemented(network_id).into()),
            _ => Err(OffchainErr::UnknownNetworkType(network_id).into()),
        }
    }

    fn local_authorities(
        session_index: &SessionIndex,
    ) -> impl Iterator<Item = (AuthIndex, T::AuthorityId)> {
        let authorities = Authorities::<T>::get(session_index);
        let mut local_authorities = T::AuthorityId::all();
        local_authorities.sort();

        authorities
            .into_iter()
            .enumerate()
            .filter_map(move |(index, authority)| {
                local_authorities
                    .binary_search(&authority)
                    .ok()
                    .map(|location| (index as u32, local_authorities[location].clone()))
            })
    }

    fn prepare_pending_evm_requests(
        rpc_endpoints: &Vec<Vec<u8>>,
        request_body: &[u8],
    ) -> OffchainResult<T, Vec<PendingRequest>> {
        let mut pending_requests = Vec::new();
        let request_body_str = core::str::from_utf8(request_body)
            .map_err(|_| OffchainErr::UnparsableRequestBody(request_body.to_vec()))?;

        let deadline = sp_io::offchain::timestamp()
            .add(rt_offchain::Duration::from_millis(FETCH_TIMEOUT_PERIOD));

        for rpc_endpoint in rpc_endpoints.iter() {
            let rpc_endpoint_str = match core::str::from_utf8(rpc_endpoint) {
                Ok(rpc_endpoint_str) => rpc_endpoint_str,
                Err(_) => {
                    log::info!(
                        target: LOG_TARGET,
                        "👻 Could not get valid utf8 rpc endpoint from bytes; skipped \"{:?}\"",
                        rpc_endpoint,
                    );
                    continue;
                }
            };

            match rt_offchain::http::Request::post(&rpc_endpoint_str, vec![request_body_str])
                .add_header("Accept", "application/json")
                .add_header("Content-Type", "application/json")
                .add_header("User-Agent", "curl/8.9.0") // mimic the curl
                .deadline(deadline)
                .send()
            {
                Ok(pending) => pending_requests.push(pending),
                Err(_) => {
                    log::info!(
                        target: LOG_TARGET,
                        "👻 Request skipped: failed to send request to \"{}\"",
                        rpc_endpoint_str,
                    )
                }
            }
        }

        if pending_requests.len() == 0 {
            return Err(OffchainErr::NoRequestsSent);
        }

        log::info!(
            target: LOG_TARGET,
            "👻 Requests sent {} out of {} possible RPC endpoints",
            pending_requests.len(),
            rpc_endpoints.len(),
        );

        Ok(pending_requests)
    }

    fn get_balanced_evm_response(
        parsed_evm_responses: &Vec<EvmResponseType>,
        max_block_distance: ExternalBlockNumber,
    ) -> OffchainResult<T, EvmResponseType> {
        let first_evm_response = parsed_evm_responses
            .first()
            .ok_or(OffchainErr::EmptyResponses)?;

        let result = match first_evm_response {
            EvmResponseType::BlockNumber(_) => {
                let mut block_numbers = parsed_evm_responses
                    .iter()
                    .enumerate()
                    .filter_map(|(index, response)| match response {
                        EvmResponseType::BlockNumber(block) => Some((index as u32, *block)),
                        EvmResponseType::TransactionLogs(_) => None,
                    })
                    .collect::<Vec<_>>();

                let block_numbers_len = block_numbers.len() as u32;
                let median_value = Self::calculate_median_value(&mut block_numbers);

                // there is no intention to make it resistent to *ANY* type of issues around RPC
                // ednpoint. here we are trying to protect against `parsed_evm_responses.len() ==
                // 2` while one of it is malicious in order not to fall in the block backoff later
                let mid_idx =
                    block_numbers.partition_point(|&block_meta| block_meta.1 < median_value);
                let mid_block = block_numbers.get(mid_idx).map(|(_, block)| *block).ok_or(
                    OffchainErr::MissingBlockNumber(mid_idx as u32, block_numbers_len),
                )?;
                let prev_block = if block_numbers_len % 2 == 0 {
                    let prev_idx = mid_idx.saturating_sub(1);
                    let prev_block = block_numbers.get(prev_idx).map(|(_, block)| *block).ok_or(
                        OffchainErr::MissingBlockNumber(prev_idx as u32, block_numbers_len),
                    )?;
                    prev_block
                } else {
                    mid_block
                };

                if mid_block.abs_diff(prev_block) > max_block_distance {
                    return Err(OffchainErr::ContradictoryBlockMedian(
                        mid_block,
                        prev_block,
                        max_block_distance,
                    ));
                }

                EvmResponseType::BlockNumber(median_value)
            }
            EvmResponseType::TransactionLogs(_) => {
                let mut count_btree_map = BTreeMap::new();
                parsed_evm_responses.iter().for_each(|response| {
                    if let EvmResponseType::TransactionLogs(logs) = response {
                        let mut inner_logs = logs.clone();
                        inner_logs.sort_by_key(|l| l.block_number);
                        *count_btree_map.entry(inner_logs).or_insert(0) += 1;
                    }
                });

                let (best_logs_ref, max_count) = count_btree_map
                    .iter()
                    .max_by_key(|&(_, count)| count)
                    .map(|(logs, count)| (logs, count))
                    .ok_or(OffchainErr::EmptyResponses)?;

                let best_logs_count = count_btree_map
                    .values()
                    .filter(|&&count| count == *max_count)
                    .count();

                if best_logs_count > 1 {
                    return Err(OffchainErr::ContradictoryTransactionLogs(
                        *max_count,
                        best_logs_count as u32,
                    ));
                }

                EvmResponseType::TransactionLogs(best_logs_ref.clone())
            }
        };

        Ok(result)
    }

    fn check_evm_responses_correctness(
        parsed_evm_responses: &Vec<EvmResponseType>,
    ) -> OffchainResult<T, ()> {
        let first_evm_response = parsed_evm_responses
            .first()
            .ok_or(OffchainErr::EmptyResponses)?;
        let first_evm_response_type = core::mem::discriminant(first_evm_response);

        if !parsed_evm_responses.iter().all(|parsed_evm_response| {
            core::mem::discriminant(parsed_evm_response) == first_evm_response_type
        }) {
            return Err(OffchainErr::DifferentEvmResponseTypes);
        }

        Ok(())
    }

    fn fetch_multiple_evm_from_remote(pending_requests: Vec<PendingRequest>) -> Vec<Vec<u8>> {
        let mut requests_failed = 0;
        let mut responses_failed = 0;
        let mut responses_non_200 = 0;

        let pending_requests_len = pending_requests.len();
        let deadline = sp_io::offchain::timestamp()
            .add(rt_offchain::Duration::from_millis(FETCH_TIMEOUT_PERIOD));

        let parsed_evm_responses = PendingRequest::try_wait_all(pending_requests, Some(deadline))
            .into_iter()
            .filter_map(|pending_request| {
                // handle request-level errors (transport/connection failures)
                let request_result = match pending_request {
                    Ok(request) => request,
                    Err(_) => {
                        requests_failed += 1;
                        return None;
                    }
                };

                // handle response-level errors (HTTP/protocol errors)
                let response = match request_result {
                    Ok(response) => response,
                    Err(_) => {
                        responses_failed += 1;
                        return None;
                    }
                };

                if response.code != 200 {
                    responses_non_200 += 1;
                    return None;
                }

                Some(response.body().collect::<Vec<u8>>())
            })
            .collect::<Vec<Vec<u8>>>();

        log::info!(
            target: LOG_TARGET,
            "👻 Fetched {} of {}: {} failed request, {} failed responses, {} non 200 code",
            parsed_evm_responses.len(),
            pending_requests_len,
            requests_failed,
            responses_failed,
            responses_non_200,
        );

        parsed_evm_responses
    }

    fn prepare_evm_request_body_for_latest_block(network_data: &NetworkData) -> Vec<u8> {
        match network_data.network_type {
            NetworkType::Evm => {
                b"{\"id\":0,\"jsonrpc\":\"2.0\",\"method\":\"eth_blockNumber\"}".to_vec()
            }
            _ => Default::default(),
        }
    }

    fn prepare_evm_request_body_for_latest_transfers(
        from_block: ExternalBlockNumber,
        to_block: ExternalBlockNumber,
        network_data: &NetworkData,
    ) -> Vec<u8> {
        match network_data.network_type {
            NetworkType::Evm => {
                let mut body =
                    b"{\"id\":0,\"jsonrpc\":\"2.0\",\"method\":\"eth_getLogs\",\"params\":[{"
                        .to_vec();
                body.extend(b"\"fromBlock\":\"".to_vec());
                body.extend(Self::block_number_to_hexadecimal_bytes(from_block));
                body.extend(b"\",\"toBlock\":\"".to_vec());
                body.extend(Self::block_number_to_hexadecimal_bytes(to_block));
                body.extend(b"\",\"address\":\"".to_vec());
                body.extend(network_data.gatekeeper.to_vec());
                body.extend(b"\",\"topics\":[\"".to_vec());
                body.extend(network_data.topic_name.to_vec());
                body.extend(b"\"]}]}".to_vec());
                body
            }
            _ => Default::default(),
        }
    }

    fn parse_evm_response(response_bytes: &[u8]) -> OffchainResult<T, EvmResponseType> {
        let response_str = sp_std::str::from_utf8(&response_bytes)
            .map_err(|_| OffchainErr::HttpBytesParsingError)?;

        let response_result: EvmResponse =
            serde_json::from_str(&response_str).map_err(|_| OffchainErr::HttpJsonParsingError)?;

        if response_result.error.is_some() {
            return Err(OffchainErr::ErrorInEvmResponse);
        }

        Ok(response_result
            .result
            .ok_or(OffchainErr::ErrorInEvmResponse)?)
    }

    fn initialize_authorities(authorities: Vec<T::AuthorityId>) {
        let session_index = T::ValidatorSet::session_index();
        assert!(
            Authorities::<T>::get(&session_index).is_empty(),
            "Authorities are already initilized!"
        );

        let authorities_len = authorities.len();
        let bounded_authorities = WeakBoundedVec::<_, T::MaxAuthorities>::force_from(
            authorities,
            Some("slow claps reached maximum number of authorities"),
        );

        let validators = T::ValidatorSet::validators();
        let bounded_validators = WeakBoundedVec::<_, T::MaxAuthorities>::force_from(
            validators,
            Some("slow claps reached maximum number of validators"),
        );

        if let Some(target_session_index) = session_index.checked_sub(T::HistoryDepth::get()) {
            Self::clear_history(&target_session_index);
        }

        Validators::<T>::insert(&session_index, bounded_validators);
        Authorities::<T>::set(&session_index, bounded_authorities);

        let mut disabled_bitmap = BitMap::new(authorities_len as AuthIndex);
        for disabled_index in T::DisabledValidators::disabled_validators() {
            disabled_bitmap.set(disabled_index);
        }
        DisabledAuthorityIndexes::<T>::insert(session_index, disabled_bitmap);
    }

    fn clear_history(target_session_index: &SessionIndex) {
        Authorities::<T>::remove(target_session_index);
        Validators::<T>::remove(target_session_index);
        DisabledAuthorityIndexes::<T>::remove(target_session_index);

        let cursor = ApplauseDetails::<T>::clear_prefix(target_session_index, u32::MAX, None);
        debug_assert!(cursor.maybe_cursor.is_none());
    }

    fn calculate_median_value<InnerValue>(values: &mut Vec<(AuthIndex, InnerValue)>) -> InnerValue
    where
        InnerValue: AtLeast32BitUnsigned + Copy,
    {
        values.sort_by_key(|data| data.1);

        let length = values.len();
        if length % 2 == 0 {
            let mid_left = values[length / 2 - 1].1;
            let mid_right = values[length / 2].1;
            (mid_left + mid_right) / 2u32.into()
        } else {
            values[length / 2].1
        }
    }

    fn capture_deviation_in_commitments<InnerValue>(
        disabled_bitmap: &BitMap,
        mut values: &mut Vec<(AuthIndex, InnerValue)>,
        validators_len: u32,
        max_deviation: InnerValue,
    ) -> BitMap
    where
        InnerValue: AtLeast32BitUnsigned + Copy,
    {
        let mut delayed_bitmap = BitMap::new(validators_len);
        let median_value = Self::calculate_median_value(&mut values);

        values.iter().for_each(|(authority_index, value)| {
            let abs_diff = if *value > median_value {
                *value - median_value
            } else {
                median_value - *value
            };

            if !disabled_bitmap.exists(authority_index) && abs_diff > max_deviation {
                delayed_bitmap.set(*authority_index);
            }
        });

        delayed_bitmap
    }

    fn get_cumulative_missing_clapped_amount(
        disabled_bitmap: &BitMap,
        session_index: &SessionIndex,
        missed_validators: &mut BitMap,
    ) -> Perbill {
        let total_exposure = TotalExposure::<T>::get();

        let bucket_size = BitMap::size_of_bucket();
        let mut missed_percent = Perbill::default();

        for applause_details in ApplauseDetails::<T>::iter_prefix_values(session_index) {
            let max_value = applause_details.authorities.max_value();
            let clapped_amount = applause_details.clapped_amount;
            let is_finalized = applause_details.finalized as Bucket;

            for (bucket_id, bucket) in applause_details.authorities.iter_buckets() {
                let bits_in_bucket = 1 << bucket_size;
                let bucket_shift = bucket_id << bucket_size;

                let disabled_bucket = disabled_bitmap.get_bucket(bucket_id);
                let missed_authorities = bucket ^ (Bucket::MAX * is_finalized);
                let missed_authorities = missed_authorities & !disabled_bucket;

                for bit_position in 0..bits_in_bucket {
                    if (missed_authorities >> bit_position) & 1 == 0 {
                        continue;
                    }

                    let position = bucket_shift + bit_position;
                    if position > max_value {
                        continue;
                    }

                    missed_validators.set(position.into());
                    missed_percent = missed_percent.saturating_add(Perbill::from_rational(
                        total_exposure.saturating_sub(clapped_amount),
                        total_exposure,
                    ));
                }
            }
        }

        missed_percent
    }

    fn try_offend_validators(
        session_index: &SessionIndex,
        validators: &WeakBoundedVec<ValidatorId<T>, T::MaxAuthorities>,
        offence_bitmap: BitMap,
        disabled_bitmap: BitMap,
        offence_type: OffenceType,
    ) -> Weight {
        let mut weight = T::DbWeight::get().reads_writes(1, 1);
        let validator_set_count = validators.len();

        let mut offenders = Vec::with_capacity(offence_bitmap.count_ones() as usize);
        let mut reporter_indexes = Vec::with_capacity(validator_set_count);

        let disabled_or_offence_bitmap = disabled_bitmap.bitor(offence_bitmap.clone());
        let validator_set_count = validator_set_count as u32;

        for (index, id) in validators.iter().enumerate() {
            let authority_index = index as AuthIndex;

            if offence_bitmap.exists(&authority_index) {
                weight.saturating_accrue(T::DbWeight::get().reads(1));
                if let Some(full_id) = <T::ValidatorSet as ValidatorSetWithIdentification<
                    T::AccountId,
                >>::IdentificationOf::convert(id.clone())
                {
                    offenders.push((id.clone(), full_id));
                }
            } else if !disabled_or_offence_bitmap.exists(&authority_index) {
                reporter_indexes.push(index);
            }
        }

        let offenders_len = offenders.len() as u32;
        let not_enough_validators_left = validator_set_count
            .saturating_sub(disabled_or_offence_bitmap.count_ones())
            .lt(&T::MinAuthoritiesNumber::get());

        if not_enough_validators_left && offenders_len > 0 {
            Self::deposit_event(Event::<T>::BlackSwan);
            return weight;
        }

        if offenders_len == 0 {
            let equilibrium_event = match offence_type {
                OffenceType::CommitmentOffence => Event::<T>::AuthoritiesCommitmentEquilibrium,
                OffenceType::ThrottlingOffence(_) => Event::<T>::AuthoritiesApplauseEquilibrium,
            };
            Self::deposit_event(equilibrium_event);
            return weight;
        }

        weight.saturating_accrue(T::DbWeight::get().reads(1));
        let reporters = T::ExposureListener::get_accounts_by_indexes(reporter_indexes.into_iter());

        let offence_event = match offence_type {
            OffenceType::CommitmentOffence => Event::<T>::SomeAuthoritiesDelayed {
                delayed: offenders.clone(),
            },
            OffenceType::ThrottlingOffence(_) => Event::<T>::SomeAuthoritiesTrottling {
                throttling: offenders.clone(),
            },
        };

        Self::deposit_event(offence_event);

        let offence = SlowClapOffence {
            validator_set_count,
            session_index: *session_index,
            offenders,
            offence_type,
        };

        if let Err(e) = T::ReportUnresponsiveness::report_offence(reporters, offence) {
            sp_runtime::print(e);
        }

        let extra_weight = T::WeightInfo::try_offend_validators(offenders_len);

        weight.saturating_add(extra_weight)
    }
}

impl<T: Config> sp_runtime::BoundToRuntimeAppPublic for Pallet<T> {
    type Public = T::AuthorityId;
}

impl<T: Config> BlockNumberProvider for Pallet<T> {
    type BlockNumber = BlockNumberFor<T>;
    fn current_block_number() -> Self::BlockNumber {
        T::BlockNumberProvider::current_block_number()
    }
}

impl<T: Config> OneSessionHandler<T::AccountId> for Pallet<T> {
    type Key = T::AuthorityId;

    fn on_genesis_session<'a, I: 'a>(validators: I)
    where
        I: Iterator<Item = (&'a T::AccountId, T::AuthorityId)>,
    {
        let authorities = validators.map(|x| x.1).collect::<Vec<_>>();
        Self::initialize_authorities(authorities);
    }

    fn on_new_session<'a, I: 'a>(_changed: bool, validators: I, _queued_validators: I)
    where
        I: Iterator<Item = (&'a T::AccountId, T::AuthorityId)>,
    {
        for network_id in T::NetworkDataHandler::iter_indexes() {
            BlockCommitments::<T>::remove(&network_id);
        }

        let total_exposure = T::ExposureListener::get_total_exposure();
        TotalExposure::<T>::set(total_exposure);

        let authorities = validators.map(|x| x.1).collect::<Vec<_>>();
        Self::initialize_authorities(authorities);
    }

    fn on_before_session_ending() {
        let session_index = T::ValidatorSet::session_index().saturating_sub(1);

        let validators = Validators::<T>::get(&session_index);
        let validators_len = validators.len() as u32;
        let disabled_bitmap = DisabledAuthorityIndexes::<T>::get(&session_index)
            .unwrap_or(BitMap::new(validators_len));

        let mut offence_bitmap = BitMap::new(validators_len);

        let missed_percent = Self::get_cumulative_missing_clapped_amount(
            &disabled_bitmap,
            &session_index,
            &mut offence_bitmap,
        );

        let offence_type = OffenceType::ThrottlingOffence(missed_percent);
        Self::try_offend_validators(
            &session_index,
            &validators,
            offence_bitmap,
            disabled_bitmap,
            offence_type,
        );
    }

    fn on_disabled(validator_index: u32) {
        let session_index = T::ValidatorSet::session_index();
        let length = Authorities::<T>::get(&session_index).len();
        DisabledAuthorityIndexes::<T>::mutate(&session_index, |maybe_bitmap| {
            if let Some(bitmap) = maybe_bitmap {
                bitmap.set(validator_index);
            } else {
                let mut new_bitmap = BitMap::new(length as u32);
                new_bitmap.set(validator_index);
                *maybe_bitmap = Some(new_bitmap);
            }
        });
    }
}

#[derive(RuntimeDebug, TypeInfo)]
#[cfg_attr(feature = "std", derive(Clone, PartialEq, Eq))]
pub enum OffenceType {
    CommitmentOffence,
    ThrottlingOffence(Perbill),
}

#[derive(RuntimeDebug, TypeInfo)]
#[cfg_attr(feature = "std", derive(Clone, PartialEq, Eq))]
pub struct SlowClapOffence<Offender> {
    session_index: SessionIndex,
    validator_set_count: u32,
    offenders: Vec<Offender>,
    offence_type: OffenceType,
}

impl<Offender: Clone> Offence<Offender> for SlowClapOffence<Offender> {
    const ID: Kind = *b"slow-clap:throtl";
    type TimeSlot = SessionIndex;

    fn offenders(&self) -> Vec<Offender> {
        self.offenders.clone()
    }

    fn session_index(&self) -> SessionIndex {
        self.session_index
    }

    fn validator_set_count(&self) -> u32 {
        self.validator_set_count
    }

    fn time_slot(&self) -> Self::TimeSlot {
        self.session_index
    }

    fn slash_fraction(&self, offenders_count: u32) -> Perbill {
        match self.offence_type {
            OffenceType::ThrottlingOffence(missed_percent) => {
                missed_percent.saturating_mul(Perbill::from_rational(1, offenders_count))
            }
            OffenceType::CommitmentOffence => offenders_count
                .checked_sub(Perbill::from_percent(9).mul_ceil(self.validator_set_count))
                .map(|threshold| {
                    Perbill::from_rational(threshold.saturating_mul(4), self.validator_set_count)
                        .square()
                })
                .unwrap_or_default(),
        }
    }
}