add crate program_transformers

Cargo.toml

@@ -6,6 +6,7 @@ members = [
   "metaplex-rpc-proxy",
   "migration",
   "nft_ingester",
+  "program_transformers",
   "tools/acc_forwarder",
   "tools/bgtask_creator",
   "tools/fetch_trees",

program_transformers/Cargo.toml

@@ -0,0 +1,27 @@
+[package]
+name = "program_transformers"
+version = { workspace = true }
+edition = { workspace = true }
+repository = { workspace = true }
+publish = { workspace = true }
+
+[dependencies]
+blockbuster = { workspace = true }
+bs58 = { workspace = true }
+digital_asset_types = { workspace = true, features = ["json_types", "sql_types"] }
+futures = { workspace = true }
+mpl-bubblegum = { workspace = true }
+num-traits = { workspace = true }
+plerkle_serialization = { workspace = true }
+sea-orm = { workspace = true, features = [] }
+serde_json = { workspace = true }
+solana-sdk = { workspace = true }
+spl-account-compression = { workspace = true, features = ["no-entrypoint"] }
+spl-token = { workspace = true, features = ["no-entrypoint"] }
+sqlx = { workspace = true, features = [] }
+thiserror = { workspace = true }
+tokio = { workspace = true, features = ["time"] }
+tracing = { workspace = true }
+
+[lints]
+workspace = true

program_transformers/src/bubblegum/burn.rs

@@ -0,0 +1,77 @@
+use {
+    crate::{
+        bubblegum::{
+            db::{save_changelog_event, upsert_asset_with_seq},
+            u32_to_u8_array,
+        },
+        error::{ProgramTransformerError, ProgramTransformerResult},
+    },
+    blockbuster::{instruction::InstructionBundle, programs::bubblegum::BubblegumInstruction},
+    digital_asset_types::dao::asset,
+    sea_orm::{
+        entity::{ActiveValue, EntityTrait},
+        query::QueryTrait,
+        sea_query::query::OnConflict,
+        ConnectionTrait, DbBackend, TransactionTrait,
+    },
+    solana_sdk::pubkey::Pubkey,
+    tracing::debug,
+};
+
+pub async fn burn<'c, T>(
+    parsing_result: &BubblegumInstruction,
+    bundle: &InstructionBundle<'c>,
+    txn: &'c T,
+    instruction: &str,
+    cl_audits: bool,
+) -> ProgramTransformerResult<()>
+where
+    T: ConnectionTrait + TransactionTrait,
+{
+    if let Some(cl) = &parsing_result.tree_update {
+        let seq = save_changelog_event(cl, bundle.slot, bundle.txn_id, txn, instruction, cl_audits)
+            .await?;
+        let leaf_index = cl.index;
+        let (asset_id, _) = Pubkey::find_program_address(
+            &[
+                "asset".as_bytes(),
+                cl.id.as_ref(),
+                u32_to_u8_array(leaf_index).as_ref(),
+            ],
+            &mpl_bubblegum::ID,
+        );
+        debug!("Indexing burn for asset id: {:?}", asset_id);
+        let id_bytes = asset_id.to_bytes();
+
+        let asset_model = asset::ActiveModel {
+            id: ActiveValue::Set(id_bytes.to_vec()),
+            burnt: ActiveValue::Set(true),
+            ..Default::default()
+        };
+
+        // Begin a transaction.  If the transaction goes out of scope (i.e. one of the executions has
+        // an error and this function returns it using the `?` operator), then the transaction is
+        // automatically rolled back.
+        let multi_txn = txn.begin().await?;
+
+        // Upsert asset table `burnt` column.  Note we don't check for decompression (asset.seq = 0)
+        // because we know if the item was burnt it could not have been decompressed later.
+        let query = asset::Entity::insert(asset_model)
+            .on_conflict(
+                OnConflict::columns([asset::Column::Id])
+                    .update_columns([asset::Column::Burnt])
+                    .to_owned(),
+            )
+            .build(DbBackend::Postgres);
+        multi_txn.execute(query).await?;
+
+        upsert_asset_with_seq(&multi_txn, id_bytes.to_vec(), seq as i64).await?;
+
+        multi_txn.commit().await?;
+
+        return Ok(());
+    }
+    Err(ProgramTransformerError::ParsingError(
+        "Ix not parsed correctly".to_string(),
+    ))
+}

program_transformers/src/bubblegum/cancel_redeem.rs

@@ -0,0 +1,85 @@
+use {
+    crate::{
+        bubblegum::db::{
+            save_changelog_event, upsert_asset_with_leaf_info,
+            upsert_asset_with_owner_and_delegate_info, upsert_asset_with_seq,
+        },
+        error::{ProgramTransformerError, ProgramTransformerResult},
+    },
+    blockbuster::{
+        instruction::InstructionBundle,
+        programs::bubblegum::{BubblegumInstruction, LeafSchema},
+    },
+    sea_orm::{ConnectionTrait, TransactionTrait},
+};
+
+pub async fn cancel_redeem<'c, T>(
+    parsing_result: &BubblegumInstruction,
+    bundle: &InstructionBundle<'c>,
+    txn: &'c T,
+    instruction: &str,
+    cl_audits: bool,
+) -> ProgramTransformerResult<()>
+where
+    T: ConnectionTrait + TransactionTrait,
+{
+    if let (Some(le), Some(cl)) = (&parsing_result.leaf_update, &parsing_result.tree_update) {
+        let seq = save_changelog_event(cl, bundle.slot, bundle.txn_id, txn, instruction, cl_audits)
+            .await?;
+        match le.schema {
+            LeafSchema::V1 {
+                id,
+                owner,
+                delegate,
+                ..
+            } => {
+                let id_bytes = id.to_bytes();
+                let owner_bytes = owner.to_bytes().to_vec();
+                let delegate = if owner == delegate || delegate.to_bytes() == [0; 32] {
+                    None
+                } else {
+                    Some(delegate.to_bytes().to_vec())
+                };
+                let tree_id = cl.id.to_bytes();
+                let nonce = cl.index as i64;
+
+                // Begin a transaction.  If the transaction goes out of scope (i.e. one of the executions has
+                // an error and this function returns it using the `?` operator), then the transaction is
+                // automatically rolled back.
+                let multi_txn = txn.begin().await?;
+
+                // Partial update of asset table with just leaf.
+                upsert_asset_with_leaf_info(
+                    &multi_txn,
+                    id_bytes.to_vec(),
+                    nonce,
+                    tree_id.to_vec(),
+                    le.leaf_hash.to_vec(),
+                    le.schema.data_hash(),
+                    le.schema.creator_hash(),
+                    seq as i64,
+                )
+                .await?;
+
+                // Partial update of asset table with just leaf owner and delegate.
+                upsert_asset_with_owner_and_delegate_info(
+                    &multi_txn,
+                    id_bytes.to_vec(),
+                    owner_bytes,
+                    delegate,
+                    seq as i64,
+                )
+                .await?;
+
+                upsert_asset_with_seq(&multi_txn, id_bytes.to_vec(), seq as i64).await?;
+
+                multi_txn.commit().await?;
+
+                return Ok(());
+            }
+        }
+    }
+    Err(ProgramTransformerError::ParsingError(
+        "Ix not parsed correctly".to_string(),
+    ))
+}

program_transformers/src/bubblegum/collection_verification.rs

@@ -0,0 +1,95 @@
+use {
+    crate::{
+        bubblegum::db::{
+            save_changelog_event, upsert_asset_with_leaf_info, upsert_asset_with_seq,
+            upsert_collection_info,
+        },
+        error::{ProgramTransformerError, ProgramTransformerResult},
+    },
+    blockbuster::{
+        instruction::InstructionBundle,
+        programs::bubblegum::{BubblegumInstruction, LeafSchema, Payload},
+    },
+    mpl_bubblegum::types::Collection,
+    sea_orm::{ConnectionTrait, TransactionTrait},
+    tracing::debug,
+};
+
+pub async fn process<'c, T>(
+    parsing_result: &BubblegumInstruction,
+    bundle: &InstructionBundle<'c>,
+    txn: &'c T,
+    instruction: &str,
+    cl_audits: bool,
+) -> ProgramTransformerResult<()>
+where
+    T: ConnectionTrait + TransactionTrait,
+{
+    if let (Some(le), Some(cl), Some(payload)) = (
+        &parsing_result.leaf_update,
+        &parsing_result.tree_update,
+        &parsing_result.payload,
+    ) {
+        let (collection, verify) = match payload {
+            Payload::CollectionVerification {
+                collection, verify, ..
+            } => (collection, verify),
+            _ => {
+                return Err(ProgramTransformerError::ParsingError(
+                    "Ix not parsed correctly".to_string(),
+                ));
+            }
+        };
+        debug!(
+            "Handling collection verification event for {} (verify: {}): {}",
+            collection, verify, bundle.txn_id
+        );
+        let seq = save_changelog_event(cl, bundle.slot, bundle.txn_id, txn, instruction, cl_audits)
+            .await?;
+        let id_bytes = match le.schema {
+            LeafSchema::V1 { id, .. } => id.to_bytes().to_vec(),
+        };
+
+        let tree_id = cl.id.to_bytes();
+        let nonce = cl.index as i64;
+
+        // Begin a transaction.  If the transaction goes out of scope (i.e. one of the executions has
+        // an error and this function returns it using the `?` operator), then the transaction is
+        // automatically rolled back.
+        let multi_txn = txn.begin().await?;
+
+        // Partial update of asset table with just leaf.
+        upsert_asset_with_leaf_info(
+            &multi_txn,
+            id_bytes.to_vec(),
+            nonce,
+            tree_id.to_vec(),
+            le.leaf_hash.to_vec(),
+            le.schema.data_hash(),
+            le.schema.creator_hash(),
+            seq as i64,
+        )
+        .await?;
+
+        upsert_asset_with_seq(&multi_txn, id_bytes.to_vec(), seq as i64).await?;
+
+        upsert_collection_info(
+            &multi_txn,
+            id_bytes.to_vec(),
+            Some(Collection {
+                key: *collection,
+                verified: *verify,
+            }),
+            bundle.slot as i64,
+            seq as i64,
+        )
+        .await?;
+
+        multi_txn.commit().await?;
+
+        return Ok(());
+    };
+    Err(ProgramTransformerError::ParsingError(
+        "Ix not parsed correctly".to_string(),
+    ))
+}