Init project

.env.example

@@ -0,0 +1 @@
+CRYPTOHUNTER_DATABASE_PASSWORD=Passw0rd

.gitignore

@@ -0,0 +1,5 @@
+/target
+
+/snapshots
+.env
+config.yaml

Cargo.toml

@@ -0,0 +1,21 @@
+[package]
+name = "cryptohunter"
+version = "0.1.7"
+edition = "2024"
+
+[dependencies]
+serde = { version = "1.0", features = ["derive"] }
+serde_yaml = "0.9"
+clap = { version = "4.0", features = ["derive"] }
+postgres = "0.19"
+rand = "0.8"
+bitcoin = { version = "0.30", features = ["rand"] }
+reqwest = { version = "0.11", features = ["blocking", "json"] }
+tokio = { version = "1", features = ["full"] }
+csv = "1.1"
+crossbeam-channel = "0.5"
+log = "0.4"
+simple_logger = "4.0"
+lazy_static = "1.4"
+indicatif = "0.17"
+rayon = "1.10.0"

Dockerfile

@@ -0,0 +1,26 @@
+FROM rust:1.87.0-slim-bullseye AS builder
+WORKDIR /app
+
+RUN apt-get update && \
+    apt-get install -y \
+    pkg-config \
+    libssl-dev \
+    && rm -rf /var/lib/apt/lists/*
+COPY Cargo.toml Cargo.lock ./
+COPY ./src ./src
+RUN cargo build --release
+
+
+FROM debian:bullseye-slim
+WORKDIR /app
+
+RUN apt-get update && \
+    apt-get install -y \
+    libpq5 \
+    openssl \
+    ca-certificates \
+    && update-ca-certificates --fresh \
+    && rm -rf /var/lib/apt/lists/*
+COPY --from=builder /app/target/release/cryptohunter /app/cryptohunter
+
+ENTRYPOINT ["/app/cryptohunter"]

Makefile

@@ -0,0 +1,2 @@
+-include .env
+export

docker-compose.yaml

@@ -0,0 +1,29 @@
+services:
+  app:
+    build: .
+    environment:
+      - RUST_LOG=info
+    volumes:
+      - ./config.yaml:/app/config.yaml
+      - ./snapshots:/snapshots
+    depends_on:
+      database:
+        condition: service_healthy
+    command: search bitcoin run
+
+  database:
+    image: postgres:16-alpine
+    environment:
+      POSTGRES_USER: cryptohunter
+      POSTGRES_PASSWORD: ${CRYPTOHUNTER_DATABASE_PASSWORD:-12345678}
+      POSTGRES_DB: cryptohunter
+    volumes:
+      - database-data:/var/lib/postgresql/data
+    healthcheck:
+      test: ["CMD-SHELL", "pg_isready -U cryptohunter -d cryptohunter"]
+      interval: 5s
+      timeout: 5s
+      retries: 5
+
+volumes:
+  database-data:

src/cli.rs

@@ -0,0 +1,41 @@
+use clap::{Parser, Subcommand};
+
+#[derive(Parser)]
+#[command(version, about, long_about = None)]
+pub struct Cli {
+    #[arg(short, long, default_value = "config.yaml")]
+    pub config: String,
+
+    #[command(subcommand)]
+    pub command: Commands,
+}
+
+#[derive(Subcommand)]
+pub enum Commands {
+    /// Search for wallets
+    Search {
+        network: Option<String>,
+        
+        #[command(subcommand)]
+        command: SearchSubcommand,
+    },
+    /// Manage snapshots
+    Snapshots {
+        network: String,
+        
+        #[command(subcommand)]
+        command: SnapshotSubcommand,
+    },
+}
+
+#[derive(Subcommand)]
+pub enum SearchSubcommand {
+    /// Run search process
+    Run,
+}
+
+#[derive(Subcommand)]
+pub enum SnapshotSubcommand {
+    /// Load snapshot into database
+    Load { path: String },
+}

src/config.rs

@@ -0,0 +1,35 @@
+use serde::Deserialize;
+use std::collections::HashMap;
+use std::fs::File;
+use std::io::Read;
+use std::path::Path;
+
+#[derive(Debug, Deserialize)]
+pub struct Config {
+    pub networks: HashMap<String, NetworkConfig>,
+    pub notifications: NotificationConfig,
+}
+
+#[derive(Debug, Deserialize, Clone)]
+pub struct NetworkConfig {
+    pub key_generator_tasks: usize,
+    pub balance_checker_tasks: usize,
+    #[serde(flatten)]
+    pub params: HashMap<String, serde_yaml::Value>,
+}
+
+#[derive(Debug, Deserialize, Clone)]
+pub struct NotificationConfig {
+    pub telegram_bot_token: String,
+    pub telegram_user_id: String,
+}
+
+impl Config {
+    pub fn from_file<P: AsRef<Path>>(path: P) -> Result<Self, Box<dyn std::error::Error>> {
+        let mut file = File::open(path)?;
+        let mut contents = String::new();
+        file.read_to_string(&mut contents)?;
+        let config: Config = serde_yaml::from_str(&contents)?;
+        Ok(config)
+    }
+}

src/main.rs

@@ -0,0 +1,176 @@
+mod cli;
+mod config;
+mod networks;
+mod notification;
+
+use clap::Parser;
+use cli::Cli;
+use config::Config;
+use crossbeam_channel::{bounded, unbounded};
+use lazy_static::lazy_static;
+use networks::create_network;
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::thread;
+use std::time::Duration;
+
+fn main() -> Result<(), Box<dyn std::error::Error>> {
+    simple_logger::init_with_level(log::Level::Info)?;
+    let cli = Cli::parse();
+    let config = Config::from_file(&cli.config)?;
+
+    match &cli.command {
+        cli::Commands::Snapshots { network, command } => {
+            let network_config = config
+                .networks
+                .get(network)
+                .ok_or(format!("Config for network {} not found", network))?;
+            let network_obj = create_network(network, &network_config.params)?;
+
+            match command {
+                cli::SnapshotSubcommand::Load { path } => {
+                    network_obj.load_snapshot(path)?;
+                    log::info!("Snapshot loaded successfully for {}", network);
+                }
+            }
+        }
+        cli::Commands::Search { network, command } => {
+            match command {
+                cli::SearchSubcommand::Run => {
+                    let networks_to_run: HashMap<_, _> = match network {
+                        Some(net) => {
+                            let cfg = config.networks.get(net).ok_or("Network not found")?;
+                            [(net.as_str(), cfg)].into()
+                        }
+                        None => config
+                            .networks
+                            .iter()
+                            .map(|(k, v)| (k.as_str(), v))
+                            .collect(),
+                    };
+
+                    for (name, net_cfg) in networks_to_run {
+                        run_network_pipeline(name, net_cfg, &config.notifications)?;
+                    }
+                },
+            }
+        }
+    }
+
+    Ok(())
+}
+
+fn run_network_pipeline(
+    network_name: &str,
+    config: &config::NetworkConfig,
+    notification_cfg: &config::NotificationConfig,
+) -> Result<(), Box<dyn std::error::Error>> {
+    log::info!("Starting pipeline for {}", network_name);
+
+    let network = Arc::new(create_network(network_name, &config.params)?);
+    let (keypair_sender, keypair_receiver) = bounded(100);
+    let (notification_sender, notification_receiver) = unbounded();
+
+    // Клонируем имя сети для использования в замыканиях
+    let network_name_str = network_name.to_string();
+
+    // Key generation workers
+    for _ in 0..config.key_generator_tasks {
+        let sender = keypair_sender.clone();
+        let net = network.clone();
+        thread::spawn(move || loop {
+            let keypair = net.generate_keypair();
+            if let Err(e) = sender.send(keypair) {
+                log::error!("Keypair send error: {}", e);
+                break;
+            }
+        });
+    }
+
+    // Balance checker workers
+    for _ in 0..config.balance_checker_tasks {
+        let receiver = keypair_receiver.clone();
+        let sender = notification_sender.clone();
+        let net = network.clone();
+        let name = network_name_str.clone(); // Клонируем String
+        thread::spawn(move || loop {
+            match receiver.recv() {
+                Ok((private_key, address)) => {
+                    match net.check_balance(&address) {
+                        Ok(balance) if balance > 0 => {
+                            let formatted_balance = format_balance(balance);
+                            let message = format!(
+                                "💰 *Balance found!*\n\n*Network:* {}\n*Address:* `{}`\n*Private Key:* `{}`\n*Balance:* {}",
+                                name, address, private_key, formatted_balance,
+                            );
+                            if let Err(e) = sender.send(message) {
+                                log::error!("Notification send error: {}", e);
+                            }
+                        }
+                        Ok(_) => {} // Баланс нулевой, ничего не делаем
+                        Err(e) => log::error!("Balance check error: {}", e),
+                    }
+                }
+                Err(e) => {
+                    log::error!("Keypair receive error: {}", e);
+                    break;
+                }
+            }
+        });
+    }
+
+    // Notification worker
+    let notifier_cfg = notification_cfg.clone();
+    thread::spawn(move || {
+        for message in notification_receiver {
+            if let Err(e) = notification::send_telegram_message(
+                &notifier_cfg.telegram_bot_token,
+                &notifier_cfg.telegram_user_id,
+                &message,
+            ) {
+                log::error!("Telegram send error: {}", e);
+            }
+        }
+    });
+
+    log::info!("Pipeline started for {}", network_name);
+    loop {
+        thread::sleep(Duration::from_secs(60));
+    }
+}
+
+lazy_static! {
+    static ref THRESHOLDS: Vec<(u64, &'static str)> = vec![
+        (1_000_000_000_000_000_000u64, " E"),
+        (1_000_000_000_000_000u64, " P"),
+        (1_000_000_000_000u64, " T"),
+        (1_000_000_000u64, " G"),
+        (1_000_000u64, " M"),
+        (1_000u64, " k"),
+    ];
+}
+
+fn format_balance(balance: i64) -> String {
+    let abs_balance = balance.abs() as u64;
+
+    let mut divider = 1;
+    let mut suffix = " sat";
+    
+    for (thresh, s) in THRESHOLDS.iter() {
+        if abs_balance >= *thresh {
+            divider = thresh.clone();
+            suffix = s;
+            break;
+        }
+    }
+    
+    let quotient = abs_balance / divider;
+    let remainder = abs_balance % divider;
+    
+    if remainder == 0 {
+        format!("{}{}", quotient, suffix)
+    } else {
+        let fractional = remainder as f64 / divider as f64;
+        format!("{:.4}{}", quotient as f64 + fractional, suffix)
+    }
+}

src/networks/bitcoin.rs

@@ -0,0 +1,187 @@
+use bitcoin::{
+    secp256k1, Address, Network, PrivateKey, PublicKey,
+};
+use indicatif::{MultiProgress, ProgressBar, ProgressStyle};
+use postgres::{Client, NoTls};
+use serde_yaml::Value;
+use std::collections::HashMap;
+use std::error::Error;
+use std::fs::File;
+use std::io::{BufReader, Read, Write};
+use std::time::{Duration, Instant};
+
+pub struct Bitcoin {
+    database_url: String,
+    database_table: String,
+}
+
+impl Bitcoin {
+    pub fn new(params: &HashMap<String, Value>) -> Result<Self, Box<dyn Error>> {
+        let database_url = params
+            .get("database_url")
+            .and_then(Value::as_str)
+            .ok_or("Missing database_url for Bitcoin")?
+            .to_string();
+            
+        let database_table = params
+            .get("database_table")
+            .and_then(Value::as_str)
+            .ok_or("Missing database_table for Bitcoin")?
+            .to_string();
+
+        Ok(Self {
+            database_url,
+            database_table,
+        })
+    }
+
+    fn connect(&self) -> Result<Client, Box<dyn Error>> {
+        let client = Client::connect(&self.database_url, NoTls)?;
+        Ok(client)
+    }
+}
+
+impl super::Network for Bitcoin {
+    fn generate_keypair(&self) -> (String, String) {
+        let secp = secp256k1::Secp256k1::new();
+        let private_key = PrivateKey::new(secp256k1::SecretKey::new(&mut rand::thread_rng()), Network::Bitcoin);
+        let public_key = PublicKey::from_private_key(&secp, &private_key);
+        let address = Address::p2pkh(&public_key, Network::Bitcoin).to_string();
+
+        (private_key.to_wif(), address)
+    }
+
+    fn check_balance(&self, address: &str) -> Result<i64, Box<dyn Error>> {
+        let mut client = self.connect()?;
+        let row = client.query_opt(
+            &format!(
+                "SELECT balance FROM {} WHERE address = $1",
+                self.database_table
+            ),
+            &[&address],
+        )?;
+
+        Ok(row.map(|r| r.get(0)).unwrap_or(0))
+    }
+
+    fn load_snapshot(&self, snapshot_path: &str) -> Result<(), Box<dyn Error>> {
+        let start_time = Instant::now();
+        let mut client = self.connect()?;
+        let file = File::open(snapshot_path)?;
+        let file_size = file.metadata()?.len();
+        let index_name = format!("{}__address__ix", self.database_table);
+        let multi_progress = MultiProgress::new();
+
+        // 1. Preparing database
+        let prep_pb = multi_progress.add(ProgressBar::new_spinner());
+        prep_pb.set_style(
+            ProgressStyle::default_spinner()
+                .tick_chars("⠁⠂⠄⡀⢀⠠⠐⠈ ")
+                .template("{spinner} {msg}").unwrap()
+        );
+        prep_pb.set_message("⚙️ Preparing database...");
+        prep_pb.enable_steady_tick(Duration::from_millis(100));
+
+        client.execute("SET synchronous_commit = off", &[])?;
+        client.execute("SET maintenance_work_mem = '4GB'", &[])?;
+        client.execute("SET work_mem = '2GB'", &[])?;
+        client.execute(
+            &format!("DROP TABLE IF EXISTS {}", self.database_table),
+            &[],
+        )?;
+        client.execute(
+            &format!("DROP INDEX IF EXISTS {}", index_name),
+            &[],
+        )?;
+        client.execute(
+            &format!(
+                "CREATE TABLE {} (address TEXT, balance BIGINT)",
+                self.database_table
+            ),
+            &[],
+        )?;
+
+        prep_pb.finish_with_message("✅ Database prepared");
+
+        // 2. Copy data to database
+        let copy_pb = multi_progress.add(ProgressBar::new(file_size));
+        copy_pb.set_style(ProgressStyle::default_bar()
+            .template("{spinner:.green} [{elapsed_precise}] [{bar:40.cyan/blue}] {bytes}/{total_bytes} ({eta}) | {binary_bytes_per_sec}")
+            .unwrap()
+            .progress_chars("#>-"));
+        copy_pb.set_message("️📥 Copy data to database...");
+
+        let copy_stmt = format!(
+            "COPY {} FROM STDIN WITH (FORMAT csv, DELIMITER E'\t', HEADER)",
+            self.database_table,
+        );
+        let mut writer = client.copy_in(&copy_stmt)?;
+        
+        let file = File::open(snapshot_path)?;
+        let mut reader = BufReader::new(file);
+        let mut buffer = [0u8; 65536]; // 64KB буфер
+        let mut total_bytes = 0;
+        
+        loop {
+            let bytes_read = reader.read(&mut buffer)?;
+            if bytes_read == 0 {
+                break;
+            }
+            
+            writer.write_all(&buffer[..bytes_read])?;
+            total_bytes += bytes_read as u64;
+            copy_pb.set_position(total_bytes);
+        }
+        
+        writer.finish()?;
+        copy_pb.set_position(file_size);
+        copy_pb.finish_with_message("✅ Data copied to database");
+
+        // 3. Creating index
+        let index_pb = multi_progress.add(ProgressBar::new_spinner());
+        index_pb.set_message("🔀 Creating index...");
+        index_pb.enable_steady_tick(Duration::from_millis(100));
+
+        client.execute(
+            &format!(
+                "CREATE INDEX {} ON {} USING HASH (address)",
+                index_name,
+                self.database_table,
+            ),
+            &[],
+        )?;
+
+        index_pb.finish_with_message("✅ Index created");
+
+        // 4. Reseting temporary settings
+        let final_pb = multi_progress.add(ProgressBar::new_spinner());
+        final_pb.set_message("🔁 Reset database settings");
+        final_pb.enable_steady_tick(Duration::from_millis(100));
+
+        client.execute("RESET synchronous_commit", &[])?;
+        client.execute("RESET maintenance_work_mem", &[])?;
+        client.execute("RESET work_mem", &[])?;
+
+        final_pb.finish_with_message("✅ Database settings reseted");
+        
+        // 5. Final
+        multi_progress.clear()?;
+
+        let records_count: i64 = client.query_one(
+            &format!(
+                "SELECT COUNT(*) FROM {}",
+                self.database_table,
+            ),
+            &[],
+        ).map(|row| row.get(0))?;
+
+        let data_size: f64 = file_size as f64 / (1024 * 1024 * 1024) as f64;
+        println!("\n🎉 Snapshot loaded successfully!");
+        println!("📊 Statistics:");
+        println!("   Total records processed: {}", records_count);
+        println!("   Data size:              {:.2} GB", data_size);
+        println!("   Execution time:         {:?}", start_time.elapsed());
+
+        Ok(())
+    }
+}

src/networks/mod.rs

@@ -0,0 +1,35 @@
+pub mod bitcoin;
+
+use serde_yaml::Value;
+use std::collections::HashMap;
+use std::error::Error;
+
+pub trait Network: Send + Sync {
+    fn generate_keypair(&self) -> (String, String);
+    fn check_balance(&self, address: &str) -> Result<i64, Box<dyn Error>>;
+    // TODO: Remove
+    fn load_snapshot(&self, snapshot_path: &str) -> Result<(), Box<dyn Error>>;
+}
+
+pub trait SnapshotLoader {
+    fn load_snapshot(&self, snapshot_path: &str, format: Option<&str>) -> Result<(), Box<dyn Error>>;
+}
+
+pub fn create_network(
+    name: &str,
+    params: &HashMap<String, Value>,
+) -> Result<Box<dyn Network>, Box<dyn Error>> {
+    match name {
+        "bitcoin" => Ok(Box::new(bitcoin::Bitcoin::new(params)?)),
+        _ => Err(format!("Unsupported network: {}", name).into()),
+    }
+}
+
+pub fn create_snapshot_loader(
+    name: &str,
+    params: &HashMap<String, Value>,
+) -> Result<Box<dyn SnapshotLoader>, Box<dyn Error>> {
+    match name {
+        _ => Err(format!("Unsupported network: {}", name).into()),
+    }
+}

src/notification.rs

@@ -0,0 +1,27 @@
+use reqwest::blocking::Client;
+use std::error::Error;
+
+pub fn send_telegram_message(
+    bot_token: &str,
+    user_id: &str,
+    message: &str,
+) -> Result<(), Box<dyn Error>> {
+    let url = format!(
+        "https://api.telegram.org/bot{}/sendMessage",
+        bot_token
+    );
+    let params = [
+        ("chat_id", user_id),
+        ("text", message),
+        ("parse_mode", "Markdown"),
+    ];
+
+    let client = Client::new();
+    let response = client.post(&url).form(&params).send()?;
+    
+    if response.status().is_success() {
+        Ok(())
+    } else {
+        Err(format!("Telegram API error: {}", response.text()?).into())
+    }
+}