#[frame_support::pallet]

    #[pallet::error]

    #[pallet::pallet]

    #[pallet::hooks]

        fn on_initialize(_n: BlockNumberFor<T>) -> Weight {

            let mut weight = Weight::zero();

            // We clear this storage item to make sure its always included

            DidSetContainerAuthorData::<T>::kill();

            weight.saturating_accrue(T::DbWeight::get().writes(1));

            // The read onfinalizes

            weight.saturating_accrue(T::DbWeight::get().reads(1));

            weight

        }

        fn on_finalize(_: BlockNumberFor<T>) {

            assert!(

                <DidSetContainerAuthorData<T>>::exists(),

                "Container chain author data needs to be present in every block!"

        }

    #[pallet::call]

        ) -> DispatchResultWithPostInfo {

            ensure_none(origin)?;

            assert!(

                !<DidSetContainerAuthorData<T>>::exists(),

                "DidSetContainerAuthorData must be updated only once in a block",

            let container_chains_to_check: Vec<_> =

                T::ContainerChains::container_chains_with_collators(ForSession::Current)

                    .into_iter()

                    .filter_map(|(para_id, collators)| (!collators.is_empty()).then_some(para_id))

                    .collect();

            let mut total_weight =

                T::WeightInfo::set_latest_author_data(container_chains_to_check.len() as u32);

            // We do this first to make sure we don't do 2 reads (parachains and relay state)

            // when we have no containers registered

            // Essentially one can pass an empty proof if no container-chains are registered

            if !container_chains_to_check.is_empty() {

                let storage_reader = T::RelayOrPara::create_storage_reader(data);

                let parent_tanssi_slot = u64::from(T::SlotBeacon::slot()).into();

                let mut infos = Vec::with_capacity(container_chains_to_check.len());

                for para_id in container_chains_to_check {

                    match Self::fetch_block_info_from_proof(

                        &storage_reader,

                        para_id,

                        parent_tanssi_slot,

                    ) {

                        Ok(block_info) => {

                            LatestAuthor::<T>::mutate(

                                para_id,

                                |maybe_old_block_info: &mut Option<

                                >| {

                                    // No block number is the same as the last block number being 0:

                                    // the first block created by collators is block number 1.

                                    let old_block_number = maybe_old_block_info

                                        .as_ref()

                                        .map(|old_block_info| old_block_info.block_number)

                                        .unwrap_or(0);

                                    // We only reward author if the block increases

                                    // If there is no previous block, we should reward the author of the first block

                                    if block_info.block_number > old_block_number {

                                        let bi = block_info.clone();

                                        let info = AuthorNotingInfo {

                                            author: block_info.author,

                                            block_number: block_info.block_number,

                                            para_id,

                                        };

                                        infos.push(info);

                                        *maybe_old_block_info = Some(bi);

                                    }

                                },

                            );

                        }

                        Err(e) => log::warn!(

                            "Author-noting error {:?} found in para {:?}",

                            e,

                            u32::from(para_id)

                total_weight

                    .saturating_accrue(T::AuthorNotingHook::on_container_authors_noted(&infos));

            }

            DidSetContainerAuthorData::<T>::put(true);

            Ok(PostDispatchInfo {

                actual_weight: Some(total_weight),

                pays_fee: Pays::No,

            })

        ) -> DispatchResult {

            ensure_root(origin)?;

            LatestAuthor::<T>::insert(

                para_id,

                ContainerChainBlockInfo {

                    block_number,

                    author: author.clone(),

                    latest_slot_number,

                },

            );

            Self::deposit_event(Event::LatestAuthorChanged {

                para_id,

                block_number,

                new_author: author,

                latest_slot_number,

            });

            Ok(())

        pub fn kill_author_data(origin: OriginFor<T>, para_id: ParaId) -> DispatchResult {

            ensure_root(origin)?;

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

            Self::deposit_event(Event::RemovedAuthorData { para_id });

            Ok(())

    #[pallet::event]

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

        /// Latest author changed

        /// Removed author data

    #[pallet::storage]

    #[pallet::storage]

        fn create_inherent(data: &InherentData) -> Option<Self::Call> {

            let data = T::RelayOrPara::create_inherent_arg(data);

            Some(Call::set_latest_author_data { data })

        }

        fn is_inherent(call: &Self::Call) -> bool {

            matches!(call, Call::set_latest_author_data { .. })

        }

    fn fetch_block_info_from_proof<S: GenericStorageReader>(

        relay_state_proof: &S,

        para_id: ParaId,

        tanssi_slot: Slot,

    ) -> Result<ContainerChainBlockInfo<T::AccountId>, Error<T>> {

        let bytes = para_id.twox_64_concat();

        // CONCAT

        let key = [PARAS_HEADS_INDEX, bytes.as_slice()].concat();

        let head_data = relay_state_proof

            .read_entry::<HeadData>(key.as_slice(), None)

            .map_err(|e| match e {

                ReadEntryErr::Proof => panic!("Invalid proof provided for para head key"),

                _ => Error::<T>::FailedReading,

            })?;

        let author_header = sp_runtime::generic::Header::<BlockNumber, BlakeTwo256>::decode(

            &mut head_data.0.as_slice(),

        )

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

        let mut first_error = None;

        for aura_digest in author_header.digest().logs() {

            match Self::author_from_log(aura_digest, para_id, &author_header, tanssi_slot) {

                Ok(x) => return Ok(x),

                Err(e) => {

                    if first_error.is_none() {

                        first_error = Some(e);

                    }

        Err(first_error.unwrap_or(Error::<T>::AuraDigestFirstItem))

    }

    fn author_from_log(

        aura_digest: &DigestItem,

        para_id: ParaId,

        author_header: &sp_runtime::generic::Header<BlockNumber, BlakeTwo256>,

        tanssi_slot: Slot,

    ) -> Result<ContainerChainBlockInfo<T::AccountId>, Error<T>> {

        let (id, mut data) = aura_digest

            .as_pre_runtime()

            .ok_or(Error::<T>::AsPreRuntimeError)?;

        if id == AURA_ENGINE_ID {

            let slot = InherentType::decode(&mut data).map_err(|_| Error::<T>::NonDecodableSlot)?;

            let author = T::ContainerChainAuthor::author_for_slot(slot, para_id)

                .ok_or(Error::<T>::AuthorNotFound)?;

            Ok(ContainerChainBlockInfo {

                block_number: author_header.number,

                author,

                // We store the slot number of the current tanssi block to have a time-based notion

                // of when the last block of a container chain was included.

                // Note that this is not the slot of the container chain block, and it does not

                // indicate when that block was created, but when it was included in tanssi.

                latest_slot_number: tanssi_slot,

            })

            Err(Error::<T>::NonAuraDigest)

    }

    pub fn latest_author(para_id: ParaId) -> Option<ContainerChainBlockInfo<T::AccountId>> {

        LatestAuthor::<T>::get(para_id)

    }

    fn get_latest_author_info(para_id: ParaId) -> Option<ContainerChainBlockInfo<T::AccountId>> {

        LatestAuthor::<T>::get(para_id)

    }

    fn create_storage_reader(_data: Self::InherentArg) -> Self::GenericStorageReader {

        NativeStorageReader

    }

    fn create_inherent_arg(data: &InherentData) -> Self::InherentArg {

        data.get_data(&INHERENT_IDENTIFIER)

            .ok()

            .flatten()

            .expect("there is not data to be posted; qed")

    }

    fn create_storage_reader(data: Self::InherentArg) -> Self::GenericStorageReader {

        let tp_author_noting_inherent::OwnParachainInherentData {

            relay_storage_proof,

        } = data;

        let relay_chain_state = RCSP::current_relay_chain_state();

        let relay_storage_root = relay_chain_state.state_root;

        GenericStateProof::new(relay_storage_root, relay_storage_proof)

            .expect("Invalid relay chain state proof")

    }