blvm-protocol 0.1.35

//! Bitcoin P2P Network Protocol (Orange Paper Section 10)
//!
//! This module provides Bitcoin P2P protocol message types and processing.
//! Protocol-specific limits and validation are handled here, with consensus
//! validation delegated to the consensus layer.

use crate::bip152::ShortTxId;
use crate::error::ProtocolError;
use crate::validation::ProtocolValidationContext;
use crate::{BitcoinProtocolEngine, ProtocolConfig, Result};
use blvm_consensus::error::ConsensusError;
use blvm_consensus::types::UtxoSet;
use blvm_consensus::types::{Block, BlockHeader, Hash, Transaction, ValidationResult};
use std::sync::Arc;
use thiserror::Error;

// Commons module is always available (ban list sharing doesn't require utxo-commitments)
pub mod commons {
    pub use crate::commons::*;
}

// BIP324: v2 encrypted transport
#[cfg(all(
    feature = "bip324",
    any(target_arch = "x86_64", target_arch = "aarch64")
))]
pub mod v2_transport {
    pub use crate::v2_transport::*;
}

#[cfg(test)]
mod bip155_tests;

/// NetworkMessage: Bitcoin P2P protocol message types
///
/// Network message types for Bitcoin P2P protocol
#[derive(Debug, Clone, PartialEq)]
pub enum NetworkMessage {
    Version(VersionMessage),
    VerAck,
    Addr(AddrMessage),
    AddrV2(AddrV2Message), // BIP155: Extended address format
    Inv(InvMessage),
    GetData(GetDataMessage),
    GetHeaders(GetHeadersMessage),
    Headers(HeadersMessage),
    /// Full block message.  Witnesses are required for correct SegWit/Taproot validation and
    /// wire round-tripping.  The second field is the per-tx, per-input witness stack.
    Block(Arc<Block>, Vec<Vec<blvm_consensus::segwit::Witness>>),
    Tx(Arc<Transaction>),
    Ping(PingMessage),
    Pong(PongMessage),
    MemPool,
    FeeFilter(FeeFilterMessage),
    // Additional core P2P messages
    GetBlocks(GetBlocksMessage),
    GetAddr,
    NotFound(NotFoundMessage),
    Reject(RejectMessage),
    SendHeaders,
    // BIP152 Compact Block Relay
    SendCmpct(SendCmpctMessage),
    CmpctBlock(CmpctBlockMessage),
    GetBlockTxn(GetBlockTxnMessage),
    BlockTxn(BlockTxnMessage),
    // Commons-specific protocol extensions
    #[cfg(feature = "utxo-commitments")]
    GetUTXOSet(commons::GetUTXOSetMessage),
    #[cfg(feature = "utxo-commitments")]
    UTXOSet(commons::UTXOSetMessage),
    #[cfg(feature = "utxo-commitments")]
    GetFilteredBlock(commons::GetFilteredBlockMessage),
    #[cfg(feature = "utxo-commitments")]
    FilteredBlock(commons::FilteredBlockMessage),
    GetBanList(commons::GetBanListMessage),
    BanList(commons::BanListMessage),
}

/// Version message for initial handshake
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct VersionMessage {
    pub version: u32,
    pub services: u64,
    pub timestamp: i64,
    pub addr_recv: NetworkAddress,
    pub addr_from: NetworkAddress,
    pub nonce: u64,
    pub user_agent: String,
    pub start_height: i32,
    pub relay: bool,
}

/// Block P2P message: consensus block paired with its segwit witness stack.
///
/// The consensus [`Block`] carries transactions without witness data; witnesses are kept
/// alongside so the node can round-trip the full segwit wire format without altering the
/// consensus type.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct BlockMessage {
    pub block: Block,
    /// One `Vec<Witness>` per transaction, one `Witness` per input.
    #[serde(skip_serializing_if = "Vec::is_empty", default)]
    pub witnesses: Vec<Vec<blvm_consensus::segwit::Witness>>,
}

/// Transaction P2P message (thin wrapper for dispatch).
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct TxMessage {
    pub transaction: Transaction,
    /// One witness stack per input. Empty when the wire encoding has no witness flag.
    #[serde(skip_serializing_if = "Vec::is_empty", default)]
    pub witnesses: Vec<crate::segwit::Witness>,
}

/// Compact block P2P message: decoded BIP152 compact block.
///
/// Holds the [`crate::bip152::CompactBlock`] abstraction used for efficient block relay.
/// The node converts between this and the wire [`CmpctBlockMessage`] on ingress/egress.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct CompactBlockMessage {
    pub compact_block: crate::bip152::CompactBlock,
}

/// Address message containing peer addresses
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct AddrMessage {
    pub addresses: Vec<NetworkAddress>,
}

/// Inventory message listing available objects
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct InvMessage {
    pub inventory: Vec<InventoryVector>,
}

/// GetData message requesting specific objects
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct GetDataMessage {
    pub inventory: Vec<InventoryVector>,
}

/// GetHeaders message requesting block headers
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct GetHeadersMessage {
    pub version: u32,
    pub block_locator_hashes: Vec<Hash>,
    pub hash_stop: Hash,
}

/// Headers message containing block headers
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct HeadersMessage {
    pub headers: Vec<BlockHeader>,
}

/// Ping message for connection keepalive
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct PingMessage {
    pub nonce: u64,
}

/// Pong message responding to ping
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct PongMessage {
    pub nonce: u64,
}

/// FeeFilter message setting minimum fee rate
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct FeeFilterMessage {
    pub feerate: u64,
}

/// GetBlocks message requesting blocks by locator
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct GetBlocksMessage {
    pub version: u32,
    pub block_locator_hashes: Vec<Hash>,
    pub hash_stop: Hash,
}

/// NotFound message indicating requested object not found
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct NotFoundMessage {
    pub inventory: Vec<InventoryVector>,
}

/// Reject message rejecting a message with reason
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct RejectMessage {
    pub message: String,          // Command name of rejected message
    pub code: u8, // Rejection code (0x01=malformed, 0x10=invalid, 0x11=obsolete, 0x12=duplicate, 0x40=nonstandard, 0x41=dust, 0x42=insufficientfee, 0x43=checkpoint)
    pub reason: String, // Human-readable reason
    pub extra_data: Option<Hash>, // Optional hash for rejected object
}

/// SendCmpct message - Negotiate compact block relay support (BIP152)
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct SendCmpctMessage {
    /// Compact block version (1 or 2)
    pub version: u64,
    /// Whether to prefer compact blocks (1) or regular blocks (0)
    pub prefer_cmpct: u8,
}

/// CmpctBlock message - Compact block data (BIP152)
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CmpctBlockMessage {
    /// Block header
    pub header: BlockHeader,
    /// Nonce for short transaction ID calculation (SipHash key derivation)
    pub nonce: u64,
    /// Short transaction IDs (6 bytes each)
    pub short_ids: Vec<ShortTxId>,
    /// Prefilled transactions (transactions that are likely missing)
    pub prefilled_txs: Vec<PrefilledTransaction>,
}

/// PrefilledTransaction - Transaction included in compact block
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PrefilledTransaction {
    /// Index in block (0 = coinbase)
    pub index: u16,
    /// Transaction data
    pub tx: Transaction,
    /// Witness data (one stack per input). If Some and non-empty, serializes with TX_WITH_WITNESS per BIP152/Core.
    pub witness: Option<Vec<blvm_consensus::segwit::Witness>>,
}

/// Error converting between [`crate::bip152::CompactBlock`] and wire [`CmpctBlockMessage`].
#[derive(Debug, Clone, PartialEq, Eq, Error)]
pub enum CompactBlockWireConvertError {
    #[error("prefilled transaction index {0} does not fit in BIP152 u16 index")]
    PrefilledIndexTooLarge(usize),
    #[error("duplicate prefilled transaction index {0}")]
    DuplicatePrefilledIndex(usize),
}

impl TryFrom<crate::bip152::CompactBlock> for CmpctBlockMessage {
    type Error = CompactBlockWireConvertError;

    /// Builds a cmpctblock-shaped message from the integration [`crate::bip152::CompactBlock`].
    ///
    /// Prefilled entries are **sorted by transaction index** (required for BIP152 diff-encoding on the wire).
    /// Witness stacks stored on the compact block are copied onto each prefilled transaction.
    fn try_from(mut value: crate::bip152::CompactBlock) -> std::result::Result<Self, Self::Error> {
        value.prefilled_txs.sort_by_key(|(i, _, _)| *i);
        let mut prefilled_txs = Vec::with_capacity(value.prefilled_txs.len());
        let mut prev_idx: Option<usize> = None;
        for (idx, tx, witness) in value.prefilled_txs {
            if prev_idx == Some(idx) {
                return Err(CompactBlockWireConvertError::DuplicatePrefilledIndex(idx));
            }
            prev_idx = Some(idx);
            let index = u16::try_from(idx)
                .map_err(|_| CompactBlockWireConvertError::PrefilledIndexTooLarge(idx))?;
            prefilled_txs.push(PrefilledTransaction { index, tx, witness });
        }
        Ok(CmpctBlockMessage {
            header: value.header,
            nonce: value.nonce,
            short_ids: value.short_ids,
            prefilled_txs,
        })
    }
}

impl From<&CmpctBlockMessage> for crate::bip152::CompactBlock {
    fn from(msg: &CmpctBlockMessage) -> Self {
        Self {
            header: msg.header.clone(),
            nonce: msg.nonce,
            short_ids: msg.short_ids.clone(),
            prefilled_txs: msg
                .prefilled_txs
                .iter()
                .map(|p| (usize::from(p.index), p.tx.clone(), p.witness.clone()))
                .collect(),
        }
    }
}

impl From<CmpctBlockMessage> for crate::bip152::CompactBlock {
    fn from(msg: CmpctBlockMessage) -> Self {
        Self {
            header: msg.header,
            nonce: msg.nonce,
            short_ids: msg.short_ids,
            prefilled_txs: msg
                .prefilled_txs
                .into_iter()
                .map(|p| (usize::from(p.index), p.tx, p.witness))
                .collect(),
        }
    }
}

/// GetBlockTxn message - Request missing transactions from compact block (BIP152)
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct GetBlockTxnMessage {
    /// Block hash for the compact block
    pub block_hash: Hash,
    /// Indices of transactions to request (0-indexed)
    pub indices: Vec<u16>,
}

/// BlockTxn message - Response with requested transactions (BIP152)
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct BlockTxnMessage {
    /// Block hash for the compact block
    pub block_hash: Hash,
    /// Requested transactions in order
    pub transactions: Vec<Transaction>,
    /// Witness data (one Vec<Witness> per tx, one Witness per input). If Some and len matches transactions, serializes with TX_WITH_WITNESS per BIP152/Core.
    pub witnesses: Option<Vec<Vec<blvm_consensus::segwit::Witness>>>,
}

/// Network address structure (legacy format)
#[derive(Debug, Clone, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
pub struct NetworkAddress {
    pub services: u64,
    pub ip: [u8; 16], // IPv6 address (IPv4 mapped to IPv6)
    pub port: u16,
    /// Advertised unix time. Zero is stored and is outside the relay horizon.
    pub time: u32,
}

/// BIP155: Address type for addrv2 message
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[repr(u8)]
pub enum AddressType {
    IPv4 = 1,
    IPv6 = 2,
    TorV2 = 3,
    TorV3 = 4,
    I2P = 5,
    CJDNS = 6,
}

impl AddressType {
    /// Get the expected address length in bytes for this type
    pub fn address_length(&self) -> usize {
        match self {
            AddressType::IPv4 => 4,
            AddressType::IPv6 => 16,
            AddressType::TorV2 => 10,
            AddressType::TorV3 => 32,
            AddressType::I2P => 32,
            AddressType::CJDNS => 16,
        }
    }

    /// Try to create AddressType from u8
    pub fn from_u8(value: u8) -> Option<Self> {
        match value {
            1 => Some(AddressType::IPv4),
            2 => Some(AddressType::IPv6),
            3 => Some(AddressType::TorV2),
            4 => Some(AddressType::TorV3),
            5 => Some(AddressType::I2P),
            6 => Some(AddressType::CJDNS),
            _ => None,
        }
    }
}

/// BIP155: Extended address message (addrv2)
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct AddrV2Message {
    pub addresses: Vec<NetworkAddressV2>,
}

/// BIP155: Extended network address structure
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct NetworkAddressV2 {
    pub time: u32,
    pub services: u64,
    pub address_type: AddressType,
    pub address: Vec<u8>, // Variable length based on address_type
    pub port: u16,
}

impl NetworkAddressV2 {
    /// Create a new NetworkAddressV2
    pub fn new(
        time: u32,
        services: u64,
        address_type: AddressType,
        address: Vec<u8>,
        port: u16,
    ) -> Result<Self> {
        let expected_len = address_type.address_length();
        if address.len() != expected_len {
            return Err(ProtocolError::Consensus(ConsensusError::Serialization(
                std::borrow::Cow::Owned(format!(
                    "Invalid address length for type {:?}: expected {}, got {}",
                    address_type,
                    expected_len,
                    address.len()
                )),
            )));
        }
        Ok(Self {
            time,
            services,
            address_type,
            address,
            port,
        })
    }

    /// Convert to legacy NetworkAddress (if possible)
    pub fn to_legacy(&self) -> Option<NetworkAddress> {
        match self.address_type {
            AddressType::IPv4 => {
                if self.address.len() == 4 {
                    // Map IPv4 to IPv6-mapped format
                    let mut ipv6 = [0u8; 16];
                    ipv6[10] = 0xff;
                    ipv6[11] = 0xff;
                    ipv6[12..16].copy_from_slice(&self.address);
                    Some(NetworkAddress {
                        time: self.time,
                        services: self.services,
                        ip: ipv6,
                        port: self.port,
                    })
                } else {
                    None
                }
            }
            AddressType::IPv6 => {
                if self.address.len() == 16 {
                    let mut ipv6 = [0u8; 16];
                    ipv6.copy_from_slice(&self.address);
                    Some(NetworkAddress {
                        time: self.time,
                        services: self.services,
                        ip: ipv6,
                        port: self.port,
                    })
                } else {
                    None
                }
            }
            _ => None, // Tor, I2P, CJDNS cannot be converted to legacy format
        }
    }
}

/// Inventory vector identifying objects
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct InventoryVector {
    pub inv_type: u32,
    pub hash: Hash,
}

/// Network response to a message
#[derive(Debug, Clone)]
pub enum NetworkResponse {
    Ok,
    SendMessage(Box<NetworkMessage>),
    SendMessages(Vec<NetworkMessage>),
    Reject(String),
}

/// Compact block held until `blocktxn` fills the short ids the mempool missed.
#[derive(Clone)]
pub(crate) struct PendingCompact {
    pub header: BlockHeader,
    pub nonce: u64,
    pub slots: Vec<Option<(Transaction, Vec<blvm_consensus::segwit::Witness>)>>,
    pub missing: Vec<(usize, ShortTxId)>,
}

/// Peer connection state
#[derive(Clone)]
pub struct PeerState {
    pub version: u32,
    pub services: u64,
    pub user_agent: String,
    pub start_height: i32,
    pub handshake_complete: bool,
    pub known_addresses: Vec<NetworkAddress>,
    pub ping_nonce: Option<u64>,
    pub last_pong: Option<std::time::SystemTime>,
    /// BIP133 minimum, in satoshis per 1,000 virtual bytes. `None` means the peer sent no filter.
    pub min_fee_rate: Option<u64>,
    /// BIP37 relay byte. `false` means this peer asked not to receive transaction announcements.
    /// A version message that omits the byte leaves this `true`.
    pub relay_tx: bool,
    /// BIP130: peer prefers `headers` over `inv` for block announcements.
    pub prefer_headers: bool,
    /// BIP152: negotiated compact-block version (1 or 2), if peer sent sendcmpct.
    pub compact_block_version: Option<u8>,
    /// BIP152: peer prefers compact blocks when true.
    pub prefer_compact_block: bool,
    /// Compact block waiting on `blocktxn` for the short ids the mempool did not fill.
    pub(crate) pending_compact: Option<PendingCompact>,
    /// BIP324: v2 encrypted transport state (if enabled)
    /// Note: V2Transport is not Clone, so we use Option<Box<V2Transport>> for storage
    #[cfg(all(
        feature = "bip324",
        any(target_arch = "x86_64", target_arch = "aarch64")
    ))]
    pub v2_transport: Option<std::sync::Arc<crate::v2_transport::V2Transport>>,
    /// BIP324: Whether v2 transport handshake is in progress
    #[cfg(all(
        feature = "bip324",
        any(target_arch = "x86_64", target_arch = "aarch64")
    ))]
    pub v2_handshake: Option<std::sync::Arc<crate::v2_transport::V2Handshake>>,
}

impl PeerState {
    pub fn new() -> Self {
        Self {
            version: 0,
            services: 0,
            user_agent: String::new(),
            start_height: 0,
            handshake_complete: false,
            known_addresses: Vec::new(),
            ping_nonce: None,
            last_pong: None,
            min_fee_rate: None,
            relay_tx: true,
            prefer_headers: false,
            compact_block_version: None,
            prefer_compact_block: false,
            pending_compact: None,
            #[cfg(all(
                feature = "bip324",
                any(target_arch = "x86_64", target_arch = "aarch64")
            ))]
            v2_transport: None,
            #[cfg(all(
                feature = "bip324",
                any(target_arch = "x86_64", target_arch = "aarch64")
            ))]
            v2_handshake: None,
        }
    }

    /// Check if peer supports BIP324 v2 encrypted transport
    #[cfg(all(
        feature = "bip324",
        any(target_arch = "x86_64", target_arch = "aarch64")
    ))]
    pub fn supports_v2_transport(&self) -> bool {
        use crate::service_flags::{has_flag, standard};
        has_flag(self.services, standard::NODE_V2_TRANSPORT)
    }

    /// Check if v2 transport is active
    #[cfg(all(
        feature = "bip324",
        any(target_arch = "x86_64", target_arch = "aarch64")
    ))]
    pub fn is_v2_transport_active(&self) -> bool {
        self.v2_transport.is_some()
    }
}

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

/// Chain object (block or transaction)
#[derive(Debug, Clone)]
pub enum ChainObject {
    Block(Arc<Block>),
    Transaction(Arc<Transaction>),
}

impl ChainObject {
    pub fn as_block(&self) -> Option<&Arc<Block>> {
        match self {
            ChainObject::Block(block) => Some(block),
            _ => None,
        }
    }

    pub fn as_transaction(&self) -> Option<&Arc<Transaction>> {
        match self {
            ChainObject::Transaction(tx) => Some(tx),
            _ => None,
        }
    }
}

/// Trait for chain state access (node layer implements this)
///
/// This trait allows the protocol layer to query chain state without
/// owning it. The node layer provides real implementations using its
/// storage modules (BlockStore, TxIndex, MempoolManager).
pub trait ChainStateAccess {
    /// Check if we have an object (block or transaction) by hash
    fn has_object(&self, hash: &Hash) -> bool;

    /// Get an object (block or transaction) by hash
    fn get_object(&self, hash: &Hash) -> Option<ChainObject>;

    /// Get headers for a block locator (for GetHeaders requests)
    /// This implements the Bitcoin block locator algorithm
    fn get_headers_for_locator(&self, locator: &[Hash], stop: &Hash) -> Vec<BlockHeader>;

    /// Get all mempool transactions
    fn get_mempool_transactions(&self) -> Vec<Transaction>;

    /// Witness stacks for a stored block (SegWit relay). Default: not available.
    fn get_block_witnesses(
        &self,
        _hash: &Hash,
    ) -> Option<Vec<Vec<blvm_consensus::segwit::Witness>>> {
        None
    }

    /// Ban entries this node would return from `getbanlist`. Default: none stored.
    fn ban_entries(&self) -> Vec<crate::commons::BanEntry> {
        Vec::new()
    }

    /// Apply one ban the way `setban` does: the address is the IP with port 0, and
    /// `timestamp` is the absolute unban time (`0` is permanent).
    fn apply_ban(&self, entry: &crate::commons::BanEntry) -> std::result::Result<(), String> {
        let _ = entry;
        Err("ban path is not available".into())
    }

    /// Stored UTXO commitment for a block, when this node has one.
    fn stored_utxo_commitment(
        &self,
        _block_hash: &Hash,
    ) -> std::result::Result<Option<crate::commons::UTXOCommitment>, String> {
        Ok(None)
    }

    /// Basic block filter for a block, the bytes `getblockfilter` already builds.
    fn stored_block_filter(
        &self,
        _block_hash: &Hash,
    ) -> std::result::Result<Option<(BlockHeader, Vec<u8>)>, String> {
        Ok(None)
    }
}

/// Process incoming network message
///
/// This function handles Bitcoin P2P protocol messages, applying protocol-specific
/// limits and delegating consensus validation to the protocol engine.
///
/// # Arguments
///
/// * `engine` - The protocol engine (contains consensus layer)
/// * `message` - The network message to process
/// * `peer_state` - Current peer connection state
/// * `chain_access` - Optional chain state access (node layer provides this)
/// * `utxo_set` - Optional UTXO set for block validation
/// * `height` - Optional block height for validation context
///
/// # Returns
///
/// A `NetworkResponse` indicating the result of processing
pub fn process_network_message(
    engine: &BitcoinProtocolEngine,
    message: &NetworkMessage,
    peer_state: &mut PeerState,
    chain_access: Option<&dyn ChainStateAccess>,
    utxo_set: Option<&UtxoSet>,
    height: Option<u64>,
) -> Result<NetworkResponse> {
    let config = engine.get_config();
    match message {
        NetworkMessage::Version(version) => process_version_message(version, peer_state, config),
        NetworkMessage::VerAck => process_verack_message(peer_state),
        NetworkMessage::Addr(addr) => process_addr_message(addr, peer_state, config),
        NetworkMessage::AddrV2(addrv2) => process_addrv2_message(addrv2, peer_state, config),
        NetworkMessage::Inv(inv) => process_inv_message(inv, chain_access, config),
        NetworkMessage::GetData(getdata) => process_getdata_message(getdata, chain_access, config),
        NetworkMessage::GetHeaders(getheaders) => {
            process_getheaders_message(getheaders, chain_access, config)
        }
        NetworkMessage::Headers(headers) => process_headers_message(headers, config),
        NetworkMessage::Block(block, witnesses) => {
            process_block_message(engine, block, witnesses, utxo_set, height, config)
        }
        NetworkMessage::Tx(tx) => process_tx_message(engine, tx, height),
        NetworkMessage::Ping(ping) => process_ping_message(ping, peer_state),
        NetworkMessage::Pong(pong) => process_pong_message(pong, peer_state),
        NetworkMessage::MemPool => process_mempool_message(chain_access),
        NetworkMessage::FeeFilter(feefilter) => process_feefilter_message(feefilter, peer_state),
        NetworkMessage::GetBlocks(getblocks) => {
            process_getblocks_message(getblocks, chain_access, config)
        }
        NetworkMessage::GetAddr => process_getaddr_message(peer_state, config),
        NetworkMessage::NotFound(notfound) => process_notfound_message(notfound, config),
        NetworkMessage::Reject(reject) => process_reject_message(reject, config),
        NetworkMessage::SendHeaders => process_sendheaders_message(peer_state),
        NetworkMessage::SendCmpct(sendcmpct) => {
            process_sendcmpct_message(sendcmpct, peer_state, config)
        }
        NetworkMessage::CmpctBlock(cmpctblock) => process_cmpctblock_message(
            engine,
            cmpctblock,
            peer_state,
            chain_access,
            utxo_set,
            height,
            config,
        ),
        NetworkMessage::GetBlockTxn(getblocktxn) => {
            process_getblocktxn_message(getblocktxn, chain_access, config)
        }
        NetworkMessage::BlockTxn(blocktxn) => {
            process_blocktxn_message(engine, blocktxn, peer_state, utxo_set, height, config)
        }
        #[cfg(feature = "utxo-commitments")]
        NetworkMessage::GetUTXOSet(getutxoset) => {
            process_getutxoset_message(getutxoset, chain_access)
        }
        #[cfg(feature = "utxo-commitments")]
        NetworkMessage::UTXOSet(utxoset) => process_utxoset_message(utxoset, chain_access),
        #[cfg(feature = "utxo-commitments")]
        NetworkMessage::GetFilteredBlock(getfiltered) => {
            process_getfilteredblock_message(getfiltered, chain_access)
        }
        #[cfg(feature = "utxo-commitments")]
        NetworkMessage::FilteredBlock(filtered) => {
            process_filteredblock_message(filtered, chain_access)
        }
        NetworkMessage::GetBanList(getbanlist) => {
            process_getbanlist_message(getbanlist, chain_access)
        }
        NetworkMessage::BanList(banlist) => process_banlist_message(banlist, chain_access),
    }
}

/// Process version message
fn process_version_message(
    version: &VersionMessage,
    peer_state: &mut PeerState,
    config: &ProtocolConfig,
) -> Result<NetworkResponse> {
    // Validate version message
    if version.version < 70001 {
        return Ok(NetworkResponse::Reject("Version too old".into()));
    }

    // Validate user agent length (from config)
    if version.user_agent.len() > config.network_limits.max_user_agent_length {
        return Ok(NetworkResponse::Reject(format!(
            "User agent too long (max {} bytes)",
            config.network_limits.max_user_agent_length
        )));
    }

    // Update peer state
    peer_state.version = version.version;
    peer_state.services = version.services;
    peer_state.user_agent = version.user_agent.clone();
    peer_state.start_height = version.start_height;

    // BIP324: Check if peer supports v2 transport and we should negotiate it
    #[cfg(all(
        feature = "bip324",
        any(target_arch = "x86_64", target_arch = "aarch64")
    ))]
    {
        use crate::service_flags::{has_flag, standard};
        let peer_supports_v2 = has_flag(version.services, standard::NODE_V2_TRANSPORT);
        let we_support_v2 = config.service_flags.node_v2_transport;

        if peer_supports_v2 && we_support_v2 {
            // Initiate v2 handshake (responder side)
            let handshake = crate::v2_transport::V2Handshake::new_responder();
            peer_state.v2_handshake = Some(std::sync::Arc::new(handshake));
            // Note: Actual handshake happens at connection level, not in version message
            // This just records that we should use v2 transport
        }
    }

    // Send verack response
    Ok(NetworkResponse::SendMessage(Box::new(
        NetworkMessage::VerAck,
    )))
}

/// Process verack message
fn process_verack_message(peer_state: &mut PeerState) -> Result<NetworkResponse> {
    peer_state.handshake_complete = true;
    Ok(NetworkResponse::Ok)
}

/// Process addr message
fn process_addr_message(
    addr: &AddrMessage,
    peer_state: &mut PeerState,
    config: &ProtocolConfig,
) -> Result<NetworkResponse> {
    // Validate address count (from config)
    if addr.addresses.len() > config.network_limits.max_addr_addresses {
        return Ok(NetworkResponse::Reject(format!(
            "Too many addresses (max {})",
            config.network_limits.max_addr_addresses
        )));
    }

    // Store addresses for future use
    peer_state.known_addresses.extend(addr.addresses.clone());

    Ok(NetworkResponse::Ok)
}

/// Process addrv2 message (BIP155)
fn process_addrv2_message(
    addrv2: &AddrV2Message,
    peer_state: &mut PeerState,
    config: &ProtocolConfig,
) -> Result<NetworkResponse> {
    // Validate address count (from config)
    if addrv2.addresses.len() > config.network_limits.max_addr_addresses {
        return Ok(NetworkResponse::Reject(format!(
            "Too many addresses (max {})",
            config.network_limits.max_addr_addresses
        )));
    }

    // Convert addrv2 addresses to legacy format where possible and store
    for addr_v2 in &addrv2.addresses {
        if let Some(legacy_addr) = addr_v2.to_legacy() {
            peer_state.known_addresses.push(legacy_addr);
        }
        // Note: Tor v3, I2P, CJDNS addresses cannot be converted to legacy format
        // but we still process them (they're stored in addrv2 format in node layer)
    }

    Ok(NetworkResponse::Ok)
}

/// Process inv message
fn process_inv_message(
    inv: &InvMessage,
    chain_access: Option<&dyn ChainStateAccess>,
    config: &ProtocolConfig,
) -> Result<NetworkResponse> {
    // Validate inventory count (from config)
    if inv.inventory.len() > config.network_limits.max_inv_items {
        return Ok(NetworkResponse::Reject(format!(
            "Too many inventory items (max {})",
            config.network_limits.max_inv_items
        )));
    }

    // Check which items we need (if chain access provided)
    if let Some(chain) = chain_access {
        let mut needed_items = Vec::with_capacity(inv.inventory.len());
        for item in &inv.inventory {
            if !chain.has_object(&item.hash) {
                needed_items.push(item.clone());
            }
        }

        if !needed_items.is_empty() {
            return Ok(NetworkResponse::SendMessage(Box::new(
                NetworkMessage::GetData(GetDataMessage {
                    inventory: needed_items,
                }),
            )));
        }
    }

    Ok(NetworkResponse::Ok)
}

/// Process getdata message
fn process_getdata_message(
    getdata: &GetDataMessage,
    chain_access: Option<&dyn ChainStateAccess>,
    config: &ProtocolConfig,
) -> Result<NetworkResponse> {
    // Validate request count (from config)
    if getdata.inventory.len() > config.network_limits.max_inv_items {
        return Ok(NetworkResponse::Reject(format!(
            "Too many getdata items (max {})",
            config.network_limits.max_inv_items
        )));
    }

    // Send requested objects (if chain access provided)
    if let Some(chain) = chain_access {
        let mut responses = Vec::with_capacity(getdata.inventory.len());
        for item in &getdata.inventory {
            if let Some(obj) = chain.get_object(&item.hash) {
                match item.inv_type {
                    1 => {
                        // MSG_TX
                        if let Some(tx) = obj.as_transaction() {
                            responses.push(NetworkMessage::Tx(Arc::clone(tx)));
                        }
                    }
                    2 => {
                        // MSG_BLOCK
                        if let Some(block) = obj.as_block() {
                            let Some(witnesses) = chain.get_block_witnesses(&item.hash) else {
                                // REV-P-08: do not synthesize empty witness stacks when storage
                                // has no witness data (node layer serves notfound instead).
                                continue;
                            };
                            responses.push(NetworkMessage::Block(Arc::clone(block), witnesses));
                        }
                    }
                    _ => {
                        // Unknown inventory type - skip
                    }
                }
            }
        }

        if !responses.is_empty() {
            return Ok(NetworkResponse::SendMessages(responses));
        }
    }

    Ok(NetworkResponse::Ok)
}

/// Process getheaders message
fn process_getheaders_message(
    getheaders: &GetHeadersMessage,
    chain_access: Option<&dyn ChainStateAccess>,
    config: &ProtocolConfig,
) -> Result<NetworkResponse> {
    // Validate block locator size (from config)
    if getheaders.block_locator_hashes.len() > config.validation.max_locator_hashes {
        return Ok(NetworkResponse::Reject(format!(
            "Too many locator hashes (max {})",
            config.validation.max_locator_hashes
        )));
    }

    // Use chain access to find headers (if provided)
    if let Some(chain) = chain_access {
        let headers =
            chain.get_headers_for_locator(&getheaders.block_locator_hashes, &getheaders.hash_stop);
        return Ok(NetworkResponse::SendMessage(Box::new(
            NetworkMessage::Headers(HeadersMessage { headers }),
        )));
    }

    Ok(NetworkResponse::Reject("Chain access not available".into()))
}

/// Process headers message
fn process_headers_message(
    headers: &HeadersMessage,
    config: &ProtocolConfig,
) -> Result<NetworkResponse> {
    // Validate header count (from config)
    if headers.headers.len() > config.network_limits.max_headers {
        return Ok(NetworkResponse::Reject(format!(
            "Too many headers (max {})",
            config.network_limits.max_headers
        )));
    }

    // Header validation is consensus logic, not protocol
    // Node layer will validate headers using consensus layer
    Ok(NetworkResponse::Ok)
}

/// Process block message
fn process_block_message(
    engine: &BitcoinProtocolEngine,
    block: &Block,
    witnesses: &[Vec<blvm_consensus::segwit::Witness>],
    utxo_set: Option<&UtxoSet>,
    height: Option<u64>,
    config: &ProtocolConfig,
) -> Result<NetworkResponse> {
    // Check protocol limits first (from config)
    if block.transactions.len() > config.validation.max_txs_per_block {
        return Err(crate::error::ProtocolError::MessageTooLarge {
            size: block.transactions.len(),
            max: config.validation.max_txs_per_block,
        });
    }

    // Delegate to consensus via protocol engine (requires utxo_set and height)
    if let (Some(utxos), Some(h)) = (utxo_set, height) {
        let context = ProtocolValidationContext::new(engine.get_protocol_version(), h)?;
        // Pass witnesses for full BIP141/Taproot weight and script validation.
        let result = engine
            .validate_block_with_protocol_and_witnesses(block, witnesses, utxos, h, &context)?;

        match result {
            ValidationResult::Valid => Ok(NetworkResponse::Ok),
            ValidationResult::Invalid(reason) => {
                Ok(NetworkResponse::Reject(format!("Invalid block: {reason}")))
            }
        }
    } else {
        Err(crate::error::ProtocolError::Configuration(
            "Missing validation context (utxo_set and height required)".into(),
        ))
    }
}

/// Process transaction message
fn process_tx_message(
    engine: &BitcoinProtocolEngine,
    tx: &Transaction,
    height: Option<u64>,
) -> Result<NetworkResponse> {
    // Check protocol limits and validate
    let context =
        ProtocolValidationContext::new(engine.get_protocol_version(), height.unwrap_or(0))?;
    let result = engine.validate_transaction_with_protocol(tx, &context)?;

    match result {
        ValidationResult::Valid => Ok(NetworkResponse::Ok),
        ValidationResult::Invalid(reason) => Ok(NetworkResponse::Reject(format!(
            "Invalid transaction: {reason}"
        ))),
    }
}

/// Process ping message
fn process_ping_message(
    ping: &PingMessage,
    _peer_state: &mut PeerState,
) -> Result<NetworkResponse> {
    let pong = NetworkMessage::Pong(PongMessage { nonce: ping.nonce });
    Ok(NetworkResponse::SendMessage(Box::new(pong)))
}

/// Process pong message
fn process_pong_message(pong: &PongMessage, peer_state: &mut PeerState) -> Result<NetworkResponse> {
    // Validate pong nonce matches our ping
    if peer_state.ping_nonce == Some(pong.nonce) {
        peer_state.ping_nonce = None;
        peer_state.last_pong = Some(std::time::SystemTime::now());
    }

    Ok(NetworkResponse::Ok)
}

/// Process mempool message
fn process_mempool_message(chain_access: Option<&dyn ChainStateAccess>) -> Result<NetworkResponse> {
    // Send all mempool transactions (if chain access provided)
    if let Some(chain) = chain_access {
        let mempool_txs = chain.get_mempool_transactions();
        let mut responses = Vec::with_capacity(mempool_txs.len());

        for tx in mempool_txs {
            responses.push(NetworkMessage::Tx(Arc::new(tx)));
        }

        if !responses.is_empty() {
            return Ok(NetworkResponse::SendMessages(responses));
        }
    }

    Ok(NetworkResponse::Ok)
}

/// Process feefilter message
fn process_feefilter_message(
    feefilter: &FeeFilterMessage,
    peer_state: &mut PeerState,
) -> Result<NetworkResponse> {
    peer_state.min_fee_rate = Some(feefilter.feerate);
    Ok(NetworkResponse::Ok)
}

/// Process getblocks message
fn process_getblocks_message(
    getblocks: &GetBlocksMessage,
    chain_access: Option<&dyn ChainStateAccess>,
    config: &ProtocolConfig,
) -> Result<NetworkResponse> {
    // Validate block locator size (from config)
    if getblocks.block_locator_hashes.len() > config.validation.max_locator_hashes {
        return Ok(NetworkResponse::Reject(format!(
            "Too many locator hashes (max {})",
            config.validation.max_locator_hashes
        )));
    }

    if let Some(chain) = chain_access {
        use blvm_consensus::block::block_header_hash;

        // Core-compatible: walk locator to fork, then advertise block hashes after fork.
        const MAX_GETBLOCKS_INV: usize = 500;
        let headers =
            chain.get_headers_for_locator(&getblocks.block_locator_hashes, &getblocks.hash_stop);
        if headers.is_empty() {
            return Ok(NetworkResponse::Ok);
        }

        let inventory: Vec<InventoryVector> = headers
            .into_iter()
            .take(MAX_GETBLOCKS_INV)
            .map(|header| InventoryVector {
                inv_type: 2, // MSG_BLOCK
                hash: block_header_hash(&header),
            })
            .collect();

        return Ok(NetworkResponse::SendMessage(Box::new(NetworkMessage::Inv(
            InvMessage { inventory },
        ))));
    }

    Ok(NetworkResponse::Ok)
}

/// Process getaddr message
fn process_getaddr_message(
    peer_state: &mut PeerState,
    config: &ProtocolConfig,
) -> Result<NetworkResponse> {
    // Return known addresses (if any)
    if !peer_state.known_addresses.is_empty() {
        // Limit to configured max addresses
        let max_addrs = config
            .network_limits
            .max_addr_addresses
            .min(peer_state.known_addresses.len());
        let mut addresses = Vec::with_capacity(max_addrs);
        addresses.extend(peer_state.known_addresses.iter().take(max_addrs).cloned());

        return Ok(NetworkResponse::SendMessage(Box::new(
            NetworkMessage::Addr(AddrMessage { addresses }),
        )));
    }

    Ok(NetworkResponse::Ok)
}

/// Process notfound message
fn process_notfound_message(
    notfound: &NotFoundMessage,
    config: &ProtocolConfig,
) -> Result<NetworkResponse> {
    // Validate inventory count (from config)
    if notfound.inventory.len() > config.network_limits.max_inv_items {
        return Ok(NetworkResponse::Reject(format!(
            "Too many notfound items (max {})",
            config.network_limits.max_inv_items
        )));
    }

    // NotFound is informational - just acknowledge
    Ok(NetworkResponse::Ok)
}

/// Process reject message
fn process_reject_message(
    reject: &RejectMessage,
    _config: &ProtocolConfig,
) -> Result<NetworkResponse> {
    // Validate message name length (Bitcoin protocol limit: 12 bytes)
    // This is a fixed protocol limit, not configurable
    if reject.message.len() > 12 {
        return Ok(NetworkResponse::Reject(
            "Invalid reject message name".into(),
        ));
    }

    // Validate reason length (Bitcoin protocol limit: 111 bytes)
    // This is a fixed protocol limit, not configurable
    if reject.reason.len() > 111 {
        return Ok(NetworkResponse::Reject("Reject reason too long".into()));
    }

    // Reject is informational - log and acknowledge
    // In production, this would trigger appropriate handling (ban peer, etc.)
    Ok(NetworkResponse::Ok)
}

/// Process sendheaders message
fn process_sendheaders_message(peer_state: &mut PeerState) -> Result<NetworkResponse> {
    peer_state.prefer_headers = true;
    Ok(NetworkResponse::Ok)
}

/// Process sendcmpct message (BIP152)
fn process_sendcmpct_message(
    sendcmpct: &SendCmpctMessage,
    peer_state: &mut PeerState,
    config: &ProtocolConfig,
) -> Result<NetworkResponse> {
    // Validate version (must be 1 or 2, or match configured preferred version)
    let valid_versions = [1, 2];
    if !valid_versions.contains(&sendcmpct.version) {
        return Ok(NetworkResponse::Reject(
            "Invalid compact block version".into(),
        ));
    }

    // Check if compact blocks are enabled
    if !config.compact_blocks.enabled {
        return Ok(NetworkResponse::Reject("Compact blocks not enabled".into()));
    }

    // Store compact block preference in peer state
    peer_state.compact_block_version = Some(sendcmpct.version as u8);
    peer_state.prefer_compact_block = sendcmpct.prefer_cmpct != 0;
    Ok(NetworkResponse::Ok)
}

fn txid_of(tx: &Transaction) -> Hash {
    use sha2::{Digest, Sha256};
    let bytes = crate::wire::serialize_tx(tx).unwrap_or_default();
    let first = Sha256::digest(&bytes);
    let second = Sha256::digest(first);
    let mut hash = [0u8; 32];
    hash.copy_from_slice(&second);
    hash
}

fn header_hash(header: &BlockHeader) -> Hash {
    use sha2::{Digest, Sha256};
    let bytes = crate::serialization::serialize_block_header(header);
    let first = Sha256::digest(&bytes);
    let second = Sha256::digest(first);
    let mut hash = [0u8; 32];
    hash.copy_from_slice(&second);
    hash
}

fn short_id_hex(id: &ShortTxId) -> String {
    hex::encode(id)
}

/// Place prefilled transactions and unique mempool matches. Missing short ids stay empty.
fn begin_pending_compact(
    msg: &CmpctBlockMessage,
    mempool: &[Transaction],
) -> std::result::Result<PendingCompact, String> {
    let n = msg.short_ids.len() + msg.prefilled_txs.len();
    let mut slots: Vec<Option<(Transaction, Vec<blvm_consensus::segwit::Witness>)>> = vec![None; n];
    let mut prefilled = std::collections::HashSet::new();
    for item in &msg.prefilled_txs {
        let index = usize::from(item.index);
        if index >= n || !prefilled.insert(index) {
            return Err(format!(
                "prefilled compact-block index {index} is not usable"
            ));
        }
        slots[index] = Some((item.tx.clone(), item.witness.clone().unwrap_or_default()));
    }

    let mut missing = Vec::new();
    let mut next = 0usize;
    for short_id in &msg.short_ids {
        while prefilled.contains(&next) {
            next = next.saturating_add(1);
        }
        let block_index = next;
        next = next.saturating_add(1);
        if block_index >= n {
            return Err("compact block short ids ran past the transaction list".into());
        }
        let mut found: Option<(Transaction, Vec<blvm_consensus::segwit::Witness>)> = None;
        let mut hits = 0usize;
        for tx in mempool {
            let id = crate::bip152::short_tx_id(&msg.header, &txid_of(tx), msg.nonce);
            if &id == short_id {
                hits += 1;
                found = Some((tx.clone(), Vec::new()));
            }
        }
        if hits == 1 {
            slots[block_index] = found;
        } else {
            missing.push((block_index, *short_id));
        }
    }
    Ok(PendingCompact {
        header: msg.header.clone(),
        nonce: msg.nonce,
        slots,
        missing,
    })
}

fn finish_compact_block(
    engine: &BitcoinProtocolEngine,
    pending: PendingCompact,
    utxo_set: Option<&UtxoSet>,
    height: Option<u64>,
    config: &ProtocolConfig,
) -> Result<NetworkResponse> {
    if !pending.missing.is_empty() {
        let id = short_id_hex(&pending.missing[0].1);
        return Ok(NetworkResponse::Reject(format!(
            "short id {id} matches neither the mempool nor blocktxn"
        )));
    }
    let mut transactions = Vec::with_capacity(pending.slots.len());
    let mut witnesses = Vec::with_capacity(pending.slots.len());
    for slot in pending.slots {
        let Some((tx, witness)) = slot else {
            return Ok(NetworkResponse::Reject(
                "compact block still has an empty transaction slot".into(),
            ));
        };
        transactions.push(tx);
        witnesses.push(witness);
    }
    let block = Block::from_parts(pending.header, transactions.into_boxed_slice());
    match process_block_message(engine, &block, &witnesses, utxo_set, height, config) {
        Ok(response) => Ok(response),
        Err(err) if err.to_string().contains("Missing validation context") => {
            Ok(NetworkResponse::SendMessage(Box::new(
                NetworkMessage::Block(Arc::new(block), witnesses),
            )))
        }
        Err(err) => Err(err),
    }
}

/// Process cmpctblock message (BIP152)
fn process_cmpctblock_message(
    engine: &BitcoinProtocolEngine,
    cmpctblock: &CmpctBlockMessage,
    peer_state: &mut PeerState,
    chain_access: Option<&dyn ChainStateAccess>,
    utxo_set: Option<&UtxoSet>,
    height: Option<u64>,
    config: &ProtocolConfig,
) -> Result<NetworkResponse> {
    if !config.compact_blocks.enabled {
        return Ok(NetworkResponse::Reject("Compact blocks not enabled".into()));
    }
    let mempool = chain_access
        .map(|chain| chain.get_mempool_transactions())
        .unwrap_or_default();
    let pending = match begin_pending_compact(cmpctblock, &mempool) {
        Ok(pending) => pending,
        Err(reason) => return Ok(NetworkResponse::Reject(reason)),
    };
    if pending.missing.is_empty() {
        peer_state.pending_compact = None;
        return finish_compact_block(engine, pending, utxo_set, height, config);
    }
    let indices: Vec<u16> = pending
        .missing
        .iter()
        .map(|(index, _)| u16::try_from(*index).unwrap_or(u16::MAX))
        .collect();
    let block_hash = header_hash(&pending.header);
    peer_state.pending_compact = Some(pending);
    Ok(NetworkResponse::SendMessage(Box::new(
        NetworkMessage::GetBlockTxn(GetBlockTxnMessage {
            block_hash,
            indices,
        }),
    )))
}

/// Process getblocktxn message (BIP152)
fn process_getblocktxn_message(
    getblocktxn: &GetBlockTxnMessage,
    chain_access: Option<&dyn ChainStateAccess>,
    config: &ProtocolConfig,
) -> Result<NetworkResponse> {
    // Validate indices count (from config)
    if getblocktxn.indices.len() > config.compact_blocks.max_blocktxn_indices {
        return Ok(NetworkResponse::Reject(format!(
            "Too many transaction indices (max {})",
            config.compact_blocks.max_blocktxn_indices
        )));
    }

    // Use chain access to get requested transactions and witnesses
    if let Some(chain) = chain_access {
        if let Some(obj) = chain.get_object(&getblocktxn.block_hash) {
            if let Some(block) = obj.as_block() {
                let stored_witnesses = chain.get_block_witnesses(&getblocktxn.block_hash);
                let mut transactions = Vec::with_capacity(getblocktxn.indices.len());
                let mut req_witnesses = Vec::with_capacity(getblocktxn.indices.len());

                for &index in &getblocktxn.indices {
                    let idx = index as usize;
                    if idx >= block.transactions.len() {
                        return Ok(NetworkResponse::Reject(
                            "getblocktxn index past end of block".into(),
                        ));
                    }
                    transactions.push(block.transactions[idx].clone());
                    let tx_witness = stored_witnesses
                        .as_ref()
                        .and_then(|w| w.get(idx).cloned())
                        .unwrap_or_else(|| {
                            block.transactions[idx]
                                .inputs
                                .iter()
                                .map(|_| Vec::new())
                                .collect()
                        });
                    req_witnesses.push(tx_witness);
                }

                if !transactions.is_empty() {
                    return Ok(NetworkResponse::SendMessage(Box::new(
                        NetworkMessage::BlockTxn(BlockTxnMessage {
                            block_hash: getblocktxn.block_hash,
                            transactions,
                            witnesses: Some(req_witnesses),
                        }),
                    )));
                }
            }
        }
    }

    Ok(NetworkResponse::Ok)
}

/// Process blocktxn message (BIP152)
fn process_blocktxn_message(
    engine: &BitcoinProtocolEngine,
    blocktxn: &BlockTxnMessage,
    peer_state: &mut PeerState,
    utxo_set: Option<&UtxoSet>,
    height: Option<u64>,
    config: &ProtocolConfig,
) -> Result<NetworkResponse> {
    let Some(mut pending) = peer_state.pending_compact.clone() else {
        return Ok(NetworkResponse::Reject(
            "blocktxn does not match a pending compact block".into(),
        ));
    };
    if header_hash(&pending.header) != blocktxn.block_hash {
        return Ok(NetworkResponse::Reject(
            "blocktxn does not match a pending compact block".into(),
        ));
    }
    if blocktxn.transactions.len() != pending.missing.len() {
        let id = pending
            .missing
            .first()
            .map(|(_, short_id)| short_id_hex(short_id))
            .unwrap_or_else(|| "none".into());
        return Ok(NetworkResponse::Reject(format!(
            "short id {id} matches neither the mempool nor blocktxn"
        )));
    }
    for (i, (index, expected)) in pending.missing.iter().enumerate() {
        let tx = &blocktxn.transactions[i];
        let got = crate::bip152::short_tx_id(&pending.header, &txid_of(tx), pending.nonce);
        if &got != expected {
            return Ok(NetworkResponse::Reject(format!(
                "short id {} matches neither the mempool nor blocktxn",
                short_id_hex(expected)
            )));
        }
        let witness = blocktxn
            .witnesses
            .as_ref()
            .and_then(|all| all.get(i).cloned())
            .unwrap_or_default();
        pending.slots[*index] = Some((tx.clone(), witness));
    }
    pending.missing.clear();
    peer_state.pending_compact = None;
    finish_compact_block(engine, pending, utxo_set, height, config)
}

#[cfg(feature = "utxo-commitments")]
/// Process getutxoset message
fn process_getutxoset_message(
    getutxoset: &commons::GetUTXOSetMessage,
    chain_access: Option<&dyn ChainStateAccess>,
) -> Result<NetworkResponse> {
    let Some(chain) = chain_access else {
        return Err(ProtocolError::Validation(
            "utxo set is not available".into(),
        ));
    };
    match chain.stored_utxo_commitment(&getutxoset.block_hash) {
        Ok(Some(commitment)) => Ok(NetworkResponse::SendMessage(Box::new(
            NetworkMessage::UTXOSet(commons::UTXOSetMessage {
                request_id: 0,
                utxo_count: commitment.utxo_count,
                is_complete: true,
                chunk_id: None,
                commitment,
            }),
        ))),
        Ok(None) => Err(ProtocolError::Validation(
            format!(
                "no stored UTXO set for block {}",
                hex::encode(getutxoset.block_hash)
            )
            .into(),
        )),
        Err(reason) => Err(ProtocolError::Validation(reason.into())),
    }
}

#[cfg(feature = "utxo-commitments")]
/// Process utxoset message
fn process_utxoset_message(
    utxoset: &commons::UTXOSetMessage,
    chain_access: Option<&dyn ChainStateAccess>,
) -> Result<NetworkResponse> {
    let Some(chain) = chain_access else {
        return Ok(NetworkResponse::Reject(
            "utxo commitment does not match the local set".into(),
        ));
    };
    match chain.stored_utxo_commitment(&utxoset.commitment.block_hash) {
        Ok(Some(local)) if local == utxoset.commitment => Ok(NetworkResponse::Ok),
        Ok(_) => Ok(NetworkResponse::Reject(format!(
            "utxo commitment does not match the local set for {}",
            hex::encode(utxoset.commitment.block_hash)
        ))),
        Err(reason) => Ok(NetworkResponse::Reject(reason)),
    }
}

#[cfg(feature = "utxo-commitments")]
/// Process getfilteredblock message
fn process_getfilteredblock_message(
    getfiltered: &commons::GetFilteredBlockMessage,
    chain_access: Option<&dyn ChainStateAccess>,
) -> Result<NetworkResponse> {
    let Some(chain) = chain_access else {
        return Err(ProtocolError::Validation(
            "block filter is not available".into(),
        ));
    };
    match chain.stored_block_filter(&getfiltered.block_hash) {
        Ok(Some((header, filter))) => {
            let commitment = chain
                .stored_utxo_commitment(&getfiltered.block_hash)
                .ok()
                .flatten()
                .unwrap_or(commons::UTXOCommitment {
                    merkle_root: [0u8; 32],
                    total_supply: 0,
                    utxo_count: 0,
                    block_height: 0,
                    block_hash: getfiltered.block_hash,
                });
            Ok(NetworkResponse::SendMessage(Box::new(
                NetworkMessage::FilteredBlock(commons::FilteredBlockMessage {
                    request_id: getfiltered.request_id,
                    header,
                    commitment,
                    transactions: Vec::new(),
                    transaction_indices: Vec::new(),
                    spam_summary: commons::SpamSummary {
                        filtered_count: 0,
                        filtered_value: 0,
                        filter_reasons: 0,
                    },
                    bip158_filter: Some(filter),
                }),
            )))
        }
        Ok(None) => Err(ProtocolError::Validation(
            format!(
                "no block filter for {}",
                hex::encode(getfiltered.block_hash)
            )
            .into(),
        )),
        Err(reason) => Err(ProtocolError::Validation(reason.into())),
    }
}

#[cfg(feature = "utxo-commitments")]
/// Process filteredblock message
fn process_filteredblock_message(
    filtered: &commons::FilteredBlockMessage,
    chain_access: Option<&dyn ChainStateAccess>,
) -> Result<NetworkResponse> {
    let Some(chain) = chain_access else {
        return Ok(NetworkResponse::Reject(
            "filtered block does not match the local filter".into(),
        ));
    };
    let block_hash = header_hash(&filtered.header);
    match chain.stored_block_filter(&block_hash) {
        Ok(Some((_, local))) if filtered.bip158_filter.as_ref() == Some(&local) => {
            Ok(NetworkResponse::Ok)
        }
        Ok(_) => Ok(NetworkResponse::Reject(format!(
            "filtered block does not match the local filter for {}",
            hex::encode(block_hash)
        ))),
        Err(reason) => Ok(NetworkResponse::Reject(reason)),
    }
}

/// Process getbanlist message
fn process_getbanlist_message(
    getbanlist: &commons::GetBanListMessage,
    chain_access: Option<&dyn ChainStateAccess>,
) -> Result<NetworkResponse> {
    let mut entries = chain_access
        .map(|chain| chain.ban_entries())
        .unwrap_or_default();
    if let Some(min_score) = getbanlist.min_score {
        entries.retain(|entry| entry.score >= min_score);
    }
    Ok(NetworkResponse::SendMessage(Box::new(
        NetworkMessage::BanList(commons::BanListMessage {
            request_id: getbanlist.request_id,
            entries,
            list_signature: None,
        }),
    )))
}

/// Process banlist message
fn process_banlist_message(
    banlist: &commons::BanListMessage,
    chain_access: Option<&dyn ChainStateAccess>,
) -> Result<NetworkResponse> {
    if banlist.entries.is_empty() {
        return Ok(NetworkResponse::Ok);
    }
    let Some(chain) = chain_access else {
        return Ok(NetworkResponse::Reject("ban path is not available".into()));
    };
    for entry in &banlist.entries {
        if let Err(reason) = chain.apply_ban(entry) {
            return Ok(NetworkResponse::Reject(reason));
        }
    }
    Ok(NetworkResponse::Ok)
}

#[cfg(test)]
fn assert_getblocktxn_past_end_rejects() {
    struct TwoTx;
    impl ChainStateAccess for TwoTx {
        fn has_object(&self, _hash: &Hash) -> bool {
            true
        }
        fn get_object(&self, hash: &Hash) -> Option<ChainObject> {
            let tx = Transaction {
                version: 1,
                inputs: vec![].into(),
                outputs: vec![].into(),
                lock_time: 0,
            };
            let block = Block {
                header: BlockHeader {
                    version: 1,
                    prev_block_hash: [0; 32],
                    merkle_root: [0; 32],
                    timestamp: 1,
                    bits: 1,
                    nonce: 1,
                },
                transactions: vec![tx.clone(), tx].into_boxed_slice(),
            };
            if hash == &[9u8; 32] {
                Some(ChainObject::Block(Arc::new(block)))
            } else {
                None
            }
        }
        fn get_headers_for_locator(&self, _locator: &[Hash], _stop: &Hash) -> Vec<BlockHeader> {
            Vec::new()
        }
        fn get_mempool_transactions(&self) -> Vec<Transaction> {
            Vec::new()
        }
    }
    let msg = GetBlockTxnMessage {
        block_hash: [9u8; 32],
        indices: vec![5],
    };
    let response =
        process_getblocktxn_message(&msg, Some(&TwoTx), &ProtocolConfig::default()).unwrap();
    assert!(
        matches!(response, NetworkResponse::Reject(_)),
        "expected reject, got {response:?}"
    );
}

#[cfg(test)]
#[test]
fn getblocktxn_index_past_block_returns_reject_and_no_blocktxn() {
    assert_getblocktxn_past_end_rejects();
}

#[cfg(test)]
fn sample_tx(tag: u8) -> Transaction {
    Transaction {
        version: 1,
        inputs: vec![blvm_consensus::types::TransactionInput {
            prevout: blvm_consensus::types::OutPoint {
                hash: [tag; 32],
                index: 0,
            },
            script_sig: vec![tag],
            sequence: 0xffffffff,
        }]
        .into(),
        outputs: vec![blvm_consensus::types::TransactionOutput {
            value: 1000,
            script_pubkey: vec![0x51],
        }]
        .into(),
        lock_time: 0,
    }
}

#[cfg(test)]
fn sample_header() -> BlockHeader {
    BlockHeader {
        version: 1,
        prev_block_hash: [2u8; 32],
        merkle_root: [3u8; 32],
        timestamp: 1_700_000_000,
        bits: 0x1d00ffff,
        nonce: 4,
    }
}

#[cfg(test)]
struct MempoolChain(Transaction);

#[cfg(test)]
impl ChainStateAccess for MempoolChain {
    fn has_object(&self, _hash: &Hash) -> bool {
        false
    }
    fn get_object(&self, _hash: &Hash) -> Option<ChainObject> {
        None
    }
    fn get_headers_for_locator(&self, _locator: &[Hash], _stop: &Hash) -> Vec<BlockHeader> {
        Vec::new()
    }
    fn get_mempool_transactions(&self) -> Vec<Transaction> {
        vec![self.0.clone()]
    }
}

#[cfg(test)]
#[test]
fn compact_block_rebuilds_from_mempool_and_names_a_miss() {
    use crate::{BitcoinProtocolEngine, ProtocolVersion};
    let engine = BitcoinProtocolEngine::new(ProtocolVersion::BitcoinV1).unwrap();
    let header = sample_header();
    let coinbase = sample_tx(0);
    let pooled = sample_tx(7);
    let nonce = 9u64;
    let short = crate::bip152::short_tx_id(&header, &txid_of(&pooled), nonce);
    let cmpct = CmpctBlockMessage {
        header: header.clone(),
        nonce,
        short_ids: vec![short],
        prefilled_txs: vec![PrefilledTransaction {
            index: 0,
            tx: coinbase.clone(),
            witness: None,
        }],
    };
    let mut peer = PeerState::new();
    let response = process_cmpctblock_message(
        &engine,
        &cmpct,
        &mut peer,
        Some(&MempoolChain(pooled.clone())),
        None,
        None,
        &ProtocolConfig::default(),
    )
    .unwrap();
    match response {
        NetworkResponse::SendMessage(message) => match *message {
            NetworkMessage::Block(block, _) => {
                assert_eq!(block.header, header);
                assert_eq!(block.transactions.len(), 2);
                assert_eq!(block.transactions[0], coinbase);
                assert_eq!(block.transactions[1], pooled);
            }
            other => panic!("expected the reconstructed block, got {other:?}"),
        },
        other => panic!("expected the reconstructed block, got {other:?}"),
    }

    let missing = [0xabu8; 6];
    let cmpct = CmpctBlockMessage {
        header: header.clone(),
        nonce,
        short_ids: vec![missing],
        prefilled_txs: vec![PrefilledTransaction {
            index: 0,
            tx: coinbase,
            witness: None,
        }],
    };
    let mut peer = PeerState::new();
    let asked = process_cmpctblock_message(
        &engine,
        &cmpct,
        &mut peer,
        Some(&MempoolChain(pooled.clone())),
        None,
        None,
        &ProtocolConfig::default(),
    )
    .unwrap();
    assert!(matches!(
        asked,
        NetworkResponse::SendMessage(ref message) if matches!(**message, NetworkMessage::GetBlockTxn(_))
    ));
    let wrong = BlockTxnMessage {
        block_hash: header_hash(&header),
        transactions: vec![pooled],
        witnesses: None,
    };
    let rejected = process_blocktxn_message(
        &engine,
        &wrong,
        &mut peer,
        None,
        None,
        &ProtocolConfig::default(),
    )
    .unwrap();
    match rejected {
        NetworkResponse::Reject(reason) => {
            assert!(reason.contains(&short_id_hex(&missing)), "{reason}");
        }
        other => panic!("expected a named short-id reject, got {other:?}"),
    }
}

#[cfg(test)]
struct BanBook {
    entries: std::sync::Mutex<Vec<crate::commons::BanEntry>>,
}

#[cfg(test)]
impl ChainStateAccess for BanBook {
    fn has_object(&self, _hash: &Hash) -> bool {
        false
    }
    fn get_object(&self, _hash: &Hash) -> Option<ChainObject> {
        None
    }
    fn get_headers_for_locator(&self, _locator: &[Hash], _stop: &Hash) -> Vec<BlockHeader> {
        Vec::new()
    }
    fn get_mempool_transactions(&self) -> Vec<Transaction> {
        Vec::new()
    }
    fn ban_entries(&self) -> Vec<crate::commons::BanEntry> {
        self.entries.lock().expect("ban book").clone()
    }
    fn apply_ban(&self, entry: &crate::commons::BanEntry) -> std::result::Result<(), String> {
        self.entries.lock().expect("ban book").push(entry.clone());
        Ok(())
    }
}

#[cfg(test)]
#[test]
fn ban_messages_list_and_apply_without_the_old_reject() {
    let book = BanBook {
        entries: std::sync::Mutex::new(Vec::new()),
    };
    let empty = process_getbanlist_message(
        &crate::commons::GetBanListMessage {
            request_id: 1,
            min_score: None,
        },
        Some(&book),
    )
    .unwrap();
    match empty {
        NetworkResponse::SendMessage(message) => match *message {
            NetworkMessage::BanList(list) => assert!(list.entries.is_empty()),
            other => panic!("expected an empty ban list, got {other:?}"),
        },
        other => panic!("expected an empty ban list, got {other:?}"),
    }
    let entry = crate::commons::BanEntry {
        ip: [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0xff, 0xff, 1, 2, 3, 4],
        score: 100,
        reason: 1,
        timestamp: 1_700_000_000,
        signature: None,
    };
    process_banlist_message(
        &crate::commons::BanListMessage {
            request_id: 2,
            entries: vec![entry.clone()],
            list_signature: None,
        },
        Some(&book),
    )
    .unwrap();
    assert_eq!(book.entries.lock().unwrap().as_slice(), &[entry]);
    let none = process_getbanlist_message(
        &crate::commons::GetBanListMessage {
            request_id: 3,
            min_score: None,
        },
        None,
    )
    .unwrap();
    assert!(matches!(none, NetworkResponse::SendMessage(_)));
}

#[cfg(all(test, feature = "utxo-commitments"))]
struct UtxoBook {
    hash: Hash,
    commitment: crate::commons::UTXOCommitment,
    header: BlockHeader,
    filter: Vec<u8>,
}

#[cfg(all(test, feature = "utxo-commitments"))]
impl ChainStateAccess for UtxoBook {
    fn has_object(&self, _hash: &Hash) -> bool {
        false
    }
    fn get_object(&self, _hash: &Hash) -> Option<ChainObject> {
        None
    }
    fn get_headers_for_locator(&self, _locator: &[Hash], _stop: &Hash) -> Vec<BlockHeader> {
        Vec::new()
    }
    fn get_mempool_transactions(&self) -> Vec<Transaction> {
        Vec::new()
    }
    fn stored_utxo_commitment(
        &self,
        block_hash: &Hash,
    ) -> std::result::Result<Option<crate::commons::UTXOCommitment>, String> {
        if block_hash == &self.hash {
            Ok(Some(self.commitment.clone()))
        } else {
            Ok(None)
        }
    }
    fn stored_block_filter(
        &self,
        block_hash: &Hash,
    ) -> std::result::Result<Option<(BlockHeader, Vec<u8>)>, String> {
        if block_hash == &self.hash {
            Ok(Some((self.header.clone(), self.filter.clone())))
        } else {
            Ok(None)
        }
    }
}

#[cfg(all(test, feature = "utxo-commitments"))]
#[test]
fn utxo_and_filter_messages_use_the_stored_set() {
    let hash = [9u8; 32];
    let commitment = crate::commons::UTXOCommitment {
        merkle_root: [4u8; 32],
        total_supply: 50,
        utxo_count: 2,
        block_height: 10,
        block_hash: hash,
    };
    let book = UtxoBook {
        hash,
        commitment: commitment.clone(),
        header: sample_header(),
        filter: vec![0x01, 0x02, 0x03],
    };
    let got = process_getutxoset_message(
        &crate::commons::GetUTXOSetMessage {
            height: 10,
            block_hash: hash,
        },
        Some(&book),
    )
    .unwrap();
    match got {
        NetworkResponse::SendMessage(message) => match *message {
            NetworkMessage::UTXOSet(msg) => assert_eq!(msg.commitment, commitment),
            other => panic!("expected the stored commitment, got {other:?}"),
        },
        other => panic!("expected the stored commitment, got {other:?}"),
    }
    let missing = process_getutxoset_message(
        &crate::commons::GetUTXOSetMessage {
            height: 11,
            block_hash: [1u8; 32],
        },
        Some(&book),
    );
    let err = missing.expect_err("a missing set is an error");
    assert!(!err.to_string().contains("not implemented"));

    let matched = process_utxoset_message(
        &crate::commons::UTXOSetMessage {
            request_id: 1,
            commitment: commitment.clone(),
            utxo_count: 2,
            is_complete: true,
            chunk_id: None,
        },
        Some(&book),
    )
    .unwrap();
    assert!(matches!(matched, NetworkResponse::Ok));
    let mut other_commitment = commitment.clone();
    other_commitment.total_supply = 1;
    let rejected = process_utxoset_message(
        &crate::commons::UTXOSetMessage {
            request_id: 1,
            commitment: other_commitment,
            utxo_count: 2,
            is_complete: true,
            chunk_id: None,
        },
        Some(&book),
    )
    .unwrap();
    assert!(matches!(rejected, NetworkResponse::Reject(_)));

    let filter_msg = process_getfilteredblock_message(
        &crate::commons::GetFilteredBlockMessage {
            request_id: 7,
            block_hash: hash,
            filter_preferences: crate::commons::FilterPreferences {
                filter_ordinals: false,
                filter_dust: false,
                filter_brc20: false,
                min_output_value: 0,
            },
            include_bip158_filter: true,
        },
        Some(&book),
    )
    .unwrap();
    let sent_filter = match filter_msg {
        NetworkResponse::SendMessage(message) => match *message {
            NetworkMessage::FilteredBlock(msg) => msg.bip158_filter.expect("filter bytes"),
            other => panic!("expected a filter, got {other:?}"),
        },
        other => panic!("expected a filter, got {other:?}"),
    };
    assert_eq!(sent_filter, vec![0x01, 0x02, 0x03]);

    let header = sample_header();
    let stored_hash = header_hash(&header);
    let book = UtxoBook {
        hash: stored_hash,
        commitment,
        header: header.clone(),
        filter: vec![0x01, 0x02, 0x03],
    };
    let ok = process_filteredblock_message(
        &crate::commons::FilteredBlockMessage {
            request_id: 7,
            header: header.clone(),
            commitment: book.commitment.clone(),
            transactions: Vec::new(),
            transaction_indices: Vec::new(),
            spam_summary: crate::commons::SpamSummary {
                filtered_count: 0,
                filtered_value: 0,
                filter_reasons: 0,
            },
            bip158_filter: Some(vec![0x01, 0x02, 0x03]),
        },
        Some(&book),
    )
    .unwrap();
    assert!(matches!(ok, NetworkResponse::Ok));
    let bad = process_filteredblock_message(
        &crate::commons::FilteredBlockMessage {
            request_id: 7,
            header,
            commitment: book.commitment.clone(),
            transactions: Vec::new(),
            transaction_indices: Vec::new(),
            spam_summary: crate::commons::SpamSummary {
                filtered_count: 0,
                filtered_value: 0,
                filter_reasons: 0,
            },
            bip158_filter: Some(vec![0xff]),
        },
        Some(&book),
    )
    .unwrap();
    assert!(matches!(bad, NetworkResponse::Reject(_)));
}