feat(routing): Enable volume split for dynamic routing

config/development.toml

@@ -790,9 +790,10 @@ card_networks = "Visa, AmericanExpress, Mastercard"
 [network_tokenization_supported_connectors]
 connector_list = "cybersource"
 
-[grpc_client.dynamic_routing_client] 
-host = "localhost" 
-port = 7000       
+[grpc_client.dynamic_routing_client]
+host = "localhost"
+port = 7000
+service = "dynamo"
 
 [theme_storage]
 file_storage_backend = "file_system"

crates/api_models/src/events/routing.rs

@@ -4,9 +4,9 @@ use crate::routing::{
     LinkedRoutingConfigRetrieveResponse, MerchantRoutingAlgorithm, ProfileDefaultRoutingConfig,
     RoutingAlgorithmId, RoutingConfigRequest, RoutingDictionaryRecord, RoutingKind,
     RoutingLinkWrapper, RoutingPayloadWrapper, RoutingRetrieveLinkQuery,
-    RoutingRetrieveLinkQueryWrapper, RoutingRetrieveQuery, SuccessBasedRoutingConfig,
-    SuccessBasedRoutingPayloadWrapper, SuccessBasedRoutingUpdateConfigQuery,
-    ToggleDynamicRoutingQuery, ToggleDynamicRoutingWrapper,
+    RoutingRetrieveLinkQueryWrapper, RoutingRetrieveQuery, RoutingVolumeSplitWrapper,
+    SuccessBasedRoutingConfig, SuccessBasedRoutingPayloadWrapper,
+    SuccessBasedRoutingUpdateConfigQuery, ToggleDynamicRoutingQuery, ToggleDynamicRoutingWrapper,
 };
 
 impl ApiEventMetric for RoutingKind {
@@ -108,3 +108,9 @@ impl ApiEventMetric for SuccessBasedRoutingUpdateConfigQuery {
         Some(ApiEventsType::Routing)
     }
 }
+
+impl ApiEventMetric for RoutingVolumeSplitWrapper {
+    fn get_api_event_type(&self) -> Option<ApiEventsType> {
+        Some(ApiEventsType::Routing)
+    }
+}

crates/api_models/src/routing.rs

@@ -522,6 +522,7 @@ pub struct DynamicAlgorithmWithTimestamp<T> {
 #[derive(Debug, Default, Clone, serde::Serialize, serde::Deserialize)]
 pub struct DynamicRoutingAlgorithmRef {
     pub success_based_algorithm: Option<SuccessBasedAlgorithm>,
+    pub dynamic_routing_volume_split: Option<u8>,
     pub elimination_routing_algorithm: Option<EliminationRoutingAlgorithm>,
 }
 
@@ -554,32 +555,6 @@ impl DynamicRoutingAlgoAccessor for EliminationRoutingAlgorithm {
     }
 }
 
-impl EliminationRoutingAlgorithm {
-    pub fn new(
-        algorithm_id_with_timestamp: DynamicAlgorithmWithTimestamp<
-            common_utils::id_type::RoutingId,
-        >,
-    ) -> Self {
-        Self {
-            algorithm_id_with_timestamp,
-            enabled_feature: DynamicRoutingFeatures::None,
-        }
-    }
-}
-
-impl SuccessBasedAlgorithm {
-    pub fn new(
-        algorithm_id_with_timestamp: DynamicAlgorithmWithTimestamp<
-            common_utils::id_type::RoutingId,
-        >,
-    ) -> Self {
-        Self {
-            algorithm_id_with_timestamp,
-            enabled_feature: DynamicRoutingFeatures::None,
-        }
-    }
-}
-
 impl DynamicRoutingAlgorithmRef {
     pub fn update(&mut self, new: Self) {
         if let Some(elimination_routing_algorithm) = new.elimination_routing_algorithm {
@@ -608,8 +583,62 @@ impl DynamicRoutingAlgorithmRef {
             }
         }
     }
+
+    pub fn update_volume_split(&mut self, volume: Option<u8>) {
+        self.dynamic_routing_volume_split = volume
+    }
+}
+
+impl EliminationRoutingAlgorithm {
+    pub fn new(
+        algorithm_id_with_timestamp: DynamicAlgorithmWithTimestamp<
+            common_utils::id_type::RoutingId,
+        >,
+    ) -> Self {
+        Self {
+            algorithm_id_with_timestamp,
+            enabled_feature: DynamicRoutingFeatures::None,
+        }
+    }
+}
+
+impl SuccessBasedAlgorithm {
+    pub fn new(
+        algorithm_id_with_timestamp: DynamicAlgorithmWithTimestamp<
+            common_utils::id_type::RoutingId,
+        >,
+    ) -> Self {
+        Self {
+            algorithm_id_with_timestamp,
+            enabled_feature: DynamicRoutingFeatures::None,
+        }
+    }
 }
 
+#[derive(Debug, Default, Clone, serde::Serialize, serde::Deserialize)]
+pub struct RoutingVolumeSplit {
+    pub routing_type: RoutingType,
+    pub split: u8,
+}
+
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+pub struct RoutingVolumeSplitWrapper {
+    pub routing_info: RoutingVolumeSplit,
+    pub profile_id: common_utils::id_type::ProfileId,
+}
+
+#[derive(Debug, Default, Clone, serde::Serialize, serde::Deserialize, Eq, PartialEq)]
+pub enum RoutingType {
+    #[default]
+    Static,
+    Dynamic,
+}
+
+impl RoutingType {
+    pub fn is_dynamic_routing(&self) -> bool {
+        self == &Self::Dynamic
+    }
+}
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 pub struct SuccessBasedAlgorithm {
     pub algorithm_id_with_timestamp:
@@ -673,6 +702,11 @@ pub struct ToggleDynamicRoutingQuery {
     pub enable: DynamicRoutingFeatures,
 }
 
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, ToSchema)]
+pub struct DynamicRoutingVolumeSplitQuery {
+    pub split: u8,
+}
+
 #[derive(Debug, Default, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq, ToSchema)]
 #[serde(rename_all = "snake_case")]
 pub enum DynamicRoutingFeatures {

crates/router/src/consts.rs

@@ -211,3 +211,6 @@ pub const VAULT_DELETE_FLOW_TYPE: &str = "delete_from_vault";
 /// Vault Fingerprint fetch flow type
 #[cfg(all(feature = "v2", feature = "payment_methods_v2"))]
 pub const VAULT_GET_FINGERPRINT_FLOW_TYPE: &str = "get_fingerprint_vault";
+
+/// Max volume split for Dynamic routing
+pub const DYNAMIC_ROUTING_MAX_VOLUME: u8 = 100;

crates/router/src/core/payments.rs

@@ -6026,47 +6026,75 @@ where
     // dynamic success based connector selection
     #[cfg(all(feature = "v1", feature = "dynamic_routing"))]
     let connectors = {
-        if business_profile.dynamic_routing_algorithm.is_some() {
-            let success_based_routing_config_params_interpolator =
-                routing_helpers::SuccessBasedRoutingConfigParamsInterpolator::new(
-                    payment_data.get_payment_attempt().payment_method,
-                    payment_data.get_payment_attempt().payment_method_type,
-                    payment_data.get_payment_attempt().authentication_type,
-                    payment_data.get_payment_attempt().currency,
-                    payment_data
-                        .get_billing_address()
-                        .and_then(|address| address.address)
-                        .and_then(|address| address.country),
-                    payment_data
-                        .get_payment_attempt()
-                        .payment_method_data
-                        .as_ref()
-                        .and_then(|data| data.as_object())
-                        .and_then(|card| card.get("card"))
-                        .and_then(|data| data.as_object())
-                        .and_then(|card| card.get("card_network"))
-                        .and_then(|network| network.as_str())
-                        .map(|network| network.to_string()),
-                    payment_data
-                        .get_payment_attempt()
-                        .payment_method_data
-                        .as_ref()
-                        .and_then(|data| data.as_object())
-                        .and_then(|card| card.get("card"))
-                        .and_then(|data| data.as_object())
-                        .and_then(|card| card.get("card_isin"))
-                        .and_then(|card_isin| card_isin.as_str())
-                        .map(|card_isin| card_isin.to_string()),
-                );
-            routing::perform_success_based_routing(
-                state,
-                connectors.clone(),
-                business_profile,
-                success_based_routing_config_params_interpolator,
-            )
-            .await
-            .map_err(|e| logger::error!(success_rate_routing_error=?e))
-            .unwrap_or(connectors)
+        if let Some(algo) = business_profile.dynamic_routing_algorithm.clone() {
+            let dynamic_routing_config: api_models::routing::DynamicRoutingAlgorithmRef = algo
+                .parse_value("DynamicRoutingAlgorithmRef")
+                .change_context(errors::ApiErrorResponse::InternalServerError)
+                .attach_printable("unable to deserialize DynamicRoutingAlgorithmRef from JSON")?;
+            let dynamic_split = api_models::routing::RoutingVolumeSplit {
+                routing_type: api_models::routing::RoutingType::Dynamic,
+                split: dynamic_routing_config
+                    .dynamic_routing_volume_split
+                    .unwrap_or_default(),
+            };
+            let static_split: api_models::routing::RoutingVolumeSplit =
+                api_models::routing::RoutingVolumeSplit {
+                    routing_type: api_models::routing::RoutingType::Static,
+                    split: crate::consts::DYNAMIC_ROUTING_MAX_VOLUME
+                        - dynamic_routing_config
+                            .dynamic_routing_volume_split
+                            .unwrap_or_default(),
+                };
+            let volume_split_vec = vec![dynamic_split, static_split];
+            let routing_choice =
+                routing::perform_dynamic_routing_volume_split(volume_split_vec, None)
+                    .change_context(errors::ApiErrorResponse::InternalServerError)
+                    .attach_printable("failed to perform volume split on routing type")?;
+
+            if routing_choice.routing_type.is_dynamic_routing() {
+                let success_based_routing_config_params_interpolator =
+                    routing_helpers::SuccessBasedRoutingConfigParamsInterpolator::new(
+                        payment_data.get_payment_attempt().payment_method,
+                        payment_data.get_payment_attempt().payment_method_type,
+                        payment_data.get_payment_attempt().authentication_type,
+                        payment_data.get_payment_attempt().currency,
+                        payment_data
+                            .get_billing_address()
+                            .and_then(|address| address.address)
+                            .and_then(|address| address.country),
+                        payment_data
+                            .get_payment_attempt()
+                            .payment_method_data
+                            .as_ref()
+                            .and_then(|data| data.as_object())
+                            .and_then(|card| card.get("card"))
+                            .and_then(|data| data.as_object())
+                            .and_then(|card| card.get("card_network"))
+                            .and_then(|network| network.as_str())
+                            .map(|network| network.to_string()),
+                        payment_data
+                            .get_payment_attempt()
+                            .payment_method_data
+                            .as_ref()
+                            .and_then(|data| data.as_object())
+                            .and_then(|card| card.get("card"))
+                            .and_then(|data| data.as_object())
+                            .and_then(|card| card.get("card_isin"))
+                            .and_then(|card_isin| card_isin.as_str())
+                            .map(|card_isin| card_isin.to_string()),
+                    );
+                routing::perform_success_based_routing(
+                    state,
+                    connectors.clone(),
+                    business_profile,
+                    success_based_routing_config_params_interpolator,
+                )
+                .await
+                .map_err(|e| logger::error!(success_rate_routing_error=?e))
+                .unwrap_or(connectors)
+            } else {
+                connectors
+            }
         } else {
             connectors
         }

crates/router/src/core/payments/routing.rs

@@ -489,6 +489,36 @@ pub async fn refresh_routing_cache_v1(
     Ok(arc_cached_algorithm)
 }
 
+pub fn perform_dynamic_routing_volume_split(
+    splits: Vec<api_models::routing::RoutingVolumeSplit>,
+    rng_seed: Option<&str>,
+) -> RoutingResult<api_models::routing::RoutingVolumeSplit> {
+    let weights: Vec<u8> = splits.iter().map(|sp| sp.split).collect();
+    let weighted_index = distributions::WeightedIndex::new(weights)
+        .change_context(errors::RoutingError::VolumeSplitFailed)
+        .attach_printable("Error creating weighted distribution for volume split")?;
+
+    let idx = if let Some(seed) = rng_seed {
+        let mut hasher = hash_map::DefaultHasher::new();
+        seed.hash(&mut hasher);
+        let hash = hasher.finish();
+
+        let mut rng = rand_chacha::ChaCha8Rng::seed_from_u64(hash);
+        weighted_index.sample(&mut rng)
+    } else {
+        let mut rng = rand::thread_rng();
+        weighted_index.sample(&mut rng)
+    };
+
+    let routing_choice = splits
+        .get(idx)
+        .ok_or(errors::RoutingError::VolumeSplitFailed)
+        .attach_printable("Volume split index lookup failed")?
+        .clone();
+
+    Ok(routing_choice)
+}
+
 pub fn perform_volume_split(
     mut splits: Vec<routing_types::ConnectorVolumeSplit>,
     rng_seed: Option<&str>,

crates/router/src/core/routing.rs

@@ -15,7 +15,9 @@ use error_stack::ResultExt;
 use external_services::grpc_client::dynamic_routing::SuccessBasedDynamicRouting;
 use hyperswitch_domain_models::{mandates, payment_address};
 #[cfg(all(feature = "v1", feature = "dynamic_routing"))]
-use router_env::{logger, metrics::add_attributes};
+use router_env::logger;
+#[cfg(feature = "v1")]
+use router_env::metrics::add_attributes;
 use rustc_hash::FxHashSet;
 #[cfg(all(feature = "v1", feature = "dynamic_routing"))]
 use storage_impl::redis::cache;
@@ -1271,6 +1273,69 @@ pub async fn toggle_specific_dynamic_routing(
     }
 }
 
+#[cfg(feature = "v1")]
+pub async fn configure_dynamic_routing_volume_split(
+    state: SessionState,
+    merchant_account: domain::MerchantAccount,
+    key_store: domain::MerchantKeyStore,
+    profile_id: common_utils::id_type::ProfileId,
+    routing_info: routing::RoutingVolumeSplit,
+) -> RouterResponse<()> {
+    metrics::ROUTING_CREATE_REQUEST_RECEIVED.add(
+        &metrics::CONTEXT,
+        1,
+        &add_attributes([("profile_id", profile_id.get_string_repr().to_owned())]),
+    );
+    let db = state.store.as_ref();
+    let key_manager_state = &(&state).into();
+
+    utils::when(
+        routing_info.split > crate::consts::DYNAMIC_ROUTING_MAX_VOLUME,
+        || {
+            Err(errors::ApiErrorResponse::InvalidRequestData {
+                message: "Dynamic routing volume split should be less than 100".to_string(),
+            })
+        },
+    )?;
+
+    let business_profile: domain::Profile = core_utils::validate_and_get_business_profile(
+        db,
+        key_manager_state,
+        &key_store,
+        Some(&profile_id),
+        merchant_account.get_id(),
+    )
+    .await?
+    .get_required_value("Profile")
+    .change_context(errors::ApiErrorResponse::ProfileNotFound {
+        id: profile_id.get_string_repr().to_owned(),
+    })?;
+
+    let mut 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 deserialize dynamic routing algorithm ref from business profile",
+        )?
+        .unwrap_or_default();
+
+    dynamic_routing_algo_ref.update_volume_split(Some(routing_info.split));
+
+    helpers::update_business_profile_active_dynamic_algorithm_ref(
+        db,
+        &((&state).into()),
+        &key_store,
+        business_profile.clone(),
+        dynamic_routing_algo_ref.clone(),
+    )
+    .await?;
+
+    Ok(service_api::ApplicationResponse::StatusOk)
+}
+
 #[cfg(all(feature = "v1", feature = "dynamic_routing"))]
 pub async fn success_based_routing_update_configs(
     state: SessionState,