//! ETH namespace RPC endpoint implementations.

use crate::{
    config::{EvmBlockContext, StorageRpcCtx, gas_oracle},
    eth::{
        error::EthError,
        helpers::{
            AddrWithBlock, BlockParams, CfgFiller, FeeHistoryArgs, StorageAtArgs, SubscribeArgs,
            TxParams, await_handler, build_receipt, build_rpc_transaction, hot_reader_at_block,
            normalize_gas_stateless,
        },
        types::{
            BlockTransactions, EmptyArray, LazyReceipts, RpcBlock, RpcHeader, RpcReceipt,
            RpcTransaction,
        },
    },
    interest::{FilterOutput, InterestKind},
};
use ajj::HandlerCtx;
use alloy::{
    consensus::Transaction,
    eips::{
        BlockId, BlockNumberOrTag,
        eip1559::BaseFeeParams,
        eip2718::{Decodable2718, Encodable2718},
        eip2930::AccessListResult,
    },
    primitives::{B256, U64, U256},
    rpc::types::{FeeHistory, Filter, Log},
};
use revm_inspectors::access_list::AccessListInspector;
use serde::Serialize;
use signet_cold::{HeaderSpecifier, ReceiptSpecifier};
use signet_hot::{HistoryRead, HotKv, db::HotDbRead, model::HotKvRead};
use tracing::{Instrument, debug, trace, trace_span};
use trevm::{
    EstimationResult, revm::context::result::ExecutionResult, revm::database::DBErrorMarker,
};

/// Response for `eth_syncing`.
///
/// Returns `false` when the node is fully synced, or a sync-status
/// object when it is still catching up.
#[derive(Debug, Clone, Serialize)]
#[serde(untagged)]
pub(crate) enum SyncingResponse {
    /// Node is fully synced.
    NotSyncing(bool),
    /// Node is still syncing.
    Syncing {
        /// Block number the node started syncing from.
        starting_block: U64,
        /// Current block the node has synced to.
        current_block: U64,
        /// Highest known block number on the network.
        highest_block: U64,
    },
}

/// `eth_syncing` — returns sync status or `false` when fully synced.
pub(crate) async fn syncing<H: HotKv>(ctx: StorageRpcCtx<H>) -> Result<SyncingResponse, ()> {
    match ctx.tags().sync_status() {
        Some(status) => Ok(SyncingResponse::Syncing {
            starting_block: U64::from(status.starting_block),
            current_block: U64::from(status.current_block),
            highest_block: U64::from(status.highest_block),
        }),
        None => Ok(SyncingResponse::NotSyncing(false)),
    }
}

/// Uncle count is always zero — Signet has no uncle blocks.
pub(crate) async fn uncle_count() -> Result<U64, ()> {
    Ok(U64::ZERO)
}

/// Uncle block is always absent — Signet has no uncle blocks.
pub(crate) async fn uncle_block() -> Result<Option<()>, ()> {
    Ok(None)
}

/// `eth_protocolVersion` — returns the Ethereum wire protocol version.
///
/// Signet does not implement devp2p. Returns a fixed value corresponding
/// to eth/68.
pub(crate) async fn protocol_version() -> Result<String, ()> {
    Ok("0x44".to_owned())
}

// ---------------------------------------------------------------------------
// Simple Queries
// ---------------------------------------------------------------------------

/// `eth_blockNumber` — returns the latest block number from block tags.
pub(crate) async fn block_number<H: HotKv>(ctx: StorageRpcCtx<H>) -> Result<U64, EthError> {
    Ok(U64::from(ctx.tags().latest()))
}

/// `eth_chainId` — returns the configured chain ID.
pub(crate) async fn chain_id<H: HotKv>(ctx: StorageRpcCtx<H>) -> Result<U64, ()> {
    Ok(U64::from(ctx.chain_id()))
}

// ---------------------------------------------------------------------------
// Gas & Fee Queries
// ---------------------------------------------------------------------------

/// `eth_gasPrice` — suggests gas price based on recent block tips + base fee.
pub(crate) async fn gas_price<H>(hctx: HandlerCtx, ctx: StorageRpcCtx<H>) -> Result<U256, EthError>
where
    H: HotKv + Send + Sync + 'static,
    <H::RoTx as HotKvRead>::Error: DBErrorMarker,
{
    let task = async move {
        let latest = ctx.tags().latest();
        let cold = ctx.cold();

        let tip = gas_oracle::suggest_tip_cap(&cold, latest, ctx.config(), ctx.gas_cache()).await?;

        let base_fee = cold
            .get_header_by_number(latest)
            .await?
            .and_then(|h| h.base_fee_per_gas)
            .unwrap_or_default();

        Ok(tip + U256::from(base_fee))
    };

    await_handler!(hctx.spawn(task), EthError::task_panic())
}

/// `eth_maxPriorityFeePerGas` — suggests priority fee from recent block tips.
pub(crate) async fn max_priority_fee_per_gas<H>(
    hctx: HandlerCtx,
    ctx: StorageRpcCtx<H>,
) -> Result<U256, EthError>
where
    H: HotKv + Send + Sync + 'static,
    <H::RoTx as HotKvRead>::Error: DBErrorMarker,
{
    let task = async move {
        let latest = ctx.tags().latest();
        Ok(gas_oracle::suggest_tip_cap(&ctx.cold(), latest, ctx.config(), ctx.gas_cache()).await?)
    };

    await_handler!(hctx.spawn(task), EthError::task_panic())
}

/// `eth_feeHistory` — returns base fee and reward percentile data.
pub(crate) async fn fee_history<H>(
    hctx: HandlerCtx,
    FeeHistoryArgs(block_count, newest, reward_percentiles): FeeHistoryArgs,
    ctx: StorageRpcCtx<H>,
) -> Result<FeeHistory, EthError>
where
    H: HotKv + Send + Sync + 'static,
    <H::RoTx as HotKvRead>::Error: DBErrorMarker,
{
    let task = async move {
        let mut block_count = block_count.to::<u64>();

        if block_count == 0 {
            return Ok(FeeHistory::default());
        }

        let max_fee_history = if reward_percentiles.is_none() {
            ctx.config().max_header_history
        } else {
            ctx.config().max_block_history
        };

        block_count = block_count.min(max_fee_history);

        let newest = if newest.is_pending() {
            block_count = block_count.saturating_sub(1);
            BlockNumberOrTag::Latest
        } else {
            newest
        };

        let end_block = ctx.resolve_block_tag(newest);
        let end_block_plus = end_block + 1;

        block_count = block_count.min(end_block_plus);

        // Validate percentiles
        if let Some(percentiles) = &reward_percentiles
            && percentiles.windows(2).any(|w| w[0] > w[1] || w[0] > 100.)
        {
            return Err(EthError::InvalidParams("invalid reward percentiles".into()));
        }

        let start_block = end_block_plus - block_count;
        let cold = ctx.cold();

        let specs: Vec<_> = (start_block..=end_block).map(HeaderSpecifier::Number).collect();
        let headers = cold.get_headers(specs).await?;

        let mut base_fee_per_gas: Vec<u128> = Vec::with_capacity(headers.len() + 1);
        let mut gas_used_ratio: Vec<f64> = Vec::with_capacity(headers.len());
        let mut rewards: Vec<Vec<u128>> = Vec::new();

        for (offset, maybe_header) in headers.iter().enumerate() {
            let Some(header) = maybe_header else {
                return Err(EthError::Internal(format!(
                    "missing header at block {}",
                    start_block + offset as u64
                )));
            };

            base_fee_per_gas.push(header.base_fee_per_gas.unwrap_or_default() as u128);
            gas_used_ratio.push(if header.gas_limit > 0 {
                header.gas_used as f64 / header.gas_limit as f64
            } else {
                0.0
            });

            if let Some(percentiles) = &reward_percentiles {
                let block_num = start_block + offset as u64;

                let (txs, receipts) = tokio::try_join!(
                    cold.get_transactions_in_block(block_num),
                    cold.get_receipts_in_block(block_num),
                )?;

                let block_rewards = calculate_reward_percentiles(
                    percentiles,
                    header.gas_used,
                    header.base_fee_per_gas.unwrap_or_default(),
                    &txs,
                    &receipts,
                );
                rewards.push(block_rewards);
            }
        }

        // Next block base fee
        if let Some(last_header) = headers.last().and_then(|h| h.as_ref()) {
            base_fee_per_gas.push(
                last_header.next_block_base_fee(BaseFeeParams::ethereum()).unwrap_or_default()
                    as u128,
            );
        }

        // NB: Signet has no blob transactions; zero-filled for client compatibility.
        let base_fee_per_blob_gas = vec![0; base_fee_per_gas.len()];
        let blob_gas_used_ratio = vec![0.; gas_used_ratio.len()];

        Ok(FeeHistory {
            base_fee_per_gas,
            gas_used_ratio,
            base_fee_per_blob_gas,
            blob_gas_used_ratio,
            oldest_block: start_block,
            reward: reward_percentiles.map(|_| rewards),
        })
    };

    await_handler!(hctx.spawn(task), EthError::task_panic())
}

/// Calculate reward percentiles for a single block.
///
/// Sorts transactions by effective tip ascending, then walks
/// cumulative gas used to find the tip value at each percentile.
fn calculate_reward_percentiles(
    percentiles: &[f64],
    gas_used: u64,
    base_fee: u64,
    txs: &[signet_storage_types::RecoveredTx],
    receipts: &[signet_cold::ColdReceipt],
) -> Vec<u128> {
    if gas_used == 0 || txs.is_empty() {
        return vec![0; percentiles.len()];
    }

    // Pair each tx's effective tip with its gas used.
    let mut tx_gas_and_tip: Vec<(u64, u128)> = txs
        .iter()
        .zip(receipts.iter())
        .map(|(tx, receipt)| {
            let tip = tx.effective_tip_per_gas(base_fee).unwrap_or_default();
            (receipt.gas_used, tip)
        })
        .collect();

    // Sort by tip ascending
    tx_gas_and_tip.sort_by_key(|&(_, tip)| tip);

    let mut result = Vec::with_capacity(percentiles.len());
    let mut cumulative_gas: u64 = 0;
    let mut tx_idx = 0;

    for &percentile in percentiles {
        let threshold = (gas_used as f64 * percentile / 100.0) as u64;

        while tx_idx < tx_gas_and_tip.len() {
            cumulative_gas += tx_gas_and_tip[tx_idx].0;
            if cumulative_gas >= threshold {
                break;
            }
            tx_idx += 1;
        }

        let (_, tip) = tx_gas_and_tip.get(tx_idx).unwrap_or(tx_gas_and_tip.last().unwrap());
        result.push(*tip);
    }

    result
}

// ---------------------------------------------------------------------------
// Block Queries
// ---------------------------------------------------------------------------

/// `eth_getBlockByHash` / `eth_getBlockByNumber` — resolve block, fetch
/// header + transactions from cold storage, assemble RPC block response.
pub(crate) async fn block<T, H>(
    hctx: HandlerCtx,
    BlockParams(t, full): BlockParams<T>,
    ctx: StorageRpcCtx<H>,
) -> Result<Option<RpcBlock>, EthError>
where
    T: Into<BlockId>,
    H: HotKv + Send + Sync + 'static,
    <H::RoTx as HotKvRead>::Error: DBErrorMarker,
{
    let id = t.into();
    let full = full.unwrap_or(false);

    let task = async move {
        let cold = ctx.cold();
        let block_num = ctx.resolve_block_id(id)?;

        let (header, txs) = tokio::try_join!(
            cold.get_header_by_number(block_num),
            cold.get_transactions_in_block(block_num),
        )?;

        let Some(header) = header else {
            return Ok(None);
        };

        let block_hash = header.hash();
        let base_fee = header.base_fee_per_gas;

        let transactions = if full {
            BlockTransactions::Full { txs, block_num, block_hash, base_fee }
        } else {
            BlockTransactions::Hashes(txs)
        };

        Ok(Some(RpcBlock {
            header: alloy::rpc::types::Header {
                inner: header.into_inner(),
                hash: block_hash,
                total_difficulty: None,
                size: None,
            },
            transactions,
            uncles: EmptyArray,
        }))
    };

    await_handler!(hctx.spawn(task), EthError::task_panic())
}

/// `eth_getBlockTransactionCount*` — transaction count in a block.
pub(crate) async fn block_tx_count<T, H>(
    hctx: HandlerCtx,
    (t,): (T,),
    ctx: StorageRpcCtx<H>,
) -> Result<Option<U64>, EthError>
where
    T: Into<BlockId>,
    H: HotKv + Send + Sync + 'static,
    <H::RoTx as HotKvRead>::Error: DBErrorMarker,
{
    let id = t.into();

    let task = async move {
        let cold = ctx.cold();
        let block_num = ctx.resolve_block_id(id)?;

        Ok(Some(U64::from(cold.get_transaction_count(block_num).await?)))
    };

    await_handler!(hctx.spawn(task), EthError::task_panic())
}

/// `eth_getBlockReceipts` — all receipts in a block.
pub(crate) async fn block_receipts<H>(
    hctx: HandlerCtx,
    (id,): (BlockId,),
    ctx: StorageRpcCtx<H>,
) -> Result<Option<LazyReceipts>, EthError>
where
    H: HotKv + Send + Sync + 'static,
    <H::RoTx as HotKvRead>::Error: DBErrorMarker,
{
    let task = async move {
        let cold = ctx.cold();
        let block_num = ctx.resolve_block_id(id)?;

        let (header, txs, receipts) = tokio::try_join!(
            cold.get_header_by_number(block_num),
            cold.get_transactions_in_block(block_num),
            cold.get_receipts_in_block(block_num),
        )?;

        let Some(header) = header else {
            return Ok(None);
        };

        let base_fee = header.base_fee_per_gas;

        Ok(Some(LazyReceipts { txs, receipts, base_fee }))
    };

    await_handler!(hctx.spawn(task), EthError::task_panic())
}

/// `eth_getBlockHeaderByHash` / `eth_getBlockHeaderByNumber`.
pub(crate) async fn header_by<T, H>(
    hctx: HandlerCtx,
    (t,): (T,),
    ctx: StorageRpcCtx<H>,
) -> Result<Option<RpcHeader>, EthError>
where
    T: Into<BlockId>,
    H: HotKv + Send + Sync + 'static,
    <H::RoTx as HotKvRead>::Error: DBErrorMarker,
{
    let id = t.into();

    let task = async move {
        Ok(ctx.resolve_header(id)?.map(|sh| {
            let hash = sh.hash();
            alloy::rpc::types::Header {
                inner: sh.into_inner(),
                hash,
                total_difficulty: None,
                size: None,
            }
        }))
    };

    await_handler!(hctx.spawn_blocking(task), EthError::task_panic())
}

// ---------------------------------------------------------------------------
// Transaction Queries
// ---------------------------------------------------------------------------

/// `eth_getTransactionByHash` — look up transaction by hash from cold storage.
pub(crate) async fn transaction_by_hash<H>(
    hctx: HandlerCtx,
    (hash,): (B256,),
    ctx: StorageRpcCtx<H>,
) -> Result<Option<RpcTransaction>, EthError>
where
    H: HotKv + Send + Sync + 'static,
    <H::RoTx as HotKvRead>::Error: DBErrorMarker,
{
    let task = async move {
        let cold = ctx.cold();
        let Some(confirmed) = cold.get_tx_by_hash(hash).await? else {
            return Ok(None);
        };

        let (tx, meta) = confirmed.into_parts();

        // Fetch header for base_fee
        let header = cold.get_header_by_number(meta.block_number()).await?;
        let base_fee = header.and_then(|h| h.base_fee_per_gas);

        Ok(Some(build_rpc_transaction(&tx, &meta, base_fee)))
    };

    await_handler!(hctx.spawn(task), EthError::task_panic())
}

/// `eth_getRawTransactionByHash` — RLP-encoded transaction bytes.
pub(crate) async fn raw_transaction_by_hash<H>(
    hctx: HandlerCtx,
    (hash,): (B256,),
    ctx: StorageRpcCtx<H>,
) -> Result<Option<alloy::primitives::Bytes>, EthError>
where
    H: HotKv + Send + Sync + 'static,
    <H::RoTx as HotKvRead>::Error: DBErrorMarker,
{
    let task = async move {
        Ok(ctx.cold().get_tx_by_hash(hash).await?.map(|c| c.into_inner().encoded_2718().into()))
    };

    await_handler!(hctx.spawn(task), EthError::task_panic())
}

/// `eth_getTransactionByBlock*AndIndex` — transaction by position in block.
pub(crate) async fn transaction_by_block_and_index<T, H>(
    hctx: HandlerCtx,
    (t, index): (T, U64),
    ctx: StorageRpcCtx<H>,
) -> Result<Option<RpcTransaction>, EthError>
where
    T: Into<BlockId>,
    H: HotKv + Send + Sync + 'static,
    <H::RoTx as HotKvRead>::Error: DBErrorMarker,
{
    let id = t.into();

    let task = async move {
        let cold = ctx.cold();
        let block_num = ctx.resolve_block_id(id)?;

        let Some(confirmed) = cold.get_tx_by_block_and_index(block_num, index.to::<u64>()).await?
        else {
            return Ok(None);
        };

        let (tx, meta) = confirmed.into_parts();
        let header = cold.get_header_by_number(meta.block_number()).await?;
        let base_fee = header.and_then(|h| h.base_fee_per_gas);

        Ok(Some(build_rpc_transaction(&tx, &meta, base_fee)))
    };

    await_handler!(hctx.spawn(task), EthError::task_panic())
}

/// `eth_getRawTransactionByBlock*AndIndex` — raw RLP bytes by position.
pub(crate) async fn raw_transaction_by_block_and_index<T, H>(
    hctx: HandlerCtx,
    (t, index): (T, U64),
    ctx: StorageRpcCtx<H>,
) -> Result<Option<alloy::primitives::Bytes>, EthError>
where
    T: Into<BlockId>,
    H: HotKv + Send + Sync + 'static,
    <H::RoTx as HotKvRead>::Error: DBErrorMarker,
{
    let id = t.into();

    let task = async move {
        let cold = ctx.cold();
        let block_num = ctx.resolve_block_id(id)?;

        Ok(cold
            .get_tx_by_block_and_index(block_num, index.to::<u64>())
            .await?
            .map(|c| c.into_inner().encoded_2718().into()))
    };

    await_handler!(hctx.spawn(task), EthError::task_panic())
}

/// `eth_getTransactionReceipt` — receipt by tx hash. Fetches the receipt,
/// then the associated transaction and header for derived fields.
pub(crate) async fn transaction_receipt<H>(
    hctx: HandlerCtx,
    (hash,): (B256,),
    ctx: StorageRpcCtx<H>,
) -> Result<Option<RpcReceipt>, EthError>
where
    H: HotKv + Send + Sync + 'static,
    <H::RoTx as HotKvRead>::Error: DBErrorMarker,
{
    let task = async move {
        let cold = ctx.cold();

        let Some(cr) = cold.get_receipt(ReceiptSpecifier::TxHash(hash)).await? else {
            return Ok(None);
        };

        let (tx, header) = tokio::try_join!(
            cold.get_tx_by_hash(cr.tx_hash),
            cold.get_header_by_number(cr.block_number),
        )?;

        let tx = tx.ok_or(EthError::TransactionMissing(cr.tx_hash))?.into_inner();
        let base_fee = header.and_then(|h| h.base_fee_per_gas);

        Ok(Some(build_receipt(&cr, &tx, base_fee)))
    };

    await_handler!(hctx.spawn(task), EthError::task_panic())
}

// ---------------------------------------------------------------------------
// Account State (Hot Storage)
// ---------------------------------------------------------------------------

/// `eth_getBalance` — account balance at a given block from hot storage.
pub(crate) async fn balance<H>(
    hctx: HandlerCtx,
    AddrWithBlock(address, block): AddrWithBlock,
    ctx: StorageRpcCtx<H>,
) -> Result<U256, EthError>
where
    H: HotKv + Send + Sync + 'static,
    <H::RoTx as HotKvRead>::Error: DBErrorMarker,
{
    let id = block.unwrap_or(BlockId::latest());

    let task = async move {
        let (reader, height) = hot_reader_at_block(&ctx, id)?;
        let acct = reader
            .get_account_at_height(&address, Some(height))
            .map_err(|e| EthError::Internal(e.to_string()))?;

        Ok(acct.map(|a| a.balance).unwrap_or_default())
    };

    await_handler!(hctx.spawn_blocking(task), EthError::task_panic())
}

/// `eth_getStorageAt` — contract storage slot at a given block.
pub(crate) async fn storage_at<H>(
    hctx: HandlerCtx,
    StorageAtArgs(address, key, block): StorageAtArgs,
    ctx: StorageRpcCtx<H>,
) -> Result<B256, EthError>
where
    H: HotKv + Send + Sync + 'static,
    <H::RoTx as HotKvRead>::Error: DBErrorMarker,
{
    let id = block.unwrap_or(BlockId::latest());

    let task = async move {
        let (reader, height) = hot_reader_at_block(&ctx, id)?;
        let val = reader
            .get_storage_at_height(&address, &key, Some(height))
            .map_err(|e| EthError::Internal(e.to_string()))?;

        Ok(val.unwrap_or_default().to_be_bytes().into())
    };

    await_handler!(hctx.spawn_blocking(task), EthError::task_panic())
}

/// `eth_getTransactionCount` — account nonce at a given block.
pub(crate) async fn addr_tx_count<H>(
    hctx: HandlerCtx,
    AddrWithBlock(address, block): AddrWithBlock,
    ctx: StorageRpcCtx<H>,
) -> Result<U64, EthError>
where
    H: HotKv + Send + Sync + 'static,
    <H::RoTx as HotKvRead>::Error: DBErrorMarker,
{
    let id = block.unwrap_or(BlockId::latest());

    let task = async move {
        let (reader, height) = hot_reader_at_block(&ctx, id)?;
        let acct = reader
            .get_account_at_height(&address, Some(height))
            .map_err(|e| EthError::Internal(e.to_string()))?;

        Ok(U64::from(acct.map(|a| a.nonce).unwrap_or_default()))
    };

    await_handler!(hctx.spawn_blocking(task), EthError::task_panic())
}

/// `eth_getCode` — contract bytecode at a given block.
pub(crate) async fn code_at<H>(
    hctx: HandlerCtx,
    AddrWithBlock(address, block): AddrWithBlock,
    ctx: StorageRpcCtx<H>,
) -> Result<alloy::primitives::Bytes, EthError>
where
    H: HotKv + Send + Sync + 'static,
    <H::RoTx as HotKvRead>::Error: DBErrorMarker,
{
    let id = block.unwrap_or(BlockId::latest());

    let task = async move {
        let (reader, height) = hot_reader_at_block(&ctx, id)?;
        let acct = reader
            .get_account_at_height(&address, Some(height))
            .map_err(|e| EthError::Internal(e.to_string()))?;

        let Some(acct) = acct else {
            return Ok(alloy::primitives::Bytes::new());
        };

        let Some(code_hash) = acct.bytecode_hash else {
            return Ok(alloy::primitives::Bytes::new());
        };

        let code =
            reader.get_bytecode(&code_hash).map_err(|e| EthError::Internal(e.to_string()))?;

        Ok(code.map(|c| c.original_bytes()).unwrap_or_default())
    };

    await_handler!(hctx.spawn_blocking(task), EthError::task_panic())
}

// ---------------------------------------------------------------------------
// EVM Execution
// ---------------------------------------------------------------------------

/// Shared EVM call execution used by `eth_call` and `eth_estimateGas`.
///
/// Resolves the block, builds a revm instance with the requested state
/// and block overrides, then executes the transaction request.
pub(crate) async fn run_call<H>(
    hctx: HandlerCtx,
    TxParams(request, block, state_overrides, block_overrides): TxParams,
    ctx: StorageRpcCtx<H>,
) -> Result<ExecutionResult, EthError>
where
    H: HotKv + Send + Sync + 'static,
    <H::RoTx as HotKvRead>::Error: DBErrorMarker,
{
    let id = block.unwrap_or(BlockId::latest());
    let span = trace_span!("run_call", block_id = %id);

    let task = async move {
        let EvmBlockContext { header, db, spec_id } = ctx.resolve_evm_block(id)?;

        let mut trevm = signet_evm::signet_evm(db, ctx.constants().clone());
        trevm.set_spec_id(spec_id);
        let trevm = trevm.fill_cfg(&CfgFiller(ctx.chain_id())).fill_block(&header);

        let trevm = trevm
            .maybe_apply_state_overrides(state_overrides.as_ref())
            .map_err(|e| EthError::Internal(e.to_string()))?
            .maybe_apply_block_overrides(block_overrides.as_deref())
            .fill_tx(&request);

        let mut trevm = trevm;
        let new_gas = trevm.cap_tx_gas().map_err(|e| EthError::Internal(e.to_string()))?;
        if Some(new_gas) != request.gas {
            debug!(req_gas = ?request.gas, new_gas, "capping gas for call");
        }

        let result = trevm.call().map_err(|e| EthError::Internal(e.into_error().to_string()))?;
        Ok(result.0)
    }
    .instrument(span);

    await_handler!(hctx.spawn_blocking(task), EthError::task_panic())
}

/// `eth_call` — execute a call and return the output bytes.
///
/// Delegates to [`run_call`], then maps the execution result to raw
/// output bytes, revert data, or halt reason.
pub(crate) async fn call<H>(
    hctx: HandlerCtx,
    mut params: TxParams,
    ctx: StorageRpcCtx<H>,
) -> Result<alloy::primitives::Bytes, EthError>
where
    H: HotKv + Send + Sync + 'static,
    <H::RoTx as HotKvRead>::Error: DBErrorMarker,
{
    let max_gas = ctx.config().rpc_gas_cap;
    normalize_gas_stateless(&mut params.0, max_gas);

    await_handler!(
        hctx.spawn_with_ctx(|hctx| async move {
            let res = run_call(hctx, params, ctx).await?;

            match res {
                ExecutionResult::Success { output, .. } => Ok(output.data().clone()),
                ExecutionResult::Revert { output, .. } => {
                    Err(EthError::EvmRevert { output: output.clone() })
                }
                ExecutionResult::Halt { reason, .. } => {
                    Err(EthError::EvmHalt { reason: format!("{reason:?}") })
                }
            }
        }),
        EthError::task_panic()
    )
}

/// `eth_estimateGas` — estimate gas required for a transaction.
pub(crate) async fn estimate_gas<H>(
    hctx: HandlerCtx,
    TxParams(mut request, block, state_overrides, block_overrides): TxParams,
    ctx: StorageRpcCtx<H>,
) -> Result<U64, EthError>
where
    H: HotKv + Send + Sync + 'static,
    <H::RoTx as HotKvRead>::Error: DBErrorMarker,
{
    let max_gas = ctx.config().rpc_gas_cap;
    normalize_gas_stateless(&mut request, max_gas);

    let id = block.unwrap_or(BlockId::pending());
    let span = trace_span!("eth_estimateGas", block_id = %id);

    let task = async move {
        let EvmBlockContext { header, db, spec_id } = ctx.resolve_evm_block(id)?;

        let mut trevm = signet_evm::signet_evm(db, ctx.constants().clone());
        trevm.set_spec_id(spec_id);
        let trevm = trevm.fill_cfg(&CfgFiller(ctx.chain_id())).fill_block(&header);

        let trevm = trevm
            .maybe_apply_state_overrides(state_overrides.as_ref())
            .map_err(|e| EthError::Internal(e.to_string()))?
            .maybe_apply_block_overrides(block_overrides.as_deref())
            .fill_tx(&request);

        let (estimate, _) =
            trevm.estimate_gas().map_err(|e| EthError::Internal(e.into_error().to_string()))?;

        match estimate {
            EstimationResult::Success { limit, .. } => Ok(U64::from(limit)),
            EstimationResult::Revert { reason, .. } => {
                Err(EthError::EvmRevert { output: reason.clone() })
            }
            EstimationResult::Halt { reason, .. } => {
                Err(EthError::EvmHalt { reason: format!("{reason:?}") })
            }
        }
    }
    .instrument(span);

    await_handler!(hctx.spawn_blocking(task), EthError::task_panic())
}

/// `eth_createAccessList` — generate an access list for a transaction.
pub(crate) async fn create_access_list<H>(
    hctx: HandlerCtx,
    TxParams(mut request, block, state_overrides, block_overrides): TxParams,
    ctx: StorageRpcCtx<H>,
) -> Result<AccessListResult, EthError>
where
    H: HotKv + Send + Sync + 'static,
    <H::RoTx as HotKvRead>::Error: DBErrorMarker,
{
    let max_gas = ctx.config().rpc_gas_cap;
    normalize_gas_stateless(&mut request, max_gas);

    let id = block.unwrap_or(BlockId::pending());
    let span = trace_span!("eth_createAccessList", block_id = %id);

    let task = async move {
        let EvmBlockContext { header, db, spec_id } = ctx.resolve_evm_block(id)?;

        let mut trevm = signet_evm::signet_evm(db, ctx.constants().clone());
        trevm.set_spec_id(spec_id);
        let trevm = trevm.fill_cfg(&CfgFiller(ctx.chain_id())).fill_block(&header);

        let trevm = trevm
            .maybe_apply_state_overrides(state_overrides.as_ref())
            .map_err(|e| EthError::Internal(e.to_string()))?
            .maybe_apply_block_overrides(block_overrides.as_deref())
            .fill_tx(&request);

        let initial = request.access_list.clone().unwrap_or_default();
        let mut inspector = AccessListInspector::new(initial);

        let result = trevm
            .try_with_inspector(&mut inspector, |trevm| trevm.run())
            .map_err(signet_evm::EvmErrored::into_error);

        let (gas_used, error) = match result {
            Ok(ref trevm) => (U256::from(trevm.gas_used()), None),
            Err(ref e) => (U256::ZERO, Some(e.to_string())),
        };

        let access_list = inspector.into_access_list();

        Ok(AccessListResult { access_list, gas_used, error })
    }
    .instrument(span);

    await_handler!(hctx.spawn_blocking(task), EthError::task_panic())
}

// ---------------------------------------------------------------------------
// Transaction Submission
// ---------------------------------------------------------------------------

/// `eth_sendRawTransaction` — decode and forward a signed transaction.
///
/// The transaction is forwarded to the tx cache in a fire-and-forget
/// task; the hash is returned immediately.
pub(crate) async fn send_raw_transaction<H>(
    hctx: HandlerCtx,
    (tx,): (alloy::primitives::Bytes,),
    ctx: StorageRpcCtx<H>,
) -> Result<B256, EthError>
where
    H: HotKv + Send + Sync + 'static,
    <H::RoTx as HotKvRead>::Error: DBErrorMarker,
{
    let Some(tx_cache) = ctx.tx_cache().cloned() else {
        return Err(EthError::Internal("tx-cache URL not provided".into()));
    };

    let task = |hctx: HandlerCtx| async move {
        let envelope = alloy::consensus::TxEnvelope::decode_2718(&mut tx.as_ref())
            .map_err(|e| EthError::InvalidParams(e.to_string()))?;

        let hash = *envelope.tx_hash();
        hctx.spawn(async move {
            if let Err(e) = tx_cache.forward_raw_transaction(envelope).await {
                tracing::warn!(error = %e, %hash, "failed to forward raw transaction");
            }
        });

        Ok(hash)
    };

    await_handler!(hctx.spawn_blocking_with_ctx(task), EthError::task_panic())
}

// ---------------------------------------------------------------------------
// Logs
// ---------------------------------------------------------------------------

/// Drain a [`signet_cold::LogStream`] into a `Vec<Log>`.
///
/// Errors from the stream (deadline exceeded, too many logs, reorg) are
/// propagated as the first encountered error.
async fn collect_log_stream(stream: signet_cold::LogStream) -> signet_cold::ColdResult<Vec<Log>> {
    use tokio_stream::StreamExt;
    let mut logs = Vec::new();
    let mut stream = std::pin::pin!(stream);
    while let Some(log) = stream.next().await {
        logs.push(log?);
    }
    Ok(logs)
}

/// `eth_getLogs` — query logs from cold storage with filter criteria.
///
/// Uses `stream_logs` for deadline enforcement and dedicated concurrency
/// control. The stream is collected into a `Vec` for the JSON-RPC response.
pub(crate) async fn get_logs<H>(
    hctx: HandlerCtx,
    (filter,): (Filter,),
    ctx: StorageRpcCtx<H>,
) -> Result<Vec<Log>, EthError>
where
    H: HotKv + Send + Sync + 'static,
    <H::RoTx as HotKvRead>::Error: DBErrorMarker,
{
    let task = async move {
        let cold = ctx.cold();

        let resolved_filter = match filter.block_option {
            alloy::rpc::types::FilterBlockOption::AtBlockHash(_) => filter,
            alloy::rpc::types::FilterBlockOption::Range { from_block, to_block } => {
                let from = from_block
                    .map(|b| ctx.resolve_block_tag(b))
                    .unwrap_or_else(|| ctx.tags().latest());
                let to = to_block
                    .map(|b| ctx.resolve_block_tag(b))
                    .unwrap_or_else(|| ctx.tags().latest());

                if from > to {
                    return Err(EthError::InvalidParams(
                        "fromBlock must not exceed toBlock".into(),
                    ));
                }
                let max_blocks = ctx.config().max_blocks_per_filter;
                if to - from >= max_blocks {
                    return Err(EthError::InvalidParams(format!(
                        "query exceeds max block range ({max_blocks})"
                    )));
                }

                Filter {
                    block_option: alloy::rpc::types::FilterBlockOption::Range {
                        from_block: Some(BlockNumberOrTag::Number(from)),
                        to_block: Some(BlockNumberOrTag::Number(to)),
                    },
                    ..filter
                }
            }
        };

        let max_logs = ctx.config().max_logs_per_response;
        let deadline = ctx.config().max_log_query_deadline;

        let stream = cold.stream_logs(resolved_filter, max_logs, deadline).await?;

        Ok(collect_log_stream(stream).await?)
    };

    await_handler!(hctx.spawn(task), EthError::task_panic())
}

// ---------------------------------------------------------------------------
// Filters
// ---------------------------------------------------------------------------

/// `eth_newFilter` — install a log filter for polling.
pub(crate) async fn new_filter<H>(
    hctx: HandlerCtx,
    (filter,): (Filter,),
    ctx: StorageRpcCtx<H>,
) -> Result<U64, EthError>
where
    H: HotKv + Send + Sync + 'static,
    <H::RoTx as HotKvRead>::Error: DBErrorMarker,
{
    let task = async move {
        let latest = ctx.tags().latest();
        Ok(ctx.filter_manager().install_log_filter(latest, filter))
    };

    await_handler!(hctx.spawn_blocking(task), EthError::task_panic())
}

/// `eth_newBlockFilter` — install a block hash filter for polling.
pub(crate) async fn new_block_filter<H>(
    hctx: HandlerCtx,
    ctx: StorageRpcCtx<H>,
) -> Result<U64, EthError>
where
    H: HotKv + Send + Sync + 'static,
    <H::RoTx as HotKvRead>::Error: DBErrorMarker,
{
    let task = async move {
        let latest = ctx.tags().latest();
        Ok(ctx.filter_manager().install_block_filter(latest))
    };

    await_handler!(hctx.spawn_blocking(task), EthError::task_panic())
}

/// `eth_uninstallFilter` — remove a filter.
pub(crate) async fn uninstall_filter<H>(
    hctx: HandlerCtx,
    (id,): (U64,),
    ctx: StorageRpcCtx<H>,
) -> Result<bool, EthError>
where
    H: HotKv + Send + Sync + 'static,
    <H::RoTx as HotKvRead>::Error: DBErrorMarker,
{
    let task = async move { Ok(ctx.filter_manager().uninstall(id).is_some()) };
    await_handler!(hctx.spawn_blocking(task), EthError::task_panic())
}

/// `eth_getFilterChanges` / `eth_getFilterLogs` — poll a filter for new
/// results since the last poll. Fetches matching data from cold storage.
pub(crate) async fn get_filter_changes<H>(
    hctx: HandlerCtx,
    (id,): (U64,),
    ctx: StorageRpcCtx<H>,
) -> Result<FilterOutput, EthError>
where
    H: HotKv + Send + Sync + 'static,
    <H::RoTx as HotKvRead>::Error: DBErrorMarker,
{
    let task = async move {
        let fm = ctx.filter_manager();
        let mut entry = fm
            .get_mut(id)
            .ok_or_else(|| EthError::InvalidParams(format!("filter not found: {id}")))?;

        // Scan the global reorg ring buffer for notifications received
        // since this filter's last poll, then lazily compute removed logs
        // and rewind `next_start_block`.
        let reorgs = fm.reorgs_since(entry.last_poll_time());
        let removed = entry.compute_removed_logs(&reorgs);
        if !removed.is_empty() {
            trace!(count = removed.len(), "computed removed logs from reorg ring buffer");
        }

        let latest = ctx.tags().latest();
        let start = entry.next_start_block();

        // Implicit reorg detection: if latest has moved backward past our
        // window, a reorg occurred that we missed (e.g. broadcast lagged).
        // Return any removed logs we do have, then reset.
        if latest + 1 < start {
            trace!(latest, start, "implicit reorg detected, resetting filter");
            entry.touch_poll_time();
            return Ok(if removed.is_empty() {
                entry.empty_output()
            } else {
                FilterOutput::from(removed)
            });
        }

        if start > latest {
            entry.touch_poll_time();
            return Ok(if removed.is_empty() {
                entry.empty_output()
            } else {
                FilterOutput::from(removed)
            });
        }

        let cold = ctx.cold();

        if entry.is_block() {
            let specs: Vec<_> = (start..=latest).map(HeaderSpecifier::Number).collect();
            let headers = cold.get_headers(specs).await?;
            let hashes: Vec<B256> = headers.into_iter().flatten().map(|h| h.hash()).collect();
            entry.mark_polled(latest);
            Ok(FilterOutput::from(hashes))
        } else {
            let stored = entry.as_filter().cloned().unwrap();
            let resolved = Filter {
                block_option: alloy::rpc::types::FilterBlockOption::Range {
                    from_block: Some(BlockNumberOrTag::Number(start)),
                    to_block: Some(BlockNumberOrTag::Number(latest)),
                },
                ..stored
            };

            let max_logs = ctx.config().max_logs_per_response;
            let deadline = ctx.config().max_log_query_deadline;

            let stream = cold.stream_logs(resolved, max_logs, deadline).await?;

            let mut logs = collect_log_stream(stream).await?;

            // Prepend removed logs so the client sees removals before
            // the replacement data.
            if !removed.is_empty() {
                let mut combined = removed;
                combined.append(&mut logs);
                logs = combined;
            }

            entry.mark_polled(latest);
            Ok(FilterOutput::from(logs))
        }
    };

    await_handler!(hctx.spawn(task), EthError::task_panic())
}

// ---------------------------------------------------------------------------
// Subscriptions
// ---------------------------------------------------------------------------

/// `eth_subscribe` — register a push-based subscription (WebSocket/SSE).
pub(crate) async fn subscribe<H>(
    hctx: HandlerCtx,
    sub: SubscribeArgs,
    ctx: StorageRpcCtx<H>,
) -> Result<U64, EthError>
where
    H: HotKv + Send + Sync + 'static,
    <H::RoTx as HotKvRead>::Error: DBErrorMarker,
{
    let interest: InterestKind = sub.try_into()?;

    ctx.sub_manager()
        .subscribe(&hctx, interest)
        .ok_or_else(|| EthError::Internal("notifications not enabled on this transport".into()))
}

/// `eth_unsubscribe` — cancel a push-based subscription.
pub(crate) async fn unsubscribe<H>(
    hctx: HandlerCtx,
    (id,): (U64,),
    ctx: StorageRpcCtx<H>,
) -> Result<bool, EthError>
where
    H: HotKv + Send + Sync + 'static,
    <H::RoTx as HotKvRead>::Error: DBErrorMarker,
{
    let task = async move { Ok(ctx.sub_manager().unsubscribe(id)) };
    await_handler!(hctx.spawn_blocking(task), EthError::task_panic())
}
