kabu_core_node/
defaults.rs

1//! Default component builders that create standard Kabu components
2
3use crate::context::KabuContext;
4use crate::traits::{ComponentBuilder, DisabledComponent};
5use alloy_network::Ethereum;
6use alloy_provider::Provider;
7use eyre::Result;
8use std::sync::Arc;
9
10use kabu_broadcast_broadcaster::FlashbotsBroadcastComponent;
11use kabu_core_block_history::BlockHistoryComponent;
12use kabu_core_components::Component;
13use kabu_core_router::SwapRouterComponent;
14use kabu_defi_market::{HistoryPoolLoaderComponent, ProtocolPoolLoaderComponent};
15use kabu_defi_pools::PoolLoadersBuilder;
16use kabu_defi_preloader::MarketStatePreloadedOneShotComponent;
17use kabu_defi_price::PriceComponent;
18use kabu_evm_db::{DatabaseKabuExt, KabuDBError};
19use kabu_execution_estimator::EvmEstimatorComponent;
20use kabu_node_debug_provider::DebugProviderExt;
21use kabu_strategy_backrun::StateChangeArbComponent;
22use kabu_strategy_merger::DiffPathMergerComponent;
23use kabu_types_entities::BlockHistoryState;
24use reth::revm::{Database, DatabaseCommit, DatabaseRef};
25use reth_chain_state::CanonStateSubscriptions;
26use reth_ethereum_primitives::EthPrimitives;
27
28#[cfg(feature = "defi-health-monitor")]
29use kabu_defi_health_monitor::PoolHealthMonitorComponent;
30
31// ================================================================================================
32// Network Component
33// ================================================================================================
34
35/// Default network builder that creates a BlockHistoryComponent
36pub struct DefaultNetwork;
37
38impl<R, P, DB, Evm> ComponentBuilder<KabuContext<R, P, DB, Evm>> for DefaultNetwork
39where
40    R: CanonStateSubscriptions<Primitives = EthPrimitives> + Send + Sync + Clone + 'static,
41    P: Provider<Ethereum> + DebugProviderExt<Ethereum> + Send + Sync + Clone + 'static,
42    DB: DatabaseRef<Error = KabuDBError>
43        + Database<Error = KabuDBError>
44        + DatabaseCommit
45        + DatabaseKabuExt
46        + BlockHistoryState<EthPrimitives>
47        + Send
48        + Sync
49        + Clone
50        + Default
51        + 'static,
52    Evm: Clone + Send + Sync + 'static,
53{
54    type Component = BlockHistoryComponent<R, P, DB>;
55
56    fn build(self, ctx: &KabuContext<R, P, DB, Evm>) -> Result<Self::Component> {
57        Ok(BlockHistoryComponent::new(ctx.reth_provider.clone(), ctx.provider.clone()).with_config(
58            ctx.config.chain_params.clone(),
59            ctx.market_state.clone(),
60            ctx.block_history.clone(),
61            ctx.channels.market_events.clone(),
62        ))
63    }
64}
65
66// ================================================================================================
67// Executor Component
68// ================================================================================================
69
70/// Default executor builder that creates a SwapRouterComponent
71pub struct DefaultExecutor;
72
73impl<R, P, DB, Evm> ComponentBuilder<KabuContext<R, P, DB, Evm>> for DefaultExecutor
74where
75    R: Send + Sync + Clone + 'static,
76    P: Provider<Ethereum> + Send + Sync + Clone + 'static,
77    DB: DatabaseRef<Error = KabuDBError>
78        + Database<Error = KabuDBError>
79        + DatabaseCommit
80        + DatabaseKabuExt
81        + BlockHistoryState<EthPrimitives>
82        + Send
83        + Sync
84        + Clone
85        + Default
86        + 'static,
87    Evm: Clone + Send + Sync + 'static,
88{
89    type Component = SwapRouterComponent<DB, EthPrimitives>;
90
91    fn build(self, ctx: &KabuContext<R, P, DB, Evm>) -> Result<Self::Component> {
92        Ok(SwapRouterComponent::new(ctx.channels.signers.clone(), ctx.channels.account_state.clone(), ctx.channels.swap_compose.clone()))
93    }
94}
95
96// ================================================================================================
97// Strategy Component
98// ================================================================================================
99
100/// Default strategy builder that creates a StateChangeArbComponent
101pub struct DefaultStrategy;
102
103impl<R, P, DB, Evm> ComponentBuilder<KabuContext<R, P, DB, Evm>> for DefaultStrategy
104where
105    R: Send + Sync + Clone + 'static,
106    P: Provider<Ethereum> + DebugProviderExt<Ethereum> + Send + Sync + Clone + 'static,
107    DB: DatabaseRef<Error = KabuDBError>
108        + Database<Error = KabuDBError>
109        + DatabaseCommit
110        + DatabaseKabuExt
111        + BlockHistoryState<EthPrimitives>
112        + Send
113        + Sync
114        + Clone
115        + Default
116        + 'static,
117    Evm: Clone + Send + Sync + 'static,
118{
119    type Component = StateChangeArbComponent<P, Ethereum, DB, EthPrimitives>;
120
121    fn build(self, ctx: &KabuContext<R, P, DB, Evm>) -> Result<Self::Component> {
122        let mut component = StateChangeArbComponent::new(ctx.provider.clone(), ctx.config.backrun_config.clone())
123            .with_market(ctx.market.clone())
124            .with_mempool(ctx.mempool.clone())
125            .with_market_state(ctx.market_state.clone())
126            .with_block_history(ctx.block_history.clone())
127            .with_mempool_events_channel(ctx.channels.mempool_events.clone())
128            .with_market_events_channel(ctx.channels.market_events.clone())
129            .with_swap_compose_channel(ctx.channels.swap_compose.clone())
130            .with_pool_health_monitor_channel(ctx.channels.health_events.clone());
131
132        if let Some(channel) = ctx.blockchain.influxdb_write_channel() {
133            component = component.with_influxdb_channel(channel);
134        }
135
136        Ok(component)
137    }
138}
139
140// ================================================================================================
141// Signer Bridge Component
142// ================================================================================================
143
144/// Default signer bridge builder
145pub struct DefaultSigner;
146
147impl<R, P, DB, Evm> ComponentBuilder<KabuContext<R, P, DB, Evm>> for DefaultSigner
148where
149    R: Send + Sync + Clone + 'static,
150    P: Provider<Ethereum> + Send + Sync + Clone + 'static,
151    DB: DatabaseRef<Error = KabuDBError>
152        + Database<Error = KabuDBError>
153        + DatabaseCommit
154        + DatabaseKabuExt
155        + BlockHistoryState<EthPrimitives>
156        + Send
157        + Sync
158        + Clone
159        + Default
160        + 'static,
161    Evm: Clone + Send + Sync + 'static,
162{
163    type Component = crate::signer::SignerBridgeComponent<DB>;
164
165    fn build(self, ctx: &KabuContext<R, P, DB, Evm>) -> Result<Self::Component> {
166        Ok(crate::signer::SignerBridgeComponent::new().with_channels(ctx.channels.swap_compose.clone(), ctx.channels.tx_compose.clone()))
167    }
168}
169
170// ================================================================================================
171// Market Components (Composite)
172// ================================================================================================
173
174/// Composite component that combines all market-related components
175#[derive(Clone)]
176pub struct MarketComponents<P, DB>
177where
178    P: Provider<Ethereum> + Send + Sync + Clone + 'static,
179    DB: Send + Sync + Clone + 'static,
180{
181    preloader: MarketStatePreloadedOneShotComponent<P, Ethereum, DB>,
182    price: PriceComponent<P, Ethereum>,
183    protocol_loader: ProtocolPoolLoaderComponent<P, P, Ethereum>,
184    history_loader: HistoryPoolLoaderComponent<P, P, Ethereum>,
185}
186
187impl<P, DB> Component for MarketComponents<P, DB>
188where
189    P: Provider<Ethereum> + DebugProviderExt<Ethereum> + Send + Sync + Clone + 'static,
190    DB: DatabaseRef<Error = KabuDBError>
191        + Database<Error = KabuDBError>
192        + DatabaseCommit
193        + DatabaseKabuExt
194        + BlockHistoryState<EthPrimitives>
195        + Send
196        + Sync
197        + Clone
198        + Default
199        + 'static,
200{
201    fn spawn(self, executor: reth_tasks::TaskExecutor) -> Result<()> {
202        self.preloader.spawn(executor.clone())?;
203        self.price.spawn(executor.clone())?;
204        self.protocol_loader.spawn(executor.clone())?;
205        self.history_loader.spawn(executor)?;
206        Ok(())
207    }
208
209    fn name(&self) -> &'static str {
210        "MarketComponents"
211    }
212}
213
214/// Default market builder
215pub struct DefaultMarket;
216
217impl<R, P, DB, Evm> ComponentBuilder<KabuContext<R, P, DB, Evm>> for DefaultMarket
218where
219    R: Send + Sync + Clone + 'static,
220    P: Provider<Ethereum> + DebugProviderExt<Ethereum> + Send + Sync + Clone + 'static,
221    DB: DatabaseRef<Error = KabuDBError>
222        + Database<Error = KabuDBError>
223        + DatabaseCommit
224        + DatabaseKabuExt
225        + BlockHistoryState<EthPrimitives>
226        + Send
227        + Sync
228        + Clone
229        + Default
230        + 'static,
231    Evm: Clone + Send + Sync + 'static,
232{
233    type Component = MarketComponents<P, DB>;
234
235    fn build(self, ctx: &KabuContext<R, P, DB, Evm>) -> Result<Self::Component> {
236        let pool_loaders = Arc::new(PoolLoadersBuilder::<_, _, EthPrimitives>::default_pool_loaders(
237            ctx.provider.clone(),
238            ctx.config.pools_config.clone(),
239        ));
240
241        let preloader = MarketStatePreloadedOneShotComponent::new(ctx.provider.clone())
242            .with_copied_account(ctx.config.swap_encoder.get_contract_address())
243            .with_signers(ctx.channels.signers.clone())
244            .with_market_state(ctx.market_state.clone());
245
246        let price = PriceComponent::new(ctx.provider.clone()).only_once().with_market(ctx.market.clone());
247
248        let protocol_loader = ProtocolPoolLoaderComponent::new(ctx.provider.clone(), pool_loaders.clone());
249
250        let history_loader = HistoryPoolLoaderComponent::new(
251            ctx.provider.clone(),
252            pool_loaders,
253            0,    // start_block
254            1000, // block_batch_size
255            10,   // max_batches
256        );
257
258        Ok(MarketComponents { preloader, price, protocol_loader, history_loader })
259    }
260}
261
262// ================================================================================================
263// Broadcaster Component
264// ================================================================================================
265
266/// Default broadcaster builder
267pub struct DefaultBroadcaster;
268
269impl<R, P, DB, Evm> ComponentBuilder<KabuContext<R, P, DB, Evm>> for DefaultBroadcaster
270where
271    R: Send + Sync + Clone + 'static,
272    P: Provider<Ethereum> + Send + Sync + Clone + 'static,
273    DB: DatabaseRef<Error = KabuDBError>
274        + Database<Error = KabuDBError>
275        + DatabaseCommit
276        + DatabaseKabuExt
277        + BlockHistoryState<EthPrimitives>
278        + Send
279        + Sync
280        + Clone
281        + Default
282        + 'static,
283    Evm: Clone + Send + Sync + 'static,
284{
285    type Component = FlashbotsBroadcastComponent;
286
287    fn build(self, ctx: &KabuContext<R, P, DB, Evm>) -> Result<Self::Component> {
288        // Convert builders from config to relay format
289        let relays = ctx
290            .config
291            .config
292            .builders
293            .iter()
294            .map(|b| kabu_broadcast_broadcaster::RelayConfig {
295                id: b.id as u64,
296                name: b.name.clone(),
297                url: b.url.clone(),
298                no_sign: Some(b.no_sign),
299            })
300            .collect::<Vec<_>>();
301
302        let mut component = FlashbotsBroadcastComponent::new(None, !relays.is_empty())?;
303        component = component.with_relays(relays)?;
304        component = component.with_channel(ctx.channels.tx_compose.clone());
305        Ok(component)
306    }
307}
308
309// ================================================================================================
310// Estimator Component
311// ================================================================================================
312
313/// Default estimator builder
314pub struct DefaultEstimator;
315
316impl<R, P, DB, Evm> ComponentBuilder<KabuContext<R, P, DB, Evm>> for DefaultEstimator
317where
318    R: Send + Sync + Clone + 'static,
319    P: Provider<Ethereum> + DebugProviderExt<Ethereum> + Send + Sync + Clone + 'static,
320    DB: DatabaseRef<Error = KabuDBError>
321        + Database<Error = KabuDBError>
322        + DatabaseCommit
323        + DatabaseKabuExt
324        + BlockHistoryState<EthPrimitives>
325        + Send
326        + Sync
327        + Clone
328        + Default
329        + 'static,
330    Evm: Clone + Send + Sync + 'static,
331{
332    type Component = EvmEstimatorComponent<P, Ethereum, kabu_execution_multicaller::MulticallerSwapEncoder, DB>;
333
334    fn build(self, ctx: &KabuContext<R, P, DB, Evm>) -> Result<Self::Component> {
335        Ok(EvmEstimatorComponent::new_with_provider(ctx.config.swap_encoder.clone(), Some(ctx.provider.clone()))
336            .with_swap_compose_channel(ctx.channels.swap_compose.clone()))
337    }
338}
339
340// ================================================================================================
341// Health Monitor Component
342// ================================================================================================
343
344/// Default health monitor builder
345pub struct DefaultHealthMonitor;
346
347impl<R, P, DB, Evm> ComponentBuilder<KabuContext<R, P, DB, Evm>> for DefaultHealthMonitor
348where
349    R: Send + Sync + Clone + 'static,
350    P: Provider<Ethereum> + Send + Sync + Clone + 'static,
351    DB: DatabaseRef<Error = KabuDBError>
352        + Database<Error = KabuDBError>
353        + DatabaseCommit
354        + DatabaseKabuExt
355        + BlockHistoryState<EthPrimitives>
356        + Send
357        + Sync
358        + Clone
359        + Default
360        + 'static,
361    Evm: Clone + Send + Sync + 'static,
362{
363    #[cfg(feature = "defi-health-monitor")]
364    type Component = PoolHealthMonitorComponent;
365
366    #[cfg(not(feature = "defi-health-monitor"))]
367    type Component = DisabledComponent;
368
369    fn build(self, ctx: &KabuContext<R, P, DB, Evm>) -> Result<Self::Component> {
370        #[cfg(feature = "defi-health-monitor")]
371        {
372            Ok(PoolHealthMonitorComponent::new().with_channels(
373                ctx.market.clone(),
374                ctx.channels.health_events.clone(),
375                ctx.blockchain.influxdb_write_channel(),
376            ))
377        }
378
379        #[cfg(not(feature = "defi-health-monitor"))]
380        {
381            let _ = ctx;
382            Ok(DisabledComponent)
383        }
384    }
385}
386
387// ================================================================================================
388// Merger Components (Composite)
389// ================================================================================================
390
391/// Composite merger components
392#[derive(Clone)]
393pub struct MergerComponents<DB>
394where
395    DB: DatabaseRef<Error = KabuDBError> + Database<Error = KabuDBError> + DatabaseCommit + Send + Sync + Clone + Default + 'static,
396{
397    diff_path: DiffPathMergerComponent<DB>,
398}
399
400impl<DB> Component for MergerComponents<DB>
401where
402    DB: DatabaseRef<Error = KabuDBError> + Database<Error = KabuDBError> + DatabaseCommit + Send + Sync + Clone + Default + 'static,
403{
404    fn spawn(self, executor: reth_tasks::TaskExecutor) -> Result<()> {
405        self.diff_path.spawn(executor)?;
406        Ok(())
407    }
408
409    fn name(&self) -> &'static str {
410        "MergerComponents"
411    }
412}
413
414/// Default merger builder
415pub struct DefaultMerger;
416
417impl<R, P, DB, Evm> ComponentBuilder<KabuContext<R, P, DB, Evm>> for DefaultMerger
418where
419    R: Send + Sync + Clone + 'static,
420    P: Provider<Ethereum> + Send + Sync + Clone + 'static,
421    DB: DatabaseRef<Error = KabuDBError>
422        + Database<Error = KabuDBError>
423        + DatabaseCommit
424        + DatabaseKabuExt
425        + BlockHistoryState<EthPrimitives>
426        + Send
427        + Sync
428        + Clone
429        + Default
430        + 'static,
431    Evm: Clone + Send + Sync + 'static,
432{
433    type Component = MergerComponents<DB>;
434
435    fn build(self, ctx: &KabuContext<R, P, DB, Evm>) -> Result<Self::Component> {
436        let diff_path = DiffPathMergerComponent::<DB>::new()
437            .with_market_events_channel(ctx.channels.market_events.clone())
438            .with_compose_channel(ctx.channels.swap_compose.clone());
439
440        Ok(MergerComponents { diff_path })
441    }
442}
443
444// ================================================================================================
445// Monitoring Component
446// ================================================================================================
447
448/// Default monitoring builder
449pub struct DefaultMonitoring;
450
451impl<R, P, DB, Evm> ComponentBuilder<KabuContext<R, P, DB, Evm>> for DefaultMonitoring
452where
453    R: Send + Sync + Clone + 'static,
454    P: Provider<Ethereum> + Send + Sync + Clone + 'static,
455    DB: DatabaseRef<Error = KabuDBError>
456        + Database<Error = KabuDBError>
457        + DatabaseCommit
458        + DatabaseKabuExt
459        + BlockHistoryState<EthPrimitives>
460        + Send
461        + Sync
462        + Clone
463        + Default
464        + 'static,
465    Evm: Clone + Send + Sync + 'static,
466{
467    type Component = DisabledComponent; // Or InfluxDbWriterComponent if configured
468
469    fn build(self, ctx: &KabuContext<R, P, DB, Evm>) -> Result<Self::Component> {
470        // For now, return disabled. Could check config for InfluxDB settings
471        let _ = ctx;
472        Ok(DisabledComponent)
473    }
474}
kabu_core_node/defaults.rs

kabu_core_node/
defaults.rs
