Bug: Integration of dynamic connector selection in core flows



## Description
This will integrate the dynamic_connector_selection in core flows, enabling the success_based connector selection for payments (with the profiles, enabled with dynamic_connector_selection feature).
diff --git a/crates/api_models/src/routing.rs b/crates/api_models/src/routing.rs
index fc65ab037c2..47d75b2e835 100644
--- a/crates/api_models/src/routing.rs
+++ b/crates/api_models/src/routing.rs
@@ -522,22 +522,51 @@ pub struct DynamicAlgorithmWithTimestamp<T> {
 
 #[derive(Debug, Default, Clone, serde::Serialize, serde::Deserialize)]
 pub struct DynamicRoutingAlgorithmRef {
-    pub success_based_algorithm:
-        Option<DynamicAlgorithmWithTimestamp<common_utils::id_type::RoutingId>>,
+    pub success_based_algorithm: Option<SuccessBasedAlgorithm>,
+}
+
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+pub struct SuccessBasedAlgorithm {
+    pub algorithm_id_with_timestamp:
+        DynamicAlgorithmWithTimestamp<common_utils::id_type::RoutingId>,
+    #[serde(default)]
+    pub enabled_feature: SuccessBasedRoutingFeatures,
+}
+
+impl SuccessBasedAlgorithm {
+    pub fn update_enabled_features(&mut self, feature_to_enable: SuccessBasedRoutingFeatures) {
+        self.enabled_feature = feature_to_enable
+    }
 }
 
 impl DynamicRoutingAlgorithmRef {
-    pub fn update_algorithm_id(&mut self, new_id: common_utils::id_type::RoutingId) {
-        self.success_based_algorithm = Some(DynamicAlgorithmWithTimestamp {
-            algorithm_id: Some(new_id),
-            timestamp: common_utils::date_time::now_unix_timestamp(),
+    pub fn update_algorithm_id(
+        &mut self,
+        new_id: common_utils::id_type::RoutingId,
+        enabled_feature: SuccessBasedRoutingFeatures,
+    ) {
+        self.success_based_algorithm = Some(SuccessBasedAlgorithm {
+            algorithm_id_with_timestamp: DynamicAlgorithmWithTimestamp {
+                algorithm_id: Some(new_id),
+                timestamp: common_utils::date_time::now_unix_timestamp(),
+            },
+            enabled_feature,
         })
     }
 }
 
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize, ToSchema)]
 pub struct ToggleSuccessBasedRoutingQuery {
-    pub status: bool,
+    pub enable: SuccessBasedRoutingFeatures,
+}
+
+#[derive(Debug, Default, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq, ToSchema)]
+#[serde(rename_all = "snake_case")]
+pub enum SuccessBasedRoutingFeatures {
+    Metrics,
+    DynamicConnectorSelection,
+    #[default]
+    None,
 }
 
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize, ToSchema)]
@@ -551,7 +580,7 @@ pub struct SuccessBasedRoutingUpdateConfigQuery {
 #[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
 pub struct ToggleSuccessBasedRoutingWrapper {
     pub profile_id: common_utils::id_type::ProfileId,
-    pub status: bool,
+    pub feature_to_enable: SuccessBasedRoutingFeatures,
 }
 
 #[derive(Clone, Debug, serde::Deserialize, serde::Serialize, ToSchema)]
diff --git a/crates/openapi/src/openapi.rs b/crates/openapi/src/openapi.rs
index a76726eb8ac..b459cee7ad4 100644
--- a/crates/openapi/src/openapi.rs
+++ b/crates/openapi/src/openapi.rs
@@ -159,7 +159,6 @@ Never share your secret api keys. Keep them guarded and secure.
         routes::routing::routing_retrieve_default_config_for_profiles,
         routes::routing::routing_update_default_config_for_profile,
         routes::routing::toggle_success_based_routing,
-        routes::routing::success_based_routing_update_configs,
 
         // Routes for blocklist
         routes::blocklist::remove_entry_from_blocklist,
@@ -559,6 +558,7 @@ Never share your secret api keys. Keep them guarded and secure.
         api_models::routing::RoutingDictionaryRecord,
         api_models::routing::RoutingKind,
         api_models::routing::RoutableConnectorChoice,
+        api_models::routing::SuccessBasedRoutingFeatures,
         api_models::routing::LinkedRoutingConfigRetrieveResponse,
         api_models::routing::RoutingRetrieveResponse,
         api_models::routing::ProfileDefaultRoutingConfig,
@@ -570,11 +570,6 @@ Never share your secret api keys. Keep them guarded and secure.
         api_models::routing::ConnectorVolumeSplit,
         api_models::routing::ConnectorSelection,
         api_models::routing::ToggleSuccessBasedRoutingQuery,
-        api_models::routing::SuccessBasedRoutingConfig,
-        api_models::routing::SuccessBasedRoutingConfigParams,
-        api_models::routing::SuccessBasedRoutingConfigBody,
-        api_models::routing::CurrentBlockThreshold,
-        api_models::routing::SuccessBasedRoutingUpdateConfigQuery,
         api_models::routing::ToggleSuccessBasedRoutingPath,
         api_models::routing::ast::RoutableChoiceKind,
         api_models::enums::RoutableConnectors,
diff --git a/crates/openapi/src/routes/routing.rs b/crates/openapi/src/routes/routing.rs
index 009708d860d..0bb79a2bbe4 100644
--- a/crates/openapi/src/routes/routing.rs
+++ b/crates/openapi/src/routes/routing.rs
@@ -266,7 +266,7 @@ pub async fn routing_update_default_config_for_profile() {}
     params(
         ("account_id" = String, Path, description = "Merchant id"),
         ("profile_id" = String, Path, description = "Profile id under which Dynamic routing needs to be toggled"),
-        ("status" = bool, Query, description = "Boolean value for mentioning the expected state of dynamic routing"),
+        ("enable" = SuccessBasedRoutingFeatures, Query, description = "Feature to enable for success based routing"),
     ),
     responses(
         (status = 200, description = "Routing Algorithm created", body = RoutingDictionaryRecord),
@@ -281,30 +281,3 @@ pub async fn routing_update_default_config_for_profile() {}
    security(("api_key" = []), ("jwt_key" = []))
 )]
 pub async fn toggle_success_based_routing() {}
-
-#[cfg(feature = "v1")]
-/// Routing - Update config for success based dynamic routing
-///
-/// Update config for success based dynamic routing
-#[utoipa::path(
-    patch,
-    path = "/account/:account_id/business_profile/:profile_id/dynamic_routing/success_based/config/:algorithm_id",
-    request_body = SuccessBasedRoutingConfig,
-    params(
-        ("account_id" = String, Path, description = "Merchant id"),
-        ("profile_id" = String, Path, description = "The unique identifier for a profile"),
-        ("algorithm_id" = String, Path, description = "The unique identifier for routing algorithm"),
-    ),
-    responses(
-        (status = 200, description = "Routing Algorithm updated", body = RoutingDictionaryRecord),
-        (status = 400, description = "Request body is malformed"),
-        (status = 500, description = "Internal server error"),
-        (status = 404, description = "Resource missing"),
-        (status = 422, description = "Unprocessable request"),
-        (status = 403, description = "Forbidden"),
-    ),
-   tag = "Routing",
-   operation_id = "Update configs for success based dynamic routing algorithm",
-   security(("admin_api_key" = []))
-)]
-pub async fn success_based_routing_update_configs() {}
diff --git a/crates/router/src/core/errors.rs b/crates/router/src/core/errors.rs
index c56bfaca913..18af1b064cd 100644
--- a/crates/router/src/core/errors.rs
+++ b/crates/router/src/core/errors.rs
@@ -318,6 +318,20 @@ pub enum RoutingError {
     VolumeSplitFailed,
     #[error("Unable to parse metadata")]
     MetadataParsingError,
+    #[error("Unable to retrieve success based routing config")]
+    SuccessBasedRoutingConfigError,
+    #[error("Unable to calculate success based routing config from dynamic routing service")]
+    SuccessRateCalculationError,
+    #[error("Success rate client from dynamic routing gRPC service not initialized")]
+    SuccessRateClientInitializationError,
+    #[error("Unable to convert from '{from}' to '{to}'")]
+    GenericConversionError { from: String, to: String },
+    #[error("Invalid success based connector label received from dynamic routing service: '{0}'")]
+    InvalidSuccessBasedConnectorLabel(String),
+    #[error("unable to find '{field}'")]
+    GenericNotFoundError { field: String },
+    #[error("Unable to deserialize from '{from}' to '{to}'")]
+    DeserializationError { from: String, to: String },
 }
 
 #[derive(Debug, Clone, thiserror::Error)]
diff --git a/crates/router/src/core/payments.rs b/crates/router/src/core/payments.rs
index 9f904116171..568fcb20902 100644
--- a/crates/router/src/core/payments.rs
+++ b/crates/router/src/core/payments.rs
@@ -4591,6 +4591,19 @@ where
     .change_context(errors::ApiErrorResponse::InternalServerError)
     .attach_printable("failed eligibility analysis and fallback")?;
 
+    // dynamic success based connector selection
+    #[cfg(all(feature = "v1", feature = "dynamic_routing"))]
+    let connectors = {
+        if business_profile.dynamic_routing_algorithm.is_some() {
+            routing::perform_success_based_routing(state, connectors.clone(), business_profile)
+                .await
+                .map_err(|e| logger::error!(success_rate_routing_error=?e))
+                .unwrap_or(connectors)
+        } else {
+            connectors
+        }
+    };
+
     let connector_data = connectors
         .into_iter()
         .map(|conn| {
diff --git a/crates/router/src/core/payments/operations/payment_response.rs b/crates/router/src/core/payments/operations/payment_response.rs
index 6a5299db8c4..8a6fad6a60a 100644
--- a/crates/router/src/core/payments/operations/payment_response.rs
+++ b/crates/router/src/core/payments/operations/payment_response.rs
@@ -1713,8 +1713,7 @@ async fn payment_response_update_tracker<F: Clone, T: types::Capturable>(
 
     #[cfg(all(feature = "v1", feature = "dynamic_routing"))]
     {
-        if let Some(dynamic_routing_algorithm) = business_profile.dynamic_routing_algorithm.clone()
-        {
+        if business_profile.dynamic_routing_algorithm.is_some() {
             let state = state.clone();
             let business_profile = business_profile.clone();
             let payment_attempt = payment_attempt.clone();
@@ -1725,7 +1724,6 @@ async fn payment_response_update_tracker<F: Clone, T: types::Capturable>(
                         &payment_attempt,
                         routable_connectors,
                         &business_profile,
-                        dynamic_routing_algorithm,
                     )
                     .await
                     .map_err(|e| logger::error!(dynamic_routing_metrics_error=?e))
diff --git a/crates/router/src/core/payments/routing.rs b/crates/router/src/core/payments/routing.rs
index 04080a42730..dbb08e770a1 100644
--- a/crates/router/src/core/payments/routing.rs
+++ b/crates/router/src/core/payments/routing.rs
@@ -7,6 +7,8 @@ use std::{
     sync::Arc,
 };
 
+#[cfg(all(feature = "v1", feature = "dynamic_routing"))]
+use api_models::routing as api_routing;
 use api_models::{
     admin as admin_api,
     enums::{self as api_enums, CountryAlpha2},
@@ -21,6 +23,10 @@ use euclid::{
     enums as euclid_enums,
     frontend::{ast, dir as euclid_dir},
 };
+#[cfg(all(feature = "v1", feature = "dynamic_routing"))]
+use external_services::grpc_client::dynamic_routing::{
+    success_rate::CalSuccessRateResponse, SuccessBasedDynamicRouting,
+};
 use kgraph_utils::{
     mca as mca_graph,
     transformers::{IntoContext, IntoDirValue},
@@ -1202,3 +1208,114 @@ pub fn make_dsl_input_for_surcharge(
     };
     Ok(backend_input)
 }
+
+/// success based dynamic routing
+#[cfg(all(feature = "v1", feature = "dynamic_routing"))]
+pub async fn perform_success_based_routing(
+    state: &SessionState,
+    routable_connectors: Vec<api_routing::RoutableConnectorChoice>,
+    business_profile: &domain::Profile,
+) -> RoutingResult<Vec<api_routing::RoutableConnectorChoice>> {
+    let success_based_dynamic_routing_algo_ref: api_routing::DynamicRoutingAlgorithmRef =
+        business_profile
+            .dynamic_routing_algorithm
+            .clone()
+            .map(|val| val.parse_value("DynamicRoutingAlgorithmRef"))
+            .transpose()
+            .change_context(errors::RoutingError::DeserializationError {
+                from: "JSON".to_string(),
+                to: "DynamicRoutingAlgorithmRef".to_string(),
+            })
+            .attach_printable("unable to deserialize DynamicRoutingAlgorithmRef from JSON")?
+            .unwrap_or_default();
+
+    let success_based_algo_ref = success_based_dynamic_routing_algo_ref
+        .success_based_algorithm
+        .ok_or(errors::RoutingError::GenericNotFoundError { field: "success_based_algorithm".to_string() })
+        .attach_printable(
+            "success_based_algorithm not found in dynamic_routing_algorithm from business_profile table",
+        )?;
+
+    if success_based_algo_ref.enabled_feature
+        == api_routing::SuccessBasedRoutingFeatures::DynamicConnectorSelection
+    {
+        logger::debug!(
+            "performing success_based_routing for profile {}",
+            business_profile.get_id().get_string_repr()
+        );
+        let client = state
+            .grpc_client
+            .dynamic_routing
+            .success_rate_client
+            .as_ref()
+            .ok_or(errors::RoutingError::SuccessRateClientInitializationError)
+            .attach_printable("success_rate gRPC client not found")?;
+
+        let success_based_routing_configs = routing::helpers::fetch_success_based_routing_configs(
+            state,
+            business_profile,
+            success_based_algo_ref
+                .algorithm_id_with_timestamp
+                .algorithm_id
+                .ok_or(errors::RoutingError::GenericNotFoundError {
+                    field: "success_based_routing_algorithm_id".to_string(),
+                })
+                .attach_printable(
+                    "success_based_routing_algorithm_id not found in business_profile",
+                )?,
+        )
+        .await
+        .change_context(errors::RoutingError::SuccessBasedRoutingConfigError)
+        .attach_printable("unable to fetch success_rate based dynamic routing configs")?;
+
+        let tenant_business_profile_id = routing::helpers::generate_tenant_business_profile_id(
+            &state.tenant.redis_key_prefix,
+            business_profile.get_id().get_string_repr(),
+        );
+
+        let success_based_connectors: CalSuccessRateResponse = client
+            .calculate_success_rate(
+                tenant_business_profile_id,
+                success_based_routing_configs,
+                routable_connectors,
+            )
+            .await
+            .change_context(errors::RoutingError::SuccessRateCalculationError)
+            .attach_printable(
+                "unable to calculate/fetch success rate from dynamic routing service",
+            )?;
+
+        let mut connectors = Vec::with_capacity(success_based_connectors.labels_with_score.len());
+        for label_with_score in success_based_connectors.labels_with_score {
+            let (connector, merchant_connector_id) = label_with_score.label
+                .split_once(':')
+                .ok_or(errors::RoutingError::InvalidSuccessBasedConnectorLabel(label_with_score.label.to_string()))
+                .attach_printable(
+                    "unable to split connector_name and mca_id from the label obtained by the dynamic routing service",
+                )?;
+            connectors.push(api_routing::RoutableConnectorChoice {
+                choice_kind: api_routing::RoutableChoiceKind::FullStruct,
+                connector: common_enums::RoutableConnectors::from_str(connector)
+                    .change_context(errors::RoutingError::GenericConversionError {
+                        from: "String".to_string(),
+                        to: "RoutableConnectors".to_string(),
+                    })
+                    .attach_printable("unable to convert String to RoutableConnectors")?,
+                merchant_connector_id: Some(
+                    common_utils::id_type::MerchantConnectorAccountId::wrap(
+                        merchant_connector_id.to_string(),
+                    )
+                    .change_context(errors::RoutingError::GenericConversionError {
+                        from: "String".to_string(),
+                        to: "MerchantConnectorAccountId".to_string(),
+                    })
+                    .attach_printable("unable to convert MerchantConnectorAccountId from string")?,
+                ),
+            });
+        }
+        logger::debug!(success_based_routing_connectors=?connectors);
+        Ok(connectors)
+    } else {
+        Ok(routable_connectors)
+    }
+}
diff --git a/crates/router/src/core/routing.rs b/crates/router/src/core/routing.rs
index b7b1c13ea80..35c4b8d9621 100644
--- a/crates/router/src/core/routing.rs
+++ b/crates/router/src/core/routing.rs
@@ -434,9 +434,13 @@ pub async fn link_routing_config(
             utils::when(
                 matches!(
                     dynamic_routing_ref.success_based_algorithm,
-                    Some(routing_types::DynamicAlgorithmWithTimestamp {
-                        algorithm_id: Some(ref id),
-                        timestamp: _
+                    Some(routing::SuccessBasedAlgorithm {
+                        algorithm_id_with_timestamp:
+                        routing_types::DynamicAlgorithmWithTimestamp {
+                            algorithm_id: Some(ref id),
+                            timestamp: _
+                        },
+                        enabled_feature: _
                     }) if id == &algorithm_id
                 ),
                 || {
@@ -446,7 +450,17 @@ pub async fn link_routing_config(
                 },
             )?;
 
-            dynamic_routing_ref.update_algorithm_id(algorithm_id);
+            dynamic_routing_ref.update_algorithm_id(
+                algorithm_id,
+                dynamic_routing_ref
+                    .success_based_algorithm
+                    .clone()
+                    .ok_or(errors::ApiErrorResponse::InternalServerError)
+                    .attach_printable(
+                        "missing success_based_algorithm in dynamic_algorithm_ref from business_profile table",
+                    )?
+                    .enabled_feature,
+            );
             helpers::update_business_profile_active_dynamic_algorithm_ref(
                 db,
                 key_manager_state,
@@ -1162,7 +1176,7 @@ pub async fn toggle_success_based_routing(
     state: SessionState,
     merchant_account: domain::MerchantAccount,
     key_store: domain::MerchantKeyStore,
-    status: bool,
+    feature_to_enable: routing::SuccessBasedRoutingFeatures,
     profile_id: common_utils::id_type::ProfileId,
 ) -> RouterResponse<routing_types::RoutingDictionaryRecord> {
     metrics::ROUTING_CREATE_REQUEST_RECEIVED.add(
@@ -1198,115 +1212,158 @@ pub async fn toggle_success_based_routing(
             )?
             .unwrap_or_default();
 
-    if status {
-        let default_success_based_routing_config = routing::SuccessBasedRoutingConfig::default();
-        let algorithm_id = common_utils::generate_routing_id_of_default_length();
-        let timestamp = common_utils::date_time::now();
-        let algo = RoutingAlgorithm {
-            algorithm_id: algorithm_id.clone(),
-            profile_id: business_profile.get_id().to_owned(),
-            merchant_id: merchant_account.get_id().to_owned(),
-            name: "Dynamic routing algorithm".to_string(),
-            description: None,
-            kind: diesel_models::enums::RoutingAlgorithmKind::Dynamic,
-            algorithm_data: serde_json::json!(default_success_based_routing_config),
-            created_at: timestamp,
-            modified_at: timestamp,
-            algorithm_for: common_enums::TransactionType::Payment,
-        };
-
-        let record = db
-            .insert_routing_algorithm(algo)
-            .await
-            .change_context(errors::ApiErrorResponse::InternalServerError)
-            .attach_printable("Unable to insert record in routing algorithm table")?;
-
-        success_based_dynamic_routing_algo_ref.update_algorithm_id(algorithm_id);
-        helpers::update_business_profile_active_dynamic_algorithm_ref(
-            db,
-            key_manager_state,
-            &key_store,
-            business_profile,
-            success_based_dynamic_routing_algo_ref,
-        )
-        .await?;
-
-        let new_record = record.foreign_into();
-
-        metrics::ROUTING_CREATE_SUCCESS_RESPONSE.add(
-            &metrics::CONTEXT,
-            1,
-            &add_attributes([("profile_id", profile_id.get_string_repr().to_owned())]),
-        );
-        Ok(service_api::ApplicationResponse::Json(new_record))
-    } else {
-        let timestamp = common_utils::date_time::now_unix_timestamp();
-        match success_based_dynamic_routing_algo_ref.success_based_algorithm {
-            Some(algorithm_ref) => {
-                if let Some(algorithm_id) = algorithm_ref.algorithm_id {
-                    let dynamic_routing_algorithm = routing_types::DynamicRoutingAlgorithmRef {
-                        success_based_algorithm: Some(
-                            routing_types::DynamicAlgorithmWithTimestamp {
-                                algorithm_id: None,
-                                timestamp,
-                            },
-                        ),
-                    };
-
-                    // redact cache for success based routing configs
-                    let cache_key = format!(
-                        "{}_{}",
-                        business_profile.get_id().get_string_repr(),
-                        algorithm_id.get_string_repr()
-                    );
-                    let cache_entries_to_redact =
-                        vec![cache::CacheKind::SuccessBasedDynamicRoutingCache(
-                            cache_key.into(),
-                        )];
-                    let _ = cache::publish_into_redact_channel(
-                        state.store.get_cache_store().as_ref(),
-                        cache_entries_to_redact,
-                    )
-                    .await
-                    .map_err(|e| {
-                        logger::error!(
-                            "unable to publish into the redact channel for evicting the success based routing config cache {e:?}"
+    match feature_to_enable {
+        routing::SuccessBasedRoutingFeatures::Metrics
+        | routing::SuccessBasedRoutingFeatures::DynamicConnectorSelection => {
+            if let Some(ref mut algo_with_timestamp) =
+                success_based_dynamic_routing_algo_ref.success_based_algorithm
+            {
+                match algo_with_timestamp
+                    .algorithm_id_with_timestamp
+                    .algorithm_id
+                    .clone()
+                {
+                    Some(algorithm_id) => {
+                        // algorithm is already present in profile
+                        if algo_with_timestamp.enabled_feature == feature_to_enable {
+                            // algorithm already has the required feature
+                            Err(errors::ApiErrorResponse::PreconditionFailed {
+                                message: "Success rate based routing is already enabled"
+                                    .to_string(),
+                            })?
+                        } else {
+                            // enable the requested feature for the algorithm
+                            algo_with_timestamp.update_enabled_features(feature_to_enable);
+                            let record = db
+                                .find_routing_algorithm_by_profile_id_algorithm_id(
+                                    business_profile.get_id(),
+                                    &algorithm_id,
+                                )
+                                .await
+                                .to_not_found_response(
+                                    errors::ApiErrorResponse::ResourceIdNotFound,
+                                )?;
+                            let response = record.foreign_into();
+                            helpers::update_business_profile_active_dynamic_algorithm_ref(
+                                db,
+                                key_manager_state,
+                                &key_store,
+                                business_profile,
+                                success_based_dynamic_routing_algo_ref,
+                            )
+                            .await?;
+
+                            metrics::ROUTING_CREATE_SUCCESS_RESPONSE.add(
+                                &metrics::CONTEXT,
+                                1,
+                                &add_attributes([(
+                                    "profile_id",
+                                    profile_id.get_string_repr().to_owned(),
+                                )]),
+                            );
+                            Ok(service_api::ApplicationResponse::Json(response))
+                        }
+                    }
+                    None => {
+                        // algorithm isn't present in profile
+                        helpers::default_success_based_routing_setup(
+                            &state,
+                            key_store,
+                            business_profile,
+                            feature_to_enable,
+                            merchant_account.get_id().to_owned(),
+                            success_based_dynamic_routing_algo_ref,
                         )
-                    });
+                        .await
+                    }
+                }
+            } else {
+                // algorithm isn't present in profile
+                helpers::default_success_based_routing_setup(
+                    &state,
+                    key_store,
+                    business_profile,
+                    feature_to_enable,
+                    merchant_account.get_id().to_owned(),
+                    success_based_dynamic_routing_algo_ref,
+                )
+                .await
+            }
+        }
+        routing::SuccessBasedRoutingFeatures::None => {
+            // disable success based routing for the requested profile
+            let timestamp = common_utils::date_time::now_unix_timestamp();
+            match success_based_dynamic_routing_algo_ref.success_based_algorithm {
+                Some(algorithm_ref) => {
+                    if let Some(algorithm_id) =
+                        algorithm_ref.algorithm_id_with_timestamp.algorithm_id
+                    {
+                        let dynamic_routing_algorithm = routing_types::DynamicRoutingAlgorithmRef {
+                            success_based_algorithm: Some(routing::SuccessBasedAlgorithm {
+                                algorithm_id_with_timestamp:
+                                    routing_types::DynamicAlgorithmWithTimestamp {
+                                        algorithm_id: None,
+                                        timestamp,
+                                    },
+                                enabled_feature: routing::SuccessBasedRoutingFeatures::None,
+                            }),
+                        };
 
-                    let record = db
-                        .find_routing_algorithm_by_profile_id_algorithm_id(
-                            business_profile.get_id(),
-                            &algorithm_id,
+                        // redact cache for success based routing configs
+                        let cache_key = format!(
+                            "{}_{}",
+                            business_profile.get_id().get_string_repr(),
+                            algorithm_id.get_string_repr()
+                        );
+                        let cache_entries_to_redact =
+                            vec![cache::CacheKind::SuccessBasedDynamicRoutingCache(
+                                cache_key.into(),
+                            )];
+                        let _ = cache::publish_into_redact_channel(
+                            state.store.get_cache_store().as_ref(),
+                            cache_entries_to_redact,
                         )
                         .await
-                        .to_not_found_response(errors::ApiErrorResponse::ResourceIdNotFound)?;
-                    let response = record.foreign_into();
-                    helpers::update_business_profile_active_dynamic_algorithm_ref(
-                        db,
-                        key_manager_state,
-                        &key_store,
-                        business_profile,
-                        dynamic_routing_algorithm,
-                    )
-                    .await?;
-
-                    metrics::ROUTING_UNLINK_CONFIG_SUCCESS_RESPONSE.add(
-                        &metrics::CONTEXT,
-                        1,
-                        &add_attributes([("profile_id", profile_id.get_string_repr().to_owned())]),
-                    );
-
-                    Ok(service_api::ApplicationResponse::Json(response))
-                } else {
-                    Err(errors::ApiErrorResponse::PreconditionFailed {
-                        message: "Algorithm is already inactive".to_string(),
-                    })?
+                        .change_context(errors::ApiErrorResponse::InternalServerError)
+                        .attach_printable("unable to publish into the redact channel for evicting the success based routing config cache")?;
+
+                        let record = db
+                            .find_routing_algorithm_by_profile_id_algorithm_id(
+                                business_profile.get_id(),
+                                &algorithm_id,
+                            )
+                            .await
+                            .to_not_found_response(errors::ApiErrorResponse::ResourceIdNotFound)?;
+                        let response = record.foreign_into();
+                        helpers::update_business_profile_active_dynamic_algorithm_ref(
+                            db,
+                            key_manager_state,
+                            &key_store,
+                            business_profile,
+                            dynamic_routing_algorithm,
+                        )
+                        .await?;
+
+                        metrics::ROUTING_UNLINK_CONFIG_SUCCESS_RESPONSE.add(
+                            &metrics::CONTEXT,
+                            1,
+                            &add_attributes([(
+                                "profile_id",
+                                profile_id.get_string_repr().to_owned(),
+                            )]),
+                        );
+
+                        Ok(service_api::ApplicationResponse::Json(response))
+                    } else {
+                        Err(errors::ApiErrorResponse::PreconditionFailed {
+                            message: "Algorithm is already inactive".to_string(),
+                        })?
+                    }
                 }
+                None => Err(errors::ApiErrorResponse::PreconditionFailed {
+                    message: "Success rate based routing is already disabled".to_string(),
+                })?,
             }
-            None => Err(errors::ApiErrorResponse::PreconditionFailed {
-                message: "Algorithm is already inactive".to_string(),
-            })?,
         }
     }
 }
diff --git a/crates/router/src/core/routing/helpers.rs b/crates/router/src/core/routing/helpers.rs
index 22966208dd4..6b584583174 100644
--- a/crates/router/src/core/routing/helpers.rs
+++ b/crates/router/src/core/routing/helpers.rs
@@ -12,10 +12,16 @@ use api_models::routing as routing_types;
 use common_utils::ext_traits::ValueExt;
 use common_utils::{ext_traits::Encode, id_type, types::keymanager::KeyManagerState};
 use diesel_models::configs;
+#[cfg(feature = "v1")]
+use diesel_models::routing_algorithm;
 use error_stack::ResultExt;
-#[cfg(feature = "dynamic_routing")]
+#[cfg(all(feature = "dynamic_routing", feature = "v1"))]
 use external_services::grpc_client::dynamic_routing::SuccessBasedDynamicRouting;
+#[cfg(feature = "v1")]
+use hyperswitch_domain_models::api::ApplicationResponse;
 #[cfg(all(feature = "dynamic_routing", feature = "v1"))]
+use router_env::logger;
+#[cfg(any(feature = "dynamic_routing", feature = "v1"))]
 use router_env::{instrument, metrics::add_attributes, tracing};
 use rustc_hash::FxHashSet;
 use storage_impl::redis::cache;
@@ -29,8 +35,10 @@ use crate::{
     types::{domain, storage},
     utils::StringExt,
 };
-#[cfg(all(feature = "dynamic_routing", feature = "v1"))]
-use crate::{core::metrics as core_metrics, routes::metrics};
+#[cfg(feature = "v1")]
+use crate::{core::metrics as core_metrics, routes::metrics, types::transformers::ForeignInto};
+pub const SUCCESS_BASED_DYNAMIC_ROUTING_ALGORITHM: &str =
+    "Success rate based dynamic routing algorithm";
 
 /// Provides us with all the configured configs of the Merchant in the ascending time configured
 /// manner and chooses the first of them
@@ -594,28 +602,8 @@ pub async fn refresh_success_based_routing_cache(
 pub async fn fetch_success_based_routing_configs(
     state: &SessionState,
     business_profile: &domain::Profile,
-    dynamic_routing_algorithm: serde_json::Value,
+    success_based_routing_id: id_type::RoutingId,
 ) -> RouterResult<routing_types::SuccessBasedRoutingConfig> {
-    let dynamic_routing_algorithm_ref = dynamic_routing_algorithm
-        .parse_value::<routing_types::DynamicRoutingAlgorithmRef>("DynamicRoutingAlgorithmRef")
-        .change_context(errors::ApiErrorResponse::InternalServerError)
-        .attach_printable("unable to parse dynamic_routing_algorithm_ref")?;
-
-    let success_based_routing_id = dynamic_routing_algorithm_ref
-        .success_based_algorithm
-        .ok_or(errors::ApiErrorResponse::GenericNotFoundError {
-            message: "success_based_algorithm not found in dynamic_routing_algorithm_ref"
-                .to_string(),
-        })?
-        .algorithm_id
-        // error can be possible when the feature is toggled off.
-        .ok_or(errors::ApiErrorResponse::GenericNotFoundError {
-            message: format!(
-                "unable to find algorithm_id in success based algorithm config as the feature is disabled for profile_id: {}",
-                business_profile.get_id().get_string_repr()
-            ),
-        })?;
-
     let key = format!(
         "{}_{}",
         business_profile.get_id().get_string_repr(),
@@ -657,156 +645,185 @@ pub async fn push_metrics_for_success_based_routing(
     payment_attempt: &storage::PaymentAttempt,
     routable_connectors: Vec<routing_types::RoutableConnectorChoice>,
     business_profile: &domain::Profile,
-    dynamic_routing_algorithm: serde_json::Value,
 ) -> RouterResult<()> {
-    let client = state
-        .grpc_client
-        .dynamic_routing
-        .success_rate_client
-        .as_ref()
-        .ok_or(errors::ApiErrorResponse::GenericNotFoundError {
-            message: "success_rate gRPC client not found".to_string(),
-        })?;
-
-    let payment_connector = &payment_attempt.connector.clone().ok_or(
-        errors::ApiErrorResponse::GenericNotFoundError {
-            message: "unable to derive payment connector from payment attempt".to_string(),
-        },
-    )?;
-
-    let success_based_routing_configs =
-        fetch_success_based_routing_configs(state, business_profile, dynamic_routing_algorithm)
-            .await
+    let success_based_dynamic_routing_algo_ref: routing_types::DynamicRoutingAlgorithmRef =
+        business_profile
+            .dynamic_routing_algorithm
+            .clone()
+            .map(|val| val.parse_value("DynamicRoutingAlgorithmRef"))
+            .transpose()
             .change_context(errors::ApiErrorResponse::InternalServerError)
-            .attach_printable("unable to retrieve success_rate based dynamic routing configs")?;
+            .attach_printable("Failed to deserialize DynamicRoutingAlgorithmRef from JSON")?
+            .unwrap_or_default();
 
-    let tenant_business_profile_id = format!(
-        "{}:{}",
-        state.tenant.redis_key_prefix,
-        business_profile.get_id().get_string_repr()
-    );
+    let success_based_algo_ref = success_based_dynamic_routing_algo_ref
+        .success_based_algorithm
+        .ok_or(errors::ApiErrorResponse::InternalServerError)
+        .attach_printable("success_based_algorithm not found in dynamic_routing_algorithm from business_profile table")?;
+
+    if success_based_algo_ref.enabled_feature != routing_types::SuccessBasedRoutingFeatures::None {
+        let client = state
+            .grpc_client
+            .dynamic_routing
+            .success_rate_client
+            .as_ref()
+            .ok_or(errors::ApiErrorResponse::GenericNotFoundError {
+                message: "success_rate gRPC client not found".to_string(),
+            })?;
+
+        let payment_connector = &payment_attempt.connector.clone().ok_or(
+            errors::ApiErrorResponse::GenericNotFoundError {
+                message: "unable to derive payment connector from payment attempt".to_string(),
+            },
+        )?;
 
-    let success_based_connectors = client
-        .calculate_success_rate(
-            tenant_business_profile_id.clone(),
-            success_based_routing_configs.clone(),
-            routable_connectors.clone(),
+        let success_based_routing_configs = fetch_success_based_routing_configs(
+            state,
+            business_profile,
+            success_based_algo_ref
+                .algorithm_id_with_timestamp
+                .algorithm_id
+                .ok_or(errors::ApiErrorResponse::InternalServerError)
+                .attach_printable(
+                    "success_based_routing_algorithm_id not found in business_profile",
+                )?,
         )
         .await
         .change_context(errors::ApiErrorResponse::InternalServerError)
-        .attach_printable("unable to calculate/fetch success rate from dynamic routing service")?;
-
-    let payment_status_attribute =
-        get_desired_payment_status_for_success_routing_metrics(&payment_attempt.status);
-
-    let first_success_based_connector_label = &success_based_connectors
-        .labels_with_score
-        .first()
-        .ok_or(errors::ApiErrorResponse::InternalServerError)
-        .attach_printable(
-            "unable to fetch the first connector from list of connectors obtained from dynamic routing service",
-        )?
-        .label
-        .to_string();
-
-    let (first_success_based_connector, merchant_connector_id) = first_success_based_connector_label
-        .split_once(':')
-        .ok_or(errors::ApiErrorResponse::InternalServerError)
-        .attach_printable(
-            "unable to split connector_name and mca_id from the first connector obtained from dynamic routing service",
-        )?;
-
-    let outcome = get_success_based_metrics_outcome_for_payment(
-        &payment_status_attribute,
-        payment_connector.to_string(),
-        first_success_based_connector.to_string(),
-    );
-
-    core_metrics::DYNAMIC_SUCCESS_BASED_ROUTING.add(
-        &metrics::CONTEXT,
-        1,
-        &add_attributes([
-            ("tenant", state.tenant.name.clone()),
-            (
-                "merchant_id",
-                payment_attempt.merchant_id.get_string_repr().to_string(),
-            ),
-            (
-                "profile_id",
-                payment_attempt.profile_id.get_string_repr().to_string(),
-            ),
-            ("merchant_connector_id", merchant_connector_id.to_string()),
-            (
-                "payment_id",
-                payment_attempt.payment_id.get_string_repr().to_string(),
-            ),
-            (
-                "success_based_routing_connector",
-                first_success_based_connector.to_string(),
-            ),
-            ("payment_connector", payment_connector.to_string()),
-            (
-                "currency",
-                payment_attempt
-                    .currency
-                    .map_or_else(|| "None".to_string(), |currency| currency.to_string()),
-            ),
-            (
-                "payment_method",
-                payment_attempt.payment_method.map_or_else(
-                    || "None".to_string(),
-                    |payment_method| payment_method.to_string(),
+        .attach_printable("unable to retrieve success_rate based dynamic routing configs")?;
+
+        let tenant_business_profile_id = generate_tenant_business_profile_id(
+            &state.tenant.redis_key_prefix,
+            business_profile.get_id().get_string_repr(),
+        );
+
+        let success_based_connectors = client
+            .calculate_success_rate(
+                tenant_business_profile_id.clone(),
+                success_based_routing_configs.clone(),
+                routable_connectors.clone(),
+            )
+            .await
+            .change_context(errors::ApiErrorResponse::InternalServerError)
+            .attach_printable(
+                "unable to calculate/fetch success rate from dynamic routing service",
+            )?;
+
+        let payment_status_attribute =
+            get_desired_payment_status_for_success_routing_metrics(&payment_attempt.status);
+
+        let first_success_based_connector_label = &success_based_connectors
+            .labels_with_score
+            .first()
+            .ok_or(errors::ApiErrorResponse::InternalServerError)
+            .attach_printable(
+                "unable to fetch the first connector from list of connectors obtained from dynamic routing service",
+            )?
+            .label
+            .to_string();
+
+        let (first_success_based_connector, merchant_connector_id) = first_success_based_connector_label
+            .split_once(':')
+            .ok_or(errors::ApiErrorResponse::InternalServerError)
+            .attach_printable(
+                "unable to split connector_name and mca_id from the first connector obtained from dynamic routing service",
+            )?;
+
+        let outcome = get_success_based_metrics_outcome_for_payment(
+            &payment_status_attribute,
+            payment_connector.to_string(),
+            first_success_based_connector.to_string(),
+        );
+
+        core_metrics::DYNAMIC_SUCCESS_BASED_ROUTING.add(
+            &metrics::CONTEXT,
+            1,
+            &add_attributes([
+                ("tenant", state.tenant.name.clone()),
+                (
+                    "merchant_id",
+                    payment_attempt.merchant_id.get_string_repr().to_string(),
                 ),
-            ),
-            (
-                "payment_method_type",
-                payment_attempt.payment_method_type.map_or_else(
-                    || "None".to_string(),
-                    |payment_method_type| payment_method_type.to_string(),
+                (
+                    "profile_id",
+                    payment_attempt.profile_id.get_string_repr().to_string(),
                 ),
-            ),
-            (
-                "capture_method",
-                payment_attempt.capture_method.map_or_else(
-                    || "None".to_string(),
-                    |capture_method| capture_method.to_string(),
+                ("merchant_connector_id", merchant_connector_id.to_string()),
+                (
+                    "payment_id",
+                    payment_attempt.payment_id.get_string_repr().to_string(),
                 ),
-            ),
-            (
-                "authentication_type",
-                payment_attempt.authentication_type.map_or_else(
-                    || "None".to_string(),
-                    |authentication_type| authentication_type.to_string(),
+                (
+                    "success_based_routing_connector",
+                    first_success_based_connector.to_string(),
                 ),
-            ),
-            ("payment_status", payment_attempt.status.to_string()),
-            ("conclusive_classification", outcome.to_string()),
-        ]),
-    );
-
-    client
-        .update_success_rate(
-            tenant_business_profile_id,
-            success_based_routing_configs,
-            vec![routing_types::RoutableConnectorChoiceWithStatus::new(
-                routing_types::RoutableConnectorChoice {
-                    choice_kind: api_models::routing::RoutableChoiceKind::FullStruct,
-                    connector: common_enums::RoutableConnectors::from_str(
-                        payment_connector.as_str(),
-                    )
-                    .change_context(errors::ApiErrorResponse::InternalServerError)
-                    .attach_printable("unable to infer routable_connector from connector")?,
-                    merchant_connector_id: payment_attempt.merchant_connector_id.clone(),
-                },
-                payment_status_attribute == common_enums::AttemptStatus::Charged,
-            )],
-        )
-        .await
-        .change_context(errors::ApiErrorResponse::InternalServerError)
-        .attach_printable(
-            "unable to update success based routing window in dynamic routing service",
-        )?;
-    Ok(())
+                ("payment_connector", payment_connector.to_string()),
+                (
+                    "currency",
+                    payment_attempt
+                        .currency
+                        .map_or_else(|| "None".to_string(), |currency| currency.to_string()),
+                ),
+                (
+                    "payment_method",
+                    payment_attempt.payment_method.map_or_else(
+                        || "None".to_string(),
+                        |payment_method| payment_method.to_string(),
+                    ),
+                ),
+                (
+                    "payment_method_type",
+                    payment_attempt.payment_method_type.map_or_else(
+                        || "None".to_string(),
+                        |payment_method_type| payment_method_type.to_string(),
+                    ),
+                ),
+                (
+                    "capture_method",
+                    payment_attempt.capture_method.map_or_else(
+                        || "None".to_string(),
+                        |capture_method| capture_method.to_string(),
+                    ),
+                ),
+                (
+                    "authentication_type",
+                    payment_attempt.authentication_type.map_or_else(
+                        || "None".to_string(),
+                        |authentication_type| authentication_type.to_string(),
+                    ),
+                ),
+                ("payment_status", payment_attempt.status.to_string()),
+                ("conclusive_classification", outcome.to_string()),
+            ]),
+        );
+        logger::debug!("successfully pushed success_based_routing metrics");
+
+        client
+            .update_success_rate(
+                tenant_business_profile_id,
+                success_based_routing_configs,
+                vec![routing_types::RoutableConnectorChoiceWithStatus::new(
+                    routing_types::RoutableConnectorChoice {
+                        choice_kind: api_models::routing::RoutableChoiceKind::FullStruct,
+                        connector: common_enums::RoutableConnectors::from_str(
+                            payment_connector.as_str(),
+                        )
+                        .change_context(errors::ApiErrorResponse::InternalServerError)
+                        .attach_printable("unable to infer routable_connector from connector")?,
+                        merchant_connector_id: payment_attempt.merchant_connector_id.clone(),
+                    },
+                    payment_status_attribute == common_enums::AttemptStatus::Charged,
+                )],
+            )
+            .await
+            .change_context(errors::ApiErrorResponse::InternalServerError)
+            .attach_printable(
+                "unable to update success based routing window in dynamic routing service",
+            )?;
+        Ok(())
+    } else {
+        Ok(())
+    }
 }
 
 #[cfg(all(feature = "v1", feature = "dynamic_routing"))]
@@ -875,3 +892,67 @@ fn get_success_based_metrics_outcome_for_payment(
         _ => common_enums::SuccessBasedRoutingConclusiveState::NonDeterministic,
     }
 }
+
+/// generates cache key with tenant's redis key prefix and profile_id
+pub fn generate_tenant_business_profile_id(
+    redis_key_prefix: &str,
+    business_profile_id: &str,
+) -> String {
+    format!("{}:{}", redis_key_prefix, business_profile_id)
+}
+
+/// default config setup for success_based_routing
+#[cfg(feature = "v1")]
+#[instrument(skip_all)]
+pub async fn default_success_based_routing_setup(
+    state: &SessionState,
+    key_store: domain::MerchantKeyStore,
+    business_profile: domain::Profile,
+    feature_to_enable: routing_types::SuccessBasedRoutingFeatures,
+    merchant_id: id_type::MerchantId,
+    mut success_based_dynamic_routing_algo: routing_types::DynamicRoutingAlgorithmRef,
+) -> RouterResult<ApplicationResponse<routing_types::RoutingDictionaryRecord>> {
+    let db = state.store.as_ref();
+    let key_manager_state = &state.into();
+    let profile_id = business_profile.get_id().to_owned();
+    let default_success_based_routing_config = routing_types::SuccessBasedRoutingConfig::default();
+    let algorithm_id = common_utils::generate_routing_id_of_default_length();
+    let timestamp = common_utils::date_time::now();
+    let algo = routing_algorithm::RoutingAlgorithm {
+        algorithm_id: algorithm_id.clone(),
+        profile_id: profile_id.clone(),
+        merchant_id,
+        name: SUCCESS_BASED_DYNAMIC_ROUTING_ALGORITHM.to_string(),
+        description: None,
+        kind: diesel_models::enums::RoutingAlgorithmKind::Dynamic,
+        algorithm_data: serde_json::json!(default_success_based_routing_config),
+        created_at: timestamp,
+        modified_at: timestamp,
+        algorithm_for: common_enums::TransactionType::Payment,
+    };
+
+    let record = db
+        .insert_routing_algorithm(algo)
+        .await
+        .change_context(errors::ApiErrorResponse::InternalServerError)
+        .attach_printable("Unable to insert record in routing algorithm table")?;
+
+    success_based_dynamic_routing_algo.update_algorithm_id(algorithm_id, feature_to_enable);
+    update_business_profile_active_dynamic_algorithm_ref(
+        db,
+        key_manager_state,
+        &key_store,
+        business_profile,
+        success_based_dynamic_routing_algo,
+    )
+    .await?;
+
+    let new_record = record.foreign_into();
+
+    core_metrics::ROUTING_CREATE_SUCCESS_RESPONSE.add(
+        &metrics::CONTEXT,
+        1,
+        &add_attributes([("profile_id", profile_id.get_string_repr().to_string())]),
+    );
+    Ok(ApplicationResponse::Json(new_record))
+}
diff --git a/crates/router/src/routes/routing.rs b/crates/router/src/routes/routing.rs
index 4c8d89fa87d..cc4a9510ab6 100644
--- a/crates/router/src/routes/routing.rs
+++ b/crates/router/src/routes/routing.rs
@@ -989,7 +989,7 @@ pub async fn toggle_success_based_routing(
 ) -> impl Responder {
     let flow = Flow::ToggleDynamicRouting;
     let wrapper = routing_types::ToggleSuccessBasedRoutingWrapper {
-        status: query.into_inner().status,
+        feature_to_enable: query.into_inner().enable,
         profile_id: path.into_inner().profile_id,
     };
     Box::pin(oss_api::server_wrap(
@@ -1005,7 +1005,7 @@ pub async fn toggle_success_based_routing(
                 state,
                 auth.merchant_account,
                 auth.key_store,
-                wrapper.status,
+                wrapper.feature_to_enable,
                 wrapper.profile_id,
             )
         },

## 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 -->
This will add dynamic_routing in core flows.
After which once the toggle route is hit with `feature = dynamic_connector_selection`, routable connectors will be evaluated on basis of success_based routing.

### Additional Changes

- [x] 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).
-->
Will enable success_based routing

## 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)?
-->
### The testing flows are as follows:
1. Toggle the route with feature(`metrics`, `dynamic_conenctor_selection`, `none`) as `dynamic_connector_selection`.
```
curl --location --request POST 'http://localhost:8080/account/merchant_id/business_profile/profile_id/dynamic_routing/success_based/toggle?enable=dynamic_connector_selection' \
--header 'api-key: api_key'
```
2. Check the business_profile table for the enabled dynamic routing feature.
3. Moreover if the dynamic_connector_selection is enabled for business_profile, success_based_routing will be performed, which can be checked by having a log filter with `success_based_routing connectors`.
4. If metrics feature is enabled, there will be no logs from the 3rd step instead there will be a log something like `successfully pushed success_based_routing metrics`, NOTE: this log will also be present in step 3rd.
5. if None is selected no logs regarding `success_based_routing` will be there.

## 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
