feat: move backfill to bubblegum crate and verfiy to it

Cargo.toml

@@ -1,6 +1,6 @@
 [workspace]
 members = [
-  "backfill",
+  "bubblegum",
   "blockbuster",
   "core",
   "das_api",
@@ -48,7 +48,12 @@ cargo-lock = "9.0.0"
 chrono = "0.4.19"
 clap = "4.2.2"
 das-core = { path = "core" }
+<<<<<<< Updated upstream
 das-backfill = { path = "backfill" }
+=======
+das-metadata-json = { path = "metadata_json" }
+das-bubblegum = { path = "bubblegum" }
+>>>>>>> Stashed changes
 das_api = { path = "das_api" }
 derive_more = { version = "0.99.17" }
 digital_asset_types = { path = "digital_asset_types" }
@@ -78,7 +83,7 @@ lru = "0.12.3"
 metrics = "0.20.1"
 migration = { path = "migration" }
 mime_guess = "2.0.4"
-mpl-bubblegum = "1.2.0"
+mpl-bubblegum = "1.4.0"
 mpl-core = { version = "0.8.0-beta.1", features = ["serde"] }
 mpl-token-metadata = "4.1.1"
 nft_ingester = { path = "nft_ingester" }
@@ -116,7 +121,7 @@ solana-program = "~1.18"
 solana-sdk = "~1.18"
 solana-transaction-status = "~1.18"
 solana-zk-token-sdk = "1.17.16"
-spl-account-compression = "0.3.0"
+spl-account-compression = "0.4.0"
 spl-associated-token-account = ">= 1.1.3, < 3.0"
 spl-concurrent-merkle-tree = "0.2.0"
 spl-noop = "0.2.0"

backfill/Cargo.toml → bubblegum/Cargo.toml

@@ -1,5 +1,5 @@
 [package]
-name = "das-backfill"
+name = "das-bubblegum"
 version = { workspace = true }
 edition = { workspace = true }
 repository = { workspace = true }

backfill/src/gap.rs → bubblegum/src/backfill/gap.rs

@@ -1,5 +1,4 @@
-use super::ErrorKind;
-use crate::Rpc;
+use crate::{error::ErrorKind, Rpc};
 use anyhow::Result;
 use clap::Args;
 use sea_orm::{DatabaseConnection, DbBackend, FromQueryResult, Statement, Value};

bubblegum/src/backfill/mod.rs

@@ -0,0 +1,2 @@
+pub mod gap;
+pub mod worker;

backfill/src/worker/gap.rs → bubblegum/src/backfill/worker/gap.rs

@@ -9,8 +9,7 @@ use tokio::{
     task::JoinHandle,
 };
 
-use crate::gap::TreeGapFill;
-use crate::BubblegumBackfillContext;
+use crate::{backfill::gap::TreeGapFill, BubblegumContext};
 
 #[derive(Parser, Debug, Clone)]
 pub struct GapWorkerArgs {
@@ -26,7 +25,7 @@ pub struct GapWorkerArgs {
 impl GapWorkerArgs {
     pub fn start(
         &self,
-        context: BubblegumBackfillContext,
+        context: BubblegumContext,
         forward: Sender<Signature>,
     ) -> Result<(JoinHandle<()>, Sender<TreeGapFill>)> {
         let (gap_sender, mut gap_receiver) = channel::<TreeGapFill>(self.gap_channel_size);

backfill/src/worker/program_transformer.rs → bubblegum/src/backfill/worker/program_transformer.rs

@@ -6,7 +6,7 @@ use program_transformers::{ProgramTransformer, TransactionInfo};
 use tokio::sync::mpsc::{channel, Sender, UnboundedSender};
 use tokio::task::JoinHandle;
 
-use crate::BubblegumBackfillContext;
+use crate::BubblegumContext;
 
 #[derive(Parser, Debug, Clone)]
 pub struct ProgramTransformerWorkerArgs {
@@ -17,7 +17,7 @@ pub struct ProgramTransformerWorkerArgs {
 impl ProgramTransformerWorkerArgs {
     pub fn start(
         &self,
-        context: BubblegumBackfillContext,
+        context: BubblegumContext,
         forwarder: UnboundedSender<DownloadMetadataInfo>,
     ) -> Result<(JoinHandle<()>, Sender<TransactionInfo>)> {
         let (sender, mut receiver) =
@@ -35,7 +35,7 @@ impl ProgramTransformerWorkerArgs {
                 transactions.push(gap);
             }
 
-            transactions.sort_by(|a, b| b.signature.cmp(&a.signature));
+            transactions.sort_by(|(a, _), (b, _)| a.cmp(b));
 
             for transaction in transactions {
                 if let Err(e) = program_transformer.handle_transaction(&transaction).await {

backfill/src/worker/transaction.rs → bubblegum/src/backfill/worker/transaction.rs

@@ -1,4 +1,4 @@
-use crate::error::ErrorKind;
+use crate::{error::ErrorKind, BubblegumContext};
 use anyhow::Result;
 use clap::Parser;
 use das_core::Rpc;
@@ -143,11 +143,13 @@ pub struct SignatureWorkerArgs {
     pub signature_worker_count: usize,
 }
 
+type TransactionSender = Sender<(Option<i64>, TransactionInfo)>; 
+
 impl SignatureWorkerArgs {
     pub fn start(
         &self,
-        context: crate::BubblegumBackfillContext,
-        forwarder: Sender<TransactionInfo>,
+        context: BubblegumContext,
+        forwarder: TransactionSender
     ) -> Result<(JoinHandle<()>, Sender<Signature>)> {
         let (sig_sender, mut sig_receiver) = channel::<Signature>(self.signature_channel_size);
         let worker_count = self.signature_worker_count;

backfill/src/worker/tree.rs → bubblegum/src/backfill/worker/tree.rs

@@ -1,7 +1,7 @@
 use crate::{
-    gap::{TreeGapFill, TreeGapModel},
+    backfill::gap::{TreeGapFill, TreeGapModel},
     tree::TreeResponse,
-    BubblegumBackfillContext,
+    BubblegumContext,
 };
 use anyhow::Result;
 use clap::Parser;
@@ -32,11 +32,7 @@ pub struct TreeWorkerArgs {
     pub force: bool,
 }
 impl TreeWorkerArgs {
-    pub fn start(
-        &self,
-        context: BubblegumBackfillContext,
-        tree: TreeResponse,
-    ) -> JoinHandle<Result<()>> {
+    pub fn start(&self, context: BubblegumContext, tree: TreeResponse) -> JoinHandle<Result<()>> {
         let db_pool = context.database_pool.clone();
         let metadata_json_download_db_pool = context.database_pool.clone();
 

backfill/src/lib.rs → bubblegum/src/lib.rs

@@ -1,24 +1,22 @@
+mod backfill;
 mod error;
-mod gap;
 mod tree;
-mod worker;
-
-pub use error::ErrorKind;
+mod verify;
 
 use anyhow::Result;
+use backfill::worker::TreeWorkerArgs;
 use clap::Parser;
 use das_core::Rpc;
 use futures::{stream::FuturesUnordered, StreamExt};
 use tree::TreeResponse;
-use worker::TreeWorkerArgs;
 
 #[derive(Clone)]
-pub struct BubblegumBackfillContext {
+pub struct BubblegumContext {
     pub database_pool: sqlx::PgPool,
     pub solana_rpc: Rpc,
 }
 
-impl BubblegumBackfillContext {
+impl BubblegumContext {
     pub const fn new(database_pool: sqlx::PgPool, solana_rpc: Rpc) -> Self {
         Self {
             database_pool,
@@ -28,7 +26,7 @@ impl BubblegumBackfillContext {
 }
 
 #[derive(Debug, Parser, Clone)]
-pub struct BubblegumBackfillArgs {
+pub struct BackfillArgs {
     /// Number of tree crawler workers
     #[arg(long, env, default_value = "20")]
     pub tree_crawler_count: usize,
@@ -41,10 +39,14 @@ pub struct BubblegumBackfillArgs {
     pub tree_worker: TreeWorkerArgs,
 }
 
-pub async fn start_bubblegum_backfill(
-    context: BubblegumBackfillContext,
-    args: BubblegumBackfillArgs,
-) -> Result<()> {
+#[derive(Debug, Parser, Clone)]
+pub struct VerifyArgs {
+    /// The list of trees to verify. If not specified, all trees will be crawled.
+    #[arg(long, env, use_value_delimiter = true)]
+    pub only_trees: Option<Vec<String>>,
+}
+
+pub async fn start_backfill(context: BubblegumContext, args: BackfillArgs) -> Result<()> {
     let trees = if let Some(ref only_trees) = args.only_trees {
         TreeResponse::find(&context.solana_rpc, only_trees.clone()).await?
     } else {
@@ -67,3 +69,17 @@ pub async fn start_bubblegum_backfill(
 
     Ok(())
 }
+
+pub async fn verify_bubblegum(context: BubblegumContext, args: VerifyArgs) -> Result<()> {
+    let trees = if let Some(ref only_trees) = args.only_trees {
+        TreeResponse::find(&context.solana_rpc, only_trees.clone()).await?
+    } else {
+        TreeResponse::all(&context.solana_rpc).await?
+    };
+
+    for tree in trees {
+        verify::check(context.clone(), tree).await?;
+    }
+
+    Ok(())
+}

backfill/src/tree.rs → bubblegum/src/tree.rs

@@ -1,4 +1,4 @@
-use super::ErrorKind;
+use super::error::ErrorKind;
 use anyhow::Result;
 use borsh::BorshDeserialize;
 use das_core::Rpc;
@@ -10,7 +10,7 @@ use spl_account_compression::state::{
 };
 use std::str::FromStr;
 
-#[derive(Debug, Clone)]
+#[derive(Clone)]
 pub struct TreeHeaderResponse {
     pub max_depth: u32,
     pub max_buffer_size: u32,
@@ -33,10 +33,11 @@ impl TryFrom<ConcurrentMerkleTreeHeader> for TreeHeaderResponse {
     }
 }
 
-#[derive(Debug, Clone)]
 pub struct TreeResponse {
     pub pubkey: Pubkey,
     pub tree_header: TreeHeaderResponse,
+    pub concurrent_merkle_tree_header: ConcurrentMerkleTreeHeader,
+    pub bytes: Vec<u8>,
     pub seq: u64,
 }
 
@@ -47,6 +48,8 @@ impl TreeResponse {
         let (header_bytes, rest) = bytes.split_at(CONCURRENT_MERKLE_TREE_HEADER_SIZE_V1);
         let header: ConcurrentMerkleTreeHeader =
             ConcurrentMerkleTreeHeader::try_from_slice(header_bytes)?;
+        let concurrent_merkle_tree_header: ConcurrentMerkleTreeHeader =
+            ConcurrentMerkleTreeHeader::try_from_slice(header_bytes)?;
 
         let merkle_tree_size = merkle_tree_get_size(&header)?;
         let (tree_bytes, _canopy_bytes) = rest.split_at(merkle_tree_size);
@@ -63,6 +66,8 @@ impl TreeResponse {
         Ok(Self {
             pubkey,
             tree_header,
+            concurrent_merkle_tree_header,
+            bytes: tree_bytes.to_vec(),
             seq,
         })
     }

bubblegum/src/verify.rs

@@ -0,0 +1,86 @@
+use super::BubblegumContext;
+use crate::error::ErrorKind;
+use crate::tree::TreeResponse;
+use anyhow::{anyhow, Result};
+use digital_asset_types::dapi::get_proof_for_asset;
+use digital_asset_types::rpc::AssetProof;
+use mpl_bubblegum::accounts::TreeConfig;
+use sea_orm::SqlxPostgresConnector;
+use solana_sdk::pubkey::Pubkey;
+use spl_account_compression::concurrent_tree_wrapper::{merkle_tree_prove_leaf, ProveLeafArgs};
+
+trait TryFromAssetProof {
+    fn try_from_asset_proof(proof: AssetProof) -> Result<Self, anyhow::Error>
+    where
+        Self: Sized;
+}
+
+impl TryFromAssetProof for ProveLeafArgs {
+    fn try_from_asset_proof(proof: AssetProof) -> Result<Self, anyhow::Error> {
+        Ok(ProveLeafArgs {
+            current_root: bs58::decode(&proof.root)
+                .into_vec()
+                .map_err(|e| anyhow!(e))?
+                .try_into()
+                .map_err(|_| anyhow!("Invalid root length"))?,
+            leaf: bs58::decode(&proof.leaf)
+                .into_vec()
+                .map_err(|e| anyhow!(e))?
+                .try_into()
+                .map_err(|_| anyhow!("Invalid leaf length"))?,
+            proof_vec: proof
+                .proof
+                .iter()
+                .map(|p| {
+                    bs58::decode(p)
+                        .into_vec()
+                        .map_err(|e| anyhow!(e))
+                        .and_then(|v| v.try_into().map_err(|_| anyhow!("Invalid proof length")))
+                })
+                .collect::<Result<Vec<[u8; 32]>>>()?,
+            index: proof.node_index as u32,
+        })
+    }
+}
+pub async fn check(context: BubblegumContext, tree: TreeResponse) -> Result<()> {
+    let TreeResponse {
+        pubkey,
+        concurrent_merkle_tree_header,
+        bytes,
+        ..
+    } = tree;
+
+    let db = SqlxPostgresConnector::from_sqlx_postgres_pool(context.database_pool);
+
+    let (tree_config_pubkey, _) = TreeConfig::find_pda(&tree.pubkey);
+
+    let account = context.solana_rpc.get_account(&tree_config_pubkey).await?;
+    let account = account
+        .value
+        .ok_or_else(|| ErrorKind::Generic("Account not found".to_string()))?;
+
+    let tree_config = TreeConfig::from_bytes(account.data.as_slice())?;
+
+    for i in 0..tree_config.num_minted {
+        let (asset, _) = Pubkey::find_program_address(
+            &[b"asset", &pubkey.to_bytes(), &i.to_le_bytes()],
+            &mpl_bubblegum::ID,
+        );
+        let proof = get_proof_for_asset(&db, asset.to_bytes().to_vec()).await?;
+
+        let prove_leaf_args = ProveLeafArgs::try_from_asset_proof(proof)?;
+
+        let result = merkle_tree_prove_leaf(
+            &concurrent_merkle_tree_header,
+            pubkey,
+            &bytes,
+            &prove_leaf_args,
+        );
+
+        if let Err(e) = &result {
+            log::error!("Proof not valid for asset pubkey {}: {:?}", asset, e);
+        }
+    }
+
+    Ok(())
+}

ops/Cargo.toml

@@ -16,7 +16,7 @@ bs58 = { workspace = true }
 cadence = { workspace = true }
 cadence-macros = { workspace = true }
 clap = { workspace = true, features = ["derive", "cargo", "env"] }
-das-backfill = { workspace = true }
+das-bubblegum = { workspace = true }
 das-core = { workspace = true }
 digital_asset_types = { workspace = true }
 env_logger = { workspace = true }

ops/src/bubblegum/backfiller.rs

@@ -1,13 +1,13 @@
 use anyhow::Result;
 use clap::Parser;
-use das_backfill::{start_bubblegum_backfill, BubblegumBackfillArgs, BubblegumBackfillContext};
+use das_bubblegum::{start_backfill, BackfillArgs, BubblegumContext};
 use das_core::{connect_db, PoolArgs, Rpc, SolanaRpcArgs};
 
 #[derive(Debug, Parser, Clone)]
 pub struct Args {
     /// Backfill Bubblegum Args
     #[clap(flatten)]
-    pub backfill_bubblegum: BubblegumBackfillArgs,
+    pub backfill_bubblegum: BackfillArgs,
 
     /// Database configuration
     #[clap(flatten)]
@@ -47,7 +47,7 @@ pub async fn run(config: Args) -> Result<()> {
     let database_pool = connect_db(&config.database).await?;
 
     let solana_rpc = Rpc::from_config(&config.solana);
-    let context = BubblegumBackfillContext::new(database_pool, solana_rpc);
+    let context = BubblegumContext::new(database_pool, solana_rpc);
 
-    start_bubblegum_backfill(context, config.backfill_bubblegum).await
+    start_backfill(context, config.backfill_bubblegum).await
 }