Enhance indexer transaction data analysis and storage

client/src/client.rs

@@ -64,7 +64,7 @@ impl ConnectionDelegate<RemoteAddr, Session> for Delegate {
         TcpStream::connect(remote).unwrap_or_else(|err| {
             #[cfg(feature = "log")]
             log::error!("Unable to connect BP Node {remote} due to {err}");
-            eprintln!("Unable to connect BP Node {remote}");
+            eprintln!("Unable to connect BP Node {remote} due to {err}");
             exit(1);
         })
     }
@@ -74,9 +74,9 @@ impl ConnectionDelegate<RemoteAddr, Session> for Delegate {
         log::info!("connection to the server is established");
     }
 
-    fn on_disconnect(&mut self, err: Error, _attempt: usize) -> OnDisconnect {
+    fn on_disconnect(&mut self, _err: Error, _attempt: usize) -> OnDisconnect {
         #[cfg(feature = "log")]
-        log::error!("disconnected due to {err}");
+        log::error!("disconnected due to {_err}");
         OnDisconnect::Terminate
     }
 

client/src/exporter.rs

@@ -57,22 +57,27 @@ impl ConnectionDelegate<RemoteAddr, Session> for BlockExporter {
 
     fn connect(&mut self, remote: &RemoteAddr) -> Session {
         TcpStream::connect(remote).unwrap_or_else(|err| {
+            #[cfg(feature = "log")]
             log::error!(target: NAME, "Unable to connect BP Node {remote} due to {err}");
+            #[cfg(feature = "log")]
             log::warn!(target: NAME, "Stopping RPC import thread");
             exit(1);
         })
     }
 
     fn on_established(&mut self, remote: SocketAddr, _attempt: usize) {
+        #[cfg(feature = "log")]
         log::info!(target: NAME, "Connected to BP Node {remote}, sending `hello(...)`");
     }
 
     fn on_disconnect(&mut self, err: std::io::Error, _attempt: usize) -> OnDisconnect {
+        #[cfg(feature = "log")]
         log::error!(target: NAME, "BP Node got disconnected due to {err}");
         exit(1)
     }
 
     fn on_io_error(&mut self, err: reactor::Error<ImpossibleResource, NetTransport<Session>>) {
+        #[cfg(feature = "log")]
         log::error!(target: NAME, "I/O error in communicating with BP Node: {err}");
         self.disconnect();
     }
@@ -85,21 +90,25 @@ impl ClientDelegate<RemoteAddr, Session> for BlockExporter {
         match msg {
             ImporterReply::Filters(filters) => {
                 if self.filters_received {
+                    #[cfg(feature = "log")]
                     log::warn!(target: NAME, "Received duplicate filters");
                 } else {
+                    #[cfg(feature = "log")]
                     log::info!(target: NAME, "Received filters");
                 }
                 self.filters = filters;
                 self.filters_received = true;
             }
             ImporterReply::Error(failure) => {
+                #[cfg(feature = "log")]
                 log::error!(target: NAME, "Received error from BP Node: {failure}");
                 self.disconnect();
             }
         }
     }
 
     fn on_reply_unparsable(&mut self, err: <Self::Reply as Frame>::Error) {
+        #[cfg(feature = "log")]
         log::error!("Invalid message from BP Node: {err}");
     }
 }

client/src/main.rs

@@ -56,7 +56,9 @@ fn cb(reply: Response) {
         Response::Failure(failure) => {
             println!("Failure: {failure}");
         }
-        Response::Pong(_noise) => {}
+        Response::Pong(_noise) => {
+            println!("Pong from server");
+        }
         Response::Status(status) => {
             println!("{}", serde_yaml::to_string(&status).unwrap());
         }

providers/bitcoincore/src/main.rs

@@ -39,7 +39,11 @@ use strict_encoding::Ident;
 
 pub const AGENT: &str = "BC_BP";
 
-pub const BLOCK_SEPARATOR: [u8; 4] = [0xF9, 0xBE, 0xB4, 0xD9];
+pub const BITCOIN_BLOCK_SEPARATOR: [u8; 4] = [0xF9, 0xBE, 0xB4, 0xD9];
+pub const TESTNET_BLOCK_SEPARATOR: [u8; 4] = [0x0B, 0x11, 0x09, 0x07];
+pub const TESTNET4_BLOCK_SEPARATOR: [u8; 4] = [0x1c, 0x16, 0x3f, 0x28];
+pub const SIGNET_BLOCK_SEPARATOR: [u8; 4] = [0x0A, 0x03, 0xCF, 0x40];
+pub const REGTEST_BLOCK_SEPARATOR: [u8; 4] = [0xFA, 0xBF, 0xB5, 0xDA];
 
 /// Command-line arguments
 #[derive(Parser)]
@@ -104,6 +108,15 @@ fn read_blocks(client: Client<ExporterPub>, args: Args) {
         exit(1);
     }
 
+    // Select the correct block separator according to the network type
+    let block_separator = match args.network {
+        Network::Mainnet => BITCOIN_BLOCK_SEPARATOR,
+        Network::Testnet3 => TESTNET_BLOCK_SEPARATOR,
+        Network::Testnet4 => TESTNET4_BLOCK_SEPARATOR,
+        Network::Signet => SIGNET_BLOCK_SEPARATOR,
+        Network::Regtest => REGTEST_BLOCK_SEPARATOR,
+    };
+
     let mut file_no: u32 = 0;
     let mut total_blocks: u32 = 0;
     let mut total_tx: u64 = 0;
@@ -138,7 +151,13 @@ fn read_blocks(client: Client<ExporterPub>, args: Args) {
                     exit(4);
                 }
             }
-            if buf != BLOCK_SEPARATOR {
+
+            if buf == [0x00, 0x00, 0x00, 0x00] {
+                log::info!("Reached end of block file");
+                break;
+            }
+
+            if buf != block_separator {
                 log::error!(
                     "Invalid block separator 0x{:02X}{:02X}{:02X}{:02X} before block #{block_no}",
                     buf[0],

src/bin/bpd.rs

@@ -27,16 +27,30 @@ extern crate clap;
 mod opts;
 
 use std::fs;
+use std::path::Path;
 use std::process::{ExitCode, Termination, exit};
 
 pub use bpnode;
-use bpnode::{Broker, BrokerError, Config, PATH_INDEXDB};
+use bpnode::{Broker, BrokerError, Config, PATH_INDEXDB, initialize_db_tables};
+use bpwallet::Network;
 use clap::Parser;
 use loglevel::LogLevel;
 use redb::Database;
 
 use crate::opts::{Command, Opts};
 
+// Exit status codes for different error conditions
+// see also constants in `db.rs`
+const EXIT_PATH_ACCESS_ERROR: i32 = 1;
+const EXIT_DB_EXISTS_ERROR: i32 = 2;
+const EXIT_DIR_CREATE_ERROR: i32 = 3;
+const EXIT_DB_CREATE_ERROR: i32 = 4;
+const EXIT_DB_OPEN_ERROR: i32 = 5;
+const EXIT_NETWORK_MISMATCH: i32 = 10;
+const EXIT_NO_NETWORK_INFO: i32 = 11;
+const EXIT_DB_NOT_FOUND: i32 = 12;
+
+/// Wrapper for result status to implement Termination trait
 struct Status(Result<(), BrokerError>);
 
 impl Termination for Status {
@@ -58,37 +72,124 @@ fn main() -> Status {
     log::debug!("Command-line arguments: {:#?}", &opts);
 
     match opts.command {
-        Some(Command::Init) => {
-            eprint!("Initializing ... ");
-            let index_path = opts.general.data_dir.join(PATH_INDEXDB);
-            match fs::exists(&index_path) {
-                Err(err) => {
-                    eprintln!("unable to access path '{}': {err}", index_path.display());
-                    exit(1);
-                }
-                Ok(true) => {
-                    eprintln!("index database directory already exists, cancelling");
-                    exit(2);
-                }
-                Ok(false) => {}
-            }
-            if let Err(err) = fs::create_dir_all(&opts.general.data_dir) {
+        Some(Command::Init) => initialize_database(&opts),
+        None => run_node(opts),
+    }
+}
+
+/// Initialize a new database for the BP Node
+fn initialize_database(opts: &Opts) -> Status {
+    eprint!("Initializing ... ");
+
+    // Prepare the database path
+    let index_path = opts.general.data_dir.join(PATH_INDEXDB);
+
+    // Check if database already exists
+    if let Err(err) = check_db_path(&index_path, false) {
+        return err;
+    }
+
+    // Create data directory if needed
+    if let Err(err) = fs::create_dir_all(&opts.general.data_dir) {
+        eprintln!(
+            "Unable to create data directory at '{}'\n{err}",
+            opts.general.data_dir.display()
+        );
+        exit(EXIT_DIR_CREATE_ERROR);
+    }
+
+    // Create the database
+    let db = match Database::create(&index_path) {
+        Ok(db) => db,
+        Err(err) => {
+            eprintln!("Unable to create index database.\n{err}");
+            exit(EXIT_DB_CREATE_ERROR);
+        }
+    };
+
+    // Initialize database with network information and create all tables
+    let network = opts.general.network;
+    initialize_db_tables(&db, network);
+
+    eprintln!("Index database initialized for {} network, exiting", network);
+    Status(Ok(()))
+}
+
+/// Run the BP Node service
+fn run_node(opts: Opts) -> Status {
+    let conf = Config::from(opts);
+    let index_path = conf.data_dir.join(PATH_INDEXDB);
+
+    // Check if database exists
+    if let Err(err) = check_db_path(&index_path, true) {
+        return err;
+    }
+
+    // Verify network configuration
+    verify_network_configuration(&index_path, &conf.network);
+
+    // Start the broker service
+    Status(Broker::start(conf).and_then(|runtime| runtime.run()))
+}
+
+/// Check if database path exists or not, depending on expected state
+fn check_db_path(index_path: &Path, should_exist: bool) -> Result<(), Status> {
+    match fs::exists(index_path) {
+        Err(err) => {
+            eprintln!("Unable to access path '{}': {err}", index_path.display());
+            exit(EXIT_PATH_ACCESS_ERROR);
+        }
+        Ok(exists) => {
+            if exists && !should_exist {
+                eprintln!("Index database directory already exists, cancelling");
+                exit(EXIT_DB_EXISTS_ERROR);
+            } else if !exists && should_exist {
                 eprintln!(
-                    "unable to create data directory at '{}'\n{err}",
-                    opts.general.data_dir.display()
+                    "ERROR: Database not found! Please initialize with 'bpd init' command first."
                 );
-                exit(3);
+                exit(EXIT_DB_NOT_FOUND);
             }
-            if let Err(err) = Database::create(&index_path) {
-                eprintln!("unable to create index database.\n{err}");
-                exit(4);
-            }
-            eprintln!("index database initialized, exiting");
-            Status(Ok(()))
-        }
-        None => {
-            let conf = Config::from(opts);
-            Status(Broker::start(conf).and_then(|runtime| runtime.run()))
         }
     }
+    Ok(())
+}
+
+/// Verify that database network configuration matches the configured network
+fn verify_network_configuration(index_path: &Path, configured_network: &Network) {
+    let Ok(db) = Database::open(index_path)
+        .inspect_err(|err| eprintln!("Error: could not open the database due to {err}"))
+    else {
+        exit(EXIT_DB_OPEN_ERROR)
+    };
+    let Ok(tx) = db
+        .begin_read()
+        .inspect_err(|err| eprintln!("Error: could not access the database due to {err}"))
+    else {
+        exit(EXIT_DB_OPEN_ERROR)
+    };
+    let Ok(main_table) = tx
+        .open_table(bpnode::db::TABLE_MAIN)
+        .inspect_err(|err| eprintln!("Error: could not open the main table due to {err}"))
+    else {
+        exit(EXIT_DB_OPEN_ERROR)
+    };
+    let Ok(Some(network_rec)) = main_table.get(bpnode::REC_NETWORK) else {
+        // Network information isn't found in the database
+        eprintln!("ERROR: Database exists but doesn't contain network information.");
+        eprintln!("Please reinitialize the database with `bpd init` command.");
+        exit(EXIT_NO_NETWORK_INFO);
+    };
+    let stored_network = String::from_utf8_lossy(network_rec.value());
+    if stored_network != configured_network.to_string() {
+        eprintln!("ERROR: Database network mismatch!");
+        eprintln!("Configured network: {}", configured_network);
+        eprintln!("Database network: {}", stored_network);
+        eprintln!("Each BP-Node instance works with a single chain.");
+        eprintln!(
+            "To use a different network, create a separate instance with a different data \
+             directory."
+        );
+        exit(EXIT_NETWORK_MISMATCH);
+    }
+    log::info!("Database network matches configured network: {}", stored_network);
 }