riftenlabs-indexer/src/db/bcmr/mod.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 crate::bcmr::parsedbcmr::{FileMeta, ParsedBCMR, Token, Uris, SOURCE_ON_CHAIN};
use crate::db::blob::{display_hex_to_blob, FromBlob, ToBlob};
use anyhow::*;

use bitcoincash::{BlockHash, TokenID, Txid};
use log::info;
use riftenlabs_defi::chainutil::OutPointHash;
use sqlx::{Row, SqlitePool};

const MAX_DOWNLOAD_ATTEMPTS: usize = 100;

pub struct AuthChainEntry {
    pub utxo: OutPointHash,
    pub txid: Txid,
    pub token_id: TokenID,
    pub height: usize,
    pub bcmr_data: Option<Vec<u8>>,
}

pub async fn prepare_tables(pool: &SqlitePool) {
    sqlx::query(
        "CREATE TABLE auth_chain_entry (
            token_id  BLOB NOT NULL,
            utxo      BLOB NOT NULL,
            blockhash BLOB NOT NULL,
            txid      BLOB NOT NULL,
            height    INT NOT NULL,
            bcmr_data TEXT,
            PRIMARY KEY (token_id, utxo)
        )",
    )
    .execute(pool)
    .await
    .expect("failed to create auth_chain_entry table");

    sqlx::query(
        "CREATE TABLE bcmr_data (
            token_id     BLOB NOT NULL,
            utxo         BLOB NOT NULL,
            symbol       TEXT NOT NULL,
            decimals     INT NOT NULL,
            name         TEXT NOT NULL,
            description  TEXT NOT NULL,
            icon         TEXT NOT NULL,
            web          TEXT NOT NULL,
            expected_hash TEXT NOT NULL,
            actual_hash   TEXT NOT NULL,
            PRIMARY KEY (token_id, utxo),
            FOREIGN KEY (token_id, utxo)
                REFERENCES auth_chain_entry(token_id, utxo)
                ON DELETE CASCADE
        )",
    )
    .execute(pool)
    .await
    .expect("failed to create bcmr_data table");

    sqlx::query(
        "CREATE TABLE bcmr_failure (
            token_id     BLOB NOT NULL,
            utxo         BLOB NOT NULL,
            txid         BLOB NOT NULL,
            last_attempt INT NOT NULL,
            attempts     INT NOT NULL,
            error_message TEXT,
            give_up      BOOLEAN,
            PRIMARY KEY (token_id, utxo, txid)
        )",
    )
    .execute(pool)
    .await
    .expect("failed to create bcmr_failure table");

    sqlx::query("CREATE INDEX IF NOT EXISTS idx_bcmr_failure_txid ON bcmr_failure(txid)")
        .execute(pool)
        .await
        .unwrap();

    sqlx::query(
        "CREATE TABLE bcmr_well_known (
            source TEXT NOT NULL,
            symbol TEXT NOT NULL,
            token_id BLOB NOT NULL,
            decimals INT NOT NULL,
            name TEXT NOT NULL,
            description TEXT NOT NULL,
            icon TEXT NOT NULL,
            web TEXT NOT NULL,
            PRIMARY KEY (source, token_id)
        )",
    )
    .execute(pool)
    .await
    .unwrap();

    sqlx::query("CREATE INDEX idx_auth_utxo ON auth_chain_entry(utxo);")
        .execute(pool)
        .await
        .expect("failed to create index");
    sqlx::query("CREATE INDEX IF NOT EXISTS idx_auth_chain_token_bcmr_height ON auth_chain_entry(token_id, height DESC) WHERE bcmr_data IS NOT NULL")
        .execute(pool).await.expect("failed to create index");
    sqlx::query("CREATE INDEX IF NOT EXISTS idx_auth_token_height ON auth_chain_entry(token_id, height DESC)")
        .execute(pool).await.expect("failed to create index");
    sqlx::query("CREATE INDEX IF NOT EXISTS idx_bcmr_data_utxo ON bcmr_data(utxo)")
        .execute(pool)
        .await
        .expect("failed to create index");
    sqlx::query(
        "CREATE INDEX IF NOT EXISTS idx_bcmr_failure_last_attempt ON bcmr_failure(last_attempt)",
    )
    .execute(pool)
    .await
    .expect("failed to create index");
}

/// Delete all well-known BCMR entries for the given source.
/// Accepts both `&SqlitePool` and `&mut SqliteConnection` (including transactions).
pub async fn delete_entries_for_well_known<'e>(
    executor: impl sqlx::Executor<'e, Database = sqlx::Sqlite>,
    source: &str,
) -> Result<()> {
    sqlx::query("DELETE FROM bcmr_well_known WHERE source = ?")
        .bind(source)
        .execute(executor)
        .await?;
    Ok(())
}

pub async fn delete_entries_for_block(pool: &SqlitePool, blockhash: &BlockHash) -> Result<bool> {
    let r = sqlx::query("DELETE FROM auth_chain_entry WHERE blockhash = ?")
        .bind(blockhash.to_blob())
        .execute(pool)
        .await?;
    Ok(r.rows_affected() != 0)
}

/// Add or update config entry
pub async fn insert_authheader(
    pool: &SqlitePool,
    utxo: &OutPointHash,
    blockhash: &BlockHash,
    txid: &Txid,
    token_id: &TokenID,
    height: usize,
    bcmr_data: Option<Vec<u8>>,
) -> Result<()> {
    sqlx::query(
        "INSERT OR REPLACE INTO auth_chain_entry
         (utxo, blockhash, txid, token_id, height, bcmr_data)
         VALUES (?, ?, ?, ?, ?, ?)",
    )
    .bind(utxo.to_blob())
    .bind(blockhash.to_blob())
    .bind(txid.to_blob())
    .bind(token_id.to_blob())
    .bind(height as i64)
    .bind(bcmr_data.map(|bcmr| hex::encode(&bcmr)))
    .execute(pool)
    .await?;
    Ok(())
}

/// Get config entry
#[allow(dead_code)]
pub async fn get_authheader(
    pool: &SqlitePool,
    utxo: &OutPointHash,
) -> Result<Option<AuthChainEntry>> {
    let row = sqlx::query(
        "SELECT token_id, txid, height, bcmr_data FROM auth_chain_entry WHERE utxo = ?",
    )
    .bind(utxo.to_blob())
    .fetch_optional(pool)
    .await?;

    if let Some(r) = row {
        let token_blob: Vec<u8> = r.get(0);
        let txid_blob: Vec<u8> = r.get(1);
        let height: i64 = r.get(2);
        let bcmr_data_hex: Option<String> = r.get(3);
        let bcmr_data = if let Some(data) = bcmr_data_hex {
            Some(hex::decode(data).context("failed to decode bcmr data")?)
        } else {
            None
        };

        Ok(Some(AuthChainEntry {
            utxo: *utxo,
            token_id: TokenID::from_blob(&token_blob).context("failed to decode token blob")?,
            txid: Txid::from_blob(&txid_blob).context("failed to decode txid blob")?,
            height: height as usize,
            bcmr_data,
        }))
    } else {
        Ok(None)
    }
}

pub async fn get_entries_missing_bcmr_download(pool: &SqlitePool) -> Result<Vec<AuthChainEntry>> {
    let sql = r#"
    WITH ranked AS (
        SELECT
            ace.utxo, ace.token_id, ace.txid, ace.height, ace.bcmr_data,
            ROW_NUMBER() OVER (PARTITION BY ace.token_id ORDER BY ace.height DESC) AS rn
        FROM auth_chain_entry ace
        WHERE ace.bcmr_data IS NOT NULL
    )
    SELECT r.token_id, r.txid, r.height, r.bcmr_data, r.utxo
    FROM ranked r
    LEFT JOIN bcmr_data bd ON bd.utxo = r.utxo AND bd.token_id = r.token_id
    LEFT JOIN bcmr_failure bf ON bf.utxo = r.utxo AND bf.txid = r.txid AND bf.token_id = r.token_id
          AND (bf.give_up = 1 OR (strftime('%s','now') - bf.last_attempt) < 1800)
    WHERE r.rn = 1 AND bd.utxo IS NULL AND bf.utxo IS NULL
    ORDER BY r.height DESC
    "#;

    let rows = sqlx::query(sql).fetch_all(pool).await?;
    let mut matches: Vec<AuthChainEntry> = Vec::new();

    for r in rows {
        let token_blob: Vec<u8> = r.get(0);
        let txid_blob: Vec<u8> = r.get(1);
        let height: i64 = r.get(2);
        let bcmr_data_hex: Option<String> = r.get(3);
        let bcmr_data = if let Some(data) = bcmr_data_hex {
            Some(hex::decode(&data).context("failed to decode bcmr data")?)
        } else {
            None
        };
        let utxo_blob: Vec<u8> = r.get(4);
        let utxo = OutPointHash::from_blob(&utxo_blob)?;

        matches.push(AuthChainEntry {
            utxo,
            token_id: TokenID::from_blob(&token_blob).context("failed to decode token blob")?,
            txid: Txid::from_blob(&txid_blob).context("failed to decode txid blob")?,
            height: height as usize,
            bcmr_data,
        });
    }

    if !matches.is_empty() {
        log::info!(
            "bcmr: {} head(s) need download; first few: {:?}",
            matches.len(),
            matches
                .iter()
                .take(3)
                .map(|e| (e.token_id.to_string(), e.height, e.utxo.to_string()))
                .collect::<Vec<_>>()
        );
    }
    Ok(matches)
}

pub async fn insert_bcmr_data(
    pool: &SqlitePool,
    token_id: &TokenID,
    utxo: &OutPointHash,
    bcmr: &ParsedBCMR,
) -> Result<()> {
    let sql = "INSERT OR REPLACE INTO bcmr_data (
                   token_id, utxo, symbol, decimals, name, description, icon, web,
                   expected_hash, actual_hash
               ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";

    info!("Trying to insert {}", utxo);
    let empty_string = "".to_owned();

    sqlx::query(sql)
        .bind(token_id.to_blob())
        .bind(utxo.to_blob())
        .bind(&bcmr.token.symbol)
        .bind(bcmr.token.decimals as i64)
        .bind(&bcmr.name)
        .bind(&bcmr.description)
        .bind(bcmr.uris.icon.as_ref().unwrap_or(&empty_string))
        .bind(bcmr.uris.web.as_ref().unwrap_or(&empty_string))
        .bind(bcmr.filemeta.expected_hash.as_deref().unwrap_or(""))
        .bind(bcmr.filemeta.actual_hash.as_deref().unwrap_or(""))
        .execute(pool)
        .await?;

    Ok(())
}

/// Insert or replace a well-known BCMR entry.
/// Accepts both `&SqlitePool` and `&mut SqliteConnection` (including transactions).
pub async fn insert_well_known_bcmr<'e>(
    executor: impl sqlx::Executor<'e, Database = sqlx::Sqlite>,
    source: &str,
    bcmr: &ParsedBCMR,
) -> Result<()> {
    let sql = "INSERT OR REPLACE INTO bcmr_well_known (source, symbol, token_id, decimals, name, description, icon, web)
               VALUES (?, ?, ?, ?, ?, ?, ?, ?)";

    let empty_string = "".to_owned();
    let token_blob = display_hex_to_blob::<TokenID>(&bcmr.token.category)?;

    sqlx::query(sql)
        .bind(source)
        .bind(&bcmr.token.symbol)
        .bind(token_blob)
        .bind(bcmr.token.decimals as i64)
        .bind(&bcmr.name)
        .bind(&bcmr.description)
        .bind(bcmr.uris.icon.as_ref().unwrap_or(&empty_string))
        .bind(bcmr.uris.web.as_ref().unwrap_or(&empty_string))
        .execute(executor)
        .await?;

    Ok(())
}

pub async fn update_bcmr_failure(
    pool: &SqlitePool,
    utxo: &OutPointHash,
    txid: &Txid,
    token_id: &TokenID,
    error_message: &str,
    give_up: bool,
) -> Result<()> {
    let sql = format!(
        "INSERT INTO bcmr_failure (token_id, utxo, txid, last_attempt, attempts, error_message, give_up)
        VALUES (?1, ?2, ?3, strftime('%s','now'), 1, ?4, ?5)
        ON CONFLICT(token_id, utxo, txid) DO UPDATE SET
            last_attempt  = excluded.last_attempt,
            attempts      = bcmr_failure.attempts + 1,
            error_message = excluded.error_message,
            give_up       = CASE
                WHEN bcmr_failure.attempts + 1 > {max} THEN 1
                ELSE excluded.give_up
            END",
        max = MAX_DOWNLOAD_ATTEMPTS,
    );
    sqlx::query(&sql)
        .bind(token_id.to_blob())
        .bind(utxo.to_blob())
        .bind(txid.to_blob())
        .bind(error_message)
        .bind(give_up)
        .execute(pool)
        .await?;
    Ok(())
}

/// Return the current authhead of a token's auth chain: the txid of the latest (highest
/// block height) auth_chain_entry. The authhead UTXO is always output 0 of that tx
/// (see `compute_outpoint_hash(&txid, 0)` in the indexer), so the outpoint is `<txid>:0`.
/// Returns None if the token has no auth chain entry.
pub async fn get_current_authhead(
    pool: &SqlitePool,
    token_hex: &str,
) -> Result<Option<(Txid, usize)>> {
    // txid is a deterministic tiebreaker: a forked/branched chain can have two entries at the
    // same max height for one token, and without it SQLite's LIMIT 1 could flip between them.
    let sql = "SELECT txid, height FROM auth_chain_entry
               WHERE token_id = ? ORDER BY height DESC, txid DESC LIMIT 1";
    let token_blob = display_hex_to_blob::<TokenID>(token_hex)?;

    let row = sqlx::query(sql)
        .bind(token_blob)
        .fetch_optional(pool)
        .await?;

    if let Some(r) = row {
        let txid_blob: Vec<u8> = r.get(0);
        let height: i64 = r.get(1);
        let txid = Txid::from_blob(&txid_blob).context("failed to decode txid blob")?;
        Ok(Some((txid, height as usize)))
    } else {
        Ok(None)
    }
}

pub async fn get_token_bcmr(pool: &SqlitePool, token_hex: &str) -> Result<Option<ParsedBCMR>> {
    let sql = r#"
    SELECT symbol, decimals, name, description, icon, web, actual_hash, expected_hash
    FROM bcmr_data
    WHERE token_id = ?
    ORDER BY ROWID DESC
    LIMIT 1;
    "#;
    let token_blob = display_hex_to_blob::<TokenID>(token_hex)?;

    let row = sqlx::query(sql)
        .bind(token_blob)
        .fetch_optional(pool)
        .await?;

    if let Some(r) = row {
        let symbol: String = r.get(0);
        let decimals: i64 = r.get(1);
        let name: String = r.get(2);
        let description: String = r.get(3);
        let icon: String = r.get(4);
        let web: String = r.get(5);
        let actual_hash: String = r.get(6);
        let expected_hash: String = r.get(7);

        Ok(Some(ParsedBCMR {
            name,
            description,
            token: Token {
                category: token_hex.to_owned(),
                symbol,
                decimals: decimals as usize,
            },
            uris: Uris {
                web: if web.is_empty() { None } else { Some(web) },
                icon: if icon.is_empty() { None } else { Some(icon) },
            },
            filemeta: FileMeta {
                expected_hash: Some(expected_hash),
                actual_hash: Some(actual_hash),
                source: SOURCE_ON_CHAIN.into(),
            },
        }))
    } else {
        Ok(None)
    }
}

pub async fn get_well_known_bcmr(pool: &SqlitePool, token_hex: &str) -> Result<Vec<ParsedBCMR>> {
    let sql = "SELECT source, symbol, decimals, name, description, icon, web
            FROM bcmr_well_known WHERE token_id = ?";
    let token_blob = display_hex_to_blob::<TokenID>(token_hex)?;

    let rows = sqlx::query(sql).bind(token_blob).fetch_all(pool).await?;

    let mut entries: Vec<ParsedBCMR> = Vec::default();

    for r in rows {
        let source: String = r.get(0);
        let symbol: String = r.get(1);
        let decimals: i64 = r.get(2);
        let name: String = r.get(3);
        let description: String = r.get(4);
        let icon: String = r.get(5);
        let web: String = r.get(6);

        entries.push(ParsedBCMR {
            name,
            description,
            token: Token {
                category: token_hex.to_owned(),
                symbol,
                decimals: decimals as usize,
            },
            uris: Uris {
                web: if web.is_empty() { None } else { Some(web) },
                icon: if icon.is_empty() { None } else { Some(icon) },
            },
            filemeta: FileMeta {
                expected_hash: None,
                actual_hash: None,
                source,
            },
        })
    }

    Ok(entries)
}

/// Given a list of token blobs, returns which ones have BCMR data in bcmr_data table.
/// Uses batched IN queries to handle large lists efficiently.
pub async fn filter_tokens_with_bcmr(
    pool: &SqlitePool,
    token_blobs: &[Vec<u8>],
) -> Result<Vec<Vec<u8>>> {
    if token_blobs.is_empty() {
        return Ok(Vec::new());
    }

    let mut result = Vec::new();
    const BATCH_SIZE: usize = 500;

    for chunk in token_blobs.chunks(BATCH_SIZE) {
        let placeholders: Vec<&str> = chunk.iter().map(|_| "?").collect();
        let sql = format!(
            "SELECT DISTINCT token_id FROM bcmr_data WHERE token_id IN ({})",
            placeholders.join(", ")
        );

        let mut query = sqlx::query(&sql);
        for blob in chunk {
            query = query.bind(blob);
        }

        let rows = query.fetch_all(pool).await?;
        for row in rows {
            let blob: Vec<u8> = row.get(0);
            result.push(blob);
        }
    }

    Ok(result)
}

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

    #[tokio::test]
    async fn test_update_bcmr_failure_multiple_tokens_same_utxo_txid() {
        let pool = SqlitePool::connect("sqlite::memory:").await.unwrap();
        prepare_tables(&pool).await;

        let token_a = "0101010101010101010101010101010101010101010101010101010101010101"
            .parse::<TokenID>()
            .unwrap();
        let token_b = "0202020202020202020202020202020202020202020202020202020202020202"
            .parse::<TokenID>()
            .unwrap();

        let utxo = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
            .parse::<OutPointHash>()
            .unwrap();
        let txid = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
            .parse::<Txid>()
            .unwrap();

        update_bcmr_failure(&pool, &utxo, &txid, &token_a, "error A", false)
            .await
            .expect("first update_bcmr_failure should succeed");

        update_bcmr_failure(&pool, &utxo, &txid, &token_b, "error B", false)
            .await
            .expect("second update_bcmr_failure should succeed");

        let count: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM bcmr_failure")
            .fetch_one(&pool)
            .await
            .unwrap();
        assert_eq!(count.0, 2, "should have 2 failure entries");

        update_bcmr_failure(&pool, &utxo, &txid, &token_a, "error A updated", false)
            .await
            .expect("updating token A failure should succeed");
        update_bcmr_failure(&pool, &utxo, &txid, &token_b, "error B updated", false)
            .await
            .expect("updating token B failure should succeed");

        let count: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM bcmr_failure")
            .fetch_one(&pool)
            .await
            .unwrap();
        assert_eq!(count.0, 2, "should still have 2 failure entries");
    }
}