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 nautilus_common::{
17    clients::DataClient,
18    defi::RequestPoolSnapshot,
19    live::get_runtime,
20    messages::{
21        DataEvent,
22        defi::{
23            DefiDataCommand, DefiRequestCommand, DefiSubscribeCommand, DefiUnsubscribeCommand,
24            SubscribeBlocks, SubscribePool, SubscribePoolFeeCollects, SubscribePoolFlashEvents,
25            SubscribePoolLiquidityUpdates, SubscribePoolSwaps, UnsubscribeBlocks, UnsubscribePool,
26            UnsubscribePoolFeeCollects, UnsubscribePoolFlashEvents,
27            UnsubscribePoolLiquidityUpdates, UnsubscribePoolSwaps,
28        },
29    },
30};
31use nautilus_model::{
32    defi::{DefiData, PoolIdentifier, SharedChain, validation::validate_address},
33    identifiers::{ClientId, Venue},
34};
35use ustr::Ustr;
36
37use crate::{
38    config::BlockchainDataClientConfig,
39    data::core::BlockchainDataClientCore,
40    exchanges::get_dex_extended,
41    rpc::{BlockchainRpcClient, types::BlockchainMessage},
42};
43
44/// A client for interacting with blockchain data from multiple sources.
45///
46/// The `BlockchainDataClient` serves as a facade that coordinates between different blockchain
47/// data providers, caching mechanisms, and contract interactions. It provides a unified interface
48/// for retrieving and processing blockchain data, particularly focused on DeFi protocols.
49///
50/// This client supports two primary data sources:
51/// 1. Direct RPC connections to blockchain nodes (via WebSocket).
52/// 2. HyperSync API for efficient historical data queries.
53#[derive(Debug)]
54pub struct BlockchainDataClient {
55    /// The client ID used to identify this client with the data engine.
56    pub client_id: ClientId,
57    /// The blockchain being targeted by this client instance.
58    pub chain: SharedChain,
59    /// Configuration parameters for the blockchain data client.
60    pub config: BlockchainDataClientConfig,
61    /// The core client instance that handles blockchain operations.
62    /// Wrapped in Option to allow moving it into the background processing task.
63    pub core_client: Option<BlockchainDataClientCore>,
64    /// Channel receiver for messages from the HyperSync client.
65    hypersync_rx: Option<tokio::sync::mpsc::UnboundedReceiver<BlockchainMessage>>,
66    /// Channel sender for messages to the HyperSync client.
67    hypersync_tx: Option<tokio::sync::mpsc::UnboundedSender<BlockchainMessage>>,
68    /// Channel sender for commands to be processed asynchronously.
69    command_tx: tokio::sync::mpsc::UnboundedSender<DefiDataCommand>,
70    /// Channel receiver for commands to be processed asynchronously.
71    command_rx: Option<tokio::sync::mpsc::UnboundedReceiver<DefiDataCommand>>,
72    /// Background task for processing messages.
73    process_task: Option<tokio::task::JoinHandle<()>>,
74    /// Cancellation token for graceful shutdown of background tasks.
75    cancellation_token: tokio_util::sync::CancellationToken,
76}
77
78impl BlockchainDataClient {
79    /// Creates a new [`BlockchainDataClient`] instance for the specified configuration.
80    #[must_use]
81    pub fn new(client_id: ClientId, config: BlockchainDataClientConfig) -> Self {
82        let chain = config.chain.clone();
83        let (command_tx, command_rx) = tokio::sync::mpsc::unbounded_channel();
84        let (hypersync_tx, hypersync_rx) = tokio::sync::mpsc::unbounded_channel();
85        Self {
86            client_id,
87            chain,
88            core_client: None,
89            config,
90            hypersync_rx: Some(hypersync_rx),
91            hypersync_tx: Some(hypersync_tx),
92            command_tx,
93            command_rx: Some(command_rx),
94            process_task: None,
95            cancellation_token: tokio_util::sync::CancellationToken::new(),
96        }
97    }
98
99    /// Spawns the main processing task that handles commands and blockchain data.
100    ///
101    /// This method creates a background task that:
102    /// 1. Processes subscription/unsubscription commands from the command channel
103    /// 2. Handles incoming blockchain data from HyperSync
104    /// 3. Processes RPC messages if RPC client is configured
105    /// 4. Routes processed data to subscribers
106    fn spawn_process_task(&mut self) {
107        let command_rx = if let Some(r) = self.command_rx.take() {
108            r
109        } else {
110            log::error!("Command receiver already taken, not spawning handler");
111            return;
112        };
113
114        let cancellation_token = self.cancellation_token.clone();
115
116        let data_tx = nautilus_common::live::runner::get_data_event_sender();
117
118        let mut hypersync_rx = self.hypersync_rx.take().unwrap();
119        let hypersync_tx = self.hypersync_tx.take();
120
121        let mut core_client = BlockchainDataClientCore::new(
122            self.config.clone(),
123            hypersync_tx,
124            Some(data_tx),
125            cancellation_token.clone(),
126        );
127
128        let handle = get_runtime().spawn(async move {
129            log::debug!("Started task 'process'");
130
131            if let Err(e) = core_client.connect().await {
132                // TODO: connect() could return more granular error types to distinguish
133                // cancellation from actual failures without string matching
134                if e.to_string().contains("cancelled") || e.to_string().contains("Sync cancelled") {
135                    log::warn!("Blockchain core client connection interrupted: {e}");
136                } else {
137                    log::error!("Failed to connect blockchain core client: {e}");
138                }
139                return;
140            }
141
142            let mut command_rx = command_rx;
143
144            loop {
145                tokio::select! {
146                    () = cancellation_token.cancelled() => {
147                        log::debug!("Received cancellation signal in Blockchain data client process task");
148                        core_client.disconnect().await;
149                        break;
150                    }
151                    command = command_rx.recv() => {
152                        if let Some(cmd) = command {
153                            match cmd {
154                                DefiDataCommand::Subscribe(cmd) => {
155                                    let chain = cmd.blockchain();
156                                    if chain != core_client.chain.name {
157                                        log::error!("Incorrect blockchain for subscribe command: {chain}");
158                                        continue;
159                                    }
160
161                                      if let Err(e) = Self::handle_subscribe_command(cmd, &mut core_client).await{
162                                        log::error!("Error processing subscribe command: {e}");
163                                    }
164                                }
165                                DefiDataCommand::Unsubscribe(cmd) => {
166                                    let chain = cmd.blockchain();
167                                    if chain != core_client.chain.name {
168                                        log::error!("Incorrect blockchain for subscribe command: {chain}");
169                                        continue;
170                                    }
171
172                                    if let Err(e) = Self::handle_unsubscribe_command(cmd, &mut core_client).await{
173                                        log::error!("Error processing subscribe command: {e}");
174                                    }
175                                }
176                                DefiDataCommand::Request(cmd) => {
177                                    if let Err(e) = Self::handle_request_command(cmd, &mut core_client).await {
178                                        log::error!("Error processing request command: {e}");
179                                    }
180                                }
181                            }
182                        } else {
183                            log::debug!("Command channel closed");
184                            break;
185                        }
186                    }
187                    data = hypersync_rx.recv() => {
188                        if let Some(msg) = data {
189                            let data_event = match msg {
190                                BlockchainMessage::Block(block) => {
191                                    // Fetch and process all subscribed events per DEX
192                                    for dex in core_client.cache.get_registered_dexes(){
193                                        let addresses = core_client.subscription_manager.get_subscribed_dex_contract_addresses(&dex);
194                                        if !addresses.is_empty() {
195                                            core_client.hypersync_client.process_block_dex_contract_events(
196                                                &dex,
197                                                block.number,
198                                                &addresses,
199                                                core_client.subscription_manager.get_dex_pool_swap_event_signature(&dex).unwrap(),
200                                                core_client.subscription_manager.get_dex_pool_mint_event_signature(&dex).unwrap(),
201                                                core_client.subscription_manager.get_dex_pool_burn_event_signature(&dex).unwrap(),
202                                            );
203                                        }
204                                    }
205
206                                    // Cache the block before its events are processed,
207                                    // so conversion can resolve ts_event.
208                                    if let Err(e) = core_client.cache.add_block(block.clone()).await {
209                                        log::error!("Failed to cache block {}: {e}", block.number);
210                                    }
211
212                                    Some(DataEvent::DeFi(DefiData::Block(block)))
213                                }
214                                BlockchainMessage::SwapEvent(swap_event) => {
215                                    match core_client.get_pool(&swap_event.pool_identifier) {
216                                        Ok(pool) => {
217                                            match core_client.process_pool_swap_event(&swap_event, pool){
218                                                Ok(swap) => Some(DataEvent::DeFi(DefiData::PoolSwap(swap))),
219                                                Err(e) => {
220                                                    log::error!("Error processing pool swap event: {e}");
221                                                    None
222                                                }
223                                            }
224                                        }
225                                        Err(e) => {
226                                            log::error!("Failed to get pool {} with error {:?}", swap_event.pool_identifier, e);
227                                            None
228                                        }
229                                    }
230                                }
231                                BlockchainMessage::BurnEvent(burn_event) => {
232                                    match core_client.get_pool(&burn_event.pool_identifier) {
233                                        Ok(pool) => {
234                                            let dex_extended = get_dex_extended(core_client.chain.name, &pool.dex.name).expect("Failed to get dex extended");
235                                            match core_client.process_pool_burn_event(
236                                                &burn_event,
237                                                pool,
238                                                dex_extended,
239                                            ){
240                                                Ok(update) => Some(DataEvent::DeFi(DefiData::PoolLiquidityUpdate(update))),
241                                                Err(e) => {
242                                                    log::error!("Error processing pool burn event: {e}");
243                                                    None
244                                                }
245                                            }
246                                        }
247                                        Err(e) => {
248                                            log::error!("Failed to get pool {} with error {:?}", burn_event.pool_identifier, e);
249                                            None
250                                        }
251                                    }
252                                }
253                                BlockchainMessage::MintEvent(mint_event) => {
254                                    match core_client.get_pool(&mint_event.pool_identifier) {
255                                        Ok(pool) => {
256                                            let dex_extended = get_dex_extended(core_client.chain.name,&pool.dex.name).expect("Failed to get dex extended");
257                                            match core_client.process_pool_mint_event(
258                                                &mint_event,
259                                                pool,
260                                                dex_extended,
261                                            ){
262                                                Ok(update) => Some(DataEvent::DeFi(DefiData::PoolLiquidityUpdate(update))),
263                                                Err(e) => {
264                                                    log::error!("Error processing pool mint event: {e}");
265                                                    None
266                                                }
267                                            }
268                                        }
269                                        Err(e) => {
270                                            log::error!("Failed to get pool {} with error {:?}", mint_event.pool_identifier, e);
271                                            None
272                                        }
273                                    }
274                                }
275                                BlockchainMessage::CollectEvent(collect_event) => {
276                                    match core_client.get_pool(&collect_event.pool_identifier) {
277                                        Ok(pool) => {
278                                            let dex_extended = get_dex_extended(core_client.chain.name, &pool.dex.name).expect("Failed to get dex extended");
279                                            match core_client.process_pool_collect_event(
280                                                &collect_event,
281                                                pool,
282                                                dex_extended,
283                                            ){
284                                                Ok(update) => Some(DataEvent::DeFi(DefiData::PoolFeeCollect(update))),
285                                                Err(e) => {
286                                                    log::error!("Error processing pool collect event: {e}");
287                                                    None
288                                                }
289                                            }
290                                        }
291                                        Err(e) => {
292                                            log::error!("Failed to get pool {} with error {:?}", collect_event.pool_identifier, e);
293                                            None
294                                        }
295                                    }
296                                }
297                            BlockchainMessage::FlashEvent(flash_event) => {
298                                    match core_client.get_pool(&flash_event.pool_identifier) {
299                                        Ok(pool) => {
300                                            match core_client.process_pool_flash_event(&flash_event,pool){
301                                                Ok(flash) => Some(DataEvent::DeFi(DefiData::PoolFlash(flash))),
302                                                Err(e) => {
303                                                    log::error!("Error processing pool flash event: {e}");
304                                                    None
305                                                }
306                                            }
307                                        }
308                                        Err(e) => {
309                                            log::error!("Failed to get pool {} with error {:?}", flash_event.pool_identifier, e);
310                                            None
311                                        }
312                                    }
313                                }
314                            };
315
316                            if let Some(event) = data_event {
317                                core_client.send_data(event);
318                            }
319                        } else {
320                            log::debug!("HyperSync data channel closed");
321                            break;
322                        }
323                    }
324                    msg = async {
325                        match core_client.rpc_client {
326                            Some(ref mut rpc_client) => rpc_client.next_rpc_message().await,
327                            None => std::future::pending().await,  // Never resolves
328                        }
329                    } => {
330                        // This branch only fires when we actually receive a message
331                        match msg {
332                            Ok(BlockchainMessage::Block(block)) => {
333                                let data = DataEvent::DeFi(DefiData::Block(block));
334                                core_client.send_data(data);
335                            },
336                            Ok(BlockchainMessage::SwapEvent(_)) => {
337                                log::warn!("RPC swap events are not yet supported");
338                            }
339                            Ok(BlockchainMessage::MintEvent(_)) => {
340                                log::warn!("RPC mint events are not yet supported");
341                            }
342                            Ok(BlockchainMessage::BurnEvent(_)) => {
343                                log::warn!("RPC burn events are not yet supported");
344                            }
345                            Ok(BlockchainMessage::CollectEvent(_)) => {
346                                log::warn!("RPC collect events are not yet supported");
347                            }
348                            Ok(BlockchainMessage::FlashEvent(_)) => {
349                                log::warn!("RPC flash events are not yet supported");
350                            }
351                            Err(e) => {
352                                log::error!("Error processing RPC message: {e}");
353                            }
354                        }
355                    }
356                }
357            }
358
359            log::debug!("Stopped task 'process'");
360        });
361
362        self.process_task = Some(handle);
363    }
364
365    /// Processes DeFi subscription commands to start receiving specific blockchain data.
366    async fn handle_subscribe_command(
367        command: DefiSubscribeCommand,
368        core_client: &mut BlockchainDataClientCore,
369    ) -> anyhow::Result<()> {
370        match command {
371            DefiSubscribeCommand::Blocks(_cmd) => {
372                log::debug!("Processing subscribe blocks command");
373
374                // Try RPC client first if available, otherwise use HyperSync
375                if let Some(ref mut rpc) = core_client.rpc_client {
376                    if let Err(e) = rpc.subscribe_blocks().await {
377                        log::warn!(
378                            "RPC blocks subscription failed: {e}, falling back to HyperSync"
379                        );
380                        core_client.hypersync_client.subscribe_blocks();
381                        tokio::task::yield_now().await;
382                    } else {
383                        log::debug!("Successfully subscribed to blocks via RPC");
384                    }
385                } else {
386                    log::debug!("Subscribing to blocks via HyperSync");
387                    core_client.hypersync_client.subscribe_blocks();
388                    tokio::task::yield_now().await;
389                }
390
391                Ok(())
392            }
393            DefiSubscribeCommand::Pool(cmd) => {
394                log::debug!(
395                    "Processing subscribe pool command for {}",
396                    cmd.instrument_id
397                );
398
399                if let Some(ref mut _rpc) = core_client.rpc_client {
400                    log::warn!("RPC pool subscription not yet implemented, using HyperSync");
401                }
402
403                if let Ok((_, dex)) = cmd.instrument_id.venue.parse_dex() {
404                    let pool_address = validate_address(cmd.instrument_id.symbol.as_str())
405                        .map_err(|e| {
406                            anyhow::anyhow!(
407                                "Invalid pool address '{}' failed with error: {:?}",
408                                cmd.instrument_id,
409                                e
410                            )
411                        })?;
412
413                    // Subscribe to all pool event types
414                    core_client
415                        .subscription_manager
416                        .subscribe_swaps(dex, pool_address);
417                    core_client
418                        .subscription_manager
419                        .subscribe_burns(dex, pool_address);
420                    core_client
421                        .subscription_manager
422                        .subscribe_mints(dex, pool_address);
423                    core_client
424                        .subscription_manager
425                        .subscribe_collects(dex, pool_address);
426                    core_client
427                        .subscription_manager
428                        .subscribe_flashes(dex, pool_address);
429
430                    log::debug!(
431                        "Subscribed to all pool events for {} at address {}",
432                        cmd.instrument_id,
433                        pool_address
434                    );
435                } else {
436                    anyhow::bail!(
437                        "Invalid venue {}, expected Blockchain DEX format",
438                        cmd.instrument_id.venue
439                    )
440                }
441
442                Ok(())
443            }
444            DefiSubscribeCommand::PoolSwaps(cmd) => {
445                log::debug!(
446                    "Processing subscribe pool swaps command for {}",
447                    cmd.instrument_id
448                );
449
450                if let Some(ref mut _rpc) = core_client.rpc_client {
451                    log::warn!("RPC pool swaps subscription not yet implemented, using HyperSync");
452                }
453
454                if let Ok((_, dex)) = cmd.instrument_id.venue.parse_dex() {
455                    let pool_address = validate_address(cmd.instrument_id.symbol.as_str())
456                        .map_err(|e| {
457                            anyhow::anyhow!(
458                                "Invalid pool swap address '{}' failed with error: {:?}",
459                                cmd.instrument_id,
460                                e
461                            )
462                        })?;
463                    core_client
464                        .subscription_manager
465                        .subscribe_swaps(dex, pool_address);
466                } else {
467                    anyhow::bail!(
468                        "Invalid venue {}, expected Blockchain DEX format",
469                        cmd.instrument_id.venue
470                    )
471                }
472
473                Ok(())
474            }
475            DefiSubscribeCommand::PoolLiquidityUpdates(cmd) => {
476                log::debug!(
477                    "Processing subscribe pool liquidity updates command for address: {}",
478                    cmd.instrument_id
479                );
480
481                if let Some(ref mut _rpc) = core_client.rpc_client {
482                    log::warn!(
483                        "RPC pool liquidity updates subscription not yet implemented, using HyperSync"
484                    );
485                }
486
487                if let Ok((_, dex)) = cmd.instrument_id.venue.parse_dex() {
488                    let pool_address = validate_address(cmd.instrument_id.symbol.as_str())
489                        .map_err(|_| {
490                            anyhow::anyhow!("Invalid pool swap address: {}", cmd.instrument_id)
491                        })?;
492                    core_client
493                        .subscription_manager
494                        .subscribe_burns(dex, pool_address);
495                    core_client
496                        .subscription_manager
497                        .subscribe_mints(dex, pool_address);
498                } else {
499                    anyhow::bail!(
500                        "Invalid venue {}, expected Blockchain DEX format",
501                        cmd.instrument_id.venue
502                    )
503                }
504
505                Ok(())
506            }
507            DefiSubscribeCommand::PoolFeeCollects(cmd) => {
508                log::debug!(
509                    "Processing subscribe pool fee collects command for address: {}",
510                    cmd.instrument_id
511                );
512
513                if let Some(ref mut _rpc) = core_client.rpc_client {
514                    log::warn!(
515                        "RPC pool fee collects subscription not yet implemented, using HyperSync"
516                    );
517                }
518
519                if let Ok((_, dex)) = cmd.instrument_id.venue.parse_dex() {
520                    let pool_address = validate_address(cmd.instrument_id.symbol.as_str())
521                        .map_err(|_| {
522                            anyhow::anyhow!(
523                                "Invalid pool fee collect address: {}",
524                                cmd.instrument_id
525                            )
526                        })?;
527                    core_client
528                        .subscription_manager
529                        .subscribe_collects(dex, pool_address);
530                } else {
531                    anyhow::bail!(
532                        "Invalid venue {}, expected Blockchain DEX format",
533                        cmd.instrument_id.venue
534                    )
535                }
536
537                Ok(())
538            }
539            DefiSubscribeCommand::PoolFlashEvents(cmd) => {
540                log::debug!(
541                    "Processing subscribe pool flash command for address: {}",
542                    cmd.instrument_id
543                );
544
545                if let Some(ref mut _rpc) = core_client.rpc_client {
546                    log::warn!(
547                        "RPC pool fee collects subscription not yet implemented, using HyperSync"
548                    );
549                }
550
551                if let Ok((_, dex)) = cmd.instrument_id.venue.parse_dex() {
552                    let pool_address = validate_address(cmd.instrument_id.symbol.as_str())
553                        .map_err(|_| {
554                            anyhow::anyhow!(
555                                "Invalid pool flash subscribe address: {}",
556                                cmd.instrument_id
557                            )
558                        })?;
559                    core_client
560                        .subscription_manager
561                        .subscribe_flashes(dex, pool_address);
562                } else {
563                    anyhow::bail!(
564                        "Invalid venue {}, expected Blockchain DEX format",
565                        cmd.instrument_id.venue
566                    )
567                }
568
569                Ok(())
570            }
571        }
572    }
573
574    /// Processes DeFi unsubscription commands to stop receiving specific blockchain data.
575    async fn handle_unsubscribe_command(
576        command: DefiUnsubscribeCommand,
577        core_client: &mut BlockchainDataClientCore,
578    ) -> anyhow::Result<()> {
579        match command {
580            DefiUnsubscribeCommand::Blocks(_cmd) => {
581                log::debug!("Processing unsubscribe blocks command");
582
583                // TODO: Implement RPC unsubscription when available
584                if core_client.rpc_client.is_some() {
585                    log::warn!("RPC blocks unsubscription not yet implemented");
586                }
587
588                // Use HyperSync client for unsubscription
589                core_client.hypersync_client.unsubscribe_blocks().await;
590                log::debug!("Unsubscribed from blocks via HyperSync");
591
592                Ok(())
593            }
594            DefiUnsubscribeCommand::Pool(cmd) => {
595                log::debug!(
596                    "Processing unsubscribe pool command for {}",
597                    cmd.instrument_id
598                );
599
600                if let Ok((_, dex)) = cmd.instrument_id.venue.parse_dex() {
601                    let pool_address = validate_address(cmd.instrument_id.symbol.as_str())
602                        .map_err(|_| {
603                            anyhow::anyhow!("Invalid pool address: {}", cmd.instrument_id)
604                        })?;
605
606                    // Unsubscribe from all pool event types
607                    core_client
608                        .subscription_manager
609                        .unsubscribe_swaps(dex, pool_address);
610                    core_client
611                        .subscription_manager
612                        .unsubscribe_burns(dex, pool_address);
613                    core_client
614                        .subscription_manager
615                        .unsubscribe_mints(dex, pool_address);
616                    core_client
617                        .subscription_manager
618                        .unsubscribe_collects(dex, pool_address);
619                    core_client
620                        .subscription_manager
621                        .unsubscribe_flashes(dex, pool_address);
622
623                    log::debug!(
624                        "Unsubscribed from all pool events for {} at address {}",
625                        cmd.instrument_id,
626                        pool_address
627                    );
628                } else {
629                    anyhow::bail!(
630                        "Invalid venue {}, expected Blockchain DEX format",
631                        cmd.instrument_id.venue
632                    )
633                }
634
635                Ok(())
636            }
637            DefiUnsubscribeCommand::PoolSwaps(cmd) => {
638                log::debug!("Processing unsubscribe pool swaps command");
639
640                if let Ok((_, dex)) = cmd.instrument_id.venue.parse_dex() {
641                    let pool_address = validate_address(cmd.instrument_id.symbol.as_str())
642                        .map_err(|_| {
643                            anyhow::anyhow!("Invalid pool swap address: {}", cmd.instrument_id)
644                        })?;
645                    core_client
646                        .subscription_manager
647                        .unsubscribe_swaps(dex, pool_address);
648                } else {
649                    anyhow::bail!(
650                        "Invalid venue {}, expected Blockchain DEX format",
651                        cmd.instrument_id.venue
652                    )
653                }
654
655                Ok(())
656            }
657            DefiUnsubscribeCommand::PoolLiquidityUpdates(cmd) => {
658                log::debug!(
659                    "Processing unsubscribe pool liquidity updates command for {}",
660                    cmd.instrument_id
661                );
662
663                if let Ok((_, dex)) = cmd.instrument_id.venue.parse_dex() {
664                    let pool_address = validate_address(cmd.instrument_id.symbol.as_str())
665                        .map_err(|_| {
666                            anyhow::anyhow!("Invalid pool swap address: {}", cmd.instrument_id)
667                        })?;
668                    core_client
669                        .subscription_manager
670                        .unsubscribe_burns(dex, pool_address);
671                    core_client
672                        .subscription_manager
673                        .unsubscribe_mints(dex, pool_address);
674                } else {
675                    anyhow::bail!(
676                        "Invalid venue {}, expected Blockchain DEX format",
677                        cmd.instrument_id.venue
678                    )
679                }
680
681                Ok(())
682            }
683            DefiUnsubscribeCommand::PoolFeeCollects(cmd) => {
684                log::debug!(
685                    "Processing unsubscribe pool fee collects command for {}",
686                    cmd.instrument_id
687                );
688
689                if let Ok((_, dex)) = cmd.instrument_id.venue.parse_dex() {
690                    let pool_address = validate_address(cmd.instrument_id.symbol.as_str())
691                        .map_err(|_| {
692                            anyhow::anyhow!(
693                                "Invalid pool fee collect address: {}",
694                                cmd.instrument_id
695                            )
696                        })?;
697                    core_client
698                        .subscription_manager
699                        .unsubscribe_collects(dex, pool_address);
700                } else {
701                    anyhow::bail!(
702                        "Invalid venue {}, expected Blockchain DEX format",
703                        cmd.instrument_id.venue
704                    )
705                }
706
707                Ok(())
708            }
709            DefiUnsubscribeCommand::PoolFlashEvents(cmd) => {
710                log::debug!(
711                    "Processing unsubscribe pool flash command for {}",
712                    cmd.instrument_id
713                );
714
715                if let Ok((_, dex)) = cmd.instrument_id.venue.parse_dex() {
716                    let pool_address = validate_address(cmd.instrument_id.symbol.as_str())
717                        .map_err(|_| {
718                            anyhow::anyhow!("Invalid pool flash address: {}", cmd.instrument_id)
719                        })?;
720                    core_client
721                        .subscription_manager
722                        .unsubscribe_flashes(dex, pool_address);
723                } else {
724                    anyhow::bail!(
725                        "Invalid venue {}, expected Blockchain DEX format",
726                        cmd.instrument_id.venue
727                    )
728                }
729
730                Ok(())
731            }
732        }
733    }
734
735    /// Processes DeFi request commands to fetch specific blockchain data.
736    async fn handle_request_command(
737        command: DefiRequestCommand,
738        core_client: &mut BlockchainDataClientCore,
739    ) -> anyhow::Result<()> {
740        match command {
741            DefiRequestCommand::PoolSnapshot(cmd) => {
742                log::debug!("Processing pool snapshot request for {}", cmd.instrument_id);
743
744                let pool_address =
745                    validate_address(cmd.instrument_id.symbol.as_str()).map_err(|e| {
746                        anyhow::anyhow!(
747                            "Invalid pool address '{}' failed with error: {:?}",
748                            cmd.instrument_id,
749                            e
750                        )
751                    })?;
752
753                let pool_identifier =
754                    PoolIdentifier::Address(Ustr::from(&pool_address.to_string()));
755
756                match core_client.get_pool(&pool_identifier) {
757                    Ok(pool) => {
758                        let pool = pool.clone();
759                        log::debug!("Found pool for snapshot request: {}", cmd.instrument_id);
760
761                        // Send the pool definition
762                        let pool_data = DataEvent::DeFi(DefiData::Pool(pool.as_ref().clone()));
763                        core_client.send_data(pool_data);
764
765                        match core_client
766                            .bootstrap_latest_pool_profiler(&pool, None)
767                            .await
768                        {
769                            Ok((profiler, already_valid)) => match profiler.extract_snapshot() {
770                                Ok(snapshot) => {
771                                    log::debug!(
772                                        "Saving pool snapshot with {} positions and {} ticks to database...",
773                                        snapshot.positions.len(),
774                                        snapshot.ticks.len()
775                                    );
776                                    core_client
777                                        .cache
778                                        .add_pool_snapshot(
779                                            &pool.dex.name,
780                                            &pool.pool_identifier,
781                                            &snapshot,
782                                        )
783                                        .await?;
784
785                                    // If the snapshot is usable, send it back to the data engine.
786                                    if core_client
787                                        .check_snapshot_validity(&profiler, already_valid)
788                                        .await?
789                                        .is_usable()
790                                    {
791                                        let snapshot_data =
792                                            DataEvent::DeFi(DefiData::PoolSnapshot(snapshot));
793                                        core_client.send_data(snapshot_data);
794                                    }
795                                }
796                                Err(e) => log::error!(
797                                    "Failed to extract snapshot for {}: {e}",
798                                    cmd.instrument_id
799                                ),
800                            },
801                            Err(e) => log::error!(
802                                "Failed to bootstrap pool profiler for {} and extract snapshot with error {e}",
803                                cmd.instrument_id
804                            ),
805                        }
806                    }
807                    Err(e) => {
808                        log::warn!("Pool {} not found in cache: {e}", cmd.instrument_id);
809                    }
810                }
811
812                Ok(())
813            }
814        }
815    }
816
817    /// Waits for the background processing task to complete.
818    ///
819    /// This method blocks until the spawned process task finishes execution,
820    /// which typically happens after a shutdown signal is sent.
821    pub async fn await_process_task_close(&mut self) {
822        if let Some(handle) = self.process_task.take()
823            && let Err(e) = handle.await
824        {
825            log::error!("Process task join error: {e}");
826        }
827    }
828}
829
830#[async_trait::async_trait(?Send)]
831impl DataClient for BlockchainDataClient {
832    fn client_id(&self) -> ClientId {
833        self.client_id
834    }
835
836    fn venue(&self) -> Option<Venue> {
837        // Blockchain data clients don't map to a single venue since they can provide
838        // data for multiple DEXs across the blockchain
839        None
840    }
841
842    fn start(&mut self) -> anyhow::Result<()> {
843        log::info!(
844            "Starting blockchain data client: chain_name={}, dex_ids={:?}, use_hypersync_for_live_data={}, proxy_url={:?}",
845            self.chain.name,
846            self.config.dex_ids,
847            self.config.use_hypersync_for_live_data,
848            self.config.proxy_url
849        );
850        Ok(())
851    }
852
853    fn stop(&mut self) -> anyhow::Result<()> {
854        log::info!(
855            "Stopping blockchain data client for '{chain_name}'",
856            chain_name = self.chain.name
857        );
858        self.cancellation_token.cancel();
859
860        // Create fresh token for next start cycle
861        self.cancellation_token = tokio_util::sync::CancellationToken::new();
862        Ok(())
863    }
864
865    fn reset(&mut self) -> anyhow::Result<()> {
866        log::info!(
867            "Resetting blockchain data client for '{chain_name}'",
868            chain_name = self.chain.name
869        );
870        self.cancellation_token = tokio_util::sync::CancellationToken::new();
871        Ok(())
872    }
873
874    fn dispose(&mut self) -> anyhow::Result<()> {
875        log::info!(
876            "Disposing blockchain data client for '{chain_name}'",
877            chain_name = self.chain.name
878        );
879        Ok(())
880    }
881
882    async fn connect(&mut self) -> anyhow::Result<()> {
883        log::info!(
884            "Connecting blockchain data client for '{}'",
885            self.chain.name
886        );
887
888        if self.process_task.is_none() {
889            self.spawn_process_task();
890        }
891
892        Ok(())
893    }
894
895    async fn disconnect(&mut self) -> anyhow::Result<()> {
896        log::info!(
897            "Disconnecting blockchain data client for '{}'",
898            self.chain.name
899        );
900
901        self.cancellation_token.cancel();
902        self.await_process_task_close().await;
903
904        // Create fresh token and channels for next connect cycle
905        self.cancellation_token = tokio_util::sync::CancellationToken::new();
906        let (hypersync_tx, hypersync_rx) = tokio::sync::mpsc::unbounded_channel();
907        self.hypersync_tx = Some(hypersync_tx);
908        self.hypersync_rx = Some(hypersync_rx);
909        let (command_tx, command_rx) = tokio::sync::mpsc::unbounded_channel();
910        self.command_tx = command_tx;
911        self.command_rx = Some(command_rx);
912
913        Ok(())
914    }
915
916    fn is_connected(&self) -> bool {
917        // TODO: Improve connection detection
918        // For now, we'll assume connected if we have either RPC or HyperSync configured
919        true
920    }
921
922    fn is_disconnected(&self) -> bool {
923        !self.is_connected()
924    }
925
926    fn subscribe_blocks(&mut self, cmd: SubscribeBlocks) -> anyhow::Result<()> {
927        let command = DefiDataCommand::Subscribe(DefiSubscribeCommand::Blocks(cmd));
928        self.command_tx.send(command)?;
929        Ok(())
930    }
931
932    fn subscribe_pool(&mut self, cmd: SubscribePool) -> anyhow::Result<()> {
933        let command = DefiDataCommand::Subscribe(DefiSubscribeCommand::Pool(cmd));
934        self.command_tx.send(command)?;
935        Ok(())
936    }
937
938    fn subscribe_pool_swaps(&mut self, cmd: SubscribePoolSwaps) -> anyhow::Result<()> {
939        let command = DefiDataCommand::Subscribe(DefiSubscribeCommand::PoolSwaps(cmd));
940        self.command_tx.send(command)?;
941        Ok(())
942    }
943
944    fn subscribe_pool_liquidity_updates(
945        &mut self,
946        cmd: SubscribePoolLiquidityUpdates,
947    ) -> anyhow::Result<()> {
948        let command = DefiDataCommand::Subscribe(DefiSubscribeCommand::PoolLiquidityUpdates(cmd));
949        self.command_tx.send(command)?;
950        Ok(())
951    }
952
953    fn subscribe_pool_fee_collects(&mut self, cmd: SubscribePoolFeeCollects) -> anyhow::Result<()> {
954        let command = DefiDataCommand::Subscribe(DefiSubscribeCommand::PoolFeeCollects(cmd));
955        self.command_tx.send(command)?;
956        Ok(())
957    }
958
959    fn subscribe_pool_flash_events(&mut self, cmd: SubscribePoolFlashEvents) -> anyhow::Result<()> {
960        let command = DefiDataCommand::Subscribe(DefiSubscribeCommand::PoolFlashEvents(cmd));
961        self.command_tx.send(command)?;
962        Ok(())
963    }
964
965    fn unsubscribe_blocks(&mut self, cmd: &UnsubscribeBlocks) -> anyhow::Result<()> {
966        let command = DefiDataCommand::Unsubscribe(DefiUnsubscribeCommand::Blocks(cmd.clone()));
967        self.command_tx.send(command)?;
968        Ok(())
969    }
970
971    fn unsubscribe_pool(&mut self, cmd: &UnsubscribePool) -> anyhow::Result<()> {
972        let command = DefiDataCommand::Unsubscribe(DefiUnsubscribeCommand::Pool(cmd.clone()));
973        self.command_tx.send(command)?;
974        Ok(())
975    }
976
977    fn unsubscribe_pool_swaps(&mut self, cmd: &UnsubscribePoolSwaps) -> anyhow::Result<()> {
978        let command = DefiDataCommand::Unsubscribe(DefiUnsubscribeCommand::PoolSwaps(cmd.clone()));
979        self.command_tx.send(command)?;
980        Ok(())
981    }
982
983    fn unsubscribe_pool_liquidity_updates(
984        &mut self,
985        cmd: &UnsubscribePoolLiquidityUpdates,
986    ) -> anyhow::Result<()> {
987        let command =
988            DefiDataCommand::Unsubscribe(DefiUnsubscribeCommand::PoolLiquidityUpdates(cmd.clone()));
989        self.command_tx.send(command)?;
990        Ok(())
991    }
992
993    fn unsubscribe_pool_fee_collects(
994        &mut self,
995        cmd: &UnsubscribePoolFeeCollects,
996    ) -> anyhow::Result<()> {
997        let command =
998            DefiDataCommand::Unsubscribe(DefiUnsubscribeCommand::PoolFeeCollects(cmd.clone()));
999        self.command_tx.send(command)?;
1000        Ok(())
1001    }
1002
1003    fn unsubscribe_pool_flash_events(
1004        &mut self,
1005        cmd: &UnsubscribePoolFlashEvents,
1006    ) -> anyhow::Result<()> {
1007        let command =
1008            DefiDataCommand::Unsubscribe(DefiUnsubscribeCommand::PoolFlashEvents(cmd.clone()));
1009        self.command_tx.send(command)?;
1010        Ok(())
1011    }
1012
1013    fn request_pool_snapshot(&self, cmd: RequestPoolSnapshot) -> anyhow::Result<()> {
1014        let command = DefiDataCommand::Request(DefiRequestCommand::PoolSnapshot(cmd));
1015        self.command_tx.send(command)?;
1016        Ok(())
1017    }
1018}
1019
1020#[cfg(test)]
1021mod tests {
1022    use std::sync::Arc;
1023
1024    use alloy::primitives::address;
1025    use nautilus_common::defi::RequestPoolSnapshot;
1026    use nautilus_core::{UUID4, UnixNanos};
1027    use nautilus_model::{
1028        defi::{Chain, DexType, Pool, PoolIdentifier, Token},
1029        identifiers::ClientId,
1030    };
1031    use tokio_util::sync::CancellationToken;
1032
1033    use super::*;
1034
1035    const WETH_USDT_CREATION_BLOCK: u64 = 12_375_326;
1036
1037    #[tokio::test(flavor = "multi_thread")]
1038    #[ignore = "requires ENVIO_API_TOKEN and live HyperSync access"]
1039    async fn pool_snapshot_request_does_not_emit_snapshot_when_bootstrap_fails() {
1040        std::env::var("ENVIO_API_TOKEN").expect("ENVIO_API_TOKEN must be set");
1041
1042        let pool = weth_usdt_pool();
1043        let instrument_id = pool.instrument_id;
1044        let (hypersync_tx, _hypersync_rx) = tokio::sync::mpsc::unbounded_channel();
1045        let (data_tx, mut data_rx) = tokio::sync::mpsc::unbounded_channel();
1046        let config = BlockchainDataClientConfig::builder()
1047            .chain(pool.chain.clone())
1048            .dex_ids(vec![DexType::UniswapV3])
1049            .http_rpc_url("http://127.0.0.1:9".to_string())
1050            .use_hypersync_for_live_data(true)
1051            .maybe_from_block(Some(WETH_USDT_CREATION_BLOCK))
1052            .build();
1053        let mut core = BlockchainDataClientCore::new(
1054            config,
1055            Some(hypersync_tx),
1056            Some(data_tx),
1057            CancellationToken::new(),
1058        );
1059        core.cache
1060            .add_pool(pool.as_ref().clone())
1061            .await
1062            .expect("Pool should be added to in-memory cache");
1063
1064        let request = RequestPoolSnapshot::new(
1065            instrument_id,
1066            Some(ClientId::new("BLOCKCHAIN")),
1067            UUID4::new(),
1068            UnixNanos::default(),
1069            None,
1070        );
1071
1072        BlockchainDataClient::handle_request_command(
1073            DefiRequestCommand::PoolSnapshot(request),
1074            &mut core,
1075        )
1076        .await
1077        .expect("Bootstrap failure should not fail the request handler");
1078
1079        let mut events = Vec::new();
1080        while let Ok(event) = data_rx.try_recv() {
1081            events.push(event);
1082        }
1083
1084        assert_eq!(events.len(), 1);
1085        match &events[0] {
1086            DataEvent::DeFi(DefiData::Pool(pool)) => {
1087                assert_eq!(pool.instrument_id, instrument_id);
1088            }
1089            _ => panic!("expected only the pool definition event"),
1090        }
1091        assert!(
1092            events
1093                .iter()
1094                .all(|event| !matches!(event, DataEvent::DeFi(DefiData::PoolSnapshot(_))))
1095        );
1096    }
1097
1098    fn weth_usdt_pool() -> Arc<Pool> {
1099        let chain = Arc::new(
1100            Chain::from_chain_id(1)
1101                .expect("Ethereum chain should exist")
1102                .clone(),
1103        );
1104        let dex = get_dex_extended(chain.name, &DexType::UniswapV3)
1105            .expect("Ethereum UniswapV3 should be registered")
1106            .dex
1107            .clone();
1108        let pool_address = address!("4e68ccd3e89f51c3074ca5072bbac773960dfa36");
1109        let token0 = Token::new(
1110            chain.clone(),
1111            address!("C02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2"),
1112            "Wrapped Ether".to_string(),
1113            "WETH".to_string(),
1114            18,
1115        );
1116        let token1 = Token::new(
1117            chain.clone(),
1118            address!("dAC17F958D2ee523a2206206994597C13D831ec7"),
1119            "Tether USD".to_string(),
1120            "USDT".to_string(),
1121            6,
1122        );
1123
1124        Arc::new(Pool::new(
1125            chain,
1126            dex,
1127            pool_address,
1128            PoolIdentifier::from_address(pool_address),
1129            WETH_USDT_CREATION_BLOCK,
1130            token0,
1131            token1,
1132            Some(3_000),
1133            Some(60),
1134            UnixNanos::default(),
1135        ))
1136    }
1137}
nautilus_blockchain/data/client.rs

nautilus_blockchain/data/
client.rs
