Issue:
Bug: feat(routing): add invalidate window as a service for SR based routing



add invalidate window as a service for SR based routing

Pull request description:
## Type of Change
<!-- Put an `x` in the boxes that apply -->

- [ ] Bugfix
- [ ] New feature
- [ ] Enhancement
- [ ] Refactoring
- [ ] Dependency updates
- [ ] Documentation
- [ ] CI/CD

## Description
add invalidate window as a service for SR based routing


### 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?
 
 - Hit the route, to update the config
 ```
 curl --location --request PATCH 'http://localhost:8080/account/merchant_1730123706/business_profile/pro_TtvgDfAdgudrReoeeUJb/dynamic_routing/success_based/config/routing_lmPpwoTevEaYMVogesHs' \
--header 'Content-Type: application/json' \
--header 'api-key: test_admin' \
--data '{
  "params": [
    "Currency"
  ],
  "config": {
    "min_aggregates_size": 8,
    "current_block_threshold": {
        "max_total_count": 22
    }
  }
}'
```
-  The old config would get invalidated
<img width="1725" alt="Screenshot 2024-10-28 at 11 25 13 PM" src="https://github.com/user-attachments/assets/54e9c142-9419-4e57-9bb9-36f92938c80a">

## 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


Testing:
 - Hit the route, to update the config
 ```
 curl --location --request PATCH 'http://localhost:8080/account/merchant_1730123706/business_profile/pro_TtvgDfAdgudrReoeeUJb/dynamic_routing/success_based/config/routing_lmPpwoTevEaYMVogesHs' \
--header 'Content-Type: application/json' \
--header 'api-key: test_admin' \
--data '{
  "params": [
    "Currency"
  ],
  "config": {
    "min_aggregates_size": 8,
    "current_block_threshold": {
        "max_total_count": 22
    }
  }
}'
```
-  The old config would get invalidated
<img width="1725" alt="Screenshot 2024-10-28 at 11 25 13 PM" src="https://github.com/user-attachments/assets/54e9c142-9419-4e57-9bb9-36f92938c80a">

Patch:
diff --git a/config/development.toml b/config/development.toml
index 8f3ffdbcbb5..5a6a1250637 100644
--- a/config/development.toml
+++ b/config/development.toml
@@ -775,3 +775,7 @@ card_networks = "Visa, AmericanExpress, Mastercard"
 
 [network_tokenization_supported_connectors]
 connector_list = "cybersource"
+
+[grpc_client.dynamic_routing_client] 
+host = "localhost" 
+port = 7000       
diff --git a/crates/external_services/src/grpc_client/dynamic_routing.rs b/crates/external_services/src/grpc_client/dynamic_routing.rs
index 0546d05ba7c..3264f065b51 100644
--- a/crates/external_services/src/grpc_client/dynamic_routing.rs
+++ b/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,
-    UpdateSuccessRateWindowConfig, UpdateSuccessRateWindowRequest, UpdateSuccessRateWindowResponse,
+    CurrentBlockThreshold as DynamicCurrentThreshold, InvalidateWindowsRequest,
+    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 = client
+        let response = self
+            .clone()
             .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 = client
+        let response = self
+            .clone()
             .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 {
diff --git a/crates/router/src/core/routing.rs b/crates/router/src/core/routing.rs
index 0bd38918ee7..f42bdb4af98 100644
--- a/crates/router/src/core/routing.rs
+++ b/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")]
-use router_env::logger;
-use router_env::metrics::add_attributes;
+#[cfg(all(feature = "v1", feature = "dynamic_routing"))]
+use router_env::{logger, 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")]
@@ -1178,7 +1181,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,
@@ -1375,7 +1378,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,
@@ -1445,6 +1448,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))
 }
 
diff --git a/crates/router/src/routes/app.rs b/crates/router/src/routes/app.rs
index d747f96b9fb..e7f4385612d 100644
--- a/crates/router/src/routes/app.rs
+++ b/crates/router/src/routes/app.rs
@@ -1709,47 +1709,54 @@ 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}")
-                    .service(
-                        web::scope("/dynamic_routing").service(
-                            web::scope("/success_based")
-                                .service(
-                                    web::resource("/toggle").route(
-                                        web::post().to(routing::toggle_success_based_routing),
-                                    ),
-                                )
-                                .service(web::resource("/config/{algorithm_id}").route(
-                                    web::patch().to(|state, req, path, payload| {
-                                        routing::success_based_routing_update_configs(
-                                            state, req, path, payload,
-                                        )
-                                    }),
-                                )),
-                        ),
-                    )
-                    .service(
-                        web::resource("")
-                            .route(web::get().to(profiles::profile_retrieve))
-                            .route(web::post().to(profiles::profile_update))
-                            .route(web::delete().to(profiles::profile_delete)),
-                    )
-                    .service(
-                        web::resource("/toggle_extended_card_info")
-                            .route(web::post().to(profiles::toggle_extended_card_info)),
-                    )
-                    .service(
-                        web::resource("/toggle_connector_agnostic_mit")
-                            .route(web::post().to(profiles::toggle_connector_agnostic_mit)),
+            );
+
+        #[cfg(feature = "dynamic_routing")]
+        {
+            route =
+                route.service(
+                    web::scope("/{profile_id}/dynamic_routing").service(
+                        web::scope("/success_based")
+                            .service(
+                                web::resource("/toggle")
+                                    .route(web::post().to(routing::toggle_success_based_routing)),
+                            )
+                            .service(web::resource("/config/{algorithm_id}").route(
+                                web::patch().to(|state, req, path, payload| {
+                                    routing::success_based_routing_update_configs(
+                                        state, req, path, payload,
+                                    )
+                                }),
+                            )),
                     ),
-            )
+                );
+        }
+
+        route = route.service(
+            web::scope("/{profile_id}")
+                .service(
+                    web::resource("")
+                        .route(web::get().to(profiles::profile_retrieve))
+                        .route(web::post().to(profiles::profile_update))
+                        .route(web::delete().to(profiles::profile_delete)),
+                )
+                .service(
+                    web::resource("/toggle_extended_card_info")
+                        .route(web::post().to(profiles::toggle_extended_card_info)),
+                )
+                .service(
+                    web::resource("/toggle_connector_agnostic_mit")
+                        .route(web::post().to(profiles::toggle_connector_agnostic_mit)),
+                ),
+        );
+
+        route
     }
 }
 
diff --git a/crates/router/src/routes/routing.rs b/crates/router/src/routes/routing.rs
index f0269bd029a..cf6d0371820 100644
--- a/crates/router/src/routes/routing.rs
+++ b/crates/router/src/routes/routing.rs
@@ -1013,7 +1013,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>,
@@ -1056,7 +1056,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>,
diff --git a/proto/success_rate.proto b/proto/success_rate.proto
index 8018f6d5fe4..38e56e36c0f 100644
--- a/proto/success_rate.proto
+++ b/proto/success_rate.proto
@@ -1,57 +1,68 @@
 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;
- }
+service SuccessRateCalculator {
+    rpc FetchSuccessRate (CalSuccessRateRequest) returns (CalSuccessRateResponse);
+
+    rpc UpdateSuccessRateWindow (UpdateSuccessRateWindowRequest) returns (UpdateSuccessRateWindowResponse);
+
+    rpc InvalidateWindows (InvalidateWindowsRequest) returns (InvalidateWindowsResponse);
+}
+
+// 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
+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;
+}
+
+ // API-3 types
+message InvalidateWindowsRequest {
+    string id = 1;
+}
+
+message InvalidateWindowsResponse {
+    string message = 1;
+}
\ No newline at end of file