sync upstream

Cargo.toml

@@ -7,6 +7,7 @@ members = [
   "digital_asset_types",
   "integration_tests",
   "metaplex-rpc-proxy",
+  "metadata_json",
   "migration",
   "nft_ingester",
   "ops",
@@ -48,6 +49,7 @@ cargo-lock = "9.0.0"
 chrono = "0.4.19"
 clap = "4.2.2"
 das-core = { path = "core" }
+das-metadata-json = { path = "metadata_json" }
 das-bubblegum = { path = "bubblegum" }
 das_api = { path = "das_api" }
 derive_more = { version = "0.99.17" }

core/src/metadata_json.rs

@@ -83,7 +83,7 @@ pub async fn create_download_metadata_notifier(
     })
 }
 
-#[derive(Parser, Clone, Debug)]
+#[derive(Parser, Clone, Debug, PartialEq, Eq)]
 pub struct MetadataJsonDownloadWorkerArgs {
     /// The number of worker threads
     #[arg(long, env, default_value = "25")]

das_api/src/api/api_impl.rs

@@ -40,7 +40,7 @@ pub struct DasApi {
 impl DasApi {
     pub async fn from_config(config: Config) -> Result<Self, DasApiError> {
         let pool = PgPoolOptions::new()
-            .max_connections(250)
+            .max_connections(config.max_database_connections.unwrap_or(250))
             .connect(&config.database_url)
             .await?;
 

das_api/src/config.rs

@@ -7,6 +7,8 @@ use {
 #[derive(Deserialize, Default)]
 pub struct Config {
     pub database_url: String,
+    pub max_database_connections: Option<u32>,
+    pub max_request_connections: Option<u32>,
     pub metrics_port: Option<u16>,
     pub metrics_host: Option<String>,
     pub server_port: u16,

das_api/src/main.rs

@@ -134,6 +134,7 @@ async fn main() -> Result<(), DasApiError> {
 
     let server = ServerBuilder::default()
         .set_middleware(middleware)
+        .max_connections(config.max_request_connections.unwrap_or(100))
         .set_logger(MetricMiddleware)
         .build(addr)
         .await?;

digital_asset_types/src/dao/generated/tree_transactions.rs

@@ -0,0 +1,67 @@
+//! `SeaORM` Entity. Generated by sea-orm-codegen 0.10.5
+
+use sea_orm::entity::prelude::*;
+use serde::{Deserialize, Serialize};
+
+#[derive(Copy, Clone, Default, Debug, DeriveEntity)]
+pub struct Entity;
+
+impl EntityName for Entity {
+    fn table_name(&self) -> &str {
+        "tree_transactions"
+    }
+}
+
+#[derive(Clone, Debug, PartialEq, DeriveModel, DeriveActiveModel, Eq, Serialize, Deserialize)]
+pub struct Model {
+    pub signature: String,
+    pub tree: String,
+    pub slot: i64,
+    pub created_at: Option<DateTimeWithTimeZone>,
+    pub processed_at: Option<DateTimeWithTimeZone>,
+}
+
+#[derive(Copy, Clone, Debug, EnumIter, DeriveColumn)]
+pub enum Column {
+    Signature,
+    Tree,
+    Slot,
+    CreatedAt,
+    ProcessedAt,
+}
+
+#[derive(Copy, Clone, Debug, EnumIter, DerivePrimaryKey)]
+pub enum PrimaryKey {
+    Signature,
+}
+
+impl PrimaryKeyTrait for PrimaryKey {
+    type ValueType = String;
+    fn auto_increment() -> bool {
+        false
+    }
+}
+
+#[derive(Copy, Clone, Debug, EnumIter)]
+pub enum Relation {}
+
+impl ColumnTrait for Column {
+    type EntityName = Entity;
+    fn def(&self) -> ColumnDef {
+        match self {
+            Self::Signature => ColumnType::Text.def(),
+            Self::Tree => ColumnType::Text.def(),
+            Self::Slot => ColumnType::BigInteger.def(),
+            Self::CreatedAt => ColumnType::TimestampWithTimeZone.def().null(),
+            Self::ProcessedAt => ColumnType::TimestampWithTimeZone.def().null(),
+        }
+    }
+}
+
+impl RelationTrait for Relation {
+    fn def(&self) -> RelationDef {
+        panic!("No RelationDef")
+    }
+}
+
+impl ActiveModelBehavior for ActiveModel {}

digital_asset_types/src/dapi/signatures_for_asset.rs

@@ -0,0 +1,36 @@
+use crate::dao::scopes;
+use crate::dao::PageOptions;
+use crate::rpc::filter::AssetSorting;
+use crate::rpc::response::TransactionSignatureList;
+use sea_orm::DatabaseConnection;
+use sea_orm::DbErr;
+use super::common::build_transaction_signatures_response;
+use super::common::{build_asset_response, create_pagination, create_sorting};
+
+
+pub async fn get_signatures_for_asset(
+    db: &DatabaseConnection,
+    asset_id: Option<Vec<u8>>,
+    tree: Option<Vec<u8>>,
+    leaf_idx: Option<i64>,
+    sorting: AssetSorting,
+    page_options: &PageOptions,
+) -> Result<TransactionSignatureList, DbErr> {
+    let pagination = create_pagination(&page_options)?;
+    let (sort_direction, sort_column) = create_sorting(sorting);
+    let transactions = scopes::asset::get_signatures_for_asset(
+        db,
+        asset_id,
+        tree,
+        leaf_idx,
+        sort_direction,
+        &pagination,
+        page_options.limit
+    )
+    .await?;
+    Ok(build_transaction_signatures_response(
+        transactions,
+        page_options.limit,
+        &pagination,
+    ))
+}

digital_asset_types/src/rpc/display_options.rs

@@ -0,0 +1,9 @@
+use schemars::JsonSchema;
+use serde::{Deserialize, Serialize};
+
+#[derive(Serialize, Deserialize, Clone, Debug, Eq, PartialEq, JsonSchema, Default)]
+#[serde(deny_unknown_fields, rename_all = "camelCase")]
+pub struct DisplayOptions {
+    #[serde(default)]
+    pub show_unverified_collections: bool,
+}

grpc-ingest/src/grpc.rs

@@ -262,7 +262,6 @@ impl SubscriptionTask {
                                                                     "*",
                                                                     account.encode_to_vec(),
                                                                 );
-                                                                debug!(target: "grpc2redis", action = "process_account_update", label = ?label, stream = ?stream, maxlen = ?stream_maxlen);
                                                             }
 
                                                             UpdateOneof::Transaction(transaction) => {
@@ -272,7 +271,6 @@ impl SubscriptionTask {
                                                                     "*",
                                                                     transaction.encode_to_vec(),
                                                                 );
-                                                                debug!(target: "grpc2redis", action = "process_transaction_update", label = ?label, stream = ?stream, maxlen = ?stream_maxlen);
                                                             }
                                                             _ => {
                                                                 warn!(target: "grpc2redis", action = "unknown_update_variant", label = ?label, message = "Unknown update variant");
@@ -315,14 +313,14 @@ impl SubscriptionTask {
                                         }
                                     }
                                 }
-                                _ => {
-                                    break;
-                                }
-                            }
+                                None | Some(Err(_)) => {
+                                    warn!(target: "grpc2redis", action = "stream_closed", message = "Stream closed, stopping subscription task", ?label);
 
-                            let mut global_shutdown_tx = global_shutdown_tx.lock().await;
-                            if let Some(global_shutdown_tx) = global_shutdown_tx.take() {
-                                let _ = global_shutdown_tx.send(());
+                                    let mut global_shutdown_tx = global_shutdown_tx.lock().await;
+                                    if let Some(global_shutdown_tx) = global_shutdown_tx.take() {
+                                        let _ = global_shutdown_tx.send(());
+                                    }
+                                }
                             }
                         }
                         _ = &mut shutdown_rx => {

integration_tests/tests/integration_tests/snapshots/integration_tests__fungibles_and_token_extensions_tests__fungible_token_get_asset_scenario_1.snap

@@ -1,10 +1,9 @@
 ---
 source: integration_tests/tests/integration_tests/fungibles_and_token_extensions_tests.rs
 expression: response
-snapshot_kind: text
 ---
 {
-  "interface": "FungibleToken",
+  "interface": "Custom",
   "id": "EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v",
   "content": {
     "$schema": "https://schema.metaplex.com/nft1.0.json",