near-primitives 0.36.0

use crate::action::GlobalContractIdentifier;
use crate::errors::EpochError;
use crate::hash::CryptoHash;
use crate::shard_layout::ShardLayout;
use crate::transaction::{Action, TransferAction};
use crate::types::{AccountId, Balance, BlockHeight, ShardId};
use borsh::{BorshDeserialize, BorshSerialize};
use itertools::Itertools;
use near_crypto::{KeyType, PublicKey};
use near_fmt::AbbrBytes;
use near_primitives_core::types::{Gas, ProtocolVersion};
use near_primitives_core::version::ProtocolFeature;
use near_schema_checker_lib::ProtocolSchema;
use serde_with::base64::Base64;
use serde_with::serde_as;
use std::borrow::Cow;
use std::collections::{BTreeMap, HashMap};
use std::fmt;
use std::io::{self, Read};
use std::io::{Error, ErrorKind};
use std::sync::Arc;

/// The outgoing (egress) data which will be transformed
/// to a `DataReceipt` to be sent to a `receipt.receiver`
#[derive(
    BorshSerialize,
    BorshDeserialize,
    Hash,
    Clone,
    Debug,
    PartialEq,
    Eq,
    serde::Serialize,
    serde::Deserialize,
    ProtocolSchema,
)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
pub struct DataReceiver {
    pub data_id: CryptoHash,
    pub receiver_id: AccountId,
}

/// Receipts are used for a cross-shard communication.
/// Receipts could be 2 types (determined by a `ReceiptEnum`): `ReceiptEnum::Action` of `ReceiptEnum::Data`.
#[derive(
    BorshSerialize,
    BorshDeserialize,
    Debug,
    PartialEq,
    Eq,
    Clone,
    serde::Serialize,
    serde::Deserialize,
    ProtocolSchema,
)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
pub struct ReceiptV0 {
    /// An issuer account_id of a particular receipt.
    /// `predecessor_id` could be either `Transaction` `signer_id` or intermediate contract's `account_id`.
    pub predecessor_id: AccountId,
    /// `receiver_id` is a receipt destination.
    pub receiver_id: AccountId,
    /// An unique id for the receipt
    pub receipt_id: CryptoHash,
    /// A receipt type
    pub receipt: ReceiptEnum,
}

/// A Receipt wrapper. Currently only V0 is used. V1 was prepared for a priority
/// feature that was never activated and has been removed.
#[derive(Debug, PartialEq, Eq, Clone, serde::Serialize, serde::Deserialize, ProtocolSchema)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
#[serde(untagged)]
pub enum Receipt {
    V0(ReceiptV0),
    // To introduce a V1 here, we need custom implementation of BorshSerialize
    // and BorshDeserialize (aka "hackery") to maintain backwards compatibility.
    //
    // Prior this was used: First field in `ReceiptV0` is an `AccountId`, which
    // is at most 64 bytes. For all valid `ReceiptV0`, the second byte must be 0
    // because of the little endian encoding of the length of the account id.
    // On the other hand, for `ReceiptV1`, always set the first byte to 1, and
    // `AccountId` must have nonzero length, so the second byte must not be
    // zero.
}

/// A receipt that is stored in the state with added metadata. A receipt may be
/// stored in the state as a delayed receipt, buffered receipt or a promise
/// yield receipt. The metadata contains additional information about receipt
///
/// Please note that the StateStoredReceipt implements custom serialization and
/// deserialization. Please see the comment on [ReceiptOrStateStoredReceipt]
/// for more details.
///
/// This struct is versioned so that it can be enhanced in the future.
#[derive(PartialEq, Eq, Debug, ProtocolSchema)]
pub enum StateStoredReceipt<'a> {
    V0(StateStoredReceiptV0<'a>),
    V1(StateStoredReceiptV1<'a>),
}

/// The V0 of StateStoredReceipt. It contains the receipt and metadata.
#[derive(BorshDeserialize, BorshSerialize, PartialEq, Eq, Debug, ProtocolSchema)]
pub struct StateStoredReceiptV0<'a> {
    /// The receipt.
    pub receipt: Cow<'a, Receipt>,
    pub metadata: StateStoredReceiptMetadata,
}

/// The V1 of StateStoredReceipt.
/// The data is the same as in V0.
/// Outgoing buffer metadata is updated only for versions V1 and higher.
/// The receipts start being stored as V1 after the protocol change that introduced
/// outgoing buffer metadata. Having a separate variant makes it clear whether the
/// outgoing buffer metadata should be updated when a receipt is stored/removed.
#[derive(BorshDeserialize, BorshSerialize, PartialEq, Eq, Debug, ProtocolSchema)]
pub struct StateStoredReceiptV1<'a> {
    pub receipt: Cow<'a, Receipt>,
    pub metadata: StateStoredReceiptMetadata,
}

/// The metadata associated with the receipt stored in state.
#[derive(BorshDeserialize, BorshSerialize, PartialEq, Eq, Debug, ProtocolSchema)]
pub struct StateStoredReceiptMetadata {
    /// The congestion gas of the receipt when it was stored in the state.
    /// Please see [compute_receipt_congestion_gas] for more details.
    pub congestion_gas: Gas,
    /// The congestion size of the receipt when it was stored in the state.
    /// Please see [compute_receipt_size] for more details.
    pub congestion_size: u64,
}

/// The tag that is used to differentiate between the Receipt and StateStoredReceipt.
const STATE_STORED_RECEIPT_TAG: u8 = u8::MAX;

/// This is a convenience struct for handling the migration from [Receipt] to
/// [StateStoredReceipt]. Both variants can be directly serialized and
/// deserialized to this struct.
///
/// This structure is only meant as a migration vehicle and should not be used
/// for other purposes. In order to make any changes to how receipts are stored
/// in state the StateStoredReceipt should be used. It supports versioning.
///
/// The receipt in both variants is stored as a Cow to allow for both owned and
/// borrowed ownership. The owned receipt should be used when pulling receipts
/// from the state. The borrowed ownership can be used when pushing receipts
/// into the state. In that case the receipt should never need to be cloned. The
/// serialization only needs a reference.
#[derive(PartialEq, Eq, Debug, ProtocolSchema)]
pub enum ReceiptOrStateStoredReceipt<'a> {
    Receipt(Cow<'a, Receipt>),
    StateStoredReceipt(StateStoredReceipt<'a>),
}

impl ReceiptOrStateStoredReceipt<'_> {
    pub fn into_receipt(self) -> Receipt {
        match self {
            ReceiptOrStateStoredReceipt::Receipt(receipt) => receipt.into_owned(),
            ReceiptOrStateStoredReceipt::StateStoredReceipt(receipt) => receipt.into_receipt(),
        }
    }

    pub fn get_receipt(&self) -> &Receipt {
        match self {
            ReceiptOrStateStoredReceipt::Receipt(receipt) => receipt,
            ReceiptOrStateStoredReceipt::StateStoredReceipt(receipt) => receipt.get_receipt(),
        }
    }

    pub fn should_update_outgoing_metadatas(&self) -> bool {
        match self {
            ReceiptOrStateStoredReceipt::Receipt(_) => false,
            ReceiptOrStateStoredReceipt::StateStoredReceipt(state_stored_receipt) => {
                state_stored_receipt.should_update_outgoing_metadatas()
            }
        }
    }
}

impl<'a> StateStoredReceipt<'a> {
    pub fn new_owned(receipt: Receipt, metadata: StateStoredReceiptMetadata) -> Self {
        let receipt = Cow::Owned(receipt);
        Self::V1(StateStoredReceiptV1 { receipt, metadata })
    }

    pub fn new_borrowed(receipt: &'a Receipt, metadata: StateStoredReceiptMetadata) -> Self {
        let receipt = Cow::Borrowed(receipt);

        Self::V1(StateStoredReceiptV1 { receipt, metadata })
    }

    pub fn into_receipt(self) -> Receipt {
        match self {
            StateStoredReceipt::V0(v0) => v0.receipt.into_owned(),
            StateStoredReceipt::V1(v1) => v1.receipt.into_owned(),
        }
    }

    pub fn get_receipt(&self) -> &Receipt {
        match self {
            StateStoredReceipt::V0(v0) => &v0.receipt,
            StateStoredReceipt::V1(v1) => &v1.receipt,
        }
    }

    pub fn metadata(&self) -> &StateStoredReceiptMetadata {
        match self {
            StateStoredReceipt::V0(v0) => &v0.metadata,
            StateStoredReceipt::V1(v1) => &v1.metadata,
        }
    }

    pub fn should_update_outgoing_metadatas(&self) -> bool {
        match self {
            StateStoredReceipt::V0(_) => false,
            StateStoredReceipt::V1(_) => true,
        }
    }
}

impl BorshSerialize for Receipt {
    fn serialize<W: io::Write>(&self, writer: &mut W) -> io::Result<()> {
        match self {
            Receipt::V0(receipt) => BorshSerialize::serialize(&receipt, writer),
        }
    }
}

impl BorshDeserialize for Receipt {
    fn deserialize_reader<R: Read>(reader: &mut R) -> std::io::Result<Self> {
        Ok(Receipt::V0(ReceiptV0::deserialize_reader(reader)?))
    }
}

impl BorshSerialize for StateStoredReceipt<'_> {
    fn serialize<W: io::Write>(&self, writer: &mut W) -> io::Result<()> {
        // The serialization format for StateStored receipt is as follows:
        // Byte 1: STATE_STORED_RECEIPT_TAG
        // Byte 2: STATE_STORED_RECEIPT_TAG
        // Byte 3: enum version (e.g. V0 => 0_u8)
        // serialized variant value

        BorshSerialize::serialize(&STATE_STORED_RECEIPT_TAG, writer)?;
        BorshSerialize::serialize(&STATE_STORED_RECEIPT_TAG, writer)?;
        match self {
            StateStoredReceipt::V0(v0) => {
                BorshSerialize::serialize(&0_u8, writer)?;
                BorshSerialize::serialize(&v0, writer)?;
            }
            StateStoredReceipt::V1(v1) => {
                BorshSerialize::serialize(&1_u8, writer)?;
                BorshSerialize::serialize(&v1, writer)?;
            }
        }
        Ok(())
    }
}

impl BorshDeserialize for StateStoredReceipt<'_> {
    fn deserialize_reader<R: Read>(reader: &mut R) -> io::Result<Self> {
        let u1 = u8::deserialize_reader(reader)?;
        let u2 = u8::deserialize_reader(reader)?;
        let u3 = u8::deserialize_reader(reader)?;

        if u1 != STATE_STORED_RECEIPT_TAG || u2 != STATE_STORED_RECEIPT_TAG {
            let error = format!(
                "Invalid tag found when deserializing StateStoredReceipt. Found: {}, {}. Expected: {}, {}",
                u1, u2, STATE_STORED_RECEIPT_TAG, STATE_STORED_RECEIPT_TAG
            );
            let error = Error::new(ErrorKind::Other, error);
            return Err(io::Error::new(ErrorKind::InvalidData, error));
        }

        match u3 {
            0 => {
                let v0 = StateStoredReceiptV0::deserialize_reader(reader)?;
                Ok(StateStoredReceipt::V0(v0))
            }
            1 => {
                let v1 = StateStoredReceiptV1::deserialize_reader(reader)?;
                Ok(StateStoredReceipt::V1(v1))
            }
            v => {
                let error = format!(
                    "Invalid version found when deserializing StateStoredReceipt. Found: {}. Expected: 0",
                    v
                );
                let error = Error::new(ErrorKind::Other, error);
                Err(io::Error::new(ErrorKind::InvalidData, error))
            }
        }
    }
}

impl BorshSerialize for ReceiptOrStateStoredReceipt<'_> {
    fn serialize<W: io::Write>(&self, writer: &mut W) -> io::Result<()> {
        // This is custom serialization in order to provide backwards
        // compatibility for migration from Receipt to StateStoredReceipt.

        // Please see the comment in deserialize_reader for more details.
        match self {
            ReceiptOrStateStoredReceipt::Receipt(receipt) => {
                BorshSerialize::serialize(receipt, writer)
            }
            ReceiptOrStateStoredReceipt::StateStoredReceipt(receipt) => {
                BorshSerialize::serialize(receipt, writer)
            }
        }
    }
}

impl BorshDeserialize for ReceiptOrStateStoredReceipt<'_> {
    fn deserialize_reader<R: Read>(reader: &mut R) -> io::Result<Self> {
        // This is custom deserialization in order to provide backwards
        // compatibility for migration from Receipt to StateStoredReceipt.

        // Both variants (Receipt and StateStoredReceipt) need to be directly
        // deserializable into the ReceiptOrStateStoredReceipt.

        // Read the first two bytes in order to discriminate between the Receipt
        // and StateStoredReceipt.
        // The StateStored receipt has the tag as the first two bytes.
        // The Receipt::V0 has 0 as the second byte.
        // Using 1 as the first byte is reserved in case Receipt::V1 is needed again.
        let u1 = u8::deserialize_reader(reader)?;
        let u2 = u8::deserialize_reader(reader)?;

        // Put the read bytes back into the reader by chaining.
        let prefix = [u1, u2];
        let mut reader = prefix.chain(reader);

        if u1 == STATE_STORED_RECEIPT_TAG && u2 == STATE_STORED_RECEIPT_TAG {
            let receipt = StateStoredReceipt::deserialize_reader(&mut reader)?;
            Ok(ReceiptOrStateStoredReceipt::StateStoredReceipt(receipt))
        } else {
            let receipt = Receipt::deserialize_reader(&mut reader)?;
            let receipt = Cow::Owned(receipt);
            Ok(ReceiptOrStateStoredReceipt::Receipt(receipt))
        }
    }
}

impl Receipt {
    pub fn from_tx(
        receipt_id: CryptoHash,
        tx_signer_id: AccountId,
        tx_receiver_id: AccountId,
        signer_public_key: PublicKey,
        gas_price: Balance,
        actions: Vec<Action>,
    ) -> Self {
        Receipt::V0(ReceiptV0 {
            predecessor_id: tx_signer_id.clone(),
            receiver_id: tx_receiver_id,
            receipt_id,
            receipt: ReceiptEnum::Action(ActionReceipt {
                signer_id: tx_signer_id,
                signer_public_key,
                gas_price,
                output_data_receivers: vec![],
                input_data_ids: vec![],
                actions,
            }),
        })
    }

    pub fn receiver_id(&self) -> &AccountId {
        let Receipt::V0(receipt) = self;
        &receipt.receiver_id
    }

    pub fn set_receiver_id(&mut self, receiver_id: AccountId) {
        let Receipt::V0(receipt) = self;
        receipt.receiver_id = receiver_id;
    }

    pub fn predecessor_id(&self) -> &AccountId {
        let Receipt::V0(receipt) = self;
        &receipt.predecessor_id
    }

    pub fn set_predecessor_id(&mut self, predecessor_id: AccountId) {
        let Receipt::V0(receipt) = self;
        receipt.predecessor_id = predecessor_id;
    }

    pub fn receipt(&self) -> &ReceiptEnum {
        let Receipt::V0(receipt) = self;
        &receipt.receipt
    }

    pub fn versioned_receipt(&self) -> VersionedReceiptEnum<'_> {
        let Receipt::V0(receipt) = self;
        VersionedReceiptEnum::from(&receipt.receipt)
    }

    pub fn receipt_mut(&mut self) -> &mut ReceiptEnum {
        let Receipt::V0(receipt) = self;
        &mut receipt.receipt
    }

    pub fn take_versioned_receipt<'a>(self) -> VersionedReceiptEnum<'a> {
        let Receipt::V0(receipt) = self;
        VersionedReceiptEnum::from(receipt.receipt)
    }

    pub fn receipt_id(&self) -> &CryptoHash {
        let Receipt::V0(receipt) = self;
        &receipt.receipt_id
    }

    pub fn set_receipt_id(&mut self, receipt_id: CryptoHash) {
        let Receipt::V0(receipt) = self;
        receipt.receipt_id = receipt_id;
    }

    pub fn refund_to(&self) -> &Option<AccountId> {
        match self.receipt() {
            ReceiptEnum::Action(_)
            | ReceiptEnum::Data(_)
            | ReceiptEnum::PromiseYield(_)
            | ReceiptEnum::PromiseResume(_)
            | ReceiptEnum::GlobalContractDistribution(_) => &None,
            ReceiptEnum::ActionV2(action_receipt_v2)
            | ReceiptEnum::PromiseYieldV2(action_receipt_v2) => &action_receipt_v2.refund_to,
        }
    }

    pub fn balance_refund_receiver(&self) -> &AccountId {
        self.refund_to().as_ref().unwrap_or_else(|| self.predecessor_id())
    }

    /// It's not a content hash, but receipt_id is unique.
    pub fn get_hash(&self) -> CryptoHash {
        *self.receipt_id()
    }

    pub fn receiver_shard_id(&self, shard_layout: &ShardLayout) -> Result<ShardId, EpochError> {
        let shard_id = match self.receipt() {
            ReceiptEnum::Action(_)
            | ReceiptEnum::ActionV2(_)
            | ReceiptEnum::Data(_)
            | ReceiptEnum::PromiseYield(_)
            | ReceiptEnum::PromiseYieldV2(_)
            | ReceiptEnum::PromiseResume(_) => {
                shard_layout.account_id_to_shard_id(self.receiver_id())
            }
            ReceiptEnum::GlobalContractDistribution(receipt) => {
                let target_shard = receipt.target_shard();
                if shard_layout.shard_ids().contains(&target_shard) {
                    target_shard
                } else {
                    // The target shard may be from an arbitrarily old layout (the receipt could
                    // have been delayed across multiple resharding events). resolve_to_current_shard
                    // will find a shard descendant in the current layout.
                    let Some(current_shard) = shard_layout.resolve_to_current_shard(target_shard)
                    else {
                        return Err(EpochError::ShardingError(format!(
                            "Shard {target_shard} does not exist in the shard layout or its split history",
                        )));
                    };
                    current_shard
                }
            }
        };
        Ok(shard_id)
    }

    /// An instant receipt is a receipt which should be processed immediately after the receipt that
    /// produced it, in the same chunk, irrespective of the gas limit.
    /// The expectation is that applying an instant receipt is a quick operation (e.g. setting a few values in the state).
    /// Instant receipts generally shouldn't emit new instant receipts, as it could lead to
    /// infinitely many receipts being executed in a single chunk.
    pub fn is_instant_receipt(&self, protocol_version: ProtocolVersion) -> bool {
        match self.versioned_receipt() {
            VersionedReceiptEnum::PromiseYield(_) => {
                // PromiseYield receipts are instant receipts.
                // Applying a PromiseYield receipt is one trie write, it's okay to make it an instant receipt.
                ProtocolFeature::InstantPromiseYield.enabled(protocol_version)
            }
            VersionedReceiptEnum::Action(action_receipt) => {
                // Action receipts containing a single DeleteAccount action and no input
                // promises are instant receipts.
                // Deleting an account is a quick trie operation, it's okay to make it instant.
                ProtocolFeature::InstantDeleteAccount.enabled(protocol_version)
                    && matches!(action_receipt.actions(), [Action::DeleteAccount(_)])
                    && action_receipt.input_data_ids().is_empty()
            }
            VersionedReceiptEnum::Data(_)
            | VersionedReceiptEnum::PromiseResume(_)
            | VersionedReceiptEnum::GlobalContractDistribution(_) => false,
        }
    }

    /// Generates a receipt with a transfer from system for a given balance without a receipt_id.
    /// This should be used for token refunds instead of gas refunds.
    /// It doesn't refund the allowance of the access key. For gas refunds use `new_gas_refund`.
    pub fn new_balance_refund(receiver_id: &AccountId, refund: Balance) -> Self {
        Receipt::V0(ReceiptV0 {
            predecessor_id: "system".parse().unwrap(),
            receiver_id: receiver_id.clone(),
            receipt_id: CryptoHash::default(),
            receipt: ReceiptEnum::Action(ActionReceipt {
                signer_id: "system".parse().unwrap(),
                signer_public_key: PublicKey::empty(KeyType::ED25519),
                gas_price: Balance::ZERO,
                output_data_receivers: vec![],
                input_data_ids: vec![],
                actions: vec![Action::Transfer(TransferAction { deposit: refund })],
            }),
        })
    }

    /// Generates a receipt with a transfer action from system for a given balance without a
    /// receipt_id. It contains `signer_id` and `signer_public_key` to indicate this is a gas
    /// refund. The execution of this receipt will try to refund the allowance of the
    /// access key with the given public key.
    /// NOTE: The access key may be replaced by the owner, so the execution can't rely that the
    /// access key is the same and it should use best effort for the refund.
    pub fn new_gas_refund(
        receiver_id: &AccountId,
        refund: Balance,
        signer_public_key: PublicKey,
    ) -> Self {
        Receipt::V0(ReceiptV0 {
            predecessor_id: "system".parse().unwrap(),
            receiver_id: receiver_id.clone(),
            receipt_id: CryptoHash::default(),
            receipt: ReceiptEnum::Action(ActionReceipt {
                signer_id: receiver_id.clone(),
                signer_public_key,
                gas_price: Balance::ZERO,
                output_data_receivers: vec![],
                input_data_ids: vec![],
                actions: vec![Action::Transfer(TransferAction { deposit: refund })],
            }),
        })
    }

    pub fn new_global_contract_distribution(
        predecessor_id: AccountId,
        receipt: GlobalContractDistributionReceipt,
    ) -> Self {
        Self::V0(ReceiptV0 {
            predecessor_id,
            receiver_id: "system".parse().unwrap(),
            receipt_id: CryptoHash::default(),
            receipt: ReceiptEnum::GlobalContractDistribution(receipt),
        })
    }
}

/// Receipt could be either ActionReceipt or DataReceipt
#[derive(
    BorshSerialize,
    BorshDeserialize,
    Clone,
    Debug,
    PartialEq,
    Eq,
    serde::Serialize,
    serde::Deserialize,
    ProtocolSchema,
)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
#[borsh(use_discriminant = true)]
#[repr(u8)]
pub enum ReceiptEnum {
    Action(ActionReceipt) = 0,
    Data(DataReceipt) = 1,
    PromiseYield(ActionReceipt) = 2,
    PromiseResume(DataReceipt) = 3,
    GlobalContractDistribution(GlobalContractDistributionReceipt) = 4,
    ActionV2(ActionReceiptV2) = 5,
    PromiseYieldV2(ActionReceiptV2) = 6,
}

/// ActionReceipt is derived from an Action from `Transaction or from Receipt`
/// TODO(#14709): deprecate in favor of `ActionReceiptV2`.
#[derive(
    BorshSerialize,
    BorshDeserialize,
    Debug,
    PartialEq,
    Eq,
    Clone,
    serde::Serialize,
    serde::Deserialize,
    ProtocolSchema,
)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
pub struct ActionReceipt {
    /// A signer of the original transaction
    pub signer_id: AccountId,
    /// An access key which was used to sign the original transaction
    pub signer_public_key: PublicKey,
    /// A gas_price which has been used to buy gas in the original transaction
    pub gas_price: Balance,
    /// If present, where to route the output data
    pub output_data_receivers: Vec<DataReceiver>,
    /// A list of the input data dependencies for this Receipt to process.
    /// If all `input_data_ids` for this receipt are delivered to the account
    /// that means we have all the `ReceivedData` input which will be than converted to a
    /// `PromiseResult::Successful(value)` or `PromiseResult::Failed`
    /// depending on `ReceivedData` is `Some(_)` or `None`
    pub input_data_ids: Vec<CryptoHash>,
    /// A list of actions to process when all input_data_ids are filled
    pub actions: Vec<Action>,
}

/// ActionReceipt is derived from a set of Actions from `Transaction or from Receipt`
#[derive(
    BorshSerialize,
    BorshDeserialize,
    Debug,
    PartialEq,
    Eq,
    Clone,
    serde::Serialize,
    serde::Deserialize,
    ProtocolSchema,
)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
pub struct ActionReceiptV2 {
    /// A signer of the original transaction
    pub signer_id: AccountId,
    /// The receiver of any balance refunds form this receipt if it is different from receiver_id.
    pub refund_to: Option<AccountId>,
    /// An access key which was used to sign the original transaction
    pub signer_public_key: PublicKey,
    /// A gas_price which has been used to buy gas in the original transaction
    pub gas_price: Balance,
    /// If present, where to route the output data
    pub output_data_receivers: Vec<DataReceiver>,
    /// A list of the input data dependencies for this Receipt to process.
    /// If all `input_data_ids` for this receipt are delivered to the account
    /// that means we have all the `ReceivedData` input which will be than converted to a
    /// `PromiseResult::Successful(value)` or `PromiseResult::Failed`
    /// depending on `ReceivedData` is `Some(_)` or `None`
    pub input_data_ids: Vec<CryptoHash>,
    /// A list of actions to process when all input_data_ids are filled
    pub actions: Vec<Action>,
}

/// Convenience wrapper for common logic accessing fields on action receipts of
/// different versions.
#[derive(Debug, PartialEq, Eq, Clone)]
pub enum VersionedActionReceipt<'a> {
    V1(Cow<'a, ActionReceipt>),
    V2(Cow<'a, ActionReceiptV2>),
}

impl VersionedActionReceipt<'_> {
    /// A signer of the original transaction
    pub fn signer_id(&self) -> &AccountId {
        match self {
            VersionedActionReceipt::V1(action_receipt) => &action_receipt.signer_id,
            VersionedActionReceipt::V2(action_receipt) => &action_receipt.signer_id,
        }
    }

    /// The receiver of any balance refunds form this receipt if it is different from receiver_id.
    pub fn refund_to(&self) -> &Option<AccountId> {
        match self {
            VersionedActionReceipt::V1(_) => &None,
            VersionedActionReceipt::V2(action_receipt) => &action_receipt.refund_to,
        }
    }

    /// An access key which was used to sign the original transaction
    pub fn signer_public_key(&self) -> &PublicKey {
        match self {
            VersionedActionReceipt::V1(action_receipt) => &action_receipt.signer_public_key,
            VersionedActionReceipt::V2(action_receipt) => &action_receipt.signer_public_key,
        }
    }

    /// A gas_price which has been used to buy gas in the original transaction
    pub fn gas_price(&self) -> Balance {
        match self {
            VersionedActionReceipt::V1(action_receipt) => action_receipt.gas_price,
            VersionedActionReceipt::V2(action_receipt) => action_receipt.gas_price,
        }
    }

    /// If present, where to route the output data
    pub fn output_data_receivers(&self) -> &[DataReceiver] {
        match self {
            VersionedActionReceipt::V1(action_receipt) => &action_receipt.output_data_receivers,
            VersionedActionReceipt::V2(action_receipt) => &action_receipt.output_data_receivers,
        }
    }

    /// A list of the input data dependencies for this Receipt to process.
    /// If all `input_data_ids` for this receipt are delivered to the account
    /// that means we have all the `ReceivedData` input which will be than converted to a
    /// `PromiseResult::Successful(value)` or `PromiseResult::Failed`
    /// depending on `ReceivedData` is `Some(_)` or `None`
    pub fn input_data_ids(&self) -> &[CryptoHash] {
        match self {
            VersionedActionReceipt::V1(action_receipt) => &action_receipt.input_data_ids,
            VersionedActionReceipt::V2(action_receipt) => &action_receipt.input_data_ids,
        }
    }

    /// A list of actions to process when all input_data_ids are filled
    pub fn actions(&self) -> &[Action] {
        match self {
            VersionedActionReceipt::V1(action_receipt) => &action_receipt.actions,
            VersionedActionReceipt::V2(action_receipt) => &action_receipt.actions,
        }
    }
}

impl<'a> From<&'a ActionReceipt> for VersionedActionReceipt<'a> {
    fn from(other: &'a ActionReceipt) -> Self {
        VersionedActionReceipt::V1(Cow::Borrowed(other))
    }
}

impl From<ActionReceipt> for VersionedActionReceipt<'_> {
    fn from(other: ActionReceipt) -> Self {
        VersionedActionReceipt::V1(Cow::Owned(other))
    }
}

impl<'a> From<&'a ActionReceiptV2> for VersionedActionReceipt<'a> {
    fn from(other: &'a ActionReceiptV2) -> Self {
        VersionedActionReceipt::V2(Cow::Borrowed(other))
    }
}

impl From<ActionReceiptV2> for VersionedActionReceipt<'_> {
    fn from(other: ActionReceiptV2) -> Self {
        VersionedActionReceipt::V2(Cow::Owned(other))
    }
}

/// Convenience wrapper for common logic accessing fields on [`ReceiptEnum`] of
/// different versions of its variants.
///
/// Note: So far, only action receipts are versioned. Other versioned variants
/// can be added later if and when necessary. Then, the Cow pointer might be
/// better pushed down, as done it [`VersionedActionReceipt`]. (Cow pointers are
/// useful here to allow using the wrapper with owned and borrowed values
/// without cloning.)
#[derive(Debug, PartialEq, Eq, Clone)]
pub enum VersionedReceiptEnum<'a> {
    Action(VersionedActionReceipt<'a>),
    Data(Cow<'a, DataReceipt>),
    PromiseYield(VersionedActionReceipt<'a>),
    PromiseResume(Cow<'a, DataReceipt>),
    GlobalContractDistribution(Cow<'a, GlobalContractDistributionReceipt>),
}

impl<'a> From<&'a ReceiptEnum> for VersionedReceiptEnum<'a> {
    fn from(other: &'a ReceiptEnum) -> Self {
        match other {
            ReceiptEnum::Action(action_receipt) => {
                VersionedReceiptEnum::Action(action_receipt.into())
            }
            ReceiptEnum::Data(data_receipt) => {
                VersionedReceiptEnum::Data(Cow::Borrowed(data_receipt))
            }
            ReceiptEnum::PromiseYield(action_receipt) => {
                VersionedReceiptEnum::PromiseYield(action_receipt.into())
            }
            ReceiptEnum::PromiseResume(data_receipt) => {
                VersionedReceiptEnum::PromiseResume(Cow::Borrowed(data_receipt))
            }
            ReceiptEnum::GlobalContractDistribution(global_contract_distribution_receipt) => {
                VersionedReceiptEnum::GlobalContractDistribution(Cow::Borrowed(
                    global_contract_distribution_receipt,
                ))
            }
            ReceiptEnum::ActionV2(action_receipt) => {
                VersionedReceiptEnum::Action(action_receipt.into())
            }
            ReceiptEnum::PromiseYieldV2(action_receipt) => {
                VersionedReceiptEnum::PromiseYield(action_receipt.into())
            }
        }
    }
}

impl<'a> From<ReceiptEnum> for VersionedReceiptEnum<'a> {
    fn from(other: ReceiptEnum) -> Self {
        match other {
            ReceiptEnum::Action(action_receipt) => {
                VersionedReceiptEnum::Action(action_receipt.into())
            }
            ReceiptEnum::Data(data_receipt) => VersionedReceiptEnum::Data(Cow::Owned(data_receipt)),
            ReceiptEnum::PromiseYield(action_receipt) => {
                VersionedReceiptEnum::PromiseYield(action_receipt.into())
            }
            ReceiptEnum::PromiseResume(data_receipt) => {
                VersionedReceiptEnum::PromiseResume(Cow::Owned(data_receipt))
            }
            ReceiptEnum::GlobalContractDistribution(global_contract_distribution_receipt) => {
                VersionedReceiptEnum::GlobalContractDistribution(Cow::Owned(
                    global_contract_distribution_receipt,
                ))
            }
            ReceiptEnum::ActionV2(action_receipt) => {
                VersionedReceiptEnum::Action(action_receipt.into())
            }
            ReceiptEnum::PromiseYieldV2(action_receipt) => {
                VersionedReceiptEnum::PromiseYield(action_receipt.into())
            }
        }
    }
}

/// An incoming (ingress) `DataReceipt` which is going to a Receipt's `receiver` input_data_ids
/// Which will be converted to `PromiseResult::Successful(value)` or `PromiseResult::Failed`
#[serde_as]
#[derive(
    BorshSerialize,
    BorshDeserialize,
    Hash,
    PartialEq,
    Eq,
    Clone,
    serde::Serialize,
    serde::Deserialize,
    ProtocolSchema,
)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
pub struct DataReceipt {
    pub data_id: CryptoHash,
    #[serde_as(as = "Option<Base64>")]
    #[cfg_attr(feature = "schemars", schemars(with = "Option<String>"))]
    pub data: Option<Vec<u8>>,
}

impl fmt::Debug for DataReceipt {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("DataReceipt")
            .field("data_id", &self.data_id)
            .field("data", &format_args!("{}", AbbrBytes(self.data.as_deref())))
            .finish()
    }
}

/// A temporary data which is created by processing of DataReceipt
/// stored in a state trie with a key = `account_id` + `data_id` until
/// `input_data_ids` of all incoming Receipts are satisfied
/// None means data retrieval was failed
#[derive(BorshSerialize, BorshDeserialize, Hash, PartialEq, Eq, Clone, ProtocolSchema)]
pub struct ReceivedData {
    pub data: Option<Vec<u8>>,
}

impl fmt::Debug for ReceivedData {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("ReceivedData")
            .field("data", &format_args!("{}", AbbrBytes(self.data.as_deref())))
            .finish()
    }
}

#[derive(
    BorshSerialize,
    BorshDeserialize,
    Hash,
    PartialEq,
    Eq,
    Clone,
    Debug,
    serde::Deserialize,
    serde::Serialize,
    ProtocolSchema,
)]
#[borsh(use_discriminant = true)]
#[repr(u8)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
pub enum GlobalContractDistributionReceipt {
    V1(GlobalContractDistributionReceiptV1) = 0,
    V2(GlobalContractDistributionReceiptV2) = 1,
}

impl GlobalContractDistributionReceipt {
    pub fn new(
        id: GlobalContractIdentifier,
        target_shard: ShardId,
        already_delivered_shards: Vec<ShardId>,
        code: Arc<[u8]>,
        nonce: u64,
        protocol_version: ProtocolVersion,
    ) -> Self {
        if ProtocolFeature::GlobalContractDistributionNonce.enabled(protocol_version) {
            Self::new_v2(id, target_shard, already_delivered_shards, code, nonce)
        } else {
            Self::new_v1(id, target_shard, already_delivered_shards, code)
        }
    }

    pub fn new_v1(
        id: GlobalContractIdentifier,
        target_shard: ShardId,
        already_delivered_shards: Vec<ShardId>,
        code: Arc<[u8]>,
    ) -> Self {
        Self::V1(GlobalContractDistributionReceiptV1 {
            id,
            target_shard,
            already_delivered_shards,
            code,
        })
    }

    pub fn new_v2(
        id: GlobalContractIdentifier,
        target_shard: ShardId,
        already_delivered_shards: Vec<ShardId>,
        code: Arc<[u8]>,
        nonce: u64,
    ) -> Self {
        Self::V2(GlobalContractDistributionReceiptV2 {
            id,
            target_shard,
            already_delivered_shards,
            code,
            nonce,
        })
    }

    pub fn id(&self) -> &GlobalContractIdentifier {
        match &self {
            Self::V1(v1) => &v1.id,
            Self::V2(v2) => &v2.id,
        }
    }

    pub fn target_shard(&self) -> ShardId {
        match &self {
            Self::V1(v1) => v1.target_shard,
            Self::V2(v2) => v2.target_shard,
        }
    }

    pub fn already_delivered_shards(&self) -> &[ShardId] {
        match &self {
            Self::V1(v1) => &v1.already_delivered_shards,
            Self::V2(v2) => &v2.already_delivered_shards,
        }
    }

    pub fn code(&self) -> &Arc<[u8]> {
        match &self {
            Self::V1(v1) => &v1.code,
            Self::V2(v2) => &v2.code,
        }
    }

    /// Returns the nonce of the distribution.
    /// V1 receipts return 0, V2 receipts return their stored nonce.
    pub fn nonce(&self) -> u64 {
        match &self {
            Self::V1(_) => 0,
            Self::V2(v2) => v2.nonce,
        }
    }

    /// Returns the nonce of the distribution.
    /// V1 receipts return None, V2 receipts return their stored nonce.
    pub fn maybe_nonce(&self) -> Option<u64> {
        match &self {
            Self::V1(_) => None,
            Self::V2(v2) => Some(v2.nonce),
        }
    }

    /// Clones the receipt with an updated target shard and already-delivered shards list.
    pub fn forward(&self, target_shard: ShardId, already_delivered_shards: Vec<ShardId>) -> Self {
        match self {
            Self::V1(v1) => Self::V1(GlobalContractDistributionReceiptV1 {
                target_shard,
                already_delivered_shards,
                ..v1.clone()
            }),
            Self::V2(v2) => Self::V2(GlobalContractDistributionReceiptV2 {
                target_shard,
                already_delivered_shards,
                ..v2.clone()
            }),
        }
    }
}

#[serde_as]
#[derive(
    BorshSerialize,
    BorshDeserialize,
    Hash,
    PartialEq,
    Eq,
    Clone,
    serde::Deserialize,
    serde::Serialize,
    ProtocolSchema,
)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
pub struct GlobalContractDistributionReceiptV1 {
    id: GlobalContractIdentifier,
    target_shard: ShardId,
    already_delivered_shards: Vec<ShardId>,
    #[serde_as(as = "Base64")]
    #[cfg_attr(feature = "schemars", schemars(with = "String"))]
    code: Arc<[u8]>,
}

impl fmt::Debug for GlobalContractDistributionReceiptV1 {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("GlobalContractDistributionReceipt")
            .field("id", &self.id)
            .field("target_shard", &self.target_shard)
            .field("already_delivered_shards", &self.already_delivered_shards)
            .field("code", &..)
            .finish()
    }
}

#[serde_as]
#[derive(
    BorshSerialize,
    BorshDeserialize,
    Hash,
    PartialEq,
    Eq,
    Clone,
    serde::Deserialize,
    serde::Serialize,
    ProtocolSchema,
)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
pub struct GlobalContractDistributionReceiptV2 {
    id: GlobalContractIdentifier,
    target_shard: ShardId,
    already_delivered_shards: Vec<ShardId>,
    #[serde_as(as = "Base64")]
    #[cfg_attr(feature = "schemars", schemars(with = "String"))]
    code: Arc<[u8]>,
    nonce: u64,
}

impl fmt::Debug for GlobalContractDistributionReceiptV2 {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("GlobalContractDistributionReceiptV2")
            .field("id", &self.id)
            .field("target_shard", &self.target_shard)
            .field("already_delivered_shards", &self.already_delivered_shards)
            .field("code", &..)
            .field("nonce", &self.nonce)
            .finish()
    }
}

/// Stores indices for a persistent queue for delayed receipts that didn't fit into a block.
#[derive(Default, BorshSerialize, BorshDeserialize, Clone, PartialEq, Debug, ProtocolSchema)]
pub struct DelayedReceiptIndices {
    // First inclusive index in the queue.
    pub first_index: u64,
    // Exclusive end index of the queue
    pub next_available_index: u64,
}

impl DelayedReceiptIndices {
    pub fn len(&self) -> u64 {
        self.next_available_index - self.first_index
    }
}

/// Stores indices for a persistent queue for PromiseYield timeouts.
#[derive(Default, BorshSerialize, BorshDeserialize, Clone, PartialEq, Debug, ProtocolSchema)]
pub struct PromiseYieldIndices {
    // First inclusive index in the queue.
    pub first_index: u64,
    // Exclusive end index of the queue
    pub next_available_index: u64,
}

impl PromiseYieldIndices {
    pub fn len(&self) -> u64 {
        self.next_available_index - self.first_index
    }
}

/// Entries in the queue of PromiseYield timeouts.
#[derive(BorshSerialize, BorshDeserialize, Clone, PartialEq, Debug, ProtocolSchema)]
pub struct PromiseYieldTimeout {
    /// The account on which the yielded promise was created
    pub account_id: AccountId,
    /// The `data_id` used to identify the awaited input data
    pub data_id: CryptoHash,
    /// The block height before which the data must be submitted
    pub expires_at: BlockHeight,
}

/// Stores indices for a persistent queue in the state trie.
#[derive(Default, BorshSerialize, BorshDeserialize, Clone, PartialEq, Debug, ProtocolSchema)]
pub struct TrieQueueIndices {
    // First inclusive index in the queue.
    pub first_index: u64,
    // Exclusive end index of the queue
    pub next_available_index: u64,
}

impl TrieQueueIndices {
    pub fn len(&self) -> u64 {
        self.next_available_index - self.first_index
    }

    pub fn is_default(&self) -> bool {
        self.next_available_index == 0
    }
}

impl From<DelayedReceiptIndices> for TrieQueueIndices {
    fn from(other: DelayedReceiptIndices) -> Self {
        Self { first_index: other.first_index, next_available_index: other.next_available_index }
    }
}

impl From<TrieQueueIndices> for DelayedReceiptIndices {
    fn from(other: TrieQueueIndices) -> Self {
        Self { first_index: other.first_index, next_available_index: other.next_available_index }
    }
}

/// Stores indices for a persistent queue for buffered receipts that couldn't be
/// forwarded.
///
/// This is the singleton value stored in the `BUFFERED_RECEIPT_INDICES` trie
/// column.
#[derive(Default, BorshSerialize, BorshDeserialize, Clone, PartialEq, Debug, ProtocolSchema)]
pub struct BufferedReceiptIndices {
    pub shard_buffers: BTreeMap<ShardId, TrieQueueIndices>,
}

/// Map of shard to list of receipts to send to it.
pub type ReceiptResult = HashMap<ShardId, Vec<Receipt>>;

#[cfg(test)]
mod tests {
    use super::*;

    fn get_receipt_v0() -> Receipt {
        let receipt_v0 = Receipt::V0(ReceiptV0 {
            predecessor_id: "predecessor_id".parse().unwrap(),
            receiver_id: "receiver_id".parse().unwrap(),
            receipt_id: CryptoHash::default(),
            receipt: ReceiptEnum::Action(ActionReceipt {
                signer_id: "signer_id".parse().unwrap(),
                signer_public_key: PublicKey::empty(KeyType::ED25519),
                gas_price: Balance::ZERO,
                output_data_receivers: vec![],
                input_data_ids: vec![],
                actions: vec![Action::Transfer(TransferAction { deposit: Balance::ZERO })],
            }),
        });
        receipt_v0
    }

    #[test]
    fn test_receipt_v0_serialization() {
        let receipt_v0 = get_receipt_v0();
        let serialized_receipt = borsh::to_vec(&receipt_v0).unwrap();
        let receipt2 = Receipt::try_from_slice(&serialized_receipt).unwrap();
        assert_eq!(receipt_v0, receipt2);
    }

    #[test]
    fn test_state_stored_receipt_serialization() {
        let receipt = get_receipt_v0();
        let metadata =
            StateStoredReceiptMetadata { congestion_gas: Gas::from_gas(42), congestion_size: 43 };
        let receipt = StateStoredReceipt::new_owned(receipt, metadata);

        let serialized_receipt = borsh::to_vec(&receipt).unwrap();
        let deserialized_receipt = StateStoredReceipt::try_from_slice(&serialized_receipt).unwrap();

        assert_eq!(receipt, deserialized_receipt);
    }

    #[test]
    fn test_global_contract_distribution_receipt_v1_nonce() {
        let receipt = GlobalContractDistributionReceipt::new_v1(
            GlobalContractIdentifier::AccountId("test.near".parse().unwrap()),
            ShardId::new(0),
            vec![],
            vec![0u8; 10].into(),
        );
        // V1 receipts have no nonce, should return 0
        assert_eq!(receipt.nonce(), 0);
    }

    #[test]
    fn test_global_contract_distribution_receipt_v2_nonce() {
        let receipt = GlobalContractDistributionReceipt::new_v2(
            GlobalContractIdentifier::AccountId("test.near".parse().unwrap()),
            ShardId::new(0),
            vec![],
            vec![0u8; 10].into(),
            42,
        );
        assert_eq!(receipt.nonce(), 42);
    }

    #[test]
    fn test_global_contract_distribution_receipt_v2_serialization() {
        let receipt = GlobalContractDistributionReceipt::new_v2(
            GlobalContractIdentifier::AccountId("test.near".parse().unwrap()),
            ShardId::new(1),
            vec![ShardId::new(0)],
            vec![1u8, 2, 3].into(),
            100,
        );
        let serialized = borsh::to_vec(&receipt).unwrap();
        let deserialized = GlobalContractDistributionReceipt::try_from_slice(&serialized).unwrap();
        assert_eq!(receipt, deserialized);
        assert_eq!(deserialized.nonce(), 100);
        assert_eq!(deserialized.target_shard(), ShardId::new(1));
    }

    #[test]
    fn test_receipt_or_state_stored_receipt_serialization() {
        // Case 1:
        // Receipt V0 can be deserialized as ReceiptOrStateStoredReceipt
        {
            let receipt = get_receipt_v0();
            let receipt = Cow::Owned(receipt);

            let serialized_receipt = borsh::to_vec(&receipt).unwrap();
            let deserialized_receipt =
                ReceiptOrStateStoredReceipt::try_from_slice(&serialized_receipt).unwrap();

            assert_eq!(ReceiptOrStateStoredReceipt::Receipt(receipt), deserialized_receipt);
        }

        // Case 2:
        // StateStoredReceipt can be deserialized as ReceiptOrStateStoredReceipt
        {
            let receipt = get_receipt_v0();
            let metadata = StateStoredReceiptMetadata {
                congestion_gas: Gas::from_gas(42),
                congestion_size: 43,
            };
            let state_stored_receipt = StateStoredReceipt::new_owned(receipt, metadata);

            let serialized_receipt = borsh::to_vec(&state_stored_receipt).unwrap();
            let deserialized_receipt =
                ReceiptOrStateStoredReceipt::try_from_slice(&serialized_receipt).unwrap();

            assert_eq!(
                ReceiptOrStateStoredReceipt::StateStoredReceipt(state_stored_receipt),
                deserialized_receipt
            );
        }

        // Case 3:
        // ReceiptOrStateStoredReceipt::Receipt
        {
            let receipt = get_receipt_v0();
            let receipt = Cow::Owned(receipt);

            let receipt_or_state_stored_receipt = ReceiptOrStateStoredReceipt::Receipt(receipt);

            let serialized_receipt = borsh::to_vec(&receipt_or_state_stored_receipt).unwrap();
            let deserialized_receipt =
                ReceiptOrStateStoredReceipt::try_from_slice(&serialized_receipt).unwrap();

            assert_eq!(receipt_or_state_stored_receipt, deserialized_receipt);
        }

        // Case 4:
        // ReceiptOrStateStoredReceipt::StateStoredReceipt
        {
            let receipt = get_receipt_v0();
            let metadata = StateStoredReceiptMetadata {
                congestion_gas: Gas::from_gas(42),
                congestion_size: 43,
            };
            let state_stored_receipt = StateStoredReceipt::new_owned(receipt, metadata);
            let receipt_or_state_stored_receipt =
                ReceiptOrStateStoredReceipt::StateStoredReceipt(state_stored_receipt);

            let serialized_receipt = borsh::to_vec(&receipt_or_state_stored_receipt).unwrap();
            let deserialized_receipt =
                ReceiptOrStateStoredReceipt::try_from_slice(&serialized_receipt).unwrap();

            assert_eq!(receipt_or_state_stored_receipt, deserialized_receipt);
        }
    }

    #[test]
    fn receipt_to_tx_info_borsh_roundtrip_from_transaction() {
        let info = ReceiptToTxInfo::V1(ReceiptToTxInfoV1 {
            origin: ReceiptOrigin::FromTransaction(ReceiptOriginTransaction {
                tx_hash: CryptoHash::hash_bytes(b"tx"),
                sender_account_id: "alice.near".parse().unwrap(),
            }),
            receiver_account_id: "bob.near".parse().unwrap(),
            shard_id: ShardId::new(0),
        });
        let bytes = borsh::to_vec(&info).unwrap();
        let decoded: ReceiptToTxInfo = borsh::from_slice(&bytes).unwrap();
        assert_eq!(info, decoded);
    }

    #[test]
    fn receipt_to_tx_info_borsh_roundtrip_from_receipt() {
        let info = ReceiptToTxInfo::V1(ReceiptToTxInfoV1 {
            origin: ReceiptOrigin::FromReceipt(ReceiptOriginReceipt {
                parent_receipt_id: CryptoHash::hash_bytes(b"parent"),
                parent_predecessor_id: "alice.near".parse().unwrap(),
            }),
            receiver_account_id: "contract.near".parse().unwrap(),
            shard_id: ShardId::new(1),
        });
        let bytes = borsh::to_vec(&info).unwrap();
        let decoded: ReceiptToTxInfo = borsh::from_slice(&bytes).unwrap();
        assert_eq!(info, decoded);
    }
}

/// Source of a processed receipt, used to track how a receipt was applied.
///
/// Stored in `DBCol::ProcessedReceiptIds` (local-only, not part of consensus).
/// Adding new variants is safe for protocol but may cause deserialization
/// failures if the node rolls back to an older binary. Recovery via epoch sync.
#[derive(BorshSerialize, BorshDeserialize, Clone, Debug, PartialEq, Eq, ProtocolSchema)]
#[borsh(use_discriminant = true)]
#[repr(u8)]
pub enum ReceiptSource {
    Local = 0,
    Delayed = 1,
    Instant = 2,
    /// Marker for receipt IDs that only need their `ReceiptToTx` index entry
    /// garbage-collected. Used for receipts whose `ReceiptToTx` mapping was saved
    /// on the source shard but that don't appear in `OutcomeIds` (e.g. data
    /// receipts, PromiseResume, cross-shard receipts on a source-only node,
    /// GlobalContractDistribution receipts).
    ReceiptToTxGc = 3,
}

/// A processed receipt together with its source. Runtime-only struct, not serialized to DB.
#[derive(Debug)]
pub struct ProcessedReceipt {
    pub receipt: Receipt,
    pub source: ReceiptSource,
}

/// Lightweight metadata about a processed receipt, stored instead of the full receipt.
#[derive(BorshSerialize, BorshDeserialize, Clone, Debug, PartialEq, Eq, ProtocolSchema)]
#[borsh(use_discriminant = true)]
#[repr(u8)]
pub enum ProcessedReceiptMetadata {
    V0(ProcessedReceiptMetadataV0) = 0,
}

impl ProcessedReceiptMetadata {
    pub fn new(receipt_id: CryptoHash, source: ReceiptSource) -> Self {
        Self::V0(ProcessedReceiptMetadataV0 { receipt_id, source })
    }

    pub fn receipt_id(&self) -> &CryptoHash {
        match self {
            Self::V0(v0) => &v0.receipt_id,
        }
    }

    pub fn source(&self) -> &ReceiptSource {
        match self {
            Self::V0(v0) => &v0.source,
        }
    }
}

#[derive(BorshSerialize, BorshDeserialize, Clone, Debug, PartialEq, Eq, ProtocolSchema)]
pub struct ProcessedReceiptMetadataV0 {
    pub receipt_id: CryptoHash,
    pub source: ReceiptSource,
}

/// Describes the origin of a receipt: either created directly from a transaction,
/// or spawned as a child of another receipt.
#[derive(BorshSerialize, BorshDeserialize, Clone, Debug, PartialEq, Eq, ProtocolSchema)]
#[borsh(use_discriminant = true)]
#[repr(u8)]
pub enum ReceiptOrigin {
    FromTransaction(ReceiptOriginTransaction) = 0,
    FromReceipt(ReceiptOriginReceipt) = 1,
}

/// A receipt that was created directly from a signed transaction.
#[derive(BorshSerialize, BorshDeserialize, Clone, Debug, PartialEq, Eq, ProtocolSchema)]
pub struct ReceiptOriginTransaction {
    pub tx_hash: CryptoHash,
    pub sender_account_id: AccountId,
}

/// A receipt that was spawned as a child of another receipt.
#[derive(BorshSerialize, BorshDeserialize, Clone, Debug, PartialEq, Eq, ProtocolSchema)]
pub struct ReceiptOriginReceipt {
    pub parent_receipt_id: CryptoHash,
    pub parent_predecessor_id: AccountId,
}

/// Versioned mapping from receipt_id to its origin information.
/// Stored in `DBCol::ReceiptToTx` to enable reverse lookups from receipt to originating transaction.
#[derive(BorshSerialize, BorshDeserialize, Clone, Debug, PartialEq, Eq, ProtocolSchema)]
#[borsh(use_discriminant = true)]
#[repr(u8)]
pub enum ReceiptToTxInfo {
    V1(ReceiptToTxInfoV1) = 0,
}

#[derive(BorshSerialize, BorshDeserialize, Clone, Debug, PartialEq, Eq, ProtocolSchema)]
pub struct ReceiptToTxInfoV1 {
    pub origin: ReceiptOrigin,
    pub receiver_account_id: AccountId,
    pub shard_id: ShardId,
}