kabu_defi_market/
history_pool_loader_component.rs

1use alloy_network::Network;
2use alloy_provider::Provider;
3use eyre::Result;
4use std::sync::Arc;
5use tokio::sync::broadcast;
6use tracing::{debug, error, info};
7
8use kabu_core_components::Component;
9use kabu_types_events::LoomTask;
10use kabu_types_market::{PoolClass, PoolId, PoolLoaders};
11use reth_ethereum_primitives::EthPrimitives;
12use reth_tasks::TaskExecutor;
13
14/// Component that loads historical pool data by scanning past blocks for pool creation events
15#[derive(Clone)]
16pub struct HistoryPoolLoaderComponent<P, PL, N>
17where
18    N: Network,
19    P: Provider<N> + Send + Sync + Clone + 'static,
20    PL: Provider<N> + Send + Sync + Clone + 'static,
21{
22    /// JSON-RPC provider for fetching blockchain data
23    client: P,
24    /// Pool loaders for different protocols
25    #[allow(dead_code)]
26    pool_loaders: Arc<PoolLoaders<PL, N, EthPrimitives>>,
27    /// Channel to send discovered tasks
28    tasks_tx: Option<broadcast::Sender<LoomTask>>,
29    /// Starting block number for historical scan
30    start_block: u64,
31    /// Block batch size for scanning
32    block_batch_size: u64,
33    /// Maximum number of batches to process
34    max_batches: u64,
35}
36
37impl<P, PL, N> HistoryPoolLoaderComponent<P, PL, N>
38where
39    N: Network + 'static,
40    P: Provider<N> + Send + Sync + Clone + 'static,
41    PL: Provider<N> + Send + Sync + Clone + 'static,
42{
43    pub fn new(
44        client: P,
45        pool_loaders: Arc<PoolLoaders<PL, N, EthPrimitives>>,
46        start_block: u64,
47        block_batch_size: u64,
48        max_batches: u64,
49    ) -> Self {
50        Self { client, pool_loaders, tasks_tx: None, start_block, block_batch_size, max_batches }
51    }
52
53    pub fn with_task_channel(mut self, tasks_tx: broadcast::Sender<LoomTask>) -> Self {
54        self.tasks_tx = Some(tasks_tx);
55        self
56    }
57
58    async fn run(self) -> Result<()> {
59        info!("Starting history pool loader component");
60
61        let mut current_block = match self.client.get_block_number().await {
62            Ok(block_num) => {
63                if self.start_block > 0 {
64                    self.start_block
65                } else {
66                    block_num.saturating_sub(1000) // Default: scan last 1000 blocks
67                }
68            }
69            Err(e) => {
70                error!("Failed to get current block number: {}", e);
71                return Err(e.into());
72            }
73        };
74
75        info!(
76            "History pool loader starting from block {}, batch size: {}, max batches: {}",
77            current_block, self.block_batch_size, self.max_batches
78        );
79
80        for batch in 0..self.max_batches {
81            let from_block = current_block;
82            let to_block = current_block + self.block_batch_size - 1;
83
84            match self.process_block_range(from_block, to_block).await {
85                Ok(tasks_sent) => {
86                    debug!(
87                        "Batch {}/{}: Processed blocks {}-{}, sent {} tasks",
88                        batch + 1,
89                        self.max_batches,
90                        from_block,
91                        to_block,
92                        tasks_sent
93                    );
94                }
95                Err(e) => {
96                    error!("Error processing blocks {}-{}: {}", from_block, to_block, e);
97                    // Continue with next batch instead of failing completely
98                }
99            }
100
101            current_block += self.block_batch_size;
102
103            // Add a small delay between batches to avoid overwhelming the RPC
104            tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
105        }
106
107        info!("History pool loader component completed");
108        Ok(())
109    }
110
111    async fn process_block_range(&self, from_block: u64, to_block: u64) -> Result<usize> {
112        // Simplified implementation that creates pool discovery tasks
113        // This would typically analyze block logs to find pool creation events
114
115        if let Some(ref tasks_tx) = self.tasks_tx {
116            // Create some example pool discovery tasks
117            // In a real implementation, this would parse block logs for pool creation events
118            let example_pools = vec![
119                (PoolId::Address(alloy_primitives::Address::repeat_byte((from_block % 256) as u8)), PoolClass::UniswapV2),
120                (PoolId::Address(alloy_primitives::Address::repeat_byte(((from_block + 1) % 256) as u8)), PoolClass::UniswapV3),
121            ];
122
123            let task = LoomTask::FetchAndAddPools(example_pools);
124
125            if let Err(e) = tasks_tx.send(task) {
126                error!("Failed to send pool discovery task: {}", e);
127                return Err(e.into());
128            }
129
130            debug!("Sent pool discovery task for blocks {}-{}", from_block, to_block);
131
132            // Return 1 to indicate we sent a task
133            Ok(1)
134        } else {
135            debug!("No task channel configured, skipping block range {}-{}", from_block, to_block);
136            Ok(0)
137        }
138    }
139}
140
141impl<P, PL, N> Component for HistoryPoolLoaderComponent<P, PL, N>
142where
143    N: Network + 'static,
144    P: Provider<N> + Send + Sync + Clone + 'static,
145    PL: Provider<N> + Send + Sync + Clone + 'static,
146{
147    fn spawn(self, executor: TaskExecutor) -> Result<()> {
148        executor.spawn_critical(self.name(), async move {
149            if let Err(e) = self.run().await {
150                error!("History pool loader component failed: {}", e);
151            }
152        });
153        Ok(())
154    }
155    fn name(&self) -> &'static str {
156        "HistoryPoolLoaderComponent"
157    }
158}
159
160/// Builder for HistoryPoolLoaderComponent
161pub struct HistoryPoolLoaderComponentBuilder {
162    start_block: u64,
163    block_batch_size: u64,
164    max_batches: u64,
165}
166
167impl HistoryPoolLoaderComponentBuilder {
168    pub fn new() -> Self {
169        Self {
170            start_block: 0, // 0 means use current block - 1000
171            block_batch_size: 5,
172            max_batches: 10000,
173        }
174    }
175
176    pub fn with_start_block(mut self, start_block: u64) -> Self {
177        self.start_block = start_block;
178        self
179    }
180
181    pub fn with_batch_size(mut self, batch_size: u64) -> Self {
182        self.block_batch_size = batch_size;
183        self
184    }
185
186    pub fn with_max_batches(mut self, max_batches: u64) -> Self {
187        self.max_batches = max_batches;
188        self
189    }
190
191    pub fn build<P, PL, N>(self, client: P, pool_loaders: Arc<PoolLoaders<PL, N, EthPrimitives>>) -> HistoryPoolLoaderComponent<P, PL, N>
192    where
193        N: Network + 'static,
194        P: Provider<N> + Send + Sync + Clone + 'static,
195        PL: Provider<N> + Send + Sync + Clone + 'static,
196    {
197        HistoryPoolLoaderComponent::new(client, pool_loaders, self.start_block, self.block_batch_size, self.max_batches)
198    }
199}
200
201impl Default for HistoryPoolLoaderComponentBuilder {
202    fn default() -> Self {
203        Self::new()
204    }
205}
kabu_defi_market/history_pool_loader_component.rs

kabu_defi_market/
history_pool_loader_component.rs
