Struct BlockchainDataClientCore 

pub struct BlockchainDataClientCore {
    pub chain: SharedChain,
    pub config: BlockchainDataClientConfig,
    pub cache: BlockchainCache,
    pub hypersync_client: HyperSyncClient,
    pub rpc_client: Option<BlockchainRpcClientAny>,
    pub subscription_manager: DefiDataSubscriptionManager,
    /* private fields */
}

Core blockchain data client responsible for fetching, processing, and caching blockchain data.

This struct encapsulates the core functionality for interacting with blockchain networks, including syncing historical data, processing real-time events, and managing cached entities.

Fields

chain: SharedChain

The blockchain being targeted by this client instance.

config: BlockchainDataClientConfig

The configuration for the data client.

cache: BlockchainCache

Local cache for blockchain entities.

hypersync_client: HyperSyncClient

Client for the HyperSync data indexing service.

rpc_client: Option<BlockchainRpcClientAny>

Optional WebSocket RPC client for direct blockchain node communication.

subscription_manager: DefiDataSubscriptionManager

Manages subscriptions for various DEX events (swaps, mints, burns).

Implementations

impl BlockchainDataClientCore

pub fn new( config: BlockchainDataClientConfig, hypersync_tx: Option<UnboundedSender<BlockchainMessage>>, data_tx: Option<UnboundedSender<DataEvent>>, cancellation_token: CancellationToken, ) -> Self

Creates a new instance of BlockchainDataClientCore.

Panics

Panics if use_hypersync_for_live_data is false but wss_rpc_url is None.

pub async fn initialize_cache_database(&mut self)

Initializes the database connection for the blockchain cache.

pub async fn connect(&mut self) -> Result<()>

Establishes connections to all configured data sources and initializes the cache.

Errors

Returns an error if cache initialization or connection setup fails.

pub async fn sync_blocks_checked( &mut self, from_block: u64, to_block: Option<u64>, ) -> Result<()>

Syncs blocks with consistency checks to ensure data integrity.

Errors

Returns an error if block syncing fails or if consistency checks fail.

pub async fn sync_blocks( &mut self, from_block: u64, to_block: Option<u64>, use_copy_command: bool, ) -> Result<()>

Synchronizes blockchain data by fetching and caching all blocks from the starting block to the current chain head.

Errors

Returns an error if block fetching, caching, or database operations fail.

pub async fn sync_pool_events( &mut self, dex: &DexType, pool_address: &Address, from_block: Option<u64>, to_block: Option<u64>, reset: bool, ) -> Result<()>

Synchronizes all events for a specific pool within the given block range.

Errors

Returns an error if event syncing, parsing, or database operations fail.

pub fn process_pool_swap_event( &self, swap_event: &SwapEvent, pool: &SharedPool, dex_extended: &DexExtended, ) -> Result<PoolSwap>

Returns an error if swap event processing fails.

Panics

Panics if swap event conversion to trade data fails.

pub fn process_pool_mint_event( &self, mint_event: &MintEvent, pool: &SharedPool, dex_extended: &DexExtended, ) -> Result<PoolLiquidityUpdate>

Processes a mint event (liquidity addition) and converts it to a PoolLiquidityUpdate.

Errors

Returns an error if mint event processing fails or if the liquidity update creation fails.

pub fn process_pool_burn_event( &self, burn_event: &BurnEvent, pool: &SharedPool, dex_extended: &DexExtended, ) -> Result<PoolLiquidityUpdate>

Processes a burn event (liquidity removal) and converts it to a PoolLiquidityUpdate. Processes a pool burn event and converts it to a liquidity update.

Errors

Returns an error if the burn event processing fails or if the liquidity update creation fails.

pub fn process_pool_collect_event( &self, collect_event: &CollectEvent, pool: &SharedPool, dex_extended: &DexExtended, ) -> Result<PoolFeeCollect>

Processes a pool collect event and converts it to a fee collection.

Errors

Returns an error if the collect event processing fails or if the fee collection creation fails.

pub fn process_pool_flash_event( &self, flash_event: &FlashEvent, pool: &SharedPool, ) -> Result<PoolFlash>

Processes a pool flash event and converts it to a flash loan.

Errors

Returns an error if the flash event processing fails or if the flash loan creation fails.

pub async fn sync_exchange_pools( &mut self, dex: &DexType, from_block: u64, to_block: Option<u64>, reset: bool, ) -> Result<()>

Synchronizes all pools and their tokens for a specific DEX within the given block range.

This method performs a comprehensive sync of:

1. Pool creation events from the DEX factory
2. Token metadata for all tokens in discovered pools
3. Pool entities with proper token associations

Errors

Returns an error if syncing pools, tokens, or DEX operations fail.

pub async fn register_dex_exchange(&mut self, dex_id: DexType) -> Result<()>

Registers a decentralized exchange for data collection and event monitoring.

Registration involves:

1. Adding the DEX to the cache
2. Loading existing pools for the DEX
3. Configuring event signatures for subscriptions

Errors

Returns an error if DEX registration, cache operations, or pool loading fails.

pub async fn bootstrap_latest_pool_profiler( &mut self, pool: &SharedPool, ) -> Result<(PoolProfiler, bool)>

Bootstraps a [PoolProfiler] with the latest state for a given pool.

Uses two paths depending on whether the pool has been synced to the database:

• Never synced: Streams events from HyperSync → restores from on-chain RPC → returns (profiler, true)
• Previously synced: Syncs new events to DB → streams from DB → returns (profiler, false)

Both paths restore from the latest valid snapshot first (if available), otherwise initialize with pool’s initial price.

Returns

• PoolProfiler: Hydrated profiler with current pool state
• bool: true if constructed from RPC (already valid), false if from DB (needs validation)

Errors

Returns an error if database is not initialized or event processing fails.

pub async fn check_snapshot_validity( &self, profiler: &PoolProfiler, already_validated: bool, ) -> Result<bool>

Validates a pool profiler’s state against on-chain data for accuracy verification.

This method performs integrity checking by comparing the profiler’s internal state (positions, ticks, liquidity) with the actual on-chain smart contract state. For UniswapV3 pools, it fetches current on-chain data and verifies that the profiler’s tracked state matches. If validation succeeds or is bypassed, the snapshot is marked as valid in the database.

Errors

Returns an error if database operations fail when marking the snapshot as valid.

pub async fn replay_pool_events( &self, pool: &Pool, dex: &SharedDex, ) -> Result<()>

Replays historical events for a pool to hydrate its profiler state.

Streams all historical swap, liquidity, and fee collect events from the database and sends them through the normal data event pipeline to build up pool profiler state.

Errors

Returns an error if database streaming fails or event processing fails.

pub fn get_pool(&self, pool_address: &Address) -> Result<&SharedPool>

Retrieves a pool from the cache by its address.

Errors

Returns an error if the pool is not registered in the cache.

pub fn send_data(&self, data: DataEvent)

Sends a data event to all subscribers through the data channel.

pub async fn disconnect(&mut self)

Disconnects all active connections and cleanup resources.

This method should be called when shutting down the client to ensure proper cleanup of network connections and background tasks.

Trait Implementations

impl Debug for BlockchainDataClientCore

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.