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

config/config.example.toml

@@ -1148,12 +1148,18 @@ url = "http://localhost:8080"           # Open Router URL
 base_url = "http://localhost:8000"      # Unified Connector Service Base URL
 connection_timeout = 10                 # Connection Timeout Duration in Seconds
 
+[grpc_client.recovery_decider_client] # Revenue recovery client base url
+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
 
 [revenue_recovery]
-monitoring_threshold_in_seconds = 2592000 # 30*24*60*60 secs , threshold for monitoring the retry system
-retry_algorithm_type = "cascading" # type of retry algorithm
+monitoring_threshold_in_seconds = 60  # 30*24*60*60 secs , threshold for monitoring the retry system
+retry_algorithm_type = "cascading"  # type of retry algorithm
+
+[revenue_recovery.recovery_timestamp] # Timestamp configuration for Revenue Recovery
+initial_timestamp_in_hours = 1          # number of hours added to start time for Decider service of Revenue Recovery
 
 [clone_connector_allowlist]
 merchant_ids = "merchant_ids"           # Comma-separated list of allowed merchant IDs

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] # Revenue recovery client base url
+base_url = "http://127.0.0.1:8080"  #Base URL
+
 [theme.storage]
 file_storage_backend = "aws_s3" # Theme storage backend to be used
 
@@ -382,6 +385,9 @@ connector_names = "connector_names"     # Comma-separated list of allowed connec
 base_url = "http://localhost:8000"      # Unified Connector Service Base URL
 connection_timeout = 10                 # Connection Timeout Duration in Seconds
 
+[revenue_recovery.recovery_timestamp] # Timestamp configuration for Revenue Recovery
+initial_timestamp_in_hours = 1        # number of hours added to start time for Decider service of Revenue Recovery
+
 [chat]
 enabled = false                                # Enable or disable chat features
 hyperswitch_ai_host = "http://0.0.0.0:8000"    # Hyperswitch ai workflow host

config/deployments/integration_test.toml

@@ -814,7 +814,7 @@ billing_connectors_which_requires_invoice_sync_call = "recurly"
 
 
 [revenue_recovery]
-monitoring_threshold_in_seconds = 2592000
+monitoring_threshold_in_seconds = 60
 retry_algorithm_type = "cascading"
 
 [authentication_providers]

config/deployments/production.toml

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

config/deployments/sandbox.toml

@@ -832,5 +832,5 @@ billing_connectors_which_requires_invoice_sync_call = "recurly"
 click_to_pay = {connector_list = "adyen, cybersource, trustpay"}
 
 [revenue_recovery]
-monitoring_threshold_in_seconds = 2592000
+monitoring_threshold_in_seconds = 60
 retry_algorithm_type = "cascading"

config/development.toml

@@ -1222,6 +1222,9 @@ host = "localhost"
 port = 8000
 service = "dynamo"
 
+[grpc_client.recovery_decider_client] # Revenue recovery client base url
+base_url = "http://127.0.0.1:8080" #Base URL
+
 [theme.storage]
 file_storage_backend = "file_system" # Theme storage backend to be used
 
@@ -1255,9 +1258,12 @@ ucs_only_connectors = [
 ]
 
 [revenue_recovery]
-monitoring_threshold_in_seconds = 2592000
+monitoring_threshold_in_seconds = 60
 retry_algorithm_type = "cascading"
 
+[revenue_recovery.recovery_timestamp]
+initial_timestamp_in_hours = 1
+
 [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

config/docker_compose.toml

@@ -1150,7 +1150,7 @@ hyperswitch_ai_host = "http://0.0.0.0:8000"
 click_to_pay = {connector_list = "adyen, cybersource, trustpay"}
 
 [revenue_recovery]
-monitoring_threshold_in_seconds = 2592000 # threshold for monitoring the retry system
+monitoring_threshold_in_seconds = 60 # threshold for monitoring the retry system
 retry_algorithm_type = "cascading" # type of retry algorithm
 
 [clone_connector_allowlist]

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",
+    "dep:prost-types",
+]
 dynamic_routing = [
     "dep:prost",
     "dep:api_models",
@@ -43,6 +53,8 @@ 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-reflection = "0.13.1"

crates/external_services/build.rs

@@ -1,9 +1,23 @@
 #[allow(clippy::expect_used)]
 fn main() -> Result<(), Box<dyn std::error::Error>> {
+    // Compilation for revenue recovery 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 recovery_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")
+            .compile_protos(&recovery_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")]
+#[cfg(any(feature = "dynamic_routing", feature = "revenue_recovery"))]
 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::{
+    recovery_decider_client::{
+        DeciderRequest, DeciderResponse, RecoveryDeciderClientConfig,
+        RecoveryDeciderClientInterface, RecoveryDeciderError, RecoveryDeciderResult,
+    },
+    GrpcRecoveryHeaders,
+};
 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 Decider 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,7 @@ 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")]
+        #[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 +93,48 @@ 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 = {
+            match &self.recovery_decider_client {
+                Some(config) => {
+                    // Validate the config first
+                    config
+                        .validate()
+                        .expect("Recovery Decider configuration validation failed");
+
+                    // Create the client
+                    let client = config
+                        .get_recovery_decider_connection(client.clone())
+                        .expect(
+                            "Failed to establish a connection with the Recovery Decider Server",
+                        );
+
+                    logger::info!("Recovery Decider gRPC client successfully initialized");
+                    // Some(Box::new(client) as Box<dyn RecoveryDeciderClientInterface>)
+                    let boxed_client: Box<dyn RecoveryDeciderClientInterface> = Box::new(client);
+                    Some(boxed_client)
+                }
+                None => {
+                    logger::debug!("Recovery Decider client configuration not provided, client will be disabled");
+                    None
+                }
+            }
+        };
+
         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 +193,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,49 @@
+/// Recovery Decider client
+pub mod recovery_decider_client;
+
+use std::fmt::Debug;
+
+use common_utils::consts;
+use router_env::logger;
+
+/// 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
+}