crawl BTC txs

btc_txs_crawler.rs

use anyhow::{anyhow, Result};
use cuckoofilter::CuckooFilter;
use futures::future::join_all;
use serde_json::Value;
use std::collections::hash_map::DefaultHasher;
use tracing::debug;

const REST_URL: &str = "https://btc.getblock.io/<YOUR OWN API KEY>/testnet";
const LAST_BLOCK_HASH: &str = "00000000000010f9d43bb5c329aef7c5ab0c4f115cb5a0f4484b0267769ee0ff";
const ACCOUNTS: [&str; 1] = ["tb1q95e88yfvytshvkk8pv7eguzj05qatlty0wm687"];

#[derive(Debug)]
enum TxDirection {
    Send,
    Receive,
}

#[derive(Debug)]
struct Tx<'a> {
    addr: &'a str,
    direction: TxDirection,
    value: &'a str,
    txid: &'a str,
}

pub fn get_accounts_filters() -> Result<CuckooFilter<DefaultHasher>> {
    let mut cf = CuckooFilter::new();
    for account in ACCOUNTS {
        cf.add(account.to_lowercase().as_str())?;
    }
    Ok(cf)
}

pub async fn handle_bitcoin_block(hash: &str) -> Result<()> {
    let block = reqwest::get(format!("{}/blockbook/api/block/{}", REST_URL, hash))
        .await?
        .json::<Value>()
        .await?;
    let block_height = block["height"].as_u64().ok_or(anyhow!(
        "an error occurred while parsing block: {:?}",
        block
    ))?;

    debug!("handling block: {}, height: {}", hash, block_height);
    let accounts_filter = get_accounts_filters()?;
    let trans = block["txs"]
        .as_array()
        .ok_or(anyhow!("expected an array"))?
        .into_iter()
        .filter_map(|tx| {
            let txid = tx["txid"].as_str()?;

            let mut vins = tx["vin"]
                .as_array()?
                .into_iter()
                .filter_map(|vin| {
                    let value = vin["value"].as_str()?;
                    vin["addresses"].as_array().and_then(|addresses| {
                        Some(
                            addresses
                                .into_iter()
                                .filter_map(|address| {
                                    Some(Tx {
                                        addr: address.as_str()?,
                                        direction: TxDirection::Send,
                                        value,
                                        txid,
                                    })
                                })
                                .collect::<Vec<Tx>>(),
                        )
                    })
                })
                .flatten()
                .filter(|item| accounts_filter.contains(item.addr.to_lowercase().as_str()))
                .collect::<Vec<_>>();
            let mut vouts = tx["vout"]
                .as_array()
                .unwrap()
                .into_iter()
                .filter_map(|vout| {
                    let value = vout["value"].as_str()?;
                    vout["scriptPubKey"]["addresses"]
                        .as_array()
                        .and_then(|addresses| {
                            Some(
                                addresses
                                    .into_iter()
                                    .filter_map(|address| {
                                        Some(Tx {
                                            addr: address.as_str()?,
                                            direction: TxDirection::Receive,
                                            value,
                                            txid,
                                        })
                                    })
                                    .collect::<Vec<Tx>>(),
                            )
                        })
                })
                .flatten()
                .filter(|item| accounts_filter.contains(item.addr.to_lowercase().as_str()))
                .collect::<Vec<_>>();

            vins.append(&mut vouts);
            Some(vins)
        })
        .flatten()
        .collect::<Vec<_>>();

    debug!("{:?}", trans);
    Ok(())
}

#[tokio::main]
async fn main() -> Result<()> {
    let mut last_block_hash = LAST_BLOCK_HASH;

    let response = reqwest::get(format!(
        "{}/rest/headers/{}/{}.json",
        REST_URL, 10, last_block_hash
    ))
    .await?
    .json::<Value>()
    .await?;

    let block_headers = response.as_array().ok_or(anyhow!("expected an array"))?;

    // handle bulk of blocks concurrently
    join_all(block_headers.iter().filter_map(|block| {
        block
            .get("nextblockhash")
            .and_then(|v| v.as_str())
            .and_then(|v| Some(last_block_hash = v));
        Some(handle_bitcoin_block(block["hash"].as_str()?))
    }))
    .await;

    Ok(())
}