blvm-node 0.1.2

//! Mempool manager
//!
//! Handles transaction mempool management, validation, and relay.

use crate::config::{MempoolPolicyConfig, RbfConfig};
use crate::node::event_publisher::EventPublisher;
use crate::utils::MEMPOOL_LOOP_SLEEP;
use anyhow::Result;
use blvm_protocol::mempool::{has_conflict_with_tx, replacement_checks, signals_rbf, Mempool};
use blvm_protocol::{Hash, OutPoint, Transaction, UtxoSet};
use std::cmp::Reverse;
use std::collections::{BTreeMap, HashMap, HashSet};
use std::sync::{Arc, Mutex, RwLock};
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use tracing::{debug, info, warn};

/// RBF tracking information for a transaction
#[derive(Debug, Clone)]
struct RbfTracking {
    /// Number of times this transaction has been replaced
    replacement_count: u32,
    /// Timestamp of last replacement (Unix timestamp)
    last_replacement_time: u64,
    /// Original transaction hash (before any replacements)
    original_tx_hash: Hash,
}

/// Core mempool state (transactions + spent outputs)
/// Wrapped in Mutex for add_transaction from Arc context (re-broadcast, sendrawtransaction)
struct MempoolPool {
    transactions: HashMap<Hash, Transaction>,
    spent_outputs: HashSet<OutPoint>,
}

/// Mempool manager
pub struct MempoolManager {
    /// Transaction mempool + spent outputs (interior mutability for Arc<MempoolManager>)
    pool: Mutex<MempoolPool>,
    /// Legacy mempool (HashSet of hashes) for compatibility
    #[allow(dead_code)]
    mempool: RwLock<Mempool>,
    #[allow(dead_code)]
    utxo_set: UtxoSet,
    /// Event callback for mempool events (optional)
    /// Called when transactions are added/removed from mempool
    #[allow(dead_code)]
    event_callback: Option<Box<dyn Fn(Hash, String, usize) + Send + Sync>>,
    /// Sorted index by fee rate (descending) - Reverse<u64> for descending order
    /// Maps fee_rate -> Vec<Hash> (multiple transactions can have same fee rate)
    /// Uses RwLock for interior mutability to allow &self methods
    fee_index: RwLock<BTreeMap<Reverse<u64>, Vec<Hash>>>,
    /// Cache fee rates per transaction hash
    /// Uses RwLock for interior mutability to allow &self methods
    fee_cache: RwLock<HashMap<Hash, u64>>,
    /// RBF configuration (optional)
    /// Uses RwLock for interior mutability to allow setting config after Arc sharing
    rbf_config: RwLock<Option<RbfConfig>>,
    /// Mempool policy configuration (optional)
    /// Uses RwLock for interior mutability to allow setting config after Arc sharing
    policy_config: RwLock<Option<MempoolPolicyConfig>>,
    /// RBF tracking: transaction hash -> RBF tracking info
    /// Uses RwLock for interior mutability
    rbf_tracking: RwLock<HashMap<Hash, RbfTracking>>,
    /// Transaction timestamps: when each transaction was added
    /// Uses RwLock for interior mutability
    tx_timestamps: RwLock<HashMap<Hash, u64>>,
    /// Transaction dependency graph: child -> parent relationships
    /// Maps transaction hash to set of parent transaction hashes (transactions it depends on)
    /// Uses RwLock for interior mutability
    tx_dependencies: RwLock<HashMap<Hash, HashSet<Hash>>>,
    /// Reverse dependency graph: parent -> children relationships
    /// Maps transaction hash to set of child transaction hashes (transactions that depend on it)
    /// Uses RwLock for interior mutability
    tx_descendants: RwLock<HashMap<Hash, HashSet<Hash>>>,
    /// UTXO set hash for change detection (optimization: only recalculate when UTXO set changes)
    /// Uses RwLock for interior mutability
    utxo_set_hash: RwLock<Option<u64>>,
    /// Event publisher for mempool events (optional)
    /// Uses Arc for shared ownership and interior mutability
    event_publisher: RwLock<Option<Arc<EventPublisher>>>,
}

impl MempoolManager {
    /// Create a new mempool manager
    pub fn new() -> Self {
        Self {
            pool: Mutex::new(MempoolPool {
                transactions: HashMap::new(),
                spent_outputs: HashSet::new(),
            }),
            mempool: RwLock::new(Mempool::new()),
            utxo_set: UtxoSet::default(),
            event_callback: None,
            fee_index: RwLock::new(BTreeMap::new()),
            fee_cache: RwLock::new(HashMap::new()),
            rbf_config: RwLock::new(None),
            policy_config: RwLock::new(None),
            rbf_tracking: RwLock::new(HashMap::new()),
            tx_timestamps: RwLock::new(HashMap::new()),
            tx_dependencies: RwLock::new(HashMap::new()),
            tx_descendants: RwLock::new(HashMap::new()),
            utxo_set_hash: RwLock::new(None),
            event_publisher: RwLock::new(None),
        }
    }

    /// Create a new mempool manager with RBF configuration
    pub fn with_rbf_config(rbf_config: Option<RbfConfig>) -> Self {
        Self {
            pool: Mutex::new(MempoolPool {
                transactions: HashMap::new(),
                spent_outputs: HashSet::new(),
            }),
            mempool: RwLock::new(Mempool::new()),
            utxo_set: UtxoSet::default(),
            event_callback: None,
            fee_index: RwLock::new(BTreeMap::new()),
            fee_cache: RwLock::new(HashMap::new()),
            rbf_config: RwLock::new(rbf_config),
            policy_config: RwLock::new(None),
            rbf_tracking: RwLock::new(HashMap::new()),
            tx_timestamps: RwLock::new(HashMap::new()),
            tx_dependencies: RwLock::new(HashMap::new()),
            tx_descendants: RwLock::new(HashMap::new()),
            utxo_set_hash: RwLock::new(None),
            event_publisher: RwLock::new(None),
        }
    }

    /// Set event publisher for mempool events
    /// Uses interior mutability so it can be called even when MempoolManager is in an Arc
    pub fn set_event_publisher(&self, event_publisher: Option<Arc<EventPublisher>>) {
        *self.event_publisher.write().unwrap() = event_publisher;
    }

    /// Set RBF configuration
    /// Uses interior mutability so it can be called even when MempoolManager is in an Arc
    pub fn set_rbf_config(&self, rbf_config: Option<RbfConfig>) {
        *self.rbf_config.write().unwrap() = rbf_config;
    }

    /// Set mempool policy configuration
    /// Uses interior mutability so it can be called even when MempoolManager is in an Arc
    pub fn set_policy_config(&self, policy_config: Option<MempoolPolicyConfig>) {
        *self.policy_config.write().unwrap() = policy_config;
    }

    /// Lock pool for access (transactions + spent_outputs)
    fn pool_lock(&self) -> std::sync::MutexGuard<'_, MempoolPool> {
        self.pool.lock().unwrap()
    }

    /// Get current timestamp (Unix seconds)
    fn current_timestamp() -> u64 {
        SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap_or(Duration::ZERO)
            .as_secs()
    }

    /// Start the mempool manager
    pub async fn start(&mut self) -> Result<()> {
        info!("Starting mempool manager");

        // Initialize mempool
        self.initialize_mempool().await?;

        // Start mempool processing loop
        self.process_loop().await?;

        Ok(())
    }

    /// Run mempool processing once (for testing)
    pub async fn process_once(&mut self) -> Result<()> {
        // Process pending transactions
        self.process_pending_transactions().await?;

        // Clean up old transactions
        self.cleanup_old_transactions().await?;

        Ok(())
    }

    /// Initialize mempool
    async fn initialize_mempool(&mut self) -> Result<()> {
        debug!("Initializing mempool");

        // Load existing mempool from storage if available
        // In a real implementation, this would restore mempool state

        Ok(())
    }

    /// Main mempool processing loop
    async fn process_loop(&mut self) -> Result<()> {
        loop {
            // Process pending transactions
            self.process_pending_transactions().await?;

            // Clean up old transactions
            self.cleanup_old_transactions().await?;

            // Small delay to prevent busy waiting
            tokio::time::sleep(MEMPOOL_LOOP_SLEEP).await;
        }
    }

    /// Process pending transactions
    async fn process_pending_transactions(&mut self) -> Result<()> {
        // In a real implementation, this would:
        // 1. Get new transactions from network
        // 2. Validate transactions using blvm-consensus
        // 3. Add valid transactions to mempool
        // 4. Relay transactions to peers

        debug!("Processing pending transactions");
        Ok(())
    }

    /// Clean up old transactions
    async fn cleanup_old_transactions(&mut self) -> Result<()> {
        // Remove transactions that are too old
        let expiry_time = {
            if let Some(ref policy) = *self.policy_config.read().unwrap() {
                policy.mempool_expiry_hours * 3600
            } else {
                // No policy config, skip expiry check
                return Ok(());
            }
        };

        let current_time = Self::current_timestamp();
        // Optimization: Pre-allocate with estimated capacity
        let estimated_removals = self.pool_lock().transactions.len() / 100; // Estimate ~1% will expire
        let mut to_remove = Vec::with_capacity(estimated_removals);

        {
            let timestamps = self.tx_timestamps.read().unwrap();
            for (hash, timestamp) in timestamps.iter() {
                if current_time.saturating_sub(*timestamp) > expiry_time {
                    to_remove.push(*hash);
                }
            }
        }

        for hash in to_remove {
            debug!("Removing expired transaction {}", hex::encode(hash));
            self.remove_transaction(&hash);
        }

        // Check mempool size limits and evict if necessary
        self.enforce_mempool_limits().await?;

        Ok(())
    }

    /// Enforce mempool size limits by evicting transactions if necessary
    async fn enforce_mempool_limits(&mut self) -> Result<()> {
        let policy = match self.policy_config.read().unwrap().as_ref() {
            Some(p) => p.clone(),
            None => return Ok(()), // No policy config, no limits
        };

        // Calculate current mempool size
        let current_size_mb = self.calculate_mempool_size_mb();
        let current_tx_count = self.pool_lock().transactions.len();

        // Check if we need to evict
        let needs_eviction =
            current_size_mb > policy.max_mempool_mb || current_tx_count > policy.max_mempool_txs;

        if !needs_eviction {
            return Ok(());
        }

        // Publish MempoolThresholdExceeded for module event subscribers
        if let Some(ref event_pub) = *self.event_publisher.read().unwrap() {
            let threshold = policy.max_mempool_txs;
            let current = current_tx_count;
            let event_pub_clone = Arc::clone(event_pub);
            tokio::spawn(async move {
                event_pub_clone
                    .publish_mempool_threshold_exceeded(current, threshold)
                    .await;
            });
        }

        debug!(
            "Mempool size limit exceeded: {} MB / {} MB, {} txs / {} txs. Evicting transactions...",
            current_size_mb, policy.max_mempool_mb, current_tx_count, policy.max_mempool_txs
        );

        // Capture min fee rate before eviction (for FeeRateChanged event)
        let old_min_fee_rate = self.get_min_fee_rate_sat_per_vb();

        // Evict transactions based on strategy
        let target_size_mb = policy.max_mempool_mb;
        let target_tx_count = policy.max_mempool_txs;
        let policy = &policy;

        match &policy.eviction_strategy {
            crate::config::EvictionStrategy::LowestFeeRate => {
                self.evict_lowest_fee_rate(target_size_mb, target_tx_count)
                    .await?;
            }
            crate::config::EvictionStrategy::OldestFirst => {
                self.evict_oldest_first(target_size_mb, target_tx_count)
                    .await?;
            }
            crate::config::EvictionStrategy::LargestFirst => {
                self.evict_largest_first(target_size_mb, target_tx_count)
                    .await?;
            }
            crate::config::EvictionStrategy::NoDescendantsFirst => {
                self.evict_no_descendants_first(target_size_mb, target_tx_count)
                    .await?;
            }
            crate::config::EvictionStrategy::Hybrid => {
                self.evict_hybrid(target_size_mb, target_tx_count).await?;
            }
            crate::config::EvictionStrategy::SpamFirst => {
                self.evict_spam_first(target_size_mb, target_tx_count)
                    .await?;
            }
        }

        // Publish FeeRateChanged when min fee rate increased due to eviction
        let new_min_fee_rate = self.get_min_fee_rate_sat_per_vb();
        if new_min_fee_rate != old_min_fee_rate {
            if let Some(ref event_pub) = *self.event_publisher.read().unwrap() {
                let old_f64 = old_min_fee_rate as f64;
                let new_f64 = new_min_fee_rate as f64;
                let mempool_size = self.pool_lock().transactions.len();
                let event_pub_clone = Arc::clone(event_pub);
                tokio::spawn(async move {
                    event_pub_clone
                        .publish_fee_rate_changed(old_f64, new_f64, mempool_size)
                        .await;
                });
            }
        }

        Ok(())
    }

    /// Get the minimum fee rate (sat/vB) in the mempool, or 0 if empty.
    fn get_min_fee_rate_sat_per_vb(&self) -> u64 {
        let fee_index = self.fee_index.read().unwrap();
        fee_index
            .iter()
            .last()
            .map(|(Reverse(r), _)| *r)
            .unwrap_or(0)
    }

    /// Calculate current mempool size in MB
    fn calculate_mempool_size_mb(&self) -> u64 {
        use blvm_protocol::serialization::transaction::serialize_transaction;

        let total_bytes: usize = self
            .pool_lock()
            .transactions
            .values()
            .map(|tx| serialize_transaction(tx).len())
            .sum();

        // Convert to MB (1 MB = 1,048,576 bytes)
        (total_bytes as u64) / 1_048_576
    }

    /// Evict transactions with lowest fee rate
    async fn evict_lowest_fee_rate(
        &mut self,
        target_size_mb: u64,
        target_tx_count: usize,
    ) -> Result<()> {
        use blvm_protocol::serialization::transaction::serialize_transaction;

        // Get all transactions sorted by fee rate (ascending - lowest first)
        let mut tx_fee_rates: Vec<(Hash, u64, usize)> = self
            .pool_lock()
            .transactions
            .iter()
            .map(|(hash, tx)| {
                let fee_rate = self
                    .fee_cache
                    .read()
                    .unwrap()
                    .get(hash)
                    .copied()
                    .unwrap_or(0);
                let size = serialize_transaction(tx).len();
                (*hash, fee_rate, size)
            })
            .collect();

        // Sort by fee rate (ascending) - lowest fee rate first
        tx_fee_rates.sort_by_key(|(_, fee_rate, _)| *fee_rate);

        // Evict until we're under limits
        let mut current_size_mb = self.calculate_mempool_size_mb();
        let mut current_tx_count = self.pool_lock().transactions.len();

        for (hash, _fee_rate, size) in tx_fee_rates {
            if current_size_mb <= target_size_mb && current_tx_count <= target_tx_count {
                break;
            }

            // Don't evict if it has descendants (would orphan them)
            let has_descendants = {
                let descendants = self.tx_descendants.read().unwrap();
                descendants
                    .get(&hash)
                    .map(|d| !d.is_empty())
                    .unwrap_or(false)
            };

            if !has_descendants {
                debug!("Evicting low fee rate transaction {}", hex::encode(hash));
                self.remove_transaction(&hash);
                current_size_mb = current_size_mb.saturating_sub((size as u64) / 1_048_576);
                current_tx_count -= 1;
            }
        }

        Ok(())
    }

    /// Evict oldest transactions first (FIFO)
    async fn evict_oldest_first(
        &mut self,
        target_size_mb: u64,
        target_tx_count: usize,
    ) -> Result<()> {
        use blvm_protocol::serialization::transaction::serialize_transaction;

        // Get all transactions with timestamps, sorted by age (oldest first)
        let mut tx_ages: Vec<(Hash, u64, usize)> = {
            let timestamps = self.tx_timestamps.read().unwrap();
            self.pool_lock()
                .transactions
                .iter()
                .filter_map(|(hash, tx)| {
                    timestamps.get(hash).map(|&timestamp| {
                        let size = serialize_transaction(tx).len();
                        (*hash, timestamp, size)
                    })
                })
                .collect()
        };

        // Sort by timestamp (ascending) - oldest first
        tx_ages.sort_by_key(|(_, timestamp, _)| *timestamp);

        // Evict until we're under limits
        let mut current_size_mb = self.calculate_mempool_size_mb();
        let mut current_tx_count = self.pool_lock().transactions.len();

        for (hash, _timestamp, size) in tx_ages {
            if current_size_mb <= target_size_mb && current_tx_count <= target_tx_count {
                break;
            }

            // Don't evict if it has descendants
            let has_descendants = {
                let descendants = self.tx_descendants.read().unwrap();
                descendants
                    .get(&hash)
                    .map(|d| !d.is_empty())
                    .unwrap_or(false)
            };

            if !has_descendants {
                debug!("Evicting old transaction {}", hex::encode(hash));
                self.remove_transaction(&hash);
                current_size_mb = current_size_mb.saturating_sub((size as u64) / 1_048_576);
                current_tx_count -= 1;
            }
        }

        Ok(())
    }

    /// Evict largest transactions first
    async fn evict_largest_first(
        &mut self,
        target_size_mb: u64,
        target_tx_count: usize,
    ) -> Result<()> {
        use blvm_protocol::serialization::transaction::serialize_transaction;

        // Get all transactions sorted by size (descending - largest first)
        let mut tx_sizes: Vec<(Hash, usize)> = self
            .pool_lock()
            .transactions
            .iter()
            .map(|(hash, tx)| {
                let size = serialize_transaction(tx).len();
                (*hash, size)
            })
            .collect();

        // Sort by size (descending) - largest first
        tx_sizes.sort_by_key(|(_, size)| std::cmp::Reverse(*size));

        // Evict until we're under limits
        let mut current_size_mb = self.calculate_mempool_size_mb();
        let mut current_tx_count = self.pool_lock().transactions.len();

        for (hash, size) in tx_sizes {
            if current_size_mb <= target_size_mb && current_tx_count <= target_tx_count {
                break;
            }

            // Don't evict if it has descendants
            let has_descendants = {
                let descendants = self.tx_descendants.read().unwrap();
                descendants
                    .get(&hash)
                    .map(|d| !d.is_empty())
                    .unwrap_or(false)
            };

            if !has_descendants {
                debug!(
                    "Evicting large transaction {} ({} bytes)",
                    hex::encode(hash),
                    size
                );
                self.remove_transaction(&hash);
                current_size_mb = current_size_mb.saturating_sub((size as u64) / 1_048_576);
                current_tx_count -= 1;
            }
        }

        Ok(())
    }

    /// Evict transactions with no descendants first (safest)
    async fn evict_no_descendants_first(
        &mut self,
        target_size_mb: u64,
        target_tx_count: usize,
    ) -> Result<()> {
        use blvm_protocol::serialization::transaction::serialize_transaction;

        // Get all transactions with no descendants, sorted by fee rate (lowest first)
        let mut tx_no_descendants: Vec<(Hash, u64, usize)> = {
            let descendants = self.tx_descendants.read().unwrap();
            let fee_cache = self.fee_cache.read().unwrap();

            self.pool_lock()
                .transactions
                .iter()
                .filter_map(|(hash, tx)| {
                    let has_descendants = descendants
                        .get(hash)
                        .map(|d| !d.is_empty())
                        .unwrap_or(false);

                    if !has_descendants {
                        let fee_rate = fee_cache.get(hash).copied().unwrap_or(0);
                        let size = serialize_transaction(tx).len();
                        Some((*hash, fee_rate, size))
                    } else {
                        None
                    }
                })
                .collect()
        };

        // Sort by fee rate (ascending) - lowest fee rate first
        tx_no_descendants.sort_by_key(|(_, fee_rate, _)| *fee_rate);

        // Evict until we're under limits
        let mut current_size_mb = self.calculate_mempool_size_mb();
        let mut current_tx_count = self.pool_lock().transactions.len();

        for (hash, _fee_rate, size) in tx_no_descendants {
            if current_size_mb <= target_size_mb && current_tx_count <= target_tx_count {
                break;
            }

            debug!(
                "Evicting transaction with no descendants {}",
                hex::encode(hash)
            );
            self.remove_transaction(&hash);
            current_size_mb = current_size_mb.saturating_sub((size as u64) / 1_048_576);
            current_tx_count -= 1;
        }

        Ok(())
    }

    /// Hybrid eviction: combine fee rate and age
    async fn evict_hybrid(&mut self, target_size_mb: u64, target_tx_count: usize) -> Result<()> {
        use blvm_protocol::serialization::transaction::serialize_transaction;

        // Calculate score: lower fee rate + older age = higher eviction priority
        let current_time = Self::current_timestamp();
        let mut tx_scores: Vec<(Hash, u64, usize)> = {
            let timestamps = self.tx_timestamps.read().unwrap();
            let fee_cache = self.fee_cache.read().unwrap();

            self.pool_lock()
                .transactions
                .iter()
                .map(|(hash, tx)| {
                    let fee_rate = fee_cache.get(hash).copied().unwrap_or(0);
                    let age = timestamps
                        .get(hash)
                        .map(|&t| current_time.saturating_sub(t))
                        .unwrap_or(0);

                    // Score: normalize fee rate (lower = higher score) + age weight
                    // Use inverse fee rate (higher for lower fees) + age in seconds
                    // Normalize fee rate: use 1 / (fee_rate + 1) to avoid division by zero
                    let fee_score = if fee_rate > 0 {
                        1_000_000 / (fee_rate + 1) // Higher score for lower fee
                    } else {
                        1_000_000 // Max score for zero fee
                    };

                    // Age weight: 1 point per hour old
                    let age_score = age / 3600;

                    // Combined score (higher = evict first)
                    let score = fee_score + age_score;

                    let size = serialize_transaction(tx).len();
                    (*hash, score, size)
                })
                .collect()
        };

        // Sort by score (descending) - highest score (most evictable) first
        tx_scores.sort_by_key(|(_, score, _)| std::cmp::Reverse(*score));

        // Evict until we're under limits
        let mut current_size_mb = self.calculate_mempool_size_mb();
        let mut current_tx_count = self.pool_lock().transactions.len();

        for (hash, _score, size) in tx_scores {
            if current_size_mb <= target_size_mb && current_tx_count <= target_tx_count {
                break;
            }

            // Don't evict if it has descendants
            let has_descendants = {
                let descendants = self.tx_descendants.read().unwrap();
                descendants
                    .get(&hash)
                    .map(|d| !d.is_empty())
                    .unwrap_or(false)
            };

            if !has_descendants {
                debug!(
                    "Evicting transaction (hybrid strategy) {}",
                    hex::encode(hash)
                );
                self.remove_transaction(&hash);
                current_size_mb = current_size_mb.saturating_sub((size as u64) / 1_048_576);
                current_tx_count -= 1;
            }
        }

        Ok(())
    }

    /// Evict spam transactions first (when mempool is full)
    async fn evict_spam_first(
        &mut self,
        target_size_mb: u64,
        target_tx_count: usize,
    ) -> Result<()> {
        use blvm_protocol::serialization::transaction::serialize_transaction;
        use blvm_protocol::spam_filter::SpamFilter;

        // Get all transactions, classify as spam or not
        let spam_filter = SpamFilter::new();
        let mut spam_txs: Vec<(Hash, u64, usize)> = Vec::new();
        let mut non_spam_txs: Vec<(Hash, u64, usize)> = Vec::new();

        let entries: Vec<(Hash, Transaction)> = {
            let pool = self.pool_lock();
            pool.transactions
                .iter()
                .map(|(h, t)| (*h, t.clone()))
                .collect()
        };
        let fee_cache = self.fee_cache.read().unwrap();
        for (hash, tx) in &entries {
            let size = serialize_transaction(tx).len();
            let fee_rate = fee_cache.get(hash).copied().unwrap_or(0);

            let result = spam_filter.is_spam(tx);
            if result.is_spam {
                spam_txs.push((*hash, fee_rate, size));
            } else {
                non_spam_txs.push((*hash, fee_rate, size));
            }
        }
        drop(fee_cache);

        // Sort spam transactions by fee rate (lowest first - evict first)
        spam_txs.sort_by_key(|(_, fee_rate, _)| *fee_rate);

        // Sort non-spam transactions by fee rate (lowest first - evict last)
        non_spam_txs.sort_by_key(|(_, fee_rate, _)| *fee_rate);

        // Evict spam transactions first, then non-spam if needed
        let mut current_size_mb = self.calculate_mempool_size_mb();
        let mut current_tx_count = self.pool_lock().transactions.len();

        // First, evict spam transactions
        for (hash, _fee_rate, size) in spam_txs {
            if current_size_mb <= target_size_mb && current_tx_count <= target_tx_count {
                break;
            }

            // Don't evict if it has descendants
            let has_descendants = {
                let descendants = self.tx_descendants.read().unwrap();
                descendants
                    .get(&hash)
                    .map(|d| !d.is_empty())
                    .unwrap_or(false)
            };

            if !has_descendants {
                debug!("Evicting spam transaction {}", hex::encode(hash));
                self.remove_transaction(&hash);
                current_size_mb = current_size_mb.saturating_sub((size as u64) / 1_048_576);
                current_tx_count -= 1;
            }
        }

        // If still over limits, evict non-spam transactions (lowest fee rate first)
        for (hash, _fee_rate, size) in non_spam_txs {
            if current_size_mb <= target_size_mb && current_tx_count <= target_tx_count {
                break;
            }

            // Don't evict if it has descendants
            let has_descendants = {
                let descendants = self.tx_descendants.read().unwrap();
                descendants
                    .get(&hash)
                    .map(|d| !d.is_empty())
                    .unwrap_or(false)
            };

            if !has_descendants {
                debug!("Evicting non-spam transaction {}", hex::encode(hash));
                self.remove_transaction(&hash);
                current_size_mb = current_size_mb.saturating_sub((size as u64) / 1_048_576);
                current_tx_count -= 1;
            }
        }

        Ok(())
    }

    /// Check if a transaction can replace an existing one (RBF)
    ///
    /// This wraps the consensus layer replacement_checks with RBF mode-specific logic
    ///
    /// `storage` is optional - if provided, can be used for conservative mode confirmation checks
    pub fn check_rbf_replacement(
        &self,
        new_tx: &Transaction,
        existing_tx: &Transaction,
        utxo_set: &UtxoSet,
        storage: Option<&crate::storage::Storage>,
    ) -> Result<bool> {
        use blvm_protocol::block::calculate_tx_id;

        let rbf_config = match self.rbf_config.read().unwrap().as_ref() {
            Some(config) => config.clone(),
            None => {
                // No RBF config - use default BIP125 behavior
                return replacement_checks(
                    new_tx,
                    existing_tx,
                    utxo_set,
                    &self.mempool.read().unwrap(),
                )
                .map_err(|e| anyhow::anyhow!("RBF check failed: {}", e));
            }
        };

        // Check if RBF is disabled
        if matches!(rbf_config.mode, crate::config::RbfMode::Disabled) {
            return Ok(false);
        }

        // Use the cloned config for the rest of the function
        let rbf_config = &rbf_config;

        // Check if existing transaction signals RBF
        if !signals_rbf(existing_tx) {
            return Ok(false);
        }

        let existing_tx_hash = calculate_tx_id(existing_tx);
        let _new_tx_hash = calculate_tx_id(new_tx);

        // Check replacement count limit
        if let Some(tracking) = self.rbf_tracking.read().unwrap().get(&existing_tx_hash) {
            if tracking.replacement_count >= rbf_config.max_replacements_per_tx {
                warn!(
                    "RBF replacement rejected: max replacements ({}) exceeded for tx {}",
                    rbf_config.max_replacements_per_tx,
                    hex::encode(existing_tx_hash)
                );
                return Ok(false);
            }

            // Check cooldown period
            let current_time = Self::current_timestamp();
            let time_since_last = current_time.saturating_sub(tracking.last_replacement_time);
            if time_since_last < rbf_config.cooldown_seconds {
                warn!(
                    "RBF replacement rejected: cooldown period not met ({}s remaining) for tx {}",
                    rbf_config.cooldown_seconds - time_since_last,
                    hex::encode(existing_tx_hash)
                );
                return Ok(false);
            }
        }

        // Calculate fees and fee rates using the same method as MempoolManager
        let new_fee = self.calculate_transaction_fee(new_tx, utxo_set) as i64;
        let existing_fee = self.calculate_transaction_fee(existing_tx, utxo_set) as i64;

        // Calculate transaction sizes (simplified - use serialization length)
        use blvm_protocol::serialization::transaction::serialize_transaction;
        let new_tx_size = serialize_transaction(new_tx).len();
        let existing_tx_size = serialize_transaction(existing_tx).len();

        if new_tx_size == 0 || existing_tx_size == 0 {
            return Ok(false);
        }

        // Check fee rate multiplier (mode-specific)
        let new_fee_scaled = (new_fee as u128)
            .checked_mul(existing_tx_size as u128)
            .ok_or_else(|| anyhow::anyhow!("Fee rate calculation overflow"))?;
        let existing_fee_scaled = (existing_fee as u128)
            .checked_mul(new_tx_size as u128)
            .ok_or_else(|| anyhow::anyhow!("Fee rate calculation overflow"))?;

        // Apply mode-specific multiplier
        let required_fee_scaled =
            (existing_fee_scaled as f64 * rbf_config.min_fee_rate_multiplier) as u128;
        if new_fee_scaled <= required_fee_scaled {
            warn!(
                "RBF replacement rejected: fee rate increase insufficient (required: {:.2}x, got: {:.2}x) for tx {}",
                rbf_config.min_fee_rate_multiplier,
                (new_fee_scaled as f64) / (existing_fee_scaled as f64),
                hex::encode(existing_tx_hash)
            );
            return Ok(false);
        }

        // Check absolute fee bump
        let min_fee_bump = rbf_config.min_fee_bump_satoshis as i64;
        if new_fee <= existing_fee + min_fee_bump {
            warn!(
                "RBF replacement rejected: absolute fee bump insufficient (required: {} sat, got: {} sat) for tx {}",
                min_fee_bump,
                new_fee - existing_fee,
                hex::encode(existing_tx_hash)
            );
            return Ok(false);
        }

        // Conservative mode: Check minimum confirmations
        // Note: Transactions in mempool have 0 confirmations. This check ensures that
        // if a transaction has been confirmed (which shouldn't be in mempool), we require
        // it to have minimum confirmations before allowing replacement.
        // In practice, mempool transactions will always have 0 confirmations, so this
        // check mainly serves as a safety mechanism.
        if matches!(rbf_config.mode, crate::config::RbfMode::Conservative)
            && rbf_config.min_confirmations > 0
        {
            if let Some(storage) = storage {
                // Check if transaction is in blockchain and has enough confirmations
                if let Ok(Some(metadata)) = storage.transactions().get_metadata(&existing_tx_hash) {
                    // Transaction is in a block - check confirmations
                    let block_hash = metadata.block_hash;
                    if let Ok(Some(block_height)) = storage.blocks().get_height_by_hash(&block_hash)
                    {
                        if let Ok(Some(tip_height)) = storage.chain().get_height() {
                            let confirmations = tip_height.saturating_sub(block_height) + 1;
                            if confirmations < rbf_config.min_confirmations as u64 {
                                warn!(
                                    "RBF replacement rejected: conservative mode requires {} confirmations, tx {} has {}",
                                    rbf_config.min_confirmations,
                                    hex::encode(existing_tx_hash),
                                    confirmations
                                );
                                return Ok(false);
                            }
                        }
                    }
                }
                // If transaction is not in blockchain (mempool only), confirmations = 0
                // For conservative mode, we might want to reject replacements of unconfirmed transactions
                // if min_confirmations > 0, but that would prevent all mempool RBF replacements.
                // So we allow it - the transaction is still in mempool and can be replaced.
            }
        }

        // Check conflict (must spend at least one input from existing tx)
        if !has_conflict_with_tx(new_tx, existing_tx) {
            return Ok(false);
        }

        // Aggressive mode: Check for package replacement support
        // Package replacement = replacing parent + child transactions together
        if matches!(rbf_config.mode, crate::config::RbfMode::Aggressive)
            && rbf_config.allow_package_replacements
        {
            // Check if new_tx has dependencies (child transactions) that should be replaced together
            // For now, we allow the replacement if the new transaction has higher fees
            // Full package replacement logic would require tracking transaction packages
            // This is a simplified implementation
            debug!(
                "Aggressive mode: allowing package replacement for tx {}",
                hex::encode(existing_tx_hash)
            );
        }

        // For the remaining BIP125 checks (new dependencies), use the consensus replacement_checks
        // but we've already applied our mode-specific fee requirements above
        // Note: replacement_checks will re-check fee rate, but we've already validated with our multiplier
        // So we call it to verify the other BIP125 rules (dependencies, etc.)
        // However, since we've already done stricter checks, if replacement_checks passes, we're good
        let bip125_result =
            replacement_checks(new_tx, existing_tx, utxo_set, &self.mempool.read().unwrap())?;
        if !bip125_result {
            // BIP125 check failed (likely new dependencies issue)
            return Ok(false);
        }

        // All checks passed
        Ok(true)
    }

    /// Check ancestor/descendant limits for a transaction
    fn check_ancestor_descendant_limits(
        &self,
        tx: &Transaction,
        tx_hash: &Hash,
        policy: &MempoolPolicyConfig,
    ) -> Result<bool> {
        use blvm_protocol::serialization::transaction::serialize_transaction;

        // Calculate transaction size
        let tx_size = serialize_transaction(tx).len() as u64;

        // Find all ancestors (transactions this tx depends on)
        // Optimization: Pre-allocate with estimated capacity (most txs have < 10 ancestors)
        let mut ancestors = HashSet::with_capacity(10);
        let mut to_process = Vec::with_capacity(10);
        to_process.push(*tx_hash);
        let mut processed = HashSet::with_capacity(10);

        while let Some(current_hash) = to_process.pop() {
            if processed.contains(&current_hash) {
                continue;
            }
            processed.insert(current_hash);

            // Single pool critical section — nested pool_lock() would deadlock (non-reentrant Mutex).
            {
                let pool = self.pool_lock();
                if let Some(current_tx) = pool.transactions.get(&current_hash) {
                    let parent_keys: Vec<Hash> = pool.transactions.keys().copied().collect();
                    for input in &current_tx.inputs {
                        for parent_hash in &parent_keys {
                            if parent_hash == &input.prevout.hash {
                                if !ancestors.contains(parent_hash) {
                                    ancestors.insert(*parent_hash);
                                    to_process.push(*parent_hash);
                                }
                                break;
                            }
                        }
                    }
                }
            }
        }

        // Calculate ancestor count and size
        let ancestor_count = ancestors.len() as u32;
        let ancestor_size: u64 = {
            let pool = self.pool_lock();
            ancestors
                .iter()
                .filter_map(|h| pool.transactions.get(h))
                .map(|t| serialize_transaction(t).len() as u64)
                .sum()
        };

        // Check ancestor limits
        if ancestor_count + 1 > policy.max_ancestor_count {
            warn!(
                "Transaction {} exceeds max ancestor count: {} > {}",
                hex::encode(tx_hash),
                ancestor_count + 1,
                policy.max_ancestor_count
            );
            return Ok(false);
        }

        if ancestor_size + tx_size > policy.max_ancestor_size {
            warn!(
                "Transaction {} exceeds max ancestor size: {} > {}",
                hex::encode(tx_hash),
                ancestor_size + tx_size,
                policy.max_ancestor_size
            );
            return Ok(false);
        }

        // Find all descendants (transactions that depend on this tx)
        // Optimization: Pre-allocate with estimated capacity (most txs have < 10 descendants)
        let mut descendants = HashSet::with_capacity(10);
        let mut to_process = Vec::with_capacity(10);
        to_process.push(*tx_hash);
        let mut processed = HashSet::with_capacity(10);

        while let Some(current_hash) = to_process.pop() {
            if processed.contains(&current_hash) {
                continue;
            }
            processed.insert(current_hash);

            {
                let pool = self.pool_lock();
                if let Some(current_tx) = pool.transactions.get(&current_hash) {
                    let output_outpoints: Vec<_> = (0..current_tx.outputs.len())
                        .map(|idx| OutPoint {
                            hash: current_hash,
                            index: idx as u32,
                        })
                        .collect();

                    for (child_hash, child_tx) in &pool.transactions {
                        for input in &child_tx.inputs {
                            if output_outpoints.contains(&input.prevout) {
                                if !descendants.contains(child_hash) {
                                    descendants.insert(*child_hash);
                                    to_process.push(*child_hash);
                                }
                                break;
                            }
                        }
                    }
                }
            }
        }

        // Calculate descendant count and size
        let descendant_count = descendants.len() as u32;
        let descendant_size: u64 = {
            let pool = self.pool_lock();
            descendants
                .iter()
                .filter_map(|h| pool.transactions.get(h))
                .map(|t| serialize_transaction(t).len() as u64)
                .sum()
        };

        // Check descendant limits
        if descendant_count + 1 > policy.max_descendant_count {
            warn!(
                "Transaction {} exceeds max descendant count: {} > {}",
                hex::encode(tx_hash),
                descendant_count + 1,
                policy.max_descendant_count
            );
            return Ok(false);
        }

        if descendant_size + tx_size > policy.max_descendant_size {
            warn!(
                "Transaction {} exceeds max descendant size: {} > {}",
                hex::encode(tx_hash),
                descendant_size + tx_size,
                policy.max_descendant_size
            );
            return Ok(false);
        }

        Ok(true)
    }

    /// Update dependency graph when a transaction is added
    fn update_dependency_graph(&self, tx: &Transaction, tx_hash: &Hash) {
        let mut dependencies = self.tx_dependencies.write().unwrap();
        let mut descendants = self.tx_descendants.write().unwrap();

        // Initialize empty sets for this transaction
        dependencies.entry(*tx_hash).or_default();
        descendants.entry(*tx_hash).or_default();

        // Find parent transactions (ancestors) - transactions that created inputs
        for input in &tx.inputs {
            // Find transaction that created this output
            for parent_hash in self.pool_lock().transactions.keys() {
                if parent_hash == &input.prevout.hash {
                    // This transaction depends on parent
                    dependencies
                        .entry(*tx_hash)
                        .or_default()
                        .insert(*parent_hash);

                    // Parent has this as a descendant
                    descendants
                        .entry(*parent_hash)
                        .or_default()
                        .insert(*tx_hash);

                    break;
                }
            }
        }
    }

    /// Add transaction to mempool
    /// Uses interior mutability so it can be called with Arc<MempoolManager> (re-broadcast, sendrawtransaction)
    pub fn add_transaction(&self, tx: Transaction) -> Result<bool> {
        debug!("Adding transaction to mempool");

        use blvm_protocol::block::calculate_tx_id;
        let tx_hash = calculate_tx_id(&tx);

        if let Some(ref policy) = *self.policy_config.read().unwrap() {
            if policy.reject_spam_in_mempool {
                use blvm_protocol::spam_filter::SpamFilter;
                let filter = policy
                    .spam_filter
                    .as_ref()
                    .map(|cfg| SpamFilter::with_config(cfg.clone().into()))
                    .unwrap_or_else(SpamFilter::new);
                if filter.is_spam(&tx).is_spam {
                    warn!(
                        "Transaction {} rejected: classified as spam (reject_spam_in_mempool)",
                        hex::encode(tx_hash)
                    );
                    return Ok(false);
                }
            }
        }

        // Check for conflicts with existing mempool transactions
        // If conflict exists, check if RBF replacement is allowed
        let mut conflicting_tx_hash: Option<Hash> = None;
        for input in &tx.inputs {
            if let Some(existing_tx) = self
                .pool_lock()
                .transactions
                .values()
                .find(|t| t.inputs.iter().any(|i| i.prevout == input.prevout))
            {
                let existing_hash = calculate_tx_id(existing_tx);
                conflicting_tx_hash = Some(existing_hash);
                break;
            }
        }

        // If there's a conflict, try RBF replacement
        if let Some(existing_hash) = conflicting_tx_hash {
            let existing_clone = {
                let pool = self.pool_lock();
                pool.transactions.get(&existing_hash).cloned()
            };
            if let Some(ref existing_tx) = existing_clone {
                // Check if RBF replacement is allowed
                // Note: Storage is not available in MempoolManager context
                // Conservative mode confirmation checks will be skipped if storage is None
                if self.check_rbf_replacement(&tx, existing_tx, &self.utxo_set, None)? {
                    debug!(
                        "RBF replacement allowed, replacing transaction {}",
                        hex::encode(existing_hash)
                    );

                    // Remove existing transaction (must not hold pool lock — remove_transaction locks pool)
                    self.remove_transaction(&existing_hash);

                    // Update RBF tracking
                    let original_hash = {
                        let tracking = self.rbf_tracking.read().unwrap();
                        tracking
                            .get(&existing_hash)
                            .map(|t| t.original_tx_hash)
                            .unwrap_or(existing_hash)
                    };

                    let replacement_count = {
                        let tracking = self.rbf_tracking.read().unwrap();
                        tracking
                            .get(&existing_hash)
                            .map(|t| t.replacement_count + 1)
                            .unwrap_or(1)
                    };

                    {
                        let mut tracking = self.rbf_tracking.write().unwrap();
                        tracking.insert(
                            tx_hash,
                            RbfTracking {
                                replacement_count,
                                last_replacement_time: Self::current_timestamp(),
                                original_tx_hash: original_hash,
                            },
                        );
                        tracking.remove(&existing_hash);
                    }
                } else {
                    debug!("Transaction conflicts with existing mempool transaction and RBF replacement not allowed");
                    return Ok(false);
                }
            }
        } else {
            // No conflict - check if inputs are already spent
            for input in &tx.inputs {
                if self.pool_lock().spent_outputs.contains(&input.prevout) {
                    debug!("Transaction conflicts with existing mempool transaction");
                    return Ok(false);
                }
            }
        }

        // Check ancestor/descendant limits before adding
        if let Some(ref policy) = *self.policy_config.read().unwrap() {
            if !self.check_ancestor_descendant_limits(&tx, &tx_hash, policy)? {
                warn!(
                    "Transaction {} rejected: exceeds ancestor/descendant limits",
                    hex::encode(tx_hash)
                );
                return Ok(false);
            }
        }

        // Add transaction to mempool (store full transaction)
        self.pool_lock().transactions.insert(tx_hash, tx.clone());
        self.mempool.write().unwrap().insert(tx_hash);

        // Track spent outputs
        for input in &tx.inputs {
            self.pool_lock().spent_outputs.insert(input.prevout);
        }

        // Update dependency graph
        self.update_dependency_graph(&tx, &tx_hash);

        // Record timestamp
        self.tx_timestamps
            .write()
            .unwrap()
            .insert(tx_hash, Self::current_timestamp());

        // Calculate and cache fee rate (will be updated when UTXO set is available)
        // For now, set to 0 - will be recalculated in get_prioritized_transactions
        let fee_rate = 0u64;
        self.fee_cache.write().unwrap().insert(tx_hash, fee_rate);
        self.fee_index
            .write()
            .unwrap()
            .entry(Reverse(fee_rate))
            .or_default()
            .push(tx_hash);

        // Publish mempool transaction added event
        if let Some(ref event_pub) = *self.event_publisher.read().unwrap() {
            let mempool_size = self.pool_lock().transactions.len();
            // Convert fee_rate from u64 to f64 (satoshis per vbyte)
            // Note: fee_rate is currently 0, will be updated later when UTXO set is available
            let fee_rate_f64 = fee_rate as f64;
            let tx_hash_clone = tx_hash;
            let event_pub_clone = Arc::clone(event_pub);
            tokio::spawn(async move {
                event_pub_clone
                    .publish_mempool_transaction_added(&tx_hash_clone, fee_rate_f64, mempool_size)
                    .await;
            });
            // NewTransaction (use publish_event to avoid ZMQ Send issues in spawn)
            let tx_hash_ev = tx_hash;
            let ep_clone = Arc::clone(event_pub);
            tokio::spawn(async move {
                let _ = ep_clone
                    .publish_event(
                        crate::module::traits::EventType::NewTransaction,
                        crate::module::ipc::protocol::EventPayload::NewTransaction {
                            tx_hash: tx_hash_ev,
                        },
                    )
                    .await;
            });
        }

        Ok(true)
    }

    /// Get mempool size
    pub fn size(&self) -> usize {
        self.pool_lock().transactions.len()
    }

    /// Get mempool transaction hashes
    pub fn transaction_hashes(&self) -> Vec<Hash> {
        self.pool_lock().transactions.keys().cloned().collect()
    }

    /// Get transaction by hash
    pub fn get_transaction(&self, hash: &Hash) -> Option<Transaction> {
        self.pool_lock().transactions.get(hash).cloned()
    }

    /// Get all transactions
    pub fn get_transactions(&self) -> Vec<Transaction> {
        self.pool_lock().transactions.values().cloned().collect()
    }

    /// Get prioritized transactions by fee rate
    ///
    /// Returns transactions sorted by fee rate (satoshis per vbyte) in descending order.
    /// Requires UTXO set to calculate fee rates.
    ///
    /// Optimization: Uses sorted index (BTreeMap) for O(log n) insertion, O(1) top-N retrieval
    /// instead of O(n log n) sort on every call.
    pub fn get_prioritized_transactions(
        &self,
        limit: usize,
        utxo_set: &UtxoSet,
    ) -> Vec<Transaction> {
        // Recalculate fee rates and update index
        // Note: In a production system, we'd track UTXO set changes and only recalculate when needed
        self.update_fee_index(utxo_set);

        // Collect hashes under fee_index read only; do not call pool_lock while holding fee_index
        // (update_fee_index may interleave with remove_transaction — opposite lock order would deadlock).
        let mut ordered_hashes: Vec<Hash> = Vec::with_capacity(limit);
        {
            let fee_index = self.fee_index.read().unwrap();
            for (Reverse(_fee_rate), tx_hashes) in fee_index.iter() {
                for tx_hash in tx_hashes {
                    ordered_hashes.push(*tx_hash);
                    if ordered_hashes.len() >= limit {
                        break;
                    }
                }
                if ordered_hashes.len() >= limit {
                    break;
                }
            }
        }

        let mut result = Vec::with_capacity(limit.min(ordered_hashes.len()));
        let pool = self.pool_lock();
        for h in ordered_hashes {
            if let Some(tx) = pool.transactions.get(&h) {
                result.push(tx.clone());
                if result.len() >= limit {
                    break;
                }
            }
        }
        result
    }

    /// Calculate a simple hash of the UTXO set for change detection
    ///
    /// Uses a fast hash of UTXO set size and a sample of keys to detect changes.
    /// This is a heuristic - not perfect but fast enough for optimization purposes.
    fn calculate_utxo_set_hash(utxo_set: &UtxoSet) -> u64 {
        use std::collections::hash_map::DefaultHasher;
        use std::hash::{Hash, Hasher};

        let mut hasher = DefaultHasher::new();
        utxo_set.len().hash(&mut hasher);

        // Sample first 10 keys for change detection (fast heuristic)
        let sample_size = utxo_set.len().min(10);
        for (i, (outpoint, utxo)) in utxo_set.iter().enumerate() {
            if i >= sample_size {
                break;
            }
            outpoint.hash(&mut hasher);
            utxo.value.hash(&mut hasher);
        }

        hasher.finish()
    }

    /// Update fee index with current UTXO set
    ///
    /// Recalculates fee rates for all transactions and rebuilds the sorted index.
    ///
    /// Optimization: Only recalculates when UTXO set changes (incremental updates)
    /// Optimization: Batch UTXO lookups across all transactions for better cache locality
    fn update_fee_index(&self, utxo_set: &UtxoSet) {
        // Calculate current UTXO set hash
        let current_hash = Self::calculate_utxo_set_hash(utxo_set);

        // Check if UTXO set changed
        let mut last_hash = self.utxo_set_hash.write().unwrap();
        if Some(current_hash) == *last_hash {
            // UTXO set unchanged - skip recalculation
            drop(last_hash);
            return;
        }

        // UTXO set changed - update hash and recalculate
        *last_hash = Some(current_hash);
        drop(last_hash);

        // Clear existing index (we'll rebuild it)
        let mut fee_index = self.fee_index.write().unwrap();
        fee_index.clear();
        drop(fee_index);

        {
            let mut fee_cache = self.fee_cache.write().unwrap();
            fee_cache.clear();
        }

        // Snapshot under pool lock only — never hold fee_cache/fee_index writes while locking pool
        // (avoids deadlock with remove_transaction: pool → fee_cache).
        let txs_snapshot: Vec<(Hash, Transaction)> = {
            let pool = self.pool_lock();
            pool.transactions
                .iter()
                .map(|(h, t)| (*h, t.clone()))
                .collect()
        };

        let all_prevouts: Vec<(Hash, OutPoint)> = txs_snapshot
            .iter()
            .flat_map(|(tx_hash, tx)| tx.inputs.iter().map(move |input| (*tx_hash, input.prevout)))
            .collect();

        let mut utxo_cache: HashMap<&OutPoint, u64> = HashMap::with_capacity(all_prevouts.len());
        for (_, prevout) in &all_prevouts {
            if let Some(utxo) = utxo_set.get(prevout) {
                utxo_cache.insert(prevout, utxo.value as u64);
            }
        }

        for (tx_hash, tx) in &txs_snapshot {
            let mut input_total = 0u64;
            for input in &tx.inputs {
                if let Some(&value) = utxo_cache.get(&input.prevout) {
                    input_total += value;
                }
            }

            let output_total: u64 = tx.outputs.iter().map(|out| out.value as u64).sum();
            let fee = input_total.saturating_sub(output_total);
            let size = self.estimate_transaction_size(tx);
            let fee_rate = if size > 0 {
                fee * 1000 / size as u64
            } else {
                0
            };

            {
                let mut fee_cache = self.fee_cache.write().unwrap();
                fee_cache.insert(*tx_hash, fee_rate);
            }
            let mut fee_index = self.fee_index.write().unwrap();
            fee_index
                .entry(Reverse(fee_rate))
                .or_default()
                .push(*tx_hash);
        }
    }

    /// Calculate transaction fee
    ///
    /// Fee = sum of inputs - sum of outputs
    ///
    /// Optimization: Uses batch UTXO lookup pattern for better cache locality
    pub fn calculate_transaction_fee(&self, tx: &Transaction, utxo_set: &UtxoSet) -> u64 {
        // Optimization: Batch UTXO lookups - collect all prevouts first, then lookup
        // This improves cache locality and reduces HashMap traversal overhead
        let prevouts: Vec<&OutPoint> = tx.inputs.iter().map(|input| &input.prevout).collect();

        // Batch UTXO lookup (single pass through HashMap)
        let mut input_total = 0u64;
        for prevout in prevouts {
            if let Some(utxo) = utxo_set.get(prevout) {
                input_total += utxo.value as u64;
            }
        }

        // Sum output values
        let output_total: u64 = tx.outputs.iter().map(|out| out.value as u64).sum();

        // Fee is difference (inputs - outputs)
        input_total.saturating_sub(output_total)
    }

    /// Estimate transaction size in vbytes
    ///
    /// Simplified estimation - in production, would use actual serialized size
    pub fn estimate_transaction_size(&self, tx: &Transaction) -> usize {
        // Base transaction size: version (4) + locktime (4) = 8 bytes
        let mut size = 8;

        // Input size: prevout (36) + script_sig (var) + sequence (4)
        for input in &tx.inputs {
            size += 36; // prevout
            size += input.script_sig.len();
            size += 4; // sequence
        }

        // Output size: value (8) + script_pubkey (var)
        for output in &tx.outputs {
            size += 8; // value
            size += output.script_pubkey.len();
        }

        // Add witness discount if segwit (simplified - assume no witness for now)
        size
    }

    /// Remove transaction from mempool
    pub fn remove_transaction(&self, hash: &Hash) -> bool {
        let tx = {
            let mut pool = self.pool_lock();
            let Some(tx) = pool.transactions.remove(hash) else {
                return false;
            };
            for input in &tx.inputs {
                pool.spent_outputs.remove(&input.prevout);
            }
            tx
        };

        self.mempool.write().unwrap().remove(hash);

        // Remove from fee index
        if let Some(fee_rate) = self.fee_cache.write().unwrap().remove(hash) {
            let mut fee_index = self.fee_index.write().unwrap();
            if let Some(tx_hashes) = fee_index.get_mut(&Reverse(fee_rate)) {
                tx_hashes.retain(|&h| h != *hash);
                if tx_hashes.is_empty() {
                    fee_index.remove(&Reverse(fee_rate));
                }
            }
        }

        // Remove RBF tracking
        self.rbf_tracking.write().unwrap().remove(hash);

        // Remove timestamp
        self.tx_timestamps.write().unwrap().remove(hash);

        // Remove from dependency graph
        {
            let mut dependencies = self.tx_dependencies.write().unwrap();
            let mut descendants = self.tx_descendants.write().unwrap();

            if let Some(children) = descendants.remove(hash) {
                for child_hash in children {
                    if let Some(parents) = dependencies.get_mut(&child_hash) {
                        parents.remove(hash);
                    }
                }
            }

            if let Some(parents) = dependencies.remove(hash) {
                for parent_hash in parents {
                    if let Some(children) = descendants.get_mut(&parent_hash) {
                        children.remove(hash);
                    }
                }
            }
        }

        if let Some(ref event_pub) = *self.event_publisher.read().unwrap() {
            let mempool_size = self.pool_lock().transactions.len();
            let hash_clone = *hash;
            let reason = "removed".to_string();
            let event_pub_clone = Arc::clone(event_pub);
            tokio::spawn(async move {
                event_pub_clone
                    .publish_mempool_transaction_removed(&hash_clone, reason, mempool_size)
                    .await;
            });
        }

        true
    }

    /// Clear mempool
    pub fn clear(&self) {
        let (cleared_count,) = {
            let mut pool = self.pool.lock().unwrap();
            let n = pool.transactions.len();
            pool.transactions.clear();
            pool.spent_outputs.clear();
            (n,)
        };
        self.mempool.write().unwrap().clear();
        self.fee_index.write().unwrap().clear();
        self.fee_cache.write().unwrap().clear();
        self.rbf_tracking.write().unwrap().clear();
        self.tx_timestamps.write().unwrap().clear();

        // Publish mempool cleared event
        if let Some(ref event_pub) = *self.event_publisher.read().unwrap() {
            let event_pub_clone = Arc::clone(event_pub);
            let cleared_count_clone = cleared_count;
            tokio::spawn(async move {
                event_pub_clone
                    .publish_mempool_cleared(cleared_count_clone)
                    .await;
            });
        }
    }

    /// Save mempool to disk for persistence
    pub fn save_to_disk<P: AsRef<std::path::Path>>(&self, path: P) -> Result<()> {
        use blvm_protocol::serialization::transaction::serialize_transaction;
        use std::fs::File;
        use std::io::Write;

        let transactions = self.get_transactions();
        let mut file = File::create(path)?;

        // Write transaction count
        file.write_all(&(transactions.len() as u32).to_le_bytes())?;

        // Write each transaction
        for tx in transactions {
            let serialized = serialize_transaction(&tx);
            file.write_all(&(serialized.len() as u32).to_le_bytes())?;
            file.write_all(&serialized)?;
        }

        file.sync_all()?;
        Ok(())
    }
}

impl Default for MempoolManager {
    fn default() -> Self {
        Self::new()
    }
}

// Implement MempoolProvider trait for integration with MiningCoordinator
// Safety: MempoolManager is thread-safe (uses Arc internally)
unsafe impl Sync for MempoolManager {}
unsafe impl Send for MempoolManager {}

impl crate::node::miner::MempoolProvider for MempoolManager {
    fn get_transactions(&self) -> Vec<blvm_protocol::Transaction> {
        self.get_transactions()
    }

    fn get_transaction(&self, hash: &[u8; 32]) -> Option<blvm_protocol::Transaction> {
        use blvm_protocol::Hash;
        let hash_array: Hash = *hash;
        self.get_transaction(&hash_array)
    }

    fn get_mempool_size(&self) -> usize {
        self.size()
    }

    fn get_prioritized_transactions(
        &self,
        limit: usize,
        utxo_set: &blvm_protocol::UtxoSet,
    ) -> Vec<blvm_protocol::Transaction> {
        self.get_prioritized_transactions(limit, utxo_set)
    }

    fn remove_transaction(&mut self, hash: &[u8; 32]) -> bool {
        use blvm_protocol::Hash;
        let hash_array: Hash = *hash;
        MempoolManager::remove_transaction(self, &hash_array)
    }
}

impl MempoolManager {
    /// Load mempool from disk
    pub fn load_from_disk<P: AsRef<std::path::Path>>(&mut self, path: P) -> Result<()> {
        use blvm_protocol::serialization::transaction::deserialize_transaction;
        use std::fs::File;
        use std::io::Read;

        let mut file = File::open(path)?;
        let mut count_bytes = [0u8; 4];
        file.read_exact(&mut count_bytes)?;
        let count = u32::from_le_bytes(count_bytes) as usize;

        for _ in 0..count {
            let mut len_bytes = [0u8; 4];
            file.read_exact(&mut len_bytes)?;
            let len = u32::from_le_bytes(len_bytes) as usize;

            let mut tx_bytes = vec![0u8; len];
            file.read_exact(&mut tx_bytes)?;

            let tx = deserialize_transaction(&tx_bytes)?;
            drop(self.add_transaction(tx));
        }

        Ok(())
    }
}