feat(routing): add invalidate window as a service for SR based routing (#6264)

Co-authored-by: hyperswitch-bot[bot] <148525504+hyperswitch-bot[bot]@users.noreply.github.com>

config/development.toml

@@ -782,3 +782,7 @@ card_networks = "Visa, AmericanExpress, Mastercard"
 
 [network_tokenization_supported_connectors]
 connector_list = "cybersource"
+
+[grpc_client.dynamic_routing_client] 
+host = "localhost" 
+port = 7000       

crates/external_services/src/grpc_client/dynamic_routing.rs

@@ -14,8 +14,9 @@ use serde;
 use success_rate::{
     success_rate_calculator_client::SuccessRateCalculatorClient, CalSuccessRateConfig,
     CalSuccessRateRequest, CalSuccessRateResponse,
-    CurrentBlockThreshold as DynamicCurrentThreshold, LabelWithStatus,
+    CurrentBlockThreshold as DynamicCurrentThreshold, InvalidateWindowsRequest,
-    UpdateSuccessRateWindowConfig, UpdateSuccessRateWindowRequest, UpdateSuccessRateWindowResponse,
+    InvalidateWindowsResponse, LabelWithStatus, UpdateSuccessRateWindowConfig,
+    UpdateSuccessRateWindowRequest, UpdateSuccessRateWindowResponse,
 };
 use tonic::Status;
 #[allow(
@@ -111,6 +112,11 @@ pub trait SuccessBasedDynamicRouting: dyn_clone::DynClone + Send + Sync {
         params: String,
         response: Vec<RoutableConnectorChoiceWithStatus>,
     ) -> DynamicRoutingResult<UpdateSuccessRateWindowResponse>;
+    /// To invalidates the success rate routing keys
+    async fn invalidate_success_rate_routing_keys(
+        &self,
+        id: String,
+    ) -> DynamicRoutingResult<InvalidateWindowsResponse>;
 }
 
 #[async_trait::async_trait]
@@ -139,9 +145,8 @@ impl SuccessBasedDynamicRouting for SuccessRateCalculatorClient<Client> {
             config,
         });
 
-        let mut client = self.clone();
+        let response = self
-
+            .clone()
-        let response = client
             .fetch_success_rate(request)
             .await
             .change_context(DynamicRoutingError::SuccessRateBasedRoutingFailure(
@@ -179,9 +184,8 @@ impl SuccessBasedDynamicRouting for SuccessRateCalculatorClient<Client> {
             config,
         });
 
-        let mut client = self.clone();
+        let response = self
-
+            .clone()
-        let response = client
             .update_success_rate_window(request)
             .await
             .change_context(DynamicRoutingError::SuccessRateBasedRoutingFailure(
@@ -191,6 +195,23 @@ impl SuccessBasedDynamicRouting for SuccessRateCalculatorClient<Client> {
 
         Ok(response)
     }
+
+    async fn invalidate_success_rate_routing_keys(
+        &self,
+        id: String,
+    ) -> DynamicRoutingResult<InvalidateWindowsResponse> {
+        let request = tonic::Request::new(InvalidateWindowsRequest { id });
+
+        let response = self
+            .clone()
+            .invalidate_windows(request)
+            .await
+            .change_context(DynamicRoutingError::SuccessRateBasedRoutingFailure(
+                "Failed to invalidate the success rate routing keys".to_string(),
+            ))?
+            .into_inner();
+        Ok(response)
+    }
 }
 
 impl ForeignTryFrom<CurrentBlockThreshold> for DynamicCurrentThreshold {

crates/router/src/core/routing.rs

@@ -7,14 +7,17 @@ use api_models::{
     routing::{self as routing_types, RoutingRetrieveQuery},
 };
 use async_trait::async_trait;
+#[cfg(all(feature = "v1", feature = "dynamic_routing"))]
+use common_utils::ext_traits::AsyncExt;
 use diesel_models::routing_algorithm::RoutingAlgorithm;
 use error_stack::ResultExt;
+#[cfg(all(feature = "v1", feature = "dynamic_routing"))]
+use external_services::grpc_client::dynamic_routing::SuccessBasedDynamicRouting;
 use hyperswitch_domain_models::{mandates, payment_address};
-#[cfg(feature = "v1")]
+#[cfg(all(feature = "v1", feature = "dynamic_routing"))]
-use router_env::logger;
+use router_env::{logger, metrics::add_attributes};
-use router_env::metrics::add_attributes;
 use rustc_hash::FxHashSet;
-#[cfg(feature = "v1")]
+#[cfg(all(feature = "v1", feature = "dynamic_routing"))]
 use storage_impl::redis::cache;
 
 #[cfg(feature = "payouts")]
@@ -1182,7 +1185,7 @@ pub async fn update_default_routing_config_for_profile(
     ))
 }
 
-#[cfg(feature = "v1")]
+#[cfg(all(feature = "v1", feature = "dynamic_routing"))]
 pub async fn toggle_success_based_routing(
     state: SessionState,
     merchant_account: domain::MerchantAccount,
@@ -1379,7 +1382,7 @@ pub async fn toggle_success_based_routing(
     }
 }
 
-#[cfg(feature = "v1")]
+#[cfg(all(feature = "v1", feature = "dynamic_routing"))]
 pub async fn success_based_routing_update_configs(
     state: SessionState,
     request: routing_types::SuccessBasedRoutingConfig,
@@ -1449,6 +1452,27 @@ pub async fn success_based_routing_update_configs(
         1,
         &add_attributes([("profile_id", profile_id.get_string_repr().to_owned())]),
     );
+
+    let prefix_of_dynamic_routing_keys = helpers::generate_tenant_business_profile_id(
+        &state.tenant.redis_key_prefix,
+        profile_id.get_string_repr(),
+    );
+    state
+        .grpc_client
+        .dynamic_routing
+        .success_rate_client
+        .as_ref()
+        .async_map(|sr_client| async {
+            sr_client
+                .invalidate_success_rate_routing_keys(prefix_of_dynamic_routing_keys)
+                .await
+                .change_context(errors::ApiErrorResponse::GenericNotFoundError {
+                    message: "Failed to invalidate the routing keys".to_string(),
+                })
+        })
+        .await
+        .transpose()?;
+
     Ok(service_api::ApplicationResponse::Json(new_record))
 }
 

crates/router/src/routes/app.rs

@@ -1724,22 +1724,23 @@ impl Profile {
 #[cfg(all(feature = "olap", feature = "v1"))]
 impl Profile {
     pub fn server(state: AppState) -> Scope {
-        web::scope("/account/{account_id}/business_profile")
+        let mut route = web::scope("/account/{account_id}/business_profile")
             .app_data(web::Data::new(state))
             .service(
                 web::resource("")
                     .route(web::post().to(profiles::profile_create))
                     .route(web::get().to(profiles::profiles_list)),
-            )
+            );
-            .service(
+
-                web::scope("/{profile_id}")
+        #[cfg(feature = "dynamic_routing")]
-                    .service(
+        {
-                        web::scope("/dynamic_routing").service(
+            route =
+                route.service(
+                    web::scope("/{profile_id}/dynamic_routing").service(
                         web::scope("/success_based")
                             .service(
-                                    web::resource("/toggle").route(
+                                web::resource("/toggle")
-                                        web::post().to(routing::toggle_success_based_routing),
+                                    .route(web::post().to(routing::toggle_success_based_routing)),
-                                    ),
                             )
                             .service(web::resource("/config/{algorithm_id}").route(
                                 web::patch().to(|state, req, path, payload| {
@@ -1749,7 +1750,11 @@ impl Profile {
                                 }),
                             )),
                     ),
-                    )
+                );
+        }
+
+        route = route.service(
+            web::scope("/{profile_id}")
                 .service(
                     web::resource("")
                         .route(web::get().to(profiles::profile_retrieve))
@@ -1764,7 +1769,9 @@ impl Profile {
                     web::resource("/toggle_connector_agnostic_mit")
                         .route(web::post().to(profiles::toggle_connector_agnostic_mit)),
                 ),
-            )
+        );
+
+        route
     }
 }
 

crates/router/src/routes/routing.rs

@@ -1009,7 +1009,7 @@ pub async fn routing_update_default_config_for_profile(
     .await
 }
 
-#[cfg(all(feature = "olap", feature = "v1"))]
+#[cfg(all(feature = "olap", feature = "v1", feature = "dynamic_routing"))]
 #[instrument(skip_all)]
 pub async fn toggle_success_based_routing(
     state: web::Data<AppState>,
@@ -1052,7 +1052,7 @@ pub async fn toggle_success_based_routing(
     .await
 }
 
-#[cfg(all(feature = "olap", feature = "v1"))]
+#[cfg(all(feature = "olap", feature = "v1", feature = "dynamic_routing"))]
 #[instrument(skip_all)]
 pub async fn success_based_routing_update_configs(
     state: web::Data<AppState>,

proto/success_rate.proto

@@ -1,57 +1,68 @@
 syntax = "proto3";
 package success_rate;
 
- service SuccessRateCalculator {
+service SuccessRateCalculator {
     rpc FetchSuccessRate (CalSuccessRateRequest) returns (CalSuccessRateResponse);
 
     rpc UpdateSuccessRateWindow (UpdateSuccessRateWindowRequest) returns (UpdateSuccessRateWindowResponse);
- }
 
- // API-1 types
+    rpc InvalidateWindows (InvalidateWindowsRequest) returns (InvalidateWindowsResponse);
- message CalSuccessRateRequest {
+}
+
+// API-1 types
+message CalSuccessRateRequest {
     string id = 1;
     string params = 2;
     repeated string labels = 3;
     CalSuccessRateConfig config = 4;
- }
+}
 
- message CalSuccessRateConfig {
+message CalSuccessRateConfig {
     uint32 min_aggregates_size = 1;
     double default_success_rate = 2;
- }
+}
 
- message CalSuccessRateResponse {
+message CalSuccessRateResponse {
     repeated LabelWithScore labels_with_score = 1;
- }
+}
 
- message LabelWithScore {
+message LabelWithScore {
     double score = 1;
     string label = 2;
- }
+}
 
  // API-2 types
- message UpdateSuccessRateWindowRequest {
+message UpdateSuccessRateWindowRequest {
     string id = 1;
     string params = 2;
     repeated LabelWithStatus labels_with_status = 3;
     UpdateSuccessRateWindowConfig config = 4;
- }
+}
 
- message LabelWithStatus {
+message LabelWithStatus {
     string label = 1;
     bool status = 2;
- }
+}
 
- message UpdateSuccessRateWindowConfig {
+message UpdateSuccessRateWindowConfig {
     uint32 max_aggregates_size = 1;
     CurrentBlockThreshold current_block_threshold = 2;
- }
+}
 
- message CurrentBlockThreshold {
+message CurrentBlockThreshold {
     optional uint64 duration_in_mins = 1;
     uint64 max_total_count = 2;
- }
+}
 
- message UpdateSuccessRateWindowResponse {
+message UpdateSuccessRateWindowResponse {
     string message = 1;
- }
+}
+
+ // API-3 types
+message InvalidateWindowsRequest {
+    string id = 1;
+}
+
+message InvalidateWindowsResponse {
+    string message = 1;
+}