kabu_defi_market/
new_pool_actor.rs

1use alloy_network::Network;
2use alloy_provider::Provider;
3use eyre::Result;
4use futures_util::StreamExt;
5use std::sync::Arc;
6use tokio::sync::broadcast;
7use tracing::{debug, error, info};
8
9use kabu_core_components::Component;
10use kabu_types_events::LoomTask;
11use kabu_types_market::PoolLoaders;
12use reth_chain_state::{CanonStateNotification, CanonStateSubscriptions};
13use reth_ethereum_primitives::EthPrimitives;
14use reth_tasks::TaskExecutor;
15
16use crate::logs_parser::process_log_entries;
17
18pub async fn new_pool_worker<P, N>(client: P, pools_loaders: Arc<PoolLoaders<P, N>>, tasks_tx: broadcast::Sender<LoomTask>)
19where
20    N: Network,
21    P: Provider<N> + CanonStateSubscriptions<Primitives = EthPrimitives> + Send + Sync + Clone + 'static,
22{
23    // Subscribe to canonical state notifications
24    let mut stream = client.canonical_state_stream();
25
26    info!("New pool loader subscribed to canonical state notifications");
27
28    while let Some(notification) = stream.next().await {
29        match notification {
30            CanonStateNotification::Reorg { old: _, new } => {
31                debug!("Processing reorg with new chain for pool discovery");
32                process_canonical_chain(new, &pools_loaders, &tasks_tx).await;
33            }
34            CanonStateNotification::Commit { new } => {
35                debug!("Processing committed chain for pool discovery");
36                process_canonical_chain(new, &pools_loaders, &tasks_tx).await;
37            }
38        }
39    }
40}
41
42async fn process_canonical_chain<P, N>(
43    chain: Arc<reth_execution_types::Chain<EthPrimitives>>,
44    pools_loaders: &Arc<PoolLoaders<P, N>>,
45    tasks_tx: &broadcast::Sender<LoomTask>,
46) where
47    N: Network,
48    P: Provider<N> + Send + Sync + Clone + 'static,
49{
50    // Get receipts from the chain's execution outcome
51    let receipts = chain.execution_outcome().receipts();
52
53    // Collect all logs from all receipts
54    let mut all_logs = Vec::new();
55    for receipt_vec in receipts {
56        for receipt in receipt_vec {
57            all_logs.extend(receipt.logs.clone());
58        }
59    }
60
61    if !all_logs.is_empty() {
62        debug!("Processing {} logs for new pool discovery", all_logs.len());
63        if let Err(e) = process_log_entries(all_logs, pools_loaders, tasks_tx.clone()).await {
64            error!("Error processing logs for new pools: {}", e);
65        }
66    }
67}
68
69pub struct NewPoolLoaderComponent<P, N>
70where
71    N: Network,
72    P: Provider<N> + CanonStateSubscriptions<Primitives = EthPrimitives> + Send + Sync + Clone + 'static,
73{
74    client: P,
75    pool_loaders: Arc<PoolLoaders<P, N>>,
76    tasks_tx: Option<broadcast::Sender<LoomTask>>,
77}
78
79impl<P, N> NewPoolLoaderComponent<P, N>
80where
81    N: Network,
82    P: Provider<N> + CanonStateSubscriptions<Primitives = EthPrimitives> + Send + Sync + Clone + 'static,
83{
84    pub fn new(client: P, pool_loaders: Arc<PoolLoaders<P, N>>) -> Self {
85        NewPoolLoaderComponent { client, pool_loaders, tasks_tx: None }
86    }
87
88    pub fn with_task_channel(self, tasks_tx: broadcast::Sender<LoomTask>) -> Self {
89        Self { tasks_tx: Some(tasks_tx), ..self }
90    }
91}
92
93impl<P, N> Component for NewPoolLoaderComponent<P, N>
94where
95    N: Network,
96    P: Provider<N> + CanonStateSubscriptions<Primitives = EthPrimitives> + Send + Sync + Clone + 'static,
97{
98    fn spawn(self, executor: TaskExecutor) -> Result<()> {
99        let name = self.name();
100
101        let tasks_tx = self.tasks_tx.ok_or_else(|| eyre::eyre!("tasks_tx not set"))?;
102
103        executor.spawn_critical(name, new_pool_worker(self.client, self.pool_loaders, tasks_tx));
104
105        Ok(())
106    }
107    fn name(&self) -> &'static str {
108        "NewPoolLoaderComponent"
109    }
110}
kabu_defi_market/new_pool_actor.rs

kabu_defi_market/
new_pool_actor.rs
