feat(gRPC): build gRPC client interface to initiate communication with recovery-decider service

config/config.example.toml

@@ -1146,6 +1146,9 @@ host = "localhost"                      # Unified Connector Service Client Host
 port = 8000                             # Unified Connector Service Client Port
 connection_timeout = 10                 # Connection Timeout Duration in Seconds
 
+[grpc_client.recovery_decider_client]
+base_url = "http://127.0.0.1:8080" #Base URL
+
 [billing_connectors_invoice_sync]
 billing_connectors_which_requires_invoice_sync_call = "recurly" # List of billing connectors which has invoice sync api call
 

config/deployments/env_specific.toml

@@ -354,6 +354,9 @@ host = "localhost" # Client Host
 port = 7000        # Client Port
 service = "dynamo" # Service name
 
+[grpc_client.recovery_decider_client]
+base_url = "http://127.0.0.1:8080" #Base URL
+
 [theme.storage]
 file_storage_backend = "aws_s3" # Theme storage backend to be used
 

config/deployments/production.toml

@@ -824,4 +824,4 @@ click_to_pay = {connector_list = "adyen, cybersource, trustpay"}
 
 [revenue_recovery]
 monitoring_threshold_in_seconds = 2592000
-retry_algorithm_type = "cascading"
+retry_algorithm_type = "cascading"

config/development.toml

@@ -1219,6 +1219,9 @@ host = "localhost"
 port = 8000
 service = "dynamo"
 
+[grpc_client.recovery_decider_client]
+base_url = "http://127.0.0.1:8080"
+
 [theme.storage]
 file_storage_backend = "file_system" # Theme storage backend to be used
 
@@ -1256,6 +1259,10 @@ ucs_only_connectors = [
 monitoring_threshold_in_seconds = 2592000
 retry_algorithm_type = "cascading"
 
+[revenue_recovery.recovery_timestamp]
+initial_timestamp_in_hours = 1
+final_timestamp_in_days = 30
+
 [clone_connector_allowlist]
 merchant_ids = "merchant_123, merchant_234"     # Comma-separated list of allowed merchant IDs
 connector_names = "stripe, adyen"               # Comma-separated list of allowed connector names

crates/external_services/Cargo.toml

@@ -13,6 +13,16 @@ email = ["dep:aws-config"]
 aws_s3 = ["dep:aws-config", "dep:aws-sdk-s3"]
 hashicorp-vault = ["dep:vaultrs"]
 v1 = ["hyperswitch_interfaces/v1", "common_utils/v1"]
+v2 = ["hyperswitch_interfaces/v2", "common_utils/v2"]
+revenue_recovery = [
+    "dep:prost",
+    "dep:router_env",
+    "tokio/macros",
+    "tokio/rt-multi-thread",
+    "dep:hyper-util",
+    "dep:http-body-util",
+    "prost-types",
+]
 dynamic_routing = [
     "dep:prost",
     "dep:api_models",
@@ -43,8 +53,10 @@ serde = { version = "1.0.219", features = ["derive"] }
 thiserror = "1.0.69"
 vaultrs = { version = "0.7.4", optional = true }
 prost = { version = "0.13", optional = true }
+prost-types = { version = "0.13", optional = true }
+time = { version = "0.3.41", features = ["serde", "serde-well-known", "std"] }
 tokio = "1.45.1"
-tonic = "0.13.1"
+tonic =  "0.13.1"
 tonic-reflection = "0.13.1"
 tonic-types = "0.13.1"
 hyper-util = { version = "0.1.12", optional = true }
@@ -68,7 +80,7 @@ api_models = { version = "0.1.0", path = "../api_models", optional = true }
 
 
 [build-dependencies]
-tonic-build = "0.13.1"
+tonic-build =  "0.13.1"
 router_env = { version = "0.1.0", path = "../router_env", default-features = false, optional = true }
 
 [lints]

crates/external_services/build.rs

@@ -1,9 +1,27 @@
 #[allow(clippy::expect_used)]
 fn main() -> Result<(), Box<dyn std::error::Error>> {
+    // Compilation for v2 protos
+    #[cfg(feature = "revenue_recovery")]
+    {
+        let proto_base_path = router_env::workspace_path().join("proto");
+        let out_dir = std::path::PathBuf::from(std::env::var("OUT_DIR")?);
+        let v2_proto_files = [proto_base_path.join("recovery_decider.proto")];
+        tonic_build::configure()
+            .out_dir(&out_dir)
+            .compile_well_known_types(true)
+            .extern_path(".google.protobuf.Timestamp", "::prost_types::Timestamp")
+            .type_attribute(
+                "google.protobuf.Timestamp",
+                "#[derive(serde::Serialize, serde::Deserialize)]",
+            )
+            .compile_protos(&v2_proto_files, &[&proto_base_path])
+            .expect("Failed to compile revenue-recovery proto files");
+    }
+
+    // Compilation for dynamic_routing protos
     #[cfg(feature = "dynamic_routing")]
     {
         // Get the directory of the current crate
-
         let proto_path = router_env::workspace_path().join("proto");
         let success_rate_proto_file = proto_path.join("success_rate.proto");
         let contract_routing_proto_file = proto_path.join("contract_routing.proto");

crates/external_services/src/grpc_client.rs

@@ -4,6 +4,10 @@ pub mod dynamic_routing;
 /// gRPC based Heath Check Client interface implementation
 #[cfg(feature = "dynamic_routing")]
 pub mod health_check_client;
+/// gRPC based Recovery Trainer Client interface implementation
+#[cfg(feature = "revenue_recovery")]
+pub mod revenue_recovery;
+
 /// gRPC based Unified Connector Service Client interface implementation
 pub mod unified_connector_service;
 use std::{fmt::Debug, sync::Arc};
@@ -14,19 +18,26 @@ use common_utils::consts;
 use dynamic_routing::{DynamicRoutingClientConfig, RoutingStrategy};
 #[cfg(feature = "dynamic_routing")]
 use health_check_client::HealthCheckClient;
-#[cfg(feature = "dynamic_routing")]
+#[cfg(any(feature = "dynamic_routing", feature = "revenue_recovery"))]
 use hyper_util::client::legacy::connect::HttpConnector;
 #[cfg(feature = "dynamic_routing")]
 use router_env::logger;
 use serde;
-#[cfg(feature = "dynamic_routing")]
+#[cfg(any(feature = "dynamic_routing", feature = "revenue_recovery"))]
 use tonic::body::Body;
 
+#[cfg(feature = "revenue_recovery")]
+pub use self::revenue_recovery::common::GrpcRecoveryHeaders;
+#[cfg(feature = "revenue_recovery")]
+pub use self::revenue_recovery::recovery_decider_client::{
+    DeciderRequest, DeciderResponse, RecoveryDeciderClientConfig, RecoveryDeciderClientInterface,
+    RecoveryDeciderError, RecoveryDeciderResult,
+};
 use crate::grpc_client::unified_connector_service::{
     UnifiedConnectorServiceClient, UnifiedConnectorServiceClientConfig,
 };
 
-#[cfg(feature = "dynamic_routing")]
+#[cfg(any(feature = "dynamic_routing", feature = "revenue_recovery"))]
 /// Hyper based Client type for maintaining connection pool for all gRPC services
 pub type Client = hyper_util::client::legacy::Client<HttpConnector, Body>;
 
@@ -39,15 +50,22 @@ pub struct GrpcClients {
     /// Health Check client for all gRPC services
     #[cfg(feature = "dynamic_routing")]
     pub health_client: HealthCheckClient,
+    /// Recovery Decoder Client
+    #[cfg(feature = "revenue_recovery")]
+    pub recovery_decider_client: Option<Box<dyn RecoveryDeciderClientInterface>>,
     /// Unified Connector Service client
     pub unified_connector_service_client: Option<UnifiedConnectorServiceClient>,
 }
+
 /// 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: Option<DynamicRoutingClientConfig>,
+    #[cfg(feature = "revenue_recovery")]
+    /// Configs for Recovery Decider Client
+    pub recovery_decider_client: Option<RecoveryDeciderClientConfig>,
     /// Configs for Unified Connector Service client
     pub unified_connector_service: Option<UnifiedConnectorServiceClientConfig>,
 }
@@ -59,7 +77,8 @@ impl GrpcClientSettings {
     /// 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")]
+        // Define the hyper client if any gRPC feature is enabled
+        #[cfg(any(feature = "dynamic_routing", feature = "revenue_recovery"))]
         let client =
             hyper_util::client::legacy::Client::builder(hyper_util::rt::TokioExecutor::new())
                 .http2_only(true)
@@ -75,18 +94,29 @@ impl GrpcClientSettings {
             .flatten();
 
         #[cfg(feature = "dynamic_routing")]
-        let health_client = HealthCheckClient::build_connections(self, client)
+        let health_client = HealthCheckClient::build_connections(self, client.clone())
             .await
             .expect("Failed to build gRPC connections");
 
         let unified_connector_service_client =
             UnifiedConnectorServiceClient::build_connections(self).await;
 
+        #[cfg(feature = "revenue_recovery")]
+        let recovery_decider_client = self
+            .recovery_decider_client
+            .as_ref()
+            .map(|config| config.get_recovery_decider_connection(client.clone()))
+            .transpose()
+            .expect("Failed to establish a connection with the Recovery Decider Server")
+            .map(|client| -> Box<dyn RecoveryDeciderClientInterface> { Box::new(client) });
+
         Arc::new(GrpcClients {
             #[cfg(feature = "dynamic_routing")]
             dynamic_routing: dynamic_routing_connection,
             #[cfg(feature = "dynamic_routing")]
             health_client,
+            #[cfg(feature = "revenue_recovery")]
+            recovery_decider_client,
             unified_connector_service_client,
         })
     }
@@ -145,7 +175,7 @@ pub(crate) fn create_grpc_request<T: Debug>(message: T, headers: GrpcHeaders) ->
     let mut request = tonic::Request::new(message);
     request.add_headers_to_grpc_request(headers);
 
-    logger::info!(dynamic_routing_request=?request);
+    logger::info!(?request);
 
     request
 }

crates/external_services/src/grpc_client/revenue_recovery.rs

@@ -0,0 +1,4 @@
+/// common file for revenue recovery
+pub mod common;
+/// Recovery Decider client
+pub mod recovery_decider_client;

crates/external_services/src/grpc_client/revenue_recovery/common.rs

@@ -0,0 +1,54 @@
+use std::fmt::Debug;
+
+use common_utils::consts;
+use router_env::logger;
+
+/// Errors in time conversion
+#[derive(Debug, thiserror::Error)]
+pub enum DateTimeConversionError {
+    #[error("Invalid timestamp value from prost Timestamp: out of representable range")]
+    /// Error for out of range
+    TimestampOutOfRange,
+}
+
+/// Contains recovery grpc headers
+#[derive(Debug)]
+pub struct GrpcRecoveryHeaders {
+    /// Request id
+    pub request_id: Option<String>,
+}
+
+/// Trait to add necessary recovery headers to the tonic Request
+pub(crate) trait AddRecoveryHeaders {
+    /// Add necessary recovery header fields to the tonic Request
+    fn add_recovery_headers(&mut self, headers: GrpcRecoveryHeaders);
+}
+
+impl<T> AddRecoveryHeaders for tonic::Request<T> {
+    #[track_caller]
+    fn add_recovery_headers(&mut self, headers: GrpcRecoveryHeaders) {
+        headers.request_id.map(|request_id| {
+            request_id
+                .parse()
+                .map(|request_id_val| {
+                    self
+                        .metadata_mut()
+                        .append(consts::X_REQUEST_ID, request_id_val)
+                })
+                .inspect_err(
+                    |err| logger::warn!(header_parse_error=?err,"invalid {} received",consts::X_REQUEST_ID),
+                )
+                .ok();
+        });
+    }
+}
+
+/// Creates a tonic::Request with recovery headers added.
+pub(crate) fn create_revenue_recovery_grpc_request<T: Debug>(
+    message: T,
+    recovery_headers: GrpcRecoveryHeaders,
+) -> tonic::Request<T> {
+    let mut request = tonic::Request::new(message);
+    request.add_recovery_headers(recovery_headers);
+    request
+}