Bug: [FEATURE]: build gRPC Client Interface to initiate communication with recovery-trainer gRPC service



Implement a gRPC client interface that facilitates communication with the recovery-trainer gRPC service. The client should be capable of initializing and managing the connection to the service, sending requests, and handling responses in accordance with the defined service contract (protobuf definitions).
diff --git a/Cargo.lock b/Cargo.lock
index c72b8f1b531..c38b3966051 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -3055,12 +3055,14 @@ dependencies = [
  "masking",
  "once_cell",
  "prost",
+ "prost-types",
  "quick-xml",
  "reqwest 0.11.27",
  "router_env",
  "rust-grpc-client",
  "serde",
  "thiserror 1.0.69",
+ "time",
  "tokio 1.45.1",
  "tonic 0.13.1",
  "tonic-build",
@@ -6793,6 +6795,7 @@ dependencies = [
  "openssl",
  "payment_methods",
  "pm_auth",
+ "prost-types",
  "rand 0.8.5",
  "rand_chacha 0.3.1",
  "rdkafka",
diff --git a/config/config.example.toml b/config/config.example.toml
index cdcbd43dcba..06302c28f26 100644
--- a/config/config.example.toml
+++ b/config/config.example.toml
@@ -1154,12 +1154,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  # 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
diff --git a/config/deployments/env_specific.toml b/config/deployments/env_specific.toml
index 46cd59a2d64..6acf697ac4d 100644
--- a/config/deployments/env_specific.toml
+++ b/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
diff --git a/config/deployments/integration_test.toml b/config/deployments/integration_test.toml
index 7ba10061926..9cca0570e7e 100644
--- a/config/deployments/integration_test.toml
+++ b/config/deployments/integration_test.toml
@@ -820,7 +820,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]
diff --git a/config/deployments/production.toml b/config/deployments/production.toml
index 1bdd2c37f9b..50ab48a47fd 100644
--- a/config/deployments/production.toml
+++ b/config/deployments/production.toml
@@ -836,8 +836,8 @@ click_to_pay = {connector_list = "adyen, cybersource, trustpay"}
 
 
 [revenue_recovery]
-monitoring_threshold_in_seconds = 2592000
+monitoring_threshold_in_seconds = 60
 retry_algorithm_type = "cascading"
 
 [list_dispute_supported_connectors]
-connector_list = "worldpayvantiv"
\ No newline at end of file
+connector_list = "worldpayvantiv"
diff --git a/config/deployments/sandbox.toml b/config/deployments/sandbox.toml
index 04797608af3..57fc5eddf6e 100644
--- a/config/deployments/sandbox.toml
+++ b/config/deployments/sandbox.toml
@@ -841,7 +841,7 @@ 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"
 
 [list_dispute_supported_connectors]
diff --git a/config/development.toml b/config/development.toml
index 9105b448e1e..d0a0c7f5a32 100644
--- a/config/development.toml
+++ b/config/development.toml
@@ -1228,6 +1228,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
 
@@ -1264,9 +1267,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
diff --git a/config/docker_compose.toml b/config/docker_compose.toml
index ffb30192bca..15d5b0abba7 100644
--- a/config/docker_compose.toml
+++ b/config/docker_compose.toml
@@ -1156,7 +1156,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]
diff --git a/crates/external_services/Cargo.toml b/crates/external_services/Cargo.toml
index 9e83f331b94..b8933003065 100644
--- a/crates/external_services/Cargo.toml
+++ b/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"
diff --git a/crates/external_services/build.rs b/crates/external_services/build.rs
index c9d9017560b..993d25e5ab4 100644
--- a/crates/external_services/build.rs
+++ b/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");
diff --git a/crates/external_services/src/grpc_client.rs b/crates/external_services/src/grpc_client.rs
index 1b93ac6bd89..875b960bf13 100644
--- a/crates/external_services/src/grpc_client.rs
+++ b/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,47 @@ 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");
+                    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 +192,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
 }
diff --git a/crates/external_services/src/grpc_client/revenue_recovery.rs b/crates/external_services/src/grpc_client/revenue_recovery.rs
new file mode 100644
index 00000000000..22c9bf7ffcb
--- /dev/null
+++ b/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
+}
diff --git a/crates/external_services/src/grpc_client/revenue_recovery/recovery_decider_client.rs b/crates/external_services/src/grpc_client/revenue_recovery/recovery_decider_client.rs
new file mode 100644
index 00000000000..877de2bc77b
--- /dev/null
+++ b/crates/external_services/src/grpc_client/revenue_recovery/recovery_decider_client.rs
@@ -0,0 +1,117 @@
+use std::fmt::Debug;
+
+use common_utils::errors::CustomResult;
+use error_stack::{Report, ResultExt};
+use router_env::logger;
+
+use crate::grpc_client::Client;
+
+#[allow(
+    missing_docs,
+    unused_qualifications,
+    clippy::unwrap_used,
+    clippy::as_conversions,
+    clippy::use_self
+)]
+pub mod decider {
+    tonic::include_proto!("decider");
+}
+
+use decider::decider_client::DeciderClient;
+pub use decider::{DeciderRequest, DeciderResponse};
+
+/// Recovery Decider result
+pub type RecoveryDeciderResult<T> = CustomResult<T, RecoveryDeciderError>;
+
+/// Recovery Decider Error
+#[derive(Debug, Clone, thiserror::Error)]
+pub enum RecoveryDeciderError {
+    /// Error establishing gRPC connection
+    #[error("Failed to establish connection with Recovery Decider service: {0}")]
+    ConnectionError(String),
+    /// Error received from the gRPC service
+    #[error("Recovery Decider service returned an error: {0}")]
+    ServiceError(String),
+    /// Missing configuration for the client
+    #[error("Recovery Decider client configuration is missing or invalid")]
+    ConfigError(String),
+}
+
+/// Recovery Decider Client type
+#[async_trait::async_trait]
+pub trait RecoveryDeciderClientInterface: dyn_clone::DynClone + Send + Sync + Debug {
+    /// fn to call gRPC service
+    async fn decide_on_retry(
+        &mut self,
+        request_payload: DeciderRequest,
+        recovery_headers: super::GrpcRecoveryHeaders,
+    ) -> RecoveryDeciderResult<DeciderResponse>;
+}
+
+dyn_clone::clone_trait_object!(RecoveryDeciderClientInterface);
+
+/// Configuration for the Recovery Decider gRPC client.
+#[derive(Debug, Default, Clone, serde::Deserialize, serde::Serialize)]
+pub struct RecoveryDeciderClientConfig {
+    /// Base URL of the Recovery Decider service
+    pub base_url: String,
+}
+
+impl RecoveryDeciderClientConfig {
+    /// Validate the configuration
+    pub fn validate(&self) -> Result<(), RecoveryDeciderError> {
+        use common_utils::fp_utils::when;
+
+        when(self.base_url.is_empty(), || {
+            Err(RecoveryDeciderError::ConfigError(
+                "Recovery Decider base URL cannot be empty when configuration is provided"
+                    .to_string(),
+            ))
+        })
+    }
+
+    /// create a connection
+    pub fn get_recovery_decider_connection(
+        &self,
+        hyper_client: Client,
+    ) -> Result<DeciderClient<Client>, Report<RecoveryDeciderError>> {
+        let uri = self
+            .base_url
+            .parse::<tonic::transport::Uri>()
+            .map_err(Report::from)
+            .change_context(RecoveryDeciderError::ConfigError(format!(
+                "Invalid URI: {}",
+                self.base_url
+            )))?;
+
+        let service_client = DeciderClient::with_origin(hyper_client, uri);
+
+        Ok(service_client)
+    }
+}
+
+#[async_trait::async_trait]
+impl RecoveryDeciderClientInterface for DeciderClient<Client> {
+    async fn decide_on_retry(
+        &mut self,
+        request_payload: DeciderRequest,
+        recovery_headers: super::GrpcRecoveryHeaders,
+    ) -> RecoveryDeciderResult<DeciderResponse> {
+        let request =
+            super::create_revenue_recovery_grpc_request(request_payload, recovery_headers);
+
+        logger::debug!(decider_request =?request);
+
+        let grpc_response = self
+            .decide(request)
+            .await
+            .change_context(RecoveryDeciderError::ServiceError(
+                "Decider service call failed".to_string(),
+            ))?
+            .into_inner();
+
+        logger::debug!(grpc_decider_response =?grpc_response);
+
+        Ok(grpc_response)
+    }
+}
diff --git a/crates/external_services/src/lib.rs b/crates/external_services/src/lib.rs
index 3d9f1bcf933..f63d2227053 100644
--- a/crates/external_services/src/lib.rs
+++ b/crates/external_services/src/lib.rs
@@ -28,6 +28,42 @@ pub mod managers;
 /// crm module
 pub mod crm;
 
+#[cfg(feature = "revenue_recovery")]
+/// date_time module
+pub mod date_time {
+    use error_stack::ResultExt;
+
+    /// 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,
+    }
+
+    /// Converts a `time::PrimitiveDateTime` to a `prost_types::Timestamp`.
+    pub fn convert_to_prost_timestamp(dt: time::PrimitiveDateTime) -> prost_types::Timestamp {
+        let odt = dt.assume_utc();
+        prost_types::Timestamp {
+            seconds: odt.unix_timestamp(),
+            // This conversion is safe as nanoseconds (0..999_999_999) always fit within an i32.
+            #[allow(clippy::as_conversions)]
+            nanos: odt.nanosecond() as i32,
+        }
+    }
+
+    /// Converts a `prost_types::Timestamp` to an `time::PrimitiveDateTime`.
+    pub fn convert_from_prost_timestamp(
+        ts: &prost_types::Timestamp,
+    ) -> error_stack::Result<time::PrimitiveDateTime, DateTimeConversionError> {
+        let timestamp_nanos = i128::from(ts.seconds) * 1_000_000_000 + i128::from(ts.nanos);
+
+        time::OffsetDateTime::from_unix_timestamp_nanos(timestamp_nanos)
+            .map(|offset_dt| time::PrimitiveDateTime::new(offset_dt.date(), offset_dt.time()))
+            .change_context(DateTimeConversionError::TimestampOutOfRange)
+    }
+}
+
 /// Crate specific constants
 pub mod consts {
     /// General purpose base64 engine
diff --git a/crates/router/Cargo.toml b/crates/router/Cargo.toml
index 4d1e9917ad1..4c59c8e952b 100644
--- a/crates/router/Cargo.toml
+++ b/crates/router/Cargo.toml
@@ -36,7 +36,7 @@ retry = []
 v2 = [ "common_default", "api_models/v2", "diesel_models/v2", "hyperswitch_domain_models/v2", "storage_impl/v2", "kgraph_utils/v2", "common_utils/v2", "hyperswitch_connectors/v2", "hyperswitch_interfaces/v2", "common_types/v2", "revenue_recovery", "scheduler/v2", "euclid/v2", "payment_methods/v2", "tokenization_v2"]
 v1 = ["common_default", "api_models/v1", "diesel_models/v1", "hyperswitch_domain_models/v1", "storage_impl/v1", "hyperswitch_interfaces/v1", "kgraph_utils/v1", "common_utils/v1", "hyperswitch_connectors/v1", "common_types/v1", "scheduler/v1", "payment_methods/v1"]
 dynamic_routing = ["external_services/dynamic_routing", "storage_impl/dynamic_routing", "api_models/dynamic_routing"]
-revenue_recovery = ["api_models/revenue_recovery", "hyperswitch_interfaces/revenue_recovery", "hyperswitch_domain_models/revenue_recovery", "hyperswitch_connectors/revenue_recovery"]
+revenue_recovery = ["api_models/revenue_recovery", "hyperswitch_interfaces/revenue_recovery", "hyperswitch_domain_models/revenue_recovery", "hyperswitch_connectors/revenue_recovery", "external_services/revenue_recovery", "dep:prost-types"]
 tokenization_v2 = ["api_models/tokenization_v2", "diesel_models/tokenization_v2", "hyperswitch_domain_models/tokenization_v2", "storage_impl/tokenization_v2"]
 
 # Partial Auth
@@ -59,6 +59,7 @@ blake3 = "1.8.2"
 bytes = "1.10.1"
 clap = { version = "4.5.38", default-features = false, features = ["std", "derive", "help", "usage"] }
 config = { version = "0.14.1", features = ["toml"] }
+prost-types = { version = "0.13", optional = true }
 cookie = "0.18.1"
 csv = "1.3.1"
 diesel = { version = "2.2.10", features = ["postgres"] }
diff --git a/crates/router/src/configs/settings.rs b/crates/router/src/configs/settings.rs
index 6bc5b35d2b8..7d897ef7c86 100644
--- a/crates/router/src/configs/settings.rs
+++ b/crates/router/src/configs/settings.rs
@@ -1121,6 +1121,15 @@ impl Settings<SecuredSecret> {
 
         self.open_router.validate()?;
 
+        // Validate gRPC client settings
+        #[cfg(feature = "revenue_recovery")]
+        self.grpc_client
+            .recovery_decider_client
+            .as_ref()
+            .map(|config| config.validate())
+            .transpose()
+            .map_err(|err| ApplicationError::InvalidConfigurationValueError(err.to_string()))?;
+
         Ok(())
     }
 }
diff --git a/crates/router/src/core/payments/operations/proxy_payments_intent.rs b/crates/router/src/core/payments/operations/proxy_payments_intent.rs
index 9170b36462a..c9721186ab8 100644
--- a/crates/router/src/core/payments/operations/proxy_payments_intent.rs
+++ b/crates/router/src/core/payments/operations/proxy_payments_intent.rs
@@ -257,6 +257,7 @@ impl<F: Send + Clone + Sync> GetTracker<F, PaymentConfirmData<F>, ProxyPaymentsR
                 ),
             ),
         };
+
         let payment_data = PaymentConfirmData {
             flow: std::marker::PhantomData,
             payment_intent,
diff --git a/crates/router/src/core/revenue_recovery/types.rs b/crates/router/src/core/revenue_recovery/types.rs
index 1e533751f0e..00a661527a1 100644
--- a/crates/router/src/core/revenue_recovery/types.rs
+++ b/crates/router/src/core/revenue_recovery/types.rs
@@ -12,14 +12,16 @@ use common_utils::{
     ext_traits::{OptionExt, ValueExt},
     id_type,
 };
-use diesel_models::{enums, process_tracker::business_status, types as diesel_types};
+use diesel_models::{
+    enums, payment_intent, process_tracker::business_status, types as diesel_types,
+};
 use error_stack::{self, ResultExt};
 use hyperswitch_domain_models::{
     business_profile, merchant_connector_account,
     merchant_context::{Context, MerchantContext},
     payments::{
-        self as domain_payments, payment_attempt, PaymentConfirmData, PaymentIntent,
-        PaymentIntentData,
+        self as domain_payments, payment_attempt::PaymentAttempt, PaymentConfirmData,
+        PaymentIntent, PaymentIntentData,
     },
     router_data_v2::{self, flow_common_types},
     router_flow_types,
@@ -97,7 +99,7 @@ impl RevenueRecoveryPaymentsAttemptStatus {
         payment_intent: &PaymentIntent,
         process_tracker: storage::ProcessTracker,
         revenue_recovery_payment_data: &storage::revenue_recovery::RevenueRecoveryPaymentData,
-        payment_attempt: payment_attempt::PaymentAttempt,
+        payment_attempt: PaymentAttempt,
         revenue_recovery_metadata: &mut PaymentRevenueRecoveryMetadata,
     ) -> Result<(), errors::ProcessTrackerError> {
         let db = &*state.store;
@@ -199,7 +201,7 @@ impl RevenueRecoveryPaymentsAttemptStatus {
                 let action = Box::pin(Action::payment_sync_call(
                     state,
                     revenue_recovery_payment_data,
-                    payment_intent.get_id(),
+                    payment_intent,
                     &process_tracker,
                     payment_attempt,
                 ))
@@ -297,10 +299,10 @@ impl Decision {
 
 #[derive(Debug, Clone)]
 pub enum Action {
-    SyncPayment(payment_attempt::PaymentAttempt),
+    SyncPayment(PaymentAttempt),
     RetryPayment(PrimitiveDateTime),
-    TerminalFailure(payment_attempt::PaymentAttempt),
-    SuccessfulPayment(payment_attempt::PaymentAttempt),
+    TerminalFailure(PaymentAttempt),
+    SuccessfulPayment(PaymentAttempt),
     ReviewPayment,
     ManualReviewAction,
 }
@@ -313,7 +315,6 @@ impl Action {
         revenue_recovery_payment_data: &storage::revenue_recovery::RevenueRecoveryPaymentData,
         revenue_recovery_metadata: &PaymentRevenueRecoveryMetadata,
     ) -> RecoveryResult<Self> {
-        let db = &*state.store;
         let response = revenue_recovery_core::api::call_proxy_api(
             state,
             payment_intent,
@@ -382,11 +383,12 @@ impl Action {
                     };
 
                     Self::decide_retry_failure_action(
-                        db,
+                        state,
                         merchant_id,
                         process.clone(),
                         revenue_recovery_payment_data,
                         &payment_data.payment_attempt,
+                        payment_intent,
                     )
                     .await
                 }
@@ -547,13 +549,13 @@ impl Action {
     pub async fn payment_sync_call(
         state: &SessionState,
         revenue_recovery_payment_data: &storage::revenue_recovery::RevenueRecoveryPaymentData,
-        global_payment_id: &id_type::GlobalPaymentId,
+        payment_intent: &PaymentIntent,
         process: &storage::ProcessTracker,
-        payment_attempt: payment_attempt::PaymentAttempt,
+        payment_attempt: PaymentAttempt,
     ) -> RecoveryResult<Self> {
         let response = revenue_recovery_core::api::call_psync_api(
             state,
-            global_payment_id,
+            payment_intent.get_id(),
             revenue_recovery_payment_data,
         )
         .await;
@@ -565,11 +567,12 @@ impl Action {
                 }
                 RevenueRecoveryPaymentsAttemptStatus::Failed => {
                     Self::decide_retry_failure_action(
-                        db,
+                        state,
                         revenue_recovery_payment_data.merchant_account.get_id(),
                         process.clone(),
                         revenue_recovery_payment_data,
                         &payment_attempt,
+                        payment_intent,
                     )
                     .await
                 }
@@ -747,15 +750,22 @@ impl Action {
     }
 
     pub(crate) async fn decide_retry_failure_action(
-        db: &dyn StorageInterface,
+        state: &SessionState,
         merchant_id: &id_type::MerchantId,
         pt: storage::ProcessTracker,
         revenue_recovery_payment_data: &storage::revenue_recovery::RevenueRecoveryPaymentData,
-        payment_attempt: &payment_attempt::PaymentAttempt,
+        payment_attempt: &PaymentAttempt,
+        payment_intent: &PaymentIntent,
     ) -> RecoveryResult<Self> {
         let next_retry_count = pt.retry_count + 1;
         let schedule_time = revenue_recovery_payment_data
-            .get_schedule_time_based_on_retry_type(db, merchant_id, next_retry_count)
+            .get_schedule_time_based_on_retry_type(
+                state,
+                merchant_id,
+                next_retry_count,
+                payment_attempt,
+                payment_intent,
+            )
             .await;
 
         match schedule_time {
@@ -769,7 +779,7 @@ impl Action {
 // TODO: Move these to impl based functions
 async fn record_back_to_billing_connector(
     state: &SessionState,
-    payment_attempt: &payment_attempt::PaymentAttempt,
+    payment_attempt: &PaymentAttempt,
     payment_intent: &PaymentIntent,
     billing_mca: &merchant_connector_account::MerchantConnectorAccount,
 ) -> RecoveryResult<()> {
@@ -822,7 +832,7 @@ async fn record_back_to_billing_connector(
 pub fn construct_recovery_record_back_router_data(
     state: &SessionState,
     billing_mca: &merchant_connector_account::MerchantConnectorAccount,
-    payment_attempt: &payment_attempt::PaymentAttempt,
+    payment_attempt: &PaymentAttempt,
     payment_intent: &PaymentIntent,
 ) -> RecoveryResult<hyperswitch_domain_models::types::RevenueRecoveryRecordBackRouterData> {
     let auth_type: types::ConnectorAuthType =
diff --git a/crates/router/src/routes/app.rs b/crates/router/src/routes/app.rs
index d4bee7276af..18044b371f5 100644
--- a/crates/router/src/routes/app.rs
+++ b/crates/router/src/routes/app.rs
@@ -13,6 +13,8 @@ use common_utils::id_type;
 use external_services::email::{
     no_email::NoEmailClient, ses::AwsSes, smtp::SmtpServer, EmailClientConfigs, EmailService,
 };
+#[cfg(all(feature = "revenue_recovery", feature = "v2"))]
+use external_services::grpc_client::revenue_recovery::GrpcRecoveryHeaders;
 use external_services::{
     file_storage::FileStorageInterface,
     grpc_client::{GrpcClients, GrpcHeaders},
@@ -151,6 +153,12 @@ impl SessionState {
             request_id: self.request_id.map(|req_id| (*req_id).to_string()),
         }
     }
+    #[cfg(all(feature = "revenue_recovery", feature = "v2"))]
+    pub fn get_recovery_grpc_headers(&self) -> GrpcRecoveryHeaders {
+        GrpcRecoveryHeaders {
+            request_id: self.request_id.map(|req_id| (*req_id).to_string()),
+        }
+    }
 }
 
 pub trait SessionStateInfo {
diff --git a/crates/router/src/types/storage/revenue_recovery.rs b/crates/router/src/types/storage/revenue_recovery.rs
index c672c5ecffb..4b0dd2f2bc8 100644
--- a/crates/router/src/types/storage/revenue_recovery.rs
+++ b/crates/router/src/types/storage/revenue_recovery.rs
@@ -1,13 +1,17 @@
 use std::fmt::Debug;
 
 use common_enums::enums;
-use common_utils::id_type;
+use common_utils::{ext_traits::ValueExt, id_type};
+use external_services::grpc_client::{self as external_grpc_client, GrpcHeaders};
 use hyperswitch_domain_models::{
     business_profile, merchant_account, merchant_connector_account, merchant_key_store,
+    payment_method_data::{Card, PaymentMethodData},
+    payments::{payment_attempt::PaymentAttempt, PaymentIntent},
 };
+use masking::PeekInterface;
 use router_env::logger;
 
-use crate::{db::StorageInterface, workflows::revenue_recovery};
+use crate::{db::StorageInterface, routes::SessionState, workflows::revenue_recovery};
 #[derive(serde::Serialize, serde::Deserialize, Debug)]
 pub struct RevenueRecoveryWorkflowTrackingData {
     pub merchant_id: id_type::MerchantId,
@@ -29,9 +33,11 @@ pub struct RevenueRecoveryPaymentData {
 impl RevenueRecoveryPaymentData {
     pub async fn get_schedule_time_based_on_retry_type(
         &self,
-        db: &dyn StorageInterface,
+        state: &SessionState,
         merchant_id: &id_type::MerchantId,
         retry_count: i32,
+        payment_attempt: &PaymentAttempt,
+        payment_intent: &PaymentIntent,
     ) -> Option<time::PrimitiveDateTime> {
         match self.retry_algorithm {
             enums::RevenueRecoveryAlgorithmType::Monitoring => {
@@ -40,21 +46,41 @@ impl RevenueRecoveryPaymentData {
             }
             enums::RevenueRecoveryAlgorithmType::Cascading => {
                 revenue_recovery::get_schedule_time_to_retry_mit_payments(
-                    db,
+                    state.store.as_ref(),
                     merchant_id,
                     retry_count,
                 )
                 .await
             }
             enums::RevenueRecoveryAlgorithmType::Smart => {
-                // TODO: Integrate the smart retry call to return back a schedule time
-                None
+                revenue_recovery::get_schedule_time_for_smart_retry(
+                    state,
+                    payment_attempt,
+                    payment_intent,
+                    retry_count,
+                )
+                .await
             }
         }
     }
 }
+
 #[derive(Debug, serde::Deserialize, Clone, Default)]
 pub struct RevenueRecoverySettings {
     pub monitoring_threshold_in_seconds: i64,
     pub retry_algorithm_type: enums::RevenueRecoveryAlgorithmType,
+    pub recovery_timestamp: RecoveryTimestamp,
+}
+
+#[derive(Debug, serde::Deserialize, Clone)]
+pub struct RecoveryTimestamp {
+    pub initial_timestamp_in_hours: i64,
+}
+
+impl Default for RecoveryTimestamp {
+    fn default() -> Self {
+        Self {
+            initial_timestamp_in_hours: 1,
+        }
+    }
 }
diff --git a/crates/router/src/workflows/revenue_recovery.rs b/crates/router/src/workflows/revenue_recovery.rs
index 792251de496..c962aeb68e5 100644
--- a/crates/router/src/workflows/revenue_recovery.rs
+++ b/crates/router/src/workflows/revenue_recovery.rs
@@ -6,9 +6,23 @@ use common_utils::{
     id_type,
 };
 #[cfg(feature = "v2")]
+use diesel_models::types::BillingConnectorPaymentMethodDetails;
+#[cfg(feature = "v2")]
 use error_stack::ResultExt;
+#[cfg(all(feature = "revenue_recovery", feature = "v2"))]
+use external_services::{
+    date_time, grpc_client::revenue_recovery::recovery_decider_client as external_grpc_client,
+};
 #[cfg(feature = "v2")]
-use hyperswitch_domain_models::payments::PaymentIntentData;
+use hyperswitch_domain_models::{
+    payment_method_data::PaymentMethodData,
+    payments::{
+        payment_attempt::PaymentAttempt, PaymentConfirmData, PaymentIntent, PaymentIntentData,
+    },
+    router_flow_types::Authorize,
+};
+#[cfg(feature = "v2")]
+use masking::{ExposeInterface, PeekInterface, Secret};
 #[cfg(feature = "v2")]
 use router_env::logger;
 use scheduler::{consumer::workflows::ProcessTrackerWorkflow, errors};
@@ -200,3 +214,279 @@ pub(crate) async fn get_schedule_time_to_retry_mit_payments(
 
     scheduler_utils::get_time_from_delta(time_delta)
 }
+
+#[cfg(feature = "v2")]
+pub(crate) async fn get_schedule_time_for_smart_retry(
+    state: &SessionState,
+    payment_attempt: &PaymentAttempt,
+    payment_intent: &PaymentIntent,
+    retry_count: i32,
+) -> Option<time::PrimitiveDateTime> {
+    let first_error_message = match payment_attempt.error.as_ref() {
+        Some(error) => error.message.clone(),
+        None => {
+            logger::error!(
+                payment_intent_id = ?payment_intent.get_id(),
+                attempt_id = ?payment_attempt.get_id(),
+                "Payment attempt error information not found - cannot proceed with smart retry"
+            );
+            return None;
+        }
+    };
+
+    let billing_state = payment_intent
+        .billing_address
+        .as_ref()
+        .and_then(|addr_enc| addr_enc.get_inner().address.as_ref())
+        .and_then(|details| details.state.as_ref())
+        .cloned();
+
+    // Check if payment_method_data itself is None
+    if payment_attempt.payment_method_data.is_none() {
+        logger::debug!(
+            payment_intent_id = ?payment_intent.get_id(),
+            attempt_id = ?payment_attempt.get_id(),
+            message = "payment_attempt.payment_method_data is None"
+        );
+    }
+
+    let billing_connector_payment_method_details = payment_intent
+        .feature_metadata
+        .as_ref()
+        .and_then(|revenue_recovery_data| {
+            revenue_recovery_data
+                .payment_revenue_recovery_metadata
+                .as_ref()
+        })
+        .and_then(|payment_metadata| {
+            payment_metadata
+                .billing_connector_payment_method_details
+                .as_ref()
+        });
+
+    let revenue_recovery_metadata = payment_intent
+        .feature_metadata
+        .as_ref()
+        .and_then(|metadata| metadata.payment_revenue_recovery_metadata.as_ref());
+
+    let card_network_str = billing_connector_payment_method_details
+        .and_then(|details| match details {
+            BillingConnectorPaymentMethodDetails::Card(card_info) => card_info.card_network.clone(),
+        })
+        .map(|cn| cn.to_string());
+
+    let card_issuer_str =
+        billing_connector_payment_method_details.and_then(|details| match details {
+            BillingConnectorPaymentMethodDetails::Card(card_info) => card_info.card_issuer.clone(),
+        });
+
+    let card_funding_str = payment_intent
+        .feature_metadata
+        .as_ref()
+        .and_then(|revenue_recovery_data| {
+            revenue_recovery_data
+                .payment_revenue_recovery_metadata
+                .as_ref()
+        })
+        .map(|payment_metadata| payment_metadata.payment_method_subtype.to_string());
+
+    let start_time_primitive = payment_intent.created_at;
+    let recovery_timestamp_config = &state.conf.revenue_recovery.recovery_timestamp;
+
+    let modified_start_time_primitive = start_time_primitive.saturating_add(time::Duration::hours(
+        recovery_timestamp_config.initial_timestamp_in_hours,
+    ));
+    let start_time_proto = date_time::convert_to_prost_timestamp(modified_start_time_primitive);
+
+    let merchant_id = Some(payment_intent.merchant_id.get_string_repr().to_string());
+    let invoice_amount = Some(
+        payment_intent
+            .amount_details
+            .order_amount
+            .get_amount_as_i64(),
+    );
+    let invoice_currency = Some(payment_intent.amount_details.currency.to_string());
+
+    let billing_country = payment_intent
+        .billing_address
+        .as_ref()
+        .and_then(|addr_enc| addr_enc.get_inner().address.as_ref())
+        .and_then(|details| details.country.as_ref())
+        .map(|country| country.to_string());
+
+    let billing_city = payment_intent
+        .billing_address
+        .as_ref()
+        .and_then(|addr_enc| addr_enc.get_inner().address.as_ref())
+        .and_then(|details| details.city.as_ref())
+        .cloned();
+
+    let attempt_currency = Some(payment_intent.amount_details.currency.to_string());
+    let attempt_status = Some(payment_attempt.status.to_string());
+    let attempt_amount = Some(
+        payment_attempt
+            .amount_details
+            .get_net_amount()
+            .get_amount_as_i64(),
+    );
+    let attempt_response_time = Some(date_time::convert_to_prost_timestamp(
+        payment_attempt.created_at,
+    ));
+    let payment_method_type = Some(payment_attempt.payment_method_type.to_string());
+    let payment_gateway = payment_attempt.connector.clone();
+
+    let pg_error_code = payment_attempt
+        .error
+        .as_ref()
+        .map(|error| error.code.clone());
+    let network_advice_code = payment_attempt
+        .error
+        .as_ref()
+        .and_then(|error| error.network_advice_code.clone());
+    let network_error_code = payment_attempt
+        .error
+        .as_ref()
+        .and_then(|error| error.network_decline_code.clone());
+
+    let first_pg_error_code = revenue_recovery_metadata
+        .and_then(|metadata| metadata.first_payment_attempt_pg_error_code.clone());
+    let first_network_advice_code = revenue_recovery_metadata
+        .and_then(|metadata| metadata.first_payment_attempt_network_advice_code.clone());
+    let first_network_error_code = revenue_recovery_metadata
+        .and_then(|metadata| metadata.first_payment_attempt_network_decline_code.clone());
+
+    let invoice_due_date = revenue_recovery_metadata
+        .and_then(|metadata| metadata.invoice_next_billing_time)
+        .map(date_time::convert_to_prost_timestamp);
+
+    let decider_request = InternalDeciderRequest {
+        first_error_message,
+        billing_state,
+        card_funding: card_funding_str,
+        card_network: card_network_str,
+        card_issuer: card_issuer_str,
+        invoice_start_time: start_time_proto,
+        retry_count: Some(retry_count.into()),
+        merchant_id,
+        invoice_amount,
+        invoice_currency,
+        invoice_due_date,
+        billing_country,
+        billing_city,
+        attempt_currency,
+        attempt_status,
+        attempt_amount,
+        pg_error_code,
+        network_advice_code,
+        network_error_code,
+        first_pg_error_code,
+        first_network_advice_code,
+        first_network_error_code,
+        attempt_response_time,
+        payment_method_type,
+        payment_gateway,
+        retry_count_left: None,
+        first_error_msg_time: None,
+        wait_time: None,
+    };
+
+    if let Some(mut client) = state.grpc_client.recovery_decider_client.clone() {
+        match client
+            .decide_on_retry(decider_request.into(), state.get_recovery_grpc_headers())
+            .await
+        {
+            Ok(grpc_response) => grpc_response
+                .retry_flag
+                .then_some(())
+                .and(grpc_response.retry_time)
+                .and_then(|prost_ts| {
+                    match date_time::convert_from_prost_timestamp(&prost_ts) {
+                        Ok(pdt) => Some(pdt),
+                        Err(e) => {
+                            logger::error!(
+                                "Failed to convert retry_time from prost::Timestamp: {e:?}"
+                            );
+                            None // If conversion fails, treat as no valid retry time
+                        }
+                    }
+                }),
+
+            Err(e) => {
+                logger::error!("Recovery decider gRPC call failed: {e:?}");
+                None
+            }
+        }
+    } else {
+        logger::debug!("Recovery decider client is not configured");
+        None
+    }
+}
+
+#[cfg(feature = "v2")]
+#[derive(Debug)]
+struct InternalDeciderRequest {
+    first_error_message: String,
+    billing_state: Option<Secret<String>>,
+    card_funding: Option<String>,
+    card_network: Option<String>,
+    card_issuer: Option<String>,
+    invoice_start_time: prost_types::Timestamp,
+    retry_count: Option<i64>,
+    merchant_id: Option<String>,
+    invoice_amount: Option<i64>,
+    invoice_currency: Option<String>,
+    invoice_due_date: Option<prost_types::Timestamp>,
+    billing_country: Option<String>,
+    billing_city: Option<String>,
+    attempt_currency: Option<String>,
+    attempt_status: Option<String>,
+    attempt_amount: Option<i64>,
+    pg_error_code: Option<String>,
+    network_advice_code: Option<String>,
+    network_error_code: Option<String>,
+    first_pg_error_code: Option<String>,
+    first_network_advice_code: Option<String>,
+    first_network_error_code: Option<String>,
+    attempt_response_time: Option<prost_types::Timestamp>,
+    payment_method_type: Option<String>,
+    payment_gateway: Option<String>,
+    retry_count_left: Option<i64>,
+    first_error_msg_time: Option<prost_types::Timestamp>,
+    wait_time: Option<prost_types::Timestamp>,
+}
+
+#[cfg(feature = "v2")]
+impl From<InternalDeciderRequest> for external_grpc_client::DeciderRequest {
+    fn from(internal_request: InternalDeciderRequest) -> Self {
+        Self {
+            first_error_message: internal_request.first_error_message,
+            billing_state: internal_request.billing_state.map(|s| s.peek().to_string()),
+            card_funding: internal_request.card_funding,
+            card_network: internal_request.card_network,
+            card_issuer: internal_request.card_issuer,
+            invoice_start_time: Some(internal_request.invoice_start_time),
+            retry_count: internal_request.retry_count,
+            merchant_id: internal_request.merchant_id,
+            invoice_amount: internal_request.invoice_amount,
+            invoice_currency: internal_request.invoice_currency,
+            invoice_due_date: internal_request.invoice_due_date,
+            billing_country: internal_request.billing_country,
+            billing_city: internal_request.billing_city,
+            attempt_currency: internal_request.attempt_currency,
+            attempt_status: internal_request.attempt_status,
+            attempt_amount: internal_request.attempt_amount,
+            pg_error_code: internal_request.pg_error_code,
+            network_advice_code: internal_request.network_advice_code,
+            network_error_code: internal_request.network_error_code,
+            first_pg_error_code: internal_request.first_pg_error_code,
+            first_network_advice_code: internal_request.first_network_advice_code,
+            first_network_error_code: internal_request.first_network_error_code,
+            attempt_response_time: internal_request.attempt_response_time,
+            payment_method_type: internal_request.payment_method_type,
+            payment_gateway: internal_request.payment_gateway,
+            retry_count_left: internal_request.retry_count_left,
+            first_error_msg_time: internal_request.first_error_msg_time,
+            wait_time: internal_request.wait_time,
+        }
+    }
+}
diff --git a/proto/recovery_decider.proto b/proto/recovery_decider.proto
new file mode 100644
index 00000000000..28543b8fa28
--- /dev/null
+++ b/proto/recovery_decider.proto
@@ -0,0 +1,45 @@
+syntax = "proto3";
+
+package decider;
+
+import "google/protobuf/timestamp.proto";
+
+service Decider {
+  rpc Decide (DeciderRequest) returns (DeciderResponse);
+}
+
+message DeciderRequest {
+  string first_error_message = 1;
+  optional string billing_state = 2;
+  optional string card_funding = 3;
+  optional string card_network = 4;
+  optional string card_issuer = 5;
+  google.protobuf.Timestamp invoice_start_time = 6;
+  optional int64 retry_count = 7;
+  optional string merchant_id = 8;
+  optional int64 invoice_amount = 9;
+  optional string invoice_currency = 10;
+  optional google.protobuf.Timestamp invoice_due_date = 11;
+  optional string billing_country = 12;
+  optional string billing_city = 13;
+  optional string attempt_currency = 14;
+  optional string attempt_status = 15;
+  optional int64 attempt_amount = 16;
+  optional string pg_error_code = 17;
+  optional string network_advice_code = 18;
+  optional string network_error_code = 19;
+  optional string first_pg_error_code = 20;
+  optional string first_network_advice_code = 21;
+  optional string first_network_error_code = 22;
+  optional google.protobuf.Timestamp attempt_response_time = 23;
+  optional string payment_method_type = 24;
+  optional string payment_gateway = 25;
+  optional int64 retry_count_left = 26;
+  optional google.protobuf.Timestamp first_error_msg_time = 27;
+  optional google.protobuf.Timestamp wait_time = 28;
+}
+
+message DeciderResponse {
+  bool retry_flag = 1;
+  google.protobuf.Timestamp retry_time = 2;
+}

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

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

## Description
<!-- Describe your changes in detail -->
I have implemented a gRPC client that facilitates communication with the recovery-decider gRPC service. The client should be capable of initializing and managing the connection to the service, sending requests, and handling responses in accordance with the defined service contract (protobuf definitions).

### 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?
<!--
Did you write an integration/unit/API test to verify the code changes?
Or did you test this change manually (provide relevant screenshots)?
-->

Decider gRPC server:-
<img width="1227" alt="Screenshot 2025-06-11 at 01 16 34" src="https://github.com/user-attachments/assets/b4887ed0-4315-406e-b393-cbd9bf480c95" />

Decider gRPC client(HS):-
<img width="1159" alt="Screenshot 2025-06-05 at 02 05 20" src="https://github.com/user-attachments/assets/bbddd857-29e8-49bc-bb73-9797f8837b5b" />

Predictor gRPC server:-
<img width="1201" alt="Screenshot 2025-06-05 at 02 03 06" src="https://github.com/user-attachments/assets/a3afdb68-33e7-4c4c-8117-8a3a501ecdb8" />

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

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


<!-- This is an auto-generated comment: release notes by coderabbit.ai -->
## Summary by CodeRabbit

- **New Features**
  - Introduced support for a gRPC client: Recovery Decider, enabling external retry decisioning.
  - Integrated dynamic scheduling for smart retry logic based on external gRPC service responses.
  - Added new protobuf definition for Recovery Decider.

- **Improvements**
  - Enhanced revenue recovery payment flow with richer payment intent context and smarter retry scheduling.
  - Expanded logging and API flow tracking with new flow types.

- **Configuration**
  - Added configuration options for new gRPC clients in the development environment.

- **Bug Fixes**
  - Improved error handling and logging for gRPC client interactions.