// Copyright (C) 2024-2026 Whiterun LLC // // This software is licensed under the GNU Affero General Public License (AGPL), version 3.0 or later. // A copy of the license can be found in the LICENSE file or at https://www.gnu.org/licenses/agpl-3.0.html use std::{ sync::{atomic::AtomicBool, atomic::Ordering, Arc}, time::Duration, }; use crate::bcmr::parsedbcmr::{parse_bcmr_json, SOURCE_ON_CHAIN}; use crate::db::bcmr::insert_bcmr_data; use crate::db::bcmr::update_bcmr_failure; use crate::{ bcmr::parse_bcmr_from_opreturn, db::bcmr::{get_entries_missing_bcmr_download, AuthChainEntry}, }; use anyhow::*; use bitcoin_hashes::hex::ToHex; use bitcoincash::Script; use log::{info, warn}; use rand::thread_rng; use serde_json::Value; use sqlx::SqlitePool; use tokio::task::JoinHandle; use rand::seq::SliceRandom; use std::result::Result::Ok; use super::utilurl::get_url; // We are generous on timeout to allow for slow ipfs gateway const DOWNLOAD_TIMEOUT: Duration = Duration::from_secs(60); const IPFS_GATEWAYS: [&str; 2] = ["https://ipfs.io/ipfs/", "https://w3s.link/ipfs/"]; const MAX_BCMR_SIZE: usize = 1024 * 1024 * 100; // 100 MB const MAX_PARALLEL_DOWNLOADS: usize = 5; pub struct BCMRDownloader { db: SqlitePool, keep_running: Arc, download_task: Option>, } const LOOP_SLEEP_TIME: Duration = Duration::from_secs(10); async fn get_download_candidates(db: &SqlitePool) -> Vec { match get_entries_missing_bcmr_download(db).await { Ok(c) => c, Err(e) => { warn!("bcmr: Failed to fetch bcmr download candidates: {e}"); Vec::default() } } } async fn fetch_bcmr( client: &reqwest::Client, urls: &[String], expected_hash: &str, ) -> ( Option<(String, String)>, String, /* error */ bool, /* fatal error (don't try again) */ ) { let mut candidate: Option<(String, String)> = None; let mut errors: Vec = Vec::new(); for url in urls { let url = if let Some(stripped) = url.strip_prefix("ipfs://") { format!( "{}{}", IPFS_GATEWAYS.choose(&mut thread_rng()).unwrap(), stripped ) } else if !url.starts_with("https://") { format!("https://{url}") } else { url.to_string() }; let (contents, actual_hash) = match get_url(client, &url, DOWNLOAD_TIMEOUT, MAX_BCMR_SIZE).await { Ok(c) => c, Err(e) => { errors.push(e.to_string()); continue; } }; if actual_hash == expected_hash { return (Some((contents, actual_hash)), String::default(), false); } else { candidate = Some((contents, actual_hash)) } } // No URL's with matching expected hash if candidate.is_some() { (candidate, String::default(), false) } else if errors.is_empty() { (None, "No URLs to fetch BCMR from".to_string(), true) } else { ( None, format!("Errors fetching BCMR: {}", errors.join("; ")), false, ) } } async fn process_entry(client: &reqwest::Client, db: &SqlitePool, entry: &AuthChainEntry) { info!( "bcmr: Downloading BCMR for token {}, txid {}", entry.token_id.to_hex(), entry.txid.to_hex() ); let Some(raw) = entry.bcmr_data.as_ref() else { warn!( "bcmr: Missing OP_RETURN BCMR for utxo {} (token {}, txid {}); skipping.", entry.utxo.to_hex(), entry.token_id.to_hex(), entry.txid.to_hex() ); return; }; let bcmr = match parse_bcmr_from_opreturn(&Script::from(raw.clone())) { Some(v) => v, None => { warn!( "bcmr: Invalid BCMR OP_RETURN for token {}, txid {}", entry.token_id.to_hex(), entry.txid.to_hex() ); return; } }; let (json_str, error, is_fatal) = fetch_bcmr(client, &bcmr.uris, &bcmr.hash.to_hex()).await; let (json_str, actual_hash) = match json_str { Some(j) => j, None => { if let Err(e) = update_bcmr_failure( db, &entry.utxo, &entry.txid, &entry.token_id, &format!("Failed to fetch BCMR: {error}"), is_fatal, ) .await { warn!("bcmr: Failed to set BCMR error: {e}"); } return; } }; let json: Value = match serde_json::from_str(&json_str) { Ok(b) => b, Err(e) => { if let Err(e2) = update_bcmr_failure( db, &entry.utxo, &entry.txid, &entry.token_id, &format!("BCMR invalid JSON error: {e}"), true, ) .await { warn!("bcmr: Failed to set BCMR error: {e2}"); } return; } }; let bcmr_parsed = match parse_bcmr_json( &json, &entry.token_id, Some(actual_hash), Some(bcmr.hash.to_hex()), SOURCE_ON_CHAIN, ) { Ok(b) => b, Err(err) => { if let Err(e2) = update_bcmr_failure( db, &entry.utxo, &entry.txid, &entry.token_id, &format!("BCMR contents error: {err}"), true, ) .await { warn!("bcmr: Failed to set BCMR error: {e2}"); } return; } }; if let Err(err) = insert_bcmr_data(db, &entry.token_id, &entry.utxo, &bcmr_parsed).await { warn!("bcmr: Failed to insert BCMR data {err}") } } impl BCMRDownloader { pub fn new(db: SqlitePool) -> Self { Self { db, keep_running: Arc::new(AtomicBool::new(true)), download_task: None, } } pub fn start(&mut self) -> Result<()> { let db = self.db.clone(); let keep_running = self.keep_running.clone(); self.download_task = Some(tokio::spawn(async move { let client = reqwest::Client::new(); loop { if !keep_running.load(Ordering::Relaxed) { info!("Exiting bcmr download task"); return; } let mut queue = get_download_candidates(&db).await; // Shuffle so we don't starve any single token if the list is large { let mut rng = thread_rng(); queue.shuffle(&mut rng); } if queue.is_empty() { tokio::time::sleep(LOOP_SLEEP_TIME).await; continue; } info!("bcmr: {} tokens need BCMR download", queue.len()); // Process up to MAX_PARALLEL_DOWNLOADS concurrently for chunk in queue.chunks(MAX_PARALLEL_DOWNLOADS) { let futures: Vec<_> = chunk .iter() .map(|entry| process_entry(&client, &db, entry)) .collect(); futures::future::join_all(futures).await; } tokio::time::sleep(LOOP_SLEEP_TIME).await; } })); Ok(()) } } impl Drop for BCMRDownloader { fn drop(&mut self) { self.keep_running .store(false, std::sync::atomic::Ordering::SeqCst); if let Some(task) = self.download_task.take() { task.abort(); info!("bcmr download task aborted"); } } }