riftenlabs-indexer/src/bcmr/bcmrdownloader.rs

// 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;

const WELL_KNOWN_PATH: &str = "/.well-known/bitcoin-cash-metadata-registry.json";

/// Resolve an on-chain BCMR URI into one or more candidate URLs to try in order.
///
/// Handles three cases:
/// - `ipfs://<cid>` → a single random IPFS gateway URL.
/// - Bare domain (authority only, no path) → BCMR v2 well-known URI first, then
///   the bare root URL as a fallback for bug-compatibility with tokens that
///   served the registry directly at `/`.
/// - Anything with a path → used as-is (with `https://` prepended if missing).
fn resolve_uri_candidates(uri: &str) -> Vec<String> {
    if let Some(stripped) = uri.strip_prefix("ipfs://") {
        return vec![format!(
            "{}{}",
            IPFS_GATEWAYS.choose(&mut thread_rng()).unwrap(),
            stripped
        )];
    }

    let (scheme, rest) = match uri.split_once("://") {
        Some((s, r)) => (s, r),
        None => ("https", uri),
    };

    let (authority, path) = match rest.split_once('/') {
        Some((a, p)) => (a, format!("/{p}")),
        None => (rest, String::new()),
    };

    if path.is_empty() || path == "/" {
        vec![
            format!("{scheme}://{authority}{WELL_KNOWN_PATH}"),
            format!("{scheme}://{authority}/"),
        ]
    } else {
        vec![format!("{scheme}://{authority}{path}")]
    }
}

pub struct BCMRDownloader {
    db: SqlitePool,
    keep_running: Arc<AtomicBool>,
    download_task: Option<JoinHandle<()>>,
}

const LOOP_SLEEP_TIME: Duration = Duration::from_secs(10);

async fn get_download_candidates(db: &SqlitePool) -> Vec<AuthChainEntry> {
    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<String> = Vec::new();

    for url in urls {
        for resolved in resolve_uri_candidates(url) {
            let (contents, actual_hash) =
                match get_url(client, &resolved, 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");
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn bare_domain_resolves_to_well_known_then_root() {
        let candidates = resolve_uri_candidates("chipnet-16.paryonusd.com");
        assert_eq!(
            candidates,
            vec![
                "https://chipnet-16.paryonusd.com/.well-known/bitcoin-cash-metadata-registry.json"
                    .to_string(),
                "https://chipnet-16.paryonusd.com/".to_string(),
            ]
        );
    }

    #[test]
    fn bare_domain_with_scheme_and_trailing_slash() {
        let candidates = resolve_uri_candidates("https://example.com/");
        assert_eq!(
            candidates,
            vec![
                "https://example.com/.well-known/bitcoin-cash-metadata-registry.json".to_string(),
                "https://example.com/".to_string(),
            ]
        );
    }

    #[test]
    fn url_with_path_is_used_as_is() {
        let candidates = resolve_uri_candidates("https://example.com/path/registry.json");
        assert_eq!(
            candidates,
            vec!["https://example.com/path/registry.json".to_string()]
        );
    }

    #[test]
    fn schemeless_url_with_path_gets_https() {
        let candidates = resolve_uri_candidates("example.com/registry.json");
        assert_eq!(
            candidates,
            vec!["https://example.com/registry.json".to_string()]
        );
    }

    #[test]
    fn ipfs_uri_uses_a_gateway() {
        let candidates = resolve_uri_candidates("ipfs://QmAbc123");
        assert_eq!(candidates.len(), 1);
        assert!(candidates[0].ends_with("/QmAbc123"));
        assert!(IPFS_GATEWAYS.iter().any(|gw| candidates[0].starts_with(gw)));
    }

    #[test]
    fn domain_with_port_is_treated_as_bare() {
        let candidates = resolve_uri_candidates("example.com:8443");
        assert_eq!(
            candidates,
            vec![
                "https://example.com:8443/.well-known/bitcoin-cash-metadata-registry.json"
                    .to_string(),
                "https://example.com:8443/".to_string(),
            ]
        );
    }
}