kabu_core_node/
signer.rs

1//! Bridge component that converts Ready SwapCompose messages to TxCompose for broadcasting
2
3use eyre::{Result, eyre};
4use kabu_core_components::Component;
5use kabu_types_events::{MessageSwapCompose, MessageTxCompose, SwapComposeMessage, TxComposeData};
6use reth_tasks::TaskExecutor;
7use tokio::sync::broadcast;
8use tracing::{debug, error, info, warn};
9
10/// Component that bridges SwapCompose Ready messages to TxCompose for the broadcaster
11pub struct SignerBridgeComponent<DB: Send + Sync + Clone + 'static> {
12    swap_compose_rx: Option<broadcast::Receiver<MessageSwapCompose<DB>>>,
13    tx_compose_tx: Option<broadcast::Sender<MessageTxCompose>>,
14}
15
16impl<DB: Send + Sync + Clone + 'static> Default for SignerBridgeComponent<DB> {
17    fn default() -> Self {
18        Self { swap_compose_rx: None, tx_compose_tx: None }
19    }
20}
21
22impl<DB: Send + Sync + Clone + 'static> SignerBridgeComponent<DB> {
23    pub fn new() -> Self {
24        Self::default()
25    }
26
27    pub fn with_channels(
28        mut self,
29        swap_compose: broadcast::Sender<MessageSwapCompose<DB>>,
30        tx_compose: broadcast::Sender<MessageTxCompose>,
31    ) -> Self {
32        self.swap_compose_rx = Some(swap_compose.subscribe());
33        self.tx_compose_tx = Some(tx_compose);
34        self
35    }
36
37    async fn run(self) -> Result<()> {
38        info!("Starting signer bridge component");
39
40        let mut swap_compose_rx = self.swap_compose_rx.ok_or_else(|| eyre!("swap_compose_rx not set"))?;
41        let tx_compose_tx = self.tx_compose_tx.ok_or_else(|| eyre!("tx_compose_tx not set"))?;
42
43        loop {
44            match swap_compose_rx.recv().await {
45                Ok(msg) => {
46                    debug!("Signer bridge received message type: {:?}", std::mem::discriminant(msg.inner()));
47                    if let SwapComposeMessage::Ready(ready_data) = msg.inner() {
48                        info!("Signer bridge converting Ready message to TxCompose for swap: {}", ready_data.swap);
49
50                        // Convert SwapComposeData to TxComposeData
51                        let tx_data = TxComposeData {
52                            eoa: ready_data.tx_compose.eoa,
53                            signer: ready_data.tx_compose.signer.clone(),
54                            nonce: ready_data.tx_compose.nonce,
55                            eth_balance: ready_data.tx_compose.eth_balance,
56                            value: ready_data.tx_compose.value,
57                            gas: ready_data.tx_compose.gas,
58                            priority_gas_fee: ready_data.tx_compose.priority_gas_fee,
59                            stuffing_txs_hashes: ready_data.tx_compose.stuffing_txs_hashes.clone(),
60                            stuffing_txs: ready_data.tx_compose.stuffing_txs.clone(),
61                            next_block_number: ready_data.tx_compose.next_block_number,
62                            next_block_timestamp: ready_data.tx_compose.next_block_timestamp,
63                            next_block_base_fee: ready_data.tx_compose.next_block_base_fee,
64                            tx_bundle: ready_data.tx_compose.tx_bundle.clone(),
65                            rlp_bundle: ready_data.tx_compose.rlp_bundle.clone(),
66                            origin: Some("SignerBridge".to_string()),
67                            swap: Some(ready_data.swap.clone()),
68                            tips: ready_data.tips,
69                        };
70
71                        // Send as broadcast message
72                        let tx_msg = MessageTxCompose::broadcast(tx_data);
73                        if let Err(e) = tx_compose_tx.send(tx_msg) {
74                            warn!("Failed to send to tx_compose: {}", e);
75                        }
76                    }
77                }
78                Err(e) => {
79                    error!("Swap compose channel error: {}", e);
80                    return Err(eyre!("Swap compose channel closed"));
81                }
82            }
83        }
84    }
85}
86
87impl<DB: Send + Sync + Clone + 'static> Component for SignerBridgeComponent<DB> {
88    fn spawn(self, executor: TaskExecutor) -> Result<()> {
89        executor.spawn_critical(self.name(), async move {
90            if let Err(e) = self.run().await {
91                error!("Signer bridge component failed: {}", e);
92            }
93        });
94        Ok(())
95    }
96
97    fn name(&self) -> &'static str {
98        "SignerBridgeComponent"
99    }
100}
kabu_core_node/signer.rs

kabu_core_node/
signer.rs
