feat: grpc client

digital_asset_types/src/dao/scopes/asset.rs

@@ -4,7 +4,7 @@ use std::string::ToString;
 use std::sync::Arc;
 
 use entities::api_req_params::{AssetSortDirection, Options};
-use entities::models::AssetSignatureWithPagination;
+use entities::models::{AssetSignatureWithPagination, OffChainData};
 use interface::asset_sigratures::AssetSignaturesGetter;
 use interface::json::{JsonDownloader, JsonPersister};
 use log::error;
@@ -17,7 +17,6 @@ use rocks_db::asset::{
     AssetAuthority, AssetCollection, AssetDynamicDetails, AssetLeaf, AssetOwner, AssetSelectedMaps,
     AssetStaticDetails,
 };
-use rocks_db::offchain_data::OffChainData;
 use rocks_db::Storage;
 use tokio::sync::Mutex;
 use tokio::task::{JoinError, JoinSet};

entities/src/models.rs

@@ -67,6 +67,12 @@ pub struct Creator {
     pub creator_share: u8,
 }
 
+#[derive(Serialize, Deserialize, Debug, Clone, Default)]
+pub struct OffChainData {
+    pub url: String,
+    pub metadata: String,
+}
+
 #[derive(Serialize, Deserialize, Debug, Clone, Default)]
 pub struct CompleteAssetDetails {
     // From AssetStaticDetails
@@ -92,6 +98,13 @@ pub struct CompleteAssetDetails {
     pub lamports: Option<Updated<u64>>,
     pub executable: Option<Updated<bool>>,
     pub metadata_owner: Option<Updated<String>>,
+    pub raw_name: Option<Updated<String>>,
+    pub plugins: Option<Updated<String>>,
+    pub unknown_plugins: Option<Updated<String>>,
+    pub rent_epoch: Option<Updated<u64>>,
+    pub num_minted: Option<Updated<u32>>,
+    pub current_size: Option<Updated<u32>>,
+    pub plugins_json_version: Option<Updated<u32>>,
 
     // From AssetAuthority as Tuple
     pub authority: Updated<Pubkey>,
@@ -113,6 +126,9 @@ pub struct CompleteAssetDetails {
     // TokenMetadataEdition
     pub edition: Option<EditionV1>,
     pub master_edition: Option<MasterEdition>,
+
+    // OffChainData
+    pub offchain_data: Option<OffChainData>,
 }
 
 /// Leaf information about compressed asset

grpc/Cargo.toml

@@ -12,9 +12,11 @@ futures = "0.3.29"
 solana-sdk = "~1.17"
 interface = { path = "../interface" }
 entities = { path = "../entities" }
+tokio = { version = "1.35.1", features = ["full"] }
+async-trait = "0.1.74"
+thiserror = "1.0.31"
 
 [dev-dependencies]
-tokio = { version = "1.35.1", features = ["full"] }
 mockall = "0.12.0"
 
 [build-dependencies]

grpc/proto/gap_filler.proto

@@ -58,6 +58,16 @@ enum UseMethod {
     SINGLE = 2;
 }
 
+enum UpdateVersion {
+    SEQUENCE = 0;
+    WRITE_VERSION = 1;
+}
+
+message UpdateVersionValue {
+    UpdateVersion type = 1;
+    uint64 value = 2;
+}
+
 message Uses {
     UseMethod use_method = 1;
     uint64 remaining = 2;
@@ -71,7 +81,7 @@ message ChainDataV1 {
     bool primary_sale_happened = 4;
     TokenStandard token_standard = 5;
     Uses uses = 6;
-    google.protobuf.UInt64Value seq_updated = 7;
+    UpdateVersionValue update_version = 7;
     uint64 slot_updated = 8;
 }
 
@@ -82,15 +92,15 @@ message AssetLeaf {
     google.protobuf.BytesValue data_hash = 4;
     google.protobuf.BytesValue creator_hash = 5;
     google.protobuf.UInt64Value leaf_seq = 6;
-    google.protobuf.UInt64Value seq_updated = 7;
+    UpdateVersionValue update_version = 7;
     uint64 slot_updated = 8;
 }
 
 message AssetCollection {
     bytes collection = 1;
     bool is_collection_verified = 2;
     google.protobuf.UInt64Value collection_seq = 3;
-    google.protobuf.UInt64Value seq_updated = 4;
+    UpdateVersionValue update_version = 4;
     uint64 slot_updated = 5;
 }
 
@@ -127,65 +137,74 @@ message AssetDetails {
     DynamicUint64Field lamports = 21;
     DynamicBoolField executable = 22;
     DynamicStringField metadata_owner = 23;
-
-    AssetLeaf asset_leaf = 24;
-    AssetCollection collection = 25;
-    ChainDataV1 chain_data = 26;
-
-    ClLeaf cl_leaf = 27;
-    repeated ClItem cl_items = 28;
+    DynamicStringField url = 24;
+    DynamicStringField raw_name = 25;
+    DynamicStringField plugins = 26;
+    DynamicStringField unknown_plugins = 27;
+    DynamicUint64Field rent_epoch = 28;
+    DynamicUint32Field num_minted = 29;
+    DynamicUint32Field current_size = 30;
+    DynamicUint32Field plugins_json_version = 31;
+
+    AssetLeaf asset_leaf = 32;
+    AssetCollection collection = 33;
+    ChainDataV1 chain_data = 34;
+
+    ClLeaf cl_leaf = 35;
+    repeated ClItem cl_items = 36;
 
     // From TokenMetadataEdition
-    EditionV1 edition = 29;
-    MasterEdition master_edition = 30;
+    EditionV1 edition = 37;
+    MasterEdition master_edition = 38;
+    OffchainData offchain_data = 39;
 }
 
 // Dynamic field messages
 message DynamicBoolField {
     bool value = 1;
-    google.protobuf.UInt64Value seq_updated = 2;
+    UpdateVersionValue update_version = 2;
     uint64 slot_updated = 3;
 }
 
 message DynamicUint64Field {
     uint64 value = 1;
-    google.protobuf.UInt64Value seq_updated = 2;
+    UpdateVersionValue update_version = 2;
     uint64 slot_updated = 3;
 }
 
 message DynamicUint32Field {
     uint32 value = 1;
-    google.protobuf.UInt64Value seq_updated = 2;
+    UpdateVersionValue update_version = 2;
     uint64 slot_updated = 3;
 }
 
 message DynamicBytesField {
     bytes value = 1;
-    google.protobuf.UInt64Value seq_updated = 2;
+    UpdateVersionValue update_version = 2;
     uint64 slot_updated = 3;
 }
 
 message DynamicStringField {
     string value = 1;
-    google.protobuf.UInt64Value seq_updated = 2;
+    UpdateVersionValue update_version = 2;
     uint64 slot_updated = 3;
 }
 
 message DynamicChainMutability {
     ChainMutability value = 1;
-    google.protobuf.UInt64Value seq_updated = 2;
+    UpdateVersionValue update_version = 2;
     uint64 slot_updated = 3;
 }
 
 message DynamicEnumField {
     OwnerType value = 1;
-    google.protobuf.UInt64Value seq_updated = 2;
+    UpdateVersionValue update_version = 2;
     uint64 slot_updated = 3;
 }
 
 message DynamicCreatorsField {
     repeated Creator creators = 1;
-    google.protobuf.UInt64Value seq_updated = 2;
+    UpdateVersionValue update_version = 2;
     uint64 slot_updated = 3;
 }
 
@@ -219,6 +238,11 @@ message MasterEdition {
     uint64 write_version = 4;
 }
 
+message OffchainData {
+    string url = 1;
+    string metadata = 2;
+}
+
 // RangeRequest and AssetDetailsResponse for data synchronization
 message RangeRequest {
     uint64 start_slot = 1;

grpc/src/client.rs

@@ -0,0 +1,56 @@
+use crate::error::GrpcError;
+use crate::gapfiller::gap_filler_service_client::GapFillerServiceClient;
+use crate::gapfiller::RangeRequest;
+use async_trait::async_trait;
+use futures::StreamExt;
+use interface::asset_streaming_and_discovery::{
+    AssetDetailsConsumer, AssetDetailsStreamNonSync, AsyncError, PeerDiscovery,
+};
+use std::str::FromStr;
+use tonic::transport::{Channel, Uri};
+use tonic::{Code, Status};
+
+pub struct Client {
+    inner: GapFillerServiceClient<Channel>,
+}
+
+impl Client {
+    pub async fn connect(peer_discovery: impl PeerDiscovery) -> Result<Self, GrpcError> {
+        let url = Uri::from_str(peer_discovery.get_gapfiller_peer_addr().as_str())
+            .map_err(|e| GrpcError::UriCreate(e.to_string()))?;
+        let channel = Channel::builder(url).connect().await?;
+
+        Ok(Self {
+            inner: GapFillerServiceClient::new(channel),
+        })
+    }
+}
+
+#[async_trait]
+impl AssetDetailsConsumer for Client {
+    async fn get_consumable_stream_in_range(
+        &mut self,
+        start_slot: u64,
+        end_slot: u64,
+    ) -> Result<AssetDetailsStreamNonSync, AsyncError> {
+        Ok(Box::pin(
+            self.inner
+                .get_assets_updated_within(RangeRequest {
+                    start_slot,
+                    end_slot,
+                })
+                .await
+                .map_err(|e| Box::new(e) as AsyncError)?
+                .into_inner()
+                .map(|stream| {
+                    stream
+                        .and_then(|asset_details| {
+                            asset_details
+                                .try_into()
+                                .map_err(|e: GrpcError| Status::new(Code::Internal, e.to_string()))
+                        })
+                        .map_err(|e| Box::new(e) as AsyncError)
+                }),
+        ))
+    }
+}

grpc/src/error/mod.rs

@@ -0,0 +1,21 @@
+use thiserror::Error;
+use tonic::transport::Error;
+
+#[derive(Error, Debug, PartialEq, Eq)]
+pub enum GrpcError {
+    #[error("Pubkey from: {0:?}")]
+    PubkeyFrom(Vec<u8>),
+    #[error("Missing field: {0}")]
+    MissingField(String),
+    #[error("Cannot cast enum: {0} {1}")]
+    EnumCast(String, String),
+    #[error("UriCreate: {0}")]
+    UriCreate(String),
+    #[error("TonicTransport: {0}")]
+    TonicTransport(String),
+}
+impl From<tonic::transport::Error> for GrpcError {
+    fn from(value: Error) -> Self {
+        Self::TonicTransport(value.to_string())
+    }
+}