nautilus-blockchain 0.61.0

// -------------------------------------------------------------------------------------------------
//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
//  https://nautechsystems.io
//
//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
//  You may not use this file except in compliance with the License.
//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
//
//  Unless required by applicable law or agreed to in writing, software
//  distributed under the License is distributed on an "AS IS" BASIS,
//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//  See the License for the specific language governing permissions and
//  limitations under the License.
// -------------------------------------------------------------------------------------------------

use std::collections::VecDeque;

use nautilus_common::{
    clients::DataClient,
    defi::RequestPoolSnapshot,
    live::get_runtime,
    messages::{
        DataEvent,
        defi::{
            DefiDataCommand, DefiRequestCommand, DefiSubscribeCommand, DefiUnsubscribeCommand,
            SubscribeBlocks, SubscribePool, SubscribePoolFeeCollects, SubscribePoolFlashEvents,
            SubscribePoolLiquidityUpdates, SubscribePoolSwaps, UnsubscribeBlocks, UnsubscribePool,
            UnsubscribePoolFeeCollects, UnsubscribePoolFlashEvents,
            UnsubscribePoolLiquidityUpdates, UnsubscribePoolSwaps,
        },
    },
};
use nautilus_model::{
    defi::{DefiData, DexType, PoolIdentifier, SharedChain, validation::validate_address},
    identifiers::{ClientId, Venue},
};
use ustr::Ustr;

use crate::{
    cache::BlockchainCache,
    config::BlockchainDataClientConfig,
    data::{
        core::BlockchainDataClientCore,
        subscription::{BlockFeedBackend, BlockFeedOwner},
    },
    exchanges::get_dex_extended,
    rpc::{
        BlockchainRpcClient,
        types::{BlockchainMessage, RpcEventType},
    },
};

const MAX_PENDING_POOL_MESSAGES: usize = 10_000;

/// A client for interacting with blockchain data from multiple sources.
///
/// The `BlockchainDataClient` serves as a facade that coordinates between different blockchain
/// data providers, caching mechanisms, and contract interactions. It provides a unified interface
/// for retrieving and processing blockchain data, particularly focused on DeFi protocols.
///
/// This client supports two primary data sources:
/// 1. Direct RPC connections to blockchain nodes (via WebSocket).
/// 2. HyperSync API for efficient historical data queries.
#[derive(Debug)]
pub struct BlockchainDataClient {
    /// The client ID used to identify this client with the data engine.
    pub client_id: ClientId,
    /// The blockchain being targeted by this client instance.
    pub chain: SharedChain,
    /// Configuration parameters for the blockchain data client.
    pub config: BlockchainDataClientConfig,
    /// The core client instance that handles blockchain operations.
    /// Wrapped in Option to allow moving it into the background processing task.
    pub core_client: Option<BlockchainDataClientCore>,
    /// Channel receiver for messages from the HyperSync client.
    hypersync_rx: Option<tokio::sync::mpsc::UnboundedReceiver<BlockchainMessage>>,
    /// Channel sender for messages to the HyperSync client.
    hypersync_tx: Option<tokio::sync::mpsc::UnboundedSender<BlockchainMessage>>,
    /// Channel sender for commands to be processed asynchronously.
    command_tx: tokio::sync::mpsc::UnboundedSender<DefiDataCommand>,
    /// Channel receiver for commands to be processed asynchronously.
    command_rx: Option<tokio::sync::mpsc::UnboundedReceiver<DefiDataCommand>>,
    /// Background task for processing messages.
    process_task: Option<tokio::task::JoinHandle<()>>,
    /// Cancellation token for graceful shutdown of background tasks.
    cancellation_token: tokio_util::sync::CancellationToken,
}

impl BlockchainDataClient {
    /// Creates a new [`BlockchainDataClient`] instance for the specified configuration.
    #[must_use]
    pub fn new(client_id: ClientId, config: BlockchainDataClientConfig) -> Self {
        let chain = config.chain.clone();
        let (command_tx, command_rx) = tokio::sync::mpsc::unbounded_channel();
        let (hypersync_tx, hypersync_rx) = tokio::sync::mpsc::unbounded_channel();
        Self {
            client_id,
            chain,
            core_client: None,
            config,
            hypersync_rx: Some(hypersync_rx),
            hypersync_tx: Some(hypersync_tx),
            command_tx,
            command_rx: Some(command_rx),
            process_task: None,
            cancellation_token: tokio_util::sync::CancellationToken::new(),
        }
    }

    /// Spawns the main processing task that handles commands and blockchain data.
    ///
    /// This method creates a background task that:
    /// 1. Processes subscription/unsubscription commands from the command channel
    /// 2. Handles incoming blockchain data from HyperSync
    /// 3. Processes RPC messages if RPC client is configured
    /// 4. Routes processed data to subscribers
    fn spawn_process_task(&mut self) {
        let command_rx = if let Some(r) = self.command_rx.take() {
            r
        } else {
            log::error!("Command receiver already taken, not spawning handler");
            return;
        };

        let cancellation_token = self.cancellation_token.clone();

        let data_tx = nautilus_common::live::runner::get_data_event_sender();

        let mut hypersync_rx = self.hypersync_rx.take().unwrap();
        let hypersync_tx = self.hypersync_tx.take();

        let mut core_client = BlockchainDataClientCore::new(
            self.config.clone(),
            hypersync_tx,
            Some(data_tx),
            cancellation_token.clone(),
        );

        let handle = get_runtime().spawn(async move {
            log::debug!("Started task 'process'");

            if let Err(e) = core_client.connect().await {
                // TODO: connect() could return more granular error types to distinguish
                // cancellation from actual failures without string matching
                if e.to_string().contains("cancelled") || e.to_string().contains("Sync cancelled") {
                    log::warn!("Blockchain core client connection interrupted: {e}");
                } else {
                    log::error!("Failed to connect blockchain core client: {e}");
                }
                return;
            }

            let mut command_rx = command_rx;
            let mut pending_pool_messages = VecDeque::new();

            loop {
                tokio::select! {
                    () = cancellation_token.cancelled() => {
                        log::debug!("Received cancellation signal in Blockchain data client process task");
                        core_client.disconnect().await;
                        break;
                    }
                    command = command_rx.recv() => {
                        if let Some(cmd) = command {
                            match cmd {
                                DefiDataCommand::Subscribe(cmd) => {
                                    let chain = cmd.blockchain();
                                    if chain != core_client.chain.name {
                                        log::error!("Incorrect blockchain for subscribe command: {chain}");
                                        continue;
                                    }

                                      if let Err(e) = Self::handle_subscribe_command(cmd, &mut core_client).await{
                                        log::error!("Error processing subscribe command: {e}");
                                    }
                                }
                                DefiDataCommand::Unsubscribe(cmd) => {
                                    let chain = cmd.blockchain();
                                    if chain != core_client.chain.name {
                                        log::error!("Incorrect blockchain for subscribe command: {chain}");
                                        continue;
                                    }

                                    if let Err(e) = Self::handle_unsubscribe_command(cmd, &mut core_client).await{
                                        log::error!("Error processing subscribe command: {e}");
                                    }
                                }
                                DefiDataCommand::Request(cmd) => {
                                    if let Err(e) = Self::handle_request_command(cmd, &mut core_client).await {
                                        log::error!("Error processing request command: {e}");
                                    }
                                }
                            }
                        } else {
                            log::debug!("Command channel closed");
                            break;
                        }
                    }
                    data = hypersync_rx.recv() => {
                        if let Some(msg) = data {
                            let is_block = matches!(&msg, BlockchainMessage::Block(_));
                            let Some(msg) = Self::ready_live_blockchain_message(
                                msg,
                                &core_client.cache,
                                &mut pending_pool_messages,
                            ) else {
                                continue;
                            };

                            let data_event =
                                Self::data_event_from_blockchain_message(msg, &mut core_client)
                                    .await;

                            if let Some(event) = data_event {
                                core_client.send_data(event);
                            }

                            if is_block {
                                for data in Self::drain_pending_pool_messages(
                                    &mut core_client,
                                    &mut pending_pool_messages,
                                )
                                .await
                                {
                                    core_client.send_data(data);
                                }
                            }
                        } else {
                            log::debug!("HyperSync data channel closed");
                            break;
                        }
                    }
                    msg = async {
                        match core_client.rpc_client {
                            Some(ref mut rpc_client) => rpc_client.next_rpc_message().await,
                            None => std::future::pending().await,  // Never resolves
                        }
                    } => {
                        // This branch only fires when we actually receive a message
                        match msg {
                            Ok(msg) => {
                                let is_block = matches!(&msg, BlockchainMessage::Block(_));

                                let Some(msg) = Self::ready_live_blockchain_message(
                                    msg,
                                    &core_client.cache,
                                    &mut pending_pool_messages,
                                ) else {
                                    continue;
                                };

                                if let Some(data) =
                                    Self::data_event_from_blockchain_message(msg, &mut core_client)
                                        .await
                                {
                                    core_client.send_data(data);
                                }

                                if is_block {
                                    for data in Self::drain_pending_pool_messages(
                                        &mut core_client,
                                        &mut pending_pool_messages,
                                    )
                                    .await
                                    {
                                        core_client.send_data(data);
                                    }
                                }
                            }
                            Err(e) => {
                                log::error!("Error processing RPC message: {e}");
                            }
                        }
                    }
                }
            }

            log::debug!("Stopped task 'process'");
        });

        self.process_task = Some(handle);
    }

    async fn drain_pending_pool_messages(
        core_client: &mut BlockchainDataClientCore,
        pending_pool_messages: &mut VecDeque<BlockchainMessage>,
    ) -> Vec<DataEvent> {
        let ready_messages = Self::drain_pending_pool_messages_with_cached_timestamps(
            pending_pool_messages,
            &core_client.cache,
        );
        let mut data_events = Vec::with_capacity(ready_messages.len());

        for msg in ready_messages {
            if let Some(data) = Self::data_event_from_blockchain_message(msg, core_client).await {
                data_events.push(data);
            }
        }

        data_events
    }

    fn drain_pending_pool_messages_with_cached_timestamps(
        pending_pool_messages: &mut VecDeque<BlockchainMessage>,
        cache: &BlockchainCache,
    ) -> Vec<BlockchainMessage> {
        let pending_count = pending_pool_messages.len();
        let mut ready_messages = Vec::new();

        for _ in 0..pending_count {
            let Some(msg) = pending_pool_messages.pop_front() else {
                break;
            };

            if Self::pool_event_missing_block_timestamp(&msg, cache).is_some() {
                pending_pool_messages.push_back(msg);
                continue;
            }

            ready_messages.push(msg);
        }

        ready_messages
    }

    fn queue_pending_pool_message(
        pending_pool_messages: &mut VecDeque<BlockchainMessage>,
        msg: BlockchainMessage,
    ) {
        if pending_pool_messages.len() >= MAX_PENDING_POOL_MESSAGES
            && let Some(dropped_msg) = pending_pool_messages.pop_front()
            && let Some(block_number) = Self::pool_event_block_number(&dropped_msg)
        {
            log::warn!(
                "Dropping oldest live pool event waiting for uncached block {block_number}; pending buffer reached {MAX_PENDING_POOL_MESSAGES} messages"
            );
        }

        pending_pool_messages.push_back(msg);
    }

    fn ready_live_blockchain_message(
        msg: BlockchainMessage,
        cache: &BlockchainCache,
        pending_pool_messages: &mut VecDeque<BlockchainMessage>,
    ) -> Option<BlockchainMessage> {
        if let Some(block_number) = Self::pool_event_missing_block_timestamp(&msg, cache) {
            log::debug!("Deferring live pool event until block {block_number} timestamp is cached");
            Self::queue_pending_pool_message(pending_pool_messages, msg);
            None
        } else {
            Some(msg)
        }
    }

    fn pool_event_missing_block_timestamp(
        msg: &BlockchainMessage,
        cache: &BlockchainCache,
    ) -> Option<u64> {
        let block_number = Self::pool_event_block_number(msg)?;
        cache
            .get_block_timestamp(block_number)
            .is_none()
            .then_some(block_number)
    }

    fn pool_event_block_number(msg: &BlockchainMessage) -> Option<u64> {
        match msg {
            BlockchainMessage::SwapEvent(event) => Some(event.block_number),
            BlockchainMessage::MintEvent(event) => Some(event.block_number),
            BlockchainMessage::BurnEvent(event) => Some(event.block_number),
            BlockchainMessage::CollectEvent(event) => Some(event.block_number),
            BlockchainMessage::FlashEvent(event) => Some(event.block_number),
            BlockchainMessage::FeeProtocolUpdateEvent(event) => Some(event.block_number),
            BlockchainMessage::FeeProtocolCollectEvent(event) => Some(event.block_number),
            BlockchainMessage::Block(_) => None,
        }
    }

    async fn data_event_from_blockchain_message(
        msg: BlockchainMessage,
        core_client: &mut BlockchainDataClientCore,
    ) -> Option<DataEvent> {
        match msg {
            BlockchainMessage::Block(block) => {
                if let Err(e) = core_client.cache.add_block(block.clone()).await {
                    log::error!("Failed to cache block {}: {e}", block.number);
                }

                Some(DataEvent::DeFi(DefiData::Block(block)))
            }
            BlockchainMessage::SwapEvent(swap_event) => {
                match core_client.get_pool(&swap_event.pool_identifier) {
                    Ok(pool) => match core_client.process_pool_swap_event(&swap_event, pool) {
                        Ok(swap) => Some(DataEvent::DeFi(DefiData::PoolSwap(swap))),
                        Err(e) => {
                            log::error!("Error processing pool swap event: {e}");
                            None
                        }
                    },
                    Err(e) => {
                        log::error!(
                            "Failed to get pool {} with error {:?}",
                            swap_event.pool_identifier,
                            e
                        );
                        None
                    }
                }
            }
            BlockchainMessage::BurnEvent(burn_event) => {
                match core_client.get_pool(&burn_event.pool_identifier) {
                    Ok(pool) => {
                        let dex_extended = get_dex_extended(core_client.chain.name, &pool.dex.name)
                            .expect("Failed to get dex extended");

                        match core_client.process_pool_burn_event(&burn_event, pool, dex_extended) {
                            Ok(update) => {
                                Some(DataEvent::DeFi(DefiData::PoolLiquidityUpdate(update)))
                            }
                            Err(e) => {
                                log::error!("Error processing pool burn event: {e}");
                                None
                            }
                        }
                    }
                    Err(e) => {
                        log::error!(
                            "Failed to get pool {} with error {:?}",
                            burn_event.pool_identifier,
                            e
                        );
                        None
                    }
                }
            }
            BlockchainMessage::MintEvent(mint_event) => {
                match core_client.get_pool(&mint_event.pool_identifier) {
                    Ok(pool) => {
                        let dex_extended = get_dex_extended(core_client.chain.name, &pool.dex.name)
                            .expect("Failed to get dex extended");

                        match core_client.process_pool_mint_event(&mint_event, pool, dex_extended) {
                            Ok(update) => {
                                Some(DataEvent::DeFi(DefiData::PoolLiquidityUpdate(update)))
                            }
                            Err(e) => {
                                log::error!("Error processing pool mint event: {e}");
                                None
                            }
                        }
                    }
                    Err(e) => {
                        log::error!(
                            "Failed to get pool {} with error {:?}",
                            mint_event.pool_identifier,
                            e
                        );
                        None
                    }
                }
            }
            BlockchainMessage::CollectEvent(collect_event) => {
                match core_client.get_pool(&collect_event.pool_identifier) {
                    Ok(pool) => {
                        let dex_extended = get_dex_extended(core_client.chain.name, &pool.dex.name)
                            .expect("Failed to get dex extended");

                        match core_client.process_pool_collect_event(
                            &collect_event,
                            pool,
                            dex_extended,
                        ) {
                            Ok(update) => Some(DataEvent::DeFi(DefiData::PoolFeeCollect(update))),
                            Err(e) => {
                                log::error!("Error processing pool collect event: {e}");
                                None
                            }
                        }
                    }
                    Err(e) => {
                        log::error!(
                            "Failed to get pool {} with error {:?}",
                            collect_event.pool_identifier,
                            e
                        );
                        None
                    }
                }
            }
            BlockchainMessage::FlashEvent(flash_event) => {
                match core_client.get_pool(&flash_event.pool_identifier) {
                    Ok(pool) => match core_client.process_pool_flash_event(&flash_event, pool) {
                        Ok(flash) => Some(DataEvent::DeFi(DefiData::PoolFlash(flash))),
                        Err(e) => {
                            log::error!("Error processing pool flash event: {e}");
                            None
                        }
                    },
                    Err(e) => {
                        log::error!(
                            "Failed to get pool {} with error {:?}",
                            flash_event.pool_identifier,
                            e
                        );
                        None
                    }
                }
            }
            BlockchainMessage::FeeProtocolUpdateEvent(update_event) => {
                match core_client.get_pool(&update_event.pool_identifier) {
                    Ok(pool) => match core_client
                        .process_pool_fee_protocol_update_event(&update_event, pool)
                    {
                        Ok(update) => {
                            Some(DataEvent::DeFi(DefiData::PoolFeeProtocolUpdate(update)))
                        }
                        Err(e) => {
                            log::error!("Error processing pool fee-protocol update event: {e}");
                            None
                        }
                    },
                    Err(e) => {
                        log::error!(
                            "Failed to get pool {} with error {:?}",
                            update_event.pool_identifier,
                            e
                        );
                        None
                    }
                }
            }
            BlockchainMessage::FeeProtocolCollectEvent(collect_event) => {
                match core_client.get_pool(&collect_event.pool_identifier) {
                    Ok(pool) => match core_client
                        .process_pool_fee_protocol_collect_event(&collect_event, pool)
                    {
                        Ok(collect) => {
                            Some(DataEvent::DeFi(DefiData::PoolFeeProtocolCollect(collect)))
                        }
                        Err(e) => {
                            log::error!("Error processing pool fee-protocol collect event: {e}");
                            None
                        }
                    },
                    Err(e) => {
                        log::error!(
                            "Failed to get pool {} with error {:?}",
                            collect_event.pool_identifier,
                            e
                        );
                        None
                    }
                }
            }
        }
    }

    /// Processes DeFi subscription commands to start receiving specific blockchain data.
    async fn handle_subscribe_command(
        command: DefiSubscribeCommand,
        core_client: &mut BlockchainDataClientCore,
    ) -> anyhow::Result<()> {
        match command {
            DefiSubscribeCommand::Blocks(_cmd) => {
                log::debug!("Processing subscribe blocks command");

                Self::subscribe_block_feed(core_client, BlockFeedOwner::Explicit).await?;

                Ok(())
            }
            DefiSubscribeCommand::Pool(cmd) => {
                log::debug!(
                    "Processing subscribe pool command for {}",
                    cmd.instrument_id
                );

                if let Ok((_, dex)) = cmd.instrument_id.venue.parse_dex() {
                    let pool_address = validate_address(cmd.instrument_id.symbol.as_str())
                        .map_err(|e| {
                            anyhow::anyhow!(
                                "Invalid pool address '{}' failed with error: {:?}",
                                cmd.instrument_id,
                                e
                            )
                        })?;

                    // Subscribe to all pool event types
                    core_client
                        .subscription_manager
                        .subscribe_swaps(dex, pool_address);
                    core_client
                        .subscription_manager
                        .subscribe_burns(dex, pool_address);
                    core_client
                        .subscription_manager
                        .subscribe_mints(dex, pool_address);
                    core_client
                        .subscription_manager
                        .subscribe_collects(dex, pool_address);
                    core_client
                        .subscription_manager
                        .subscribe_flashes(dex, pool_address);
                    core_client
                        .subscription_manager
                        .subscribe_fee_protocol_updates(dex, pool_address);
                    core_client
                        .subscription_manager
                        .subscribe_fee_protocol_collects(dex, pool_address);
                    Self::update_rpc_pool_event_subscriptions(core_client, dex).await?;
                    Self::update_hypersync_pool_event_stream(core_client, dex).await?;

                    log::debug!(
                        "Subscribed to all pool events for {} at address {}",
                        cmd.instrument_id,
                        pool_address
                    );
                } else {
                    anyhow::bail!(
                        "Invalid venue {}, expected Blockchain DEX format",
                        cmd.instrument_id.venue
                    )
                }

                Ok(())
            }
            DefiSubscribeCommand::PoolSwaps(cmd) => {
                log::debug!(
                    "Processing subscribe pool swaps command for {}",
                    cmd.instrument_id
                );

                if let Ok((_, dex)) = cmd.instrument_id.venue.parse_dex() {
                    let pool_address = validate_address(cmd.instrument_id.symbol.as_str())
                        .map_err(|e| {
                            anyhow::anyhow!(
                                "Invalid pool swap address '{}' failed with error: {:?}",
                                cmd.instrument_id,
                                e
                            )
                        })?;
                    core_client
                        .subscription_manager
                        .subscribe_swaps(dex, pool_address);
                    Self::update_rpc_pool_event_subscriptions(core_client, dex).await?;
                    Self::update_hypersync_pool_event_stream(core_client, dex).await?;
                } else {
                    anyhow::bail!(
                        "Invalid venue {}, expected Blockchain DEX format",
                        cmd.instrument_id.venue
                    )
                }

                Ok(())
            }
            DefiSubscribeCommand::PoolLiquidityUpdates(cmd) => {
                log::debug!(
                    "Processing subscribe pool liquidity updates command for address: {}",
                    cmd.instrument_id
                );

                if let Ok((_, dex)) = cmd.instrument_id.venue.parse_dex() {
                    let pool_address = validate_address(cmd.instrument_id.symbol.as_str())
                        .map_err(|_| {
                            anyhow::anyhow!("Invalid pool swap address: {}", cmd.instrument_id)
                        })?;
                    core_client
                        .subscription_manager
                        .subscribe_burns(dex, pool_address);
                    core_client
                        .subscription_manager
                        .subscribe_mints(dex, pool_address);
                    Self::update_rpc_pool_event_subscriptions(core_client, dex).await?;
                    Self::update_hypersync_pool_event_stream(core_client, dex).await?;
                } else {
                    anyhow::bail!(
                        "Invalid venue {}, expected Blockchain DEX format",
                        cmd.instrument_id.venue
                    )
                }

                Ok(())
            }
            DefiSubscribeCommand::PoolFeeCollects(cmd) => {
                log::debug!(
                    "Processing subscribe pool fee collects command for address: {}",
                    cmd.instrument_id
                );

                if let Ok((_, dex)) = cmd.instrument_id.venue.parse_dex() {
                    let pool_address = validate_address(cmd.instrument_id.symbol.as_str())
                        .map_err(|_| {
                            anyhow::anyhow!(
                                "Invalid pool fee collect address: {}",
                                cmd.instrument_id
                            )
                        })?;
                    core_client
                        .subscription_manager
                        .subscribe_collects(dex, pool_address);
                    Self::update_rpc_pool_event_subscriptions(core_client, dex).await?;
                    Self::update_hypersync_pool_event_stream(core_client, dex).await?;
                } else {
                    anyhow::bail!(
                        "Invalid venue {}, expected Blockchain DEX format",
                        cmd.instrument_id.venue
                    )
                }

                Ok(())
            }
            DefiSubscribeCommand::PoolFlashEvents(cmd) => {
                log::debug!(
                    "Processing subscribe pool flash command for address: {}",
                    cmd.instrument_id
                );

                if let Ok((_, dex)) = cmd.instrument_id.venue.parse_dex() {
                    let pool_address = validate_address(cmd.instrument_id.symbol.as_str())
                        .map_err(|_| {
                            anyhow::anyhow!(
                                "Invalid pool flash subscribe address: {}",
                                cmd.instrument_id
                            )
                        })?;
                    core_client
                        .subscription_manager
                        .subscribe_flashes(dex, pool_address);
                    Self::update_rpc_pool_event_subscriptions(core_client, dex).await?;
                    Self::update_hypersync_pool_event_stream(core_client, dex).await?;
                } else {
                    anyhow::bail!(
                        "Invalid venue {}, expected Blockchain DEX format",
                        cmd.instrument_id.venue
                    )
                }

                Ok(())
            }
        }
    }

    /// Processes DeFi unsubscription commands to stop receiving specific blockchain data.
    async fn handle_unsubscribe_command(
        command: DefiUnsubscribeCommand,
        core_client: &mut BlockchainDataClientCore,
    ) -> anyhow::Result<()> {
        match command {
            DefiUnsubscribeCommand::Blocks(_cmd) => {
                log::debug!("Processing unsubscribe blocks command");

                Self::unsubscribe_block_feed(core_client, BlockFeedOwner::Explicit).await?;

                Ok(())
            }
            DefiUnsubscribeCommand::Pool(cmd) => {
                log::debug!(
                    "Processing unsubscribe pool command for {}",
                    cmd.instrument_id
                );

                if let Ok((_, dex)) = cmd.instrument_id.venue.parse_dex() {
                    let pool_address = validate_address(cmd.instrument_id.symbol.as_str())
                        .map_err(|_| {
                            anyhow::anyhow!("Invalid pool address: {}", cmd.instrument_id)
                        })?;

                    // Unsubscribe from all pool event types
                    core_client
                        .subscription_manager
                        .unsubscribe_swaps(dex, pool_address);
                    core_client
                        .subscription_manager
                        .unsubscribe_burns(dex, pool_address);
                    core_client
                        .subscription_manager
                        .unsubscribe_mints(dex, pool_address);
                    core_client
                        .subscription_manager
                        .unsubscribe_collects(dex, pool_address);
                    core_client
                        .subscription_manager
                        .unsubscribe_flashes(dex, pool_address);
                    core_client
                        .subscription_manager
                        .unsubscribe_fee_protocol_updates(dex, pool_address);
                    core_client
                        .subscription_manager
                        .unsubscribe_fee_protocol_collects(dex, pool_address);
                    Self::update_rpc_pool_event_subscriptions(core_client, dex).await?;
                    Self::update_hypersync_pool_event_stream(core_client, dex).await?;

                    log::debug!(
                        "Unsubscribed from all pool events for {} at address {}",
                        cmd.instrument_id,
                        pool_address
                    );
                } else {
                    anyhow::bail!(
                        "Invalid venue {}, expected Blockchain DEX format",
                        cmd.instrument_id.venue
                    )
                }

                Ok(())
            }
            DefiUnsubscribeCommand::PoolSwaps(cmd) => {
                log::debug!("Processing unsubscribe pool swaps command");

                if let Ok((_, dex)) = cmd.instrument_id.venue.parse_dex() {
                    let pool_address = validate_address(cmd.instrument_id.symbol.as_str())
                        .map_err(|_| {
                            anyhow::anyhow!("Invalid pool swap address: {}", cmd.instrument_id)
                        })?;
                    core_client
                        .subscription_manager
                        .unsubscribe_swaps(dex, pool_address);
                    Self::update_rpc_pool_event_subscriptions(core_client, dex).await?;
                    Self::update_hypersync_pool_event_stream(core_client, dex).await?;
                } else {
                    anyhow::bail!(
                        "Invalid venue {}, expected Blockchain DEX format",
                        cmd.instrument_id.venue
                    )
                }

                Ok(())
            }
            DefiUnsubscribeCommand::PoolLiquidityUpdates(cmd) => {
                log::debug!(
                    "Processing unsubscribe pool liquidity updates command for {}",
                    cmd.instrument_id
                );

                if let Ok((_, dex)) = cmd.instrument_id.venue.parse_dex() {
                    let pool_address = validate_address(cmd.instrument_id.symbol.as_str())
                        .map_err(|_| {
                            anyhow::anyhow!("Invalid pool swap address: {}", cmd.instrument_id)
                        })?;
                    core_client
                        .subscription_manager
                        .unsubscribe_burns(dex, pool_address);
                    core_client
                        .subscription_manager
                        .unsubscribe_mints(dex, pool_address);
                    Self::update_rpc_pool_event_subscriptions(core_client, dex).await?;
                    Self::update_hypersync_pool_event_stream(core_client, dex).await?;
                } else {
                    anyhow::bail!(
                        "Invalid venue {}, expected Blockchain DEX format",
                        cmd.instrument_id.venue
                    )
                }

                Ok(())
            }
            DefiUnsubscribeCommand::PoolFeeCollects(cmd) => {
                log::debug!(
                    "Processing unsubscribe pool fee collects command for {}",
                    cmd.instrument_id
                );

                if let Ok((_, dex)) = cmd.instrument_id.venue.parse_dex() {
                    let pool_address = validate_address(cmd.instrument_id.symbol.as_str())
                        .map_err(|_| {
                            anyhow::anyhow!(
                                "Invalid pool fee collect address: {}",
                                cmd.instrument_id
                            )
                        })?;
                    core_client
                        .subscription_manager
                        .unsubscribe_collects(dex, pool_address);
                    Self::update_rpc_pool_event_subscriptions(core_client, dex).await?;
                    Self::update_hypersync_pool_event_stream(core_client, dex).await?;
                } else {
                    anyhow::bail!(
                        "Invalid venue {}, expected Blockchain DEX format",
                        cmd.instrument_id.venue
                    )
                }

                Ok(())
            }
            DefiUnsubscribeCommand::PoolFlashEvents(cmd) => {
                log::debug!(
                    "Processing unsubscribe pool flash command for {}",
                    cmd.instrument_id
                );

                if let Ok((_, dex)) = cmd.instrument_id.venue.parse_dex() {
                    let pool_address = validate_address(cmd.instrument_id.symbol.as_str())
                        .map_err(|_| {
                            anyhow::anyhow!("Invalid pool flash address: {}", cmd.instrument_id)
                        })?;
                    core_client
                        .subscription_manager
                        .unsubscribe_flashes(dex, pool_address);
                    Self::update_rpc_pool_event_subscriptions(core_client, dex).await?;
                    Self::update_hypersync_pool_event_stream(core_client, dex).await?;
                } else {
                    anyhow::bail!(
                        "Invalid venue {}, expected Blockchain DEX format",
                        cmd.instrument_id.venue
                    )
                }

                Ok(())
            }
        }
    }

    async fn update_rpc_pool_event_subscriptions(
        core_client: &mut BlockchainDataClientCore,
        dex: DexType,
    ) -> anyhow::Result<()> {
        let updates = vec![
            (
                RpcEventType::PoolSwap(dex),
                core_client
                    .subscription_manager
                    .get_subscribed_pool_swap_addresses(&dex),
                core_client
                    .subscription_manager
                    .get_dex_pool_swap_event_signature(&dex),
            ),
            (
                RpcEventType::PoolMint(dex),
                core_client
                    .subscription_manager
                    .get_subscribed_pool_mint_addresses(&dex),
                core_client
                    .subscription_manager
                    .get_dex_pool_mint_event_signature(&dex),
            ),
            (
                RpcEventType::PoolBurn(dex),
                core_client
                    .subscription_manager
                    .get_subscribed_pool_burn_addresses(&dex),
                core_client
                    .subscription_manager
                    .get_dex_pool_burn_event_signature(&dex),
            ),
            (
                RpcEventType::PoolCollect(dex),
                core_client
                    .subscription_manager
                    .get_subscribed_pool_collect_addresses(&dex),
                core_client
                    .subscription_manager
                    .get_dex_pool_collect_event_signature(&dex),
            ),
            (
                RpcEventType::PoolFlash(dex),
                core_client
                    .subscription_manager
                    .get_subscribed_pool_flash_addresses(&dex),
                core_client
                    .subscription_manager
                    .get_dex_pool_flash_event_signature(&dex),
            ),
            (
                RpcEventType::PoolFeeProtocolUpdate(dex),
                core_client
                    .subscription_manager
                    .get_subscribed_pool_fee_protocol_update_addresses(&dex),
                core_client
                    .subscription_manager
                    .get_dex_pool_fee_protocol_update_event_signature(&dex),
            ),
            (
                RpcEventType::PoolFeeProtocolCollect(dex),
                core_client
                    .subscription_manager
                    .get_subscribed_pool_fee_protocol_collect_addresses(&dex),
                core_client
                    .subscription_manager
                    .get_dex_pool_fee_protocol_collect_event_signature(&dex),
            ),
        ];

        let has_pool_event_subscriptions = Self::has_active_pool_event_subscriptions(core_client);

        if core_client.rpc_client.is_none() {
            return Ok(());
        }

        if has_pool_event_subscriptions {
            Self::subscribe_block_feed(core_client, BlockFeedOwner::PoolEvents).await?;
        }

        if let Some(ref mut rpc) = core_client.rpc_client {
            for (event_type, addresses, event_signature) in updates {
                if let Some(event_signature) = event_signature {
                    rpc.subscribe_pool_events(event_type, &addresses, event_signature)
                        .await?;
                }
            }
        }

        if !has_pool_event_subscriptions {
            Self::unsubscribe_block_feed(core_client, BlockFeedOwner::PoolEvents).await?;
        }

        Ok(())
    }

    async fn update_hypersync_pool_event_stream(
        core_client: &mut BlockchainDataClientCore,
        dex: DexType,
    ) -> anyhow::Result<()> {
        if core_client.rpc_client.is_some() {
            return Ok(());
        }

        let addresses = core_client
            .subscription_manager
            .get_subscribed_dex_contract_addresses(&dex);
        let event_signatures = core_client
            .subscription_manager
            .get_active_subscribed_dex_event_signatures(&dex);
        let has_pool_event_subscriptions = Self::has_active_pool_event_subscriptions(core_client);

        if has_pool_event_subscriptions {
            Self::subscribe_block_feed(core_client, BlockFeedOwner::PoolEvents).await?;
        }

        core_client
            .hypersync_client
            .update_dex_event_stream(dex, addresses, event_signatures)
            .await;

        if !has_pool_event_subscriptions {
            Self::unsubscribe_block_feed(core_client, BlockFeedOwner::PoolEvents).await?;
        }

        Ok(())
    }

    fn has_active_pool_event_subscriptions(core_client: &BlockchainDataClientCore) -> bool {
        core_client
            .subscription_manager
            .has_pool_event_subscriptions()
    }

    async fn subscribe_block_feed(
        core_client: &mut BlockchainDataClientCore,
        owner: BlockFeedOwner,
    ) -> anyhow::Result<()> {
        let preferred_backend = if core_client.rpc_client.is_some() {
            BlockFeedBackend::Rpc
        } else {
            BlockFeedBackend::HyperSync
        };
        let Some(backend) = core_client
            .subscription_manager
            .add_block_demand(owner, preferred_backend)
        else {
            return Ok(());
        };

        let started_backend = match backend {
            BlockFeedBackend::Rpc => {
                let Some(rpc) = core_client.rpc_client.as_mut() else {
                    anyhow::bail!("RPC block feed selected without an RPC client")
                };

                match rpc.subscribe_blocks().await {
                    Ok(()) => {
                        log::debug!("Successfully subscribed to blocks via RPC");
                        BlockFeedBackend::Rpc
                    }
                    Err(e) if owner == BlockFeedOwner::Explicit => {
                        log::warn!(
                            "RPC blocks subscription failed: {e}, falling back to HyperSync"
                        );
                        core_client.hypersync_client.subscribe_blocks();
                        tokio::task::yield_now().await;
                        BlockFeedBackend::HyperSync
                    }
                    Err(e) => return Err(e.into()),
                }
            }
            BlockFeedBackend::HyperSync => {
                log::debug!("Subscribing to blocks via HyperSync");
                core_client.hypersync_client.subscribe_blocks();
                tokio::task::yield_now().await;
                BlockFeedBackend::HyperSync
            }
        };

        core_client
            .subscription_manager
            .block_feed_started(started_backend);
        Ok(())
    }

    async fn unsubscribe_block_feed(
        core_client: &mut BlockchainDataClientCore,
        owner: BlockFeedOwner,
    ) -> anyhow::Result<()> {
        let Some(backend) = core_client.subscription_manager.remove_block_demand(owner) else {
            log::debug!("Keeping block subscription active while another owner remains");
            return Ok(());
        };

        match backend {
            BlockFeedBackend::Rpc => {
                let Some(rpc) = core_client.rpc_client.as_mut() else {
                    anyhow::bail!("RPC block feed active without an RPC client")
                };
                rpc.unsubscribe_blocks().await?;
                log::debug!("Unsubscribed from blocks via RPC");
            }
            BlockFeedBackend::HyperSync => {
                core_client.hypersync_client.unsubscribe_blocks().await;
                log::debug!("Unsubscribed from blocks via HyperSync");
            }
        }

        core_client.subscription_manager.block_feed_stopped(backend);
        Ok(())
    }

    /// Processes DeFi request commands to fetch specific blockchain data.
    async fn handle_request_command(
        command: DefiRequestCommand,
        core_client: &mut BlockchainDataClientCore,
    ) -> anyhow::Result<()> {
        match command {
            DefiRequestCommand::PoolSnapshot(cmd) => {
                log::debug!("Processing pool snapshot request for {}", cmd.instrument_id);

                let pool_address =
                    validate_address(cmd.instrument_id.symbol.as_str()).map_err(|e| {
                        anyhow::anyhow!(
                            "Invalid pool address '{}' failed with error: {:?}",
                            cmd.instrument_id,
                            e
                        )
                    })?;

                let pool_identifier =
                    PoolIdentifier::Address(Ustr::from(&pool_address.to_string()));

                match core_client.get_pool(&pool_identifier) {
                    Ok(pool) => {
                        let pool = pool.clone();
                        log::debug!("Found pool for snapshot request: {}", cmd.instrument_id);

                        // Send the pool definition
                        let pool_data = DataEvent::DeFi(DefiData::Pool(pool.as_ref().clone()));
                        core_client.send_data(pool_data);

                        match core_client
                            .bootstrap_latest_pool_profiler(&pool, None)
                            .await
                        {
                            Ok((profiler, already_valid)) => match profiler.extract_snapshot() {
                                Ok(snapshot) => {
                                    log::debug!(
                                        "Saving pool snapshot with {} positions and {} ticks to database...",
                                        snapshot.positions.len(),
                                        snapshot.ticks.len()
                                    );
                                    core_client
                                        .cache
                                        .add_pool_snapshot(
                                            &pool.dex.name,
                                            &pool.pool_identifier,
                                            &snapshot,
                                        )
                                        .await?;

                                    // If the snapshot is usable, send it back to the data engine.
                                    if core_client
                                        .check_snapshot_validity(&profiler, already_valid)
                                        .await?
                                        .is_usable()
                                    {
                                        let snapshot_data =
                                            DataEvent::DeFi(DefiData::PoolSnapshot(snapshot));
                                        core_client.send_data(snapshot_data);
                                    }
                                }
                                Err(e) => log::error!(
                                    "Failed to extract snapshot for {}: {e}",
                                    cmd.instrument_id
                                ),
                            },
                            Err(e) => log::error!(
                                "Failed to bootstrap pool profiler for {} and extract snapshot with error {e}",
                                cmd.instrument_id
                            ),
                        }
                    }
                    Err(e) => {
                        log::warn!("Pool {} not found in cache: {e}", cmd.instrument_id);
                    }
                }

                Ok(())
            }
        }
    }

    /// Waits for the background processing task to complete.
    ///
    /// This method blocks until the spawned process task finishes execution,
    /// which typically happens after a shutdown signal is sent.
    pub async fn await_process_task_close(&mut self) {
        if let Some(handle) = self.process_task.take()
            && let Err(e) = handle.await
        {
            log::error!("Process task join error: {e}");
        }
    }
}

#[async_trait::async_trait(?Send)]
impl DataClient for BlockchainDataClient {
    fn client_id(&self) -> ClientId {
        self.client_id
    }

    fn venue(&self) -> Option<Venue> {
        // Blockchain data clients don't map to a single venue since they can provide
        // data for multiple DEXs across the blockchain
        None
    }

    fn start(&mut self) -> anyhow::Result<()> {
        log::info!(
            "Starting blockchain data client: chain_name={}, dex_ids={:?}, use_hypersync_for_live_data={}, proxy_url={:?}",
            self.chain.name,
            self.config.dex_ids,
            self.config.use_hypersync_for_live_data,
            self.config.proxy_url
        );
        Ok(())
    }

    fn stop(&mut self) -> anyhow::Result<()> {
        log::info!(
            "Stopping blockchain data client for '{chain_name}'",
            chain_name = self.chain.name
        );
        self.cancellation_token.cancel();

        // Create fresh token for next start cycle
        self.cancellation_token = tokio_util::sync::CancellationToken::new();
        Ok(())
    }

    fn reset(&mut self) -> anyhow::Result<()> {
        log::info!(
            "Resetting blockchain data client for '{chain_name}'",
            chain_name = self.chain.name
        );
        self.cancellation_token = tokio_util::sync::CancellationToken::new();
        Ok(())
    }

    fn dispose(&mut self) -> anyhow::Result<()> {
        log::info!(
            "Disposing blockchain data client for '{chain_name}'",
            chain_name = self.chain.name
        );
        Ok(())
    }

    async fn connect(&mut self) -> anyhow::Result<()> {
        log::info!(
            "Connecting blockchain data client for '{}'",
            self.chain.name
        );

        if self.process_task.is_none() {
            self.spawn_process_task();
        }

        Ok(())
    }

    async fn disconnect(&mut self) -> anyhow::Result<()> {
        log::info!(
            "Disconnecting blockchain data client for '{}'",
            self.chain.name
        );

        self.cancellation_token.cancel();
        self.await_process_task_close().await;

        // Create fresh token and channels for next connect cycle
        self.cancellation_token = tokio_util::sync::CancellationToken::new();
        let (hypersync_tx, hypersync_rx) = tokio::sync::mpsc::unbounded_channel();
        self.hypersync_tx = Some(hypersync_tx);
        self.hypersync_rx = Some(hypersync_rx);
        let (command_tx, command_rx) = tokio::sync::mpsc::unbounded_channel();
        self.command_tx = command_tx;
        self.command_rx = Some(command_rx);

        Ok(())
    }

    fn is_connected(&self) -> bool {
        // TODO: Improve connection detection
        // For now, we'll assume connected if we have either RPC or HyperSync configured
        true
    }

    fn is_disconnected(&self) -> bool {
        !self.is_connected()
    }

    fn subscribe_blocks(&mut self, cmd: SubscribeBlocks) -> anyhow::Result<()> {
        let command = DefiDataCommand::Subscribe(DefiSubscribeCommand::Blocks(cmd));
        self.command_tx.send(command)?;
        Ok(())
    }

    fn subscribe_pool(&mut self, cmd: SubscribePool) -> anyhow::Result<()> {
        let command = DefiDataCommand::Subscribe(DefiSubscribeCommand::Pool(cmd));
        self.command_tx.send(command)?;
        Ok(())
    }

    fn subscribe_pool_swaps(&mut self, cmd: SubscribePoolSwaps) -> anyhow::Result<()> {
        let command = DefiDataCommand::Subscribe(DefiSubscribeCommand::PoolSwaps(cmd));
        self.command_tx.send(command)?;
        Ok(())
    }

    fn subscribe_pool_liquidity_updates(
        &mut self,
        cmd: SubscribePoolLiquidityUpdates,
    ) -> anyhow::Result<()> {
        let command = DefiDataCommand::Subscribe(DefiSubscribeCommand::PoolLiquidityUpdates(cmd));
        self.command_tx.send(command)?;
        Ok(())
    }

    fn subscribe_pool_fee_collects(&mut self, cmd: SubscribePoolFeeCollects) -> anyhow::Result<()> {
        let command = DefiDataCommand::Subscribe(DefiSubscribeCommand::PoolFeeCollects(cmd));
        self.command_tx.send(command)?;
        Ok(())
    }

    fn subscribe_pool_flash_events(&mut self, cmd: SubscribePoolFlashEvents) -> anyhow::Result<()> {
        let command = DefiDataCommand::Subscribe(DefiSubscribeCommand::PoolFlashEvents(cmd));
        self.command_tx.send(command)?;
        Ok(())
    }

    fn unsubscribe_blocks(&mut self, cmd: &UnsubscribeBlocks) -> anyhow::Result<()> {
        let command = DefiDataCommand::Unsubscribe(DefiUnsubscribeCommand::Blocks(cmd.clone()));
        self.command_tx.send(command)?;
        Ok(())
    }

    fn unsubscribe_pool(&mut self, cmd: &UnsubscribePool) -> anyhow::Result<()> {
        let command = DefiDataCommand::Unsubscribe(DefiUnsubscribeCommand::Pool(cmd.clone()));
        self.command_tx.send(command)?;
        Ok(())
    }

    fn unsubscribe_pool_swaps(&mut self, cmd: &UnsubscribePoolSwaps) -> anyhow::Result<()> {
        let command = DefiDataCommand::Unsubscribe(DefiUnsubscribeCommand::PoolSwaps(cmd.clone()));
        self.command_tx.send(command)?;
        Ok(())
    }

    fn unsubscribe_pool_liquidity_updates(
        &mut self,
        cmd: &UnsubscribePoolLiquidityUpdates,
    ) -> anyhow::Result<()> {
        let command =
            DefiDataCommand::Unsubscribe(DefiUnsubscribeCommand::PoolLiquidityUpdates(cmd.clone()));
        self.command_tx.send(command)?;
        Ok(())
    }

    fn unsubscribe_pool_fee_collects(
        &mut self,
        cmd: &UnsubscribePoolFeeCollects,
    ) -> anyhow::Result<()> {
        let command =
            DefiDataCommand::Unsubscribe(DefiUnsubscribeCommand::PoolFeeCollects(cmd.clone()));
        self.command_tx.send(command)?;
        Ok(())
    }

    fn unsubscribe_pool_flash_events(
        &mut self,
        cmd: &UnsubscribePoolFlashEvents,
    ) -> anyhow::Result<()> {
        let command =
            DefiDataCommand::Unsubscribe(DefiUnsubscribeCommand::PoolFlashEvents(cmd.clone()));
        self.command_tx.send(command)?;
        Ok(())
    }

    fn request_pool_snapshot(&self, cmd: RequestPoolSnapshot) -> anyhow::Result<()> {
        let command = DefiDataCommand::Request(DefiRequestCommand::PoolSnapshot(cmd));
        self.command_tx.send(command)?;
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use std::{sync::Arc, time::Duration};

    use alloy::primitives::{I256, U160, U256, address};
    use nautilus_common::defi::RequestPoolSnapshot;
    use nautilus_core::{UUID4, UnixNanos};
    use nautilus_model::{
        defi::{Block, Blockchain, Chain, DexType, Pool, PoolIdentifier, Token},
        identifiers::{ClientId, InstrumentId},
    };
    use rstest::rstest;
    use tokio_util::sync::CancellationToken;

    use super::*;
    use crate::events::{flash::FlashEvent, swap::SwapEvent};

    const WETH_USDT_CREATION_BLOCK: u64 = 12_375_326;

    #[tokio::test(flavor = "multi_thread")]
    #[ignore = "requires ENVIO_API_TOKEN and live HyperSync access"]
    async fn pool_snapshot_request_does_not_emit_snapshot_when_bootstrap_fails() {
        std::env::var("ENVIO_API_TOKEN").expect("ENVIO_API_TOKEN must be set");

        let pool = weth_usdt_pool();
        let instrument_id = pool.instrument_id;
        let (hypersync_tx, _hypersync_rx) = tokio::sync::mpsc::unbounded_channel();
        let (data_tx, mut data_rx) = tokio::sync::mpsc::unbounded_channel();
        let config = BlockchainDataClientConfig::builder()
            .chain(pool.chain.clone())
            .dex_ids(vec![DexType::UniswapV3])
            .http_rpc_url("http://127.0.0.1:9".to_string())
            .use_hypersync_for_live_data(true)
            .maybe_from_block(Some(WETH_USDT_CREATION_BLOCK))
            .build();
        let mut core = BlockchainDataClientCore::new(
            config,
            Some(hypersync_tx),
            Some(data_tx),
            CancellationToken::new(),
        );
        core.cache
            .add_pool(pool.as_ref().clone())
            .await
            .expect("Pool should be added to in-memory cache");

        let request = RequestPoolSnapshot::new(
            instrument_id,
            Some(ClientId::new("BLOCKCHAIN")),
            UUID4::new(),
            UnixNanos::default(),
            None,
        );

        BlockchainDataClient::handle_request_command(
            DefiRequestCommand::PoolSnapshot(request),
            &mut core,
        )
        .await
        .expect("Bootstrap failure should not fail the request handler");

        let mut events = Vec::new();
        while let Ok(event) = data_rx.try_recv() {
            events.push(event);
        }

        assert_eq!(events.len(), 1);
        match &events[0] {
            DataEvent::DeFi(DefiData::Pool(pool)) => {
                assert_eq!(pool.instrument_id, instrument_id);
            }
            _ => panic!("expected only the pool definition event"),
        }
        assert!(
            events
                .iter()
                .all(|event| !matches!(event, DataEvent::DeFi(DefiData::PoolSnapshot(_))))
        );
    }

    #[rstest]
    fn pool_event_missing_block_timestamp_clears_after_timestamp_cache_update() {
        let chain = Arc::new(
            Chain::from_chain_id(1)
                .expect("Ethereum chain should exist")
                .clone(),
        );
        let mut cache = BlockchainCache::new(chain);
        let msg = flash_message(42);

        assert_eq!(
            BlockchainDataClient::pool_event_block_number(&msg),
            Some(42)
        );
        assert_eq!(
            BlockchainDataClient::pool_event_missing_block_timestamp(&msg, &cache),
            Some(42)
        );

        cache.cache_block_timestamp(42, UnixNanos::from(1_700_000_000_000_000_000));

        assert_eq!(
            BlockchainDataClient::pool_event_missing_block_timestamp(&msg, &cache),
            None
        );
    }

    #[rstest]
    fn queue_pending_pool_message_drops_oldest_entry_at_cap() {
        let mut pending = VecDeque::new();

        for block_number in 0..MAX_PENDING_POOL_MESSAGES {
            BlockchainDataClient::queue_pending_pool_message(
                &mut pending,
                swap_message(block_number as u64),
            );
        }

        BlockchainDataClient::queue_pending_pool_message(
            &mut pending,
            swap_message(MAX_PENDING_POOL_MESSAGES as u64),
        );

        assert_eq!(pending.len(), MAX_PENDING_POOL_MESSAGES);
        assert_eq!(
            BlockchainDataClient::pool_event_block_number(pending.front().unwrap()),
            Some(1)
        );
        assert_eq!(
            BlockchainDataClient::pool_event_block_number(pending.back().unwrap()),
            Some(MAX_PENDING_POOL_MESSAGES as u64)
        );
    }

    #[rstest]
    fn drain_pending_pool_messages_releases_events_after_timestamps_are_cached() {
        let chain = Arc::new(
            Chain::from_chain_id(1)
                .expect("Ethereum chain should exist")
                .clone(),
        );
        let mut cache = BlockchainCache::new(chain);
        cache.cache_block_timestamp(42, UnixNanos::from(1_700_000_000_000_000_000));
        let mut pending = VecDeque::from([flash_message(41), flash_message(42)]);

        let ready_messages =
            BlockchainDataClient::drain_pending_pool_messages_with_cached_timestamps(
                &mut pending,
                &cache,
            );

        assert_eq!(ready_messages.len(), 1);
        assert_eq!(
            BlockchainDataClient::pool_event_block_number(&ready_messages[0]),
            Some(42)
        );
        assert_eq!(pending.len(), 1);
        assert_eq!(
            BlockchainDataClient::pool_event_block_number(pending.front().unwrap()),
            Some(41)
        );

        cache.cache_block_timestamp(41, UnixNanos::from(1_700_000_000_000_000_001));

        let ready_messages =
            BlockchainDataClient::drain_pending_pool_messages_with_cached_timestamps(
                &mut pending,
                &cache,
            );

        assert_eq!(ready_messages.len(), 1);
        assert_eq!(
            BlockchainDataClient::pool_event_block_number(&ready_messages[0]),
            Some(41)
        );
        assert!(pending.is_empty());
    }

    #[rstest]
    fn ready_live_blockchain_message_queues_until_block_timestamp_is_cached() {
        let chain = Arc::new(
            Chain::from_chain_id(1)
                .expect("Ethereum chain should exist")
                .clone(),
        );
        let mut cache = BlockchainCache::new(chain);
        let mut pending = VecDeque::new();

        let message = BlockchainDataClient::ready_live_blockchain_message(
            flash_message(42),
            &cache,
            &mut pending,
        );

        assert!(message.is_none());
        assert_eq!(pending.len(), 1);
        assert_eq!(
            BlockchainDataClient::pool_event_block_number(pending.front().unwrap()),
            Some(42)
        );

        cache.cache_block_timestamp(42, UnixNanos::from(1_700_000_000_000_000_000));
        let block = Block::new(
            "0x1".to_string(),
            "0x0".to_string(),
            42,
            Ustr::from("0x0000000000000000000000000000000000000000"),
            30_000_000,
            21_000,
            UnixNanos::from(1_700_000_000_000_000_000),
            Some(Blockchain::Ethereum),
        );

        let message = BlockchainDataClient::ready_live_blockchain_message(
            BlockchainMessage::Block(block),
            &cache,
            &mut pending,
        );
        let ready_messages =
            BlockchainDataClient::drain_pending_pool_messages_with_cached_timestamps(
                &mut pending,
                &cache,
            );

        assert!(matches!(message, Some(BlockchainMessage::Block(_))));
        assert_eq!(ready_messages.len(), 1);
        assert_eq!(
            BlockchainDataClient::pool_event_block_number(&ready_messages[0]),
            Some(42)
        );
        assert!(pending.is_empty());
    }

    #[tokio::test(flavor = "multi_thread")]
    #[ignore = "requires ENVIO_API_TOKEN and live HyperSync access"]
    async fn live_hypersync_pool_swap_subscription_receives_tip_event_and_unsubscribes() {
        std::env::var("ENVIO_API_TOKEN").expect("ENVIO_API_TOKEN must be set");

        let chain = Arc::new(
            Chain::from_chain_id(42161)
                .expect("Arbitrum chain should exist")
                .clone(),
        );
        let dex_extended = get_dex_extended(chain.name, &DexType::UniswapV3)
            .expect("Arbitrum UniswapV3 should be registered");
        let pool_address = address!("C31E54c7A869B9FcBEcc14363CF510d1c41fa443");
        let instrument_id_value = format!("{}.Arbitrum:UniswapV3", pool_address.to_checksum(None));
        let instrument_id = InstrumentId::from(instrument_id_value.as_str());
        let expected_pool_id = PoolIdentifier::from_address(pool_address);
        let (hypersync_tx, mut hypersync_rx) = tokio::sync::mpsc::unbounded_channel();
        let config = BlockchainDataClientConfig::builder()
            .chain(chain)
            .dex_ids(vec![DexType::UniswapV3])
            .http_rpc_url("http://127.0.0.1:9".to_string())
            .use_hypersync_for_live_data(true)
            .build();
        let mut core = BlockchainDataClientCore::new(
            config,
            Some(hypersync_tx),
            None,
            CancellationToken::new(),
        );
        core.cache
            .add_dex(dex_extended.dex.clone())
            .await
            .expect("DEX should be added to in-memory cache");
        core.subscription_manager.register_dex_for_subscriptions(
            DexType::UniswapV3,
            dex_extended.swap_created_event.as_ref(),
            dex_extended.mint_created_event.as_ref(),
            dex_extended.burn_created_event.as_ref(),
            dex_extended.collect_created_event.as_ref(),
            dex_extended.flash_created_event.as_deref(),
        );
        core.subscription_manager.register_dex_fee_protocol_events(
            DexType::UniswapV3,
            dex_extended.fee_protocol_update_event.as_deref(),
            dex_extended.fee_protocol_collect_event.as_deref(),
        );

        BlockchainDataClient::handle_subscribe_command(
            DefiSubscribeCommand::PoolSwaps(SubscribePoolSwaps::new(
                instrument_id,
                Some(ClientId::new("BLOCKCHAIN")),
                UUID4::new(),
                UnixNanos::default(),
                None,
            )),
            &mut core,
        )
        .await
        .expect("live HyperSync pool swap subscribe should succeed");

        let event = tokio::time::timeout(Duration::from_secs(240), async {
            loop {
                let msg = hypersync_rx
                    .recv()
                    .await
                    .expect("HyperSync live stream channel should stay open");

                if let BlockchainMessage::SwapEvent(event) = msg
                    && event.pool_identifier == expected_pool_id
                {
                    break event;
                }
            }
        })
        .await
        .expect("expected a live Arbitrum UniswapV3 swap within 240s");

        BlockchainDataClient::handle_unsubscribe_command(
            DefiUnsubscribeCommand::PoolSwaps(UnsubscribePoolSwaps::new(
                instrument_id,
                Some(ClientId::new("BLOCKCHAIN")),
                UUID4::new(),
                UnixNanos::default(),
                None,
            )),
            &mut core,
        )
        .await
        .expect("live HyperSync pool swap unsubscribe should succeed");
        core.disconnect().await;

        assert_eq!(event.pool_identifier, expected_pool_id);
        assert!(event.block_number > 0);
    }

    fn swap_message(block_number: u64) -> BlockchainMessage {
        let pool = weth_usdt_pool();
        let address = address!("1111111111111111111111111111111111111111");

        BlockchainMessage::SwapEvent(SwapEvent::new(
            pool.dex.clone(),
            pool.pool_identifier,
            block_number,
            "0x1".to_string(),
            0,
            0,
            address,
            address,
            I256::ZERO,
            I256::ZERO,
            U160::ZERO,
            0,
            0,
        ))
    }

    fn flash_message(block_number: u64) -> BlockchainMessage {
        let pool = weth_usdt_pool();
        let address = address!("1111111111111111111111111111111111111111");

        BlockchainMessage::FlashEvent(FlashEvent::new(
            pool.dex.clone(),
            pool.pool_identifier,
            block_number,
            "0x1".to_string(),
            0,
            0,
            address,
            address,
            U256::ZERO,
            U256::ZERO,
            U256::ZERO,
            U256::ZERO,
        ))
    }

    fn weth_usdt_pool() -> Arc<Pool> {
        let chain = Arc::new(
            Chain::from_chain_id(1)
                .expect("Ethereum chain should exist")
                .clone(),
        );
        let dex = get_dex_extended(chain.name, &DexType::UniswapV3)
            .expect("Ethereum UniswapV3 should be registered")
            .dex
            .clone();
        let pool_address = address!("4e68ccd3e89f51c3074ca5072bbac773960dfa36");
        let token0 = Token::new(
            chain.clone(),
            address!("C02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2"),
            "Wrapped Ether".to_string(),
            "WETH".to_string(),
            18,
        );
        let token1 = Token::new(
            chain.clone(),
            address!("dAC17F958D2ee523a2206206994597C13D831ec7"),
            "Tether USD".to_string(),
            "USDT".to_string(),
            6,
        );

        Arc::new(Pool::new(
            chain,
            dex,
            pool_address,
            PoolIdentifier::from_address(pool_address),
            WETH_USDT_CREATION_BLOCK,
            token0,
            token1,
            Some(3_000),
            Some(60),
            UnixNanos::default(),
        ))
    }
}