feat: ability to download hord.sqlite

Author: Ludo Galabru

components/chainhook-cli/src/archive/mod.rs

@@ -1,5 +1,6 @@
 use crate::config::Config;
-use chainhook_types::StacksNetwork;
+use chainhook_event_observer::utils::Context;
+use chainhook_types::{BitcoinNetwork, StacksNetwork};
 use clarinet_files::FileLocation;
 use flate2::read::GzDecoder;
 use futures_util::StreamExt;
@@ -15,13 +16,21 @@ pub fn default_tsv_sha_file_path(network: &StacksNetwork) -> String {
     format!("{:?}-stacks-events.sha256", network).to_lowercase()
 }
 
+pub fn default_sqlite_file_path(_network: &BitcoinNetwork) -> String {
+    format!("hord.sqlite").to_lowercase()
+}
+
+pub fn default_sqlite_sha_file_path(_network: &BitcoinNetwork) -> String {
+    format!("hord.sqlite.sha256").to_lowercase()
+}
+
 pub async fn download_tsv_file(config: &Config) -> Result<(), String> {
     let mut destination_path = config.expected_cache_path();
     std::fs::create_dir_all(&destination_path).unwrap_or_else(|e| {
         println!("{}", e.to_string());
     });
 
-    let remote_sha_url = config.expected_remote_tsv_sha256();
+    let remote_sha_url = config.expected_remote_stacks_tsv_sha256();
     let res = reqwest::get(&remote_sha_url)
         .await
         .or(Err(format!("Failed to GET from '{}'", &remote_sha_url)))?
@@ -32,11 +41,10 @@ pub async fn download_tsv_file(config: &Config) -> Result<(), String> {
     let mut local_sha_file_path = destination_path.clone();
     local_sha_file_path.push(default_tsv_sha_file_path(&config.network.stacks_network));
 
-    println!("1");
     let local_sha_file = FileLocation::from_path(local_sha_file_path);
     let _ = local_sha_file.write_content(&res.to_vec());
 
-    let file_url = config.expected_remote_tsv_url();
+    let file_url = config.expected_remote_stacks_tsv_url();
     let res = reqwest::get(&file_url)
         .await
         .or(Err(format!("Failed to GET from '{}'", &file_url)))?;
@@ -76,6 +84,68 @@ pub async fn download_tsv_file(config: &Config) -> Result<(), String> {
     Ok(())
 }
 
+pub async fn download_sqlite_file(config: &Config) -> Result<(), String> {
+    let mut destination_path = config.expected_cache_path();
+    std::fs::create_dir_all(&destination_path).unwrap_or_else(|e| {
+        println!("{}", e.to_string());
+    });
+
+    let remote_sha_url = config.expected_remote_ordinals_sqlite_sha256();
+    let res = reqwest::get(&remote_sha_url)
+        .await
+        .or(Err(format!("Failed to GET from '{}'", &remote_sha_url)))?
+        .bytes()
+        .await
+        .or(Err(format!("Failed to GET from '{}'", &remote_sha_url)))?;
+
+    let mut local_sha_file_path = destination_path.clone();
+    local_sha_file_path.push(default_sqlite_sha_file_path(
+        &config.network.bitcoin_network,
+    ));
+
+    let local_sha_file = FileLocation::from_path(local_sha_file_path);
+    let _ = local_sha_file.write_content(&res.to_vec());
+
+    let file_url = config.expected_remote_ordinals_sqlite_url();
+    let res = reqwest::get(&file_url)
+        .await
+        .or(Err(format!("Failed to GET from '{}'", &file_url)))?;
+
+    // Download chunks
+    let (tx, rx) = flume::bounded(0);
+    destination_path.push(default_sqlite_file_path(&config.network.bitcoin_network));
+
+    let decoder_thread = std::thread::spawn(move || {
+        let input = ChannelRead::new(rx);
+        let mut decoder = GzDecoder::new(input);
+        let mut content = Vec::new();
+        let _ = decoder.read_to_end(&mut content);
+        let mut file = fs::File::create(&destination_path).unwrap();
+        if let Err(e) = file.write_all(&content[..]) {
+            println!("unable to write file: {}", e.to_string());
+            std::process::exit(1);
+        }
+    });
+
+    if res.status() == reqwest::StatusCode::OK {
+        let mut stream = res.bytes_stream();
+        while let Some(item) = stream.next().await {
+            let chunk = item.or(Err(format!("Error while downloading file")))?;
+            tx.send_async(chunk.to_vec())
+                .await
+                .map_err(|e| format!("unable to download stacks event: {}", e.to_string()))?;
+        }
+        drop(tx);
+    }
+
+    tokio::task::spawn_blocking(|| decoder_thread.join())
+        .await
+        .unwrap()
+        .unwrap();
+
+    Ok(())
+}
+
 // Wrap a channel into something that impls `io::Read`
 struct ChannelRead {
     rx: flume::Receiver<Vec<u8>>,
@@ -105,3 +175,129 @@ impl Read for ChannelRead {
         self.current.read(buf)
     }
 }
+
+pub async fn download_stacks_dataset_if_required(config: &mut Config, ctx: &Context) -> bool {
+    if config.is_initial_ingestion_required() {
+        // Download default tsv.
+        if config.rely_on_remote_stacks_tsv() && config.should_download_remote_stacks_tsv() {
+            let url = config.expected_remote_stacks_tsv_url();
+            let mut tsv_file_path = config.expected_cache_path();
+            tsv_file_path.push(default_tsv_file_path(&config.network.stacks_network));
+            let mut tsv_sha_file_path = config.expected_cache_path();
+            tsv_sha_file_path.push(default_tsv_sha_file_path(&config.network.stacks_network));
+
+            // Download archive if not already present in cache
+            // Load the local
+            let local_sha_file = FileLocation::from_path(tsv_sha_file_path).read_content();
+            let sha_url = config.expected_remote_stacks_tsv_sha256();
+
+            let remote_sha_file = match reqwest::get(&sha_url).await {
+                Ok(response) => response.bytes().await,
+                Err(e) => Err(e),
+            };
+            match (local_sha_file, remote_sha_file) {
+                (Ok(local), Ok(remote_response)) => {
+                    println!("{:?}", local);
+                    println!("{:?}", remote_response);
+                }
+                (Ok(local), _) => {
+                    // println!("Local: {:?}", local)
+                }
+                (_, _) => {
+                    // We will download the latest file
+                    println!("error reading local / remote");
+                }
+            }
+
+            if !tsv_file_path.exists() {
+                info!(ctx.expect_logger(), "Downloading {}", url);
+                match download_tsv_file(&config).await {
+                    Ok(_) => {}
+                    Err(e) => {
+                        error!(ctx.expect_logger(), "{}", e);
+                        std::process::exit(1);
+                    }
+                }
+            } else {
+                info!(
+                    ctx.expect_logger(),
+                    "Building in-memory chainstate from file {}",
+                    tsv_file_path.display()
+                );
+            }
+            config.add_local_stacks_tsv_source(&tsv_file_path);
+        }
+        true
+    } else {
+        info!(
+            ctx.expect_logger(),
+            "Streaming blocks from stacks-node {}", config.network.stacks_node_rpc_url
+        );
+        false
+    }
+}
+
+pub async fn download_ordinals_dataset_if_required(config: &Config, ctx: &Context) -> bool {
+    if config.is_initial_ingestion_required() {
+        // Download default tsv.
+        if config.rely_on_remote_ordinals_sqlite()
+            && config.should_download_remote_ordinals_sqlite()
+        {
+            let url = config.expected_remote_ordinals_sqlite_url();
+            let mut sqlite_file_path = config.expected_cache_path();
+            sqlite_file_path.push(default_sqlite_file_path(&config.network.bitcoin_network));
+            let mut tsv_sha_file_path = config.expected_cache_path();
+            tsv_sha_file_path.push(default_sqlite_sha_file_path(
+                &config.network.bitcoin_network,
+            ));
+
+            // Download archive if not already present in cache
+            // Load the local
+            let local_sha_file = FileLocation::from_path(tsv_sha_file_path).read_content();
+            let sha_url = config.expected_remote_ordinals_sqlite_sha256();
+
+            let remote_sha_file = match reqwest::get(&sha_url).await {
+                Ok(response) => response.bytes().await,
+                Err(e) => Err(e),
+            };
+            match (local_sha_file, remote_sha_file) {
+                (Ok(local), Ok(remote_response)) => {
+                    println!("{:?}", local);
+                    println!("{:?}", remote_response);
+                }
+                (Ok(local), _) => {
+                    // println!("Local: {:?}", local)
+                }
+                (_, _) => {
+                    // We will download the latest file
+                    println!("error reading local / remote");
+                }
+            }
+
+            if !sqlite_file_path.exists() {
+                info!(ctx.expect_logger(), "Downloading {}", url);
+                match download_sqlite_file(&config).await {
+                    Ok(_) => {}
+                    Err(e) => {
+                        error!(ctx.expect_logger(), "{}", e);
+                        std::process::exit(1);
+                    }
+                }
+            } else {
+                info!(
+                    ctx.expect_logger(),
+                    "Basing ordinals evaluation on database {}",
+                    sqlite_file_path.display()
+                );
+            }
+            // config.add_local_ordinals_sqlite_source(&sqlite_file_path);
+        }
+        true
+    } else {
+        info!(
+            ctx.expect_logger(),
+            "Streaming blocks from bitcoind {}", config.network.stacks_node_rpc_url
+        );
+        false
+    }
+}