feat: add storage to validator node (#3551)

Add state and chain storage to the validator node.

Still a while to go, but this should be a good start

.gitignore

@@ -44,3 +44,4 @@ node_modules
 
 # ignore output files from windows ISS
 buildtools/Output/
+/applications/tari_collectibles/src-tauri/data

Cargo.toml

@@ -14,7 +14,7 @@ members = [
     "comms/dht",
     "comms/rpc_macros",
     "dan_layer/core",
-    "dan_layer/validator_node_sqlite",
+    "dan_layer/storage_sqlite",
     "infrastructure/shutdown",
     "infrastructure/storage",
     "infrastructure/test_utils",

applications/tari_app_grpc/build.rs

@@ -3,6 +3,13 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
         .build_client(true)
         .build_server(true)
         .format(false)
-        .compile(&["proto/base_node.proto", "proto/wallet.proto"], &["proto"])?;
+        .compile(
+            &[
+                "proto/base_node.proto",
+                "proto/wallet.proto",
+                "proto/validator_node.proto",
+            ],
+            &["proto"],
+        )?;
     Ok(())
 }

applications/tari_app_grpc/proto/base_node.proto

@@ -21,7 +21,6 @@
 // USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 syntax = "proto3";
 
-
 import "types.proto";
 
 package tari.rpc;
@@ -85,6 +84,19 @@ service BaseNode {
 
     rpc GetTokens(GetTokensRequest) returns (stream GetTokensResponse);
     rpc ListAssetRegistrations(ListAssetRegistrationsRequest) returns (stream ListAssetRegistrationsResponse);
+    rpc GetAssetMetadata(GetAssetMetadataRequest) returns (GetAssetMetadataResponse);
+}
+
+message GetAssetMetadataRequest {
+    bytes asset_public_key = 1;
+}
+
+message GetAssetMetadataResponse {
+    optional string name = 2;
+    optional string description =3;
+    optional string image = 4;
+    bytes owner_commitment = 5;
+    OutputFeatures features = 6;
 }
 
 message ListAssetRegistrationsRequest {

applications/tari_validator_node/proto/grpc/validator_node.proto → applications/tari_app_grpc/proto/validator_node.proto

@@ -21,7 +21,8 @@
 // USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 syntax = "proto3";
 
-package tari.validator_node.rpc;
+import "types.proto";
+package tari.rpc;
 
 service ValidatorNode {
     rpc GetMetadata(GetMetadataRequest) returns (GetMetadataResponse);

applications/tari_base_node/src/grpc/base_node_grpc_server.rs

@@ -47,7 +47,7 @@ use tari_core::{
         StateMachineHandle,
     },
     blocks::{Block, BlockHeader, NewBlockTemplate},
-    chain_storage::ChainStorageError,
+    chain_storage::{ChainStorageError, PrunedOutput},
     consensus::{emission::Emission, ConsensusManager, NetworkConsensus},
     iterators::NonOverlappingIntegerPairIter,
     mempool::{service::LocalMempoolService, TxStorageResponse},
@@ -464,6 +464,69 @@ impl tari_rpc::base_node_server::BaseNode for BaseNodeGrpcServer {
         Ok(Response::new(rx))
     }
 
+    async fn get_asset_metadata(
+        &self,
+        request: Request<tari_rpc::GetAssetMetadataRequest>,
+    ) -> Result<Response<tari_rpc::GetAssetMetadataResponse>, Status> {
+        let request = request.into_inner();
+
+        let mut handler = self.node_service.clone();
+        let metadata = handler
+            .get_asset_metadata(
+                PublicKey::from_bytes(&request.asset_public_key)
+                    .map_err(|e| Status::invalid_argument("Not a valid asset public key"))?,
+            )
+            .await
+            .map_err(|e| Status::internal(e.to_string()))?;
+
+        if let Some(m) = metadata {
+            match m.output {
+                PrunedOutput::Pruned {
+                    output_hash,
+                    witness_hash,
+                } => return Err(Status::not_found("Output has been pruned")),
+                PrunedOutput::NotPruned { output } => {
+                    if let Some(ref asset) = output.features.asset {
+                        const ASSET_METADATA_TEMPLATE_ID: u32 = 1;
+                        if asset.template_ids_implemented.contains(&ASSET_METADATA_TEMPLATE_ID) {
+                            // TODO: move to a better location, or better yet, have the grpc caller split the metadata
+                            let m = String::from_utf8(output.features.metadata.clone()).unwrap();
+                            let mut m = m
+                                .as_str()
+                                .split('|')
+                                .map(|s| s.to_string())
+                                .collect::<Vec<String>>()
+                                .into_iter();
+                            let name = m.next();
+                            let description = m.next();
+                            let image = m.next();
+
+                            // TODO Perhaps this should just return metadata and have the client read the metadata in a
+                            // pattern described by the template
+                            return Ok(Response::new(tari_rpc::GetAssetMetadataResponse {
+                                name,
+                                description,
+                                image,
+                                owner_commitment: Vec::from(output.commitment.as_bytes()),
+                                features: Some(output.features.clone().into()),
+                            }));
+                        }
+                    }
+                    return Ok(Response::new(tari_rpc::GetAssetMetadataResponse {
+                        name: None,
+                        description: None,
+                        image: None,
+                        owner_commitment: Vec::from(output.commitment.as_bytes()),
+                        features: Some(output.features.clone().into()),
+                    }));
+                },
+            };
+            Err(Status::unknown("Could not find a matching arm"))
+        } else {
+            Err(Status::not_found("Could not find any utxo"))
+        }
+    }
+
     async fn list_asset_registrations(
         &self,
         request: Request<tari_rpc::ListAssetRegistrationsRequest>,

applications/tari_collectibles/src-tauri/Cargo.toml

@@ -28,6 +28,10 @@ serde_json = "1.0"
 serde = { version = "1.0", features = ["derive"] }
 tonic = "0.5.2"
 tauri = { version = "1.0.0-beta.8", features = ["api-all"] }
+diesel = { version = "1.4.8", features = ["sqlite"] }
+diesel_migrations = "1.4.0"
+thiserror = "1.0.30"
+uuid = {  version = "0.8.2", features = ["serde"] }
 
 [features]
 default = [ "custom-protocol" ]

applications/tari_collectibles/src-tauri/migrations/2021-11-10-113031_init_schema/down.sql

@@ -0,0 +1 @@
+-- lol, no

applications/tari_collectibles/src-tauri/migrations/2021-11-10-113031_init_schema/up.sql

@@ -0,0 +1,9 @@
+create table accounts (
+   id blob not null primary key,
+   asset_public_key blob not null unique,
+   name text,
+   description text,
+   image text,
+   committee_length integer not null,
+   committee_pub_keys blob not null
+);

applications/tari_collectibles/src-tauri/src/app_state.rs

@@ -20,12 +20,22 @@
 //  WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
 //  USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
-use crate::{base_node_client::BaseNodeClient, settings::Settings, wallet_client::WalletClient};
+use crate::{
+  clients::{BaseNodeClient, GrpcValidatorNodeClient, ValidatorNodeClient, WalletClient},
+  settings::Settings,
+  storage::{
+    sqlite::{SqliteCollectiblesStorage, SqliteDbFactory},
+    StorageError,
+  },
+};
+use diesel::SqliteConnection;
 use std::sync::Arc;
+use tari_common_types::types::PublicKey;
 use tauri::async_runtime::RwLock;
 
 pub struct AppState {
   config: Settings,
+  db_factory: SqliteDbFactory,
 }
 
 #[derive(Clone)]
@@ -35,9 +45,11 @@ pub struct ConcurrentAppState {
 
 impl ConcurrentAppState {
   pub fn new() -> Self {
+    let settings = Settings::new();
     Self {
       inner: Arc::new(RwLock::new(AppState {
-        config: Settings::new(),
+        db_factory: SqliteDbFactory::new(settings.data_dir.as_path()),
+        config: settings,
       })),
     }
   }
@@ -52,4 +64,22 @@ impl ConcurrentAppState {
       BaseNodeClient::connect(format!("http://{}", lock.config.base_node_grpc_address)).await?;
     Ok(client)
   }
+
+  pub async fn connect_validator_node_client(
+    &self,
+    _public_key: PublicKey,
+  ) -> Result<GrpcValidatorNodeClient, String> {
+    // todo: convert this GRPC to tari comms
+    let lock = self.inner.read().await;
+    let client = GrpcValidatorNodeClient::connect(format!(
+      "http://{}",
+      lock.config.validator_node_grpc_address
+    ))
+    .await?;
+    Ok(client)
+  }
+
+  pub async fn create_db(&self) -> Result<SqliteCollectiblesStorage, StorageError> {
+    self.inner.read().await.db_factory.create_db()
+  }
 }

applications/tari_collectibles/src-tauri/src/base_node_client.rs → applications/tari_collectibles/src-tauri/src/clients/base_node_client.rs

@@ -22,6 +22,8 @@
 
 use futures::StreamExt;
 use tari_app_grpc::tari_rpc as grpc;
+use tari_common_types::types::PublicKey;
+use tari_utilities::ByteArray;
 
 pub struct BaseNodeClient {
   client: grpc::base_node_client::BaseNodeClient<tonic::transport::Channel>,
@@ -63,6 +65,70 @@ impl BaseNodeClient {
     Ok(assets)
   }
 
+  pub async fn get_asset_metadata(
+    &mut self,
+    asset_public_key: &PublicKey,
+  ) -> Result<grpc::GetAssetMetadataResponse, String> {
+    let client = self.client_mut();
+    let request = grpc::GetAssetMetadataRequest {
+      asset_public_key: Vec::from(asset_public_key.as_bytes()),
+    };
+    dbg!(&request);
+    let response = client
+      .get_asset_metadata(request)
+      .await
+      .map(|response| response.into_inner())
+      .map_err(|s| format!("Could not get asset metadata: {}", s))?;
+    dbg!(&response);
+    Ok(response)
+  }
+
+  // TODO: probably can get the full checkpoint instead
+  pub async fn get_sidechain_committee(
+    &mut self,
+    asset_public_key: &PublicKey,
+  ) -> Result<Vec<PublicKey>, String> {
+    let client = self.client_mut();
+    let request = grpc::GetTokensRequest {
+      asset_public_key: Vec::from(asset_public_key.as_bytes()),
+      unique_ids: vec![vec![3u8; 32]],
+    };
+
+    dbg!(&request);
+    let mut stream = client
+      .get_tokens(request)
+      .await
+      .map(|response| response.into_inner())
+      .map_err(|s| format!("Could not get asset sidechain checkpoint"))?;
+    let mut i = 0;
+    // Could def do this better
+    #[allow(clippy::never_loop)]
+    while let Some(response) = stream.next().await {
+      i += 1;
+      if i > 10 {
+        break;
+      }
+      dbg!(&response);
+      let features = response
+        .map_err(|status| format!("Got an error status from GRPC:{}", status))?
+        .features;
+      if let Some(sidechain) = features.and_then(|f| f.sidechain_checkpoint) {
+        let pub_keys = sidechain
+          .committee
+          .iter()
+          .map(|s| PublicKey::from_bytes(s).map_err(|e| format!("Not a valid public key:{}", e)))
+          .collect::<Result<_, String>>()?;
+        return Ok(pub_keys);
+      } else {
+        return Err("Found utxo but was missing sidechain data".to_string());
+      }
+    }
+    Err(format!(
+      "No side chain tokens were found out of {} streamed",
+      i
+    ))
+  }
+
   fn client_mut(
     &mut self,
   ) -> &mut grpc::base_node_client::BaseNodeClient<tonic::transport::Channel> {