iota_types/

transaction.rs

// Copyright (c) 2021, Facebook, Inc. and its affiliates
// Copyright (c) Mysten Labs, Inc.
// Modifications Copyright (c) 2024 IOTA Stiftung
// SPDX-License-Identifier: Apache-2.0

// zkLogin/AuthenticatorStateUpdate types are kept (deprecated) for
// serialization compatibility only.

use std::{
    collections::{BTreeMap, BTreeSet, HashMap, HashSet},
    fmt::{Debug, Display, Formatter, Write},
    hash::Hash,
    iter::{self},
};

use anyhow::bail;
use fastcrypto::{encoding::Base64, hash::HashFunction};
use iota_protocol_config::ProtocolConfig;
use iota_sdk_types::{
    Address, Argument, CanceledTransaction, CertificateDigest, Command, ConsensusCommitDigest,
    ConsensusCommitPrologueV1, ConsensusDeterminedVersionAssignments, EndOfEpochTransactionKind,
    Event, GasPayment, GenesisObject, GenesisTransaction, Identifier, Input, MakeMoveVector,
    MergeCoins, MoveAuthenticator, MoveCall, MoveStruct, ObjectDigest, ObjectId, ObjectReference,
    Owner, ProgrammableTransaction, Publish, RandomnessRound, RandomnessStateUpdate,
    SenderSignedTransaction, SharedObjectReference, SplitCoins, Transaction,
    TransactionDenyRulesUpdate, TransactionDigest, TransactionExpiration, TransactionKind,
    TransactionV1, TransferObjects, TypeTag, Upgrade, UserSignature, Version,
    crypto::{Intent, IntentMessage, IntentScope, SimpleSignature},
};
use itertools::Either;
use nonempty::{NonEmpty, nonempty};
use serde::{Deserialize, Serialize};
use tap::Pipe;
use tracing::{instrument, trace};

use super::{base_types::*, error::*};
use crate::{
    IOTA_CLOCK_OBJECT_SHARED_VERSION, IOTA_SYSTEM_STATE_OBJECT_SHARED_VERSION,
    committee::{Committee, EpochId},
    crypto::{
        AuthoritySignInfo, AuthoritySignInfoTrait, AuthoritySignature,
        AuthorityStrongQuorumSignInfo, DefaultHash, EmptySignInfo, Signer, zero_ed25519_signature,
    },
    execution::SharedInput,
    message_envelope::{Envelope, Message, TrustedEnvelope, VerifiedEnvelope},
    messages_checkpoint::CheckpointTimestamp,
    move_authenticator::MoveAuthenticatorExt,
    object::Object,
    programmable_transaction_builder::ProgrammableTransactionBuilder,
    signature::VerifyParams,
    signature_verification::verify_sender_signed_data_message_signatures,
};

pub const TEST_ONLY_GAS_UNIT_FOR_TRANSFER: u64 = 10_000;
pub const TEST_ONLY_GAS_UNIT_FOR_OBJECT_BASICS: u64 = 50_000;
pub const TEST_ONLY_GAS_UNIT_FOR_PUBLISH: u64 = 50_000;
pub const TEST_ONLY_GAS_UNIT_FOR_STAKING: u64 = 50_000;
pub const TEST_ONLY_GAS_UNIT_FOR_GENERIC: u64 = 50_000;
pub const TEST_ONLY_GAS_UNIT_FOR_SPLIT_COIN: u64 = 10_000;
// For some transactions we may either perform heavy operations or touch
// objects that are storage expensive. That may happen (and often is the case)
// because the object touched are set up in genesis and carry no storage cost
// (and thus rebate) on first usage.
pub const TEST_ONLY_GAS_UNIT_FOR_HEAVY_COMPUTATION_STORAGE: u64 = 5_000_000;

pub const GAS_PRICE_FOR_SYSTEM_TX: u64 = 1;

pub const DEFAULT_VALIDATOR_GAS_PRICE: u64 = 1000;

/// The most inputs a programmable transaction may declare.
pub const MAX_PROGRAMMABLE_TX_INPUTS: usize = u16::MAX as usize;

const BLOCKED_MOVE_FUNCTIONS: [(ObjectId, &str, &str); 0] = [];

#[cfg(test)]
#[path = "unit_tests/messages_tests.rs"]
mod messages_tests;

/// Type alias for the SDK's `Input` type, used as transaction call arguments.
pub type CallArg = Input;

/// Rejects a version that a transaction names for an input object when it is
/// at or above `Version::MAX_VALID_EXCL`, the range assigned to the objects of
/// canceled transactions, or right below it. The version of a transaction's
/// outputs is one more than its largest input version, and version assignment
/// halts the node when that result is not a valid version, so both are refused
/// from the transaction bytes, before any object is loaded.
fn input_object_version_validity_check(version: Version) -> UserInputResult {
    fp_ensure!(
        version.next().is_ok_and(|next| next.is_valid()),
        UserInputError::InvalidSequenceNumber
    );

    Ok(())
}

pub fn type_tag_validity_check(
    tag: &TypeTag,
    config: &ProtocolConfig,
    starting_count: &mut usize,
) -> UserInputResult<()> {
    let mut stack = vec![(tag, 1)];
    while let Some((tag, depth)) = stack.pop() {
        *starting_count += 1;
        fp_ensure!(
            *starting_count < config.max_type_arguments() as usize,
            UserInputError::SizeLimitExceeded {
                limit: "maximum type arguments in a call transaction".to_string(),
                value: config.max_type_arguments().to_string()
            }
        );
        fp_ensure!(
            depth < config.max_type_argument_depth(),
            UserInputError::SizeLimitExceeded {
                limit: "maximum type argument depth in a call transaction".to_string(),
                value: config.max_type_argument_depth().to_string()
            }
        );
        match tag {
            TypeTag::Bool
            | TypeTag::U8
            | TypeTag::U64
            | TypeTag::U128
            | TypeTag::Address
            | TypeTag::Signer
            | TypeTag::U16
            | TypeTag::U32
            | TypeTag::U256 => (),
            TypeTag::Vector(t) => {
                stack.push((t, depth + 1));
            }
            TypeTag::Struct(s) => {
                let next_depth = depth + 1;
                if config.validate_identifier_inputs() {
                    fp_ensure!(
                        Identifier::is_valid(s.module().as_str()),
                        UserInputError::InvalidIdentifier {
                            error: s.module().as_str().to_owned()
                        }
                    );
                    fp_ensure!(
                        Identifier::is_valid(s.name().as_str()),
                        UserInputError::InvalidIdentifier {
                            error: s.name().as_str().to_owned()
                        }
                    );
                }
                stack.extend(s.type_params().iter().map(|t| (t, next_depth)));
            }
        }
    }
    Ok(())
}

/// Extension trait for [`EndOfEpochTransactionKind`] that adds methods
/// requiring iota-types-specific types (like [`InputObjectKind`] and
/// [`ProtocolConfig`]) that are not available in the SDK.
pub(crate) trait EndOfEpochTransactionKindExt {
    fn input_objects(&self) -> Vec<InputObjectKind>;
    fn validity_check(&self, config: &ProtocolConfig) -> UserInputResult;
}

impl EndOfEpochTransactionKindExt for EndOfEpochTransactionKind {
    fn input_objects(&self) -> Vec<InputObjectKind> {
        match self {
            Self::ChangeEpoch(_)
            | Self::ChangeEpochV2(_)
            | Self::ChangeEpochV3(_)
            | Self::ChangeEpochV4(_) => {
                vec![InputObjectKind::SharedMoveObject {
                    id: ObjectId::SYSTEM_STATE,
                    initial_shared_version: IOTA_SYSTEM_STATE_OBJECT_SHARED_VERSION,
                    mutable: true,
                }]
            }
            // Creates the TransactionDenyRules object; there is no input to
            // reference yet.
            Self::TransactionDenyRulesCreate => vec![],
            _ => unimplemented!(
                "a new EndOfEpochTransactionKind enum variant was added and needs to be handled"
            ),
        }
    }

    fn validity_check(&self, config: &ProtocolConfig) -> UserInputResult {
        match self {
            Self::ChangeEpoch(_) => {
                if config.protocol_defined_base_fee() {
                    return Err(UserInputError::Unsupported(
                        "protocol defined base fee not supported".to_string(),
                    ));
                }
                if config.select_committee_from_eligible_validators() {
                    return Err(UserInputError::Unsupported(
                        "selecting committee only among validators supporting the protocol version not supported".to_string(),
                    ));
                }
                if config.pass_validator_scores_to_advance_epoch() {
                    return Err(UserInputError::Unsupported(
                        "passing of validator scores not supported".to_string(),
                    ));
                }
                if config.adjust_rewards_by_score() {
                    return Err(UserInputError::Unsupported(
                        "adjusting rewards by score not supported".to_string(),
                    ));
                }
            }
            Self::ChangeEpochV2(_) => {
                if !config.protocol_defined_base_fee() {
                    return Err(UserInputError::Unsupported(
                        "protocol defined base fee required".to_string(),
                    ));
                }
                if config.select_committee_from_eligible_validators() {
                    return Err(UserInputError::Unsupported(
                        "selecting committee only among validators supporting the protocol version not supported".to_string(),
                    ));
                }
                if config.pass_validator_scores_to_advance_epoch() {
                    return Err(UserInputError::Unsupported(
                        "passing of validator scores not supported".to_string(),
                    ));
                }
                if config.adjust_rewards_by_score() {
                    return Err(UserInputError::Unsupported(
                        "adjusting rewards by score not supported".to_string(),
                    ));
                }
            }
            Self::ChangeEpochV3(_) => {
                if !config.protocol_defined_base_fee() {
                    return Err(UserInputError::Unsupported(
                        "protocol defined base fee required".to_string(),
                    ));
                }
                if !config.select_committee_from_eligible_validators() {
                    return Err(UserInputError::Unsupported(
                        "selecting committee only among validators supporting the protocol version required".to_string(),
                    ));
                }
                if config.pass_validator_scores_to_advance_epoch() {
                    return Err(UserInputError::Unsupported(
                        "passing of validator scores not supported".to_string(),
                    ));
                }
                if config.adjust_rewards_by_score() {
                    return Err(UserInputError::Unsupported(
                        "adjusting rewards by score not supported".to_string(),
                    ));
                }
            }
            Self::ChangeEpochV4(_) => {
                if !config.protocol_defined_base_fee() {
                    return Err(UserInputError::Unsupported(
                        "protocol defined base fee required".to_string(),
                    ));
                }
                if !config.select_committee_from_eligible_validators() {
                    return Err(UserInputError::Unsupported(
                        "selecting committee only among validators supporting the protocol version required".to_string(),
                    ));
                }
                if !config.pass_validator_scores_to_advance_epoch() {
                    return Err(UserInputError::Unsupported(
                        "passing of validator scores required".to_string(),
                    ));
                }
            }
            Self::TransactionDenyRulesCreate => {
                if !config.enable_pcool_flow() {
                    return Err(UserInputError::Unsupported(
                        "on-chain deny rule governance not supported at current protocol version"
                            .to_string(),
                    ));
                }
            }
            _ => unimplemented!(
                "a new EndOfEpochTransactionKind enum variant was added and needs to be handled"
            ),
        }
        Ok(())
    }
}

mod call_arg_ext {
    pub trait Sealed {}
    impl Sealed for super::CallArg {}
}

/// Extension trait for [`CallArg`] providing helper methods.
pub trait CallArgExt: Sized + call_arg_ext::Sealed {
    /// Returns the input object kind for this argument, excluding receiving
    /// objects.
    fn input_object_kind(&self) -> Option<InputObjectKind>;

    /// Validity check for this argument against the given protocol config.
    fn validity_check(&self, config: &ProtocolConfig) -> UserInputResult;
}

impl CallArgExt for CallArg {
    fn input_object_kind(&self) -> Option<InputObjectKind> {
        match self {
            CallArg::ImmutableOrOwned(object_ref) => {
                Some(InputObjectKind::ImmOrOwnedMoveObject(*object_ref))
            }
            CallArg::Shared(SharedObjectReference {
                object_id,
                initial_shared_version,
                mutable,
            }) => Some(InputObjectKind::SharedMoveObject {
                id: *object_id,
                initial_shared_version: *initial_shared_version,
                mutable: *mutable,
            }),
            CallArg::Pure(_) | CallArg::Receiving(_) => None,
            _ => unimplemented!("a new CallArg enum variant was added and needs to be handled"),
        }
    }

    fn validity_check(&self, config: &ProtocolConfig) -> UserInputResult {
        match self {
            CallArg::Pure(bytes) => {
                fp_ensure!(
                    bytes.len() < config.max_pure_argument_size() as usize,
                    UserInputError::SizeLimitExceeded {
                        limit: "maximum pure argument size".to_string(),
                        value: config.max_pure_argument_size().to_string()
                    }
                );
            }
            CallArg::ImmutableOrOwned(ObjectReference { version, .. })
            | CallArg::Receiving(ObjectReference { version, .. })
            | CallArg::Shared(SharedObjectReference {
                initial_shared_version: version,
                ..
            }) => {
                if config.validate_input_object_versions() {
                    input_object_version_validity_check(*version)?;
                }
            }
            _ => unimplemented!("a new CallArg enum variant was added and needs to be handled"),
        }
        Ok(())
    }
}

// Add package IDs, `ObjectId`, for types defined in modules.
fn add_type_tag_packages(packages: &mut BTreeSet<ObjectId>, type_argument: &TypeTag) {
    let mut stack = vec![type_argument];
    while let Some(cur) = stack.pop() {
        match cur {
            TypeTag::U8
            | TypeTag::U16
            | TypeTag::U32
            | TypeTag::U64
            | TypeTag::U128
            | TypeTag::U256
            | TypeTag::Bool
            | TypeTag::Address
            | TypeTag::Signer => (),
            TypeTag::Vector(inner) => stack.push(inner),
            TypeTag::Struct(struct_tag) => {
                packages.insert(ObjectId::new(struct_tag.address().into_bytes()));
                stack.extend(struct_tag.type_params().iter())
            }
        }
    }
}

mod move_call_ext {
    pub trait Sealed {}
    impl Sealed for super::MoveCall {}
}

pub trait MoveCallExt: Sized + move_call_ext::Sealed {
    fn input_objects(&self) -> Vec<InputObjectKind>;
    fn validity_check(&self, config: &ProtocolConfig) -> UserInputResult;
    fn is_input_arg_used(&self, arg: usize) -> bool;
}

impl MoveCallExt for MoveCall {
    fn input_objects(&self) -> Vec<InputObjectKind> {
        let mut packages = BTreeSet::from([self.package]);
        for type_argument in &self.type_arguments {
            add_type_tag_packages(&mut packages, type_argument);
        }
        packages
            .into_iter()
            .map(InputObjectKind::MovePackage)
            .collect()
    }

    fn validity_check(&self, config: &ProtocolConfig) -> UserInputResult {
        let is_blocked = BLOCKED_MOVE_FUNCTIONS.contains(&(
            self.package,
            self.module.as_str(),
            self.function.as_str(),
        ));
        fp_ensure!(!is_blocked, UserInputError::BlockedMoveFunction);
        let mut type_arguments_count = 0;
        for tag in &self.type_arguments {
            type_tag_validity_check(tag, config, &mut type_arguments_count)?;
        }
        fp_ensure!(
            self.arguments.len() < config.max_arguments() as usize,
            UserInputError::SizeLimitExceeded {
                limit: "maximum arguments in a move call".to_string(),
                value: config.max_arguments().to_string()
            }
        );
        if config.validate_identifier_inputs() {
            fp_ensure!(
                Identifier::is_valid(&self.module),
                UserInputError::InvalidIdentifier {
                    error: self.module.to_string()
                }
            );
            fp_ensure!(
                Identifier::is_valid(&self.function),
                UserInputError::InvalidIdentifier {
                    error: self.function.to_string()
                }
            );
        }
        Ok(())
    }

    fn is_input_arg_used(&self, arg: usize) -> bool {
        self.arguments
            .iter()
            .any(|a| matches!(a, Argument::Input(inp) if usize::from(*inp) == arg))
    }
}

mod command_ext {
    pub trait Sealed {}
    impl Sealed for super::Command {}
}

pub trait CommandExt: Sized + command_ext::Sealed {
    fn input_objects(&self) -> Vec<InputObjectKind>;
    fn validity_check(&self, config: &ProtocolConfig) -> UserInputResult;
    fn non_system_packages_to_be_published(&self) -> Option<&Vec<Vec<u8>>>;
    fn is_input_arg_used(&self, input_arg: usize) -> bool;
}

impl CommandExt for Command {
    fn input_objects(&self) -> Vec<InputObjectKind> {
        match self {
            Command::MoveCall(cmd) => cmd.input_objects(),
            Command::Upgrade(cmd) => cmd
                .dependencies
                .iter()
                .map(|id| InputObjectKind::MovePackage(*id))
                .chain(Some(InputObjectKind::MovePackage(cmd.package)))
                .collect(),
            Command::Publish(cmd) => cmd
                .dependencies
                .iter()
                .map(|id| InputObjectKind::MovePackage(*id))
                .collect(),
            Command::MakeMoveVector(MakeMoveVector {
                type_tag: Some(t), ..
            }) => {
                let mut packages = BTreeSet::new();
                add_type_tag_packages(&mut packages, t);
                packages
                    .into_iter()
                    .map(InputObjectKind::MovePackage)
                    .collect()
            }
            Command::MakeMoveVector(MakeMoveVector { type_tag: None, .. })
            | Command::TransferObjects(_)
            | Command::SplitCoins(_)
            | Command::MergeCoins(_) => vec![],
            _ => unimplemented!("a new Command enum variant was added and needs to be handled"),
        }
    }

    fn validity_check(&self, config: &ProtocolConfig) -> UserInputResult {
        match self {
            Command::MoveCall(call) => call.validity_check(config)?,
            Command::TransferObjects(TransferObjects { objects: args, .. })
            | Command::MergeCoins(MergeCoins {
                coins_to_merge: args,
                ..
            })
            | Command::SplitCoins(SplitCoins { amounts: args, .. }) => {
                fp_ensure!(!args.is_empty(), UserInputError::EmptyCommandInput);
                fp_ensure!(
                    args.len() < config.max_arguments() as usize,
                    UserInputError::SizeLimitExceeded {
                        limit: "maximum arguments in a programmable transaction command"
                            .to_string(),
                        value: config.max_arguments().to_string()
                    }
                );
            }
            Command::MakeMoveVector(MakeMoveVector {
                type_tag: ty_opt,
                elements: args,
            }) => {
                // ty_opt.is_none() ==> !args.is_empty()
                fp_ensure!(
                    ty_opt.is_some() || !args.is_empty(),
                    UserInputError::EmptyCommandInput
                );
                if let Some(ty) = ty_opt {
                    let mut type_arguments_count = 0;
                    type_tag_validity_check(ty, config, &mut type_arguments_count)?;
                }
                fp_ensure!(
                    args.len() < config.max_arguments() as usize,
                    UserInputError::SizeLimitExceeded {
                        limit: "maximum arguments in a programmable transaction command"
                            .to_string(),
                        value: config.max_arguments().to_string()
                    }
                );
            }
            Command::Publish(Publish {
                modules,
                dependencies,
            })
            | Command::Upgrade(Upgrade {
                modules,
                dependencies,
                ..
            }) => {
                fp_ensure!(!modules.is_empty(), UserInputError::EmptyCommandInput);
                fp_ensure!(
                    modules.len() < config.max_modules_in_publish() as usize,
                    UserInputError::SizeLimitExceeded {
                        limit: "maximum modules in a programmable transaction upgrade command"
                            .to_string(),
                        value: config.max_modules_in_publish().to_string()
                    }
                );
                if let Some(max_package_dependencies) = config.max_package_dependencies_as_option()
                {
                    fp_ensure!(
                        dependencies.len() < max_package_dependencies as usize,
                        UserInputError::SizeLimitExceeded {
                            limit: "maximum package dependencies".to_string(),
                            value: max_package_dependencies.to_string()
                        }
                    );
                };
            }
            _ => unimplemented!("a new Command enum variant was added and needs to be handled"),
        };

        Ok(())
    }

    fn non_system_packages_to_be_published(&self) -> Option<&Vec<Vec<u8>>> {
        match self {
            Command::Publish(cmd) => Some(&cmd.modules),
            Command::Upgrade(cmd) => Some(&cmd.modules),
            Command::MoveCall(_)
            | Command::TransferObjects(_)
            | Command::SplitCoins(_)
            | Command::MergeCoins(_)
            | Command::MakeMoveVector(_) => None,
            _ => unimplemented!("a new Command enum variant was added and needs to be handled"),
        }
    }

    fn is_input_arg_used(&self, input_arg: usize) -> bool {
        match self {
            Command::MoveCall(c) => c.is_input_arg_used(input_arg),
            Command::TransferObjects(TransferObjects {
                objects: args,
                address: arg,
            })
            | Command::MergeCoins(MergeCoins {
                coins_to_merge: args,
                coin: arg,
            })
            | Command::SplitCoins(SplitCoins {
                amounts: args,
                coin: arg,
            }) => args.iter().chain(iter::once(arg)).any(
                |arg| matches!(arg, Argument::Input(input) if usize::from(*input) == input_arg),
            ),
            Command::MakeMoveVector(MakeMoveVector { elements, .. }) => elements.iter().any(
                |arg| matches!(arg, Argument::Input(input) if usize::from(*input) == input_arg),
            ),
            Command::Upgrade(Upgrade { ticket, .. }) => {
                matches!(ticket, Argument::Input(input) if usize::from(*input) == input_arg)
            }
            Command::Publish(_) => false,
            _ => unimplemented!("a new Command enum variant was added and needs to be handled"),
        }
    }
}

/// Iterates over arguments in command-argument order.
fn command_arguments(command: &Command) -> Box<dyn Iterator<Item = &Argument> + '_> {
    match command {
        Command::MoveCall(c) => Box::new(c.arguments.iter()),
        Command::TransferObjects(TransferObjects { objects, address }) => {
            Box::new(objects.iter().chain(iter::once(address)))
        }
        Command::SplitCoins(SplitCoins { coin, amounts }) => {
            Box::new(iter::once(coin).chain(amounts))
        }
        Command::MergeCoins(MergeCoins {
            coin,
            coins_to_merge,
        }) => Box::new(iter::once(coin).chain(coins_to_merge)),
        Command::MakeMoveVector(MakeMoveVector { elements, .. }) => Box::new(elements.iter()),
        Command::Upgrade(Upgrade { ticket, .. }) => Box::new(iter::once(ticket)),
        Command::Publish(_) => Box::new(iter::empty()),
        _ => unimplemented!("a new Command enum variant was added and needs to be handled"),
    }
}

mod programmable_transaction_ext {
    pub trait Sealed {}
    impl Sealed for super::ProgrammableTransaction {}
}

pub trait ProgrammableTransactionExt: Sized + programmable_transaction_ext::Sealed {
    fn input_objects(&self) -> UserInputResult<Vec<InputObjectKind>>;
    fn receiving_objects(&self) -> Vec<ObjectReference>;
    fn validity_check(&self, config: &ProtocolConfig) -> UserInputResult;
    fn validate_argument_indices(&self) -> UserInputResult;
    fn shared_input_objects(&self) -> impl Iterator<Item = SharedObjectReference>;
    fn move_calls(&self) -> Vec<(&ObjectId, &str, &str)>;
    fn non_system_packages_to_be_published(&self) -> impl Iterator<Item = &Vec<Vec<u8>>>;
}

impl ProgrammableTransactionExt for ProgrammableTransaction {
    fn input_objects(&self) -> UserInputResult<Vec<InputObjectKind>> {
        let ProgrammableTransaction { inputs, commands } = self;
        let input_arg_objects = inputs
            .iter()
            .filter_map(|arg| arg.input_object_kind())
            .collect::<Vec<_>>();
        // all objects, not just mutable, must be unique
        let mut used = HashSet::new();
        if !input_arg_objects.iter().all(|o| used.insert(o.object_id())) {
            return Err(UserInputError::DuplicateObjectRefInput);
        }
        // do not duplicate packages referred to in commands
        let command_input_objects: BTreeSet<InputObjectKind> = commands
            .iter()
            .flat_map(|command| command.input_objects())
            .collect();
        Ok(input_arg_objects
            .into_iter()
            .chain(command_input_objects)
            .collect())
    }

    fn receiving_objects(&self) -> Vec<ObjectReference> {
        let ProgrammableTransaction { inputs, .. } = self;
        inputs
            .iter()
            .filter_map(|arg| arg.as_opt_receiving().copied())
            .collect()
    }

    fn validity_check(&self, config: &ProtocolConfig) -> UserInputResult {
        let ProgrammableTransaction { inputs, commands } = self;
        fp_ensure!(
            commands.len() < config.max_programmable_tx_commands() as usize,
            UserInputError::SizeLimitExceeded {
                limit: "maximum commands in a programmable transaction".to_string(),
                value: config.max_programmable_tx_commands().to_string()
            }
        );
        // `max_input_objects` below does not count pure inputs, so it does not
        // bound the list.
        fp_ensure!(
            inputs.len() <= MAX_PROGRAMMABLE_TX_INPUTS,
            UserInputError::SizeLimitExceeded {
                limit: "maximum inputs in a programmable transaction".to_string(),
                value: MAX_PROGRAMMABLE_TX_INPUTS.to_string(),
            }
        );
        let total_inputs = self.input_objects()?.len() + self.receiving_objects().len();
        fp_ensure!(
            total_inputs <= config.max_input_objects() as usize,
            UserInputError::SizeLimitExceeded {
                limit: "maximum input + receiving objects in a transaction".to_string(),
                value: config.max_input_objects().to_string()
            }
        );
        for input in inputs {
            input.validity_check(config)?
        }
        if let Some(max_publish_commands) = config.max_publish_or_upgrade_per_ptb_as_option() {
            let publish_count = commands
                .iter()
                .filter(|c| c.is_publish() || c.is_upgrade())
                .count() as u64;
            fp_ensure!(
                publish_count <= max_publish_commands,
                UserInputError::MaxPublishCountExceeded {
                    max_publish_commands,
                    publish_count,
                }
            );
        }
        for command in commands {
            command.validity_check(config)?;
        }
        if config.validate_ptb_argument_indices() {
            self.validate_argument_indices()?;
        }

        // If randomness is used, it must be enabled by protocol config.
        // A command that uses Random can only be followed by TransferObjects or
        // MergeCoins.
        if let Some(random_index) = inputs.iter().position(|obj| {
            matches!(obj, CallArg::Shared(SharedObjectReference { object_id, .. }) if *object_id == ObjectId::RANDOMNESS_STATE)
        }) {
            let mut used_random_object = false;
            for command in commands {
                if !used_random_object {
                    used_random_object = command.is_input_arg_used(random_index);
                } else {
                    fp_ensure!(
                        command.is_transfer_objects() || command.is_merge_coins(),
                        UserInputError::PostRandomCommandRestrictions
                    );
                }
            }
        }

        Ok(())
    }

    fn validate_argument_indices(&self) -> UserInputResult {
        for (command_idx, command) in self.commands.iter().enumerate() {
            for (argument_idx, argument) in command_arguments(command).enumerate() {
                let index = match argument {
                    Argument::Input(index) if *index as usize >= self.inputs.len() => *index,
                    Argument::Result(index) | Argument::NestedResult(index, _)
                        if *index as usize >= command_idx =>
                    {
                        *index
                    }
                    Argument::Gas
                    | Argument::Input(_)
                    | Argument::Result(_)
                    | Argument::NestedResult(_, _) => continue,
                    _ => unimplemented!(
                        "a new Argument enum variant was added and needs to be handled"
                    ),
                };
                return Err(UserInputError::InvalidArgumentIndex {
                    command_idx,
                    argument_idx,
                    index,
                });
            }
        }
        Ok(())
    }

    fn shared_input_objects(&self) -> impl Iterator<Item = SharedObjectReference> {
        self.inputs.iter().filter_map(|arg| match arg {
            CallArg::Shared(shared) => Some(*shared),
            CallArg::Pure(_) | CallArg::Receiving(_) | CallArg::ImmutableOrOwned(_) => None,
            _ => unimplemented!("a new CallArg enum variant was added and needs to be handled"),
        })
    }

    fn move_calls(&self) -> Vec<(&ObjectId, &str, &str)> {
        self.commands
            .iter()
            .filter_map(|command| match command {
                Command::MoveCall(m) => Some((&m.package, m.module.as_str(), m.function.as_str())),
                _ => None,
            })
            .collect()
    }

    fn non_system_packages_to_be_published(&self) -> impl Iterator<Item = &Vec<Vec<u8>>> {
        self.commands
            .iter()
            .filter_map(|q| q.non_system_packages_to_be_published())
    }
}

/// Merges `other` into `this` shared input object.
/// If there is a conflict in mutability, the resulting object will be
/// mutable. Errors if the id or initial_shared_version do not match.
fn left_union_shared_input_objects(
    this: &mut SharedObjectReference,
    other: &SharedObjectReference,
) -> UserInputResult<()> {
    fp_ensure!(
        this.object_id == other.object_id,
        UserInputError::SharedObjectIdMismatch
    );
    fp_ensure!(
        this.initial_shared_version == other.initial_shared_version,
        UserInputError::SharedObjectStartingVersionMismatch
    );

    if !this.mutable && other.mutable {
        this.mutable = other.mutable;
    }

    Ok(())
}

mod transaction_kind_ext {
    pub trait Sealed {}
    impl Sealed for super::TransactionKind {}
}

pub trait TransactionKindExt: Sized + transaction_kind_ext::Sealed {
    /// If this is an advance epoch transaction, returns (total gas charged,
    /// total gas rebated). TODO: We should use `GasCostSummary` directly in
    /// `ChangeEpoch` struct, and return that directly.
    fn get_advance_epoch_tx_gas_summary(&self) -> Option<(u64, u64)>;
    /// Returns `true` if the transaction contains at least one shared object.
    fn contains_shared_object(&self) -> bool;
    /// Returns an iterator of all shared input objects used by this
    /// transaction.
    fn shared_input_objects(&self) -> impl Iterator<Item = SharedObjectReference> + '_;
    /// Returns the move calls made by this transaction as a list of
    /// (package, module, function) tuples.
    fn move_calls(&self) -> Vec<(&ObjectId, &str, &str)>;
    /// Returns the objects received by this transaction.
    fn receiving_objects(&self) -> Vec<ObjectReference>;
    /// Return the metadata of each of the input objects for the transaction.
    /// For a Move object, we attach the object reference;
    /// for a Move package, we provide the object id only since they never
    /// change on chain. TODO: use an iterator over references here instead
    /// of a `Vec` to avoid allocations.
    fn input_objects(&self) -> UserInputResult<Vec<InputObjectKind>>;
    /// Validates the transaction against the given protocol config.
    fn validity_check(&self, config: &ProtocolConfig) -> UserInputResult;
    /// Returns an iterator over the commands in this transaction.
    fn iter_commands(&self) -> impl Iterator<Item = &Command>;
    /// Returns a human-readable name for this transaction kind.
    fn name(&self) -> &'static str;
}

impl TransactionKindExt for TransactionKind {
    fn get_advance_epoch_tx_gas_summary(&self) -> Option<(u64, u64)> {
        match self {
            Self::EndOfEpoch(txns) => {
                match txns.last().expect("at least one end-of-epoch txn required") {
                    EndOfEpochTransactionKind::ChangeEpoch(e) => {
                        Some((e.computation_charge + e.storage_charge, e.storage_rebate))
                    }
                    EndOfEpochTransactionKind::ChangeEpochV2(e) => {
                        Some((e.computation_charge + e.storage_charge, e.storage_rebate))
                    }
                    EndOfEpochTransactionKind::ChangeEpochV3(e) => {
                        Some((e.computation_charge + e.storage_charge, e.storage_rebate))
                    }
                    EndOfEpochTransactionKind::ChangeEpochV4(e) => {
                        Some((e.computation_charge + e.storage_charge, e.storage_rebate))
                    }
                    // Never the last end-of-epoch kind; a change-epoch kind
                    // always follows it.
                    EndOfEpochTransactionKind::TransactionDenyRulesCreate => None,
                    _ => unimplemented!(
                        "a new EndOfEpochTransactionKind enum variant was added and needs to be handled"
                    ),
                }
            }
            _ => None,
        }
    }

    fn contains_shared_object(&self) -> bool {
        self.shared_input_objects().next().is_some()
    }

    fn shared_input_objects(&self) -> impl Iterator<Item = SharedObjectReference> + '_ {
        match &self {
            Self::ConsensusCommitPrologueV1(_) => Either::Left(Either::Left(iter::once(
                SharedObjectReference::new(ObjectId::CLOCK, IOTA_CLOCK_OBJECT_SHARED_VERSION, true),
            ))),
            #[allow(deprecated)]
            Self::AuthenticatorStateUpdateV1Deprecated => {
                // Deprecated: Authenticator state (JWK) is deprecated and
                // was never enabled. These transaction kinds are retained
                // only for BCS enum variant compatibility.
                Either::Right(Either::Right(iter::empty()))
            }
            Self::RandomnessStateUpdate(update) => {
                Either::Left(Either::Left(iter::once(SharedObjectReference::new(
                    ObjectId::RANDOMNESS_STATE,
                    update.randomness_obj_initial_shared_version,
                    true,
                ))))
            }
            Self::TransactionDenyRulesUpdate(update) => {
                Either::Left(Either::Left(iter::once(SharedObjectReference::new(
                    ObjectId::TRANSACTION_DENY_RULES,
                    update.deny_rules_obj_initial_shared_version,
                    true,
                ))))
            }
            Self::EndOfEpoch(txns) => Either::Left(Either::Right(
                txns.iter().flat_map(|txn| txn.shared_input_objects()),
            )),
            Self::Programmable(pt) => Either::Right(Either::Left(pt.shared_input_objects())),
            _ => Either::Right(Either::Right(iter::empty())),
        }
    }

    fn move_calls(&self) -> Vec<(&ObjectId, &str, &str)> {
        match &self {
            Self::Programmable(pt) => pt.move_calls(),
            _ => vec![],
        }
    }

    fn receiving_objects(&self) -> Vec<ObjectReference> {
        match &self {
            #[allow(deprecated)]
            TransactionKind::Genesis(_)
            | TransactionKind::ConsensusCommitPrologueV1(_)
            | TransactionKind::AuthenticatorStateUpdateV1Deprecated
            | TransactionKind::RandomnessStateUpdate(_)
            | TransactionKind::TransactionDenyRulesUpdate(_)
            | TransactionKind::EndOfEpoch(_) => vec![],
            TransactionKind::Programmable(pt) => pt.receiving_objects(),
            _ => unimplemented!(
                "a new TransactionKind enum variant was added and needs to be handled"
            ),
        }
    }

    fn input_objects(&self) -> UserInputResult<Vec<InputObjectKind>> {
        let input_objects = match &self {
            Self::Genesis(_) => {
                vec![]
            }
            Self::ConsensusCommitPrologueV1(_) => {
                vec![InputObjectKind::SharedMoveObject {
                    id: ObjectId::CLOCK,
                    initial_shared_version: IOTA_CLOCK_OBJECT_SHARED_VERSION,
                    mutable: true,
                }]
            }
            #[allow(deprecated)]
            Self::AuthenticatorStateUpdateV1Deprecated => {
                // Deprecated: Authenticator state (JWK) is deprecated and
                // was never enabled. These transaction kinds are retained
                // only for BCS enum variant compatibility.
                vec![]
            }
            Self::RandomnessStateUpdate(update) => {
                vec![InputObjectKind::SharedMoveObject {
                    id: ObjectId::RANDOMNESS_STATE,
                    initial_shared_version: update.randomness_obj_initial_shared_version,
                    mutable: true,
                }]
            }
            Self::TransactionDenyRulesUpdate(update) => {
                vec![InputObjectKind::SharedMoveObject {
                    id: ObjectId::TRANSACTION_DENY_RULES,
                    initial_shared_version: update.deny_rules_obj_initial_shared_version,
                    mutable: true,
                }]
            }
            Self::EndOfEpoch(txns) => {
                // Dedup since transactions may have an overlap in input objects.
                // Note: it's critical to ensure the order of inputs are deterministic.
                let before_dedup: Vec<_> =
                    txns.iter().flat_map(|txn| txn.input_objects()).collect();
                let mut has_seen = HashSet::new();
                let mut after_dedup = vec![];
                for obj in before_dedup {
                    if has_seen.insert(obj) {
                        after_dedup.push(obj);
                    }
                }
                after_dedup
            }
            Self::Programmable(p) => return p.input_objects(),
            _ => unimplemented!(
                "a new TransactionKind enum variant was added and needs to be handled"
            ),
        };
        // Ensure that there are no duplicate inputs. This cannot be removed because:
        // In [`AuthorityState::check_locks`], we check that there are no duplicate
        // mutable input objects, which would have made this check here
        // unnecessary. However, we do plan to allow shared objects show up more
        // than once in multiple single transactions down the line. Once we have
        // that, we need check here to make sure the same shared object doesn't
        // show up more than once in the same single transaction.
        let mut used = HashSet::new();
        if !input_objects.iter().all(|o| used.insert(o.object_id())) {
            return Err(UserInputError::DuplicateObjectRefInput);
        }
        Ok(input_objects)
    }

    fn validity_check(&self, config: &ProtocolConfig) -> UserInputResult {
        match self {
            TransactionKind::Programmable(p) => p.validity_check(config)?,
            // All transaction kinds below are assumed to be system,
            // and no validity or limit checks are performed.
            TransactionKind::Genesis(_) | TransactionKind::ConsensusCommitPrologueV1(_) => (),
            TransactionKind::EndOfEpoch(txns) => {
                for tx in txns {
                    tx.validity_check(config)?;
                }
            }

            #[allow(deprecated)]
            TransactionKind::AuthenticatorStateUpdateV1Deprecated => {
                // Deprecated: Authenticator state (JWK) is deprecated and
                // was never enabled. These transaction kinds are retained
                // only for BCS enum variant compatibility.
                return Err(UserInputError::Unsupported(
                    "authenticator state transactions are deprecated and were never created on IOTA"
                        .to_string(),
                ));
            }
            TransactionKind::RandomnessStateUpdate(_) => (),
            TransactionKind::TransactionDenyRulesUpdate(_) => {
                if !config.enable_pcool_flow() {
                    return Err(UserInputError::Unsupported(
                        "on-chain deny rule governance not supported at current protocol version"
                            .to_string(),
                    ));
                }
            }
            _ => unimplemented!(
                "a new TransactionKind enum variant was added and needs to be handled"
            ),
        };
        Ok(())
    }

    fn iter_commands(&self) -> impl Iterator<Item = &Command> {
        match self {
            TransactionKind::Programmable(pt) => pt.commands.iter(),
            _ => [].iter(),
        }
    }

    fn name(&self) -> &'static str {
        match self {
            Self::Genesis(_) => "Genesis",
            Self::ConsensusCommitPrologueV1(_) => "ConsensusCommitPrologueV1",
            Self::Programmable(_) => "Programmable",
            #[allow(deprecated)]
            Self::AuthenticatorStateUpdateV1Deprecated => "AuthenticatorStateUpdateV1Deprecated",
            Self::RandomnessStateUpdate(_) => "RandomnessStateUpdate",
            Self::TransactionDenyRulesUpdate(_) => "TransactionDenyRulesUpdate",
            Self::EndOfEpoch(_) => "EndOfEpoch",
            _ => unimplemented!(
                "a new TransactionKind enum variant was added and needs to be handled"
            ),
        }
    }
}

/// API for accessing and constructing [`Transaction`].
///
/// This trait provides node-internal methods for:
/// - **Accessors**: reading transaction fields (sender, kind, gas, expiration, etc.)
/// - **Queries**: inspecting transaction properties (shared objects, Move calls, sponsorship)
/// - **Validation**: checking transaction validity against protocol config
/// - **Constructors**: building new transactions (transfers, Move calls, programmable txs, etc.)
///
/// Note: The `iota-rust-sdk` crate (`iota-sdk-types`) defines additional
/// client-facing methods on [`Transaction`] itself.
pub trait TransactionAPI {
    /// Returns the address of the transaction sender.
    fn sender(&self) -> Address;

    /// Returns a reference to the transaction kind.
    fn kind(&self) -> &TransactionKind;

    /// Returns a mutable reference to the transaction kind.
    fn kind_mut(&mut self) -> &mut TransactionKind;

    /// Consumes self and returns the transaction kind.
    fn into_kind(self) -> TransactionKind;

    /// Returns the transaction signer(s). Includes both the sender and the gas
    /// owner if they differ (i.e. for sponsored transactions).
    fn signers(&self) -> NonEmpty<Address>;

    /// Returns a reference to the gas data (owner, payment objects, price,
    /// budget).
    fn gas_data(&self) -> &GasPayment;

    /// Returns the address that owns the gas payment objects.
    fn gas_owner(&self) -> Address;

    /// Returns the gas payment object references.
    fn gas(&self) -> &[ObjectReference];

    /// Returns the gas price for this transaction.
    fn gas_price(&self) -> u64;

    /// Returns the gas budget for this transaction.
    fn gas_budget(&self) -> u64;

    /// Returns the transaction expiration.
    fn expiration(&self) -> &TransactionExpiration;

    /// Returns a list of the transaction data shared input objects.
    ///
    /// IMPORTANT: This function does not return shared objects associated with
    /// `MoveAuthenticator` signatures. To check those objects as well, use the
    /// corresponding function from `SenderSignedTransaction`.
    fn shared_input_objects(&self) -> Vec<SharedObjectReference>;

    /// Returns a list of Move calls as `(package_id, module_name,
    /// function_name)` tuples.
    fn move_calls(&self) -> Vec<(&ObjectId, &str, &str)>;

    /// Returns the input objects required by the transaction body.
    ///
    /// Note: this does NOT include the objects read by any `MoveAuthenticator`s
    /// (abstract-account authenticators); those are carried by the transaction
    /// envelope, not by `TransactionData`. For the full set including
    /// authenticator inputs, use `SenderSignedData::input_objects` or
    /// `SenderSignedData::collect_all_input_object_kind_for_reading`.
    fn input_objects(&self) -> UserInputResult<Vec<InputObjectKind>>;

    /// Returns object references for all objects being received in this
    /// transaction.
    fn receiving_objects(&self) -> Vec<ObjectReference>;

    /// Validates the transaction data against the given protocol config,
    /// including gas checks.
    fn validity_check(&self, config: &ProtocolConfig) -> UserInputResult;

    /// Validates the transaction data against the given protocol config,
    /// skipping gas-related checks.
    fn validity_check_no_gas_check(&self, config: &ProtocolConfig) -> UserInputResult;

    /// Checks the BCS size of the transaction data against the protocol's
    /// `max_tx_size_bytes`.
    fn check_serialized_size(&self, config: &ProtocolConfig) -> IotaResult;

    /// Checks the gas payment against the protocol's cap on how many objects it
    /// may name.
    fn check_gas_payment_size(&self, config: &ProtocolConfig) -> UserInputResult;

    /// Check if the transaction is compliant with sponsorship.
    fn check_sponsorship(&self) -> UserInputResult;

    /// Returns `true` if this is a system transaction.
    fn is_system_tx(&self) -> bool;
    /// Returns `true` if this is the genesis transaction.
    fn is_genesis_tx(&self) -> bool;

    /// returns true if the transaction is one that is specially sequenced to
    /// run at the very end of the epoch
    fn is_end_of_epoch_tx(&self) -> bool;

    /// Check if the transaction is sponsored (namely gas owner != sender)
    fn is_sponsored_tx(&self) -> bool;

    /// Returns a mutable reference to the sender address. **Testing only.**
    fn sender_mut_for_testing(&mut self) -> &mut Address;

    /// Returns a mutable reference to the gas data.
    fn gas_data_mut(&mut self) -> &mut GasPayment;

    /// Returns a mutable reference to the expiration. **Testing only.**
    fn expiration_mut_for_testing(&mut self) -> &mut TransactionExpiration;

    /// Creates a new system transaction with no gas payment. Used for
    /// validator-initiated transactions (epoch changes, checkpoints, etc.).
    fn new_system_transaction(kind: TransactionKind) -> Transaction;

    /// Creates a new transaction with a single gas payment coin. The sender
    /// is also the gas owner.
    #[allow(clippy::new_ret_no_self)]
    fn new(
        kind: TransactionKind,
        sender: Address,
        gas_payment: ObjectReference,
        gas_budget: u64,
        gas_price: u64,
    ) -> Transaction;

    /// Creates a new transaction with multiple gas payment coins. The sender
    /// is also the gas owner.
    fn new_with_gas_coins(
        kind: TransactionKind,
        sender: Address,
        gas_payment: Vec<ObjectReference>,
        gas_budget: u64,
        gas_price: u64,
    ) -> Transaction;

    /// Creates a new transaction with multiple gas payment coins and a
    /// separate gas sponsor. Use this for sponsored transactions where
    /// the gas owner differs from the sender.
    fn new_with_gas_coins_allow_sponsor(
        kind: TransactionKind,
        sender: Address,
        gas_payment: Vec<ObjectReference>,
        gas_budget: u64,
        gas_price: u64,
        gas_sponsor: Address,
    ) -> Transaction;

    /// Creates a new transaction from a pre-built [`GasPayment`] struct.
    fn new_with_gas_data(
        kind: TransactionKind,
        sender: Address,
        gas_data: GasPayment,
    ) -> Transaction;

    /// Creates a transaction that calls a single Move function with a single
    /// gas payment coin.
    fn new_move_call(
        sender: Address,
        package: ObjectId,
        module: Identifier,
        function: Identifier,
        type_arguments: Vec<TypeTag>,
        gas_payment: ObjectReference,
        arguments: Vec<CallArg>,
        gas_budget: u64,
        gas_price: u64,
    ) -> anyhow::Result<Transaction>;

    /// Creates a transaction that calls a single Move function with multiple
    /// gas payment coins.
    fn new_move_call_with_gas_coins(
        sender: Address,
        package: ObjectId,
        module: Identifier,
        function: Identifier,
        type_arguments: Vec<TypeTag>,
        gas_payment: Vec<ObjectReference>,
        arguments: Vec<CallArg>,
        gas_budget: u64,
        gas_price: u64,
    ) -> anyhow::Result<Transaction>;

    /// Creates a transaction that transfers an object to a recipient.
    fn new_transfer(
        recipient: Address,
        object_ref: ObjectReference,
        sender: Address,
        gas_payment: ObjectReference,
        gas_budget: u64,
        gas_price: u64,
    ) -> Transaction;

    /// Creates a transaction that transfers IOTA coins to a recipient.
    /// If `amount` is `None`, the entire gas coin balance (minus gas fees)
    /// is transferred.
    fn new_transfer_iota(
        recipient: Address,
        sender: Address,
        amount: Option<u64>,
        gas_payment: ObjectReference,
        gas_budget: u64,
        gas_price: u64,
    ) -> Transaction;

    /// Creates a sponsored transaction that transfers IOTA coins to a
    /// recipient. If `amount` is `None`, the entire gas coin balance
    /// (minus gas fees) is transferred.
    fn new_transfer_iota_allow_sponsor(
        recipient: Address,
        sender: Address,
        amount: Option<u64>,
        gas_payment: ObjectReference,
        gas_budget: u64,
        gas_price: u64,
        gas_sponsor: Address,
    ) -> Transaction;

    /// Creates a transaction that pays multiple recipients from a set of
    /// input coins. The coins are merged and then split to satisfy the
    /// specified amounts.
    fn new_pay(
        sender: Address,
        coins: Vec<ObjectReference>,
        recipients: Vec<Address>,
        amounts: Vec<u64>,
        gas_payment: ObjectReference,
        gas_budget: u64,
        gas_price: u64,
    ) -> anyhow::Result<Transaction>;

    /// Creates a transaction that pays multiple recipients using IOTA coins.
    /// Similar to [`Self::new_pay`] but the gas coin is also used as an
    /// input coin.
    fn new_pay_iota(
        sender: Address,
        coins: Vec<ObjectReference>,
        recipients: Vec<Address>,
        amounts: Vec<u64>,
        gas_payment: ObjectReference,
        gas_budget: u64,
        gas_price: u64,
    ) -> anyhow::Result<Transaction>;

    /// Creates a transaction that sends all IOTA from the given coins to a
    /// single recipient. The gas coin is included as an input coin.
    fn new_pay_all_iota(
        sender: Address,
        coins: Vec<ObjectReference>,
        recipient: Address,
        gas_payment: ObjectReference,
        gas_budget: u64,
        gas_price: u64,
    ) -> Transaction;

    /// Creates a transaction that splits a coin into multiple coins with the
    /// specified amounts.
    fn new_split_coin(
        sender: Address,
        coin: ObjectReference,
        amounts: Vec<u64>,
        gas_payment: ObjectReference,
        gas_budget: u64,
        gas_price: u64,
    ) -> Transaction;

    /// Creates a transaction that publishes new Move modules.
    fn new_module(
        sender: Address,
        gas_payment: ObjectReference,
        modules: Vec<Vec<u8>>,
        dep_ids: Vec<ObjectId>,
        gas_budget: u64,
        gas_price: u64,
    ) -> Transaction;

    /// Creates a transaction that upgrades an existing Move package.
    /// Requires the upgrade capability object and the upgrade policy.
    fn new_upgrade(
        sender: Address,
        gas_payment: ObjectReference,
        package_id: ObjectId,
        modules: Vec<Vec<u8>>,
        dep_ids: Vec<ObjectId>,
        upgrade_capability_and_owner: (ObjectReference, Owner),
        upgrade_policy: u8,
        digest: Vec<u8>,
        gas_budget: u64,
        gas_price: u64,
    ) -> anyhow::Result<Transaction>;

    /// Creates a programmable transaction with multiple gas payment coins.
    /// The sender is also the gas owner.
    fn new_programmable(
        sender: Address,
        gas_payment: Vec<ObjectReference>,
        pt: ProgrammableTransaction,
        gas_budget: u64,
        gas_price: u64,
    ) -> Transaction;

    /// Creates a programmable transaction with multiple gas payment coins
    /// and a separate gas sponsor.
    fn new_programmable_allow_sponsor(
        sender: Address,
        gas_payment: Vec<ObjectReference>,
        pt: ProgrammableTransaction,
        gas_budget: u64,
        gas_price: u64,
        sponsor: Address,
    ) -> Transaction;

    /// Returns the internal message version number.
    fn message_version(&self) -> u64;

    /// Consumes self and returns the transaction kind, sender address, and
    /// gas payment object references as a tuple.
    fn execution_parts(&self) -> (TransactionKind, Address, GasPayment);
}

fn tx_bcs_size<T: Serialize>(tx: &T) -> IotaResult<usize> {
    bcs::serialized_size(tx).map_err(|e| IotaError::TransactionSerialization {
        error: e.to_string(),
    })
}

/// Checks the BCS size of `transaction` against the protocol's
/// `max_tx_size_bytes` and returns it.
fn check_transaction_size<T: Serialize>(
    transaction: &T,
    config: &ProtocolConfig,
) -> IotaResult<usize> {
    let tx_size = tx_bcs_size(transaction)?;
    let max_tx_size_bytes = config.max_tx_size_bytes();
    fp_ensure!(
        tx_size as u64 <= max_tx_size_bytes,
        IotaError::UserInput {
            error: UserInputError::SizeLimitExceeded {
                limit: format!(
                    "serialized transaction size exceeded maximum of {max_tx_size_bytes}"
                ),
                value: tx_size.to_string(),
            }
        }
    );
    Ok(tx_size)
}

impl TransactionAPI for Transaction {
    fn check_serialized_size(&self, config: &ProtocolConfig) -> IotaResult {
        check_transaction_size(self, config).map(|_| ())
    }

    fn sender(&self) -> Address {
        match self {
            Self::V1(v1) => v1.sender,
            _ => unimplemented!("a new Transaction enum variant was added and needs to be handled"),
        }
    }

    fn kind(&self) -> &TransactionKind {
        match self {
            Self::V1(v1) => &v1.kind,
            _ => unimplemented!("a new Transaction enum variant was added and needs to be handled"),
        }
    }

    fn kind_mut(&mut self) -> &mut TransactionKind {
        match self {
            Self::V1(v1) => &mut v1.kind,
            _ => unimplemented!("a new Transaction enum variant was added and needs to be handled"),
        }
    }

    fn into_kind(self) -> TransactionKind {
        match self {
            Self::V1(v1) => v1.kind,
            _ => unimplemented!("a new Transaction enum variant was added and needs to be handled"),
        }
    }

    fn signers(&self) -> NonEmpty<Address> {
        let mut signers = nonempty![self.sender()];
        if self.gas_owner() != self.sender() {
            signers.push(self.gas_owner());
        }
        signers
    }

    fn gas_data(&self) -> &GasPayment {
        match self {
            Self::V1(v1) => &v1.gas_payment,
            _ => unimplemented!("a new Transaction enum variant was added and needs to be handled"),
        }
    }

    fn gas_owner(&self) -> Address {
        self.gas_data().owner
    }

    fn gas(&self) -> &[ObjectReference] {
        &self.gas_data().objects
    }

    fn gas_price(&self) -> u64 {
        self.gas_data().price
    }

    fn gas_budget(&self) -> u64 {
        self.gas_data().budget
    }

    fn expiration(&self) -> &TransactionExpiration {
        match self {
            Self::V1(v1) => &v1.expiration,
            _ => unimplemented!("a new Transaction enum variant was added and needs to be handled"),
        }
    }

    fn shared_input_objects(&self) -> Vec<SharedObjectReference> {
        self.kind().shared_input_objects().collect()
    }

    fn move_calls(&self) -> Vec<(&ObjectId, &str, &str)> {
        self.kind().move_calls()
    }

    fn input_objects(&self) -> UserInputResult<Vec<InputObjectKind>> {
        let mut inputs = self.kind().input_objects()?;

        if !self.kind().is_system() {
            inputs.extend(
                self.gas()
                    .iter()
                    .map(|obj_ref| InputObjectKind::ImmOrOwnedMoveObject(*obj_ref)),
            );
        }
        Ok(inputs)
    }

    fn receiving_objects(&self) -> Vec<ObjectReference> {
        self.kind().receiving_objects()
    }

    fn validity_check(&self, config: &ProtocolConfig) -> UserInputResult {
        fp_ensure!(!self.gas().is_empty(), UserInputError::MissingGasPayment);
        self.check_gas_payment_size(config)?;
        if config.validate_input_object_versions() {
            for gas_object in self.gas() {
                input_object_version_validity_check(gas_object.version)?;
            }
        }
        self.validity_check_no_gas_check(config)
    }

    #[instrument(level = "trace", skip_all)]
    fn validity_check_no_gas_check(&self, config: &ProtocolConfig) -> UserInputResult {
        self.kind().validity_check(config)?;
        self.check_sponsorship()
    }

    fn check_gas_payment_size(&self, config: &ProtocolConfig) -> UserInputResult {
        fp_ensure!(
            self.gas().len() < config.max_gas_payment_objects() as usize,
            UserInputError::SizeLimitExceeded {
                limit: "maximum number of gas payment objects".to_string(),
                value: config.max_gas_payment_objects().to_string()
            }
        );
        Ok(())
    }

    fn is_sponsored_tx(&self) -> bool {
        self.gas_owner() != self.sender()
    }

    fn check_sponsorship(&self) -> UserInputResult {
        if self.gas_owner() == self.sender() {
            return Ok(());
        }
        if matches!(self.kind(), TransactionKind::Programmable(_)) {
            return Ok(());
        }
        Err(UserInputError::UnsupportedSponsoredTransactionKind)
    }

    fn is_end_of_epoch_tx(&self) -> bool {
        matches!(self.kind(), TransactionKind::EndOfEpoch(_))
    }

    fn is_system_tx(&self) -> bool {
        self.kind().is_system()
    }

    fn is_genesis_tx(&self) -> bool {
        matches!(self.kind(), TransactionKind::Genesis(_))
    }

    fn sender_mut_for_testing(&mut self) -> &mut Address {
        match self {
            Self::V1(v1) => &mut v1.sender,
            _ => unimplemented!("a new Transaction enum variant was added and needs to be handled"),
        }
    }

    fn gas_data_mut(&mut self) -> &mut GasPayment {
        match self {
            Self::V1(v1) => &mut v1.gas_payment,
            _ => unimplemented!("a new Transaction enum variant was added and needs to be handled"),
        }
    }

    fn expiration_mut_for_testing(&mut self) -> &mut TransactionExpiration {
        match self {
            Self::V1(v1) => &mut v1.expiration,
            _ => unimplemented!("a new Transaction enum variant was added and needs to be handled"),
        }
    }

    fn new_system_transaction(kind: TransactionKind) -> Transaction {
        assert!(kind.is_system());
        let sender = Address::ZERO;
        Transaction::V1(TransactionV1 {
            kind,
            sender,
            gas_payment: GasPayment {
                price: GAS_PRICE_FOR_SYSTEM_TX,
                owner: sender,
                objects: vec![ObjectReference::new(
                    ObjectId::ZERO,
                    Version::default(),
                    ObjectDigest::MIN,
                )],
                budget: 0,
            },
            expiration: TransactionExpiration::None,
        })
    }

    fn new(
        kind: TransactionKind,
        sender: Address,
        gas_payment: ObjectReference,
        gas_budget: u64,
        gas_price: u64,
    ) -> Transaction {
        Transaction::V1(TransactionV1 {
            kind,
            sender,
            gas_payment: GasPayment {
                price: gas_price,
                owner: sender,
                objects: vec![gas_payment],
                budget: gas_budget,
            },
            expiration: TransactionExpiration::None,
        })
    }

    fn new_with_gas_coins(
        kind: TransactionKind,
        sender: Address,
        gas_payment: Vec<ObjectReference>,
        gas_budget: u64,
        gas_price: u64,
    ) -> Transaction {
        Transaction::new_with_gas_coins_allow_sponsor(
            kind,
            sender,
            gas_payment,
            gas_budget,
            gas_price,
            sender,
        )
    }

    fn new_with_gas_coins_allow_sponsor(
        kind: TransactionKind,
        sender: Address,
        gas_payment: Vec<ObjectReference>,
        gas_budget: u64,
        gas_price: u64,
        gas_sponsor: Address,
    ) -> Transaction {
        Transaction::V1(TransactionV1 {
            kind,
            sender,
            gas_payment: GasPayment {
                price: gas_price,
                owner: gas_sponsor,
                objects: gas_payment,
                budget: gas_budget,
            },
            expiration: TransactionExpiration::None,
        })
    }

    fn new_with_gas_data(
        kind: TransactionKind,
        sender: Address,
        gas_data: GasPayment,
    ) -> Transaction {
        Transaction::V1(TransactionV1 {
            kind,
            sender,
            gas_payment: gas_data,
            expiration: TransactionExpiration::None,
        })
    }

    fn new_move_call(
        sender: Address,
        package: ObjectId,
        module: Identifier,
        function: Identifier,
        type_arguments: Vec<TypeTag>,
        gas_payment: ObjectReference,
        arguments: Vec<CallArg>,
        gas_budget: u64,
        gas_price: u64,
    ) -> anyhow::Result<Transaction> {
        Transaction::new_move_call_with_gas_coins(
            sender,
            package,
            module,
            function,
            type_arguments,
            vec![gas_payment],
            arguments,
            gas_budget,
            gas_price,
        )
    }

    fn new_move_call_with_gas_coins(
        sender: Address,
        package: ObjectId,
        module: Identifier,
        function: Identifier,
        type_arguments: Vec<TypeTag>,
        gas_payment: Vec<ObjectReference>,
        arguments: Vec<CallArg>,
        gas_budget: u64,
        gas_price: u64,
    ) -> anyhow::Result<Transaction> {
        let pt = {
            let mut builder = ProgrammableTransactionBuilder::new();
            builder.move_call(package, module, function, type_arguments, arguments)?;
            builder.finish()
        };
        Ok(Transaction::new_programmable(
            sender,
            gas_payment,
            pt,
            gas_budget,
            gas_price,
        ))
    }

    fn new_transfer(
        recipient: Address,
        object_ref: ObjectReference,
        sender: Address,
        gas_payment: ObjectReference,
        gas_budget: u64,
        gas_price: u64,
    ) -> Transaction {
        let pt = {
            let mut builder = ProgrammableTransactionBuilder::new();
            builder.transfer_object(recipient, object_ref).unwrap();
            builder.finish()
        };
        Transaction::new_programmable(sender, vec![gas_payment], pt, gas_budget, gas_price)
    }

    fn new_transfer_iota(
        recipient: Address,
        sender: Address,
        amount: Option<u64>,
        gas_payment: ObjectReference,
        gas_budget: u64,
        gas_price: u64,
    ) -> Transaction {
        Transaction::new_transfer_iota_allow_sponsor(
            recipient,
            sender,
            amount,
            gas_payment,
            gas_budget,
            gas_price,
            sender,
        )
    }

    fn new_transfer_iota_allow_sponsor(
        recipient: Address,
        sender: Address,
        amount: Option<u64>,
        gas_payment: ObjectReference,
        gas_budget: u64,
        gas_price: u64,
        gas_sponsor: Address,
    ) -> Transaction {
        let pt = {
            let mut builder = ProgrammableTransactionBuilder::new();
            builder.transfer_iota(recipient, amount);
            builder.finish()
        };
        Transaction::new_programmable_allow_sponsor(
            sender,
            vec![gas_payment],
            pt,
            gas_budget,
            gas_price,
            gas_sponsor,
        )
    }

    fn new_pay(
        sender: Address,
        coins: Vec<ObjectReference>,
        recipients: Vec<Address>,
        amounts: Vec<u64>,
        gas_payment: ObjectReference,
        gas_budget: u64,
        gas_price: u64,
    ) -> anyhow::Result<Transaction> {
        let pt = {
            let mut builder = ProgrammableTransactionBuilder::new();
            builder.pay(coins, recipients, amounts)?;
            builder.finish()
        };
        Ok(Transaction::new_programmable(
            sender,
            vec![gas_payment],
            pt,
            gas_budget,
            gas_price,
        ))
    }

    fn new_pay_iota(
        sender: Address,
        mut coins: Vec<ObjectReference>,
        recipients: Vec<Address>,
        amounts: Vec<u64>,
        gas_payment: ObjectReference,
        gas_budget: u64,
        gas_price: u64,
    ) -> anyhow::Result<Transaction> {
        coins.insert(0, gas_payment);
        let pt = {
            let mut builder = ProgrammableTransactionBuilder::new();
            builder.pay_iota(recipients, amounts)?;
            builder.finish()
        };
        Ok(Transaction::new_programmable(
            sender, coins, pt, gas_budget, gas_price,
        ))
    }

    fn new_pay_all_iota(
        sender: Address,
        mut coins: Vec<ObjectReference>,
        recipient: Address,
        gas_payment: ObjectReference,
        gas_budget: u64,
        gas_price: u64,
    ) -> Transaction {
        coins.insert(0, gas_payment);
        let pt = {
            let mut builder = ProgrammableTransactionBuilder::new();
            builder.pay_all_iota(recipient);
            builder.finish()
        };
        Transaction::new_programmable(sender, coins, pt, gas_budget, gas_price)
    }

    fn new_split_coin(
        sender: Address,
        coin: ObjectReference,
        amounts: Vec<u64>,
        gas_payment: ObjectReference,
        gas_budget: u64,
        gas_price: u64,
    ) -> Transaction {
        let pt = {
            let mut builder = ProgrammableTransactionBuilder::new();
            builder.split_coin(sender, coin, amounts);
            builder.finish()
        };
        Transaction::new_programmable(sender, vec![gas_payment], pt, gas_budget, gas_price)
    }

    fn new_module(
        sender: Address,
        gas_payment: ObjectReference,
        modules: Vec<Vec<u8>>,
        dep_ids: Vec<ObjectId>,
        gas_budget: u64,
        gas_price: u64,
    ) -> Transaction {
        let pt = {
            let mut builder = ProgrammableTransactionBuilder::new();
            let upgrade_cap = builder.publish_upgradeable(modules, dep_ids);
            builder.transfer_arg(sender, upgrade_cap);
            builder.finish()
        };
        Transaction::new_programmable(sender, vec![gas_payment], pt, gas_budget, gas_price)
    }

    fn new_upgrade(
        sender: Address,
        gas_payment: ObjectReference,
        package_id: ObjectId,
        modules: Vec<Vec<u8>>,
        dep_ids: Vec<ObjectId>,
        (upgrade_capability, capability_owner): (ObjectReference, Owner),
        upgrade_policy: u8,
        digest: Vec<u8>,
        gas_budget: u64,
        gas_price: u64,
    ) -> anyhow::Result<Transaction> {
        let pt = {
            let mut builder = ProgrammableTransactionBuilder::new();
            let capability_arg = match capability_owner {
                Owner::Address(_) => CallArg::ImmutableOrOwned(upgrade_capability),
                Owner::Shared(initial_shared_version) => {
                    CallArg::Shared(SharedObjectReference::new(
                        upgrade_capability.object_id,
                        initial_shared_version,
                        true,
                    ))
                }
                Owner::Immutable => {
                    bail!("Upgrade capability is stored immutably and cannot be used for upgrades");
                }
                Owner::Object(_) => {
                    bail!("Upgrade capability controlled by object");
                }
                _ => unimplemented!("a new Owner enum variant was added and needs to be handled"),
            };
            builder.obj(capability_arg).unwrap();
            let upgrade_arg = builder.pure(upgrade_policy).unwrap();
            let digest_arg = builder.pure(digest).unwrap();
            let upgrade_ticket = builder.programmable_move_call(
                ObjectId::FRAMEWORK,
                Identifier::PACKAGE_MODULE,
                Identifier::from_static("authorize_upgrade"),
                vec![],
                vec![Argument::Input(0), upgrade_arg, digest_arg],
            );
            let upgrade_receipt = builder.upgrade(package_id, upgrade_ticket, dep_ids, modules);

            builder.programmable_move_call(
                ObjectId::FRAMEWORK,
                Identifier::PACKAGE_MODULE,
                Identifier::from_static("commit_upgrade"),
                vec![],
                vec![Argument::Input(0), upgrade_receipt],
            );

            builder.finish()
        };
        Ok(Transaction::new_programmable(
            sender,
            vec![gas_payment],
            pt,
            gas_budget,
            gas_price,
        ))
    }

    fn new_programmable(
        sender: Address,
        gas_payment: Vec<ObjectReference>,
        pt: ProgrammableTransaction,
        gas_budget: u64,
        gas_price: u64,
    ) -> Transaction {
        Transaction::new_programmable_allow_sponsor(
            sender,
            gas_payment,
            pt,
            gas_budget,
            gas_price,
            sender,
        )
    }

    fn new_programmable_allow_sponsor(
        sender: Address,
        gas_payment: Vec<ObjectReference>,
        pt: ProgrammableTransaction,
        gas_budget: u64,
        gas_price: u64,
        sponsor: Address,
    ) -> Transaction {
        let kind = TransactionKind::Programmable(pt);
        Transaction::new_with_gas_coins_allow_sponsor(
            kind,
            sender,
            gas_payment,
            gas_budget,
            gas_price,
            sponsor,
        )
    }

    fn message_version(&self) -> u64 {
        match self {
            Transaction::V1(_) => 1,
            _ => unimplemented!("a new Transaction enum variant was added and needs to be handled"),
        }
    }

    fn execution_parts(&self) -> (TransactionKind, Address, GasPayment) {
        (self.kind().clone(), self.sender(), self.gas_data().clone())
    }
}

pub struct TxValidityCheckContext<'a> {
    pub config: &'a ProtocolConfig,
    pub epoch: EpochId,
}

/// Merge every [`MoveAuthenticator`]'s input objects into `input_objects`.
///
/// Objects not yet present are appended; for an object that appears in both
/// sets the kinds are checked for consistency and unioned via
/// [`InputObjectKind::left_union_with_checks`] (in particular, a shared object
/// may differ in mutability but not in initial shared version).
pub fn merge_authenticator_input_objects<'a>(
    move_authenticators: impl IntoIterator<Item = &'a MoveAuthenticator>,
    input_objects: &mut Vec<InputObjectKind>,
) -> UserInputResult<()> {
    for move_authenticator in move_authenticators {
        for auth_object in move_authenticator.input_objects() {
            let entry = input_objects
                .iter_mut()
                .find(|o| o.object_id() == auth_object.object_id());

            match entry {
                None => input_objects.push(auth_object),
                Some(existing) => existing.left_union_with_checks(&auth_object)?,
            }
        }
    }
    Ok(())
}

/// API for accessing and constructing [`SenderSignedTransaction`].
///
/// This trait provides node-internal methods on the SDK's
/// [`SenderSignedTransaction`], which carries the transaction data together
/// with the signatures of all transaction participants. A non-participant
/// signature must not be present, and the signature order does not matter.
pub trait SenderSignedTransactionAPI {
    /// Creates a new [`SenderSignedTransaction`] with a single sender
    /// signature.
    fn new_from_sender_signature(
        tx: Transaction,
        tx_signature: SimpleSignature,
    ) -> SenderSignedTransaction;

    /// Adds a signature. Does not check the validity of the signature or
    /// perform any de-dup checks.
    fn add_signature(&mut self, new_signature: SimpleSignature);

    /// Returns a mapping from the address each signature commits to, to the
    /// signature itself.
    fn get_signer_sig_mapping(&self) -> IotaResult<BTreeMap<Address, &UserSignature>>;

    /// Returns `true` if any signature is a multisig.
    fn has_multisig(&self) -> bool;

    /// Returns a mutable reference to the transaction. **Testing only.**
    fn transaction_mut_for_testing(&mut self) -> &mut Transaction;

    /// Returns a mutable reference to the signatures. **Testing only.**
    fn tx_signatures_mut_for_testing(&mut self) -> &mut Vec<UserSignature>;

    /// Returns the BCS serialized size in bytes.
    fn serialized_size(&self) -> IotaResult<usize>;

    /// Validate untrusted user transaction, including its size, input count,
    /// command count, etc.
    /// Returns the certificate serialised bytes size.
    fn validity_check(&self, context: &TxValidityCheckContext<'_>) -> Result<usize, IotaError>;

    /// Returns all unique input objects including those from
    /// `MoveAuthenticator`s if any for reading, in the order the transaction
    /// and then each authenticator name them.
    ///
    /// Although some shared objects(with a different mutability flag, for
    /// example) can be duplicated in the transaction and authenticators, we
    /// load them independently to make it possible to analyze the inputs in
    /// the transaction checkers.
    fn collect_all_input_object_kind_for_reading(&self) -> IotaResult<Vec<InputObjectKind>>;

    /// Splits the provided input objects into groups:
    /// 1. Input objects required by the transaction itself; may contain duplicates if an IOTA coin
    ///    is used both as an input and a gas coin.
    /// 2. A list of input objects required by each `MoveAuthenticator`( including the object to
    ///    authenticate) + the object to authenticate.
    fn split_input_objects_into_groups_for_reading(
        &self,
        input_objects: InputObjects,
    ) -> IotaResult<(InputObjects, Vec<(InputObjects, ObjectReadResult)>)>;

    /// Checks if [`SenderSignedTransaction`] contains at least one shared
    /// object. This function checks shared objects from the
    /// `MoveAuthenticator`s if any.
    ///
    /// Unlike [`Self::shared_input_objects`], it does not combine the shared
    /// objects of the transaction and its authenticators, so it does not panic
    /// when two of them name the same object with different initial versions.
    fn contains_shared_object(&self) -> bool;

    /// Returns an iterator over all shared input objects related to this
    /// transaction, including those from `MoveAuthenticator`s if any.
    ///
    /// If a shared object appears with the same version but different
    /// mutability, only one instance which is mutable is returned.
    ///
    /// Panics if there are shared objects with the same ID but different
    /// initial versions.
    fn shared_input_objects(&self) -> Vec<SharedObjectReference>;

    /// Returns an iterator over all input objects related to this
    /// transaction, including those from the `MoveAuthenticator`s if any.
    ///
    /// If an IOTA coin is used both as an input and as a gas coin, it will
    /// appear two times in the returned iterator.
    ///
    /// If a shared object appears both in the transaction and authenticator
    /// with different mutability, only one instance which is mutable is
    /// returned.
    ///
    /// Shared objects with the same ID but different versions are not allowed.
    fn input_objects(&self) -> IotaResult<Vec<InputObjectKind>>;

    /// Checks if [`SenderSignedTransaction`] contains the `Random` object as an
    /// input.
    /// This function checks shared objects from the `MoveAuthenticator`s if
    /// any.
    fn uses_randomness(&self) -> bool;
}

impl SenderSignedTransactionAPI for SenderSignedTransaction {
    fn new_from_sender_signature(
        tx: Transaction,
        tx_signature: SimpleSignature,
    ) -> SenderSignedTransaction {
        Self::new(tx, vec![tx_signature.into()])
    }

    fn add_signature(&mut self, new_signature: SimpleSignature) {
        self.signed_transaction_mut()
            .signatures
            .push(new_signature.into());
    }

    fn get_signer_sig_mapping(&self) -> IotaResult<BTreeMap<Address, &UserSignature>> {
        let mut mapping = BTreeMap::new();
        for sig in &self.signed_transaction().signatures {
            let address = sig.derive_address();
            mapping.insert(address, sig);
        }
        Ok(mapping)
    }

    fn has_multisig(&self) -> bool {
        self.signatures().iter().any(|sig| sig.is_multisig())
    }

    fn transaction_mut_for_testing(&mut self) -> &mut Transaction {
        &mut self.signed_transaction_mut().transaction
    }

    fn tx_signatures_mut_for_testing(&mut self) -> &mut Vec<UserSignature> {
        &mut self.signed_transaction_mut().signatures
    }

    fn serialized_size(&self) -> IotaResult<usize> {
        tx_bcs_size(self)
    }

    fn validity_check(&self, context: &TxValidityCheckContext<'_>) -> Result<usize, IotaError> {
        // Check that the features used by the user signatures are enabled on the
        // network.
        check_user_signature_protocol_compatibility(self, context.config)?;

        // CRITICAL!!
        // Users cannot send system transactions.
        let tx = self.transaction();
        fp_ensure!(
            !tx.is_system_tx(),
            IotaError::UserInput {
                error: UserInputError::Unsupported(
                    "SenderSignedTransaction must not contain system transaction".to_string()
                )
            }
        );

        // Checks to see if the transaction has expired
        if match &tx.expiration() {
            TransactionExpiration::None => false,
            TransactionExpiration::Epoch(exp_poch) => *exp_poch < context.epoch,
            _ => unimplemented!(
                "a new TransactionExpiration enum variant was added and needs to be handled"
            ),
        } {
            return Err(IotaError::TransactionExpired);
        }

        // Enforce overall transaction size limit.
        let tx_size = check_transaction_size(self, context.config)?;

        tx.validity_check(context.config)
            .map_err(Into::<IotaError>::into)?;

        move_authenticators_validity_check(self, context.config)?;

        Ok(tx_size)
    }

    fn collect_all_input_object_kind_for_reading(&self) -> IotaResult<Vec<InputObjectKind>> {
        let mut seen = HashSet::new();
        Ok(self
            .transaction()
            .input_objects()?
            .into_iter()
            .chain(
                self.move_authenticators()
                    .into_iter()
                    .flat_map(|authenticator| authenticator.input_objects()),
            )
            .filter(|kind| seen.insert(*kind))
            .collect())
    }

    fn split_input_objects_into_groups_for_reading(
        &self,
        input_objects: InputObjects,
    ) -> IotaResult<(InputObjects, Vec<(InputObjects, ObjectReadResult)>)> {
        let input_objects_map = input_objects
            .iter()
            .map(|o| (&o.input_object_kind, o))
            .collect::<HashMap<_, _>>();

        let mut tx_input_objects: InputObjects = self
            .transaction()
            .input_objects()?
            .iter()
            .map(|k| {
                input_objects_map
                    .get(k)
                    .map(|&r| r.clone())
                    .expect("All transaction input objects are expected to be present")
            })
            .collect::<Vec<_>>()
            .into();
        if let Some((gas_object_id, version)) = input_objects.gas_object_cancellation() {
            tx_input_objects.set_gas_object_cancellation(gas_object_id, version);
        }

        let per_authenticator_inputs =
            self.move_authenticators()
                .into_iter()
                .map(|move_authenticator| {
                    let authenticator_input_objects = move_authenticator
                        .input_objects()
                        .iter()
                        .map(|k| {
                            input_objects_map.get(k).map(|&r| r.clone()).expect(
                                "All authenticator input objects are expected to be present",
                            )
                        })
                        .collect::<Vec<_>>()
                        .into();

                    let account_objects = move_authenticator
                        .object_to_authenticate()
                        .input_object_kind()
                        .iter()
                        .map(|k| {
                            input_objects_map
                                .get(k)
                                .map(|&r| r.clone())
                                .expect("Account object is expected to be present")
                        })
                        .collect::<Vec<_>>();

                    debug_assert!(
                        account_objects.len() == 1,
                        "Only one account object must be loaded"
                    );

                    (
                        authenticator_input_objects,
                        account_objects
                            .into_iter()
                            .next()
                            .expect("Account object is expected to be present"),
                    )
                })
                .collect();

        Ok((tx_input_objects, per_authenticator_inputs))
    }

    fn contains_shared_object(&self) -> bool {
        self.transaction().kind().contains_shared_object()
            || self
                .move_authenticators()
                .iter()
                .any(|move_authenticator| !move_authenticator.shared_objects().is_empty())
    }

    fn shared_input_objects(&self) -> Vec<SharedObjectReference> {
        // Vector is used to preserve the order of input objects.
        let mut input_objects = self.transaction().shared_input_objects();

        // Add Move authenticator shared objects if any.
        self.move_authenticators()
            .into_iter()
            .for_each(|move_authenticator| {
                for auth_shared_object in move_authenticator.shared_objects() {
                    let entry = input_objects
                        .iter_mut()
                        .find(|o| o.object_id == auth_shared_object.object_id);

                    match entry {
                        None => input_objects.push(auth_shared_object),
                        Some(existing) => {
                            left_union_shared_input_objects(existing, &auth_shared_object)
                                .expect("union of shared objects should not fail")
                        }
                    }
                }
            });

        input_objects
    }

    fn input_objects(&self) -> IotaResult<Vec<InputObjectKind>> {
        // Can contain duplicates in case of using the same IOTA coin as an input and as
        // a gas coin.
        let mut input_objects = self.transaction().input_objects()?;

        // Add the `MoveAuthenticator` shared objects if any.
        merge_authenticator_input_objects(self.move_authenticators(), &mut input_objects)?;

        Ok(input_objects)
    }

    fn uses_randomness(&self) -> bool {
        self.shared_input_objects()
            .iter()
            .any(|obj| obj.object_id == ObjectId::RANDOMNESS_STATE)
    }
}

fn check_user_signature_protocol_compatibility(
    signed_tx: &SenderSignedTransaction,
    config: &ProtocolConfig,
) -> IotaResult {
    for sig in signed_tx.signatures() {
        match sig {
            UserSignature::PasskeyAuthenticator(_) => {
                if !config.passkey_auth() {
                    return Err(IotaError::UserInput {
                        error: UserInputError::Unsupported(
                            "passkey is not enabled on this network".to_string(),
                        ),
                    });
                }
            }
            UserSignature::MoveAuthenticator(_) => {
                if !config.enable_move_authentication() {
                    return Err(IotaError::UserInput {
                        error: UserInputError::Unsupported(
                            "`Move authentication` is not enabled on this network".to_string(),
                        ),
                    });
                }
            }
            UserSignature::Simple(_) | UserSignature::Multisig(_) => (),
            _ => {
                unimplemented!("a new UserSignature variant was added and needs to be handled")
            }
        }
    }

    Ok(())
}

fn move_authenticators_validity_check(
    signed_tx: &SenderSignedTransaction,
    config: &ProtocolConfig,
) -> IotaResult {
    let authenticators = signed_tx.move_authenticators();

    // Check each `MoveAuthenticator` validity.
    authenticators
        .iter()
        .try_for_each(|authenticator| authenticator.validity_check(config))?;

    // Additional checks when `MoveAuthenticators` are present.
    let authenticators_num = authenticators.len();
    if authenticators_num > 0 {
        let tx = signed_tx.transaction();

        fp_ensure!(
            tx.kind().is_programmable(),
            UserInputError::Unsupported(
                "SenderSignedTransaction with MoveAuthenticator must be a programmable transaction"
                    .to_string(),
            )
            .into()
        );

        if !config.enable_move_authentication_for_sponsor() {
            fp_ensure!(
                authenticators_num == 1,
                UserInputError::Unsupported(
                    "SenderSignedTransaction with more than one MoveAuthenticator is not supported"
                        .to_string(),
                )
                .into()
            );

            fp_ensure!(
                signed_tx.sender_move_authenticator().is_some(),
                UserInputError::Unsupported(
                    "SenderSignedTransaction can have MoveAuthenticator only for the sender"
                        .to_string(),
                )
                .into()
            );
        }

        check_move_authenticators_input_consistency(tx, &authenticators)?;
    }

    Ok(())
}

fn check_move_authenticators_input_consistency(
    tx: &Transaction,
    authenticators: &[&MoveAuthenticator],
) -> IotaResult {
    // Get the input objects from the transaction data kind to skip the gas coins.
    let mut checked_inputs = tx
        .kind()
        .input_objects()?
        .into_iter()
        .map(|o| (o.object_id(), o))
        .collect::<HashMap<_, _>>();

    authenticators.iter().try_for_each(|authenticator| {
        authenticator
            .input_objects()
            .iter()
            .try_for_each(|auth_input_object| {
                match checked_inputs.get(&auth_input_object.object_id()) {
                    Some(existing) => {
                        auth_input_object.check_consistency_for_authentication(existing)?
                    }
                    None => {
                        checked_inputs.insert(auth_input_object.object_id(), *auth_input_object);
                    }
                };

                Ok(())
            })
    })
}

impl Message for SenderSignedTransaction {
    type DigestType = TransactionDigest;
    const SCOPE: IntentScope = IntentScope::SenderSignedTransaction;

    /// Computes the tx digest that encodes the Rust type prefix from Signable
    /// trait.
    fn digest(&self) -> Self::DigestType {
        self.transaction().digest()
    }
}

impl<S> Envelope<SenderSignedTransaction, S> {
    pub fn sender_address(&self) -> Address {
        self.data().transaction().sender()
    }

    pub fn gas(&self) -> &[ObjectReference] {
        self.data().transaction().gas()
    }

    // Returns the primary key for this transaction.
    pub fn key(&self) -> TransactionKey {
        match &self.data().transaction().kind() {
            TransactionKind::RandomnessStateUpdate(rsu) => {
                TransactionKey::RandomnessRound(rsu.epoch, rsu.randomness_round)
            }
            _ => TransactionKey::Digest(*self.digest()),
        }
    }

    // Returns non-Digest keys that could be used to refer to this transaction.
    //
    // At the moment this returns a single Option for efficiency, but if more key
    // types are added, the return type could change to Vec<TransactionKey>.
    pub fn non_digest_key(&self) -> Option<TransactionKey> {
        match &self.data().transaction().kind() {
            TransactionKind::RandomnessStateUpdate(rsu) => Some(TransactionKey::RandomnessRound(
                rsu.epoch,
                rsu.randomness_round,
            )),
            _ => None,
        }
    }

    pub fn is_system_tx(&self) -> bool {
        self.data().transaction().is_system_tx()
    }

    pub fn is_sponsored_tx(&self) -> bool {
        self.data().transaction().is_sponsored_tx()
    }
}

impl TransactionEnvelope {
    pub fn from_data_and_signer(
        tx: Transaction,
        signers: Vec<&impl iota_sdk_crypto::Signer<SimpleSignature>>,
    ) -> Self {
        let signatures = {
            let digest = tx.signing_digest();
            signers.into_iter().map(|s| s.sign(&digest)).collect()
        };
        Self::from_data(tx, signatures)
    }

    // TODO: Rename this function and above to make it clearer.
    pub fn from_data(tx: Transaction, signatures: Vec<SimpleSignature>) -> Self {
        Self::from_user_sig_data(tx, signatures.into_iter().map(|s| s.into()).collect())
    }

    pub fn signature_from_signer(
        tx: Transaction,
        intent: Intent,
        signer: &impl iota_sdk_crypto::Signer<SimpleSignature>,
    ) -> SimpleSignature {
        let digest = IntentMessage::new(intent, tx).signing_digest();
        signer.sign(&digest)
    }

    pub fn from_user_sig_data(tx: Transaction, signatures: Vec<UserSignature>) -> Self {
        Self::new(SenderSignedTransaction::new(tx, signatures))
    }

    /// Returns the Base64 encoded tx_bytes
    /// and a list of Base64 encoded [`UserSignature`].
    pub fn to_tx_bytes_and_signatures(&self) -> (Base64, Vec<Base64>) {
        (
            Base64::from_bytes(&self.data().transaction().to_bcs()),
            self.data()
                .signatures()
                .iter()
                .map(|s| Base64::from_bytes(&s.to_bytes()))
                .collect(),
        )
    }
}

impl VerifiedTransaction {
    pub fn new_genesis_transaction(objects: Vec<GenesisObject>, events: Vec<Event>) -> Self {
        GenesisTransaction { objects, events }
            .pipe(TransactionKind::Genesis)
            .pipe(Self::new_system_transaction)
    }

    pub fn new_consensus_commit_prologue_v1(
        epoch: u64,
        round: u64,
        commit_timestamp_ms: CheckpointTimestamp,
        consensus_commit_digest: ConsensusCommitDigest,
        canceled_transactions: Vec<CanceledTransaction>,
    ) -> Self {
        ConsensusCommitPrologueV1 {
            epoch,
            round,
            // sub_dag_index is reserved for when we have multi commits per round.
            sub_dag_index: None,
            commit_timestamp_ms,
            consensus_commit_digest,
            consensus_determined_version_assignments:
                ConsensusDeterminedVersionAssignments::CanceledTransactions {
                    canceled_transactions,
                },
        }
        .pipe(TransactionKind::ConsensusCommitPrologueV1)
        .pipe(Self::new_system_transaction)
    }

    pub fn new_randomness_state_update(
        epoch: u64,
        randomness_round: RandomnessRound,
        random_bytes: Vec<u8>,
        randomness_obj_initial_shared_version: Version,
    ) -> Self {
        RandomnessStateUpdate {
            epoch,
            randomness_round,
            random_bytes,
            randomness_obj_initial_shared_version,
        }
        .pipe(TransactionKind::RandomnessStateUpdate)
        .pipe(Self::new_system_transaction)
    }

    pub fn new_end_of_epoch_transaction(txns: Vec<EndOfEpochTransactionKind>) -> Self {
        TransactionKind::EndOfEpoch(txns).pipe(Self::new_system_transaction)
    }

    pub fn new_transaction_deny_rules_update(update: TransactionDenyRulesUpdate) -> Self {
        TransactionKind::TransactionDenyRulesUpdate(update).pipe(Self::new_system_transaction)
    }

    fn new_system_transaction(system_transaction: TransactionKind) -> Self {
        system_transaction
            .pipe(Transaction::new_system_transaction)
            .pipe(|data| {
                SenderSignedTransaction::new_from_sender_signature(data, zero_ed25519_signature())
            })
            .pipe(TransactionEnvelope::new)
            .pipe(Self::new_from_verified)
    }
}

impl VerifiedSignedTransaction {
    /// Use signing key to create a signed object.
    #[instrument(level = "trace", skip_all)]
    pub fn new(
        epoch: EpochId,
        transaction: VerifiedTransaction,
        authority: AuthorityName,
        secret: &dyn Signer<AuthoritySignature>,
    ) -> Self {
        Self::new_from_verified(SignedTransaction::new(
            epoch,
            transaction.into_inner().into_data(),
            secret,
            authority,
        ))
    }
}

/// A transaction that is signed by a sender but not yet by an authority.
pub type TransactionEnvelope = Envelope<SenderSignedTransaction, EmptySignInfo>;
pub type VerifiedTransaction = VerifiedEnvelope<SenderSignedTransaction, EmptySignInfo>;
pub type TrustedTransaction = TrustedEnvelope<SenderSignedTransaction, EmptySignInfo>;

/// A transaction that is signed by a sender and also by an authority.
pub type SignedTransaction = Envelope<SenderSignedTransaction, AuthoritySignInfo>;
pub type VerifiedSignedTransaction = VerifiedEnvelope<SenderSignedTransaction, AuthoritySignInfo>;

impl TransactionEnvelope {
    pub fn verify_signature_for_testing(&self, verify_params: &VerifyParams) -> IotaResult {
        verify_sender_signed_data_message_signatures(self.data(), verify_params)
    }

    pub fn try_into_verified_for_testing(
        self,
        verify_params: &VerifyParams,
    ) -> IotaResult<VerifiedTransaction> {
        self.verify_signature_for_testing(verify_params)?;
        Ok(VerifiedTransaction::new_from_verified(self))
    }

    pub fn gas_price(&self) -> u64 {
        self.data().transaction().gas_price()
    }
}

impl SignedTransaction {
    pub fn verify_signatures_authenticated_for_testing(
        &self,
        committee: &Committee,
        verify_params: &VerifyParams,
    ) -> IotaResult {
        verify_sender_signed_data_message_signatures(self.data(), verify_params)?;

        self.auth_sig().verify_secure(
            self.data(),
            Intent::iota_app(IntentScope::SenderSignedTransaction),
            committee,
        )
    }

    pub fn try_into_verified_for_testing(
        self,
        committee: &Committee,
        verify_params: &VerifyParams,
    ) -> IotaResult<VerifiedSignedTransaction> {
        self.verify_signatures_authenticated_for_testing(committee, verify_params)?;
        Ok(VerifiedSignedTransaction::new_from_verified(self))
    }
}

pub type CertifiedTransaction = Envelope<SenderSignedTransaction, AuthorityStrongQuorumSignInfo>;

impl CertifiedTransaction {
    pub fn certificate_digest(&self) -> CertificateDigest {
        let mut digest = DefaultHash::default();
        bcs::serialize_into(&mut digest, self).expect("serialization should not fail");
        let hash = digest.finalize();
        CertificateDigest::new(hash.into())
    }

    pub fn gas_price(&self) -> u64 {
        self.data().transaction().gas_price()
    }

    // TODO: Eventually we should remove all calls to verify_signature
    // and make sure they all call verify to avoid repeated verifications.
    #[instrument(level = "trace", skip_all)]
    pub fn verify_signatures_authenticated(
        &self,
        committee: &Committee,
        verify_params: &VerifyParams,
    ) -> IotaResult {
        verify_sender_signed_data_message_signatures(self.data(), verify_params)?;
        self.verify_committee_sigs_only(committee)
    }

    pub fn try_into_verified_for_testing(
        self,
        committee: &Committee,
        verify_params: &VerifyParams,
    ) -> IotaResult<VerifiedCertificate> {
        self.verify_signatures_authenticated(committee, verify_params)?;
        Ok(VerifiedCertificate::new_from_verified(self))
    }

    pub fn verify_committee_sigs_only(&self, committee: &Committee) -> IotaResult {
        self.auth_sig().verify_secure(
            self.data(),
            Intent::iota_app(IntentScope::SenderSignedTransaction),
            committee,
        )
    }
}

pub type VerifiedCertificate =
    VerifiedEnvelope<SenderSignedTransaction, AuthorityStrongQuorumSignInfo>;
pub type TrustedCertificate =
    TrustedEnvelope<SenderSignedTransaction, AuthorityStrongQuorumSignInfo>;

#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize, PartialOrd, Ord, Hash)]
pub enum InputObjectKind {
    // A Move package, must be immutable.
    MovePackage(ObjectId),
    // A Move object, either immutable, or owned mutable.
    ImmOrOwnedMoveObject(ObjectReference),
    // A Move object that's shared and mutable.
    SharedMoveObject {
        id: ObjectId,
        initial_shared_version: Version,
        mutable: bool,
    },
}

impl InputObjectKind {
    pub fn object_id(&self) -> ObjectId {
        match self {
            Self::MovePackage(id) => *id,
            Self::ImmOrOwnedMoveObject(object_ref) => object_ref.object_id,
            Self::SharedMoveObject { id, .. } => *id,
        }
    }

    pub fn version(&self) -> Option<Version> {
        match self {
            Self::MovePackage(..) => None,
            Self::ImmOrOwnedMoveObject(object_ref) => Some(object_ref.version),
            Self::SharedMoveObject { .. } => None,
        }
    }

    pub fn object_not_found_error(&self) -> UserInputError {
        match *self {
            Self::MovePackage(package_id) => {
                UserInputError::DependentPackageNotFound { package_id }
            }
            Self::ImmOrOwnedMoveObject(object_ref) => UserInputError::ObjectNotFound {
                object_id: object_ref.object_id,
                version: Some(object_ref.version),
            },
            Self::SharedMoveObject { id, .. } => UserInputError::ObjectNotFound {
                object_id: id,
                version: None,
            },
        }
    }

    /// The error for an input that a validator drops at the consensus commit
    /// for good: every validator drops it there, and it cannot become
    /// available later.
    pub fn not_available_at_commit_error(&self) -> UserInputError {
        let (object_id, version) = match *self {
            Self::MovePackage(package_id) => (package_id, None),
            Self::ImmOrOwnedMoveObject(object_ref) => {
                (object_ref.object_id, Some(object_ref.version))
            }
            Self::SharedMoveObject { id, .. } => (id, None),
        };
        UserInputError::InputObjectNotAvailableAtCommit { object_id, version }
    }

    pub fn is_shared_object(&self) -> bool {
        matches!(self, Self::SharedMoveObject { .. })
    }

    pub fn is_mutable(&self) -> bool {
        match self {
            Self::MovePackage(..) => false,
            Self::ImmOrOwnedMoveObject(_) => true,
            Self::SharedMoveObject { mutable, .. } => *mutable,
        }
    }

    /// Merges another InputObjectKind into self.
    ///
    /// For shared objects, if either is mutable, the result is mutable. Fails
    /// if the IDs or initial versions do not match.
    /// For non-shared objects, fails if they are not equal.
    pub fn left_union_with_checks(&mut self, other: &InputObjectKind) -> UserInputResult<()> {
        match self {
            InputObjectKind::MovePackage(_) | InputObjectKind::ImmOrOwnedMoveObject(_) => {
                fp_ensure!(
                    self == other,
                    UserInputError::InconsistentInput {
                        object_id: other.object_id(),
                    }
                );
            }
            InputObjectKind::SharedMoveObject {
                id,
                initial_shared_version,
                mutable,
            } => match other {
                InputObjectKind::MovePackage(_) | InputObjectKind::ImmOrOwnedMoveObject(_) => {
                    fp_bail!(UserInputError::NotSharedObject)
                }
                InputObjectKind::SharedMoveObject {
                    id: other_id,
                    initial_shared_version: other_initial_shared_version,
                    mutable: other_mutable,
                } => {
                    fp_ensure!(id == other_id, UserInputError::SharedObjectIdMismatch);
                    fp_ensure!(
                        initial_shared_version == other_initial_shared_version,
                        UserInputError::SharedObjectStartingVersionMismatch
                    );

                    if !*mutable && *other_mutable {
                        *mutable = *other_mutable;
                    }
                }
            },
        }

        Ok(())
    }

    /// Checks that `self` and `other` are equal for non-shared objects.
    /// For shared objects, checks that IDs and initial versions match while
    /// mutability can be different.
    pub fn check_consistency_for_authentication(
        &self,
        other: &InputObjectKind,
    ) -> UserInputResult<()> {
        match self {
            InputObjectKind::MovePackage(_) | InputObjectKind::ImmOrOwnedMoveObject(_) => {
                fp_ensure!(
                    self == other,
                    UserInputError::InconsistentInput {
                        object_id: self.object_id()
                    }
                );
            }
            InputObjectKind::SharedMoveObject {
                id,
                initial_shared_version,
                mutable: _,
            } => match other {
                InputObjectKind::MovePackage(_) | InputObjectKind::ImmOrOwnedMoveObject(_) => {
                    fp_bail!(UserInputError::InconsistentInput {
                        object_id: self.object_id()
                    })
                }
                InputObjectKind::SharedMoveObject {
                    id: other_id,
                    initial_shared_version: other_initial_shared_version,
                    mutable: _,
                } => {
                    fp_ensure!(
                        id == other_id,
                        UserInputError::InconsistentInput { object_id: *id }
                    );
                    fp_ensure!(
                        initial_shared_version == other_initial_shared_version,
                        UserInputError::InconsistentInput { object_id: *id }
                    );
                }
            },
        }

        Ok(())
    }
}

/// The result of reading an object for execution. Because shared objects may be
/// deleted, one possible result of reading a shared object is that
/// ObjectReadResultKind::Deleted is returned.
#[derive(Clone, Debug)]
pub struct ObjectReadResult {
    pub input_object_kind: InputObjectKind,
    pub object: ObjectReadResultKind,
}

#[derive(Clone, PartialEq)]
pub enum ObjectReadResultKind {
    Object(Object),
    // The version of the object that the transaction intended to read, and the digest of the tx
    // that deleted it.
    DeletedSharedObject(Version, TransactionDigest),
    // A shared object in a cancelled transaction. The sequence number embeds cancellation reason.
    CancelledTransactionObject(Version),
    // A shared object that exists and is shared at the declared initial version, whose contents
    // were not read. Only post-consensus validation lists it: the version and contents a
    // validator holds there differ between validators, and execution reads the object at the
    // version consensus assigned.
    UnreadSharedObject,
}

impl std::fmt::Debug for ObjectReadResultKind {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            ObjectReadResultKind::Object(obj) => {
                write!(f, "Object({:?})", obj.object_ref())
            }
            ObjectReadResultKind::DeletedSharedObject(seq, digest) => {
                write!(f, "DeletedSharedObject({seq}, {digest})")
            }
            ObjectReadResultKind::CancelledTransactionObject(seq) => {
                write!(f, "CancelledTransactionObject({seq})")
            }
            ObjectReadResultKind::UnreadSharedObject => write!(f, "UnreadSharedObject"),
        }
    }
}

impl From<Object> for ObjectReadResultKind {
    fn from(object: Object) -> Self {
        Self::Object(object)
    }
}

impl ObjectReadResult {
    pub fn new(input_object_kind: InputObjectKind, object: ObjectReadResultKind) -> Self {
        if let (
            InputObjectKind::ImmOrOwnedMoveObject(_),
            ObjectReadResultKind::DeletedSharedObject(_, _),
        ) = (&input_object_kind, &object)
        {
            panic!("only shared objects can be DeletedSharedObject");
        }

        if let (
            InputObjectKind::ImmOrOwnedMoveObject(_),
            ObjectReadResultKind::CancelledTransactionObject(_),
        ) = (&input_object_kind, &object)
        {
            panic!("only shared objects can be CancelledTransactionObject");
        }

        if let (
            InputObjectKind::ImmOrOwnedMoveObject(_),
            ObjectReadResultKind::UnreadSharedObject,
        ) = (&input_object_kind, &object)
        {
            panic!("only shared objects can be UnreadSharedObject");
        }

        Self {
            input_object_kind,
            object,
        }
    }

    pub fn id(&self) -> ObjectId {
        self.input_object_kind.object_id()
    }

    pub fn as_object(&self) -> Option<&Object> {
        match &self.object {
            ObjectReadResultKind::Object(object) => Some(object),
            ObjectReadResultKind::DeletedSharedObject(_, _) => None,
            ObjectReadResultKind::CancelledTransactionObject(_) => None,
            ObjectReadResultKind::UnreadSharedObject => None,
        }
    }

    pub fn new_from_gas_object(gas: &Object) -> Self {
        let objref = gas.object_ref();
        Self {
            input_object_kind: InputObjectKind::ImmOrOwnedMoveObject(objref),
            object: ObjectReadResultKind::Object(gas.clone()),
        }
    }

    pub fn is_mutable(&self) -> bool {
        match (&self.input_object_kind, &self.object) {
            (InputObjectKind::MovePackage(_), _) => false,
            (InputObjectKind::ImmOrOwnedMoveObject(_), ObjectReadResultKind::Object(object)) => {
                !object.is_immutable()
            }
            (
                InputObjectKind::ImmOrOwnedMoveObject(_),
                ObjectReadResultKind::DeletedSharedObject(_, _),
            ) => unreachable!(),
            (
                InputObjectKind::ImmOrOwnedMoveObject(_),
                ObjectReadResultKind::CancelledTransactionObject(_),
            ) => unreachable!(),
            (
                InputObjectKind::ImmOrOwnedMoveObject(_),
                ObjectReadResultKind::UnreadSharedObject,
            ) => unreachable!(),
            (InputObjectKind::SharedMoveObject { mutable, .. }, _) => *mutable,
        }
    }

    pub fn is_shared_object(&self) -> bool {
        self.input_object_kind.is_shared_object()
    }

    pub fn is_deleted_shared_object(&self) -> bool {
        self.deletion_info().is_some()
    }

    pub fn deletion_info(&self) -> Option<(Version, TransactionDigest)> {
        match &self.object {
            ObjectReadResultKind::DeletedSharedObject(v, tx) => Some((*v, *tx)),
            _ => None,
        }
    }

    /// Return the object ref iff the object is an owned object (i.e. not
    /// shared, not immutable).
    pub fn get_owned_objref(&self) -> Option<ObjectReference> {
        match (&self.input_object_kind, &self.object) {
            (InputObjectKind::MovePackage(_), _) => None,
            (
                InputObjectKind::ImmOrOwnedMoveObject(objref),
                ObjectReadResultKind::Object(object),
            ) => {
                if object.is_immutable() {
                    None
                } else {
                    Some(*objref)
                }
            }
            (
                InputObjectKind::ImmOrOwnedMoveObject(_),
                ObjectReadResultKind::DeletedSharedObject(_, _),
            ) => unreachable!(),
            (
                InputObjectKind::ImmOrOwnedMoveObject(_),
                ObjectReadResultKind::CancelledTransactionObject(_),
            ) => unreachable!(),
            (
                InputObjectKind::ImmOrOwnedMoveObject(_),
                ObjectReadResultKind::UnreadSharedObject,
            ) => unreachable!(),
            (InputObjectKind::SharedMoveObject { .. }, _) => None,
        }
    }

    pub fn is_owned(&self) -> bool {
        self.get_owned_objref().is_some()
    }

    pub fn to_shared_input(&self) -> Option<SharedInput> {
        match self.input_object_kind {
            InputObjectKind::MovePackage(_) => None,
            InputObjectKind::ImmOrOwnedMoveObject(_) => None,
            InputObjectKind::SharedMoveObject { id, mutable, .. } => Some(match &self.object {
                ObjectReadResultKind::Object(obj) => SharedInput::Existing(obj.object_ref()),
                ObjectReadResultKind::DeletedSharedObject(seq, digest) => {
                    SharedInput::Deleted((id, *seq, mutable, *digest))
                }
                ObjectReadResultKind::CancelledTransactionObject(seq) => {
                    SharedInput::Cancelled((id, *seq))
                }
                ObjectReadResultKind::UnreadSharedObject => {
                    unreachable!("an unread shared object has no version to execute at")
                }
            }),
        }
    }

    pub fn get_previous_transaction(&self) -> Option<TransactionDigest> {
        match &self.object {
            ObjectReadResultKind::Object(obj) => Some(obj.previous_transaction),
            ObjectReadResultKind::DeletedSharedObject(_, digest) => Some(*digest),
            ObjectReadResultKind::CancelledTransactionObject(_) => None,
            ObjectReadResultKind::UnreadSharedObject => None,
        }
    }
}

/// The input objects that carry a cancelled transaction's cancellation
/// version, and therefore the objects reported to the client as the cause.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum CancelledObjects {
    /// The transaction's shared inputs, reported as the congested objects.
    SharedObjects(Vec<ObjectId>),
    /// The gas object of a transaction without shared inputs. Consensus only
    /// assigns a cancellation version there when the execution workers are
    /// congested, so no individual object is responsible and none is
    /// reported.
    GasObject,
}

#[derive(Clone)]
pub struct InputObjects {
    objects: Vec<ObjectReadResult>,
    /// Cancellation version assigned to the gas object when a transaction
    /// without shared inputs is cancelled by consensus (execution-worker
    /// congestion). Transactions with shared inputs carry cancellation
    /// versions on their shared object read results instead. Unlike those,
    /// the gas object itself is still read normally, because the cancelled
    /// execution must charge gas to it.
    gas_object_cancellation: Option<(ObjectId, Version)>,
}

impl std::fmt::Debug for InputObjects {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match &self.gas_object_cancellation {
            None => f.debug_list().entries(self.objects.iter()).finish(),
            Some(gas_object_cancellation) => f
                .debug_struct("InputObjects")
                .field("objects", &self.objects)
                .field("gas_object_cancellation", gas_object_cancellation)
                .finish(),
        }
    }
}

// An InputObjects new-type that has been verified by iota-transaction-checks,
// and can be safely passed to execution.
pub struct CheckedInputObjects(InputObjects);

// DO NOT CALL outside of iota-transaction-checks, genesis, or replay.
//
// CheckedInputObjects should really be defined in iota-transaction-checks so
// that we can make public construction impossible. But we can't do that because
// it would result in circular dependencies.
impl CheckedInputObjects {
    // Only called by iota-transaction-checks.
    pub fn new_with_checked_transaction_inputs(inputs: InputObjects) -> Self {
        Self(inputs)
    }

    // Only called when building the genesis transaction
    pub fn new_for_genesis(input_objects: Vec<ObjectReadResult>) -> Self {
        Self(InputObjects::new(input_objects))
    }

    // Only called from the replay tool.
    pub fn new_for_replay(input_objects: InputObjects) -> Self {
        Self(input_objects)
    }

    pub fn inner(&self) -> &InputObjects {
        &self.0
    }

    pub fn into_inner(self) -> InputObjects {
        self.0
    }
}

impl From<Vec<ObjectReadResult>> for InputObjects {
    fn from(objects: Vec<ObjectReadResult>) -> Self {
        Self::new(objects)
    }
}

impl InputObjects {
    pub fn new(objects: Vec<ObjectReadResult>) -> Self {
        Self {
            objects,
            gas_object_cancellation: None,
        }
    }

    /// Marks this transaction as cancelled via the version assigned to its
    /// gas object. Only valid for transactions without shared inputs; a
    /// transaction with shared inputs carries cancellation versions on its
    /// shared object read results.
    pub fn set_gas_object_cancellation(&mut self, gas_object_id: ObjectId, version: Version) {
        assert!(
            version.is_canceled(),
            "gas object cancellation must use a cancellation version, got {version:?}"
        );
        assert!(
            !self.objects.iter().any(|obj| obj.is_shared_object()),
            "transactions with shared inputs carry cancellation on their shared objects"
        );
        self.gas_object_cancellation = Some((gas_object_id, version));
    }

    pub fn gas_object_cancellation(&self) -> Option<(ObjectId, Version)> {
        self.gas_object_cancellation
    }

    pub fn len(&self) -> usize {
        self.objects.len()
    }

    pub fn is_empty(&self) -> bool {
        self.objects.is_empty()
    }

    pub fn contains_deleted_objects(&self) -> bool {
        self.objects
            .iter()
            .any(|obj| obj.is_deleted_shared_object())
    }

    // Returns the objects responsible for a transaction being cancelled, and the
    // corresponding reason for cancellation.
    pub fn get_cancelled_objects(&self) -> Option<(CancelledObjects, Version)> {
        if let Some((_, version)) = &self.gas_object_cancellation {
            return Some((CancelledObjects::GasObject, *version));
        }

        let mut contains_cancelled = false;
        let mut cancel_reason = None;
        let mut cancelled_objects = Vec::new();
        for obj in &self.objects {
            if let ObjectReadResultKind::CancelledTransactionObject(version) = obj.object {
                contains_cancelled = true;
                if version.is_congested() || version == Version::RANDOMNESS_UNAVAILABLE {
                    // Verify we don't have multiple cancellation reasons.
                    assert!(cancel_reason.is_none() || cancel_reason == Some(version));
                    cancel_reason = Some(version);
                    cancelled_objects.push(obj.id());
                }
            }
        }

        if !cancelled_objects.is_empty() {
            Some((
                CancelledObjects::SharedObjects(cancelled_objects),
                cancel_reason
                    .expect("there should be a cancel reason if there are cancelled objects"),
            ))
        } else {
            assert!(!contains_cancelled);
            None
        }
    }

    pub fn filter_owned_objects(&self) -> Vec<ObjectReference> {
        let owned_objects: Vec<_> = self
            .objects
            .iter()
            .filter_map(|obj| obj.get_owned_objref())
            .collect();

        trace!(
            num_mutable_objects = owned_objects.len(),
            "Checked locks and found mutable objects"
        );

        owned_objects
    }

    pub fn filter_shared_objects(&self) -> Vec<SharedInput> {
        self.objects
            .iter()
            .filter(|obj| obj.is_shared_object())
            .map(|obj| {
                obj.to_shared_input()
                    .expect("already filtered for shared objects")
            })
            .collect()
    }

    pub fn transaction_dependencies(&self) -> BTreeSet<TransactionDigest> {
        self.objects
            .iter()
            .filter_map(|obj| obj.get_previous_transaction())
            .collect()
    }

    pub fn mutable_inputs(&self) -> BTreeMap<ObjectId, (VersionDigest, Owner)> {
        self.objects
            .iter()
            .filter_map(
                |ObjectReadResult {
                     input_object_kind,
                     object,
                 }| match (input_object_kind, object) {
                    (InputObjectKind::MovePackage(_), _) => None,
                    (
                        InputObjectKind::ImmOrOwnedMoveObject(object_ref),
                        ObjectReadResultKind::Object(object),
                    ) => {
                        if object.is_immutable() {
                            None
                        } else {
                            Some((
                                object_ref.object_id,
                                ((object_ref.version, object_ref.digest), object.owner),
                            ))
                        }
                    }
                    (
                        InputObjectKind::ImmOrOwnedMoveObject(_),
                        ObjectReadResultKind::DeletedSharedObject(_, _),
                    ) => {
                        unreachable!()
                    }
                    (
                        InputObjectKind::SharedMoveObject { .. },
                        ObjectReadResultKind::DeletedSharedObject(_, _),
                    ) => None,
                    (
                        InputObjectKind::SharedMoveObject { mutable, .. },
                        ObjectReadResultKind::Object(object),
                    ) => {
                        if *mutable {
                            let oref = object.object_ref();
                            Some((oref.object_id, ((oref.version, oref.digest), object.owner)))
                        } else {
                            None
                        }
                    }
                    (
                        InputObjectKind::ImmOrOwnedMoveObject(_),
                        ObjectReadResultKind::CancelledTransactionObject(_),
                    ) => {
                        unreachable!()
                    }
                    (
                        InputObjectKind::SharedMoveObject { .. },
                        ObjectReadResultKind::CancelledTransactionObject(_),
                    ) => None,
                    (
                        InputObjectKind::ImmOrOwnedMoveObject(_),
                        ObjectReadResultKind::UnreadSharedObject,
                    ) => {
                        unreachable!()
                    }
                    (
                        InputObjectKind::SharedMoveObject { .. },
                        ObjectReadResultKind::UnreadSharedObject,
                    ) => {
                        unreachable!("an unread shared object has no version, digest or owner")
                    }
                },
            )
            .collect()
    }

    /// The version to set on objects created by the computation that `self` is
    /// input to. Guaranteed to be strictly greater than the versions of all
    /// input objects and objects received in the transaction.
    pub fn lamport_timestamp(&self, receiving_objects: &[ObjectReference]) -> Version {
        let input_versions = self
            .objects
            .iter()
            .filter_map(|object| match &object.object {
                ObjectReadResultKind::Object(object) => {
                    object.data.as_opt_struct().map(MoveStruct::version)
                }
                ObjectReadResultKind::DeletedSharedObject(v, _) => Some(*v),
                ObjectReadResultKind::CancelledTransactionObject(_) => None,
                ObjectReadResultKind::UnreadSharedObject => {
                    unreachable!("an unread shared object has no version to exceed")
                }
            })
            .chain(
                receiving_objects
                    .iter()
                    .map(|object_ref| object_ref.version),
            );

        Version::lamport_increment(input_versions).unwrap()
    }

    pub fn object_kinds(&self) -> impl Iterator<Item = &InputObjectKind> {
        self.objects.iter().map(
            |ObjectReadResult {
                 input_object_kind, ..
             }| input_object_kind,
        )
    }

    pub fn into_object_map(self) -> BTreeMap<ObjectId, Object> {
        self.objects
            .into_iter()
            .filter_map(|o| o.as_object().map(|object| (o.id(), object.clone())))
            .collect()
    }

    pub fn push(&mut self, object: ObjectReadResult) {
        self.objects.push(object);
    }

    // If it contains then it returns the ObjectReadResult
    pub fn find_object_id_mut(&mut self, object_id: ObjectId) -> Option<&mut ObjectReadResult> {
        self.objects.iter_mut().find(|o| o.id() == object_id)
    }

    pub fn iter(&self) -> impl Iterator<Item = &ObjectReadResult> {
        self.objects.iter()
    }

    pub fn iter_objects(&self) -> impl Iterator<Item = &Object> {
        self.objects.iter().filter_map(|o| o.as_object())
    }
}

// Result of attempting to read a receiving object (currently only at signing
// time). Because an object may have been previously received and deleted, the
// result may be ReceivingObjectReadResultKind::PreviouslyReceivedObject.
#[derive(Clone, Debug)]
pub enum ReceivingObjectReadResultKind {
    Object(Object),
    // The object was received by some other transaction, and we were not able to read it
    PreviouslyReceivedObject,
}

impl ReceivingObjectReadResultKind {
    pub fn as_object(&self) -> Option<&Object> {
        match &self {
            Self::Object(object) => Some(object),
            Self::PreviouslyReceivedObject => None,
        }
    }
}

pub struct ReceivingObjectReadResult {
    pub object_ref: ObjectReference,
    pub object: ReceivingObjectReadResultKind,
}

impl ReceivingObjectReadResult {
    pub fn new(object_ref: ObjectReference, object: ReceivingObjectReadResultKind) -> Self {
        Self { object_ref, object }
    }

    pub fn is_previously_received(&self) -> bool {
        matches!(
            self.object,
            ReceivingObjectReadResultKind::PreviouslyReceivedObject
        )
    }
}

impl From<Object> for ReceivingObjectReadResultKind {
    fn from(object: Object) -> Self {
        Self::Object(object)
    }
}

pub struct ReceivingObjects {
    pub objects: Vec<ReceivingObjectReadResult>,
}

impl ReceivingObjects {
    pub fn iter(&self) -> impl Iterator<Item = &ReceivingObjectReadResult> {
        self.objects.iter()
    }

    pub fn iter_objects(&self) -> impl Iterator<Item = &Object> {
        self.objects.iter().filter_map(|o| o.object.as_object())
    }
}

impl From<Vec<ReceivingObjectReadResult>> for ReceivingObjects {
    fn from(objects: Vec<ReceivingObjectReadResult>) -> Self {
        Self { objects }
    }
}

impl Display for CertifiedTransaction {
    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
        let mut writer = String::new();
        writeln!(writer, "Transaction Hash: {:?}", self.digest())?;
        writeln!(
            writer,
            "Signed Authorities Bitmap : {:?}",
            self.auth_sig().signers_map
        )?;
        write!(writer, "{}", self.data().transaction().kind())?;
        write!(f, "{writer}")
    }
}

/// TransactionKey uniquely identifies a transaction across all epochs.
/// Note that a single transaction may have multiple keys, for example a
/// RandomnessStateUpdate could be identified by both `Digest` and
/// `RandomnessRound`.
#[derive(Clone, Copy, Debug, Eq, PartialEq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
pub enum TransactionKey {
    Digest(TransactionDigest),
    RandomnessRound(EpochId, RandomnessRound),
}

impl TransactionKey {
    pub fn unwrap_digest(&self) -> &TransactionDigest {
        match self {
            TransactionKey::Digest(d) => d,
            _ => panic!("called expect_digest on a non-Digest TransactionKey: {self:?}"),
        }
    }

    pub fn as_digest(&self) -> Option<&TransactionDigest> {
        match self {
            TransactionKey::Digest(d) => Some(d),
            _ => None,
        }
    }
}