11use crate :: format:: format_to_human;
22use crate :: rpc:: execute_rpc;
33use crate :: { AssetConfig , DepositResult } ;
4+ use futures:: StreamExt ;
45use hashbrown:: HashSet ;
56use rust_decimal:: prelude:: * ;
67use serde_json:: json;
@@ -9,15 +10,32 @@ use std::sync::Arc;
910use tokio:: sync:: Mutex ;
1011
1112const BTC_DECIMALS : u32 = 8 ;
12- const RPC_DELAY_MS : u64 = 100 ;
13+ const DEFAULT_RPC_DELAY_MS : u64 = 100 ;
14+ const DEFAULT_MAX_CONCURRENT : usize = 3 ;
1315
1416pub struct BtcScanner {
1517 pub rpc_urls : Vec < String > ,
1618 pub caip2 : String ,
1719 pub assets : Arc < HashMap < String , AssetConfig > > ,
20+ pub rpc_delay_ms : Option < u64 > ,
21+ pub max_concurrent : usize ,
1822}
1923
2024impl BtcScanner {
25+ pub async fn get_tip (
26+ client : & reqwest:: Client ,
27+ rpc_urls : & [ String ] ,
28+ ) -> Result < u64 , Box < dyn std:: error:: Error + Send + Sync > > {
29+ let payload = json ! ( {
30+ "jsonrpc" : "1.0" ,
31+ "id" : "rustplorer" ,
32+ "method" : "getblockcount" ,
33+ "params" : [ ]
34+ } ) ;
35+ let res = execute_rpc ( client, rpc_urls, & payload) . await ?;
36+ Ok ( res[ "result" ] . as_u64 ( ) . unwrap_or ( 0 ) )
37+ }
38+
2139 pub async fn scan (
2240 & self ,
2341 client : reqwest:: Client ,
@@ -26,83 +44,117 @@ impl BtcScanner {
2644 targets : Arc < HashSet < String > > ,
2745 results : Arc < Mutex < Vec < DepositResult > > > ,
2846 ) -> Result < ( ) , Box < dyn std:: error:: Error + Send + Sync > > {
29- for block_num in start..=end {
30- let hash_payload = json ! ( {
31- "jsonrpc" : "1.0" ,
32- "id" : "rustplorer" ,
33- "method" : "getblockhash" ,
34- "params" : [ block_num]
35- } ) ;
36-
37- let hash_res = execute_rpc ( & client, & self . rpc_urls , & hash_payload) . await ?;
38- let block_hash = match hash_res[ "result" ] . as_str ( ) {
39- Some ( h) => h,
40- None => continue ,
41- } ;
42-
43- let block_payload = json ! ( {
44- "jsonrpc" : "1.0" ,
45- "id" : "rustplorer" ,
46- "method" : "getblock" ,
47- "params" : [ block_hash, 3 ]
48- } ) ;
49-
50- let block_res = execute_rpc ( & client, & self . rpc_urls , & block_payload) . await ?;
51-
52- if let Some ( transactions) = block_res[ "result" ] [ "tx" ] . as_array ( ) {
53- for tx in transactions {
54- self . process_transaction ( tx, block_num, & targets, & results)
55- . await ;
47+ let client = Arc :: new ( client) ;
48+ let delay_ms = self . rpc_delay_ms . unwrap_or ( DEFAULT_RPC_DELAY_MS ) ;
49+ let max_concurrent = if self . max_concurrent > 0 {
50+ self . max_concurrent
51+ } else {
52+ DEFAULT_MAX_CONCURRENT
53+ } ;
54+
55+ let block_range: Vec < u64 > = ( start..=end) . collect ( ) ;
56+ let fetches = futures:: stream:: iter ( block_range. into_iter ( ) . map ( |block_num| {
57+ let client = Arc :: clone ( & client) ;
58+ let rpc_urls = self . rpc_urls . clone ( ) ;
59+ let caip2 = self . caip2 . clone ( ) ;
60+ let targets = Arc :: clone ( & targets) ;
61+ let results = Arc :: clone ( & results) ;
62+ async move {
63+ let hash_payload = json ! ( {
64+ "jsonrpc" : "1.0" ,
65+ "id" : "rustplorer" ,
66+ "method" : "getblockhash" ,
67+ "params" : [ block_num]
68+ } ) ;
69+
70+ let hash_res = match execute_rpc ( & client, & rpc_urls, & hash_payload) . await {
71+ Ok ( r) => r,
72+ Err ( e) => {
73+ eprintln ! ( "[rustplorer] [{}] block {}: {}" , caip2, block_num, e) ;
74+ return ;
75+ }
76+ } ;
77+ let block_hash = match hash_res[ "result" ] . as_str ( ) {
78+ Some ( h) => h,
79+ None => return ,
80+ } ;
81+
82+ let block_payload = json ! ( {
83+ "jsonrpc" : "1.0" ,
84+ "id" : "rustplorer" ,
85+ "method" : "getblock" ,
86+ "params" : [ block_hash, 3 ]
87+ } ) ;
88+
89+ let block_res = match execute_rpc ( & client, & rpc_urls, & block_payload) . await {
90+ Ok ( r) => r,
91+ Err ( e) => {
92+ eprintln ! ( "[rustplorer] [{}] block {}: {}" , caip2, block_num, e) ;
93+ return ;
94+ }
95+ } ;
96+
97+ if let Some ( transactions) = block_res[ "result" ] [ "tx" ] . as_array ( ) {
98+ for tx in transactions {
99+ process_transaction_static ( tx, block_num, & caip2, & targets, & results) . await ;
100+ }
56101 }
57102 }
103+ } ) )
104+ . buffer_unordered ( max_concurrent) ;
105+
106+ fetches. collect :: < Vec < _ > > ( ) . await ;
107+
108+ tokio:: time:: sleep ( tokio:: time:: Duration :: from_millis ( delay_ms) ) . await ;
58109
59- tokio:: time:: sleep ( tokio:: time:: Duration :: from_millis ( RPC_DELAY_MS ) ) . await ;
60- }
61110 Ok ( ( ) )
62111 }
112+ }
63113
64- async fn process_transaction (
65- & self ,
66- tx : & serde_json :: Value ,
67- block_num : u64 ,
68- targets : & Arc < HashSet < String > > ,
69- results : & Arc < Mutex < Vec < DepositResult > > > ,
70- ) {
71- let vouts = match tx[ "vout" ] . as_array ( ) {
72- Some ( v) => v,
73- None => return ,
74- } ;
114+ async fn process_transaction_static (
115+ tx : & serde_json :: Value ,
116+ block_num : u64 ,
117+ caip2 : & str ,
118+ targets : & Arc < HashSet < String > > ,
119+ results : & Arc < Mutex < Vec < DepositResult > > > ,
120+ ) {
121+ let vouts = match tx[ "vout" ] . as_array ( ) {
122+ Some ( v) => v,
123+ None => return ,
124+ } ;
75125
76- let mut from_address = "unknown" . to_string ( ) ;
77- if let Some ( vins) = tx[ "vin" ] . as_array ( ) {
78- if let Some ( first_vin) = vins. first ( ) {
79- if let Some ( prevout) = first_vin. get ( "prevout" ) {
80- if let Some ( addr) = extract_btc_address ( prevout) {
81- from_address = addr;
82- }
126+ let tx_hash = tx[ "txid" ] . as_str ( ) . unwrap_or ( "unknown" ) . to_string ( ) ;
127+
128+ let mut from_address = "unknown" . to_string ( ) ;
129+ if let Some ( vins) = tx[ "vin" ] . as_array ( ) {
130+ if let Some ( first_vin) = vins. first ( ) {
131+ if let Some ( prevout) = first_vin. get ( "prevout" ) {
132+ if let Some ( addr) = extract_btc_address ( prevout) {
133+ from_address = addr;
83134 }
84135 }
85136 }
137+ }
86138
87- for vout in vouts {
88- if let Some ( to_address) = extract_btc_address ( vout) {
89- if targets. contains ( & to_address) {
90- let btc_val_f64 = vout[ "value" ] . as_f64 ( ) . unwrap_or ( 0.0 ) ;
91-
92- let decimal_val = Decimal :: from_f64 ( btc_val_f64) . unwrap_or_default ( ) ;
93- let sats_decimal = decimal_val * Decimal :: new ( 100_000_000 , 0 ) ;
94- let raw_amount = sats_decimal. trunc ( ) . to_string ( ) ;
95-
96- results. lock ( ) . await . push ( DepositResult {
97- chain : self . caip2 . clone ( ) ,
98- token : "Native" . to_string ( ) ,
99- from_address : from_address. clone ( ) ,
100- to_address,
101- amount_raw : raw_amount. clone ( ) ,
102- amount_clean : format_to_human ( & raw_amount, BTC_DECIMALS ) ,
103- block_number : block_num,
104- } ) ;
105- }
139+ for vout in vouts {
140+ if let Some ( to_address) = extract_btc_address ( vout) {
141+ if targets. contains ( & to_address) {
142+ let btc_val_f64 = vout[ "value" ] . as_f64 ( ) . unwrap_or ( 0.0 ) ;
143+
144+ let decimal_val = Decimal :: from_f64 ( btc_val_f64) . unwrap_or_default ( ) ;
145+ let sats_decimal = decimal_val * Decimal :: new ( 100_000_000 , 0 ) ;
146+ let raw_amount = sats_decimal. trunc ( ) . to_string ( ) ;
147+
148+ results. lock ( ) . await . push ( DepositResult {
149+ chain : caip2. to_string ( ) ,
150+ token : "Native" . to_string ( ) ,
151+ from_address : from_address. clone ( ) ,
152+ to_address,
153+ amount_raw : raw_amount. clone ( ) ,
154+ amount_clean : format_to_human ( & raw_amount, BTC_DECIMALS ) ,
155+ block_number : block_num,
156+ tx_hash : tx_hash . clone ( ) ,
157+ } ) ;
106158 }
107159 }
108160 }
0 commit comments