nft_ingester: use program_transformers crate

Cargo.toml

@@ -27,7 +27,6 @@ anchor-lang = "0.28.0"
 anyhow = "1.0.75"
 async-std = "1.0.0"
 async-trait = "0.1.60"
-base64 = "0.21.0"
 blockbuster = "1.0.1"
 borsh = "~0.10.3"
 borsh-derive = "~0.10.3"
@@ -44,7 +43,6 @@ fake = "2.5.0"
 figment = "0.10.8"
 flatbuffers = "23.1.21"
 futures = "0.3.28"
-futures-util = "0.3.27"
 hex = "0.4.3"
 hyper = "0.14.23"
 indexmap = "1.9.3"
@@ -59,16 +57,15 @@ mpl-bubblegum = "1.2.0"
 mpl-token-metadata = "4.1.1"
 nft_ingester = { path = "nft_ingester" }
 num-derive = "0.3.3"
-num-integer = { version = "0.1.44", default_features = false }
 num-traits = "0.2.15"
 open-rpc-derive = "0.0.4"
 open-rpc-schema = "0.0.4"
 plerkle_messenger = "1.6.0"
 plerkle_serialization = "1.6.0"
+program_transformers = { path = "program_transformers" }
 prometheus = "0.13.3"
 proxy-wasm = "0.2.0"
 rand = "0.8.5"
-redis = "0.22.3"
 regex = "1.6.0"
 reqwest = "0.11.13"
 rust-crypto = "0.2.36"
@@ -81,10 +78,8 @@ serde = "1.0.137"
 serde_json = "1.0.81"
 solana-account-decoder = "~1.16.16"
 solana-client = "~1.16.16"
-solana-geyser-plugin-interface = "~1.16.16"
 solana-program = "~1.16.16"
 solana-sdk = "~1.16.16"
-solana-sdk-macro = "~1.16.16"
 solana-transaction-status = "~1.16.16"
 spl-account-compression = "0.2.0"
 spl-associated-token-account = ">= 1.1.3, < 3.0"
@@ -95,15 +90,13 @@ sqlx = "0.6.2"
 stretto = "0.7.2"
 thiserror = "1.0.31"
 tokio = "1.30.0"
-tokio-postgres = "0.7.7"
 tokio-stream = "0.1.14"
 tower = "0.4.13"
 tower-http = "0.3.5"
 tracing = "0.1.35"
 tracing-subscriber = "0.3.16"
 txn_forwarder = { path = "tools/txn_forwarder" }
 url = "2.3.1"
-uuid = "1.0.0"
 wasi = "0.7.0"
 wasm-bindgen = "0.2.83"
 

das_api/Cargo.toml

@@ -33,7 +33,6 @@ solana-sdk = { workspace = true }
 sqlx = { workspace = true, features = ["macros", "runtime-tokio-rustls", "postgres", "uuid", "offline", "json"] }
 thiserror = { workspace = true }
 tokio = { workspace = true, features = ["signal"] }
-tokio-postgres = { workspace = true }
 tower = { workspace = true }
 tower-http = { workspace = true, features = ["cors"] }
 tracing = { workspace = true }

nft_ingester/Cargo.toml

@@ -6,9 +6,7 @@ repository = { workspace = true }
 publish = { workspace = true }
 
 [dependencies]
-anchor-lang = { workspace = true }
 async-trait = { workspace = true }
-base64 = { workspace = true }
 blockbuster = { workspace = true }
 borsh = { workspace = true }
 bs58 = { workspace = true }
@@ -17,46 +15,31 @@ cadence-macros = { workspace = true }
 chrono = { workspace = true }
 clap = { workspace = true, features = ["derive", "cargo"] }
 digital_asset_types = { workspace = true, features = ["json_types", "sql_types"] }
-env_logger = { workspace = true }
 figment = { workspace = true, features = ["env", "toml", "yaml"] }
 flatbuffers = { workspace = true }
 futures = { workspace = true }
-futures-util = { workspace = true }
-hex = { workspace = true }
-lazy_static = { workspace = true }
 log = { workspace = true }
 mpl-bubblegum = { workspace = true }
-num-integer = { workspace = true }
-num-traits = { workspace = true }
 plerkle_messenger = { workspace = true, features = ["redis"] }
 plerkle_serialization = { workspace = true }
+program_transformers = { workspace = true }
 rand = { workspace = true }
-redis = { workspace = true, features = ["aio", "tokio-comp", "streams", "tokio-native-tls-comp"] }
-regex = { workspace = true }
 reqwest = { workspace = true }
 rust-crypto = { workspace = true }
 sea-orm = { workspace = true, features = ["macros", "runtime-tokio-rustls", "sqlx-postgres", "with-chrono", "mock"] }
-sea-query = { workspace = true, features = ["postgres-array"] }
 serde = { workspace = true }
 serde_json = { workspace = true }
 solana-account-decoder = { workspace = true }
 solana-client = { workspace = true }
-solana-geyser-plugin-interface = { workspace = true }
 solana-sdk = { workspace = true }
-solana-sdk-macro = { workspace = true }
 solana-transaction-status = { workspace = true }
 spl-account-compression = { workspace = true, features = ["no-entrypoint"] }
-spl-concurrent-merkle-tree = { workspace = true }
-spl-token = { workspace = true, features = ["no-entrypoint"] }
 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"] }
-tokio-postgres = { workspace = true }
-tokio-stream = { workspace = true }
 tracing-subscriber = { workspace = true, features = ["json", "env-filter", "ansi"] }
 url = { workspace = true }
-uuid = { workspace = true }
 
 [lints]
 workspace = true

nft_ingester/src/account_updates.rs

@@ -1,18 +1,22 @@
-use std::sync::Arc;
-
-use crate::{
-    metric, metrics::capture_result, program_transformers::ProgramTransformer, tasks::TaskData,
-};
-use cadence_macros::{is_global_default_set, statsd_count, statsd_time};
-use chrono::Utc;
-use log::{debug, error};
-use plerkle_messenger::{ConsumptionType, Messenger, MessengerConfig, RecvData};
-use plerkle_serialization::root_as_account_info;
-use sqlx::{Pool, Postgres};
-use tokio::{
-    sync::mpsc::UnboundedSender,
-    task::{JoinHandle, JoinSet},
-    time::Instant,
+use {
+    crate::{
+        metric,
+        metrics::capture_result,
+        tasks::{create_download_metadata_notifier, TaskData},
+    },
+    cadence_macros::{is_global_default_set, statsd_count, statsd_time},
+    chrono::Utc,
+    log::{debug, error},
+    plerkle_messenger::{ConsumptionType, Messenger, MessengerConfig, RecvData},
+    plerkle_serialization::root_as_account_info,
+    program_transformers::ProgramTransformer,
+    sqlx::{Pool, Postgres},
+    std::sync::Arc,
+    tokio::{
+        sync::mpsc::UnboundedSender,
+        task::{JoinHandle, JoinSet},
+        time::Instant,
+    },
 };
 
 pub fn account_worker<T: Messenger>(
@@ -26,7 +30,11 @@ pub fn account_worker<T: Messenger>(
     tokio::spawn(async move {
         let source = T::new(config).await;
         if let Ok(mut msg) = source {
-            let manager = Arc::new(ProgramTransformer::new(pool, bg_task_sender, false));
+            let manager = Arc::new(ProgramTransformer::new(
+                pool,
+                create_download_metadata_notifier(bg_task_sender),
+                false,
+            ));
             loop {
                 let e = msg.recv(stream_key, consumption_type.clone()).await;
                 let mut tasks = JoinSet::new();

nft_ingester/src/lib.rs

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

nft_ingester/src/main.rs

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