nautilus_blockchain/data/
client.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16use std::{collections::VecDeque, sync::Arc, time::Duration};
17
18use anyhow::Context;
19use nautilus_common::{
20    clients::DataClient,
21    defi::RequestPoolSnapshot,
22    messages::{
23        DataEvent,
24        defi::{
25            DefiDataCommand, DefiRequestCommand, DefiSubscribeCommand, DefiUnsubscribeCommand,
26            PoolSnapshotResponse, SubscribeBlocks, SubscribePool, SubscribePoolFeeCollects,
27            SubscribePoolFlashEvents, SubscribePoolLiquidityUpdates, SubscribePoolSwaps,
28            UnsubscribeBlocks, UnsubscribePool, UnsubscribePoolFeeCollects,
29            UnsubscribePoolFlashEvents, UnsubscribePoolLiquidityUpdates, UnsubscribePoolSwaps,
30        },
31    },
32};
33use nautilus_live::{
34    SocketControlFactory,
35    task::{TaskGroup, TaskGroupGuard},
36};
37use nautilus_model::{
38    defi::{DefiData, DexType, PoolIdentifier, SharedChain, validation::validate_address},
39    identifiers::{ClientId, Venue},
40};
41use ustr::Ustr;
42
43use crate::{
44    cache::BlockchainCache,
45    config::BlockchainDataClientConfig,
46    data::{
47        core::BlockchainDataClientCore,
48        subscription::{BlockFeedBackend, BlockFeedOwner},
49    },
50    exchanges::get_dex_extended,
51    rpc::{
52        BlockchainRpcClient,
53        types::{BlockchainMessage, RpcEventType},
54    },
55};
56
57const MAX_PENDING_POOL_MESSAGES: usize = 10_000;
58
59/// A client for interacting with blockchain data from multiple sources.
60///
61/// The `BlockchainDataClient` serves as a facade that coordinates between different blockchain
62/// data providers, caching mechanisms, and contract interactions. It provides a unified interface
63/// for retrieving and processing blockchain data, particularly focused on DeFi protocols.
64///
65/// This client supports two primary data sources:
66/// 1. Direct RPC connections to blockchain nodes (via WebSocket).
67/// 2. HyperSync API for efficient historical data queries.
68#[derive(Debug)]
69pub struct BlockchainDataClient {
70    /// The client ID used to identify this client with the data engine.
71    pub client_id: ClientId,
72    /// The blockchain being targeted by this client instance.
73    pub chain: SharedChain,
74    /// Configuration parameters for the blockchain data client.
75    pub config: BlockchainDataClientConfig,
76    /// The core client instance that handles blockchain operations.
77    /// Wrapped in Option to allow moving it into the background processing task.
78    pub core_client: Option<BlockchainDataClientCore>,
79    socket_factory: SocketControlFactory,
80    hypersync_rx: Option<tokio::sync::mpsc::UnboundedReceiver<BlockchainMessage>>,
81    hypersync_tx: Option<tokio::sync::mpsc::UnboundedSender<BlockchainMessage>>,
82    command_tx: tokio::sync::mpsc::UnboundedSender<DefiDataCommand>,
83    command_rx: Option<tokio::sync::mpsc::UnboundedReceiver<DefiDataCommand>>,
84    session_tasks: TaskGroup,
85    cancellation_token: tokio_util::sync::CancellationToken,
86}
87
88impl BlockchainDataClient {
89    /// Creates a new [`BlockchainDataClient`] instance for the specified configuration.
90    ///
91    /// # Errors
92    ///
93    /// Returns an error if DEX selections contain duplicates.
94    pub fn new(client_id: ClientId, config: BlockchainDataClientConfig) -> anyhow::Result<Self> {
95        config.validate()?;
96
97        let chain = Arc::clone(&config.chain);
98        let (command_tx, command_rx) = tokio::sync::mpsc::unbounded_channel();
99        let (hypersync_tx, hypersync_rx) = tokio::sync::mpsc::unbounded_channel();
100        let socket_factory = SocketControlFactory::new(client_id, None);
101        let session_tasks = TaskGroup::new();
102        Ok(Self {
103            client_id,
104            chain,
105            core_client: None,
106            socket_factory,
107            config,
108            hypersync_rx: Some(hypersync_rx),
109            hypersync_tx: Some(hypersync_tx),
110            command_tx,
111            command_rx: Some(command_rx),
112            cancellation_token: session_tasks.cancellation_token(),
113            session_tasks,
114        })
115    }
116
117    /// Spawns the main processing task that handles commands and blockchain data.
118    ///
119    /// This method creates a background task that:
120    /// 1. Processes subscription/unsubscription commands from the command channel
121    /// 2. Handles incoming blockchain data from HyperSync
122    /// 3. Processes RPC messages if RPC client is configured
123    /// 4. Routes processed data to subscribers
124    fn spawn_process_task(
125        &mut self,
126    ) -> anyhow::Result<tokio::sync::oneshot::Receiver<anyhow::Result<()>>> {
127        anyhow::ensure!(self.command_rx.is_some(), "Command receiver already taken");
128
129        let cancellation_token = self.cancellation_token.clone();
130
131        let data_tx = nautilus_common::live::runner::get_data_event_sender();
132
133        let hypersync_tx = self.hypersync_tx.clone();
134
135        let mut core_client = BlockchainDataClientCore::new(
136            self.config.clone(),
137            hypersync_tx,
138            Some(data_tx),
139            cancellation_token.clone(),
140        )?;
141        let command_rx = self
142            .command_rx
143            .take()
144            .expect("command receiver was checked");
145        let mut hypersync_rx = self
146            .hypersync_rx
147            .take()
148            .expect("HyperSync receiver is available before startup");
149        self.hypersync_tx = None;
150        core_client.set_socket_control(self.socket_factory.control("blockchain-rpc"));
151
152        let (startup_tx, startup_rx) = tokio::sync::oneshot::channel();
153        let future = async move {
154            log::debug!("Started task 'process'");
155
156            if let Err(e) = core_client.connect().await {
157                let _ = startup_tx.send(Err(e));
158                return;
159            }
160            let _ = startup_tx.send(Ok(()));
161
162            let mut command_rx = command_rx;
163            let mut pending_pool_messages = VecDeque::new();
164
165            loop {
166                tokio::select! {
167                    () = cancellation_token.cancelled() => {
168                        log::debug!("Received cancellation signal in Blockchain data client process task");
169                        core_client.disconnect().await;
170                        break;
171                    }
172                    command = command_rx.recv() => {
173                        if let Some(cmd) = command {
174                            match cmd {
175                                DefiDataCommand::Subscribe(cmd) => {
176                                    let chain = cmd.blockchain();
177                                    if chain != core_client.chain.name {
178                                        log::error!("Incorrect blockchain for subscribe command: {chain}");
179                                        continue;
180                                    }
181
182                                      if let Err(e) = Self::handle_subscribe_command(cmd, &mut core_client).await{
183                                        log::error!("Error processing subscribe command: {e}");
184                                    }
185                                }
186                                DefiDataCommand::Unsubscribe(cmd) => {
187                                    let chain = cmd.blockchain();
188                                    if chain != core_client.chain.name {
189                                        log::error!("Incorrect blockchain for subscribe command: {chain}");
190                                        continue;
191                                    }
192
193                                    if let Err(e) = Self::handle_unsubscribe_command(cmd, &mut core_client).await{
194                                        log::error!("Error processing subscribe command: {e}");
195                                    }
196                                }
197                                DefiDataCommand::Request(cmd) => {
198                                    if let Err(e) = Self::handle_request_command(cmd, &mut core_client).await {
199                                        log::error!("Error processing request command: {e}");
200                                    }
201                                }
202                            }
203                        } else {
204                            log::debug!("Command channel closed");
205                            break;
206                        }
207                    }
208                    data = hypersync_rx.recv() => {
209                        if let Some(msg) = data {
210                            Self::process_live_blockchain_message(
211                                msg,
212                                &mut core_client,
213                                &mut pending_pool_messages,
214                            )
215                            .await;
216                        } else {
217                            log::debug!("HyperSync data channel closed");
218                            break;
219                        }
220                    }
221                    msg = async {
222                        match core_client.rpc_client {
223                            Some(ref mut rpc_client) => rpc_client.next_rpc_message().await,
224                            None => std::future::pending().await,  // Never resolves
225                        }
226                    } => {
227                        // This branch only fires when we actually receive a message
228                        match msg {
229                            Ok(msg) => {
230                                Self::process_live_blockchain_message(
231                                    msg,
232                                    &mut core_client,
233                                    &mut pending_pool_messages,
234                                )
235                                .await;
236                            }
237                            Err(e) => {
238                                log::error!("Error processing RPC message: {e}");
239                            }
240                        }
241                    }
242                }
243            }
244
245            log::debug!("Stopped task 'process'");
246        };
247
248        self.session_tasks
249            .spawn(future)
250            .map_err(|e| anyhow::anyhow!("Failed to register blockchain process task: {e}"))?;
251        Ok(startup_rx)
252    }
253
254    async fn process_live_blockchain_message(
255        msg: BlockchainMessage,
256        core_client: &mut BlockchainDataClientCore,
257        pending_pool_messages: &mut VecDeque<BlockchainMessage>,
258    ) {
259        let is_block = matches!(&msg, BlockchainMessage::Block(_));
260        let Some(msg) =
261            Self::ready_live_blockchain_message(msg, &core_client.cache, pending_pool_messages)
262        else {
263            return;
264        };
265
266        if let Some(data) = Self::data_event_from_blockchain_message(msg, core_client).await {
267            core_client.send_data(data);
268        }
269
270        if is_block {
271            for data in Self::drain_pending_pool_messages(core_client, pending_pool_messages).await
272            {
273                core_client.send_data(data);
274            }
275        }
276    }
277
278    async fn drain_pending_pool_messages(
279        core_client: &mut BlockchainDataClientCore,
280        pending_pool_messages: &mut VecDeque<BlockchainMessage>,
281    ) -> Vec<DataEvent> {
282        let ready_messages = Self::drain_pending_pool_messages_with_cached_metadata(
283            pending_pool_messages,
284            &core_client.cache,
285        );
286        let mut data_events = Vec::with_capacity(ready_messages.len());
287
288        for msg in ready_messages {
289            if let Some(data) = Self::data_event_from_blockchain_message(msg, core_client).await {
290                data_events.push(data);
291            }
292        }
293
294        data_events
295    }
296
297    fn drain_pending_pool_messages_with_cached_metadata(
298        pending_pool_messages: &mut VecDeque<BlockchainMessage>,
299        cache: &BlockchainCache,
300    ) -> Vec<BlockchainMessage> {
301        let pending_count = pending_pool_messages.len();
302        let mut ready_messages = Vec::new();
303
304        for _ in 0..pending_count {
305            let Some(msg) = pending_pool_messages.pop_front() else {
306                break;
307            };
308
309            if Self::pool_event_missing_block_metadata(&msg, cache).is_some() {
310                pending_pool_messages.push_back(msg);
311                continue;
312            }
313
314            ready_messages.push(msg);
315        }
316
317        ready_messages
318    }
319
320    fn queue_pending_pool_message(
321        pending_pool_messages: &mut VecDeque<BlockchainMessage>,
322        msg: BlockchainMessage,
323    ) {
324        if pending_pool_messages.len() >= MAX_PENDING_POOL_MESSAGES
325            && let Some(dropped_msg) = pending_pool_messages.pop_front()
326            && let Some(block_number) = Self::pool_event_block_number(&dropped_msg)
327        {
328            log::warn!(
329                "Dropping oldest live pool event waiting for uncached block {block_number}; pending buffer reached {MAX_PENDING_POOL_MESSAGES} messages"
330            );
331        }
332
333        pending_pool_messages.push_back(msg);
334    }
335
336    fn ready_live_blockchain_message(
337        msg: BlockchainMessage,
338        cache: &BlockchainCache,
339        pending_pool_messages: &mut VecDeque<BlockchainMessage>,
340    ) -> Option<BlockchainMessage> {
341        if let Some(block_number) = Self::pool_event_missing_block_metadata(&msg, cache) {
342            log::debug!("Deferring live pool event until block {block_number} metadata is cached");
343            Self::queue_pending_pool_message(pending_pool_messages, msg);
344            None
345        } else {
346            Some(msg)
347        }
348    }
349
350    fn pool_event_missing_block_metadata(
351        msg: &BlockchainMessage,
352        cache: &BlockchainCache,
353    ) -> Option<u64> {
354        let block_number = Self::pool_event_block_number(msg)?;
355        (cache.get_block_timestamp(block_number).is_none()
356            || cache.get_block_hash(block_number).is_none())
357        .then_some(block_number)
358    }
359
360    fn pool_event_block_number(msg: &BlockchainMessage) -> Option<u64> {
361        match msg {
362            BlockchainMessage::SwapEvent(event) => Some(event.block_number),
363            BlockchainMessage::MintEvent(event) => Some(event.block_number),
364            BlockchainMessage::BurnEvent(event) => Some(event.block_number),
365            BlockchainMessage::CollectEvent(event) => Some(event.block_number),
366            BlockchainMessage::FlashEvent(event) => Some(event.block_number),
367            BlockchainMessage::FeeProtocolUpdateEvent(event) => Some(event.block_number),
368            BlockchainMessage::FeeProtocolCollectEvent(event) => Some(event.block_number),
369            BlockchainMessage::Block(_) => None,
370        }
371    }
372
373    async fn data_event_from_blockchain_message(
374        msg: BlockchainMessage,
375        core_client: &mut BlockchainDataClientCore,
376    ) -> Option<DataEvent> {
377        match msg {
378            BlockchainMessage::Block(block) => {
379                if let Err(e) = core_client.cache.add_block(block.clone()).await {
380                    log::error!("Failed to cache block {}: {e}", block.number);
381                }
382
383                Some(DataEvent::DeFi(DefiData::Block(block)))
384            }
385            BlockchainMessage::SwapEvent(swap_event) => {
386                match core_client.get_pool(&swap_event.pool_identifier) {
387                    Ok(pool) => match core_client.process_pool_swap_event(&swap_event, pool) {
388                        Ok(swap) => Some(DataEvent::DeFi(DefiData::PoolSwap(swap))),
389                        Err(e) => {
390                            log::error!("Error processing pool swap event: {e}");
391                            None
392                        }
393                    },
394                    Err(e) => {
395                        log::error!(
396                            "Failed to get pool {} with error {:?}",
397                            swap_event.pool_identifier,
398                            e
399                        );
400                        None
401                    }
402                }
403            }
404            BlockchainMessage::BurnEvent(burn_event) => {
405                match core_client.get_pool(&burn_event.pool_identifier) {
406                    Ok(pool) => {
407                        let dex_extended = get_dex_extended(core_client.chain.name, &pool.dex.name)
408                            .expect("Failed to get dex extended");
409
410                        match core_client.process_pool_burn_event(&burn_event, pool, dex_extended) {
411                            Ok(update) => {
412                                Some(DataEvent::DeFi(DefiData::PoolLiquidityUpdate(update)))
413                            }
414                            Err(e) => {
415                                log::error!("Error processing pool burn event: {e}");
416                                None
417                            }
418                        }
419                    }
420                    Err(e) => {
421                        log::error!(
422                            "Failed to get pool {} with error {:?}",
423                            burn_event.pool_identifier,
424                            e
425                        );
426                        None
427                    }
428                }
429            }
430            BlockchainMessage::MintEvent(mint_event) => {
431                match core_client.get_pool(&mint_event.pool_identifier) {
432                    Ok(pool) => {
433                        let dex_extended = get_dex_extended(core_client.chain.name, &pool.dex.name)
434                            .expect("Failed to get dex extended");
435
436                        match core_client.process_pool_mint_event(&mint_event, pool, dex_extended) {
437                            Ok(update) => {
438                                Some(DataEvent::DeFi(DefiData::PoolLiquidityUpdate(update)))
439                            }
440                            Err(e) => {
441                                log::error!("Error processing pool mint event: {e}");
442                                None
443                            }
444                        }
445                    }
446                    Err(e) => {
447                        log::error!(
448                            "Failed to get pool {} with error {:?}",
449                            mint_event.pool_identifier,
450                            e
451                        );
452                        None
453                    }
454                }
455            }
456            BlockchainMessage::CollectEvent(collect_event) => {
457                match core_client.get_pool(&collect_event.pool_identifier) {
458                    Ok(pool) => {
459                        let dex_extended = get_dex_extended(core_client.chain.name, &pool.dex.name)
460                            .expect("Failed to get dex extended");
461
462                        match core_client.process_pool_collect_event(
463                            &collect_event,
464                            pool,
465                            dex_extended,
466                        ) {
467                            Ok(update) => Some(DataEvent::DeFi(DefiData::PoolFeeCollect(update))),
468                            Err(e) => {
469                                log::error!("Error processing pool collect event: {e}");
470                                None
471                            }
472                        }
473                    }
474                    Err(e) => {
475                        log::error!(
476                            "Failed to get pool {} with error {:?}",
477                            collect_event.pool_identifier,
478                            e
479                        );
480                        None
481                    }
482                }
483            }
484            BlockchainMessage::FlashEvent(flash_event) => {
485                match core_client.get_pool(&flash_event.pool_identifier) {
486                    Ok(pool) => match core_client.process_pool_flash_event(&flash_event, pool) {
487                        Ok(flash) => Some(DataEvent::DeFi(DefiData::PoolFlash(flash))),
488                        Err(e) => {
489                            log::error!("Error processing pool flash event: {e}");
490                            None
491                        }
492                    },
493                    Err(e) => {
494                        log::error!(
495                            "Failed to get pool {} with error {:?}",
496                            flash_event.pool_identifier,
497                            e
498                        );
499                        None
500                    }
501                }
502            }
503            BlockchainMessage::FeeProtocolUpdateEvent(update_event) => {
504                match core_client.get_pool(&update_event.pool_identifier) {
505                    Ok(pool) => match core_client
506                        .process_pool_fee_protocol_update_event(&update_event, pool)
507                    {
508                        Ok(update) => {
509                            Some(DataEvent::DeFi(DefiData::PoolFeeProtocolUpdate(update)))
510                        }
511                        Err(e) => {
512                            log::error!("Error processing pool fee-protocol update event: {e}");
513                            None
514                        }
515                    },
516                    Err(e) => {
517                        log::error!(
518                            "Failed to get pool {} with error {:?}",
519                            update_event.pool_identifier,
520                            e
521                        );
522                        None
523                    }
524                }
525            }
526            BlockchainMessage::FeeProtocolCollectEvent(collect_event) => {
527                match core_client.get_pool(&collect_event.pool_identifier) {
528                    Ok(pool) => match core_client
529                        .process_pool_fee_protocol_collect_event(&collect_event, pool)
530                    {
531                        Ok(collect) => {
532                            Some(DataEvent::DeFi(DefiData::PoolFeeProtocolCollect(collect)))
533                        }
534                        Err(e) => {
535                            log::error!("Error processing pool fee-protocol collect event: {e}");
536                            None
537                        }
538                    },
539                    Err(e) => {
540                        log::error!(
541                            "Failed to get pool {} with error {:?}",
542                            collect_event.pool_identifier,
543                            e
544                        );
545                        None
546                    }
547                }
548            }
549        }
550    }
551
552    /// Processes DeFi subscription commands to start receiving specific blockchain data.
553    async fn handle_subscribe_command(
554        command: DefiSubscribeCommand,
555        core_client: &mut BlockchainDataClientCore,
556    ) -> anyhow::Result<()> {
557        match command {
558            DefiSubscribeCommand::Blocks(_cmd) => {
559                log::debug!("Processing subscribe blocks command");
560
561                Self::subscribe_block_feed(core_client, BlockFeedOwner::Explicit).await?;
562
563                Ok(())
564            }
565            DefiSubscribeCommand::Pool(cmd) => {
566                log::debug!(
567                    "Processing subscribe pool command for {}",
568                    cmd.instrument_id
569                );
570
571                if let Ok((_, dex)) = cmd.instrument_id.venue.parse_dex() {
572                    let pool_address = validate_address(cmd.instrument_id.symbol.as_str())
573                        .map_err(|e| {
574                            anyhow::anyhow!(
575                                "Invalid pool address '{}' failed with error: {:?}",
576                                cmd.instrument_id,
577                                e
578                            )
579                        })?;
580
581                    core_client
582                        .subscription_manager
583                        .subscribe_pool(dex, pool_address);
584                    Self::update_rpc_pool_event_subscriptions(core_client, dex).await?;
585                    Self::update_hypersync_pool_event_stream(core_client, dex).await?;
586
587                    log::debug!(
588                        "Subscribed to all pool events for {} at address {}",
589                        cmd.instrument_id,
590                        pool_address
591                    );
592                } else {
593                    anyhow::bail!(
594                        "Invalid venue {}, expected Blockchain DEX format",
595                        cmd.instrument_id.venue
596                    )
597                }
598
599                Ok(())
600            }
601            DefiSubscribeCommand::PoolSwaps(cmd) => {
602                log::debug!(
603                    "Processing subscribe pool swaps command for {}",
604                    cmd.instrument_id
605                );
606
607                if let Ok((_, dex)) = cmd.instrument_id.venue.parse_dex() {
608                    let pool_address = validate_address(cmd.instrument_id.symbol.as_str())
609                        .map_err(|e| {
610                            anyhow::anyhow!(
611                                "Invalid pool swap address '{}' failed with error: {:?}",
612                                cmd.instrument_id,
613                                e
614                            )
615                        })?;
616                    core_client
617                        .subscription_manager
618                        .subscribe_swaps(dex, pool_address);
619                    Self::update_rpc_pool_event_subscriptions(core_client, dex).await?;
620                    Self::update_hypersync_pool_event_stream(core_client, dex).await?;
621                } else {
622                    anyhow::bail!(
623                        "Invalid venue {}, expected Blockchain DEX format",
624                        cmd.instrument_id.venue
625                    )
626                }
627
628                Ok(())
629            }
630            DefiSubscribeCommand::PoolLiquidityUpdates(cmd) => {
631                log::debug!(
632                    "Processing subscribe pool liquidity updates command for address: {}",
633                    cmd.instrument_id
634                );
635
636                if let Ok((_, dex)) = cmd.instrument_id.venue.parse_dex() {
637                    let pool_address = validate_address(cmd.instrument_id.symbol.as_str())
638                        .map_err(|_| {
639                            anyhow::anyhow!("Invalid pool swap address: {}", cmd.instrument_id)
640                        })?;
641                    core_client
642                        .subscription_manager
643                        .subscribe_burns(dex, pool_address);
644                    core_client
645                        .subscription_manager
646                        .subscribe_mints(dex, pool_address);
647                    Self::update_rpc_pool_event_subscriptions(core_client, dex).await?;
648                    Self::update_hypersync_pool_event_stream(core_client, dex).await?;
649                } else {
650                    anyhow::bail!(
651                        "Invalid venue {}, expected Blockchain DEX format",
652                        cmd.instrument_id.venue
653                    )
654                }
655
656                Ok(())
657            }
658            DefiSubscribeCommand::PoolFeeCollects(cmd) => {
659                log::debug!(
660                    "Processing subscribe pool fee collects command for address: {}",
661                    cmd.instrument_id
662                );
663
664                if let Ok((_, dex)) = cmd.instrument_id.venue.parse_dex() {
665                    let pool_address = validate_address(cmd.instrument_id.symbol.as_str())
666                        .map_err(|_| {
667                            anyhow::anyhow!(
668                                "Invalid pool fee collect address: {}",
669                                cmd.instrument_id
670                            )
671                        })?;
672                    core_client
673                        .subscription_manager
674                        .subscribe_collects(dex, pool_address);
675                    Self::update_rpc_pool_event_subscriptions(core_client, dex).await?;
676                    Self::update_hypersync_pool_event_stream(core_client, dex).await?;
677                } else {
678                    anyhow::bail!(
679                        "Invalid venue {}, expected Blockchain DEX format",
680                        cmd.instrument_id.venue
681                    )
682                }
683
684                Ok(())
685            }
686            DefiSubscribeCommand::PoolFlashEvents(cmd) => {
687                log::debug!(
688                    "Processing subscribe pool flash command for address: {}",
689                    cmd.instrument_id
690                );
691
692                if let Ok((_, dex)) = cmd.instrument_id.venue.parse_dex() {
693                    let pool_address = validate_address(cmd.instrument_id.symbol.as_str())
694                        .map_err(|_| {
695                            anyhow::anyhow!(
696                                "Invalid pool flash subscribe address: {}",
697                                cmd.instrument_id
698                            )
699                        })?;
700                    core_client
701                        .subscription_manager
702                        .subscribe_flashes(dex, pool_address);
703                    Self::update_rpc_pool_event_subscriptions(core_client, dex).await?;
704                    Self::update_hypersync_pool_event_stream(core_client, dex).await?;
705                } else {
706                    anyhow::bail!(
707                        "Invalid venue {}, expected Blockchain DEX format",
708                        cmd.instrument_id.venue
709                    )
710                }
711
712                Ok(())
713            }
714        }
715    }
716
717    /// Processes DeFi unsubscription commands to stop receiving specific blockchain data.
718    async fn handle_unsubscribe_command(
719        command: DefiUnsubscribeCommand,
720        core_client: &mut BlockchainDataClientCore,
721    ) -> anyhow::Result<()> {
722        match command {
723            DefiUnsubscribeCommand::Blocks(_cmd) => {
724                log::debug!("Processing unsubscribe blocks command");
725
726                Self::unsubscribe_block_feed(core_client, BlockFeedOwner::Explicit).await?;
727
728                Ok(())
729            }
730            DefiUnsubscribeCommand::Pool(cmd) => {
731                log::debug!(
732                    "Processing unsubscribe pool command for {}",
733                    cmd.instrument_id
734                );
735
736                if let Ok((_, dex)) = cmd.instrument_id.venue.parse_dex() {
737                    let pool_address = validate_address(cmd.instrument_id.symbol.as_str())
738                        .map_err(|_| {
739                            anyhow::anyhow!("Invalid pool address: {}", cmd.instrument_id)
740                        })?;
741
742                    core_client
743                        .subscription_manager
744                        .unsubscribe_pool(dex, pool_address);
745                    Self::update_rpc_pool_event_subscriptions(core_client, dex).await?;
746                    Self::update_hypersync_pool_event_stream(core_client, dex).await?;
747
748                    log::debug!(
749                        "Unsubscribed from all pool events for {} at address {}",
750                        cmd.instrument_id,
751                        pool_address
752                    );
753                } else {
754                    anyhow::bail!(
755                        "Invalid venue {}, expected Blockchain DEX format",
756                        cmd.instrument_id.venue
757                    )
758                }
759
760                Ok(())
761            }
762            DefiUnsubscribeCommand::PoolSwaps(cmd) => {
763                log::debug!("Processing unsubscribe pool swaps command");
764
765                if let Ok((_, dex)) = cmd.instrument_id.venue.parse_dex() {
766                    let pool_address = validate_address(cmd.instrument_id.symbol.as_str())
767                        .map_err(|_| {
768                            anyhow::anyhow!("Invalid pool swap address: {}", cmd.instrument_id)
769                        })?;
770                    core_client
771                        .subscription_manager
772                        .unsubscribe_swaps(dex, pool_address);
773                    Self::update_rpc_pool_event_subscriptions(core_client, dex).await?;
774                    Self::update_hypersync_pool_event_stream(core_client, dex).await?;
775                } else {
776                    anyhow::bail!(
777                        "Invalid venue {}, expected Blockchain DEX format",
778                        cmd.instrument_id.venue
779                    )
780                }
781
782                Ok(())
783            }
784            DefiUnsubscribeCommand::PoolLiquidityUpdates(cmd) => {
785                log::debug!(
786                    "Processing unsubscribe pool liquidity updates command for {}",
787                    cmd.instrument_id
788                );
789
790                if let Ok((_, dex)) = cmd.instrument_id.venue.parse_dex() {
791                    let pool_address = validate_address(cmd.instrument_id.symbol.as_str())
792                        .map_err(|_| {
793                            anyhow::anyhow!("Invalid pool swap address: {}", cmd.instrument_id)
794                        })?;
795                    core_client
796                        .subscription_manager
797                        .unsubscribe_burns(dex, pool_address);
798                    core_client
799                        .subscription_manager
800                        .unsubscribe_mints(dex, pool_address);
801                    Self::update_rpc_pool_event_subscriptions(core_client, dex).await?;
802                    Self::update_hypersync_pool_event_stream(core_client, dex).await?;
803                } else {
804                    anyhow::bail!(
805                        "Invalid venue {}, expected Blockchain DEX format",
806                        cmd.instrument_id.venue
807                    )
808                }
809
810                Ok(())
811            }
812            DefiUnsubscribeCommand::PoolFeeCollects(cmd) => {
813                log::debug!(
814                    "Processing unsubscribe pool fee collects command for {}",
815                    cmd.instrument_id
816                );
817
818                if let Ok((_, dex)) = cmd.instrument_id.venue.parse_dex() {
819                    let pool_address = validate_address(cmd.instrument_id.symbol.as_str())
820                        .map_err(|_| {
821                            anyhow::anyhow!(
822                                "Invalid pool fee collect address: {}",
823                                cmd.instrument_id
824                            )
825                        })?;
826                    core_client
827                        .subscription_manager
828                        .unsubscribe_collects(dex, pool_address);
829                    Self::update_rpc_pool_event_subscriptions(core_client, dex).await?;
830                    Self::update_hypersync_pool_event_stream(core_client, dex).await?;
831                } else {
832                    anyhow::bail!(
833                        "Invalid venue {}, expected Blockchain DEX format",
834                        cmd.instrument_id.venue
835                    )
836                }
837
838                Ok(())
839            }
840            DefiUnsubscribeCommand::PoolFlashEvents(cmd) => {
841                log::debug!(
842                    "Processing unsubscribe pool flash command for {}",
843                    cmd.instrument_id
844                );
845
846                if let Ok((_, dex)) = cmd.instrument_id.venue.parse_dex() {
847                    let pool_address = validate_address(cmd.instrument_id.symbol.as_str())
848                        .map_err(|_| {
849                            anyhow::anyhow!("Invalid pool flash address: {}", cmd.instrument_id)
850                        })?;
851                    core_client
852                        .subscription_manager
853                        .unsubscribe_flashes(dex, pool_address);
854                    Self::update_rpc_pool_event_subscriptions(core_client, dex).await?;
855                    Self::update_hypersync_pool_event_stream(core_client, dex).await?;
856                } else {
857                    anyhow::bail!(
858                        "Invalid venue {}, expected Blockchain DEX format",
859                        cmd.instrument_id.venue
860                    )
861                }
862
863                Ok(())
864            }
865        }
866    }
867
868    async fn update_rpc_pool_event_subscriptions(
869        core_client: &mut BlockchainDataClientCore,
870        dex: DexType,
871    ) -> anyhow::Result<()> {
872        let updates = vec![
873            (
874                RpcEventType::PoolSwap(dex),
875                core_client
876                    .subscription_manager
877                    .get_subscribed_pool_swap_addresses(&dex),
878                core_client
879                    .subscription_manager
880                    .get_dex_pool_swap_event_signature(&dex),
881            ),
882            (
883                RpcEventType::PoolMint(dex),
884                core_client
885                    .subscription_manager
886                    .get_subscribed_pool_mint_addresses(&dex),
887                core_client
888                    .subscription_manager
889                    .get_dex_pool_mint_event_signature(&dex),
890            ),
891            (
892                RpcEventType::PoolBurn(dex),
893                core_client
894                    .subscription_manager
895                    .get_subscribed_pool_burn_addresses(&dex),
896                core_client
897                    .subscription_manager
898                    .get_dex_pool_burn_event_signature(&dex),
899            ),
900            (
901                RpcEventType::PoolCollect(dex),
902                core_client
903                    .subscription_manager
904                    .get_subscribed_pool_collect_addresses(&dex),
905                core_client
906                    .subscription_manager
907                    .get_dex_pool_collect_event_signature(&dex),
908            ),
909            (
910                RpcEventType::PoolFlash(dex),
911                core_client
912                    .subscription_manager
913                    .get_subscribed_pool_flash_addresses(&dex),
914                core_client
915                    .subscription_manager
916                    .get_dex_pool_flash_event_signature(&dex),
917            ),
918            (
919                RpcEventType::PoolFeeProtocolUpdate(dex),
920                core_client
921                    .subscription_manager
922                    .get_subscribed_pool_fee_protocol_update_addresses(&dex),
923                core_client
924                    .subscription_manager
925                    .get_dex_pool_fee_protocol_update_event_signature(&dex),
926            ),
927            (
928                RpcEventType::PoolFeeProtocolCollect(dex),
929                core_client
930                    .subscription_manager
931                    .get_subscribed_pool_fee_protocol_collect_addresses(&dex),
932                core_client
933                    .subscription_manager
934                    .get_dex_pool_fee_protocol_collect_event_signature(&dex),
935            ),
936        ];
937
938        let has_pool_event_subscriptions = Self::has_active_pool_event_subscriptions(core_client);
939
940        if core_client.rpc_client.is_none() {
941            return Ok(());
942        }
943
944        if has_pool_event_subscriptions {
945            Self::subscribe_block_feed(core_client, BlockFeedOwner::PoolEvents).await?;
946        }
947
948        if let Some(ref mut rpc) = core_client.rpc_client {
949            for (event_type, addresses, event_signature) in updates {
950                if let Some(event_signature) = event_signature {
951                    rpc.subscribe_pool_events(event_type, &addresses, event_signature)
952                        .await?;
953                }
954            }
955        }
956
957        if !has_pool_event_subscriptions {
958            Self::unsubscribe_block_feed(core_client, BlockFeedOwner::PoolEvents).await?;
959        }
960
961        Ok(())
962    }
963
964    async fn update_hypersync_pool_event_stream(
965        core_client: &mut BlockchainDataClientCore,
966        dex: DexType,
967    ) -> anyhow::Result<()> {
968        if core_client.rpc_client.is_some() {
969            return Ok(());
970        }
971
972        let addresses = core_client
973            .subscription_manager
974            .get_subscribed_dex_contract_addresses(&dex);
975        let event_signatures = core_client
976            .subscription_manager
977            .get_active_subscribed_dex_event_signatures(&dex);
978        let has_pool_event_subscriptions = Self::has_active_pool_event_subscriptions(core_client);
979
980        if has_pool_event_subscriptions {
981            Self::subscribe_block_feed(core_client, BlockFeedOwner::PoolEvents).await?;
982        }
983
984        core_client
985            .hypersync_client
986            .update_dex_event_stream(dex, addresses, event_signatures)
987            .await;
988
989        if !has_pool_event_subscriptions {
990            Self::unsubscribe_block_feed(core_client, BlockFeedOwner::PoolEvents).await?;
991        }
992
993        Ok(())
994    }
995
996    fn has_active_pool_event_subscriptions(core_client: &BlockchainDataClientCore) -> bool {
997        core_client
998            .subscription_manager
999            .has_pool_event_subscriptions()
1000    }
1001
1002    async fn subscribe_block_feed(
1003        core_client: &mut BlockchainDataClientCore,
1004        owner: BlockFeedOwner,
1005    ) -> anyhow::Result<()> {
1006        let preferred_backend = if core_client.rpc_client.is_some() {
1007            BlockFeedBackend::Rpc
1008        } else {
1009            BlockFeedBackend::HyperSync
1010        };
1011        let Some(backend) = core_client
1012            .subscription_manager
1013            .add_block_demand(owner, preferred_backend)
1014        else {
1015            return Ok(());
1016        };
1017
1018        let started_backend = match backend {
1019            BlockFeedBackend::Rpc => {
1020                let Some(rpc) = core_client.rpc_client.as_mut() else {
1021                    anyhow::bail!("RPC block feed selected without an RPC client")
1022                };
1023
1024                match rpc.subscribe_blocks().await {
1025                    Ok(()) => {
1026                        log::debug!("Successfully subscribed to blocks via RPC");
1027                        BlockFeedBackend::Rpc
1028                    }
1029                    Err(e) if owner == BlockFeedOwner::Explicit => {
1030                        log::warn!(
1031                            "RPC blocks subscription failed: {e}, falling back to HyperSync"
1032                        );
1033                        core_client.hypersync_client.subscribe_blocks();
1034                        tokio::task::yield_now().await;
1035                        BlockFeedBackend::HyperSync
1036                    }
1037                    Err(e) => return Err(e.into()),
1038                }
1039            }
1040            BlockFeedBackend::HyperSync => {
1041                log::debug!("Subscribing to blocks via HyperSync");
1042                core_client.hypersync_client.subscribe_blocks();
1043                tokio::task::yield_now().await;
1044                BlockFeedBackend::HyperSync
1045            }
1046        };
1047
1048        core_client
1049            .subscription_manager
1050            .block_feed_started(started_backend);
1051        Ok(())
1052    }
1053
1054    async fn unsubscribe_block_feed(
1055        core_client: &mut BlockchainDataClientCore,
1056        owner: BlockFeedOwner,
1057    ) -> anyhow::Result<()> {
1058        let Some(backend) = core_client.subscription_manager.remove_block_demand(owner) else {
1059            log::debug!("Keeping block subscription active while another owner remains");
1060            return Ok(());
1061        };
1062
1063        match backend {
1064            BlockFeedBackend::Rpc => {
1065                let Some(rpc) = core_client.rpc_client.as_mut() else {
1066                    anyhow::bail!("RPC block feed active without an RPC client")
1067                };
1068                rpc.unsubscribe_blocks().await?;
1069                log::debug!("Unsubscribed from blocks via RPC");
1070            }
1071            BlockFeedBackend::HyperSync => {
1072                core_client.hypersync_client.unsubscribe_blocks().await;
1073                log::debug!("Unsubscribed from blocks via HyperSync");
1074            }
1075        }
1076
1077        core_client.subscription_manager.block_feed_stopped(backend);
1078        Ok(())
1079    }
1080
1081    /// Processes DeFi request commands to fetch specific blockchain data.
1082    async fn handle_request_command(
1083        command: DefiRequestCommand,
1084        core_client: &mut BlockchainDataClientCore,
1085    ) -> anyhow::Result<()> {
1086        match command {
1087            DefiRequestCommand::PoolSnapshot(cmd) => {
1088                log::debug!("Processing pool snapshot request for {}", cmd.instrument_id);
1089
1090                let pool_address =
1091                    validate_address(cmd.instrument_id.symbol.as_str()).map_err(|e| {
1092                        anyhow::anyhow!(
1093                            "Invalid pool address '{}' failed with error: {:?}",
1094                            cmd.instrument_id,
1095                            e
1096                        )
1097                    })?;
1098
1099                let pool_identifier =
1100                    PoolIdentifier::Address(Ustr::from(&pool_address.to_string()));
1101
1102                match core_client.get_pool(&pool_identifier) {
1103                    Ok(pool) => {
1104                        let pool = Arc::clone(pool);
1105                        log::debug!("Found pool for snapshot request: {}", cmd.instrument_id);
1106
1107                        // Send the pool definition
1108                        let pool_data = DataEvent::DeFi(DefiData::Pool(pool.as_ref().clone()));
1109                        core_client.send_data(pool_data);
1110
1111                        match core_client
1112                            .bootstrap_latest_pool_profiler(&pool, None)
1113                            .await
1114                        {
1115                            Ok((profiler, already_valid)) => match profiler.extract_snapshot() {
1116                                Ok(snapshot) => {
1117                                    log::debug!(
1118                                        "Saving pool snapshot with {} positions and {} ticks to database...",
1119                                        snapshot.positions.len(),
1120                                        snapshot.ticks.len()
1121                                    );
1122                                    core_client
1123                                        .cache
1124                                        .add_pool_snapshot(
1125                                            &pool.dex.name,
1126                                            &pool.pool_identifier,
1127                                            &snapshot,
1128                                        )
1129                                        .await?;
1130
1131                                    // If the snapshot is usable, send it back to the data engine.
1132                                    if core_client
1133                                        .check_snapshot_validity(&profiler, already_valid)
1134                                        .await?
1135                                        .is_usable()
1136                                    {
1137                                        let response =
1138                                            PoolSnapshotResponse::new(cmd.request_id, snapshot);
1139                                        core_client
1140                                            .send_data(DataEvent::PoolSnapshotResponse(response));
1141                                    }
1142                                }
1143                                Err(e) => log::error!(
1144                                    "Failed to extract snapshot for {}: {e}",
1145                                    cmd.instrument_id
1146                                ),
1147                            },
1148                            Err(e) => log::error!(
1149                                "Failed to bootstrap pool profiler for {} and extract snapshot with error {e}",
1150                                cmd.instrument_id
1151                            ),
1152                        }
1153                    }
1154                    Err(e) => {
1155                        log::warn!("Pool {} not found in cache: {e}", cmd.instrument_id);
1156                    }
1157                }
1158
1159                Ok(())
1160            }
1161        }
1162    }
1163
1164    /// Waits for the background processing task to complete.
1165    ///
1166    /// This method blocks until the spawned process task finishes execution,
1167    /// which typically happens after a shutdown signal is sent.
1168    ///
1169    /// # Errors
1170    ///
1171    /// Returns an error if bounded task shutdown fails.
1172    pub async fn await_process_task_close(&self) -> anyhow::Result<()> {
1173        self.session_tasks
1174            .finish_shutdown(Duration::from_secs(2), Duration::from_secs(2))
1175            .await
1176            .map_err(|e| anyhow::anyhow!("Blockchain process task shutdown failed: {e}"))?;
1177        Ok(())
1178    }
1179
1180    async fn prepare_task_group(&mut self) -> anyhow::Result<()> {
1181        if !self.session_tasks.is_open() {
1182            self.session_tasks.begin_shutdown();
1183            self.await_process_task_close().await?;
1184            self.reset_channels();
1185            self.session_tasks
1186                .start_generation()
1187                .map_err(|e| anyhow::anyhow!("Failed to start blockchain task generation: {e}"))?;
1188            self.cancellation_token = self.session_tasks.cancellation_token();
1189        }
1190        Ok(())
1191    }
1192
1193    fn reset_channels(&mut self) {
1194        let (hypersync_tx, hypersync_rx) = tokio::sync::mpsc::unbounded_channel();
1195        self.hypersync_tx = Some(hypersync_tx);
1196        self.hypersync_rx = Some(hypersync_rx);
1197        let (command_tx, command_rx) = tokio::sync::mpsc::unbounded_channel();
1198        self.command_tx = command_tx;
1199        self.command_rx = Some(command_rx);
1200    }
1201}
1202
1203#[async_trait::async_trait(?Send)]
1204impl DataClient for BlockchainDataClient {
1205    fn client_id(&self) -> ClientId {
1206        self.client_id
1207    }
1208
1209    fn venue(&self) -> Option<Venue> {
1210        // Blockchain data clients don't map to a single venue since they can provide
1211        // data for multiple DEXs across the blockchain
1212        None
1213    }
1214
1215    fn start(&mut self) -> anyhow::Result<()> {
1216        log::info!(
1217            "Starting blockchain data client: chain_name={}, dex_ids={:?}, use_hypersync_for_live_data={}, proxy_url={:?}",
1218            self.chain.name,
1219            self.config.dex_ids,
1220            self.config.use_hypersync_for_live_data,
1221            self.config.proxy_url
1222        );
1223        Ok(())
1224    }
1225
1226    fn stop(&mut self) -> anyhow::Result<()> {
1227        log::info!(
1228            "Stopping blockchain data client for '{chain_name}'",
1229            chain_name = self.chain.name
1230        );
1231        self.session_tasks.begin_shutdown();
1232        Ok(())
1233    }
1234
1235    fn reset(&mut self) -> anyhow::Result<()> {
1236        log::info!(
1237            "Resetting blockchain data client for '{chain_name}'",
1238            chain_name = self.chain.name
1239        );
1240        self.session_tasks.begin_shutdown();
1241        Ok(())
1242    }
1243
1244    fn dispose(&mut self) -> anyhow::Result<()> {
1245        log::info!(
1246            "Disposing blockchain data client for '{chain_name}'",
1247            chain_name = self.chain.name
1248        );
1249        self.session_tasks.begin_shutdown();
1250        Ok(())
1251    }
1252
1253    async fn connect(&mut self) -> anyhow::Result<()> {
1254        self.config.validate()?;
1255
1256        if self.session_tasks.is_open() && !self.session_tasks.is_empty() {
1257            return Ok(());
1258        }
1259
1260        log::info!(
1261            "Connecting blockchain data client for '{}'",
1262            self.chain.name
1263        );
1264
1265        self.prepare_task_group().await?;
1266        let setup_guard = TaskGroupGuard::new(&[&self.session_tasks], || {});
1267        let startup = self.spawn_process_task()?;
1268        if let Err(e) = startup
1269            .await
1270            .context("Blockchain process task stopped before startup")?
1271        {
1272            self.session_tasks.begin_shutdown();
1273            let teardown_result = self.await_process_task_close().await;
1274            self.reset_channels();
1275
1276            if let Err(teardown_error) = teardown_result {
1277                return Err(e.context(format!(
1278                    "Blockchain data startup teardown failed: {teardown_error}"
1279                )));
1280            }
1281            return Err(e);
1282        }
1283
1284        setup_guard.disarm();
1285        Ok(())
1286    }
1287
1288    async fn disconnect(&mut self) -> anyhow::Result<()> {
1289        log::info!(
1290            "Disconnecting blockchain data client for '{}'",
1291            self.chain.name
1292        );
1293
1294        self.session_tasks.begin_shutdown();
1295        let tasks_result = self.await_process_task_close().await;
1296        self.reset_channels();
1297
1298        tasks_result
1299    }
1300
1301    fn is_connected(&self) -> bool {
1302        self.session_tasks.is_open()
1303            && !self.session_tasks.is_empty()
1304            && !self.session_tasks.all_finished()
1305    }
1306
1307    fn is_disconnected(&self) -> bool {
1308        !self.is_connected()
1309    }
1310
1311    fn subscribe_blocks(&mut self, cmd: SubscribeBlocks) -> anyhow::Result<()> {
1312        let command = DefiDataCommand::Subscribe(DefiSubscribeCommand::Blocks(cmd));
1313        self.command_tx.send(command)?;
1314        Ok(())
1315    }
1316
1317    fn subscribe_pool(&mut self, cmd: SubscribePool) -> anyhow::Result<()> {
1318        let command = DefiDataCommand::Subscribe(DefiSubscribeCommand::Pool(cmd));
1319        self.command_tx.send(command)?;
1320        Ok(())
1321    }
1322
1323    fn subscribe_pool_swaps(&mut self, cmd: SubscribePoolSwaps) -> anyhow::Result<()> {
1324        let command = DefiDataCommand::Subscribe(DefiSubscribeCommand::PoolSwaps(cmd));
1325        self.command_tx.send(command)?;
1326        Ok(())
1327    }
1328
1329    fn subscribe_pool_liquidity_updates(
1330        &mut self,
1331        cmd: SubscribePoolLiquidityUpdates,
1332    ) -> anyhow::Result<()> {
1333        let command = DefiDataCommand::Subscribe(DefiSubscribeCommand::PoolLiquidityUpdates(cmd));
1334        self.command_tx.send(command)?;
1335        Ok(())
1336    }
1337
1338    fn subscribe_pool_fee_collects(&mut self, cmd: SubscribePoolFeeCollects) -> anyhow::Result<()> {
1339        let command = DefiDataCommand::Subscribe(DefiSubscribeCommand::PoolFeeCollects(cmd));
1340        self.command_tx.send(command)?;
1341        Ok(())
1342    }
1343
1344    fn subscribe_pool_flash_events(&mut self, cmd: SubscribePoolFlashEvents) -> anyhow::Result<()> {
1345        let command = DefiDataCommand::Subscribe(DefiSubscribeCommand::PoolFlashEvents(cmd));
1346        self.command_tx.send(command)?;
1347        Ok(())
1348    }
1349
1350    fn unsubscribe_blocks(&mut self, cmd: &UnsubscribeBlocks) -> anyhow::Result<()> {
1351        let command = DefiDataCommand::Unsubscribe(DefiUnsubscribeCommand::Blocks(cmd.clone()));
1352        self.command_tx.send(command)?;
1353        Ok(())
1354    }
1355
1356    fn unsubscribe_pool(&mut self, cmd: &UnsubscribePool) -> anyhow::Result<()> {
1357        let command = DefiDataCommand::Unsubscribe(DefiUnsubscribeCommand::Pool(cmd.clone()));
1358        self.command_tx.send(command)?;
1359        Ok(())
1360    }
1361
1362    fn unsubscribe_pool_swaps(&mut self, cmd: &UnsubscribePoolSwaps) -> anyhow::Result<()> {
1363        let command = DefiDataCommand::Unsubscribe(DefiUnsubscribeCommand::PoolSwaps(cmd.clone()));
1364        self.command_tx.send(command)?;
1365        Ok(())
1366    }
1367
1368    fn unsubscribe_pool_liquidity_updates(
1369        &mut self,
1370        cmd: &UnsubscribePoolLiquidityUpdates,
1371    ) -> anyhow::Result<()> {
1372        let command =
1373            DefiDataCommand::Unsubscribe(DefiUnsubscribeCommand::PoolLiquidityUpdates(cmd.clone()));
1374        self.command_tx.send(command)?;
1375        Ok(())
1376    }
1377
1378    fn unsubscribe_pool_fee_collects(
1379        &mut self,
1380        cmd: &UnsubscribePoolFeeCollects,
1381    ) -> anyhow::Result<()> {
1382        let command =
1383            DefiDataCommand::Unsubscribe(DefiUnsubscribeCommand::PoolFeeCollects(cmd.clone()));
1384        self.command_tx.send(command)?;
1385        Ok(())
1386    }
1387
1388    fn unsubscribe_pool_flash_events(
1389        &mut self,
1390        cmd: &UnsubscribePoolFlashEvents,
1391    ) -> anyhow::Result<()> {
1392        let command =
1393            DefiDataCommand::Unsubscribe(DefiUnsubscribeCommand::PoolFlashEvents(cmd.clone()));
1394        self.command_tx.send(command)?;
1395        Ok(())
1396    }
1397
1398    fn request_pool_snapshot(&self, cmd: RequestPoolSnapshot) -> anyhow::Result<()> {
1399        let command = DefiDataCommand::Request(DefiRequestCommand::PoolSnapshot(cmd));
1400        self.command_tx.send(command)?;
1401        Ok(())
1402    }
1403}
1404
1405#[cfg(test)]
1406mod tests {
1407    use std::sync::Arc;
1408
1409    use alloy::primitives::{I256, U160, U256, address};
1410    use nautilus_core::UnixNanos;
1411    use nautilus_model::defi::{Block, Blockchain, Chain, DexType, Pool, PoolIdentifier, Token};
1412    use rstest::rstest;
1413
1414    use super::*;
1415    use crate::events::{flash::FlashEvent, swap::SwapEvent};
1416
1417    const WETH_USDT_CREATION_BLOCK: u64 = 12_375_326;
1418
1419    #[rstest]
1420    fn pool_event_missing_block_metadata_clears_after_block_cache_update() {
1421        let chain = Arc::new(
1422            Chain::from_chain_id(1)
1423                .expect("Ethereum chain should exist")
1424                .clone(),
1425        );
1426        let mut cache = BlockchainCache::new(chain);
1427        let msg = flash_message(42);
1428
1429        assert_eq!(
1430            BlockchainDataClient::pool_event_block_number(&msg),
1431            Some(42)
1432        );
1433        assert_eq!(
1434            BlockchainDataClient::pool_event_missing_block_metadata(&msg, &cache),
1435            Some(42)
1436        );
1437
1438        let block = test_block(42, "0x1", 1_700_000_000_000_000_000);
1439        cache.cache_block_metadata(&block);
1440
1441        assert_eq!(
1442            BlockchainDataClient::pool_event_missing_block_metadata(&msg, &cache),
1443            None
1444        );
1445    }
1446
1447    #[rstest]
1448    fn queue_pending_pool_message_drops_oldest_entry_at_cap() {
1449        let mut pending = VecDeque::new();
1450
1451        for block_number in 0..MAX_PENDING_POOL_MESSAGES {
1452            BlockchainDataClient::queue_pending_pool_message(
1453                &mut pending,
1454                swap_message(block_number as u64),
1455            );
1456        }
1457
1458        BlockchainDataClient::queue_pending_pool_message(
1459            &mut pending,
1460            swap_message(MAX_PENDING_POOL_MESSAGES as u64),
1461        );
1462
1463        assert_eq!(pending.len(), MAX_PENDING_POOL_MESSAGES);
1464        assert_eq!(
1465            BlockchainDataClient::pool_event_block_number(pending.front().unwrap()),
1466            Some(1)
1467        );
1468        assert_eq!(
1469            BlockchainDataClient::pool_event_block_number(pending.back().unwrap()),
1470            Some(MAX_PENDING_POOL_MESSAGES as u64)
1471        );
1472    }
1473
1474    #[rstest]
1475    fn drain_pending_pool_messages_releases_events_after_block_metadata_is_cached() {
1476        let chain = Arc::new(
1477            Chain::from_chain_id(1)
1478                .expect("Ethereum chain should exist")
1479                .clone(),
1480        );
1481        let mut cache = BlockchainCache::new(chain);
1482        let block_42 = test_block(42, "0x42", 1_700_000_000_000_000_000);
1483        cache.cache_block_metadata(&block_42);
1484        let mut pending = VecDeque::from([flash_message(41), flash_message(42)]);
1485
1486        let ready_messages = BlockchainDataClient::drain_pending_pool_messages_with_cached_metadata(
1487            &mut pending,
1488            &cache,
1489        );
1490
1491        assert_eq!(ready_messages.len(), 1);
1492        assert_eq!(
1493            BlockchainDataClient::pool_event_block_number(&ready_messages[0]),
1494            Some(42)
1495        );
1496        assert_eq!(pending.len(), 1);
1497        assert_eq!(
1498            BlockchainDataClient::pool_event_block_number(pending.front().unwrap()),
1499            Some(41)
1500        );
1501
1502        let block_41 = test_block(41, "0x41", 1_700_000_000_000_000_001);
1503        cache.cache_block_metadata(&block_41);
1504
1505        let ready_messages = BlockchainDataClient::drain_pending_pool_messages_with_cached_metadata(
1506            &mut pending,
1507            &cache,
1508        );
1509
1510        assert_eq!(ready_messages.len(), 1);
1511        assert_eq!(
1512            BlockchainDataClient::pool_event_block_number(&ready_messages[0]),
1513            Some(41)
1514        );
1515        assert!(pending.is_empty());
1516    }
1517
1518    #[rstest]
1519    fn ready_live_blockchain_message_queues_until_block_metadata_is_cached() {
1520        let chain = Arc::new(
1521            Chain::from_chain_id(1)
1522                .expect("Ethereum chain should exist")
1523                .clone(),
1524        );
1525        let mut cache = BlockchainCache::new(chain);
1526        let mut pending = VecDeque::new();
1527
1528        let message = BlockchainDataClient::ready_live_blockchain_message(
1529            flash_message(42),
1530            &cache,
1531            &mut pending,
1532        );
1533
1534        assert!(message.is_none());
1535        assert_eq!(pending.len(), 1);
1536        assert_eq!(
1537            BlockchainDataClient::pool_event_block_number(pending.front().unwrap()),
1538            Some(42)
1539        );
1540
1541        let block = test_block(42, "0x1", 1_700_000_000_000_000_000);
1542        cache.cache_block_metadata(&block);
1543
1544        let message = BlockchainDataClient::ready_live_blockchain_message(
1545            BlockchainMessage::Block(block),
1546            &cache,
1547            &mut pending,
1548        );
1549        let ready_messages = BlockchainDataClient::drain_pending_pool_messages_with_cached_metadata(
1550            &mut pending,
1551            &cache,
1552        );
1553
1554        assert!(matches!(message, Some(BlockchainMessage::Block(_))));
1555        assert_eq!(ready_messages.len(), 1);
1556        assert_eq!(
1557            BlockchainDataClient::pool_event_block_number(&ready_messages[0]),
1558            Some(42)
1559        );
1560        assert!(pending.is_empty());
1561    }
1562
1563    fn swap_message(block_number: u64) -> BlockchainMessage {
1564        let pool = weth_usdt_pool();
1565        let address = address!("1111111111111111111111111111111111111111");
1566
1567        BlockchainMessage::SwapEvent(SwapEvent::new(
1568            Arc::clone(&pool.dex),
1569            pool.pool_identifier,
1570            block_number,
1571            "0x1".to_string(),
1572            0,
1573            0,
1574            address,
1575            address,
1576            I256::ZERO,
1577            I256::ZERO,
1578            U160::ZERO,
1579            0,
1580            0,
1581        ))
1582    }
1583
1584    fn test_block(number: u64, hash: &str, timestamp: u64) -> Block {
1585        Block::new(
1586            hash.to_string(),
1587            "0x0".to_string(),
1588            number,
1589            Ustr::from("0x0000000000000000000000000000000000000000"),
1590            30_000_000,
1591            21_000,
1592            UnixNanos::from(timestamp),
1593            Some(Blockchain::Ethereum),
1594        )
1595    }
1596
1597    fn flash_message(block_number: u64) -> BlockchainMessage {
1598        let pool = weth_usdt_pool();
1599        let address = address!("1111111111111111111111111111111111111111");
1600
1601        BlockchainMessage::FlashEvent(FlashEvent::new(
1602            Arc::clone(&pool.dex),
1603            pool.pool_identifier,
1604            block_number,
1605            "0x1".to_string(),
1606            0,
1607            0,
1608            address,
1609            address,
1610            U256::ZERO,
1611            U256::ZERO,
1612            U256::ZERO,
1613            U256::ZERO,
1614        ))
1615    }
1616
1617    fn weth_usdt_pool() -> Arc<Pool> {
1618        let chain = Arc::new(
1619            Chain::from_chain_id(1)
1620                .expect("Ethereum chain should exist")
1621                .clone(),
1622        );
1623        let dex = Arc::clone(
1624            &get_dex_extended(chain.name, &DexType::UniswapV3)
1625                .expect("Ethereum UniswapV3 should be registered")
1626                .dex,
1627        );
1628        let pool_address = address!("4e68ccd3e89f51c3074ca5072bbac773960dfa36");
1629        let token0 = Token::new(
1630            Arc::clone(&chain),
1631            address!("C02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2"),
1632            "Wrapped Ether".to_string(),
1633            "WETH".to_string(),
1634            18,
1635        );
1636        let token1 = Token::new(
1637            Arc::clone(&chain),
1638            address!("dAC17F958D2ee523a2206206994597C13D831ec7"),
1639            "Tether USD".to_string(),
1640            "USDT".to_string(),
1641            6,
1642        );
1643
1644        Arc::new(Pool::new(
1645            chain,
1646            dex,
1647            pool_address,
1648            PoolIdentifier::from_address(pool_address),
1649            WETH_USDT_CREATION_BLOCK,
1650            token0,
1651            token1,
1652            Some(3_000),
1653            Some(60),
1654            UnixNanos::default(),
1655        ))
1656    }
1657}
nautilus_blockchain/data/client.rs

nautilus_blockchain/data/
client.rs
