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

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

use sp_std::prelude::*;
use sp_core::H256;
use sp_runtime::{
    offchain::{
        self as rt_offchain,
        http::{PendingRequest, Request},
        storage::StorageValueRef,
        storage_lock::{StorageLock, Time},
    },
    traits::{BlockNumberProvider, Saturating, UniqueSaturatedInto},
    Perbill, RuntimeAppPublic,
};

use ghost_helpers::{
    bounded_bitmap::{validate_bitmap_sizes, BoundedBitmap},
    get_byzantium_threshold,
    networks::{NetworkData, NetworkType},
};
use ghost_traits::{
    bounded_bitmap::{BoundedBitmapReader, BoundedBitmapWriter},
    networks::{
        NetworkDataBasicHandler, NetworkDataInspectHandler, NetworkDataMutateHandler,
        NetworkRpcResolver,
    },
};

mod errors;
mod impls;
mod types;
mod weights;

#[cfg(test)]
mod tests;

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

#[cfg(any(test, feature = "runtime-benchmarks"))]
mod mock;

use crate::errors::*;
use crate::types::*;
pub use crate::weights::WeightInfo;

pub use pallet::*;

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

    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-weaver";
const DB_PREFIX: &[u8] = b"ghost-weaver::";

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

type AuthIndex = ghost_helpers::AuthIndexU16;
type BitmapChunk = ghost_helpers::BitmapChunkU32;

type RequestId = u64;
type ThreadId = u64;
type WeavingSession = u64;
type ExternalBlockNumber = u64;

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

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

type WeavingResult<T, O> = Result<O, WeavingError<NetworkIdOf<T>>>;

type WeaversBitmap<T> = BoundedBitmap<BitmapChunk, <T as Config>::MaxAuthoritiesChunks>;

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

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

    #[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 Currency: Currency<Self::AccountId>;

        type NetworkDataHandler: NetworkDataBasicHandler
            + NetworkDataInspectHandler<NetworkData>
            + NetworkDataMutateHandler<NetworkData, BalanceOf<Self>>;

        type BlockNumberProvider: BlockNumberProvider<BlockNumber = BlockNumberFor<Self>>;

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

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

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

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

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

        type WeightInfo: WeightInfo;
    }

    #[pallet::event]
    #[pallet::generate_deposit(pub(super) fn deposit_event)]
    pub enum Event<T: Config> {
        BlockAttestated {
            network_id: NetworkIdOf<T>,
            authority_index: AuthIndex,
            external_block: ExternalBlockNumber,
        },
        HashAttestated {
            next_session: WeavingSession,
            authority_index: AuthIndex,
            network_id: NetworkIdOf<T>,
            root_hash: H256,
        },
        ThreadPulled {
            pulled_thread_key: H256,
            network_id: NetworkIdOf<T>,
            receiver: T::AccountId,
            amount: BalanceOf<T>,
        },
    }

    #[pallet::error]
    pub enum Error<T> {
        InvalidParticipantAttestation,
        NonExistentNetworkId,
        TooManyAttestations,
        TimeWentBackwards,
        WeavingIsActive,
        WeavingIsInactive,
        WeaverAlreadyExists,

        AuthorityPartOfTrapestry,
        ThreadAlreadyPulled,
        InvalidMerkleProof,
        InvalidReceiverAddress,
        CouldNotAccumulateIncomingImbalance,
        CouldNotIncreaseGatekeeperAmount,
        CouldNotAccumulateCommission,
    }

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

    #[pallet::storage]
    #[pallet::getter(fn pulled_threads)]
    pub(super) type PulledThreads<T: Config> = StorageMap<
        _,
        Twox64Concat, H256,
        (),
        OptionQuery,
    >;

    #[pallet::storage]
    #[pallet::getter(fn loom_states)]
    pub(super) type LoomStates<T: Config> = StorageDoubleMap<
        _,
        Twox64Concat, NetworkIdOf<T>,
        Twox64Concat, WeavingSession,
        H256,
        ValueQuery,
    >;

    #[pallet::storage]
    #[pallet::getter(fn tapestry_strands)]
    pub(super) type TapestryStrands<T: Config> = StorageNMap<
        _,
        (
            NMapKey<Twox64Concat, NetworkIdOf<T>>,
            NMapKey<Twox64Concat, WeavingSession>,
            NMapKey<Twox64Concat, WeavingSession>,
            NMapKey<Twox64Concat, H256>,
        ),
        WeaversBitmap<T>,
        ValueQuery,
    >;

    #[pallet::storage]
    #[pallet::getter(fn network_attestations)]
    pub(super) type NetworkAttestations<T: Config> = StorageMap<
        _,
        Twox64Concat, NetworkIdOf<T>,
        BoundedVec<Option<ExternalBlockNumber>, T::MaxAuthorities>,
        ValueQuery,
    >;

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

    #[pallet::storage]
    #[pallet::getter(fn weaving_states)]
    pub(super) type WeavingStates<T: Config> = StorageMap<
        _,
        Twox64Concat, NetworkIdOf<T>,
        WeavingState<AuthIndex>,
        ValueQuery,
    >;

    #[pallet::storage]
    #[pallet::getter(fn tapestry_drafts)]
    pub(super) type TapestryDrafts<T: Config> = StorageMap<
        _,
        Twox64Concat, NetworkIdOf<T>,
        TapestryDraft<BlockNumberFor<T>>,
        OptionQuery,
    >;

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

    #[pallet::storage]
    #[pallet::getter(fn disabled_authorities)]
    pub(super) type DisabledAuthorities<T: Config> = StorageValue<_, WeaversBitmap<T>, ValueQuery>;

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

    #[pallet::genesis_build]
    impl<T: Config> BuildGenesisConfig for GenesisConfig<T> {
        fn build(&self) {
            validate_bitmap_sizes::<WeaversBitmap<T>>(T::MaxAuthorities::get());

            self.loom_states.iter().for_each(|data| {
                if !T::NetworkDataHandler::contains_key(&data.0) {
                    log::info!(
                        target: LOG_TARGET,
                        "🕸️ ATTENTION! Network #{:?} not registered yet, could not insert root {:?} for weaving session #{:?}!",
                        data.0,
                        data.2,
                        data.1,
                    );
                    return;
                }
                LoomStates::<T>::insert(&data.0, data.1, data.2);
            });

            if !self.authorities.is_empty() {
                let authorities_vec = self
                    .authorities
                    .iter()
                    .map(|(account_id, authority)| (account_id, authority.clone()));

                Pallet::<T>::initialize_authorities(authorities_vec);
            }
        }
    }

    #[pallet::call]
    impl<T: Config> Pallet<T> {
        #[pallet::call_index(0)]
        #[pallet::weight((
            T::WeightInfo::extend_warp(),
            DispatchClass::Normal,
            Pays::No,
        ))]
        pub fn extend_warp(
            origin: OriginFor<T>,
            block_attestation: BlockAttestation<BlockNumberFor<T>, NetworkIdOf<T>, AuthIndex>,
            signature: <T::AuthorityId as RuntimeAppPublic>::Signature,
        ) -> DispatchResult {
            ensure_none(origin)?;

            let (authority_index, network_id) =
                Self::validate_attestation_signature(&block_attestation, &signature)
                    .map_err(|_| Error::<T>::InvalidParticipantAttestation)?;

            ensure!(
                !TapestryDrafts::<T>::contains_key(&network_id),
                Error::<T>::WeavingIsActive,
            );

            let external_block = block_attestation.external_block;

            let current_authorities_count = Authorities::<T>::decode_len().
                unwrap_or_default();

            NetworkAttestations::<T>::try_mutate(
                &network_id,
                |attestations| -> DispatchResult {
                    if attestations.is_empty() {
                        attestations.bounded_resize(current_authorities_count, None);
                    }

                    ensure!(
                        (authority_index as usize) < attestations.len(),
                        Error::<T>::TooManyAttestations,
                    );

                    attestations[authority_index as usize] = Some(external_block);

                    Ok(())
                })?;

            Self::deposit_event(Event::<T>::BlockAttestated {
                network_id,
                authority_index,
                external_block,
            });

            Ok(())
        }

        #[pallet::call_index(1)]
        #[pallet::weight((
            T::WeightInfo::weave_weft(),
            DispatchClass::Normal,
            Pays::No,
        ))]
        pub fn weave_weft(
            origin: OriginFor<T>,
            hash_attestation: HashAttestation<BlockNumberFor<T>, NetworkIdOf<T>, AuthIndex>,
            signature: <T::AuthorityId as RuntimeAppPublic>::Signature,
        ) -> DispatchResult {
            ensure_none(origin)?;

            let (authority_index, network_id) =
                Self::validate_attestation_signature(&hash_attestation, &signature)
                    .map_err(|_| Error::<T>::InvalidParticipantAttestation)?;

            ensure!(
                TapestryDrafts::<T>::contains_key(&network_id),
                Error::<T>::WeavingIsInactive,
            );

            let root_hash = hash_attestation.root_hash;
            let next_session = hash_attestation.next_session;
            let weaving_session = hash_attestation.weaving_session();

            let weavers_count = TapestryStrands::<T>::try_mutate(
                &(network_id, weaving_session, next_session, root_hash),
                |tapestry_strand| -> Result<AuthIndex, DispatchError> {
                    ensure!(
                        !tapestry_strand.contains(authority_index),
                        Error::<T>::AuthorityPartOfTrapestry,
                    );

                    tapestry_strand.insert(authority_index);
                    Ok(tapestry_strand.count_ones::<AuthIndex>())
                },
            )?;

            WeavingStates::<T>::mutate(&network_id, |weaving_state| {
                if weaving_state.count < weavers_count {
                    *weaving_state = WeavingState::new(root_hash, next_session, weavers_count);
                }
            });

            Self::deposit_event(Event::<T>::HashAttestated {
                authority_index,
                next_session,
                network_id,
                root_hash,
            });

            Ok(())
        }

        #[pallet::call_index(2)]
        #[pallet::weight((
            match &thread_proof {
                ThreadProof::UtxoThreadProof(_) => T::WeightInfo::pull_utxo_thread(),
                ThreadProof::EvmThreadProof(_) => T::WeightInfo::pull_evm_thread(),
            },
            DispatchClass::Normal,
            Pays::No,
        ))]
        pub fn pull_thread(
            origin: OriginFor<T>,
            network_id: NetworkIdOf<T>,
            session: WeavingSession,
            thread_proof: ThreadProof<BalanceOf<T>>,
        ) -> DispatchResult {
            ensure_none(origin)?;

            let network_data = T::NetworkDataHandler::get(&network_id)
                .ok_or(Error::<T>::NonExistentNetworkId)?;

            let root_hash = LoomStates::<T>::get(&network_id, &session);
            let pulled_thread_key = thread_proof.get_unique_key(session);

            ensure!(
                !PulledThreads::<T>::contains_key(&pulled_thread_key),
                Error::<T>::ThreadAlreadyPulled,
            );

            let receiver_account = thread_proof
                .verify_proof(root_hash, &network_data.gatekeeper)
                .ok_or(Error::<T>::InvalidMerkleProof)?;

            let amount = thread_proof.amount().clone();
            let receiver_bytes: &[u8; 32] = receiver_account.as_ref();
            let receiver = T::AccountId::decode(&mut &receiver_bytes[..])
                .map_err(|_| Error::<T>::InvalidReceiverAddress)?;

            let commission = Perbill::from_parts(network_data.incoming_fee).mul_ceil(amount);
            let pure_amount = amount.saturating_sub(commission);

            let _ = T::NetworkDataHandler::accumulate_incoming_imbalance(&pure_amount)
                .map_err(|_| Error::<T>::CouldNotAccumulateIncomingImbalance)
                .and_then(|_| {
                    T::NetworkDataHandler::increase_gatekeeper_amount(&network_id, &amount)
                        .map_err(|_| Error::<T>::CouldNotIncreaseGatekeeperAmount)
                })
                .and_then(|_| {
                    T::NetworkDataHandler::accumulate_commission(&commission)
                        .map_err(|_| Error::<T>::CouldNotAccumulateCommission)
                })?;

            let _ = T::Currency::deposit_creating(&receiver, pure_amount);
            PulledThreads::<T>::insert(pulled_thread_key, ());

            Self::deposit_event(Event::<T>::ThreadPulled {
                pulled_thread_key,
                network_id,
                receiver,
                amount,
            });

            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 converted_block: usize = current_block.unique_saturated_into();

            let (network_id, data) = match T::NetworkDataHandler::network_for_block(converted_block)
            {
                Some((network_id, data)) => (network_id, data),
                None => return weight,
            };

            weight.saturating_accrue(T::DbWeight::get().reads_writes(2, 1));
            let len = Authorities::<T>::decode_len().unwrap_or_default();
            let weaving_threshold = get_byzantium_threshold(len);

            let maybe_tapestry_drafts = TapestryDrafts::<T>::get(&network_id);

            match maybe_tapestry_drafts {
                Some(tapestry_draft) => {
                    weight.saturating_accrue(T::DbWeight::get().reads(1));
                    let weaving_state = WeavingStates::<T>::get(&network_id);

                    let time_is_out = tapestry_draft.until <= current_block;
                    let threshold_reached = weaving_threshold <= weaving_state.count as usize;

                    if time_is_out || threshold_reached {
                        weight.saturating_accrue(T::DbWeight::get().reads_writes(2, 1));

                        let current_session = CurrentWeavingSession::<T>::get(&network_id);
                        let loom_state_changed =
                            LoomStates::<T>::mutate(&network_id, &current_session, |loom_state| {
                                if *loom_state != weaving_state.hash && threshold_reached {
                                    *loom_state = weaving_state.hash;
                                    return true;
                                }
                                false
                            });

                        if loom_state_changed && weaving_state.next_session != current_session {
                            weight.saturating_accrue(T::DbWeight::get().writes(1));

                            let next_session = current_session.saturating_add(1);
                            CurrentWeavingSession::<T>::insert(&network_id, next_session);
                        }

                        weight.saturating_accrue(T::DbWeight::get().writes(1));
                        TapestryDrafts::<T>::remove(&network_id);
                    }
                }
                None => {
                    weight.saturating_accrue(T::DbWeight::get().reads(1));
                    let next_attempt_block =
                        NextConsensusAttemptBlocks::<T>::get(&network_id);

                    if current_block < next_attempt_block { return weight; }

                    weight.saturating_accrue(T::DbWeight::get().reads(1));
                    let attestation_len = NetworkAttestations::<T>::decode_len(&network_id);

                    if attestation_len.unwrap_or(0) > len {
                        weight.saturating_accrue(T::DbWeight::get().writes(1));
                        NetworkAttestations::<T>::remove(&network_id);
                        return weight;
                    }

                    let attestation_delay_converted: BlockNumberFor<T> =
                        T::AttestationDelay::get().unique_saturated_into();

                    weight.saturating_accrue(T::DbWeight::get().writes(1));
                    let next_release_block = current_block + attestation_delay_converted;
                    NextConsensusAttemptBlocks::<T>::insert(network_id, next_release_block);

                    let mut block_numbers = NetworkAttestations::<T>::get(&network_id)
                        .iter()
                        .flatten()
                        .copied()
                        .collect();

                    if let Some(median_block) = Self::validate_block_numbers_consensus(
                        &mut block_numbers,
                        data.block_deviation,
                        weaving_threshold,
                    ) {
                        let tapestry_draft = TapestryDraftBuilder::default()
                            .with_block_delay(T::WeavingDelay::get())
                            .with_current_block(current_block)
                            .with_median_external(median_block)
                            .build();

                        TapestryDrafts::<T>::insert(&network_id, tapestry_draft);
                        weight.saturating_accrue(T::DbWeight::get().writes(1));
                    }
                }
            }

            weight
        }

        fn offchain_worker(now: BlockNumberFor<T>) {
            if sp_io::offchain::is_validator() {
                log::warn!(target: LOG_TARGET, "🕸️ Weaver started at block #{:?}", now);
                match Self::start_weaving(now) {
                    Ok(_) => log::warn!(target: LOG_TARGET, "🕸️ Weaver finished gracefully at block #{:?}", now),
                    Err(e) => log::warn!(target: LOG_TARGET, "🕸️ Weaver failed at block #{:?}: {:?}", 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::extend_warp {
                    block_attestation,
                    signature,
                } => {
                    let weaver_context =
                        Self::validate_attestation_signature(block_attestation, signature)?;

                    ValidTransaction::with_tag_prefix("WeaverBlock")
                        .priority(T::UnsignedPriority::get())
                        .and_provides(weaver_context.encode())
                        .longevity(LOCK_BLOCK_EXPIRATION)
                        .propagate(true)
                        .build()
                }
                Call::weave_weft {
                    hash_attestation,
                    signature,
                } => {
                    let weaver_context =
                        Self::validate_attestation_signature(hash_attestation, signature)?;

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

impl<T: Config> Pallet<T> {
    fn validate_attestation_signature<A>(
        attestation: &A,
        signature: &<T::AuthorityId as RuntimeAppPublic>::Signature,
    ) -> Result<(AuthIndex, NetworkIdOf<T>), InvalidTransaction>
    where
        A: AttestationMetadata<BlockNumberFor<T>, NetworkIdOf<T>, AuthIndex>,
        A: Encode,
    {
        let authority_index = attestation.authority_index();
        let attestation_block = attestation.block();
        let network_id = attestation.network_id();

        if DisabledAuthorities::<T>::get().contains(authority_index) {
            return Err(InvalidTransaction::BadSigner);
        }

        if T::BlockNumberProvider::current_block_number() <= attestation_block {
            return Err(InvalidTransaction::BadProof);
        }

        Authorities::<T>::get()
            .get(authority_index as usize)
            .ok_or(InvalidTransaction::BadProof)
            .and_then(|authority| {
                attestation
                    .using_encoded(|encoded| authority.verify(&encoded, signature))
                    .then(|| ())
                    .ok_or(InvalidTransaction::BadSigner)
            })?;

        if !T::NetworkDataHandler::contains_key(&network_id) {
            return Err(InvalidTransaction::BadProof);
        }

        Ok((authority_index, network_id))
    }

    fn start_weaving(
        block_number: BlockNumberFor<T>,
    ) -> WeavingResult<T, WeavingOk<NetworkIdOf<T>>> {
        let converted_block: usize = block_number.unique_saturated_into();
        let network_in_use = T::NetworkDataHandler::network_for_block(converted_block)
            .ok_or(WeavingError::NoStoredNetworks)?;

        let network_id_encoded = network_in_use.0.encode();

        let lock_period_key = Self::create_storage_key(b"network-period", &network_id_encoded);
        let lock_period = Self::read_offchain_storage(&lock_period_key, &MIN_LOCK_GUARD_PERIOD);
        let lock_until = rt_offchain::Duration::from_millis(lock_period);

        let network_lock_key = Self::create_storage_key(b"network-lock-", &network_id_encoded);
        let mut network_lock = StorageLock::<Time>::with_deadline(&network_lock_key, lock_until);

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

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

    fn do_weaving(
        current_block: BlockNumberFor<T>,
        network_id: NetworkIdOf<T>,
        network_data: &NetworkData,
    ) -> WeavingResult<T, WeavingOk<NetworkIdOf<T>>> {
        let network_id_encoded = network_id.encode();
        let endpoint_key = Self::create_storage_key(b"endpoint-", &network_id_encoded);
        let timeout_key = Self::create_storage_key(b"timeout-", &network_id_encoded);

        let rpc_endpoints = Self::read_offchain_storage(
            &endpoint_key,
            &network_data
                .default_endpoints
                .iter()
                .map(|endpoint| endpoint.clone().into_inner())
                .collect::<Vec<Vec<u8>>>(),
        );

        let timeout = Self::read_offchain_storage(&timeout_key, &FETCH_TIMEOUT_PERIOD);
        let deadline =
            sp_io::offchain::timestamp().add(rt_offchain::Duration::from_millis(timeout));

        if rpc_endpoints.len() == 0 {
            return Err(WeavingError::NoEndpointsAvailable(network_id));
        }

        match TapestryDrafts::<T>::get(&network_id) {
            Some(tapestry_draft) => Self::try_request_and_submit_hash(
                network_id,
                current_block,
                tapestry_draft.median,
                deadline,
                &network_data,
                &rpc_endpoints,
            ),
            None => Self::try_request_and_submit_block(
                network_id,
                current_block,
                deadline,
                &network_data,
                &rpc_endpoints,
            ),
        }
    }

    fn try_request_and_submit_hash(
        network_id: NetworkIdOf<T>,
        current_block: BlockNumberFor<T>,
        target_block: ExternalBlockNumber,
        deadline: rt_offchain::Timestamp,
        network_data: &NetworkData,
        rpc_endpoints: &Vec<Vec<u8>>,
    ) -> WeavingResult<T, WeavingOk<NetworkIdOf<T>>> {
        let weaving_session = CurrentWeavingSession::<T>::get(&network_id);

        let request_id = current_block.unique_saturated_into();

        let request_body = network_data.r#type.get_hash_request_body(
            request_id,
            target_block,
            weaving_session,
            &network_data.gatekeeper,
            &network_data.selector,
        );

        let pending_requests =
            Self::prepare_pending_requests(&rpc_endpoints, deadline, &request_body)?;

        let hashes_and_next_sessions =
            network_data
                .r#type
                .parse_hash_requests(pending_requests, deadline, request_id);

        let (first_hash, first_next_session) = hashes_and_next_sessions
            .first()
            .and_then(|(first_hash, first_next_session)| {
                hashes_and_next_sessions
                    .iter()
                    .all(|(hash, next_session)| {
                        hash == first_hash && next_session == first_next_session
                    })
                    .then(|| (*first_hash, *first_next_session))
            })
            .ok_or(WeavingError::<NetworkIdOf<T>>::ContradictoryHashes(
                network_id,
            ))?;

        let tapestry_key = (network_id, weaving_session, first_next_session, first_hash);
        let tapestry_strand = TapestryStrands::<T>::get(&tapestry_key);

        let context = AttestationContext::default()
            .with_weaving_session(weaving_session)
            .with_current_block(current_block)
            .with_network_id(network_id);

        Self::local_authorities().for_each(|(authority_index, authority_key)| {
            if tapestry_strand.contains(authority_index) {
                log::error!(
                    target: LOG_TARGET,
                    "🕸️ Weft already weaved from authority #{:?} for network #{:?}.",
                    authority_index,
                    network_id,
                );
                return;
            }

            let hash_attestation = context
                .with_authority_index(authority_index)
                .build_hash_attestation(first_hash, first_next_session);

            let Some(signature) = authority_key.sign(&hash_attestation.encode()) else {
                log::error!(
                    target: LOG_TARGET,
                    "🕸️ Signing hash attestation failed from authority #{:?} for network #{:?}.",
                    authority_index,
                    network_id,
                );
                return;
            };

            let call = Call::weave_weft { hash_attestation, signature };

            if let Err(err) =
				SubmitTransaction::<T, Call<T>>::submit_unsigned_transaction(call.into()) {
    			    log::error!(
    			        target: LOG_TARGET,
    			        "🕸️ Failed to submit hash attestation from authority #{:?} for network {:?}: {:?}.",
    			        authority_index,
    			        network_id,
    			        err,
    			    );
    			}
        });

        Ok(WeavingOk::Hash(
            target_block,
            first_hash,
            first_next_session,
            network_id,
        ))
    }

    fn try_request_and_submit_block(
        network_id: NetworkIdOf<T>,
        current_block: BlockNumberFor<T>,
        deadline: rt_offchain::Timestamp,
        network_data: &NetworkData,
        rpc_endpoints: &Vec<Vec<u8>>,
    ) -> WeavingResult<T, WeavingOk<NetworkIdOf<T>>> {
        let network_id_encoded = network_id.encode();
        let distance_key = Self::create_storage_key(b"distance-", &network_id_encoded);
        let block_deviation =
            Self::read_offchain_storage(&distance_key, &network_data.block_deviation);

        let request_id = current_block.unique_saturated_into();
        let request_body = network_data.r#type.get_block_request_body(request_id);

        let pending_requests =
            Self::prepare_pending_requests(&rpc_endpoints, deadline, &request_body)?;

        let mut block_numbers =
            network_data
                .r#type
                .parse_block_requests(pending_requests, deadline, request_id);

        let total_blocks = block_numbers.len();
        let weaving_threshold = get_byzantium_threshold(total_blocks);

        let median_block = Self::validate_block_numbers_consensus(
            &mut block_numbers,
            block_deviation,
            weaving_threshold,
        )
        .ok_or(WeavingError::ContradictoryBlocks(
            total_blocks as u32,
            block_deviation,
            network_id,
        ))?;

        let finality_delay = network_data.finality_delay as ExternalBlockNumber;
        let finalized_block = median_block.saturating_sub(finality_delay);

        let context = AttestationContext::default()
            .with_current_block(current_block)
            .with_network_id(network_id);

        Self::local_authorities().for_each(|(authority_index, authority_key)| {
            let block_attestation = context
                .with_authority_index(authority_index)
                .build_block_attestation(finalized_block);

            let Some(signature) = authority_key.sign(&block_attestation.encode()) else {
                log::error!(
                    target: LOG_TARGET,
                    "🕸️ Signing block attestation failed from authority #{:?} for network {:?}.",
                    authority_index,
                    network_id,
                );
                return;
            };

            let call = Call::extend_warp { block_attestation, signature };

            if let Err(err) =
				SubmitTransaction::<T, Call<T>>::submit_unsigned_transaction(call.into()) {
    			    log::error!(
    			        target: LOG_TARGET,
                        "🕸️ Failed to submit block attestation from authority #{:?} for network {:?}: {:?}.",
    			        authority_index,
    			        network_id,
    			        err,
    			    );
    			}
        });

        Ok(WeavingOk::Block(finalized_block, network_id))
    }

    fn prepare_pending_requests(
        rpc_endpoints: &[Vec<u8>],
        deadline: rt_offchain::Timestamp,
        request_body: &[u8],
    ) -> WeavingResult<T, Vec<PendingRequest>> {
        let default_headers: Vec<(Vec<u8>, Vec<u8>)> = Vec::new();
        let request_body_str =
            core::str::from_utf8(request_body).map_err(|_| WeavingError::UnparsableRequestBody)?;

        let pending_requests = rpc_endpoints
            .iter()
            .filter_map(|rpc_endpoint| {
                let headers = Self::read_offchain_storage(&rpc_endpoint, &default_headers);

                let rpc_endpoint_str = core::str::from_utf8(rpc_endpoint)
                    .inspect_err(|err| log::error!(target: LOG_TARGET, "🕸️ Could not convert to UTF-8: {:?}", err))
                    .ok()?;

                let body_bytes: &[u8] = request_body_str.as_bytes();
                let body_slice: &[&[u8]] = &[body_bytes];

                let mut request = Request::post(&rpc_endpoint_str, body_slice)
                    .add_header("Accept", "application/json")
                    .add_header("Content-Type", "application/json");

                for (key, header) in headers.iter() {
                    let maybe_key_str = core::str::from_utf8(key);
                    let maybe_header_str = core::str::from_utf8(header);

                    match (maybe_key_str, maybe_header_str) {
                        (Ok(key_str), Ok(header_str)) => {
                            request = request.add_header(key_str, header_str);
                        }
                        _ => continue,
                    }
                }

                request
                    .deadline(deadline)
                    .send()
                    .inspect_err(|err| {
                        log::error!(
                            target: LOG_TARGET,
                            "🕸️ Failed request on {}: {:?}",
                            rpc_endpoint_str,
                            err,
                        );
                    })
                    .ok()
            })
            .collect();

        Ok(pending_requests)
    }

    fn validate_block_numbers_consensus<I>(
        block_numbers: &mut Vec<ExternalBlockNumber>,
        max_block_deviation: I,
        threshold: usize,
    ) -> Option<ExternalBlockNumber>
    where
        I: UniqueSaturatedInto<ExternalBlockNumber>,
    {
        let total_blocks = block_numbers.len();

        if block_numbers.is_empty() || total_blocks < threshold {
            return None;
        }

        block_numbers.sort();
        let max_deviation: ExternalBlockNumber = max_block_deviation
            .unique_saturated_into();

        let mut left = total_blocks / 2;
        let mut right = left;

        loop {
            let can_expand_left =
                left > 0 && (block_numbers[right] - block_numbers[left - 1] <= max_deviation);
            let can_expand_right = right < total_blocks - 1
                && (block_numbers[right + 1] - block_numbers[left] <= max_deviation);

            match (can_expand_left, can_expand_right) {
                (true, true) => {
                    let dist_in_left = block_numbers[right] - block_numbers[left - 1];
                    let dist_in_right = block_numbers[right + 1] - block_numbers[left];

                    if dist_in_left > dist_in_right {
                        right += 1;
                    } else {
                        left -= 1;
                    }
                }
                (true, false) => left -= 1,
                (false, true) => right += 1,
                (false, false) => break,
            }
        }

        let window_len = right + 1 - left;
        let mid_idx = (right + left) / 2;

        let mid_value = if window_len % 2 == 0 {
            let a = block_numbers[mid_idx];
            let b = block_numbers[mid_idx + 1];

            a + (b - a) / 2
        } else {
            block_numbers[mid_idx]
        };

        window_len.ge(&threshold).then(|| mid_value)
    }

    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_offchain_storage<R: codec::Decode + Clone>(storage_key: &[u8], default_value: &R) -> R {
        StorageValueRef::persistent(&storage_key)
            .get::<R>()
            .ok()
            .flatten()
            .unwrap_or(default_value.clone())
    }

    fn local_authorities() -> impl Iterator<Item = (AuthIndex, T::AuthorityId)> {
        let authorities = Authorities::<T>::get();
        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.unique_saturated_into(),
                            local_authorities[location].clone(),
                        )
                    })
            })
    }

    fn initialize_authorities<'a, I: 'a>(validators: I)
    where
        I: Iterator<Item = (&'a T::AccountId, T::AuthorityId)>,
    {
        let authorities = validators.map(|x| x.1).collect::<Vec<_>>();
        let bounded_authorities = WeakBoundedVec::<_, T::MaxAuthorities>::force_from(
            authorities,
            Some("ghost-weaver pallet reached maximum number of authorities."),
        );

        Authorities::<T>::set(bounded_authorities);
        DisabledAuthorities::<T>::kill();
    }
}

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)>,
    {
        Self::initialize_authorities(validators);
    }

    fn on_new_session<'a, I: 'a>(_changed: bool, validators: I, _queued_validators: I)
    where
        I: Iterator<Item = (&'a T::AccountId, T::AuthorityId)>,
    {
        Self::initialize_authorities(validators);
    }

    fn on_before_session_ending() {}

    fn on_disabled(validator_index: u32) {
        DisabledAuthorities::<T>::mutate(|bitmap| {
            bitmap.insert(validator_index);
        });
    }
}