Bug: Interfacing of client and gRPC communications to Dynamo



 Building a gRPC Client for communicating with the Dynamo server, once the dynamic routing feature has been enabled.

The dynamic routing feature  would be a compile time as well as runtime feature flag ,
 for providing the flexibility to enable and disable the dynamic routing during deployment

#### Possible Implementation:
Once dynamic routing has been enabled , we will store the default config in the algorithm column in routing table as the following jsonb:


```
{
    "id": "merchant_id", 
    "params": "card", 
    "labels": ["stripe","adyen"],
    "config": {
        "min_aggregates_size": 2,
        "default_success_rate": 100.0
    }
}
```

gRPC client will be integrated as a dynamic dispatch field of the DynamicRouting trait within the struct  SessionState, 
which would be under the dynamic_routing feature flag


```
struct SessionState
{

pub grpc: Box<dyn DynamicRouting>
}
```

The DynamicRouting trait would have following  functions: 
calculate_the_success_rate 
update_the_success_rates


```
triat DynamicRouting
{
	async fn calculate_success_rate(self,label_input : Vec<T>) -> CalSuccessRateResponse

	async fn update_success_rate(self,calculated_success_rate:CalSuccessRateResponse) -> UpdateSuccessRateWindowResponse
}
```

The trait DynamicRouting would be implemented on the the struct SuccessRateCalculator taken from the .proto file


```
impl DynamicRouting for SuccessRateCalculator
{
	async fn calculate_success_rate(self,label_input : T) ->     CalSuccessRateResponse
	{
		self.fetch_success_rate(request)
	}

	async fn update_success_rate(self,calculated_success_rate :  CalSuccessRateResponse) -> UpdateSuccessRateWindowResponse
  	{

   		self.update_success_rate_window(request)
   	}

}
```


diff --git a/Cargo.lock b/Cargo.lock
index 11036b42dc4..d90c577152d 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1433,7 +1433,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3b829e4e32b91e643de6eafe82b1d90675f5874230191a4ffbc1b336dec4d6bf"
 dependencies = [
  "async-trait",
- "axum-core",
+ "axum-core 0.3.4",
  "bitflags 1.3.2",
  "bytes 1.7.1",
  "futures-util",
@@ -1454,6 +1454,33 @@ dependencies = [
  "tower-service",
 ]
 
+[[package]]
+name = "axum"
+version = "0.7.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3a6c9af12842a67734c9a2e355436e5d03b22383ed60cf13cd0c18fbfe3dcbcf"
+dependencies = [
+ "async-trait",
+ "axum-core 0.4.3",
+ "bytes 1.7.1",
+ "futures-util",
+ "http 1.1.0",
+ "http-body 1.0.1",
+ "http-body-util",
+ "itoa",
+ "matchit",
+ "memchr",
+ "mime",
+ "percent-encoding",
+ "pin-project-lite",
+ "rustversion",
+ "serde",
+ "sync_wrapper 1.0.1",
+ "tower",
+ "tower-layer",
+ "tower-service",
+]
+
 [[package]]
 name = "axum-core"
 version = "0.3.4"
@@ -1471,6 +1498,26 @@ dependencies = [
  "tower-service",
 ]
 
+[[package]]
+name = "axum-core"
+version = "0.4.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a15c63fd72d41492dc4f497196f5da1fb04fb7529e631d73630d1b491e47a2e3"
+dependencies = [
+ "async-trait",
+ "bytes 1.7.1",
+ "futures-util",
+ "http 1.1.0",
+ "http-body 1.0.1",
+ "http-body-util",
+ "mime",
+ "pin-project-lite",
+ "rustversion",
+ "sync_wrapper 0.1.2",
+ "tower-layer",
+ "tower-service",
+]
+
 [[package]]
 name = "backtrace"
 version = "0.3.73"
@@ -3064,6 +3111,7 @@ dependencies = [
 name = "external_services"
 version = "0.1.0"
 dependencies = [
+ "api_models",
  "async-trait",
  "aws-config 0.55.3",
  "aws-sdk-kms",
@@ -3081,10 +3129,15 @@ dependencies = [
  "hyperswitch_interfaces",
  "masking",
  "once_cell",
+ "prost 0.13.2",
  "router_env",
  "serde",
  "thiserror",
  "tokio 1.40.0",
+ "tonic 0.12.2",
+ "tonic-build",
+ "tonic-reflection",
+ "tonic-types",
  "vaultrs",
 ]
 
@@ -3160,6 +3213,12 @@ version = "0.2.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "28dea519a9695b9977216879a3ebfddf92f1c08c05d984f8996aecd6ecdc811d"
 
+[[package]]
+name = "fixedbitset"
+version = "0.4.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0ce7134b9999ecaf8bcd65542e436736ef32ddca1b3e06094cb6ec5755203b80"
+
 [[package]]
 name = "flate2"
 version = "1.0.33"
@@ -3854,6 +3913,19 @@ dependencies = [
  "tokio-io-timeout",
 ]
 
+[[package]]
+name = "hyper-timeout"
+version = "0.5.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3203a961e5c83b6f5498933e78b6b263e208c197b63e9c6c53cc82ffd3f63793"
+dependencies = [
+ "hyper 1.4.1",
+ "hyper-util",
+ "pin-project-lite",
+ "tokio 1.40.0",
+ "tower-service",
+]
+
 [[package]]
 name = "hyper-tls"
 version = "0.5.0"
@@ -4727,6 +4799,12 @@ dependencies = [
  "uuid",
 ]
 
+[[package]]
+name = "multimap"
+version = "0.10.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "defc4c55412d89136f966bbb339008b474350e5e6e78d2714439c386b3137a03"
+
 [[package]]
 name = "mutually_exclusive_features"
 version = "0.0.3"
@@ -5114,10 +5192,10 @@ dependencies = [
  "http 0.2.12",
  "opentelemetry",
  "opentelemetry-proto",
- "prost",
+ "prost 0.11.9",
  "thiserror",
  "tokio 1.40.0",
- "tonic",
+ "tonic 0.8.3",
 ]
 
 [[package]]
@@ -5129,8 +5207,8 @@ dependencies = [
  "futures 0.3.30",
  "futures-util",
  "opentelemetry",
- "prost",
- "tonic",
+ "prost 0.11.9",
+ "tonic 0.8.3",
 ]
 
 [[package]]
@@ -5389,6 +5467,16 @@ dependencies = [
  "sha2",
 ]
 
+[[package]]
+name = "petgraph"
+version = "0.6.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b4c5cc86750666a3ed20bdaf5ca2a0344f9c67674cae0515bec2da16fbaa47db"
+dependencies = [
+ "fixedbitset",
+ "indexmap 2.5.0",
+]
+
 [[package]]
 name = "phf"
 version = "0.11.2"
@@ -5591,6 +5679,16 @@ dependencies = [
  "vcpkg",
 ]
 
+[[package]]
+name = "prettyplease"
+version = "0.2.20"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5f12335488a2f3b0a83b14edad48dca9879ce89b2edd10e80237e4e852dd645e"
+dependencies = [
+ "proc-macro2",
+ "syn 2.0.77",
+]
+
 [[package]]
 name = "primeorder"
 version = "0.13.6"
@@ -5679,7 +5777,38 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0b82eaa1d779e9a4bc1c3217db8ffbeabaae1dca241bf70183242128d48681cd"
 dependencies = [
  "bytes 1.7.1",
- "prost-derive",
+ "prost-derive 0.11.9",
+]
+
+[[package]]
+name = "prost"
+version = "0.13.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3b2ecbe40f08db5c006b5764a2645f7f3f141ce756412ac9e1dd6087e6d32995"
+dependencies = [
+ "bytes 1.7.1",
+ "prost-derive 0.13.2",
+]
+
+[[package]]
+name = "prost-build"
+version = "0.13.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f8650aabb6c35b860610e9cff5dc1af886c9e25073b7b1712a68972af4281302"
+dependencies = [
+ "bytes 1.7.1",
+ "heck 0.5.0",
+ "itertools 0.12.1",
+ "log",
+ "multimap",
+ "once_cell",
+ "petgraph",
+ "prettyplease",
+ "prost 0.13.2",
+ "prost-types",
+ "regex",
+ "syn 2.0.77",
+ "tempfile",
 ]
 
 [[package]]
@@ -5695,6 +5824,28 @@ dependencies = [
  "syn 1.0.109",
 ]
 
+[[package]]
+name = "prost-derive"
+version = "0.13.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "acf0c195eebb4af52c752bec4f52f645da98b6e92077a04110c7f349477ae5ac"
+dependencies = [
+ "anyhow",
+ "itertools 0.12.1",
+ "proc-macro2",
+ "quote",
+ "syn 2.0.77",
+]
+
+[[package]]
+name = "prost-types"
+version = "0.13.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "60caa6738c7369b940c3d49246a8d1749323674c65cb13010134f5c9bad5b519"
+dependencies = [
+ "prost 0.13.2",
+]
+
 [[package]]
 name = "ptr_meta"
 version = "0.1.4"
@@ -8215,7 +8366,7 @@ checksum = "8f219fad3b929bef19b1f86fbc0358d35daed8f2cac972037ac0dc10bbb8d5fb"
 dependencies = [
  "async-stream",
  "async-trait",
- "axum",
+ "axum 0.6.20",
  "base64 0.13.1",
  "bytes 1.7.1",
  "futures-core",
@@ -8224,11 +8375,11 @@ dependencies = [
  "http 0.2.12",
  "http-body 0.4.6",
  "hyper 0.14.30",
- "hyper-timeout",
+ "hyper-timeout 0.4.1",
  "percent-encoding",
  "pin-project",
- "prost",
- "prost-derive",
+ "prost 0.11.9",
+ "prost-derive 0.11.9",
  "tokio 1.40.0",
  "tokio-stream",
  "tokio-util",
@@ -8239,6 +8390,73 @@ dependencies = [
  "tracing-futures",
 ]
 
+[[package]]
+name = "tonic"
+version = "0.12.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c6f6ba989e4b2c58ae83d862d3a3e27690b6e3ae630d0deb59f3697f32aa88ad"
+dependencies = [
+ "async-stream",
+ "async-trait",
+ "axum 0.7.5",
+ "base64 0.22.1",
+ "bytes 1.7.1",
+ "h2 0.4.6",
+ "http 1.1.0",
+ "http-body 1.0.1",
+ "http-body-util",
+ "hyper 1.4.1",
+ "hyper-timeout 0.5.1",
+ "hyper-util",
+ "percent-encoding",
+ "pin-project",
+ "prost 0.13.2",
+ "socket2",
+ "tokio 1.40.0",
+ "tokio-stream",
+ "tower",
+ "tower-layer",
+ "tower-service",
+ "tracing",
+]
+
+[[package]]
+name = "tonic-build"
+version = "0.12.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fe4ee8877250136bd7e3d2331632810a4df4ea5e004656990d8d66d2f5ee8a67"
+dependencies = [
+ "prettyplease",
+ "proc-macro2",
+ "prost-build",
+ "quote",
+ "syn 2.0.77",
+]
+
+[[package]]
+name = "tonic-reflection"
+version = "0.12.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7b56b874eedb04f89907573b408eab1e87c1c1dce43aac6ad63742f57faa99ff"
+dependencies = [
+ "prost 0.13.2",
+ "prost-types",
+ "tokio 1.40.0",
+ "tokio-stream",
+ "tonic 0.12.2",
+]
+
+[[package]]
+name = "tonic-types"
+version = "0.12.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9d967793411bc1a5392accf4731114295f0fd122865d22cde46a8584b03402b2"
+dependencies = [
+ "prost 0.13.2",
+ "prost-types",
+ "tonic 0.12.2",
+]
+
 [[package]]
 name = "totp-rs"
 version = "5.6.0"
diff --git a/Dockerfile b/Dockerfile
index 29c1003c115..65ca6370364 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -4,7 +4,7 @@ ARG EXTRA_FEATURES=""
 ARG VERSION_FEATURE_SET="v1"
 
 RUN apt-get update \
-    && apt-get install -y libpq-dev libssl-dev pkg-config
+    && apt-get install -y libpq-dev libssl-dev pkg-config protobuf-compiler
 
 # Copying codebase from current dir to /router dir
 # and creating a fresh build
diff --git a/crates/api_models/src/routing.rs b/crates/api_models/src/routing.rs
index 5023d30f722..2b551eaebb7 100644
--- a/crates/api_models/src/routing.rs
+++ b/crates/api_models/src/routing.rs
@@ -197,7 +197,9 @@ pub enum RoutableChoiceSerde {
 impl std::fmt::Display for RoutableConnectorChoice {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         let base = self.connector.to_string();
-
+        if let Some(mca_id) = &self.merchant_connector_id {
+            return write!(f, "{}:{}", base, mca_id.get_string_repr());
+        }
         write!(f, "{}", base)
     }
 }
@@ -252,6 +254,11 @@ impl From<RoutableConnectorChoice> for RoutableChoiceSerde {
     }
 }
 
+#[derive(serde::Serialize, serde::Deserialize, Debug, Clone)]
+pub struct RoutableConnectorChoiceWithStatus {
+    pub routable_connector_choice: RoutableConnectorChoice,
+    pub status: bool,
+}
 #[derive(Debug, Copy, Clone, serde::Serialize, serde::Deserialize, strum::Display, ToSchema)]
 #[serde(rename_all = "snake_case")]
 #[strum(serialize_all = "snake_case")]
@@ -565,7 +572,7 @@ impl Default for SuccessBasedRoutingConfig {
     }
 }
 
-#[derive(serde::Serialize, serde::Deserialize, Debug, Clone, ToSchema)]
+#[derive(serde::Serialize, serde::Deserialize, Debug, Clone, ToSchema, strum::Display)]
 pub enum SuccessBasedRoutingConfigParams {
     PaymentMethod,
     PaymentMethodType,
diff --git a/crates/external_services/Cargo.toml b/crates/external_services/Cargo.toml
index 9051c58ea59..a882cdc9f51 100644
--- a/crates/external_services/Cargo.toml
+++ b/crates/external_services/Cargo.toml
@@ -13,6 +13,7 @@ email = ["dep:aws-config"]
 aws_s3 = ["dep:aws-config", "dep:aws-sdk-s3"]
 hashicorp-vault = ["dep:vaultrs"]
 v1 = ["hyperswitch_interfaces/v1"]
+dynamic_routing = ["dep:prost", "dep:tonic", "dep:tonic-reflection", "dep:tonic-types", "dep:api_models", "tokio/macros", "tokio/rt-multi-thread"]
 
 [dependencies]
 async-trait = "0.1.79"
@@ -31,14 +32,24 @@ hyper-proxy = "0.9.1"
 once_cell = "1.19.0"
 serde = { version = "1.0.197", features = ["derive"] }
 thiserror = "1.0.58"
-tokio = "1.37.0"
 vaultrs = { version = "0.7.2", optional = true }
+prost = { version = "0.13", optional = true }
+tokio = "1.37.0" 
+tonic = { version = "0.12.2", optional = true }
+tonic-reflection = { version = "0.12.2", optional = true }
+tonic-types = { version = "0.12.2", optional = true }
+
 
 # First party crates
 common_utils = { version = "0.1.0", path = "../common_utils" }
 hyperswitch_interfaces = { version = "0.1.0", path = "../hyperswitch_interfaces", default-features = false }
 masking = { version = "0.1.0", path = "../masking" }
 router_env = { version = "0.1.0", path = "../router_env", features = ["log_extra_implicit_fields", "log_custom_entries_to_extra"] }
+api_models = { version = "0.1.0", path = "../api_models", optional = true }
+
+
+[build-dependencies]
+tonic-build = "0.12"
 
 [lints]
 workspace = true
diff --git a/crates/external_services/build.rs b/crates/external_services/build.rs
new file mode 100644
index 00000000000..61e19f308d8
--- /dev/null
+++ b/crates/external_services/build.rs
@@ -0,0 +1,15 @@
+use std::{env, path::PathBuf};
+
+#[allow(clippy::expect_used)]
+fn main() -> Result<(), Box<dyn std::error::Error>> {
+    // Get the directory of the current crate
+    let crate_dir = env::var("CARGO_MANIFEST_DIR")?;
+    let proto_file = PathBuf::from(crate_dir)
+        .join("..")
+        .join("..")
+        .join("proto")
+        .join("success_rate.proto");
+    // Compile the .proto file
+    tonic_build::compile_protos(proto_file).expect("Failed to compile protos ");
+    Ok(())
+}
diff --git a/crates/external_services/src/grpc_client.rs b/crates/external_services/src/grpc_client.rs
new file mode 100644
index 00000000000..5afd3024551
--- /dev/null
+++ b/crates/external_services/src/grpc_client.rs
@@ -0,0 +1,48 @@
+/// Dyanimc Routing Client interface implementation
+#[cfg(feature = "dynamic_routing")]
+pub mod dynamic_routing;
+use std::{fmt::Debug, sync::Arc};
+
+#[cfg(feature = "dynamic_routing")]
+use dynamic_routing::{DynamicRoutingClientConfig, RoutingStrategy};
+use router_env::logger;
+use serde;
+
+/// Struct contains all the gRPC Clients
+#[derive(Debug, Clone)]
+pub struct GrpcClients {
+    /// The routing client
+    #[cfg(feature = "dynamic_routing")]
+    pub dynamic_routing: RoutingStrategy,
+}
+/// Type that contains the configs required to construct a  gRPC client with its respective services.
+#[derive(Debug, Clone, serde::Deserialize, serde::Serialize, Default)]
+pub struct GrpcClientSettings {
+    #[cfg(feature = "dynamic_routing")]
+    /// Configs for Dynamic Routing Client
+    pub dynamic_routing_client: DynamicRoutingClientConfig,
+}
+
+impl GrpcClientSettings {
+    /// # Panics
+    ///
+    /// This function will panic if it fails to establish a connection with the gRPC server.
+    /// This function will be called at service startup.
+    #[allow(clippy::expect_used)]
+    pub async fn get_grpc_client_interface(&self) -> Arc<GrpcClients> {
+        #[cfg(feature = "dynamic_routing")]
+        let dynamic_routing_connection = self
+            .dynamic_routing_client
+            .clone()
+            .get_dynamic_routing_connection()
+            .await
+            .expect("Failed to establish a connection with the Dynamic Routing Server");
+
+        logger::info!("Connection established with gRPC Server");
+
+        Arc::new(GrpcClients {
+            #[cfg(feature = "dynamic_routing")]
+            dynamic_routing: dynamic_routing_connection,
+        })
+    }
+}
diff --git a/crates/external_services/src/grpc_client/dynamic_routing.rs b/crates/external_services/src/grpc_client/dynamic_routing.rs
new file mode 100644
index 00000000000..17bd43ca3a8
--- /dev/null
+++ b/crates/external_services/src/grpc_client/dynamic_routing.rs
@@ -0,0 +1,265 @@
+use std::fmt::Debug;
+
+use api_models::routing::{
+    CurrentBlockThreshold, RoutableConnectorChoice, RoutableConnectorChoiceWithStatus,
+    SuccessBasedRoutingConfig, SuccessBasedRoutingConfigBody,
+};
+use common_utils::{errors::CustomResult, ext_traits::OptionExt, transformers::ForeignTryFrom};
+use error_stack::ResultExt;
+use serde;
+use success_rate::{
+    success_rate_calculator_client::SuccessRateCalculatorClient, CalSuccessRateConfig,
+    CalSuccessRateRequest, CalSuccessRateResponse,
+    CurrentBlockThreshold as DynamicCurrentThreshold, LabelWithStatus,
+    UpdateSuccessRateWindowConfig, UpdateSuccessRateWindowRequest, UpdateSuccessRateWindowResponse,
+};
+use tonic::transport::Channel;
+#[allow(
+    missing_docs,
+    unused_qualifications,
+    clippy::unwrap_used,
+    clippy::as_conversions
+)]
+pub mod success_rate {
+    tonic::include_proto!("success_rate");
+}
+/// Result type for Dynamic Routing
+pub type DynamicRoutingResult<T> = CustomResult<T, DynamicRoutingError>;
+
+/// Dynamic Routing Errors
+#[derive(Debug, Clone, thiserror::Error)]
+pub enum DynamicRoutingError {
+    /// The required input is missing
+    #[error("Missing Required Field : {field} for building the Dynamic Routing Request")]
+    MissingRequiredField {
+        /// The required field name
+        field: String,
+    },
+    /// Error from Dynamic Routing Server
+    #[error("Error from Dynamic Routing Server : {0}")]
+    SuccessRateBasedRoutingFailure(String),
+}
+
+/// Type that consists of all the services provided by the client
+#[derive(Debug, Clone)]
+pub struct RoutingStrategy {
+    /// success rate service for Dynamic Routing
+    pub success_rate_client: Option<SuccessRateCalculatorClient<Channel>>,
+}
+
+/// Contains the Dynamic Routing Client Config
+#[derive(Debug, Clone, serde::Deserialize, serde::Serialize, Default)]
+#[serde(untagged)]
+pub enum DynamicRoutingClientConfig {
+    /// If the dynamic routing client config has been enabled
+    Enabled {
+        /// The host for the client
+        host: String,
+        /// The port of the client
+        port: u16,
+    },
+    #[default]
+    /// If the dynamic routing client config has been disabled
+    Disabled,
+}
+
+impl DynamicRoutingClientConfig {
+    /// establish connection with the server
+    pub async fn get_dynamic_routing_connection(
+        self,
+    ) -> Result<RoutingStrategy, Box<dyn std::error::Error>> {
+        let success_rate_client = match self {
+            Self::Enabled { host, port } => {
+                let uri = format!("http://{}:{}", host, port);
+                let channel = tonic::transport::Endpoint::new(uri)?.connect().await?;
+                Some(SuccessRateCalculatorClient::new(channel))
+            }
+            Self::Disabled => None,
+        };
+        Ok(RoutingStrategy {
+            success_rate_client,
+        })
+    }
+}
+
+/// The trait Success Based Dynamic Routing would have the functions required to support the calculation and updation window
+#[async_trait::async_trait]
+pub trait SuccessBasedDynamicRouting: dyn_clone::DynClone + Send + Sync {
+    /// To calculate the success rate for the list of chosen connectors
+    async fn calculate_success_rate(
+        &self,
+        id: String,
+        success_rate_based_config: SuccessBasedRoutingConfig,
+        label_input: Vec<RoutableConnectorChoice>,
+    ) -> DynamicRoutingResult<CalSuccessRateResponse>;
+    /// To update the success rate with the given label
+    async fn update_success_rate(
+        &self,
+        id: String,
+        success_rate_based_config: SuccessBasedRoutingConfig,
+        response: Vec<RoutableConnectorChoiceWithStatus>,
+    ) -> DynamicRoutingResult<UpdateSuccessRateWindowResponse>;
+}
+
+#[async_trait::async_trait]
+impl SuccessBasedDynamicRouting for SuccessRateCalculatorClient<Channel> {
+    async fn calculate_success_rate(
+        &self,
+        id: String,
+        success_rate_based_config: SuccessBasedRoutingConfig,
+        label_input: Vec<RoutableConnectorChoice>,
+    ) -> DynamicRoutingResult<CalSuccessRateResponse> {
+        let params = success_rate_based_config
+            .params
+            .map(|vec| {
+                vec.into_iter().fold(String::new(), |mut acc_vec, params| {
+                    if !acc_vec.is_empty() {
+                        acc_vec.push(':')
+                    }
+                    acc_vec.push_str(params.to_string().as_str());
+                    acc_vec
+                })
+            })
+            .get_required_value("params")
+            .change_context(DynamicRoutingError::MissingRequiredField {
+                field: "params".to_string(),
+            })?;
+
+        let labels = label_input
+            .into_iter()
+            .map(|conn_choice| conn_choice.to_string())
+            .collect::<Vec<_>>();
+
+        let config = success_rate_based_config
+            .config
+            .map(ForeignTryFrom::foreign_try_from)
+            .transpose()?;
+
+        let request = tonic::Request::new(CalSuccessRateRequest {
+            id,
+            params,
+            labels,
+            config,
+        });
+
+        let mut client = self.clone();
+
+        let response = client
+            .fetch_success_rate(request)
+            .await
+            .change_context(DynamicRoutingError::SuccessRateBasedRoutingFailure(
+                "Failed to fetch the success rate".to_string(),
+            ))?
+            .into_inner();
+
+        Ok(response)
+    }
+
+    async fn update_success_rate(
+        &self,
+        id: String,
+        success_rate_based_config: SuccessBasedRoutingConfig,
+        label_input: Vec<RoutableConnectorChoiceWithStatus>,
+    ) -> DynamicRoutingResult<UpdateSuccessRateWindowResponse> {
+        let config = success_rate_based_config
+            .config
+            .map(ForeignTryFrom::foreign_try_from)
+            .transpose()?;
+
+        let labels_with_status = label_input
+            .into_iter()
+            .map(|conn_choice| LabelWithStatus {
+                label: conn_choice.routable_connector_choice.to_string(),
+                status: conn_choice.status,
+            })
+            .collect();
+
+        let params = success_rate_based_config
+            .params
+            .map(|vec| {
+                vec.into_iter().fold(String::new(), |mut acc_vec, params| {
+                    if !acc_vec.is_empty() {
+                        acc_vec.push(':')
+                    }
+                    acc_vec.push_str(params.to_string().as_str());
+                    acc_vec
+                })
+            })
+            .get_required_value("params")
+            .change_context(DynamicRoutingError::MissingRequiredField {
+                field: "params".to_string(),
+            })?;
+
+        let request = tonic::Request::new(UpdateSuccessRateWindowRequest {
+            id,
+            params,
+            labels_with_status,
+            config,
+        });
+
+        let mut client = self.clone();
+
+        let response = client
+            .update_success_rate_window(request)
+            .await
+            .change_context(DynamicRoutingError::SuccessRateBasedRoutingFailure(
+                "Failed to update the success rate window".to_string(),
+            ))?
+            .into_inner();
+
+        Ok(response)
+    }
+}
+
+impl ForeignTryFrom<CurrentBlockThreshold> for DynamicCurrentThreshold {
+    type Error = error_stack::Report<DynamicRoutingError>;
+    fn foreign_try_from(current_threshold: CurrentBlockThreshold) -> Result<Self, Self::Error> {
+        Ok(Self {
+            duration_in_mins: current_threshold.duration_in_mins,
+            max_total_count: current_threshold
+                .max_total_count
+                .get_required_value("max_total_count")
+                .change_context(DynamicRoutingError::MissingRequiredField {
+                    field: "max_total_count".to_string(),
+                })?,
+        })
+    }
+}
+
+impl ForeignTryFrom<SuccessBasedRoutingConfigBody> for UpdateSuccessRateWindowConfig {
+    type Error = error_stack::Report<DynamicRoutingError>;
+    fn foreign_try_from(config: SuccessBasedRoutingConfigBody) -> Result<Self, Self::Error> {
+        Ok(Self {
+            max_aggregates_size: config
+                .max_aggregates_size
+                .get_required_value("max_aggregate_size")
+                .change_context(DynamicRoutingError::MissingRequiredField {
+                    field: "max_aggregates_size".to_string(),
+                })?,
+            current_block_threshold: config
+                .current_block_threshold
+                .map(ForeignTryFrom::foreign_try_from)
+                .transpose()?,
+        })
+    }
+}
+
+impl ForeignTryFrom<SuccessBasedRoutingConfigBody> for CalSuccessRateConfig {
+    type Error = error_stack::Report<DynamicRoutingError>;
+    fn foreign_try_from(config: SuccessBasedRoutingConfigBody) -> Result<Self, Self::Error> {
+        Ok(Self {
+            min_aggregates_size: config
+                .min_aggregates_size
+                .get_required_value("min_aggregate_size")
+                .change_context(DynamicRoutingError::MissingRequiredField {
+                    field: "min_aggregates_size".to_string(),
+                })?,
+            default_success_rate: config
+                .default_success_rate
+                .get_required_value("default_success_rate")
+                .change_context(DynamicRoutingError::MissingRequiredField {
+                    field: "default_success_rate".to_string(),
+                })?,
+        })
+    }
+}
diff --git a/crates/external_services/src/lib.rs b/crates/external_services/src/lib.rs
index f4bcd91e343..4570a5e5960 100644
--- a/crates/external_services/src/lib.rs
+++ b/crates/external_services/src/lib.rs
@@ -14,6 +14,9 @@ pub mod hashicorp_vault;
 
 pub mod no_encryption;
 
+/// Building grpc clients to communicate with the server
+pub mod grpc_client;
+
 pub mod managers;
 
 /// Crate specific constants
diff --git a/crates/hyperswitch_interfaces/src/api.rs b/crates/hyperswitch_interfaces/src/api.rs
index 0b4545a33d7..7205865f24a 100644
--- a/crates/hyperswitch_interfaces/src/api.rs
+++ b/crates/hyperswitch_interfaces/src/api.rs
@@ -16,7 +16,6 @@ pub mod payouts;
 pub mod payouts_v2;
 pub mod refunds;
 pub mod refunds_v2;
-
 use common_enums::enums::{CallConnectorAction, CaptureMethod, PaymentAction, PaymentMethodType};
 use common_utils::{
     errors::CustomResult,
diff --git a/crates/router/Cargo.toml b/crates/router/Cargo.toml
index bae6243d2e6..442561c2ca3 100644
--- a/crates/router/Cargo.toml
+++ b/crates/router/Cargo.toml
@@ -38,6 +38,7 @@ v1 = ["common_default", "api_models/v1", "diesel_models/v1", "hyperswitch_domain
 customer_v2 = ["api_models/customer_v2", "diesel_models/customer_v2", "hyperswitch_domain_models/customer_v2", "storage_impl/customer_v2"]
 payment_v2 = ["api_models/payment_v2", "diesel_models/payment_v2", "hyperswitch_domain_models/payment_v2", "storage_impl/payment_v2"]
 payment_methods_v2 = ["api_models/payment_methods_v2", "diesel_models/payment_methods_v2", "hyperswitch_domain_models/payment_methods_v2", "storage_impl/payment_methods_v2", "common_utils/payment_methods_v2"]
+dynamic_routing = ["external_services/dynamic_routing"]
 
 # Partial Auth
 # The feature reduces the overhead of the router authenticating the merchant for every request, and trusts on `x-merchant-id` header to be present in the request.
diff --git a/crates/router/src/configs/secrets_transformers.rs b/crates/router/src/configs/secrets_transformers.rs
index 9b0bfbb40b2..604dce5047c 100644
--- a/crates/router/src/configs/secrets_transformers.rs
+++ b/crates/router/src/configs/secrets_transformers.rs
@@ -497,6 +497,7 @@ pub(crate) async fn fetch_raw_secrets(
         user_auth_methods,
         decision: conf.decision,
         locker_based_open_banking_connectors: conf.locker_based_open_banking_connectors,
+        grpc_client: conf.grpc_client,
         recipient_emails: conf.recipient_emails,
         network_tokenization_supported_card_networks: conf
             .network_tokenization_supported_card_networks,
diff --git a/crates/router/src/configs/settings.rs b/crates/router/src/configs/settings.rs
index 78ff709cc02..7d6dfd26eae 100644
--- a/crates/router/src/configs/settings.rs
+++ b/crates/router/src/configs/settings.rs
@@ -13,6 +13,7 @@ use error_stack::ResultExt;
 use external_services::email::EmailSettings;
 use external_services::{
     file_storage::FileStorageConfig,
+    grpc_client::GrpcClientSettings,
     managers::{
         encryption_management::EncryptionManagementConfig,
         secrets_management::SecretsManagementConfig,
@@ -120,6 +121,7 @@ pub struct Settings<S: SecretState> {
     pub user_auth_methods: SecretStateContainer<UserAuthMethodSettings, S>,
     pub decision: Option<DecisionConfig>,
     pub locker_based_open_banking_connectors: LockerBasedRecipientConnectorList,
+    pub grpc_client: GrpcClientSettings,
     pub recipient_emails: RecipientMails,
     pub network_tokenization_supported_card_networks: NetworkTokenizationSupportedCardNetworks,
     pub network_tokenization_service: Option<SecretStateContainer<NetworkTokenizationService, S>>,
diff --git a/crates/router/src/core/payments.rs b/crates/router/src/core/payments.rs
index 0af5bf6037a..30d619efedc 100644
--- a/crates/router/src/core/payments.rs
+++ b/crates/router/src/core/payments.rs
@@ -11,7 +11,6 @@ pub mod routing;
 pub mod tokenization;
 pub mod transformers;
 pub mod types;
-
 #[cfg(feature = "olap")]
 use std::collections::HashMap;
 use std::{
@@ -160,7 +159,6 @@ where
             &header_payload,
         )
         .await?;
-
     utils::validate_profile_id_from_auth_layer(
         profile_id_from_auth_layer,
         &payment_data.get_payment_intent().clone(),
@@ -4427,6 +4425,7 @@ where
     )
     .await
     .change_context(errors::ApiErrorResponse::InternalServerError)?;
+
     let connectors = routing::perform_eligibility_analysis_with_fallback(
         &state.clone(),
         key_store,
diff --git a/crates/router/src/routes/admin.rs b/crates/router/src/routes/admin.rs
index a68f648e178..8ba214f94b9 100644
--- a/crates/router/src/routes/admin.rs
+++ b/crates/router/src/routes/admin.rs
@@ -361,7 +361,7 @@ pub async fn connector_retrieve(
     })
     .into_inner();
 
-    api::server_wrap(
+    Box::pin(api::server_wrap(
         flow,
         state,
         &req,
@@ -384,7 +384,7 @@ pub async fn connector_retrieve(
             req.headers(),
         ),
         api_locking::LockAction::NotApplicable,
-    )
+    ))
     .await
 }
 /// Merchant Connector - Retrieve
diff --git a/crates/router/src/routes/api_keys.rs b/crates/router/src/routes/api_keys.rs
index 71632cc5749..a00e740b3e6 100644
--- a/crates/router/src/routes/api_keys.rs
+++ b/crates/router/src/routes/api_keys.rs
@@ -166,7 +166,7 @@ pub async fn api_key_update(
     payload.key_id = key_id;
     payload.merchant_id.clone_from(&merchant_id);
 
-    api::server_wrap(
+    Box::pin(api::server_wrap(
         flow,
         state,
         &req,
@@ -182,7 +182,7 @@ pub async fn api_key_update(
             req.headers(),
         ),
         api_locking::LockAction::NotApplicable,
-    )
+    ))
     .await
 }
 
diff --git a/crates/router/src/routes/app.rs b/crates/router/src/routes/app.rs
index aad5acbbade..034b3784f40 100644
--- a/crates/router/src/routes/app.rs
+++ b/crates/router/src/routes/app.rs
@@ -9,7 +9,7 @@ use common_enums::TransactionType;
 use common_utils::crypto::Blake3;
 #[cfg(feature = "email")]
 use external_services::email::{ses::AwsSes, EmailService};
-use external_services::file_storage::FileStorageInterface;
+use external_services::{file_storage::FileStorageInterface, grpc_client::GrpcClients};
 use hyperswitch_interfaces::{
     encryption_interface::EncryptionManagementInterface,
     secrets_interface::secret_state::{RawSecret, SecuredSecret},
@@ -105,6 +105,7 @@ pub struct SessionState {
     pub tenant: Tenant,
     #[cfg(feature = "olap")]
     pub opensearch_client: Arc<OpenSearchClient>,
+    pub grpc_client: Arc<GrpcClients>,
 }
 impl scheduler::SchedulerSessionState for SessionState {
     fn get_db(&self) -> Box<dyn SchedulerInterface> {
@@ -202,6 +203,7 @@ pub struct AppState {
     pub request_id: Option<RequestId>,
     pub file_storage_client: Arc<dyn FileStorageInterface>,
     pub encryption_client: Arc<dyn EncryptionManagementInterface>,
+    pub grpc_client: Arc<GrpcClients>,
 }
 impl scheduler::SchedulerAppState for AppState {
     fn get_tenants(&self) -> Vec<String> {
@@ -351,6 +353,8 @@ impl AppState {
 
             let file_storage_client = conf.file_storage.get_file_storage_client().await;
 
+            let grpc_client = conf.grpc_client.get_grpc_client_interface().await;
+
             Self {
                 flow_name: String::from("default"),
                 stores,
@@ -367,6 +371,7 @@ impl AppState {
                 request_id: None,
                 file_storage_client,
                 encryption_client,
+                grpc_client,
             }
         })
         .await
@@ -447,6 +452,7 @@ impl AppState {
             email_client: Arc::clone(&self.email_client),
             #[cfg(feature = "olap")]
             opensearch_client: Arc::clone(&self.opensearch_client),
+            grpc_client: Arc::clone(&self.grpc_client),
         })
     }
 }
diff --git a/crates/router/src/routes/dummy_connector.rs b/crates/router/src/routes/dummy_connector.rs
index 26c32776d72..e376bcd7f9e 100644
--- a/crates/router/src/routes/dummy_connector.rs
+++ b/crates/router/src/routes/dummy_connector.rs
@@ -46,7 +46,7 @@ pub async fn dummy_connector_complete_payment(
         attempt_id,
         confirm: json_payload.confirm,
     };
-    api::server_wrap(
+    Box::pin(api::server_wrap(
         flow,
         state,
         &req,
@@ -54,7 +54,7 @@ pub async fn dummy_connector_complete_payment(
         |state, _: (), req, _| core::payment_complete(state, req),
         &auth::NoAuth,
         api_locking::LockAction::NotApplicable,
-    )
+    ))
     .await
 }
 #[instrument(skip_all, fields(flow = ?types::Flow::DummyPaymentCreate))]
@@ -65,7 +65,7 @@ pub async fn dummy_connector_payment(
 ) -> impl actix_web::Responder {
     let payload = json_payload.into_inner();
     let flow = types::Flow::DummyPaymentCreate;
-    api::server_wrap(
+    Box::pin(api::server_wrap(
         flow,
         state,
         &req,
@@ -73,7 +73,7 @@ pub async fn dummy_connector_payment(
         |state, _: (), req, _| core::payment(state, req),
         &auth::NoAuth,
         api_locking::LockAction::NotApplicable,
-    )
+    ))
     .await
 }
 #[instrument(skip_all, fields(flow = ?types::Flow::DummyPaymentRetrieve))]
diff --git a/crates/router/src/routes/routing.rs b/crates/router/src/routes/routing.rs
index 86ad2078f6c..70da80c6756 100644
--- a/crates/router/src/routes/routing.rs
+++ b/crates/router/src/routes/routing.rs
@@ -612,7 +612,7 @@ pub async fn retrieve_surcharge_decision_manager_config(
     req: HttpRequest,
 ) -> impl Responder {
     let flow = Flow::DecisionManagerRetrieveConfig;
-    oss_api::server_wrap(
+    Box::pin(oss_api::server_wrap(
         flow,
         state,
         &req,
@@ -638,7 +638,7 @@ pub async fn retrieve_surcharge_decision_manager_config(
             minimum_entity_level: EntityType::Merchant,
         },
         api_locking::LockAction::NotApplicable,
-    )
+    ))
     .await
 }
 
@@ -727,7 +727,7 @@ pub async fn retrieve_decision_manager_config(
     req: HttpRequest,
 ) -> impl Responder {
     let flow = Flow::DecisionManagerRetrieveConfig;
-    oss_api::server_wrap(
+    Box::pin(oss_api::server_wrap(
         flow,
         state,
         &req,
@@ -750,7 +750,7 @@ pub async fn retrieve_decision_manager_config(
             minimum_entity_level: EntityType::Merchant,
         },
         api_locking::LockAction::NotApplicable,
-    )
+    ))
     .await
 }
 
diff --git a/docker-compose-development.yml b/docker-compose-development.yml
index ebccb114048..cf12982a9bd 100644
--- a/docker-compose-development.yml
+++ b/docker-compose-development.yml
@@ -60,7 +60,9 @@ services:
   ### Application services
   hyperswitch-server:
     image: rust:latest
-    command: cargo run --bin router -- -f ./config/docker_compose.toml
+    command: | 
+     apt-get install -y protobuf-compiler && \
+     cargo run --bin router -- -f ./config/docker_compose.toml
     working_dir: /app
     ports:
       - "8080:8080"
diff --git a/migrations/2024-09-05-160455_add_new_col_is_dynamic_routing_algorithm_in_business_profile/up.sql b/migrations/2024-09-05-160455_add_new_col_is_dynamic_routing_algorithm_in_business_profile/up.sql
index 2482fc9f591..c7a53898129 100644
--- a/migrations/2024-09-05-160455_add_new_col_is_dynamic_routing_algorithm_in_business_profile/up.sql
+++ b/migrations/2024-09-05-160455_add_new_col_is_dynamic_routing_algorithm_in_business_profile/up.sql
@@ -1,3 +1,5 @@
 -- Your SQL goes here
-ALTER TABLE business_profile
-ADD COLUMN dynamic_routing_algorithm JSON DEFAULT NULL;
+ALTER TABLE
+    business_profile
+ADD
+    COLUMN dynamic_routing_algorithm JSON DEFAULT NULL;
\ No newline at end of file
diff --git a/proto/success_rate.proto b/proto/success_rate.proto
new file mode 100644
index 00000000000..8018f6d5fe4
--- /dev/null
+++ b/proto/success_rate.proto
@@ -0,0 +1,57 @@
+syntax = "proto3";
+package success_rate;
+
+ service SuccessRateCalculator {
+     rpc FetchSuccessRate (CalSuccessRateRequest) returns (CalSuccessRateResponse);
+
+     rpc UpdateSuccessRateWindow (UpdateSuccessRateWindowRequest) returns (UpdateSuccessRateWindowResponse);
+ }
+
+ // API-1 types
+ message CalSuccessRateRequest {
+     string id = 1;
+     string params = 2;
+     repeated string labels = 3;
+     CalSuccessRateConfig config = 4;
+ }
+
+ message CalSuccessRateConfig {
+     uint32 min_aggregates_size = 1;
+     double default_success_rate = 2;
+ }
+
+ message CalSuccessRateResponse {
+     repeated LabelWithScore labels_with_score = 1;
+ }
+
+ message LabelWithScore {
+     double score = 1;
+     string label = 2;
+ }
+
+ // API-2 types
+ message UpdateSuccessRateWindowRequest {
+     string id = 1;
+     string params = 2;
+     repeated LabelWithStatus labels_with_status = 3;
+     UpdateSuccessRateWindowConfig config = 4;
+ }
+
+ message LabelWithStatus {
+     string label = 1;
+     bool status = 2;
+ }
+
+ message UpdateSuccessRateWindowConfig {
+     uint32 max_aggregates_size = 1;
+     CurrentBlockThreshold current_block_threshold = 2;
+ }
+
+ message CurrentBlockThreshold {
+     optional uint64 duration_in_mins = 1;
+     uint64 max_total_count = 2;
+ }
+
+ message UpdateSuccessRateWindowResponse {
+     string message = 1;
+ }
\ No newline at end of file

## Type of Change
<!-- Put an `x` in the boxes that apply -->

- [ ] Bugfix
- [ ] New feature
- [ ] Enhancement
- [ ] Refactoring
- [ ] Dependency updates
- [ ] Documentation
- [ ] CI/CD

## Description




Feature Description:
 Building a Grpc Client for communicating with the Dynamo server, once the dynamic routing feature has been enabled.

The dynamic routing feature  would be a compile time as well as runtime feature flag ,
 for providing the flexibility to enable and disable the dynamic routing during deployment

Possible Implementation:
Once dynamic routing has been enabled , we will store the default config in the algorithm column in routing table as the following jsonb:


```
{
    "id": "merchant_id", 
    "params": "card", 
    "labels": ["stripe","adyen"],
    "config": {
        "min_aggregates_size": 2,
        "default_success_rate": 100.0
    }
}
```


gRPC  client will be integrated as a struct of  GrpcClientInterface within the struct  SessionState, 
```
struct SessionState
{
	grpc:  GrpcClientInterface
}
```



The gRPcClientInterface  would be a struct which would have Dynamic Routing as its field and similarly any other future gRPC Client would be one of its fields

```
struct gRPcClientInterface
{
pub dynamic_routing: DynamicRouting
}
```

The DynamicRouting struct would have all the fields that would be there as its services channel

```
struct DynamicRouting
{
pub success_based_client : SuccessBasedClient<Channel>
}

```



The SuccessBasedClient would be a trait would have following  functions: 
calculate_the_success_rate 
update_the_success_rates


```
triat SuccessBasedClient 
{
	type CalCulatedRespons;
	type UpadationResponse;

	async fn calculate_success_rate(self,label_input : Vec<T>) -> CalCulatedRespons

	async fn update_success_rate(self,calculated_success_rate:CalCulatedRespons) -> UpadationResponse
}
```

The SuccessBasedClient would be implemented on client that would be a SuccessBasedClient<Channel>


```
impl SuccessBasedClient for SuccessBasedClient<Channel>
{
	async fn calculate_success_rate(self,label_input : T) ->     CalSuccessRateResponse
	{
		self.fetch_success_rate(request)
	}

	async fn update_success_rate(self,calculated_success_rate :  CalSuccessRateResponse) -> UpdateSuccessRateWindowResponse
  	{

   		self.update_success_rate_window(request)
   	}

}
```

### Additional Changes

- [ ] This PR modifies the API contract
- [ ] This PR modifies the database schema
- [ ] This PR modifies application configuration/environment variables

<!--
Provide links to the files with corresponding changes.

Following are the paths where you can find config files:
1. `config`
2. `crates/router/src/configs`
3. `loadtest/config`
-->


## Motivation and Context
<!--
Why is this change required? What problem does it solve?
If it fixes an open issue, please link to the issue here.

If you don't have an issue, we'd recommend starting with one first so the PR
can focus on the implementation (unless it is an obvious bug or documentation fix
that will have little conversation).
-->


## How did you test it?
- This cannot be tested on sbx, yet

I have tested it locally
- Started the gRPC Server
- Established the connection with the Client <-> Server
- Sent Request the gRPC Server
<img width="1711" alt="Screenshot 2024-09-17 at 7 29 17 PM" src="https://github.com/user-attachments/assets/60fb8ccf-2c4d-48d5-bcfe-f7f3018a915c">

- Response from the server
<img width="1296" alt="Screenshot 2024-09-17 at 7 30 27 PM" src="https://github.com/user-attachments/assets/5de3594b-4701-42a2-9840-a2c14a7db104">



## Checklist
<!-- Put an `x` in the boxes that apply -->

- [ ] I formatted the code `cargo +nightly fmt --all`
- [ ] I addressed lints thrown by `cargo clippy`
- [ ] I reviewed the submitted code
- [ ] I added unit tests for my changes where possible
