miden-ntx-builder 0.17.2

use miden_node_proto::DecodeMessageExt;
use std::collections::BTreeSet;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::{Duration, Instant};

use tokio::sync::RwLock;

use backon::ExponentialBuilder;
use futures::stream::{BoxStream, TryStreamExt};
use futures::{Stream, StreamExt};
use miden_node_proto::clients::{Builder, RpcClient as InnerRpcClient};
use miden_node_proto::VerifyWith;
use miden_node_proto::domain::account::{
    AccountDetails, GetAccountResponse, AccountVaultDetails, StorageMapEntries
};
use miden_node_proto::domain::encryption::{
    TransactionInputsSealer,
    TrustedTransactionEncryptionState,
};
use miden_node_proto::domain::protocol_config::ensure_protocol_config_is_present_and_matches_header;
use miden_node_proto::errors::ConversionError;
use miden_node_proto::generated::miden::node::v1::get_account_request::account_detail_request::{StorageMapDetailRequest, StorageMapDetailRequests, StorageRequest, storage_map_detail_request};
use miden_node_proto::generated::miden::node::v1::get_account_request::account_detail_request::storage_map_detail_request::MapKeys;
use miden_node_proto::generated::miden::node::v1::{
    GetBlockHeaderByNumberRequest,
    GetBlockHeaderByNumberResponse,
    BlockSubscriptionRequest,
};
use miden_node_proto::generated::{self as proto};
use miden_node_tracing::ErrorReport;
use miden_node_utils::retry::{self, Retryable};
use miden_node_tracing::{debug, info, miden_instrument, warn};
use miden_protocol::Word;
use miden_protocol::account::{
    AccountId,
    PartialAccount,
    PartialStorage,
    StorageMapKey,
    StorageMapWitness,
    StorageSlotName,
};
use miden_protocol::asset::{Asset, AssetVault, AssetId, AssetWitness, PartialVault};
use miden_protocol::block::{BlockNumber, SignedBlock};
use miden_protocol::crypto::dsa::ecdsa_k256_keccak::PublicKey as ValidatorPublicKey;
use miden_protocol::note::NoteScript;
use miden_protocol::protocol_config::ProtocolConfig;
use miden_protocol::transaction::{AccountInputs, ProvenTransaction, TransactionInputs};
use miden_protocol::utils::serde::Serializable;
use thiserror::Error;
use tonic::Status;
use tonic::metadata::AsciiMetadataValue;
use url::Url;

use crate::COMPONENT;

// RPC CLIENT
// ================================================================================================

/// A signed block paired with its stream metadata.
pub(crate) type BlockSubscriptionEvent = (SignedBlock, BlockNumber, Option<ProtocolConfig>);

/// A decoded block-subscription event.
type BlockSubscriptionItem = Result<BlockSubscriptionEvent, RpcError>;

/// Tracks the active configuration within one block-subscription connection.
///
/// The node sends a baseline with the first response. It then omits the configuration until its
/// commitment changes. A new connection creates a new tracker and requires a new baseline.
#[derive(Default)]
struct ProtocolConfigTracker {
    current: Option<ProtocolConfig>,
}

impl ProtocolConfigTracker {
    fn validate(
        &mut self,
        header: &miden_protocol::block::BlockHeader,
        config: Option<&ProtocolConfig>,
    ) -> Result<(), RpcError> {
        if let Some(config) = config {
            if config.to_commitment() != header.protocol_config_commitment() {
                return Err(RpcError::InvalidResponse(
                    "block protocol config commitment does not match its header".into(),
                ));
            }
            self.current = Some(config.clone());
            return Ok(());
        }

        let current = self.current.as_ref().ok_or_else(|| {
            RpcError::InvalidResponse(
                "first block subscription response is missing protocol config".into(),
            )
        })?;
        if current.to_commitment() != header.protocol_config_commitment() {
            return Err(RpcError::InvalidResponse(
                "block subscription omitted a changed protocol config".into(),
            ));
        }

        Ok(())
    }
}

/// Delay between block-subscription reconnect attempts, paced so a node that immediately closes the
/// connection cannot spin the reconnect loop. Connection *failures* are already backed off
/// exponentially inside [`RpcClient::block_subscription_with_retry`].
const RECONNECT_DELAY: Duration = Duration::from_secs(5);

/// How long a single `stream.next()` poll waits before re-evaluating liveness.
const BLOCK_POLL_TIMEOUT: Duration = Duration::from_secs(10);

/// Maximum time without a committed block before treating the subscription as stalled and forcing a
/// reconnect.
///
/// This is a defensive backstop for a silently dropped connection that never surfaces an error on
/// its own (client HTTP/2 keepalive is the primary, faster detector). It must be comfortably larger
/// than the node's block interval so a legitimately quiet chain is not mistaken for a stall.
const STALL_TIMEOUT: Duration = Duration::from_mins(2);

/// Thin wrapper around the node RPC gRPC service that the ntx-builder uses to consume the
/// committed-block subscription stream.
#[derive(Clone, Debug)]
pub struct RpcClient {
    inner: InnerRpcClient,
    /// Backoff schedule applied to repeated `block_subscription` connection attempts. Built once at
    /// construction time and cloned cheaply on each retry loop.
    backoff: ExponentialBuilder,
    /// Genesis commitment of the network being submitted to, bound into the associated data of
    /// sealed transaction inputs.
    genesis_commitment: Word,
    /// Cached sealer for transaction inputs, fetched on first submission.
    sealer: Arc<RwLock<Option<TransactionInputsSealer>>>,
    /// Validator signing keys read from the genesis block at bootstrap.
    trusted_validator_signing_keys: Arc<[ValidatorPublicKey]>,
}

impl RpcClient {
    /// Creates a new client with a lazy connection to the node RPC endpoint.
    ///
    /// `request_timeout` bounds each gRPC request, including establishment of the long-lived block
    /// subscription but not the lifetime of its response stream.
    ///
    /// `backoff_initial` / `backoff_max` configure the exponential backoff schedule applied to
    /// `block_subscription` retries (the only operation that retries today).
    pub fn new(
        rpc_url: Url,
        genesis_commitment: Word,
        trusted_validator_signing_keys: Vec<ValidatorPublicKey>,
        request_timeout: Duration,
        backoff_initial: Duration,
        backoff_max: Duration,
    ) -> anyhow::Result<Self> {
        Self::new_with_auth(
            rpc_url,
            None,
            genesis_commitment,
            trusted_validator_signing_keys,
            request_timeout,
            backoff_initial,
            backoff_max,
        )
    }

    /// Creates a new client with an optional metadata header for internal RPC authentication.
    ///
    /// `genesis_commitment` is sent as the `genesis` parameter of the `Accept` header so that the
    /// node accepts write RPCs such as `SubmitProvenTx`, which require a matching genesis.
    pub fn new_with_auth(
        rpc_url: Url,
        rpc_auth_header_value: Option<AsciiMetadataValue>,
        genesis_commitment: Word,
        trusted_validator_signing_keys: Vec<ValidatorPublicKey>,
        request_timeout: Duration,
        backoff_initial: Duration,
        backoff_max: Duration,
    ) -> anyhow::Result<Self> {
        info!(
            target: COMPONENT,
            "Initializing RPC client",
            dependency.name = "rpc",
            dependency.endpoint = rpc_url.to_string()
        );

        let builder = Builder::new(rpc_url)
            .with_tls()?
            .with_timeout(request_timeout)
            .without_metadata_version()
            .with_metadata_genesis(genesis_commitment);
        let builder = match rpc_auth_header_value {
            Some(value) => builder.with_auth_header_value(value),
            None => builder.without_auth_header(),
        };
        let rpc = builder.with_otel_context_injection().connect_lazy::<InnerRpcClient>();

        let backoff = retry::exponential(backoff_initial, backoff_max);

        Ok(Self {
            inner: rpc,
            backoff,
            genesis_commitment,
            sealer: Arc::new(RwLock::new(None)),
            trusted_validator_signing_keys: trusted_validator_signing_keys.into(),
        })
    }

    /// Returns a sealer for transaction inputs, fetching the encryption key if the cache is empty.
    pub(crate) async fn sealer(&self) -> Result<TransactionInputsSealer, Status> {
        if let Some(sealer) = self.sealer.read().await.clone() {
            return Ok(sealer);
        }

        let key = self
            .inner
            .clone()
            .get_transaction_encryption_key(
                proto::miden::node::v1::GetTransactionEncryptionKeyRequest {},
            )
            .await?
            .into_inner()
            .key
            .ok_or_else(|| tonic::Status::internal("missing transaction encryption key"))?;
        let verified = key
            .verify_with(TrustedTransactionEncryptionState::new(
                self.genesis_commitment,
                &self.trusted_validator_signing_keys,
            ))
            .map_err(|err| {
                Status::failed_precondition(
                    err.as_report_context("Untrusted transaction encryption key"),
                )
            })?;
        let sealer = TransactionInputsSealer::new(verified);

        let mut cached = self.sealer.write().await;
        if let Some(sealer) = cached.clone() {
            return Ok(sealer);
        }
        *cached = Some(sealer.clone());
        Ok(sealer)
    }

    /// Loads and verifies the active configuration for a persisted local header.
    pub(crate) async fn protocol_config_for_header(
        &self,
        expected_header: &miden_protocol::block::BlockHeader,
    ) -> Result<ProtocolConfig, RpcError> {
        (|| async {
            let response = self
                .inner
                .clone()
                .get_block_header_by_number(startup_header_request(expected_header.block_num()))
                .await
                .map_err(RpcError::GrpcClientError)?
                .into_inner();
            decode_startup_header_response(response, expected_header)
        })
        .retry(self.backoff)
        .when(|err| matches!(err, RpcError::GrpcClientError(_)))
        .notify(|err, dur| {
            warn!(
                err,
                target: COMPONENT,
                "RPC request failed while verifying the persisted chain tip, retrying",
                retry.delay_ms = dur.as_millis() as u64
            );
        })
        .await
    }

    /// Opens a committed-block subscription starting at `block_from`, retrying indefinitely with
    /// the client's configured exponential backoff while the initial connection attempt fails.
    ///
    /// Returns a stream that decodes each [`proto::miden::node::v1::BlockSubscriptionResponse`] into a block, the committed
    /// chain tip, and an optional protocol configuration. The configuration is present for the
    /// first response and for each transition. The committed chain tip is the latest block the node
    /// believes is committed when it emits the response.
    #[miden_instrument(
        target = COMPONENT,
        name = "rpc.client.block_subscription_with_retry",
        fields(
            block.from = block_from,
        ),
        err,
    )]
    pub async fn block_subscription_with_retry(
        &self,
        block_from: BlockNumber,
    ) -> Result<BoxStream<'static, BlockSubscriptionItem>, RpcError> {
        (|| async move {
            let request =
                tonic::Request::new(BlockSubscriptionRequest { block_from: block_from.as_u32() });
            let stream = self
                .inner
                .clone()
                .block_subscription(request)
                .await
                .map_err(RpcError::GrpcClientError)?
                .into_inner();

            // Box the stream so its type is named and explicitly `'static` (it owns the cloned
            // client, borrowing nothing from `self`). This keeps the return type from capturing
            // `&self`, so callers like `block_subscription_reconnecting` can store it freely.
            let decoded = stream
                .map_err(RpcError::GrpcClientError)
                .and_then(|response| async move {
                    response
                        // SAFETY: The builder verifies the block against its trusted parent before
                        // it writes block effects or notifies account actors.
                        .decode_and_build_unchecked()
                        .map_err(RpcError::Conversion)
                })
                .scan(ProtocolConfigTracker::default(), |tracker, item| {
                    let item = item.and_then(|event| {
                        tracker.validate(event.0.header(), event.2.as_ref())?;
                        Ok(event)
                    });
                    std::future::ready(Some(item))
                });

            Ok(decoded.boxed())
        })
        .retry(self.backoff)
        .notify(|err: &RpcError, dur| {
            warn!(
                err,
                target: COMPONENT,
                "RPC connection failed while opening block subscription, retrying",
                retry.delay_ms = dur.as_millis() as u64
            );
        })
        .await
    }

    /// Opens a committed-block subscription that transparently reconnects whenever the gRPC
    /// connection is closed.
    ///
    /// Each reconnection resumes from the block after the last one yielded, so no committed block
    /// is skipped or replayed. A closed or errored subscription is logged and re-opened, paced by
    /// the client's exponential backoff schedule.
    pub fn block_subscription_reconnecting(
        &self,
        block_from: BlockNumber,
    ) -> impl Stream<Item = BlockSubscriptionItem> + Send + 'static {
        let client = self.clone();

        futures::stream::unfold(
            (client, block_from, None, Instant::now()),
            |(client, mut next_from, mut inner, mut last_block)| async move {
                loop {
                    // Open the subscription if we don't hold a live one. The connect itself retries
                    // indefinitely with exponential backoff.
                    let stream = match &mut inner {
                        Some(stream) => stream,
                        None => match client.block_subscription_with_retry(next_from).await {
                            Ok(stream) => {
                                info!(
                                    target: COMPONENT,
                                    "block subscription connected",
                                    block.from = next_from
                                );
                                // Reset the stall clock so time spent (re)connecting is not counted
                                // against the next block's arrival.
                                last_block = Instant::now();
                                inner.insert(stream)
                            },
                            Err(err) => {
                                warn!(
                                    &err,
                                    target: COMPONENT,
                                    "failed to open block subscription, retrying",
                                    block.from = next_from
                                );
                                tokio::time::sleep(RECONNECT_DELAY).await;
                                continue;
                            },
                        },
                    };

                    // Poll on a short timeout so a silently dropped connection stays observable.
                    // Each quiet poll emits a liveness log; once no block has arrived for
                    // `STALL_TIMEOUT` the subscription is treated as stalled and reconnected.
                    match tokio::time::timeout(BLOCK_POLL_TIMEOUT, stream.next()).await {
                        Ok(Some(Ok((block, committed_tip, protocol_config)))) => {
                            next_from = block.header().block_num().child();
                            last_block = Instant::now();
                            return Some((
                                Ok((block, committed_tip, protocol_config)),
                                (client, next_from, inner, last_block),
                            ));
                        },
                        Ok(Some(Err(err))) => warn!(
                            &err,
                            target: COMPONENT,
                            "block subscription failed, reconnecting",
                            block.from = next_from
                        ),
                        Ok(None) => warn!(
                            target: COMPONENT,
                            "block subscription closed by node, reconnecting",
                            block.from = next_from
                        ),
                        Err(_elapsed) => {
                            let idle = last_block.elapsed();
                            if idle < STALL_TIMEOUT {
                                // Quiet but not yet stalled: emit a liveness signal and keep
                                // polling the same stream instead of reconnecting.
                                debug!(
                                    target: COMPONENT,
                                    "no block received recently; subscription still open",
                                    block.from = next_from,
                                    subscription.idle_ms = idle.as_millis() as u64
                                );
                                continue;
                            }
                            warn!(
                                target: COMPONENT,
                                "no block received within stall timeout; treating subscription as stalled, reconnecting",
                                block.from = next_from,
                                subscription.idle_ms = idle.as_millis() as u64,
                                subscription.stall_timeout_ms = STALL_TIMEOUT.as_millis() as u64
                            );
                        },
                    }

                    // Reached on an error/close/stall branch: discard the stream, pace the
                    // reconnect, and resume from the next un-applied block on the following
                    // iteration.
                    inner = None;
                    tokio::time::sleep(RECONNECT_DELAY).await;
                }
            },
        )
    }

    #[miden_instrument(
        target = COMPONENT,
        name = "ntx.rpc.client.submit_proven_tx",
        err,
    )]
    pub async fn submit_proven_tx(
        &self,
        proven_tx: &ProvenTransaction,
        tx_inputs: &TransactionInputs,
    ) -> Result<(), Status> {
        let transaction: proto::transaction::ProvenTransaction = proven_tx.into();
        let transaction_inputs = tx_inputs.to_bytes();
        let tx_id = proven_tx.id();
        let stale_key = AtomicBool::new(false);

        (|| {
            let mut client = self.inner.clone();
            let transaction = transaction.clone();
            let transaction_inputs = transaction_inputs.clone();
            let stale_key = &stale_key;
            async move {
                if stale_key.swap(false, Ordering::Relaxed) {
                    *self.sealer.write().await = None;
                }

                let sealer = self.sealer().await?;
                let sealed = sealer.seal(tx_id, &transaction_inputs).map_err(|err| {
                    Status::failed_precondition(
                        err.as_report_context("Failed to seal the transaction inputs"),
                    )
                })?;
                client
                    .submit_proven_tx(proto::miden::node::v1::SubmitProvenTxRequest {
                        submission: Some(proto::submission::ProvenTransactionSubmission {
                            transaction: Some(transaction),
                            sealed_transaction_inputs: Some(sealed),
                        }),
                    })
                    .await
            }
        })
        .retry(retry::constant(Duration::ZERO, Some(1)))
        .when(|status: &Status| status.code() == tonic::Code::FailedPrecondition)
        .notify(|status: &Status, _| {
            stale_key.store(true, Ordering::Relaxed);
            warn!(
                status,
                target: COMPONENT,
                "Transaction inputs rejected as stale, refreshing the encryption key and retrying",
                transaction.id = tx_id
            );
        })
        .await
        .map(|_| ())
    }
}

fn startup_header_request(block_num: BlockNumber) -> GetBlockHeaderByNumberRequest {
    GetBlockHeaderByNumberRequest {
        block_num: Some(block_num.as_u32()),
        include_mmr_proof: None,
        include_protocol_config: Some(true),
    }
}

fn decode_startup_header_response(
    response: GetBlockHeaderByNumberResponse,
    expected_header: &miden_protocol::block::BlockHeader,
) -> Result<ProtocolConfig, RpcError> {
    let header: miden_protocol::block::BlockHeader = response
        .block_header
        .ok_or_else(|| RpcError::InvalidResponse("header response is missing block header".into()))?
        // SAFETY: The commitment check below binds this header to the persisted local header.
        .decode_and_build_unchecked()
        .map_err(RpcError::Conversion)?;
    if header.commitment() != expected_header.commitment() {
        return Err(RpcError::InvalidResponse(
            "remote header does not match the persisted local header".into(),
        ));
    }

    ensure_protocol_config_is_present_and_matches_header(response.protocol_config, &header)
        .map_err(RpcError::Conversion)
}

// ACTOR-PATH METHODS
// ================================================================================================
//
// Required endpoint implementations for the NTX `DataStore` implementation
impl RpcClient {
    /// Fetches the transaction inputs for a specific account.
    ///
    /// These inputs reference a specific `block_num`, and include a minimal partial account,
    /// plus its witness.
    pub async fn get_account_inputs(
        &self,
        account_id: AccountId,
        block_num: BlockNumber,
    ) -> Result<AccountInputs, RpcError> {
        // Only request account code
        let request = proto::miden::node::v1::GetAccountRequest {
            account_id: Some(account_id.into()),
            block_num: Some(block_num.into()),
            // TODO: should these commitments be cached on the NTX builder?
            details: Some(proto::miden::node::v1::get_account_request::AccountDetailRequest {
                code_commitment: Some(Word::default().into()),
                asset_vault_commitment: None, //
                storage_request: None,
            }),
        };

        let response = self.get_account(request).await?;
        let details = response.details.as_ref().ok_or_else(|| {
            RpcError::InvalidResponse("response did not include account details".into())
        })?;
        let partial_account = build_minimal_partial_account(details)?;

        Ok(AccountInputs::new(partial_account, response.witness))
    }

    /// Fetches asset vault witnesses for the given keys at the reference block.
    pub async fn get_vault_asset_witnesses(
        &self,
        account_id: AccountId,
        vault_keys: BTreeSet<AssetId>,
        block_num: Option<BlockNumber>,
    ) -> Result<Vec<AssetWitness>, RpcError> {
        if vault_keys.is_empty() {
            return Ok(Vec::new());
        }

        let request = proto::miden::node::v1::GetAccountRequest {
            account_id: Some(account_id.into()),
            block_num: block_num.map(Into::into),
            details: Some(proto::miden::node::v1::get_account_request::AccountDetailRequest {
                code_commitment: None,
                asset_vault_commitment: Some(Word::default().into()),
                storage_request: None,
            }),
        };

        let response = self.get_account(request).await?;
        let assets: Vec<Asset> = match response.details.map(|details| details.vault_details) {
            Some(AccountVaultDetails::Assets(assets)) => assets,
            Some(AccountVaultDetails::LimitExceeded) => {
                // NOTE: in the tx kernel, `get_vault_asset_witnesses` is called either for single
                // asset keys, or when pre-loading all the assets related to input notes involved in
                // the transaction. This should never exceed the maximum amount of keys you can
                // request to RPC, but this needs double-checking. If it able to exceed them,
                // batching needs to be implemented as a workaround.
                panic!("should never exceed maximum number of requested keys")
            },
            None => Vec::new(),
        };

        let vault =
            AssetVault::new(&assets).map_err(|err| RpcError::InvalidResponse(err.as_report()))?;

        Ok(vault_keys.into_iter().map(|key| vault.open(key)).collect())
    }

    /// Fetches a storage map witness for a single key at the reference block.
    pub async fn get_storage_map_witness(
        &self,
        account_id: AccountId,
        slot_name: StorageSlotName,
        map_key: StorageMapKey,
        block_num: Option<BlockNumber>,
    ) -> Result<StorageMapWitness, RpcError> {
        let request = proto::miden::node::v1::GetAccountRequest {
            account_id: Some(account_id.into()),
            block_num: block_num.map(Into::into),
            details: Some(proto::miden::node::v1::get_account_request::AccountDetailRequest {
                code_commitment: None,
                asset_vault_commitment: None,
                storage_request: Some(StorageRequest::StorageMaps(StorageMapDetailRequests {
                    storage_maps: vec![StorageMapDetailRequest {
                        slot_name: slot_name.to_string(),
                        slot_data: Some(storage_map_detail_request::SlotData::MapKeys(MapKeys {
                            map_keys: vec![map_key.as_word().into()],
                        })),
                    }],
                })),
            }),
        };

        let response = self.get_account(request).await?;
        let details = response.details.as_ref().ok_or_else(|| {
            RpcError::InvalidResponse("response did not include account details".into())
        })?;

        let map_details = details
            .storage_details
            .map_details
            .iter()
            .find(|detail| detail.slot_name == slot_name)
            .ok_or_else(|| {
                RpcError::InvalidResponse(format!(
                    "response is missing storage map details for slot {slot_name}"
                ))
            })?;

        let StorageMapEntries::PartialMap { map_keys, partial_smt } = &map_details.entries else {
            return Err(RpcError::InvalidResponse(
                "response did not include a partial storage map".into(),
            ));
        };

        if !map_keys.contains(&map_key) {
            return Err(RpcError::InvalidResponse(
                "response partial storage map did not include the requested key".into(),
            ));
        }

        let proof = partial_smt.open(&map_key.hash().as_word()).map_err(|err| {
            RpcError::InvalidResponse(format!(
                "response did not track the requested storage map key: {err}"
            ))
        })?;

        StorageMapWitness::new(proof, [map_key])
            .map_err(|err| RpcError::InvalidResponse(err.as_report()))
    }

    /// Fetches a note script by its root, returning `None` if the node does not know it.
    #[miden_instrument(
        target = COMPONENT,
        name = "ntx.rpc.client.get_note_script_by_root",
        err,
    )]
    pub async fn get_note_script_by_root(
        &self,
        script_root: Word,
    ) -> Result<Option<NoteScript>, RpcError> {
        let request =
            proto::miden::node::v1::GetNoteScriptByRootRequest { root: Some(script_root.into()) };

        self.inner
            .clone()
            .get_note_script_by_root(request)
            .await
            .map_err(RpcError::GrpcClientError)?
            .into_inner()
            .decode_and_verify()
            .map_err(RpcError::Conversion)
    }

    /// Issues a `GetAccount` request and decodes the response into the domain
    /// [`GetAccountResponse`].
    async fn get_account(
        &self,
        request: proto::miden::node::v1::GetAccountRequest,
    ) -> Result<GetAccountResponse, RpcError> {
        let response = self
            .inner
            .clone()
            .get_account(request)
            .await
            .map_err(RpcError::GrpcClientError)?
            .into_inner();

        response.decode_and_verify().map_err(RpcError::Conversion)
    }
}

/// Builds a minimal partial account from account details.
fn build_minimal_partial_account(details: &AccountDetails) -> Result<PartialAccount, RpcError> {
    let account_code = details
        .account_code
        .clone()
        .ok_or_else(|| RpcError::InvalidResponse("response did not include account code".into()))?;

    let partial_storage = PartialStorage::new(details.storage_details.header.clone(), [])
        .map_err(|err| RpcError::InvalidResponse(err.as_report()))?;

    let partial_vault = PartialVault::new(details.account_header.vault_root());

    PartialAccount::new(
        details.account_header.id(),
        details.account_header.nonce(),
        account_code,
        partial_storage,
        partial_vault,
        None,
    )
    .map_err(|err| RpcError::InvalidResponse(err.as_report()))
}

// RPC ERROR
// ================================================================================================

#[derive(Debug, Error)]
pub enum RpcError {
    #[error("RPC gRPC call failed")]
    GrpcClientError(#[source] tonic::Status),
    #[error("failed to convert RPC response")]
    Conversion(#[source] ConversionError),
    #[error("invalid RPC response: {0}")]
    InvalidResponse(String),
}

#[cfg(test)]
mod protocol_config_tests {
    use miden_node_proto::generated::miden::node::v1::{
        BlockSubscriptionResponse,
        GetBlockHeaderByNumberResponse,
    };
    use miden_node_proto::generated::protocol_config::ProtocolConfig as ProtoProtocolConfig;
    use miden_node_proto::{BuildUnchecked, DecodeMessage};
    use miden_node_store::genesis::GenesisState;
    use miden_node_utils::fee::{test_fee_params, test_protocol_config};
    use miden_protocol::Word;
    use miden_protocol::block::{BlockHeader, BlockNumber, FeeParameters};
    use miden_protocol::protocol_config::ProtocolConfig;

    use super::{ProtocolConfigTracker, decode_startup_header_response, startup_header_request};
    use crate::test_utils::mock_genesis_block;

    fn valid_genesis_block() -> miden_protocol::block::SignedBlock {
        GenesisState::new(
            Vec::new(),
            test_fee_params(),
            0,
            mock_genesis_block().header().validator_config().clone(),
            test_protocol_config(),
        )
        .into_block()
        .expect("test genesis block must build")
        .inner()
        .clone()
    }

    fn header_for_config(block_num: u32, config: &ProtocolConfig) -> BlockHeader {
        let fixture = mock_genesis_block();
        BlockHeader::new(
            Word::empty(),
            BlockNumber::from(block_num),
            Word::empty(),
            Word::empty(),
            Word::empty(),
            Word::empty(),
            Word::empty(),
            fixture.header().validator_config().clone(),
            FeeParameters::new(0),
            config.to_commitment(),
            None,
            block_num,
        )
    }

    fn other_protocol_config() -> ProtocolConfig {
        use miden_protocol::asset::AssetId;
        use miden_protocol::testing::account_id::ACCOUNT_ID_PUBLIC_FUNGIBLE_FAUCET_1;

        ProtocolConfig::current(AssetId::new_fungible(
            ACCOUNT_ID_PUBLIC_FUNGIBLE_FAUCET_1.try_into().unwrap(),
        ))
        .unwrap()
    }

    /// A malformed configuration must stop the stream item before the builder can apply its block.
    #[test]
    fn subscription_decoder_rejects_malformed_protocol_config() {
        let block = valid_genesis_block();
        let response = BlockSubscriptionResponse {
            block: Some(block.into()),
            committed_chain_tip: 0,
            protocol_config: Some(ProtoProtocolConfig::default()),
        };

        let error = response
            .decode_fields()
            .and_then(BuildUnchecked::build_unchecked)
            .expect_err("a malformed protocol config must be rejected");

        assert!(error.to_string().contains("protocol_config"), "unexpected error: {error}");
    }

    /// A valid streamed configuration must remain attached to its decoded block event.
    #[test]
    fn subscription_decoder_preserves_optional_protocol_config() {
        let block = valid_genesis_block();
        for config in [None, Some(test_protocol_config())] {
            let response = BlockSubscriptionResponse {
                block: Some((&block).into()),
                committed_chain_tip: 0,
                protocol_config: config.as_ref().map(Into::into),
            };
            let (_, _, decoded_config) =
                response.decode_fields().and_then(BuildUnchecked::build_unchecked).unwrap();
            assert_eq!(decoded_config, config);
        }
    }

    #[test]
    fn subscription_decoder_rejects_mismatched_protocol_config() {
        let response = BlockSubscriptionResponse {
            block: Some(valid_genesis_block().into()),
            committed_chain_tip: 0,
            protocol_config: Some(other_protocol_config().into()),
        };
        let error = response.decode_fields().and_then(BuildUnchecked::build_unchecked).unwrap_err();
        assert!(error.to_string().contains("does not match header commitment"));
    }

    /// Startup must explicitly request the active configuration for the persisted tip.
    #[test]
    fn startup_header_request_includes_protocol_config() {
        let request = startup_header_request(BlockNumber::from(42));

        assert_eq!(request.block_num, Some(42));
        assert_eq!(request.include_protocol_config, Some(true));
    }

    /// Startup must reject a response without the requested configuration.
    #[test]
    fn startup_decoder_rejects_missing_protocol_config() {
        let config = test_protocol_config();
        let header = header_for_config(42, &config);
        let response = GetBlockHeaderByNumberResponse {
            block_header: Some((&header).into()),
            chain_length: None,
            mmr_path: None,
            protocol_config: None,
        };

        assert!(decode_startup_header_response(response, &header).is_err());
    }

    /// Startup must reject a configuration that does not match the persisted header commitment.
    #[test]
    fn startup_decoder_rejects_protocol_config_mismatch() {
        let config = test_protocol_config();
        let other = other_protocol_config();
        let header = header_for_config(42, &config);
        let response = GetBlockHeaderByNumberResponse {
            block_header: Some((&header).into()),
            chain_length: None,
            mmr_path: None,
            protocol_config: Some((&other).into()),
        };

        assert!(decode_startup_header_response(response, &header).is_err());
    }

    /// Startup must reject a valid remote bundle when its header is not the persisted local tip.
    #[test]
    fn startup_decoder_rejects_remote_header_mismatch() {
        let config = test_protocol_config();
        let local = header_for_config(42, &config);
        let remote = header_for_config(43, &config);
        let response = GetBlockHeaderByNumberResponse {
            block_header: Some((&remote).into()),
            chain_length: None,
            mmr_path: None,
            protocol_config: Some((&config).into()),
        };

        assert!(decode_startup_header_response(response, &local).is_err());
    }

    /// A stream accepts omission only after a baseline and only while the commitment is unchanged.
    #[test]
    fn subscription_tracker_validates_baseline_and_transitions() {
        let first = test_protocol_config();
        let second = other_protocol_config();
        let first_header = header_for_config(1, &first);
        let second_header = header_for_config(2, &second);
        let mut tracker = ProtocolConfigTracker::default();

        assert!(tracker.validate(&first_header, None).is_err());
        tracker.validate(&first_header, Some(&first)).unwrap();
        tracker.validate(&first_header, None).unwrap();
        assert!(tracker.validate(&second_header, None).is_err());
        tracker.validate(&first_header, None).unwrap();
        tracker.validate(&second_header, Some(&second)).unwrap();
        tracker.validate(&second_header, None).unwrap();
    }

    /// Every reconnect starts a new stream and therefore requires a new baseline configuration.
    #[test]
    fn subscription_tracker_requires_baseline_after_reconnect() {
        let config = test_protocol_config();
        let header = header_for_config(1, &config);
        let mut connected = ProtocolConfigTracker::default();
        connected.validate(&header, Some(&config)).unwrap();
        connected.validate(&header, None).unwrap();

        let mut reconnected = ProtocolConfigTracker::default();
        assert!(reconnected.validate(&header, None).is_err());
    }
}