//! Core proposer implementation for submitting block proposals.

use alethia_reth_consensus::eip4396::SHASTA_INITIAL_BASE_FEE;
use alethia_reth_primitives::{
    decode_shasta_proposal_id, payload::attributes::TaikoPayloadAttributes,
};
use alloy::{
    eips::BlockNumberOrTag,
    primitives::{Address, B256, Bytes, U256, aliases::U48},
    providers::Provider,
    rpc::types::{Block, Transaction},
    signers::local::PrivateKeySigner,
    transports::RpcError,
};
use alloy_consensus::{
    TxEnvelope,
    transaction::{Recovered, SignerRecoverable, TransactionInfo},
};
use alloy_eips::eip2718::{Decodable2718, Encodable2718};
use alloy_provider::RootProvider;
use alloy_rpc_types::{Transaction as RpcTransaction, TransactionReceipt};
use alloy_rpc_types_engine::{ExecutionPayloadFieldV2, ForkchoiceState};
use base_tx_manager::{SimpleTxManager, TxManager, TxManagerError};
use bindings::preconf_whitelist::PreconfWhitelist::PreconfWhitelistInstance;
use protocol::shasta::{
    AnchorTxConstructor, AnchorV4Input, PayloadAttributesInput, build_payload_attributes,
    calculate_shasta_mix_hash,
    constants::{
        PROPOSAL_MAX_BLOB_BYTES, calculate_next_block_eip4396_base_fee_for_parent,
        min_base_fee_for_chain,
    },
    encode_extra_data,
};
use rpc::{RpcClientError, client::Client};
use serde_json::from_value;
use tokio::time::{MissedTickBehavior, interval};
use tracing::{error, info, instrument, warn};

use crate::{
    config::ProposerConfigs,
    error::{ProposerError, Result},
    metrics::ProposerMetrics,
    transaction_builder::{ShastaProposalTransactionBuilder, manifest_gas_limit},
    tx_manager_adapter::{build_tx_manager, proposal_candidate},
};

/// Type alias for batches of transaction lists fetched from the txpool.
pub type TransactionLists = Vec<Vec<Transaction>>;

/// Chain-state snapshot a proposal is built against.
/// Captured once per proposal attempt — from engine-mode payload building or from the
/// proposer's own L1/L2 reads in pool mode — so transaction selection, the anchor transaction,
/// and the block manifest all describe the same parent and L1 head.
#[derive(Debug, Clone, Copy)]
pub struct EngineBuildContext {
    /// The L1 block number used for the anchor transaction.
    pub anchor_block_number: u64,
    /// The L2 parent block number used to derive the proposal payload.
    pub parent_block_number: u64,
    /// The timestamp used for the payload.
    pub timestamp: u64,
    /// The gas limit for the block.
    pub gas_limit: u64,
}

impl EngineBuildContext {
    /// Capture a build context, together with the L2 parent block it derives from, off the
    /// current L1/L2 chain heads.
    ///
    /// The timestamp is taken from the L1 head rather than the local wall clock: driver
    /// validation rejects any manifest block stamped above the proposal's L1 inclusion
    /// timestamp, degrading the whole manifest to the default empty block.
    pub async fn from_chain_heads(rpc: &Client) -> Result<(Self, Block)> {
        let parent = rpc
            .l2_provider
            .get_block_by_number(BlockNumberOrTag::Latest)
            .await?
            .ok_or(ProposerError::LatestBlockNotFound)?;

        let l1_head = rpc
            .l1_provider
            .get_block_by_number(BlockNumberOrTag::Latest)
            .await?
            .ok_or(ProposerError::LatestBlockNotFound)?;

        let context = Self {
            anchor_block_number: l1_head.header.number,
            parent_block_number: parent.number(),
            timestamp: l1_head.header.timestamp,
            gas_limit: parent.header.gas_limit,
        };

        Ok((context, parent))
    }
}

/// Proposer loop that builds and submits Shasta proposals at a fixed interval.
pub struct Proposer {
    /// RPC client bundle used for L1/L2 reads; L1 submission is signed by the tx-manager.
    rpc_provider: Client,
    /// Builder that converts txpool content into proposal transactions.
    transaction_builder: ShastaProposalTransactionBuilder,
    /// Tx-manager responsible for proposal submission and retry handling.
    tx_manager: SimpleTxManager,
    /// L1 address derived from the configured proposer private key.
    l1_proposer_address: Address,
    /// Optional anchor constructor used in engine mode.
    anchor_constructor: Option<AnchorTxConstructor<RootProvider<alloy_network::Ethereum>>>,
    /// Chain-specific minimum base fee used by EIP-4396 clamping.
    min_base_fee_to_clamp: u64,
    /// Runtime proposer configuration.
    cfg: ProposerConfigs,
}

impl Proposer {
    /// Creates a new proposer instance.
    #[instrument(skip(cfg), fields(inbox_address = ?cfg.client.inbox_address))]
    pub async fn new(cfg: ProposerConfigs) -> Result<Self> {
        info!(
            inbox_address = ?cfg.client.inbox_address,
            l2_suggested_fee_recipient = ?cfg.l2_suggested_fee_recipient,
            propose_interval = ?cfg.propose_interval,
            "initializing proposer"
        );

        let rpc_provider = Client::new(cfg.client.clone()).await?;

        let transaction_builder = ShastaProposalTransactionBuilder::new(
            rpc_provider.clone(),
            cfg.l2_suggested_fee_recipient,
        );
        // The RPC client carries no wallet; the tx-manager owns the proposer key, so all
        // L1 proposal submissions flow through tx-manager and nonce management stays on a
        // single send path.
        let tx_manager = build_tx_manager(&cfg, rpc_provider.l1_provider.root().to_owned()).await?;
        let l1_proposer_address = proposer_address_from_key(&cfg.l1_proposer_private_key)?;
        // Match proposer-side base-fee clamping to chain policy used by derivation.
        let min_base_fee_to_clamp =
            min_base_fee_for_chain(rpc_provider.l2_provider.get_chain_id().await?);

        // Initialize anchor transaction constructor only for engine mode.
        let anchor_constructor = if cfg.use_engine_mode {
            warn!(
                "engine mode is enabled: on execution clients without deferred L1-origin \
                 persistence (taikoxyz/alethia-reth#219), the build-only FCU preview persists \
                 l1_origin, head_l1_origin, and batch_to_last_block rows for blocks that never \
                 become canonical, leaving ghost rows every proposal cycle; verify the \
                 connected node includes that fix before using engine mode in production"
            );
            Some(
                AnchorTxConstructor::new(
                    rpc_provider.l2_provider.clone(),
                    *rpc_provider.shasta.anchor.address(),
                )
                .await?,
            )
        } else {
            None
        };

        Ok(Self {
            rpc_provider,
            transaction_builder,
            tx_manager,
            l1_proposer_address,
            anchor_constructor,
            min_base_fee_to_clamp,
            cfg,
        })
    }

    /// Start the proposer main loop.
    pub async fn start(&self) -> Result<()> {
        let mut interval = interval(self.cfg.propose_interval);
        // A slow confirmation must not replay the missed ticks as an immediate burst afterwards:
        // the inbox accepts at most one proposal per L1 block, so burst ticks only revert.
        interval.set_missed_tick_behavior(MissedTickBehavior::Delay);
        let mut epoch = 0;

        loop {
            interval.tick().await;
            info!(epoch, "proposer epoch");

            match self.precheck_current_preconf_operator().await {
                Ok(true) => {}
                Ok(false) => {
                    info!(
                        epoch,
                        proposer = ?self.l1_proposer_address,
                        "skipping proposal attempt because proposer is not current preconf whitelist operator"
                    );
                    epoch += 1;
                    continue;
                }
                Err(err) if is_operational_loop_error(&err) => {
                    if should_increment_loop_failure_metric(&err) {
                        ProposerMetrics::proposals_failed().inc();
                    }
                    warn!(
                        epoch,
                        error = %err,
                        "proposer precheck failed on a retryable error; continuing proposer loop"
                    );
                    epoch += 1;
                    continue;
                }
                Err(err) => return Err(err),
            }

            match self.fetch_and_propose().await {
                Ok(receipt) => {
                    info!(
                        epoch,
                        tx_hash = %receipt.transaction_hash,
                        execution_succeeded = receipt.status(),
                        "proposal attempt completed"
                    );
                }
                Err(err) if is_operational_loop_error(&err) => {
                    if should_increment_loop_failure_metric(&err) {
                        ProposerMetrics::proposals_failed().inc();
                    }
                    warn!(epoch, error = %err, "proposal attempt failed; continuing proposer loop");
                }
                Err(err) => return Err(err),
            }

            epoch += 1;
        }
    }

    /// Fetch L2 EE mempool and propose a new proposal to protocol inbox.
    pub async fn fetch_and_propose(&self) -> Result<TransactionReceipt> {
        // Fetch transactions based on mode. Both modes capture the chain-state snapshot the
        // transactions were selected against, so the manifest is built from the same snapshot.
        let (pool_content, build_ctx) = if self.cfg.use_engine_mode {
            self.fetch_payload_transactions().await?
        } else {
            self.fetch_pool_content().await?
        };

        // Record number of transactions in the pool
        let tx_count: usize = pool_content.iter().map(|list| list.len()).sum();
        ProposerMetrics::tx_pool_size().set(tx_count as f64);
        info!(
            txs_lists = pool_content.len(),
            tx_count,
            engine_mode = self.cfg.use_engine_mode,
            ?build_ctx,
            "fetched transaction pool content"
        );

        let mut proposal_tx = self.transaction_builder.build(pool_content, build_ctx).await?;

        // Set gas limit if configured, otherwise let the provider estimate it.
        if let Some(gas_limit) = self.cfg.gas_limit {
            proposal_tx = proposal_tx.with_gas_limit(gas_limit);
        }

        record_submission_attempt();
        let receipt = self.tx_manager.send(proposal_candidate(proposal_tx)).await?;
        record_submission_receipt(receipt)
    }

    /// Return a clone of the RPC client bundle used by the proposer.
    pub fn rpc_client(&self) -> Client {
        self.rpc_provider.clone()
    }

    /// Return whether the configured proposer key is the current preconfirmation whitelist
    /// operator.
    async fn precheck_current_preconf_operator(&self) -> Result<bool> {
        let inbox_config = self.rpc_provider.shasta.inbox.getConfig().call().await?;
        if self.forced_inclusion_allows_permissionless(&inbox_config).await? {
            info!(
                "allowing proposal attempt because forced inclusion processing is permissionless"
            );
            return Ok(true);
        }

        let whitelist = PreconfWhitelistInstance::new(
            inbox_config.proposerChecker,
            self.rpc_provider.l1_provider.clone(),
        );
        let current_operator = whitelist.getOperatorForCurrentEpoch().call().await?;

        Ok(current_operator == self.l1_proposer_address)
    }

    /// Return whether the oldest queued forced inclusion makes proposing permissionless.
    async fn forced_inclusion_allows_permissionless(
        &self,
        inbox_config: &bindings::inbox::IInbox::Config,
    ) -> Result<bool> {
        let forced_inclusion_state =
            self.rpc_provider.shasta.inbox.getForcedInclusionState().call().await?;
        if forced_inclusion_state.head_ == forced_inclusion_state.tail_ {
            return Ok(false);
        }

        let inclusions = self
            .rpc_provider
            .shasta
            .inbox
            .getForcedInclusions(forced_inclusion_state.head_, U48::from(1))
            .call()
            .await?;
        let Some(oldest_inclusion) = inclusions.first() else {
            return Ok(false);
        };

        let latest_l1_block = self
            .rpc_provider
            .l1_provider
            .get_block_by_number(BlockNumberOrTag::Latest)
            .await?
            .ok_or(ProposerError::LatestBlockNotFound)?;

        Ok(forced_inclusion_is_permissionless(
            oldest_inclusion.blobSlice.timestamp.to::<u64>(),
            latest_l1_block.header.timestamp,
            inbox_config.forcedInclusionDelay,
            inbox_config.permissionlessInclusionMultiplier,
        ))
    }

    /// Fetch transaction pool content from the L2 execution engine, together with the
    /// chain-state snapshot the selection ran against.
    async fn fetch_pool_content(&self) -> Result<(TransactionLists, EngineBuildContext)> {
        let (build_ctx, parent) = EngineBuildContext::from_chain_heads(&self.rpc_provider).await?;

        let base_fee_u64 =
            u64::try_from(self.calculate_next_shasta_block_base_fee_for_parent(&parent).await?)
                .map_err(|_| ProposerError::BaseFeeOverflow)?;

        let pool_content = self
            .rpc_provider
            .tx_pool_content_with_min_tip(rpc::TxPoolContentParams {
                beneficiary: self.cfg.l2_suggested_fee_recipient,
                base_fee: Some(base_fee_u64),
                // Fill up to the gas limit the manifest will actually declare (parent limit
                // minus the anchor-gas discount) instead of the protocol minimum, which
                // under-filled every list to 10M while blocks advertise ~45M. Derived from the
                // same snapshot the manifest is built against.
                block_max_gas_limit: manifest_gas_limit(
                    build_ctx.parent_block_number,
                    build_ctx.gas_limit,
                ),
                max_bytes_per_tx_list: PROPOSAL_MAX_BLOB_BYTES as u64,
                locals: vec![],
                max_transactions_lists: 1,
                min_tip: 0,
            })
            .await?;

        info!(
            txs_lists_count = pool_content.len(),
            "fetched transactions lists from L2 execution engine"
        );

        let txs_lists = pool_content
            .into_iter()
            .map(|content| {
                content
                    .tx_list
                    .into_iter()
                    .map(|tx| from_value::<Transaction>(tx).map_err(ProposerError::from))
                    .collect::<Result<Vec<_>>>()
            })
            .collect::<Result<Vec<Vec<_>>>>()?;

        Ok((txs_lists, build_ctx))
    }

    /// Calculate the base fee for the next L2 block from a specific parent snapshot.
    async fn calculate_next_shasta_block_base_fee_for_parent(
        &self,
        parent: &Block,
    ) -> Result<U256> {
        let parent_number = parent.number();
        let grandparent = if parent_number == 0 {
            None
        } else {
            let grandparent_number = parent_number.saturating_sub(1);
            Some(
                self.rpc_provider
                    .l2_provider
                    .get_block_by_hash(parent.header.parent_hash)
                    .await?
                    .ok_or(ProposerError::ParentBlockNotFound(grandparent_number))?,
            )
        };

        calculate_next_shasta_block_base_fee_from_parent(
            parent,
            grandparent.as_ref(),
            self.min_base_fee_to_clamp,
        )
    }

    /// Build forkchoice state from L2 chain.
    /// Returns the forkchoice state and the head block used.
    async fn build_forkchoice_state(&self) -> Result<(ForkchoiceState, Block)> {
        let head = self
            .rpc_provider
            .l2_provider
            .get_block_by_number(BlockNumberOrTag::Latest)
            .await?
            .ok_or(ProposerError::LatestBlockNotFound)?;

        let safe = self
            .rpc_provider
            .l2_provider
            .get_block_by_number(BlockNumberOrTag::Safe)
            .await?
            .map(|b| b.header.hash)
            .unwrap_or(head.header.hash);

        let finalized = self
            .rpc_provider
            .l2_provider
            .get_block_by_number(BlockNumberOrTag::Finalized)
            .await?
            .map(|b| b.header.hash)
            .unwrap_or(head.header.hash);

        Ok((
            ForkchoiceState {
                head_block_hash: head.header.hash,
                safe_block_hash: safe,
                finalized_block_hash: finalized,
            },
            head,
        ))
    }

    /// Build Taiko payload attributes for engine mode.
    /// Constructs the payload attributes with anchor transaction and block metadata.
    /// Returns the payload attributes and the engine payload parameters used.
    async fn build_payload_attributes(
        &self,
        parent: &Block,
    ) -> Result<(TaikoPayloadAttributes, EngineBuildContext)> {
        let block_number = parent.number() + 1;

        // Get basefee sharing percentage from inbox config.
        let inbox_config = self.rpc_provider.shasta.inbox.getConfig().call().await?;
        let basefee_sharing_pctg = inbox_config.basefeeSharingPctg;

        // Get proposal ID from parent's extra data and increment.
        let proposal_id = next_shasta_proposal_id(parent.header.number, &parent.header.extra_data)?;

        // Calculate base fee for the new block.
        let base_fee = self.calculate_next_shasta_block_base_fee_for_parent(parent).await?;

        // Get latest L1 block for anchor transaction.
        let l1_block = self
            .rpc_provider
            .l1_provider
            .get_block_by_number(BlockNumberOrTag::Latest)
            .await?
            .ok_or(ProposerError::LatestBlockNotFound)?;
        let anchor_block_number = l1_block.header.number;
        // Stamp the payload with the L1 head timestamp instead of the local wall clock: the
        // manifest timestamp must not exceed the proposal's L1 inclusion timestamp, or driver
        // validation degrades the whole manifest to the default empty block.
        let timestamp = l1_block.header.timestamp;

        // Build anchor transaction.
        let anchor_tx = self
            .anchor_constructor
            .as_ref()
            .ok_or(ProposerError::AnchorConstructorNotInitialized)?
            .assemble_anchor_v4_tx(
                parent.header.hash,
                AnchorV4Input {
                    anchor_block_number,
                    anchor_block_hash: l1_block.header.hash,
                    anchor_state_root: l1_block.header.inner.state_root,
                    l2_height: block_number,
                    base_fee,
                },
            )
            .await?;

        // Calculate mix hash.
        let mix_hash = calculate_shasta_mix_hash(parent.header.inner.mix_hash, block_number);

        let payload_attributes = build_payload_attributes(PayloadAttributesInput {
            beneficiary: self.cfg.l2_suggested_fee_recipient,
            timestamp,
            mix_hash,
            gas_limit: parent.header.gas_limit,
            // Engine mode: let the node select transactions from its mempool.
            tx_list: None,
            extra_data: encode_extra_data(basefee_sharing_pctg, proposal_id),
            base_fee_per_gas: base_fee,
            block_number,
            l1_block_height: Some(U256::from(anchor_block_number)),
            l1_block_hash: Some(l1_block.header.hash),
            is_forced_inclusion: false,
            signature: [0; 65],
            parent_beacon_block_root: None,
            anchor_transaction: Some(Bytes::from(anchor_tx.encoded_2718())),
        });

        Ok((
            payload_attributes,
            EngineBuildContext {
                anchor_block_number,
                parent_block_number: parent.header.number,
                timestamp,
                gas_limit: parent.header.gas_limit,
            },
        ))
    }

    /// Fetch transactions using Engine API (FCU + get_payload).
    /// In engine mode, we use forkchoice_updated to trigger payload building
    /// with tx_list: None, then retrieve the built payload to extract transactions.
    /// Returns the transactions and the engine payload parameters used.
    async fn fetch_payload_transactions(&self) -> Result<(TransactionLists, EngineBuildContext)> {
        // Build forkchoice state and get the head block to use as parent.
        let (forkchoice_state, parent) = self.build_forkchoice_state().await?;

        // Build payload attributes and capture the engine parameters used.
        let (payload_attributes, engine_params) = self.build_payload_attributes(&parent).await?;

        info!(
            parent_number = parent.number(),
            parent_hash = %parent.header.hash,
            anchor_block_number = engine_params.anchor_block_number,
            timestamp = engine_params.timestamp,
            gas_limit = engine_params.gas_limit,
            "sending forkchoice_updated with payload attributes"
        );

        // Send forkchoice_updated to trigger payload building.
        let fcu_response = self
            .rpc_provider
            .engine_forkchoice_updated_v2(forkchoice_state, Some(payload_attributes))
            .await
            .map_err(|e| ProposerError::FcuFailed(e.to_string()))?;

        // Check FCU response status.
        if !fcu_response.payload_status.is_valid() {
            return Err(ProposerError::FcuFailed(format!(
                "invalid payload status: {:?}",
                fcu_response.payload_status
            )));
        }

        // Get payload ID from FCU response.
        let payload_id = fcu_response.payload_id.ok_or(ProposerError::NoPayloadId)?;

        info!(payload_id = ?payload_id, "received payload ID, fetching payload");

        // Fetch the built payload.
        let payload_envelope = self.rpc_provider.engine_get_payload_v2(payload_id).await?;

        // Extract transactions from payload based on version.
        let transactions = match &payload_envelope.execution_payload {
            ExecutionPayloadFieldV2::V1(payload) => &payload.transactions,
            ExecutionPayloadFieldV2::V2(payload) => &payload.payload_inner.transactions,
        };

        // If no transactions, return empty list with engine parameters.
        if transactions.is_empty() {
            info!("payload contains no transactions");
            return Ok((vec![vec![]], engine_params));
        }

        // Skip the first transaction (anchor) and parse the rest.
        let txs: Vec<Transaction> = transactions
            .iter()
            .skip(1) // Skip anchor transaction
            .enumerate()
            .map(|(index, tx_bytes): (usize, &Bytes)| {
                // Decode the transaction from RLP bytes.
                let tx = TxEnvelope::decode_2718(&mut tx_bytes.as_ref())
                    .map_err(|source| ProposerError::TxDecode { index, source })?;

                // Recover the signer address from the transaction signature.
                let signer = tx
                    .recover_signer()
                    .map_err(|e| ProposerError::SignerRecovery { index, message: e.to_string() })?;

                Ok(RpcTransaction::from_transaction(
                    Recovered::new_unchecked(tx, signer),
                    TransactionInfo::default(),
                ))
            })
            .collect::<Result<Vec<_>>>()?;

        info!(
            tx_count = txs.len(),
            total_payload_txs = transactions.len(),
            anchor_block_number = engine_params.anchor_block_number,
            "extracted user transactions from engine payload"
        );

        Ok((vec![txs], engine_params))
    }
}

/// Calculate the next Shasta base fee from a fixed parent snapshot and its grandparent.
fn calculate_next_shasta_block_base_fee_from_parent(
    parent: &Block,
    grandparent: Option<&Block>,
    min_base_fee_to_clamp: u64,
) -> Result<U256> {
    if parent.number() == 0 {
        return Ok(U256::from(SHASTA_INITIAL_BASE_FEE));
    }

    let grandparent =
        grandparent.ok_or(ProposerError::ParentBlockNotFound(parent.number().saturating_sub(1)))?;

    calculate_next_block_eip4396_base_fee_for_parent(
        parent.header.inner.number,
        parent.header.inner.gas_limit,
        parent.header.inner.gas_used,
        parent.header.timestamp,
        parent.header.inner.base_fee_per_gas,
        grandparent.header.timestamp,
        min_base_fee_to_clamp,
    )
    .map(U256::from)
    .ok_or(ProposerError::MissingParentBaseFee { parent_block_number: parent.number() })
}

/// Record metrics and logs for a proposer submission receipt.
fn record_submission_receipt(receipt: TransactionReceipt) -> Result<TransactionReceipt> {
    if receipt.status() {
        info!(
            tx_hash = %receipt.transaction_hash,
            gas_used = receipt.gas_used,
            "proposal transaction mined successfully"
        );
        ProposerMetrics::proposals_success().inc();

        // Record gas used once the confirmed receipt shows successful execution.
        ProposerMetrics::gas_used().observe(receipt.gas_used as f64);
        Ok(receipt)
    } else {
        let tx_hash = receipt.transaction_hash;
        error!(tx_hash = %tx_hash, "proposal transaction failed");
        ProposerMetrics::proposals_failed().inc();
        Err(ProposerError::ProposalTransactionReverted { tx_hash })
    }
}

/// Record that the proposer started an L1 submission attempt for a built proposal.
fn record_submission_attempt() {
    ProposerMetrics::proposals_sent().inc();
}

/// Return the L1 account address controlled by a proposer private key.
fn proposer_address_from_key(private_key: &B256) -> Result<Address> {
    PrivateKeySigner::from_bytes(private_key).map(|signer| signer.address()).map_err(|err| {
        ProposerError::from(TxManagerError::Sign(format!(
            "failed to build proposer signer from configured private key: {err}"
        )))
    })
}

/// Return whether a forced inclusion is old enough to bypass proposer authorization.
#[must_use]
fn forced_inclusion_is_permissionless(
    oldest_timestamp: u64,
    l1_timestamp: u64,
    forced_inclusion_delay: u16,
    permissionless_inclusion_multiplier: u8,
) -> bool {
    if oldest_timestamp == 0 {
        return false;
    }

    let permissionless_timestamp = u64::from(forced_inclusion_delay)
        .saturating_mul(u64::from(permissionless_inclusion_multiplier))
        .saturating_add(oldest_timestamp);
    l1_timestamp > permissionless_timestamp
}

/// Derive the next proposal id from the parent block header.
///
/// Shasta stores the previous proposal id in the parent block extra data. On a fresh chain the
/// genesis parent may still have empty extra data, so the first proposal starts at id `1`.
fn next_shasta_proposal_id(parent_block_number: u64, parent_extra_data: &Bytes) -> Result<u64> {
    if parent_block_number == 0 {
        return Ok(1);
    }

    decode_shasta_proposal_id(parent_extra_data)
        .map(|proposal_id| proposal_id + 1)
        .ok_or(ProposerError::InvalidExtraData)
}

/// Return `true` when a surfaced proposer loop error should be retried on the next epoch.
///
/// Transport failures (network blips, timeouts, backend-gone), tx-manager execution reverts,
/// and proposer-owned reverted receipt errors are operational.
/// RPC error responses (`ErrorResp`) and local errors (decoding, unsupported features, unknown
/// functions, fatal tx-manager errors) are fatal and exit the loop.
fn is_operational_loop_error(err: &ProposerError) -> bool {
    matches!(
        err,
        ProposerError::Rpc(RpcError::Transport(_)) |
            ProposerError::RpcClient(RpcClientError::Rpc(RpcError::Transport(_))) |
            ProposerError::Contract(alloy::contract::Error::TransportError(RpcError::Transport(
                _,
            ))) |
            ProposerError::TxManager(
                TxManagerError::Rpc(_) |
                    TxManagerError::SendTimeout |
                    TxManagerError::MempoolDeadlineExpired |
                    TxManagerError::ExecutionReverted,
            ) |
            ProposerError::ProposalTransactionReverted { .. }
    )
}

/// Return whether a retryable proposer loop error should increment the loop failure counter.
///
/// Receipt-level proposal reverts are already counted when the receipt is recorded, so counting
/// here would double-count the same failure.
#[must_use]
fn should_increment_loop_failure_metric(err: &ProposerError) -> bool {
    !matches!(err, ProposerError::ProposalTransactionReverted { .. })
}

#[cfg(test)]
mod tests {
    use alethia_reth_consensus::eip4396::SHASTA_INITIAL_BASE_FEE;
    use alloy::{
        consensus::Header as ConsensusHeader,
        primitives::{Address, B256, Bytes, U256},
        transports::{RpcError, TransportErrorKind},
    };
    use alloy_consensus::{Eip658Value, Receipt, ReceiptEnvelope, ReceiptWithBloom};
    use alloy_json_rpc::ErrorPayload;
    use alloy_rpc_types::{
        TransactionReceipt,
        eth::{Block as RpcBlock, Header as RpcHeader},
    };
    use base_tx_manager::TxManagerError;

    use super::{
        calculate_next_shasta_block_base_fee_from_parent, forced_inclusion_is_permissionless,
        is_operational_loop_error, next_shasta_proposal_id, record_submission_attempt,
        record_submission_receipt, should_increment_loop_failure_metric,
    };
    use crate::{error::ProposerError, metrics::ProposerMetrics};
    use protocol::shasta::{
        constants::calculate_next_block_eip4396_base_fee_from_parent_values, encode_extra_data,
    };
    use rpc::RpcClientError;

    #[test]
    fn submission_attempt_increments_sent_metric() {
        let before = ProposerMetrics::proposals_sent().get();
        record_submission_attempt();
        assert_eq!(ProposerMetrics::proposals_sent().get(), before + 1);
    }

    #[test]
    fn successful_submission_receipt_is_returned_and_recorded() {
        let receipt = receipt_with_status(true);
        let tx_hash = receipt.transaction_hash;
        let before = ProposerMetrics::proposals_success().get();

        let recorded = record_submission_receipt(receipt)
            .expect("successful receipt should remain successful");

        assert_eq!(recorded.transaction_hash, tx_hash);
        assert_eq!(ProposerMetrics::proposals_success().get(), before + 1);
    }

    /// Serializes tests that snapshot and assert on the process-global proposal-failure
    /// counter, which would otherwise race each other under the parallel test runner.
    static PROPOSALS_FAILED_METRIC_GUARD: std::sync::Mutex<()> = std::sync::Mutex::new(());

    #[test]
    fn reverted_submission_receipt_becomes_failed_proposal_error() {
        let _metric_guard =
            PROPOSALS_FAILED_METRIC_GUARD.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
        let receipt = receipt_with_status(false);
        let tx_hash = receipt.transaction_hash;
        let before = ProposerMetrics::proposals_failed().get();

        let err = record_submission_receipt(receipt)
            .expect_err("reverted receipt should be surfaced as a proposer error");

        assert!(matches!(
            err,
            ProposerError::ProposalTransactionReverted { tx_hash: observed } if observed == tx_hash
        ));
        assert_eq!(ProposerMetrics::proposals_failed().get(), before + 1);
    }

    fn receipt_with_status(status: bool) -> TransactionReceipt {
        TransactionReceipt {
            inner: ReceiptEnvelope::Eip1559(ReceiptWithBloom {
                receipt: Receipt {
                    status: Eip658Value::Eip658(status),
                    cumulative_gas_used: 21_000,
                    logs: vec![],
                },
                logs_bloom: Default::default(),
            }),
            transaction_hash: B256::repeat_byte(if status { 0x11 } else { 0x22 }),
            transaction_index: Some(0),
            block_hash: Some(B256::repeat_byte(0x33)),
            block_number: Some(1),
            gas_used: 21_000,
            effective_gas_price: 1,
            blob_gas_used: None,
            blob_gas_price: None,
            from: Address::repeat_byte(0x44),
            to: Some(Address::repeat_byte(0x55)),
            contract_address: None,
        }
    }

    #[test]
    fn forced_inclusion_precheck_classifies_permissionless_windows() {
        // (case, (oldest inclusion timestamp, l1 timestamp, delay, multiplier), expected).
        let cases = [
            ("l1 timestamp past the permissionless window", (100, 151, 10, 5), true),
            ("l1 timestamp at the window boundary", (100, 150, 10, 5), false),
            ("l1 timestamp before the window", (100, 149, 10, 5), false),
            ("missing oldest-inclusion timestamp", (0, 1_000, 10, 5), false),
        ];

        for (case, (oldest_timestamp, l1_timestamp, delay, multiplier), expected) in cases {
            assert_eq!(
                forced_inclusion_is_permissionless(
                    oldest_timestamp,
                    l1_timestamp,
                    delay,
                    multiplier
                ),
                expected,
                "{case}"
            );
        }
    }

    #[test]
    fn loop_failure_metric_not_double_counted_for_reverted_receipts() {
        let _metric_guard =
            PROPOSALS_FAILED_METRIC_GUARD.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
        let before = ProposerMetrics::proposals_failed().get();

        let reverted_err =
            ProposerError::ProposalTransactionReverted { tx_hash: B256::repeat_byte(0x22) };
        if is_operational_loop_error(&reverted_err) &&
            should_increment_loop_failure_metric(&reverted_err)
        {
            ProposerMetrics::proposals_failed().inc();
        }
        assert_eq!(ProposerMetrics::proposals_failed().get(), before);

        let execution_reverted = ProposerError::TxManager(TxManagerError::ExecutionReverted);
        let before = ProposerMetrics::proposals_failed().get();
        if is_operational_loop_error(&execution_reverted) &&
            should_increment_loop_failure_metric(&execution_reverted)
        {
            ProposerMetrics::proposals_failed().inc();
        }
        assert_eq!(ProposerMetrics::proposals_failed().get(), before + 1);
    }

    #[test]
    fn loop_error_classification_matches_expected_retry_behavior() {
        let error_payload = || -> ErrorPayload {
            serde_json::from_str(
                r#"{"code":3,"message":"execution reverted: ","data":"0x810f00230000000000000000000000000000000000000000000000000000000000000001"}"#,
            )
            .expect("valid JSON-RPC error payload")
        };
        let revert_contract_error =
            alloy::contract::Error::TransportError(RpcError::ErrorResp(error_payload()));
        assert!(revert_contract_error.as_revert_data().is_some());

        // (case, error, expected `is_operational_loop_error` verdict).
        let cases = [
            (
                "tx-manager rpc failure is retried",
                ProposerError::TxManager(TxManagerError::Rpc("provider timed out".into())),
                true,
            ),
            (
                "tx-manager send timeout is retried",
                ProposerError::TxManager(TxManagerError::SendTimeout),
                true,
            ),
            (
                "tx-manager mempool deadline expiry is retried",
                ProposerError::TxManager(TxManagerError::MempoolDeadlineExpired),
                true,
            ),
            (
                "tx-manager execution revert is retried",
                ProposerError::TxManager(TxManagerError::ExecutionReverted),
                true,
            ),
            (
                "reverted proposal receipt is retried",
                ProposerError::ProposalTransactionReverted { tx_hash: B256::repeat_byte(0x22) },
                true,
            ),
            (
                "precheck transport failure is retried",
                ProposerError::Rpc(TransportErrorKind::backend_gone()),
                true,
            ),
            (
                "precheck contract transport failure is retried",
                ProposerError::Contract(alloy::contract::Error::TransportError(
                    TransportErrorKind::backend_gone(),
                )),
                true,
            ),
            (
                "rpc-client transport failure is retried",
                ProposerError::from(RpcClientError::from(TransportErrorKind::backend_gone())),
                true,
            ),
            (
                "rpc error response exits the loop",
                ProposerError::Rpc(RpcError::ErrorResp(error_payload())),
                false,
            ),
            (
                "contract revert response exits the loop",
                ProposerError::Contract(revert_contract_error),
                false,
            ),
            (
                "rpc-client error response exits the loop",
                ProposerError::from(RpcClientError::from(RpcError::ErrorResp(error_payload()))),
                false,
            ),
            (
                "local contract error exits the loop",
                ProposerError::Contract(alloy::contract::Error::UnknownFunction(
                    "getOperatorForCurrentEpoch".into(),
                )),
                false,
            ),
            (
                "tx-manager nonce-too-low exits the loop",
                ProposerError::TxManager(TxManagerError::NonceTooLow),
                false,
            ),
            (
                "tx-manager fee-limit breach exits the loop",
                ProposerError::TxManager(TxManagerError::FeeLimitExceeded { fee: 11, ceiling: 10 }),
                false,
            ),
            (
                "tx-manager signing failure exits the loop",
                ProposerError::TxManager(TxManagerError::Sign("wallet rejected signing".into())),
                false,
            ),
            (
                "tx-manager invalid config exits the loop",
                ProposerError::TxManager(TxManagerError::InvalidConfig("bad fee limit".into())),
                false,
            ),
        ];

        for (case, err, expected) in cases {
            assert_eq!(is_operational_loop_error(&err), expected, "{case}");
        }
    }

    #[test]
    fn next_shasta_proposal_id_starts_from_one_for_empty_genesis_extra_data() {
        assert_eq!(
            next_shasta_proposal_id(0, &Bytes::new()).expect("empty genesis extra data is valid"),
            1
        );
    }

    #[test]
    fn next_shasta_proposal_id_increments_encoded_parent_proposal_id() {
        assert_eq!(
            next_shasta_proposal_id(7, &encode_extra_data(15, 9))
                .expect("encoded proposal id should decode"),
            10
        );
    }

    #[test]
    fn next_shasta_proposal_id_rejects_non_genesis_invalid_extra_data() {
        assert!(matches!(
            next_shasta_proposal_id(1, &Bytes::from_static(&[0x12, 0x34])),
            Err(crate::error::ProposerError::InvalidExtraData)
        ));
    }

    #[test]
    fn base_fee_calculation_uses_supplied_parent_snapshot() {
        let grandparent = RpcBlock {
            header: RpcHeader {
                hash: B256::repeat_byte(0x11),
                inner: ConsensusHeader { number: 1, timestamp: 100, ..Default::default() },
                total_difficulty: None,
                size: None,
            },
            ..Default::default()
        };
        let parent = RpcBlock {
            header: RpcHeader {
                hash: B256::repeat_byte(0x22),
                inner: ConsensusHeader {
                    number: 2,
                    parent_hash: grandparent.header.hash,
                    timestamp: 112,
                    gas_limit: 45_000_000,
                    base_fee_per_gas: Some(2_000_000_000),
                    ..Default::default()
                },
                total_difficulty: None,
                size: None,
            },
            ..Default::default()
        };

        let expected = U256::from(calculate_next_block_eip4396_base_fee_from_parent_values(
            parent.header.inner.number,
            parent.header.inner.gas_limit,
            parent.header.inner.gas_used,
            parent.header.timestamp.saturating_sub(grandparent.header.timestamp),
            parent.header.inner.base_fee_per_gas.expect("parent should define a base fee"),
            1_000_000_000,
        ));

        assert_eq!(
            calculate_next_shasta_block_base_fee_from_parent(
                &parent,
                Some(&grandparent),
                1_000_000_000
            )
            .expect("parent snapshot should determine the next base fee"),
            expected
        );
    }

    #[test]
    fn base_fee_calculation_returns_initial_base_fee_for_genesis_parent() {
        let genesis = RpcBlock {
            header: RpcHeader {
                hash: B256::repeat_byte(0x11),
                inner: ConsensusHeader { number: 0, ..Default::default() },
                total_difficulty: None,
                size: None,
            },
            ..Default::default()
        };

        assert_eq!(
            calculate_next_shasta_block_base_fee_from_parent(&genesis, None, 1_000_000_000)
                .expect("genesis parent should not require a grandparent"),
            U256::from(SHASTA_INITIAL_BASE_FEE)
        );
    }

    #[test]
    fn base_fee_calculation_errors_on_missing_grandparent() {
        let parent = RpcBlock {
            header: RpcHeader {
                hash: B256::repeat_byte(0x22),
                inner: ConsensusHeader {
                    number: 2,
                    timestamp: 112,
                    gas_limit: 45_000_000,
                    base_fee_per_gas: Some(2_000_000_000),
                    ..Default::default()
                },
                total_difficulty: None,
                size: None,
            },
            ..Default::default()
        };

        assert!(matches!(
            calculate_next_shasta_block_base_fee_from_parent(&parent, None, 1_000_000_000),
            Err(ProposerError::ParentBlockNotFound(1))
        ));
    }

    #[test]
    fn base_fee_calculation_errors_on_parent_missing_base_fee() {
        let grandparent = RpcBlock {
            header: RpcHeader {
                hash: B256::repeat_byte(0x11),
                inner: ConsensusHeader { number: 1, timestamp: 100, ..Default::default() },
                total_difficulty: None,
                size: None,
            },
            ..Default::default()
        };
        let parent = RpcBlock {
            header: RpcHeader {
                hash: B256::repeat_byte(0x22),
                inner: ConsensusHeader {
                    number: 2,
                    parent_hash: grandparent.header.hash,
                    timestamp: 112,
                    gas_limit: 45_000_000,
                    base_fee_per_gas: None,
                    ..Default::default()
                },
                total_difficulty: None,
                size: None,
            },
            ..Default::default()
        };

        assert!(matches!(
            calculate_next_shasta_block_base_fee_from_parent(
                &parent,
                Some(&grandparent),
                1_000_000_000
            ),
            Err(ProposerError::MissingParentBaseFee { parent_block_number: 2 })
        ));
    }
}
