Remove program_transformers dependency on plerkle

Cargo.toml

@@ -25,13 +25,13 @@ repository = "https://github.com/metaplex-foundation/digital-asset-rpc-infrastru
 version = "0.7.2"
 
 [workspace.dependencies]
-anchor-client = "0.28.0"
-anchor-lang = "0.28.0"
+anchor-client = "0.29.0"
+anchor-lang = "0.29.0"
 anyhow = "1.0.75"
 async-std = "1.0.0"
 async-trait = "0.1.60"
 backon = "0.4.1"
-blockbuster = "1.0.1"
+blockbuster = "2.0.0"
 borsh = "~0.10.3"
 borsh-derive = "~0.10.3"
 bs58 = "0.4.0"
@@ -73,7 +73,7 @@ once_cell = "1.19.0"
 open-rpc-derive = "0.0.4"
 open-rpc-schema = "0.0.4"
 plerkle_messenger = "1.6.0"
-plerkle_serialization = "1.6.0"
+plerkle_serialization = "1.8.0"
 program_transformers = { path = "program_transformers" }
 prometheus = "0.13.3"
 proxy-wasm = "0.2.0"
@@ -89,12 +89,12 @@ sea-query = "0.28.1"
 serde = "1.0.137"
 serde_json = "1.0.81"
 serial_test = "2.0.0"
-solana-account-decoder = "~1.16.16"
-solana-client = "~1.16.16"
-solana-program = "~1.16.16"
-solana-sdk = "~1.16.16"
-solana-transaction-status = "~1.16.16"
-spl-account-compression = "0.2.0"
+solana-account-decoder = "~1.17"
+solana-client = "~1.17"
+solana-program = "~1.17"
+solana-sdk = "~1.17"
+solana-transaction-status = "~1.17"
+spl-account-compression = "0.3.0"
 spl-associated-token-account = ">= 1.1.3, < 3.0"
 spl-concurrent-merkle-tree = "0.2.0"
 spl-noop = "0.2.0"

integration_tests/tests/integration_tests/common.rs

@@ -12,12 +12,16 @@ use migration::sea_orm::{
 use migration::{Migrator, MigratorTrait};
 use mpl_token_metadata::accounts::Metadata;
 
-use nft_ingester::config;
+use nft_ingester::{
+    config,
+    plerkle::{PlerkleAccountInfo, PlerkleTransactionInfo},
+};
 use once_cell::sync::Lazy;
-use plerkle_serialization::root_as_account_info;
-use plerkle_serialization::root_as_transaction_info;
-use plerkle_serialization::serializer::serialize_account;
-use plerkle_serialization::solana_geyser_plugin_interface_shims::ReplicaAccountInfoV2;
+use plerkle_serialization::{
+    root_as_account_info, root_as_transaction_info,
+    serializer::{seralize_encoded_transaction_with_status, serialize_account},
+    solana_geyser_plugin_interface_shims::ReplicaAccountInfoV2,
+};
 use program_transformers::ProgramTransformer;
 
 use solana_client::nonblocking::rpc_client::RpcClient;
@@ -34,8 +38,6 @@ use futures::TryStreamExt;
 use tokio_stream::{self as stream};
 
 use log::{error, info};
-use plerkle_serialization::serializer::seralize_encoded_transaction_with_status;
-// use rand::seq::SliceRandom;
 use serde::de::DeserializeOwned;
 use solana_account_decoder::{UiAccount, UiAccountEncoding};
 use solana_client::{
@@ -354,9 +356,13 @@ pub async fn get_token_largest_account(client: &RpcClient, mint: Pubkey) -> anyh
 pub async fn index_account_bytes(setup: &TestSetup, account_bytes: Vec<u8>) {
     let account = root_as_account_info(&account_bytes).unwrap();
 
+    let account = PlerkleAccountInfo(account)
+        .try_into()
+        .expect("failed to parse account info");
+
     setup
         .transformer
-        .handle_account_update(account)
+        .handle_account_update(&account)
         .await
         .unwrap();
 }
@@ -419,7 +425,16 @@ async fn cached_fetch_transaction(setup: &TestSetup, sig: Signature) -> Vec<u8>
 pub async fn index_transaction(setup: &TestSetup, sig: Signature) {
     let txn_bytes: Vec<u8> = cached_fetch_transaction(setup, sig).await;
     let txn = root_as_transaction_info(&txn_bytes).unwrap();
-    setup.transformer.handle_transaction(&txn).await.unwrap();
+
+    let transaction_info = PlerkleTransactionInfo(txn)
+        .try_into()
+        .expect("failed to parse txn");
+
+    setup
+        .transformer
+        .handle_transaction(&transaction_info)
+        .await
+        .unwrap();
 }
 
 async fn cached_fetch_largest_token_account_id(client: &RpcClient, mint: Pubkey) -> Pubkey {

nft_ingester/Cargo.toml

@@ -14,7 +14,10 @@ cadence = { workspace = true }
 cadence-macros = { workspace = true }
 chrono = { workspace = true }
 clap = { workspace = true, features = ["derive", "cargo"] }
-digital_asset_types = { workspace = true, features = ["json_types", "sql_types"] }
+digital_asset_types = { workspace = true, features = [
+  "json_types",
+  "sql_types",
+] }
 figment = { workspace = true, features = ["env", "toml", "yaml"] }
 flatbuffers = { workspace = true }
 futures = { workspace = true }
@@ -26,19 +29,36 @@ program_transformers = { workspace = true }
 rand = { workspace = true }
 reqwest = { workspace = true }
 rust-crypto = { workspace = true }
-sea-orm = { workspace = true, features = ["macros", "runtime-tokio-rustls", "sqlx-postgres", "with-chrono", "mock"] }
+sea-orm = { workspace = true, features = [
+  "macros",
+  "runtime-tokio-rustls",
+  "sqlx-postgres",
+  "with-chrono",
+  "mock",
+] }
 serde = { workspace = true }
 serde_json = { workspace = true }
 solana-account-decoder = { workspace = true }
 solana-client = { workspace = true }
 solana-sdk = { workspace = true }
 solana-transaction-status = { workspace = true }
 spl-account-compression = { workspace = true, features = ["no-entrypoint"] }
-sqlx = { workspace = true, features = ["macros", "runtime-tokio-rustls", "postgres", "uuid", "offline", "json"] }
+sqlx = { workspace = true, features = [
+  "macros",
+  "runtime-tokio-rustls",
+  "postgres",
+  "uuid",
+  "offline",
+  "json",
+] }
 stretto = { workspace = true, features = ["async"] }
 thiserror = { workspace = true }
 tokio = { workspace = true, features = ["tracing"] }
-tracing-subscriber = { workspace = true, features = ["json", "env-filter", "ansi"] }
+tracing-subscriber = { workspace = true, features = [
+  "json",
+  "env-filter",
+  "ansi",
+] }
 url = { workspace = true }
 
 [lints]

nft_ingester/src/account_updates.rs

@@ -2,6 +2,7 @@ use {
     crate::{
         metric,
         metrics::capture_result,
+        plerkle::PlerkleAccountInfo,
         tasks::{create_download_metadata_notifier, TaskData},
     },
     cadence_macros::{is_global_default_set, statsd_count, statsd_time},
@@ -103,7 +104,10 @@ async fn handle_account(
             account = Some(bs58::encode(pubkey.0.as_slice()).into_string());
         }
         let begin_processing = Instant::now();
-        let res = manager.handle_account_update(account_update).await;
+
+        let account_info = PlerkleAccountInfo(account_update).try_into().ok()?;
+        let res = manager.handle_account_update(&account_info).await;
+
         let should_ack = capture_result(
             id.clone(),
             stream_key,

nft_ingester/src/lib.rs

@@ -1,10 +1,10 @@
-pub mod account_updates;
 pub mod ack;
 pub mod backfiller;
 pub mod config;
 pub mod database;
 pub mod error;
 pub mod metrics;
+pub mod plerkle;
 pub mod stream;
 pub mod tasks;
 pub mod transaction_notifications;

nft_ingester/src/main.rs

@@ -5,6 +5,7 @@ pub mod config;
 mod database;
 pub mod error;
 pub mod metrics;
+mod plerkle;
 mod stream;
 pub mod tasks;
 mod transaction_notifications;

nft_ingester/src/plerkle.rs

@@ -0,0 +1,75 @@
+use {
+    plerkle_serialization::deserializer::*,
+    program_transformers::{error::ProgramTransformerError, AccountInfo, TransactionInfo},
+};
+
+pub fn into_program_transformer_err(e: PlerkleDeserializerError) -> ProgramTransformerError {
+    ProgramTransformerError::DeserializationError(e.to_string())
+}
+
+#[derive(thiserror::Error, Clone, Debug)]
+pub enum PlerkleDeserializerError {
+    #[error("Not found")]
+    NotFound,
+    #[error("Solana error: {0}")]
+    Solana(#[from] SolanaDeserializerError),
+}
+
+pub struct PlerkleAccountInfo<'a>(pub plerkle_serialization::AccountInfo<'a>);
+
+impl<'a> TryFrom<PlerkleAccountInfo<'a>> for AccountInfo {
+    type Error = PlerkleDeserializerError;
+
+    fn try_from(value: PlerkleAccountInfo) -> Result<Self, Self::Error> {
+        let account_info = value.0;
+
+        Ok(Self {
+            slot: account_info.slot(),
+            pubkey: account_info
+                .pubkey()
+                .ok_or(PlerkleDeserializerError::NotFound)?
+                .try_into()?,
+            owner: account_info
+                .owner()
+                .ok_or(PlerkleDeserializerError::NotFound)?
+                .try_into()?,
+            data: PlerkleOptionalU8Vector(account_info.data()).try_into()?,
+        })
+    }
+}
+
+pub struct PlerkleTransactionInfo<'a>(pub plerkle_serialization::TransactionInfo<'a>);
+
+impl<'a> TryFrom<PlerkleTransactionInfo<'a>> for TransactionInfo {
+    type Error = PlerkleDeserializerError;
+
+    fn try_from(value: PlerkleTransactionInfo<'a>) -> Result<Self, Self::Error> {
+        let tx_info = value.0;
+
+        let slot = tx_info.slot();
+        let signature = PlerkleOptionalStr(tx_info.signature()).try_into()?;
+        let account_keys = PlerkleOptionalPubkeyVector(tx_info.account_keys()).try_into()?;
+        let message_instructions = PlerkleCompiledInstructionVector(
+            tx_info
+                .outer_instructions()
+                .ok_or(PlerkleDeserializerError::NotFound)?,
+        )
+        .try_into()?;
+        let compiled = tx_info.compiled_inner_instructions();
+        let inner = tx_info.inner_instructions();
+        let meta_inner_instructions = if let Some(c) = compiled {
+            PlerkleCompiledInnerInstructionVector(c).try_into()
+        } else {
+            PlerkleInnerInstructionsVector(inner.ok_or(PlerkleDeserializerError::NotFound)?)
+                .try_into()
+        }?;
+
+        Ok(Self {
+            slot,
+            signature,
+            account_keys,
+            message_instructions,
+            meta_inner_instructions,
+        })
+    }
+}

nft_ingester/src/transaction_notifications.rs

@@ -2,6 +2,7 @@ use {
     crate::{
         metric,
         metrics::capture_result,
+        plerkle::{into_program_transformer_err, PlerkleTransactionInfo},
         tasks::{create_download_metadata_notifier, TaskData},
     },
     cadence_macros::{is_global_default_set, statsd_count, statsd_time},
@@ -101,7 +102,11 @@ async fn handle_transaction(
         }
 
         let begin = Instant::now();
-        let res = manager.handle_transaction(&tx).await;
+        let transaction_info = PlerkleTransactionInfo(tx)
+            .try_into()
+            .map_err(into_program_transformer_err)
+            .ok()?;
+        let res = manager.handle_transaction(&transaction_info).await;
         let should_ack = capture_result(
             id.clone(),
             stream_key,

program_transformers/Cargo.toml

@@ -14,10 +14,10 @@ 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 }
 serde_json = { workspace = true }
 solana-sdk = { workspace = true }
+solana-transaction-status = { workspace = true }
 spl-account-compression = { workspace = true, features = ["no-entrypoint"] }
 spl-token = { workspace = true, features = ["no-entrypoint"] }
 sqlx = { workspace = true }

program_transformers/src/bubblegum/mint_v1.rs

@@ -148,7 +148,7 @@ where
                     &multi_txn,
                     id_bytes.to_vec(),
                     nonce as i64,
-                    tree_id.to_vec(),
+                    tree_id.to_bytes().to_vec(),
                     le.leaf_hash.to_vec(),
                     le.schema.data_hash(),
                     le.schema.creator_hash(),
@@ -183,7 +183,7 @@ where
                 upsert_asset_authority(
                     &multi_txn,
                     id_bytes.to_vec(),
-                    authority.to_vec(),
+                    authority.to_bytes().to_vec(),
                     seq as i64,
                     slot_i,
                 )

program_transformers/src/bubblegum/update_metadata.rs

@@ -166,7 +166,7 @@ where
                     &multi_txn,
                     id_bytes.to_vec(),
                     nonce as i64,
-                    tree_id.to_vec(),
+                    tree_id.to_bytes().to_vec(),
                     le.leaf_hash.to_vec(),
                     le.schema.data_hash(),
                     le.schema.creator_hash(),