//! RPC context wrapping [`UnifiedStorage`].
//!
//! # Consistency Model
//!
//! The RPC layer reads from two storage tiers with different consistency
//! guarantees:
//!
//! - **Hot storage** (MDBX): synchronous writes, MVCC snapshot isolation,
//!   authoritative source for state and headers. Reads open an MDBX read
//!   transaction that provides a consistent point-in-time snapshot.
//!
//! - **Cold storage** (async task): eventually consistent. Writes are
//!   dispatched asynchronously after hot storage commits via
//!   `append_blocks()`, so cold may lag by milliseconds to seconds under
//!   normal operation.
//!
//! # Query Routing
//!
//! | Query type | Resolution | Data source | Staleness risk |
//! |---|---|---|---|
//! | State (`getBalance`, `getStorageAt`, `getCode`, `getTransactionCount`) | Hot tag → height | Hot | Low — single tier, single transaction |
//! | EVM execution (`eth_call`, `estimateGas`) | Hot tag → height | Hot | Low — single tier, single transaction |
//! | Block/header queries | Hot tag → height | Cold | Medium — cold lag |
//! | Transaction queries | Hot tag or hash → height | Cold | Medium — cold lag |
//! | Receipt queries | Hot tag or hash → height | Cold | Medium — cold lag |
//! | Log queries (`getLogs`) | Hot tag → height range | Cold | Medium — cold lag |
//! | Filter changes | Hot tag → latest | Cold | Medium — reorg detection via ring buffer mitigates |
//!
//! # Resolve-then-Query Pattern
//!
//! Most endpoints follow a two-step pattern:
//!
//! 1. **Resolve** a block tag or hash to a concrete block number (reads
//!    atomic tag values or queries hot storage's `HeaderNumbers` table).
//! 2. **Query** the resolved block number against hot or cold storage.
//!
//! For hot-only queries (state, EVM), both steps share a single MDBX
//! read transaction, eliminating races between resolution and query.
//!
//! For cold queries, the resolved number is passed to cold storage.
//! Between resolution and cold query, tags can advance or a reorg can
//! replace the block at that height. The caller gets a consistent view
//! of the **resolved** height but may miss a newer block. This is
//! acceptable per JSON-RPC spec — clients retry on stale data.
//!
//! # Future Work: Hash-based Consistency Verification
//!
//! For queries where correctness is critical (e.g., `eth_call`), a
//! stronger guarantee is possible: after resolving the block number,
//! read the block hash from hot storage and pass both (number, hash) to
//! cold. Cold verifies the hash matches before returning data, catching
//! reorgs that replaced the block between resolution and query.
//!
//! This was deferred because: (a) the reorg window is small
//! (milliseconds), (b) it adds one extra hot storage read per query,
//! and (c) for read-only queries the impact of returning stale data is
//! low. See ENG-1901 for the full trade-off analysis.

use crate::{
    config::{
        ChainNotifier, GasOracleCache, StorageRpcConfig,
        resolve::{BlockTags, ResolveError},
    },
    eth::EthError,
    interest::{FilterManager, SubscriptionManager},
};
use alloy::{
    eips::{BlockId, BlockNumberOrTag},
    genesis::ChainConfig,
};
use signet_cold::ColdStorageReadHandle;
use signet_evm::EthereumHardfork;
use signet_hot::{
    HotKv,
    db::HotDbRead,
    model::{HotKvRead, RevmRead},
};
use signet_storage::UnifiedStorage;
use signet_tx_cache::TxCache;
use signet_types::constants::SignetSystemConstants;
use std::sync::Arc;
use tokio::sync::Semaphore;
use trevm::revm::{
    database::{DBErrorMarker, StateBuilder},
    primitives::hardfork::SpecId,
};

/// Resolved block context for EVM execution.
///
/// Contains the header and a revm-compatible database snapshot at the
/// resolved block height, ready for use with `signet_evm`.
#[derive(Debug)]
pub(crate) struct EvmBlockContext<Db> {
    /// The resolved block header.
    pub header: alloy::consensus::Header,
    /// The revm database at the resolved height, wrapped in [`State`] to
    /// provide [`DatabaseCommit`] support required by `eth_estimateGas`
    /// (binary-search commits), `eth_createAccessList`, `signet_callBundle`,
    /// and debug tracing.
    ///
    /// [`State`]: trevm::revm::database::State
    /// [`DatabaseCommit`]: trevm::revm::database::DatabaseCommit
    pub db: trevm::revm::database::State<Db>,
    /// The EVM hardfork spec active at this block.
    pub spec_id: SpecId,
}

/// RPC context backed by [`UnifiedStorage`].
///
/// Provides access to hot storage (state, headers), cold storage
/// (blocks, transactions, receipts, logs), block tag resolution,
/// filter/subscription management, and optional transaction forwarding.
///
/// Hot-only queries (state reads, EVM execution) open a single MDBX
/// read transaction for both block resolution and data access. Cold
/// queries resolve a block number from hot storage, then pass it to
/// the cold read handle.
///
/// See the module-level documentation for the full consistency model.
///
/// # Construction
///
/// Call [`StorageRpcCtx::new`] with unified storage, system constants,
/// a [`ChainNotifier`], an optional [`TxCache`], and [`StorageRpcConfig`].
#[derive(Debug)]
pub struct StorageRpcCtx<H: HotKv> {
    inner: Arc<StorageRpcCtxInner<H>>,
}

impl<H: HotKv> Clone for StorageRpcCtx<H> {
    fn clone(&self) -> Self {
        Self { inner: Arc::clone(&self.inner) }
    }
}

#[derive(Debug)]
struct StorageRpcCtxInner<H: HotKv> {
    storage: Arc<UnifiedStorage<H>>,
    constants: SignetSystemConstants,
    chain_config: ChainConfig,
    chain: ChainNotifier,
    tx_cache: Option<TxCache>,
    config: StorageRpcConfig,
    tracing_semaphore: Arc<Semaphore>,
    filter_manager: FilterManager,
    sub_manager: SubscriptionManager,
    gas_cache: GasOracleCache,
}

impl<H: HotKv> StorageRpcCtx<H> {
    /// Create a new storage-backed RPC context.
    ///
    /// The [`ChainNotifier`] provides block tag tracking and a broadcast
    /// sender for new block notifications. The subscription manager
    /// subscribes to this channel to push updates to WebSocket clients.
    ///
    /// The `chain_config` is the rollup genesis chain configuration, used
    /// to determine the active EVM hardfork (spec ID) for each block.
    pub fn new(
        storage: Arc<UnifiedStorage<H>>,
        constants: SignetSystemConstants,
        chain_config: ChainConfig,
        chain: ChainNotifier,
        tx_cache: Option<TxCache>,
        config: StorageRpcConfig,
    ) -> Self {
        let tracing_semaphore = Arc::new(Semaphore::new(config.max_tracing_requests));
        let filter_manager =
            FilterManager::new(chain.clone(), config.stale_filter_ttl, config.stale_filter_ttl);
        let sub_manager = SubscriptionManager::new(chain.notif_sender(), config.stale_filter_ttl);
        let gas_cache = GasOracleCache::new();
        Self {
            inner: Arc::new(StorageRpcCtxInner {
                storage,
                constants,
                chain_config,
                chain,
                tx_cache,
                config,
                tracing_semaphore,
                filter_manager,
                sub_manager,
                gas_cache,
            }),
        }
    }

    /// Access the unified storage.
    pub fn storage(&self) -> &UnifiedStorage<H> {
        &self.inner.storage
    }

    /// Get a cold storage read handle.
    pub fn cold(&self) -> ColdStorageReadHandle {
        self.inner.storage.cold_reader()
    }

    /// Get a hot storage read transaction.
    pub fn hot_reader(&self) -> signet_storage::StorageResult<H::RoTx> {
        self.inner.storage.reader()
    }

    /// Access the block tags.
    pub fn tags(&self) -> &BlockTags {
        self.inner.chain.tags()
    }

    /// Access the system constants.
    pub fn constants(&self) -> &SignetSystemConstants {
        &self.inner.constants
    }

    /// Access the chain configuration (rollup genesis forks).
    pub(crate) fn chain_config(&self) -> &ChainConfig {
        &self.inner.chain_config
    }

    /// Determine the EVM [`SpecId`] for a given block header.
    pub(crate) fn spec_id_for_header(&self, header: &alloy::consensus::Header) -> SpecId {
        EthereumHardfork::active_hardforks_at_header(self.chain_config(), header).spec_id()
    }

    /// Get the chain ID.
    pub fn chain_id(&self) -> u64 {
        self.inner.constants.ru_chain_id()
    }

    /// Access the RPC configuration.
    pub fn config(&self) -> &StorageRpcConfig {
        &self.inner.config
    }

    /// Acquire a permit from the tracing semaphore.
    ///
    /// Limits concurrent tracing/debug requests. Callers should hold
    /// the permit for the duration of their tracing operation.
    pub async fn acquire_tracing_permit(&self) -> tokio::sync::OwnedSemaphorePermit {
        Arc::clone(&self.inner.tracing_semaphore)
            .acquire_owned()
            .await
            .expect("tracing semaphore closed")
    }

    /// Access the optional tx cache.
    pub fn tx_cache(&self) -> Option<&TxCache> {
        self.inner.tx_cache.as_ref()
    }

    /// Access the filter manager.
    pub(crate) fn filter_manager(&self) -> &FilterManager {
        &self.inner.filter_manager
    }

    /// Access the subscription manager.
    pub(crate) fn sub_manager(&self) -> &SubscriptionManager {
        &self.inner.sub_manager
    }

    /// Access the gas oracle cache.
    pub(crate) fn gas_cache(&self) -> &GasOracleCache {
        &self.inner.gas_cache
    }

    /// Resolve a [`BlockNumberOrTag`] to a block number.
    ///
    /// This is synchronous — no cold storage lookup is needed.
    ///
    /// - `Latest` / `Pending` → latest tag
    /// - `Safe` → safe tag
    /// - `Finalized` → finalized tag
    /// - `Earliest` → `0`
    /// - `Number(n)` → `n`
    pub(crate) fn resolve_block_tag(&self, tag: BlockNumberOrTag) -> u64 {
        match tag {
            BlockNumberOrTag::Latest | BlockNumberOrTag::Pending => self.tags().latest(),
            BlockNumberOrTag::Safe => self.tags().safe(),
            BlockNumberOrTag::Finalized => self.tags().finalized(),
            BlockNumberOrTag::Earliest => 0,
            BlockNumberOrTag::Number(n) => n,
        }
    }

    /// Resolve a [`BlockId`] to a block number.
    ///
    /// For tag/number-based IDs, resolves synchronously via
    /// [`resolve_block_tag`](Self::resolve_block_tag). For hash-based IDs,
    /// looks up the block number from hot storage's `HeaderNumbers` table.
    pub(crate) fn resolve_block_id(&self, id: BlockId) -> Result<u64, ResolveError>
    where
        <H::RoTx as HotKvRead>::Error: std::error::Error + Send + Sync + 'static,
    {
        match id {
            BlockId::Number(tag) => Ok(self.resolve_block_tag(tag)),
            BlockId::Hash(h) => {
                let reader = self.hot_reader()?;
                reader
                    .get_header_number(&h.block_hash)
                    .map_err(|e| ResolveError::Db(Box::new(e)))?
                    .ok_or(ResolveError::HashNotFound(h.block_hash))
            }
        }
    }

    /// Resolve a [`BlockId`] to a header from hot storage.
    ///
    /// For hash-based IDs, fetches the header directly by hash. For
    /// tag/number-based IDs, resolves the tag then fetches the header by
    /// number. Returns `None` if the header is not found.
    pub(crate) fn resolve_header(
        &self,
        id: BlockId,
    ) -> Result<Option<signet_storage_types::SealedHeader>, ResolveError>
    where
        <H::RoTx as HotKvRead>::Error: std::error::Error + Send + Sync + 'static,
    {
        let reader = self.hot_reader()?;
        match id {
            BlockId::Hash(h) => {
                reader.header_by_hash(&h.block_hash).map_err(|e| ResolveError::Db(Box::new(e)))
            }
            BlockId::Number(tag) => {
                let height = self.resolve_block_tag(tag);
                reader.get_header(height).map_err(|e| ResolveError::Db(Box::new(e)))
            }
        }
    }

    /// Create a revm-compatible database at a specific block height.
    ///
    /// Wraps the underlying `RevmRead` in [`State`] so that the returned
    /// database implements both `Database` and `DatabaseCommit`. This is
    /// needed by EVM operations that mutate intermediate state (gas
    /// estimation, access-list generation, bundle simulation, debug
    /// tracing).
    ///
    /// [`State`]: trevm::revm::database::State
    pub fn revm_state_at_height(
        &self,
        height: u64,
    ) -> signet_storage::StorageResult<trevm::revm::database::State<RevmRead<H::RoTx>>>
    where
        <H::RoTx as HotKvRead>::Error: DBErrorMarker,
    {
        let revm_read = self.inner.storage.revm_reader_at_height(height)?;
        Ok(StateBuilder::new_with_database(revm_read).build())
    }

    /// Resolve a [`BlockId`] to a header and revm database in one pass.
    ///
    /// Opens a single MDBX read transaction and uses it for both the
    /// header lookup and the revm-compatible database snapshot. This
    /// ensures the header and EVM state come from the same database
    /// snapshot, eliminating races where a reorg between two separate
    /// transactions could yield a header from one fork and state from
    /// another.
    ///
    /// For `Pending` block IDs, remaps to `Latest` and synthesizes a
    /// next-block header (incremented number, timestamp +12s, projected
    /// base fee). State is loaded at the latest block in both cases.
    pub(crate) fn resolve_evm_block(
        &self,
        id: BlockId,
    ) -> Result<EvmBlockContext<RevmRead<H::RoTx>>, EthError>
    where
        <H::RoTx as HotKvRead>::Error: DBErrorMarker + std::error::Error + Send + Sync + 'static,
    {
        let pending = id.is_pending();
        let id = if pending { BlockId::latest() } else { id };

        // Single MDBX read transaction for both header lookup and
        // RevmRead construction. `hot_reader()` returns
        // `Result<H::RoTx, StorageError>` — the `?` converts through
        // `StorageError -> EthError::Hot`.
        let reader = self.hot_reader()?;

        let sealed = match id {
            BlockId::Hash(h) => {
                reader.header_by_hash(&h.block_hash).map_err(|e| ResolveError::Db(Box::new(e)))?
            }
            BlockId::Number(tag) => {
                let height = self.resolve_block_tag(tag);
                reader.get_header(height).map_err(|e| ResolveError::Db(Box::new(e)))?
            }
        }
        .ok_or(EthError::BlockNotFound(id))?;

        let height = sealed.number;
        let parent_hash = sealed.hash();
        let mut header = sealed.into_inner();

        if pending {
            header.parent_hash = parent_hash;
            header.number += 1;
            header.timestamp += self.constants().host().slot_duration();
            header.base_fee_per_gas =
                header.next_block_base_fee(alloy::eips::eip1559::BaseFeeParams::ethereum());
        }

        let spec_id = self.spec_id_for_header(&header);

        // Wrap the same reader in RevmRead — no range validation needed
        // because we already proved the height exists by fetching the header.
        let db = StateBuilder::new_with_database(RevmRead::at_height(reader, height)).build();

        Ok(EvmBlockContext { header, db, spec_id })
    }
}
