kabu_defi_market/
protocol_pool_loader_component.rs

1use std::sync::Arc;
2use tokio::sync::broadcast;
3
4use alloy_network::Network;
5use alloy_provider::Provider;
6use eyre::Result;
7use tracing::{error, info};
8
9use kabu_core_components::Component;
10use kabu_types_events::LoomTask;
11use kabu_types_market::{PoolClass, PoolId, PoolLoaders};
12use reth_tasks::TaskExecutor;
13
14/// Component that loads pools by protocol type using protocol-specific loaders
15#[derive(Clone)]
16pub struct ProtocolPoolLoaderComponent<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    #[allow(dead_code)]
24    client: P,
25    /// Pool loaders for different protocols
26    #[allow(dead_code)]
27    pool_loaders: Arc<PoolLoaders<PL, N>>,
28    /// Channel to send discovered tasks
29    tasks_tx: Option<broadcast::Sender<LoomTask>>,
30}
31
32impl<P, PL, N> ProtocolPoolLoaderComponent<P, PL, N>
33where
34    N: Network + 'static,
35    P: Provider<N> + Send + Sync + Clone + 'static,
36    PL: Provider<N> + Send + Sync + Clone + 'static,
37{
38    pub fn new(client: P, pool_loaders: Arc<PoolLoaders<PL, N>>) -> Self {
39        Self { client, pool_loaders, tasks_tx: None }
40    }
41
42    pub fn with_task_channel(mut self, tasks_tx: broadcast::Sender<LoomTask>) -> Self {
43        self.tasks_tx = Some(tasks_tx);
44        self
45    }
46
47    async fn run(self) -> Result<()> {
48        info!("Starting protocol pool loader component");
49
50        // Spawn a task for each protocol type
51        let mut tasks = Vec::new();
52
53        // Simplified: just process a few common pool classes
54        let pool_classes = vec![PoolClass::UniswapV2, PoolClass::UniswapV3, PoolClass::Curve];
55
56        for pool_class in pool_classes {
57            let tasks_tx_clone = self.tasks_tx.clone();
58
59            let task = tokio::task::spawn(async move {
60                if let Err(e) = Self::process_protocol_loader(
61                    pool_class,
62                    Arc::new(()), // Dummy loader
63                    tasks_tx_clone,
64                )
65                .await
66                {
67                    error!("Protocol loader failed for {}: {}", pool_class, e);
68                }
69            });
70
71            tasks.push(task);
72        }
73
74        // Wait for all protocol loaders to complete
75        for task in tasks {
76            if let Err(e) = task.await {
77                error!("Protocol loader task failed: {}", e);
78            }
79        }
80
81        info!("Protocol pool loader component completed");
82        Ok(())
83    }
84
85    async fn process_protocol_loader(
86        pool_class: PoolClass,
87        _pool_loader: Arc<dyn Send + Sync>,
88        tasks_tx: Option<broadcast::Sender<LoomTask>>,
89    ) -> Result<()> {
90        info!("Protocol loader started for {}", pool_class);
91
92        // Simplified implementation that creates example protocol discovery tasks
93        if let Some(tx) = tasks_tx {
94            // Create example pools for this protocol
95            let example_pools = vec![
96                (PoolId::Address(alloy_primitives::Address::repeat_byte(0x01)), pool_class),
97                (PoolId::Address(alloy_primitives::Address::repeat_byte(0x02)), pool_class),
98            ];
99
100            let task = LoomTask::FetchAndAddPools(example_pools);
101
102            if let Err(e) = tx.send(task) {
103                error!("Failed to send protocol pool task for {}: {}", pool_class, e);
104                return Err(e.into());
105            }
106
107            info!("Sent protocol pool discovery task for {}", pool_class);
108        }
109
110        Ok(())
111    }
112}
113
114impl<P, PL, N> Component for ProtocolPoolLoaderComponent<P, PL, N>
115where
116    N: Network + 'static,
117    P: Provider<N> + Send + Sync + Clone + 'static,
118    PL: Provider<N> + Send + Sync + Clone + 'static,
119{
120    fn spawn(self, executor: TaskExecutor) -> Result<()> {
121        executor.spawn_critical(self.name(), async move {
122            if let Err(e) = self.run().await {
123                error!("Protocol pool loader component failed: {}", e);
124            }
125        });
126        Ok(())
127    }
128    fn name(&self) -> &'static str {
129        "ProtocolPoolLoaderComponent"
130    }
131}
132
133/// Builder for ProtocolPoolLoaderComponent
134pub struct ProtocolPoolLoaderComponentBuilder {
135    // Add configuration options as needed
136}
137
138impl ProtocolPoolLoaderComponentBuilder {
139    pub fn new() -> Self {
140        Self {}
141    }
142
143    pub fn build<P, PL, N>(self, client: P, pool_loaders: Arc<PoolLoaders<PL, N>>) -> ProtocolPoolLoaderComponent<P, PL, N>
144    where
145        N: Network + 'static,
146        P: Provider<N> + Send + Sync + Clone + 'static,
147        PL: Provider<N> + Send + Sync + Clone + 'static,
148    {
149        ProtocolPoolLoaderComponent::new(client, pool_loaders)
150    }
151}
152
153impl Default for ProtocolPoolLoaderComponentBuilder {
154    fn default() -> Self {
155        Self::new()
156    }
157}
kabu_defi_market/protocol_pool_loader_component.rs

kabu_defi_market/
protocol_pool_loader_component.rs
