tycho_ethereum/services/
token_pre_processor.rs

1use std::sync::Arc;
2
3use alloy::{primitives::Address, rpc::types::BlockNumberOrTag, sol_types::SolCall};
4use async_trait::async_trait;
5use tracing::{instrument, warn};
6use tycho_common::{
7    models::{
8        blockchain::BlockTag,
9        token::{Token, TokenQuality},
10        Chain,
11    },
12    traits::{TokenAnalyzer, TokenOwnerFinding, TokenPreProcessor},
13    Bytes,
14};
15use unicode_segmentation::UnicodeSegmentation;
16
17use crate::{
18    erc20::{decimalsCall, symbolCall},
19    rpc::EthereumRpcClient,
20    services::token_analyzer::{call_request, EthCallDetector},
21    BytesCodec,
22};
23
24#[derive(Debug, Clone)]
25pub struct EthereumTokenPreProcessor {
26    rpc: EthereumRpcClient,
27    chain: Chain,
28    settlement_contract: Address,
29}
30
31impl EthereumTokenPreProcessor {
32    pub fn new(rpc: &EthereumRpcClient, chain: Chain, settlement_contract: Address) -> Self {
33        EthereumTokenPreProcessor { rpc: rpc.clone(), chain, settlement_contract }
34    }
35
36    async fn call_symbol(&self, token: Address) -> String {
37        let calldata = symbolCall {}.abi_encode();
38
39        let result = match self
40            .rpc
41            .eth_call(call_request(None, token, calldata), BlockNumberOrTag::Latest)
42            .await
43        {
44            Ok(result) => result,
45            Err(e) => {
46                warn!(?e, ?token, "Failed to call symbol function, using address as fallback");
47                return format!("0x{:x}", token);
48            }
49        };
50
51        match symbolCall::abi_decode_returns_validate(&result) {
52            Ok(symbol) => symbol,
53            Err(e) => {
54                warn!(
55                    ?e,
56                    ?token,
57                    "Failed to decode symbol function result, using address as fallback"
58                );
59                format!("0x{:x}", token)
60            }
61        }
62    }
63
64    async fn call_decimals(&self, token: Address) -> u8 {
65        let calldata = decimalsCall {}.abi_encode();
66
67        let result = match self
68            .rpc
69            .eth_call(call_request(None, token, calldata), BlockNumberOrTag::Latest)
70            .await
71        {
72            Ok(result) => result,
73            Err(e) => {
74                warn!(?e, ?token, "Failed to call decimals function, using default decimals 18");
75                return 18;
76            }
77        };
78
79        match decimalsCall::abi_decode_returns_validate(&result) {
80            Ok(decimals) => decimals,
81            Err(e) => {
82                warn!(
83                    ?e,
84                    ?token,
85                    "Failed to decode decimals function result, using default decimals 18"
86                );
87                18
88            }
89        }
90    }
91}
92
93#[async_trait]
94impl TokenPreProcessor for EthereumTokenPreProcessor {
95    // Named explicitly: this span is an on-chain metadata fetch and would otherwise
96    // be indistinguishable in traces from the storage-layer `get_tokens` spans.
97    #[instrument(
98        name = "fetch_onchain_token_metadata",
99        skip_all,
100        fields(n_addresses = addresses.len(), block = ?block)
101    )]
102    async fn get_tokens(
103        &self,
104        addresses: Vec<Bytes>,
105        token_finder: Arc<dyn TokenOwnerFinding>,
106        block: BlockTag,
107    ) -> Vec<Token> {
108        let mut tokens_info = Vec::new();
109
110        for address in addresses {
111            let token_address = Address::from_bytes(&address);
112
113            // Make RPC calls directly for symbol and decimals
114            let symbol = self.call_symbol(token_address).await;
115            let decimals = self.call_decimals(token_address).await;
116
117            let detector =
118                EthCallDetector::new(&self.rpc, token_finder.clone(), self.settlement_contract);
119
120            let (token_quality, gas, tax) = detector
121                .analyze(address.clone(), block)
122                .await
123                .unwrap_or_else(|e| {
124                    warn!(error=?e, "TokenDetectionFailure");
125                    (TokenQuality::bad("Detection failed"), None, None)
126                });
127
128            let mut quality = 100;
129
130            if let TokenQuality::Bad { reason } = token_quality {
131                warn!(address=?address, ?reason, "BadToken");
132                // Flag this token as bad using quality, an external script is responsible for
133                // analyzing these tokens again.
134                quality = 10;
135            };
136
137            // If quality is 100 but it's a fee token, set quality to 50
138            if quality == 100 && tax.is_some_and(|tax_value| tax_value > 0) {
139                quality = 50;
140            }
141
142            tokens_info.push(Token {
143                address,
144                symbol: symbol
145                    .replace('\0', "")
146                    .graphemes(true)
147                    .take(255)
148                    .collect::<String>(),
149                decimals: decimals.into(),
150                tax: tax.unwrap_or(0),
151                gas: gas
152                    .map(|g| vec![Some(g)])
153                    .unwrap_or_else(Vec::new),
154                chain: self.chain,
155                quality,
156            });
157        }
158
159        tokens_info
160    }
161}
162
163#[cfg(test)]
164mod tests {
165    use std::str::FromStr;
166
167    use alloy::primitives::address;
168    use tycho_common::models::token::TokenOwnerStore;
169
170    use super::*;
171    use crate::test_fixtures::{TestFixture, TEST_BLOCK_NUMBER, TOKEN_HOLDERS, USDC_STR, WETH_STR};
172
173    const COWSWAP_SETTLEMENT: Address = address!("c9f2e6ea1637E499406986ac50ddC92401ce1f58");
174
175    impl TestFixture {
176        fn create_token_preprocessor(&self) -> EthereumTokenPreProcessor {
177            // We do not enable batching as the token pre-processor does not leverage it currently
178            let rpc = self.create_rpc_client(false);
179
180            EthereumTokenPreProcessor::new(&rpc, Chain::Ethereum, COWSWAP_SETTLEMENT)
181        }
182    }
183
184    #[tokio::test]
185    #[ignore = "require RPC connection"]
186    async fn test_call_symbol() {
187        let fixture = TestFixture::new();
188        let processor = fixture.create_token_preprocessor();
189
190        // Test WETH symbol
191        let weth_address = Address::from_str(WETH_STR).expect("Failed to parse WETH address");
192        let symbol = processor
193            .call_symbol(weth_address)
194            .await;
195        assert_eq!(symbol, "WETH", "Expected WETH symbol");
196
197        // Test USDC symbol
198        let usdc_address = Address::from_str(USDC_STR).expect("Failed to parse USDC address");
199        let symbol = processor
200            .call_symbol(usdc_address)
201            .await;
202        assert_eq!(symbol, "USDC", "Expected USDC symbol");
203    }
204
205    #[tokio::test]
206    #[ignore = "require RPC connection"]
207    async fn test_call_decimals() {
208        let fixture = TestFixture::new();
209        let processor = fixture.create_token_preprocessor();
210
211        // Test WETH decimals (18)
212        let weth_address = Address::from_str(WETH_STR).expect("Failed to parse WETH address");
213        let decimals = processor
214            .call_decimals(weth_address)
215            .await;
216        assert_eq!(decimals, 18, "Expected WETH to have 18 decimals");
217
218        // Test USDC decimals (6)
219        let usdc_address = Address::from_str(USDC_STR).expect("Failed to parse USDC address");
220        let decimals = processor
221            .call_decimals(usdc_address)
222            .await;
223        assert_eq!(decimals, 6, "Expected USDC to have 6 decimals");
224    }
225
226    #[tokio::test]
227    #[ignore = "require archive RPC connection"]
228    async fn test_get_tokens() {
229        let fixture = TestFixture::new();
230        let processor = fixture.create_token_preprocessor();
231
232        let tf = TokenOwnerStore::new(TOKEN_HOLDERS.clone());
233
234        let fake_address: &str = "0xA0b86991c7456b36c1d19D4a2e9Eb0cE3606eB48";
235        let addresses = vec![
236            Bytes::from_str(WETH_STR).unwrap(),
237            Bytes::from_str(USDC_STR).unwrap(),
238            Bytes::from_str(fake_address).unwrap(),
239        ];
240
241        let results = processor
242            .get_tokens(addresses, Arc::new(tf), BlockTag::Number(TEST_BLOCK_NUMBER))
243            .await;
244        assert_eq!(results.len(), 3);
245        let relevant_attrs: Vec<(String, u32, u32)> = results
246            .iter()
247            .map(|t| (t.symbol.clone(), t.decimals, t.quality))
248            .collect();
249        assert_eq!(
250            relevant_attrs,
251            vec![
252                ("WETH".to_string(), 18, 100),
253                ("USDC".to_string(), 6, 100),
254                (fake_address.to_lowercase(), 18, 10)
255            ]
256        );
257    }
258}
tycho_ethereum/services/token_pre_processor.rs

tycho_ethereum/services/
token_pre_processor.rs
