kabu_defi_pools/loaders/
curve.rs

1use crate::protocols::CurveProtocol;
2use crate::{CurvePool, pool_loader};
3use alloy::primitives::{Bytes, Log};
4use async_stream::stream;
5use eyre::eyre;
6use futures::Stream;
7use kabu_evm_db::KabuDBError;
8use kabu_types_market::PoolClass;
9use kabu_types_market::{PoolId, PoolLoader, PoolWrapper};
10use reth_ethereum_primitives::EthPrimitives;
11use revm::DatabaseRef;
12use std::future::Future;
13use std::pin::Pin;
14use std::sync::Arc;
15use tracing::error;
16
17pool_loader!(CurvePoolLoader);
18
19impl<P, N, LDT> PoolLoader<P, N, LDT> for CurvePoolLoader<P, N, LDT>
20where
21    N: Network,
22    P: Provider<N> + Clone + 'static,
23    LDT: Send + Sync + 'static,
24{
25    fn get_pool_class_by_log(&self, _log_entry: &Log) -> Option<(PoolId, PoolClass)> {
26        None
27    }
28
29    fn fetch_pool_by_id<'a>(&'a self, pool_id: PoolId) -> Pin<Box<dyn Future<Output = eyre::Result<PoolWrapper>> + Send + 'a>> {
30        Box::pin(async move {
31            if let Some(provider) = &self.provider {
32                self.fetch_pool_by_id_from_provider(pool_id, provider.clone()).await
33            } else {
34                Err(eyre!("NO_PROVIDER"))
35            }
36        })
37    }
38
39    fn fetch_pool_by_id_from_provider<'a>(
40        &'a self,
41        pool_id: PoolId,
42        provider: P,
43    ) -> Pin<Box<dyn Future<Output = eyre::Result<PoolWrapper>> + Send + 'a>> {
44        Box::pin(async move {
45            let pool_address = match pool_id {
46                PoolId::Address(addr) => addr,
47                _ => return Err(eyre!("Pool ID must be an address for Curve pools")),
48            };
49            match CurveProtocol::get_contract_from_code(provider.clone(), pool_address).await {
50                Ok(curve_contract) => {
51                    let curve_pool = CurvePool::<P, N>::fetch_pool_data_with_default_encoder(provider.clone(), curve_contract).await?;
52
53                    Ok(PoolWrapper::new(Arc::new(curve_pool)))
54                }
55                Err(e) => {
56                    error!("Error getting curve contract from code {} : {} ", pool_address, e);
57                    Err(e)
58                }
59            }
60        })
61    }
62
63    fn fetch_pool_by_id_from_evm(&self, _pool_id: PoolId, _db: &dyn DatabaseRef<Error = KabuDBError>) -> eyre::Result<PoolWrapper> {
64        Err(eyre!("NOT_IMPLEMENTED"))
65    }
66
67    fn is_code(&self, _code: &Bytes) -> bool {
68        false
69    }
70
71    fn protocol_loader(&self) -> eyre::Result<Pin<Box<dyn Stream<Item = (PoolId, PoolClass)> + Send>>> {
72        let provider_clone = self.provider.clone();
73
74        if let Some(client) = provider_clone {
75            Ok(Box::pin(stream! {
76                let curve_contracts = CurveProtocol::get_contracts_vec(client.clone());
77                for curve_contract in curve_contracts.iter() {
78                    yield (PoolId::Address(curve_contract.get_address()), PoolClass::Curve)
79                }
80
81                for factory_idx in 0..10 {
82                    if let Ok(factory_address) = CurveProtocol::get_factory_address(client.clone(), factory_idx).await
83                        && let Ok(pool_count) = CurveProtocol::get_pool_count(client.clone(), factory_address).await {
84                            for pool_id in 0..pool_count {
85                                if let Ok(addr) = CurveProtocol::get_pool_address(client.clone(), factory_address, pool_id).await {
86                                    yield (PoolId::Address(addr), PoolClass::Curve)
87                                }
88                            }
89                    }
90                }
91            }))
92        } else {
93            Err(eyre!("NO_PROVIDER"))
94        }
95    }
96}
kabu_defi_pools/loaders/curve.rs

kabu_defi_pools/loaders/
curve.rs
