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

nautilus_blockchain/data/
client.rs
