Fog Ledger Router Admin service (#2920)

* Fog Ledger Router Admin service

* Sort itertools properly in Cargo.toml

* Fix redundant mod declaration

fog/ledger/server/Cargo.toml

@@ -50,6 +50,7 @@ clap = { version = "4.0", features = ["derive", "env"] }
 displaydoc = { version = "0.2", default-features = false }
 futures = "0.3"
 grpcio = "0.11.0"
+itertools = "0.10"
 lazy_static = "1.4"
 rand = "0.8"
 retry = "2.0"

fog/ledger/server/src/lib.rs

@@ -8,6 +8,7 @@ mod error;
 mod key_image_router_service;
 mod key_image_service;
 mod merkle_proof_service;
+mod router_admin_service;
 mod router_handlers;
 mod server;
 mod untrusted_tx_out_service;

fog/ledger/server/src/router_admin_service.rs

@@ -0,0 +1,85 @@
+// Copyright (c) 2018-2022 The MobileCoin Foundation
+
+use grpcio::{ChannelBuilder, RpcContext, RpcStatus, UnarySink};
+use itertools::Itertools;
+use mc_common::logger::{log, Logger};
+use mc_fog_api::{
+    fog_common::AddShardRequest,
+    ledger_grpc::{KeyImageStoreApiClient, LedgerRouterAdminApi},
+};
+use mc_fog_uri::KeyImageStoreUri;
+use mc_util_grpc::{
+    rpc_invalid_arg_error, rpc_logger, rpc_precondition_error, send_result,
+    ConnectionUriGrpcioChannel, Empty,
+};
+use mc_util_metrics::SVC_COUNTERS;
+use std::{
+    collections::HashMap,
+    str::FromStr,
+    sync::{Arc, RwLock},
+};
+
+#[derive(Clone)]
+pub struct LedgerRouterAdminService {
+    shard_clients: Arc<RwLock<HashMap<KeyImageStoreUri, Arc<KeyImageStoreApiClient>>>>,
+    logger: Logger,
+}
+
+impl LedgerRouterAdminService {
+    #[allow(dead_code)]
+    pub fn new(
+        shard_clients: Arc<RwLock<HashMap<KeyImageStoreUri, Arc<KeyImageStoreApiClient>>>>,
+        logger: Logger,
+    ) -> Self {
+        Self {
+            shard_clients,
+            logger,
+        }
+    }
+
+    fn add_shard_impl(&mut self, shard_uri: &str, logger: &Logger) -> Result<Empty, RpcStatus> {
+        let key_image_store_uri = KeyImageStoreUri::from_str(shard_uri).map_err(|_| {
+            rpc_invalid_arg_error(
+                "add_shard",
+                format!("Shard uri string {} is invalid", shard_uri),
+                logger,
+            )
+        })?;
+        let mut shard_clients = self.shard_clients.write().expect("RwLock Poisoned");
+        if shard_clients.keys().contains(&key_image_store_uri) {
+            let error = rpc_precondition_error(
+                "add_shard",
+                format!("Shard uri {} already exists in the shard list", shard_uri),
+                logger,
+            );
+            return Err(error);
+        }
+        let grpc_env = Arc::new(
+            grpcio::EnvBuilder::new()
+                .name_prefix("add-shard".to_string())
+                .build(),
+        );
+        let key_image_store_client = KeyImageStoreApiClient::new(
+            ChannelBuilder::default_channel_builder(grpc_env)
+                .connect_to_uri(&key_image_store_uri, logger),
+        );
+        shard_clients.insert(key_image_store_uri, Arc::new(key_image_store_client));
+
+        Ok(Empty::new())
+    }
+}
+
+impl LedgerRouterAdminApi for LedgerRouterAdminService {
+    fn add_shard(&mut self, ctx: RpcContext, request: AddShardRequest, sink: UnarySink<Empty>) {
+        log::info!(self.logger, "Request received in add_shard fn");
+        let _timer = SVC_COUNTERS.req(&ctx);
+        mc_common::logger::scoped_global_logger(&rpc_logger(&ctx, &self.logger), |logger| {
+            send_result(
+                ctx,
+                sink,
+                self.add_shard_impl(request.get_shard_uri(), logger),
+                logger,
+            );
+        });
+    }
+}