Merge branch 'main' into triton-build

Cargo.toml

@@ -1,26 +1,29 @@
 [workspace]
-resolver = "2"
 members = [
+  "core",
   "das_api",
   "digital_asset_types",
   "metadata_json",
   "metaplex-rpc-proxy",
   "migration",
   "nft_ingester",
   "tree_backfiller",
+  "ops",
   "tools/acc_forwarder",
   "tools/bgtask_creator",
   "tools/fetch_trees",
   "tools/load_generation",
   "tools/tree-status",
   "tools/txn_forwarder",
+  "integration_tests",
 ]
+resolver = "2"
 
 [workspace.package]
-version = "0.7.2"
 edition = "2021"
-repository = "https://github.com/metaplex-foundation/digital-asset-rpc-infrastructure"
 publish = false
+repository = "https://github.com/metaplex-foundation/digital-asset-rpc-infrastructure"
+version = "0.7.2"
 
 [workspace.dependencies]
 anchor-client = "0.28.0"
@@ -29,7 +32,7 @@ anyhow = "1.0.75"
 async-std = "1.0.0"
 async-trait = "0.1.60"
 base64 = "0.21.0"
-blockbuster = "=0.9.0-beta.5"
+blockbuster = "1.0.1"
 borsh = "~0.10.3"
 borsh-derive = "~0.10.3"
 bs58 = "0.4.0"
@@ -39,33 +42,38 @@ chrono = "0.4.19"
 clap = "4.2.2"
 das-metadata-json = { path = "metadata_json" }
 das-tree-backfiller = { path = "tree_backfiller" }
+das_api = { path = "das_api" }
+das-core = { path = "core" }
 digital_asset_types = { path = "digital_asset_types" }
 enum-iterator = "1.2.0"
 enum-iterator-derive = "1.1.0"
 env_logger = "0.10.0"
 fake = "2.5.0"
 figment = "0.10.8"
 flatbuffers = "23.1.21"
+function_name = "0.3.0"
 futures = "0.3.28"
 futures-util = "0.3.27"
 hex = "0.4.3"
 hyper = "0.14.23"
 indexmap = "1.9.3"
+insta = { version = "1.34.0", features = ["json"] }
+itertools = "0.10.1"
 jsonpath_lib = "0.3.0"
 jsonrpsee = "0.16.2"
 jsonrpsee-core = "0.16.2"
 lazy_static = "1.4.0"
 log = "0.4.17"
 metrics = "0.20.1"
+migration = { path = "migration" }
 mime_guess = "2.0.4"
-mpl-bubblegum = "=1.0.1-beta.4"
-mpl-candy-guard = "2.0.0"
-mpl-candy-machine-core = "2.0.1"
-mpl-token-metadata = "=2.0.0-beta.1"
+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"
+once_cell = "1.19.0"
 open-rpc-derive = "0.0.4"
 open-rpc-schema = "0.0.4"
 plerkle_messenger = "1.6.0"
@@ -79,11 +87,18 @@ reqwest = "0.11.13"
 rust-crypto = "0.2.36"
 schemars = "0.8.6"
 schemars_derive = "0.8.6"
-sea-orm = "0.10.6"
+sea-orm = { version = "0.10.6", features = [
+  "macros",
+  "runtime-tokio-rustls",
+  "sqlx-postgres",
+  "with-chrono",
+  "mock",
+] }
 sea-orm-migration = "0.10.6"
 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-geyser-plugin-interface = "~1.16.16"
@@ -95,11 +110,18 @@ spl-account-compression = "0.2.0"
 spl-associated-token-account = ">= 1.1.3, < 3.0"
 spl-concurrent-merkle-tree = "0.2.0"
 spl-noop = "0.2.0"
-spl-token = ">= 3.5.0, < 5.0"
-sqlx = "0.6.2"
+spl-token = { version = ">= 3.5.0, < 5.0", features = ["no-entrypoint"] }
+sqlx = { version = "0.6.2", features = [
+  "macros",
+  "runtime-tokio-rustls",
+  "postgres",
+  "uuid",
+  "offline",
+  "json",
+] }
 stretto = "0.7.2"
 thiserror = "1.0.31"
-tokio = "1.30.0"
+tokio = { version = "1.30.0", features = ["macros", "rt-multi-thread"] }
 tokio-postgres = "0.7.7"
 tokio-stream = "0.1.14"
 tower = "0.4.13"
@@ -118,5 +140,5 @@ missing_const_for_fn = "deny"
 trivially_copy_pass_by_ref = "deny"
 
 [profile.release]
-lto = true
 codegen-units = 1
+lto = true

core/Cargo.toml

@@ -0,0 +1,18 @@
+[package]
+name = "das-core"
+version.workspace = true
+edition.workspace = true
+repository.workspace = true
+publish.workspace = true
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+clap = { workspace = true }
+anyhow = { workspace = true }
+sqlx = { workspace = true }
+cadence = { workspace = true }
+cadence-macros = { workspace = true }
+
+[lints]
+workspace = true

core/src/db.rs

@@ -0,0 +1,38 @@
+use anyhow::Result;
+use clap::Parser;
+use sqlx::{
+    postgres::{PgConnectOptions, PgPoolOptions},
+    PgPool,
+};
+
+#[derive(Debug, Parser, Clone)]
+pub struct PoolArgs {
+    /// The database URL.
+    #[arg(long, env)]
+    pub database_url: String,
+    /// The maximum number of connections to the database.
+    #[arg(long, env, default_value = "125")]
+    pub database_max_connections: u32,
+    /// The minimum number of connections to the database.
+    #[arg(long, env, default_value = "5")]
+    pub database_min_connections: u32,
+}
+
+///// Establishes a connection to the database using the provided configuration.
+/////
+///// # Arguments
+/////
+///// * `config` - A `PoolArgs` struct containing the database URL and the minimum and maximum number of connections.
+/////
+///// # Returns
+/////
+///// * `Result<DatabaseConnection, DbErr>` - On success, returns a `DatabaseConnection`. On failure, returns a `DbErr`.
+pub async fn connect_db(config: PoolArgs) -> Result<PgPool, sqlx::Error> {
+    let options: PgConnectOptions = config.database_url.parse()?;
+
+    PgPoolOptions::new()
+        .min_connections(config.database_min_connections)
+        .max_connections(config.database_max_connections)
+        .connect_with(options)
+        .await
+}

core/src/lib.rs

@@ -0,0 +1,5 @@
+mod db;
+mod metrics;
+
+pub use db::*;
+pub use metrics::*;

core/src/metrics.rs

@@ -0,0 +1,31 @@
+use anyhow::Result;
+use cadence::{BufferedUdpMetricSink, QueuingMetricSink, StatsdClient};
+use cadence_macros::set_global_default;
+use clap::Parser;
+use std::net::UdpSocket;
+
+#[derive(Clone, Parser, Debug)]
+pub struct MetricsArgs {
+    #[arg(long, env, default_value = "127.0.0.1")]
+    pub metrics_host: String,
+    #[arg(long, env, default_value = "8125")]
+    pub metrics_port: u16,
+    #[arg(long, env, default_value = "das.backfiller")]
+    pub metrics_prefix: String,
+}
+
+pub fn setup_metrics(config: MetricsArgs) -> Result<()> {
+    let host = (config.metrics_host, config.metrics_port);
+
+    let socket = UdpSocket::bind("0.0.0.0:0")?;
+    socket.set_nonblocking(true)?;
+
+    let udp_sink = BufferedUdpMetricSink::from(host, socket)?;
+    let queuing_sink = QueuingMetricSink::from(udp_sink);
+
+    let client = StatsdClient::from_sink(&config.metrics_prefix, queuing_sink);
+
+    set_global_default(client);
+
+    Ok(())
+}

das_api/Cargo.toml

@@ -21,9 +21,7 @@ jsonrpsee-core = { workspace = true, features =["server"]}
 log = { workspace = true }
 metrics = { workspace = true }
 mpl-bubblegum = { workspace = true }
-mpl-candy-guard = { workspace = true, features = ["no-entrypoint"] }
-mpl-candy-machine-core = { workspace = true, features = ["no-entrypoint"] }
-mpl-token-metadata = { workspace = true,  features = ["serde-feature"] }
+mpl-token-metadata = { workspace = true,  features = ["serde"] }
 open-rpc-derive = { workspace = true }
 open-rpc-schema = { workspace = true }
 schemars = { workspace = true }

das_api/src/api/api_impl.rs

@@ -20,13 +20,13 @@ use digital_asset_types::{
 use open_rpc_derive::document_rpc;
 use sea_orm::{sea_query::ConditionType, ConnectionTrait, DbBackend, Statement};
 
+use crate::error::DasApiError;
 use crate::validation::{validate_opt_pubkey, validate_search_with_name};
 use open_rpc_schema::document::OpenrpcDocument;
 use {
     crate::api::*,
     crate::config::Config,
     crate::validation::validate_pubkey,
-    crate::DasApiError,
     async_trait::async_trait,
     digital_asset_types::rpc::{response::AssetList, Asset, AssetProof},
     sea_orm::{DatabaseConnection, DbErr, SqlxPostgresConnector},

das_api/src/api/mod.rs

@@ -1,4 +1,4 @@
-use crate::DasApiError;
+use crate::error::DasApiError;
 use async_trait::async_trait;
 use digital_asset_types::rpc::filter::{AssetSortDirection, SearchConditionType};
 use digital_asset_types::rpc::options::Options;
@@ -44,7 +44,7 @@ pub struct GetAssetsByOwner {
     pub cursor: Option<String>,
 }
 
-#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema, Default)]
 #[serde(deny_unknown_fields, rename_all = "camelCase")]
 pub struct GetAsset {
     pub id: String,
@@ -88,7 +88,7 @@ pub struct GetAssetsByCreator {
     pub cursor: Option<String>,
 }
 
-#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema, Default)]
 #[serde(deny_unknown_fields, rename_all = "camelCase")]
 pub struct SearchAssets {
     pub negate: Option<bool>,

das_api/src/builder.rs

@@ -1,6 +1,8 @@
+use jsonrpsee::RpcModule;
 use log::debug;
 
-use crate::{api::*, DasApiError, RpcModule};
+use crate::{api::*, error::DasApiError};
+
 pub struct RpcApiBuilder;
 
 impl RpcApiBuilder {

das_api/src/config.rs

@@ -4,7 +4,7 @@ use {
     serde::Deserialize,
 };
 
-#[derive(Deserialize)]
+#[derive(Deserialize, Default)]
 pub struct Config {
     pub database_url: String,
     pub metrics_port: Option<u16>,

das_api/src/lib.rs

@@ -0,0 +1,6 @@
+pub mod api;
+
+pub mod builder;
+pub mod config;
+pub mod error;
+pub mod validation;

das_api/src/main.rs

@@ -25,7 +25,7 @@ use tower_http::cors::{Any, CorsLayer};
 use jsonrpsee::server::{
     logger::{Logger, TransportProtocol},
     middleware::proxy_get_request::ProxyGetRequestLayer,
-    RpcModule, ServerBuilder,
+    ServerBuilder,
 };
 
 use cadence_macros::{is_global_default_set, statsd_time};

das_api/src/validation.rs

@@ -1,4 +1,4 @@
-use crate::DasApiError;
+use crate::error::DasApiError;
 use solana_sdk::pubkey::Pubkey;
 use std::str::FromStr;
 

digital_asset_types/src/dao/extensions/asset.rs

@@ -94,6 +94,10 @@ impl Default for asset::Model {
             created_at: None,
             burnt: Default::default(),
             slot_updated: None,
+            slot_updated_metadata_account: None,
+            slot_updated_mint_account: None,
+            slot_updated_token_account: None,
+            slot_updated_cnft_transaction: None,
             data_hash: None,
             creator_hash: None,
             owner_delegate_seq: None,

digital_asset_types/src/dao/generated/asset.rs

@@ -41,6 +41,10 @@ pub struct Model {
     pub created_at: Option<DateTimeWithTimeZone>,
     pub burnt: bool,
     pub slot_updated: Option<i64>,
+    pub slot_updated_metadata_account: Option<i64>,
+    pub slot_updated_mint_account: Option<i64>,
+    pub slot_updated_token_account: Option<i64>,
+    pub slot_updated_cnft_transaction: Option<i64>,
     pub data_hash: Option<String>,
     pub creator_hash: Option<String>,
     pub owner_delegate_seq: Option<i64>,
@@ -73,6 +77,10 @@ pub enum Column {
     CreatedAt,
     Burnt,
     SlotUpdated,
+    SlotUpdatedMetadataAccount,
+    SlotUpdatedMintAccount,
+    SlotUpdatedTokenAccount,
+    SlotUpdatedCnftTransaction,
     DataHash,
     CreatorHash,
     OwnerDelegateSeq,
@@ -122,6 +130,10 @@ impl ColumnTrait for Column {
             Self::CreatedAt => ColumnType::TimestampWithTimeZone.def().null(),
             Self::Burnt => ColumnType::Boolean.def(),
             Self::SlotUpdated => ColumnType::BigInteger.def().null(),
+            Self::SlotUpdatedMetadataAccount => ColumnType::BigInteger.def().null(),
+            Self::SlotUpdatedMintAccount => ColumnType::BigInteger.def().null(),
+            Self::SlotUpdatedTokenAccount => ColumnType::BigInteger.def().null(),
+            Self::SlotUpdatedCnftTransaction => ColumnType::BigInteger.def().null(),
             Self::DataHash => ColumnType::Char(Some(50u32)).def().null(),
             Self::CreatorHash => ColumnType::Char(Some(50u32)).def().null(),
             Self::OwnerDelegateSeq => ColumnType::BigInteger.def().null(),

digital_asset_types/src/dao/generated/prelude.rs

@@ -1,5 +1,7 @@
 //! SeaORM Entity. Generated by sea-orm-codegen 0.9.3
 
+#![allow(unused_imports)]
+
 pub use super::asset::Entity as Asset;
 pub use super::asset_authority::Entity as AssetAuthority;
 pub use super::asset_creators::Entity as AssetCreators;

digital_asset_types/src/json/chain_data.rs

@@ -1,4 +1,4 @@
-use blockbuster::token_metadata::state::{TokenStandard, Uses};
+use blockbuster::token_metadata::types::{TokenStandard, Uses};
 use serde::{Deserialize, Serialize};
 
 pub enum ChainData {

digital_asset_types/tests/common.rs

@@ -1,4 +1,4 @@
-use blockbuster::token_metadata::state::*;
+use blockbuster::token_metadata::types::{Collection, Creator, TokenStandard, Uses};
 use digital_asset_types::dao::sea_orm_active_enums::{
     SpecificationAssetClass, SpecificationVersions,
 };
@@ -152,6 +152,10 @@ pub fn create_asset(
             created_at: None,
             specification_asset_class: Some(SpecificationAssetClass::Nft),
             slot_updated: Some(0),
+            slot_updated_metadata_account: Some(0),
+            slot_updated_mint_account: None,
+            slot_updated_token_account: None,
+            slot_updated_cnft_transaction: None,
             data_hash: None,
             alt_id: None,
             creator_hash: None,