Bug: Create a profile level config to switch between routing engine



Hyperswitch already has support to handle euclid rules, whereas the same support has been added in [decision engine](https://github.com/juspay/decision-engine) as part of the routing unification. To migrate hyperswitch merchants to start using the decision engine results, we need this switch to migrate them slowly.
diff --git a/crates/api_models/src/open_router.rs b/crates/api_models/src/open_router.rs
index af91ece3f12..bec1f14fef6 100644
--- a/crates/api_models/src/open_router.rs
+++ b/crates/api_models/src/open_router.rs
@@ -186,6 +186,19 @@ pub struct DecisionEngineConfigSetupRequest {
     pub config: DecisionEngineConfigVariant,
 }
 
+#[derive(Debug, Serialize, Deserialize, Clone)]
+pub struct GetDecisionEngineConfigRequest {
+    pub merchant_id: String,
+    pub config: DecisionEngineDynamicAlgorithmType,
+}
+
+#[derive(Debug, Serialize, Deserialize, Clone)]
+#[serde(rename_all = "camelCase")]
+pub enum DecisionEngineDynamicAlgorithmType {
+    SuccessRate,
+    Elimination,
+}
+
 #[derive(Debug, Serialize, Deserialize, Clone)]
 #[serde(tag = "type", content = "data")]
 #[serde(rename_all = "camelCase")]
diff --git a/crates/api_models/src/routing.rs b/crates/api_models/src/routing.rs
index 3e312769e07..a8f7315be40 100644
--- a/crates/api_models/src/routing.rs
+++ b/crates/api_models/src/routing.rs
@@ -1409,6 +1409,7 @@ pub struct RuleMigrationResponse {
     pub profile_id: common_utils::id_type::ProfileId,
     pub euclid_algorithm_id: common_utils::id_type::RoutingId,
     pub decision_engine_algorithm_id: String,
+    pub is_active_rule: bool,
 }
 
 #[derive(Debug, serde::Serialize)]
@@ -1423,11 +1424,23 @@ impl RuleMigrationResponse {
         profile_id: common_utils::id_type::ProfileId,
         euclid_algorithm_id: common_utils::id_type::RoutingId,
         decision_engine_algorithm_id: String,
+        is_active_rule: bool,
     ) -> Self {
         Self {
             profile_id,
             euclid_algorithm_id,
             decision_engine_algorithm_id,
+            is_active_rule,
         }
     }
 }
+
+#[derive(Clone, Debug, serde::Deserialize, serde::Serialize, strum::Display, strum::EnumString)]
+#[serde(rename_all = "snake_case")]
+#[strum(serialize_all = "snake_case")]
+pub enum RoutingResultSource {
+    /// External Decision Engine
+    DecisionEngine,
+    /// Inbuilt Hyperswitch Routing Engine
+    HyperswitchRouting,
+}
diff --git a/crates/hyperswitch_domain_models/src/business_profile.rs b/crates/hyperswitch_domain_models/src/business_profile.rs
index 93ed2af4945..1b1b87b6b50 100644
--- a/crates/hyperswitch_domain_models/src/business_profile.rs
+++ b/crates/hyperswitch_domain_models/src/business_profile.rs
@@ -5,7 +5,7 @@ use common_utils::{
     date_time,
     encryption::Encryption,
     errors::{CustomResult, ValidationError},
-    ext_traits::OptionExt,
+    ext_traits::{OptionExt, ValueExt},
     pii, type_name,
     types::keymanager,
 };
@@ -20,8 +20,10 @@ use diesel_models::business_profile::{
 use error_stack::ResultExt;
 use masking::{ExposeInterface, PeekInterface, Secret};
 
-use crate::type_encryption::{crypto_operation, AsyncLift, CryptoOperation};
-
+use crate::{
+    errors::api_error_response,
+    type_encryption::{crypto_operation, AsyncLift, CryptoOperation},
+};
 #[cfg(feature = "v1")]
 #[derive(Clone, Debug)]
 pub struct Profile {
@@ -1194,6 +1196,61 @@ impl Profile {
     pub fn is_vault_sdk_enabled(&self) -> bool {
         self.external_vault_connector_details.is_some()
     }
+
+    #[cfg(feature = "v1")]
+    pub fn get_payment_routing_algorithm(
+        &self,
+    ) -> CustomResult<
+        Option<api_models::routing::RoutingAlgorithmRef>,
+        api_error_response::ApiErrorResponse,
+    > {
+        self.routing_algorithm
+            .clone()
+            .map(|val| {
+                val.parse_value::<api_models::routing::RoutingAlgorithmRef>("RoutingAlgorithmRef")
+            })
+            .transpose()
+            .change_context(api_error_response::ApiErrorResponse::InternalServerError)
+            .attach_printable("unable to deserialize routing algorithm ref from merchant account")
+    }
+
+    #[cfg(feature = "v1")]
+    pub fn get_payout_routing_algorithm(
+        &self,
+    ) -> CustomResult<
+        Option<api_models::routing::RoutingAlgorithmRef>,
+        api_error_response::ApiErrorResponse,
+    > {
+        self.payout_routing_algorithm
+            .clone()
+            .map(|val| {
+                val.parse_value::<api_models::routing::RoutingAlgorithmRef>("RoutingAlgorithmRef")
+            })
+            .transpose()
+            .change_context(api_error_response::ApiErrorResponse::InternalServerError)
+            .attach_printable(
+                "unable to deserialize payout routing algorithm ref from merchant account",
+            )
+    }
+
+    #[cfg(feature = "v1")]
+    pub fn get_frm_routing_algorithm(
+        &self,
+    ) -> CustomResult<
+        Option<api_models::routing::RoutingAlgorithmRef>,
+        api_error_response::ApiErrorResponse,
+    > {
+        self.frm_routing_algorithm
+            .clone()
+            .map(|val| {
+                val.parse_value::<api_models::routing::RoutingAlgorithmRef>("RoutingAlgorithmRef")
+            })
+            .transpose()
+            .change_context(api_error_response::ApiErrorResponse::InternalServerError)
+            .attach_printable(
+                "unable to deserialize frm routing algorithm ref from merchant account",
+            )
+    }
 }
 
 #[cfg(feature = "v2")]
diff --git a/crates/router/src/core/payments/routing.rs b/crates/router/src/core/payments/routing.rs
index 53a657c2e6c..2ab004cccd1 100644
--- a/crates/router/src/core/payments/routing.rs
+++ b/crates/router/src/core/payments/routing.rs
@@ -460,76 +460,72 @@ pub async fn perform_static_routing_v1(
     )
     .await?;
 
-    Ok(match cached_algorithm.as_ref() {
-        CachedAlgorithm::Single(conn) => vec![(**conn).clone()],
+    let backend_input = match transaction_data {
+        routing::TransactionData::Payment(payment_data) => make_dsl_input(payment_data)?,
+        #[cfg(feature = "payouts")]
+        routing::TransactionData::Payout(payout_data) => make_dsl_input_for_payouts(payout_data)?,
+    };
 
-        CachedAlgorithm::Priority(plist) => plist.clone(),
+    let payment_id = match transaction_data {
+        routing::TransactionData::Payment(payment_data) => payment_data
+            .payment_attempt
+            .payment_id
+            .clone()
+            .get_string_repr()
+            .to_string(),
+        #[cfg(feature = "payouts")]
+        routing::TransactionData::Payout(payout_data) => {
+            payout_data.payout_attempt.payout_id.clone()
+        }
+    };
+
+    let routing_events_wrapper = utils::RoutingEventsWrapper::new(
+        state.tenant.tenant_id.clone(),
+        state.request_id,
+        payment_id,
+        business_profile.get_id().to_owned(),
+        business_profile.merchant_id.to_owned(),
+        "DecisionEngine: Euclid Static Routing".to_string(),
+        None,
+        true,
+        false,
+    );
 
+    let de_euclid_connectors = perform_decision_euclid_routing(
+        state,
+        backend_input.clone(),
+        business_profile.get_id().get_string_repr().to_string(),
+        routing_events_wrapper
+    )
+    .await
+    .map_err(|e|
+        // errors are ignored as this is just for diff checking as of now (optional flow).
+        logger::error!(decision_engine_euclid_evaluate_error=?e, "decision_engine_euclid: error in evaluation of rule")
+    ).unwrap_or_default();
+
+    let routable_connectors = match cached_algorithm.as_ref() {
+        CachedAlgorithm::Single(conn) => vec![(**conn).clone()],
+        CachedAlgorithm::Priority(plist) => plist.clone(),
         CachedAlgorithm::VolumeSplit(splits) => perform_volume_split(splits.to_vec())
             .change_context(errors::RoutingError::ConnectorSelectionFailed)?,
-
         CachedAlgorithm::Advanced(interpreter) => {
-            let backend_input = match transaction_data {
-                routing::TransactionData::Payment(payment_data) => make_dsl_input(payment_data)?,
-                #[cfg(feature = "payouts")]
-                routing::TransactionData::Payout(payout_data) => {
-                    make_dsl_input_for_payouts(payout_data)?
-                }
-            };
-
-            let payment_id = match transaction_data {
-                routing::TransactionData::Payment(payment_data) => payment_data
-                    .payment_attempt
-                    .payment_id
-                    .clone()
-                    .get_string_repr()
-                    .to_string(),
-                #[cfg(feature = "payouts")]
-                routing::TransactionData::Payout(payout_data) => {
-                    payout_data.payout_attempt.payout_id.clone()
-                }
-            };
+            execute_dsl_and_get_connector_v1(backend_input, interpreter)?
+        }
+    };
 
-            let routing_events_wrapper = utils::RoutingEventsWrapper::new(
-                state.tenant.tenant_id.clone(),
-                state.request_id,
-                payment_id,
-                business_profile.get_id().to_owned(),
-                business_profile.merchant_id.to_owned(),
-                "DecisionEngine: Euclid Static Routing".to_string(),
-                None,
-                true,
-                false,
-            );
+    utils::compare_and_log_result(
+        de_euclid_connectors.clone(),
+        routable_connectors.clone(),
+        "evaluate_routing".to_string(),
+    );
 
-            let de_euclid_connectors = perform_decision_euclid_routing(
-                state,
-                backend_input.clone(),
-                business_profile.get_id().get_string_repr().to_string(),
-                routing_events_wrapper
-            )
-            .await
-            .map_err(|e|
-                // errors are ignored as this is just for diff checking as of now (optional flow).
-                logger::error!(decision_engine_euclid_evaluate_error=?e, "decision_engine_euclid: error in evaluation of rule")
-            ).unwrap_or_default();
-            let routable_connectors = execute_dsl_and_get_connector_v1(backend_input, interpreter)?;
-            let connectors = routable_connectors
-                .iter()
-                .map(|c| c.connector.to_string())
-                .collect::<Vec<String>>();
-            let de_connectors = de_euclid_connectors
-                .iter()
-                .map(|c| c.gateway_name.to_string())
-                .collect::<Vec<String>>();
-            utils::compare_and_log_result(
-                de_connectors,
-                connectors,
-                "evaluate_routing".to_string(),
-            );
-            routable_connectors
-        }
-    })
+    Ok(utils::select_routing_result(
+        state,
+        business_profile,
+        routable_connectors,
+        de_euclid_connectors,
+    )
+    .await)
 }
 
 async fn ensure_algorithm_cached_v1(
diff --git a/crates/router/src/core/payments/routing/utils.rs b/crates/router/src/core/payments/routing/utils.rs
index d49aada7076..4296fea819f 100644
--- a/crates/router/src/core/payments/routing/utils.rs
+++ b/crates/router/src/core/payments/routing/utils.rs
@@ -1,19 +1,26 @@
-use std::{
-    collections::{HashMap, HashSet},
-    str::FromStr,
-};
+use std::collections::{HashMap, HashSet};
 
 use api_models::{
-    open_router as or_types, routing as api_routing,
-    routing::{ConnectorSelection, RoutableConnectorChoice},
+    open_router as or_types,
+    routing::{
+        self as api_routing, ConnectorSelection, ConnectorVolumeSplit, RoutableConnectorChoice,
+    },
 };
 use async_trait::async_trait;
-use common_utils::{ext_traits::BytesExt, id_type};
+use common_enums::TransactionType;
+use common_utils::{
+    ext_traits::{BytesExt, StringExt},
+    id_type,
+};
 use diesel_models::{enums, routing_algorithm};
 use error_stack::ResultExt;
-use euclid::{backend::BackendInput, frontend::ast};
+use euclid::{
+    backend::BackendInput,
+    frontend::ast::{self},
+};
 #[cfg(all(feature = "v1", feature = "dynamic_routing"))]
 use external_services::grpc_client::dynamic_routing as ir_client;
+use hyperswitch_domain_models::business_profile;
 use hyperswitch_interfaces::events::routing_api_logs as routing_events;
 use router_env::tracing_actix_web::RequestId;
 use serde::{Deserialize, Serialize};
@@ -40,17 +47,6 @@ pub trait DecisionEngineApiHandler {
     where
         Req: Serialize + Send + Sync + 'static + Clone,
         Res: Serialize + serde::de::DeserializeOwned + Send + 'static + std::fmt::Debug + Clone;
-
-    async fn send_decision_engine_request_without_response_parsing<Req>(
-        state: &SessionState,
-        http_method: services::Method,
-        path: &str,
-        request_body: Option<Req>,
-        timeout: Option<u64>,
-        events_wrapper: Option<RoutingEventsWrapper<Req>>,
-    ) -> RoutingResult<()>
-    where
-        Req: Serialize + Send + Sync + 'static + Clone;
 }
 
 // Struct to implement the DecisionEngineApiHandler trait
@@ -71,7 +67,7 @@ pub async fn build_and_send_decision_engine_http_request<Req, Res, ErrRes>(
 ) -> RoutingResult<RoutingEventsResponse<Res>>
 where
     Req: Serialize + Send + Sync + 'static + Clone,
-    Res: Serialize + serde::de::DeserializeOwned + std::fmt::Debug + Clone,
+    Res: Serialize + serde::de::DeserializeOwned + std::fmt::Debug + Clone + 'static,
     ErrRes: serde::de::DeserializeOwned + std::fmt::Debug + Clone + DecisionEngineErrorsInterface,
 {
     let decision_engine_base_url = &state.conf.open_router.url;
@@ -109,6 +105,15 @@ where
 
                 let resp = should_parse_response
                     .then(|| {
+                        if std::any::TypeId::of::<Res>() == std::any::TypeId::of::<String>()
+                            && resp.response.is_empty()
+                        {
+                            return serde_json::from_str::<Res>("\"\"").change_context(
+                                errors::RoutingError::OpenRouterError(
+                                    "Failed to parse empty response as String".into(),
+                                ),
+                            );
+                        }
                         let response_type: Res = resp
                             .response
                             .parse_struct(std::any::type_name::<Res>())
@@ -209,35 +214,6 @@ impl DecisionEngineApiHandler for EuclidApiClient {
         logger::debug!(parsed_response = ?parsed_response, response_type = %std::any::type_name::<Res>(), euclid_request_path = %path, "decision_engine_euclid: Successfully parsed response from Euclid API");
         Ok(event_response)
     }
-
-    async fn send_decision_engine_request_without_response_parsing<Req>(
-        state: &SessionState,
-        http_method: services::Method,
-        path: &str,
-        request_body: Option<Req>,
-        timeout: Option<u64>,
-        events_wrapper: Option<RoutingEventsWrapper<Req>>,
-    ) -> RoutingResult<()>
-    where
-        Req: Serialize + Send + Sync + 'static + Clone,
-    {
-        let event_response =
-            build_and_send_decision_engine_http_request::<Req, (), DeErrorResponse>(
-                state,
-                http_method,
-                path,
-                request_body,
-                timeout,
-                "not parsing response",
-                events_wrapper,
-            )
-            .await?;
-
-        let response = event_response.response;
-
-        logger::debug!(euclid_response = ?response, euclid_request_path = %path, "decision_engine_routing: Received raw response from Euclid API");
-        Ok(())
-    }
 }
 
 #[async_trait]
@@ -275,35 +251,6 @@ impl DecisionEngineApiHandler for ConfigApiClient {
         logger::debug!(parsed_response = ?parsed_response, response_type = %std::any::type_name::<Res>(), decision_engine_request_path = %path, "decision_engine_config: Successfully parsed response from Decision Engine config API");
         Ok(events_response)
     }
-
-    async fn send_decision_engine_request_without_response_parsing<Req>(
-        state: &SessionState,
-        http_method: services::Method,
-        path: &str,
-        request_body: Option<Req>,
-        timeout: Option<u64>,
-        events_wrapper: Option<RoutingEventsWrapper<Req>>,
-    ) -> RoutingResult<()>
-    where
-        Req: Serialize + Send + Sync + 'static + Clone,
-    {
-        let event_response =
-            build_and_send_decision_engine_http_request::<Req, (), DeErrorResponse>(
-                state,
-                http_method,
-                path,
-                request_body,
-                timeout,
-                "not parsing response",
-                events_wrapper,
-            )
-            .await?;
-
-        let response = event_response.response;
-
-        logger::debug!(decision_engine_response = ?response, decision_engine_request_path = %path, "decision_engine_config: Received raw response from Decision Engine config API");
-        Ok(())
-    }
 }
 
 #[async_trait]
@@ -342,35 +289,6 @@ impl DecisionEngineApiHandler for SRApiClient {
         logger::debug!(parsed_response = ?parsed_response, response_type = %std::any::type_name::<Res>(), decision_engine_request_path = %path, "decision_engine_config: Successfully parsed response from Decision Engine config API");
         Ok(events_response)
     }
-
-    async fn send_decision_engine_request_without_response_parsing<Req>(
-        state: &SessionState,
-        http_method: services::Method,
-        path: &str,
-        request_body: Option<Req>,
-        timeout: Option<u64>,
-        events_wrapper: Option<RoutingEventsWrapper<Req>>,
-    ) -> RoutingResult<()>
-    where
-        Req: Serialize + Send + Sync + 'static + Clone,
-    {
-        let event_response =
-            build_and_send_decision_engine_http_request::<Req, (), or_types::ErrorResponse>(
-                state,
-                http_method,
-                path,
-                request_body,
-                timeout,
-                "not parsing response",
-                events_wrapper,
-            )
-            .await?;
-
-        let response = event_response.response;
-
-        logger::debug!(decision_engine_response = ?response, decision_engine_request_path = %path, "decision_engine_config: Received raw response from Decision Engine config API");
-        Ok(())
-    }
 }
 
 const EUCLID_API_TIMEOUT: u64 = 5;
@@ -380,7 +298,7 @@ pub async fn perform_decision_euclid_routing(
     input: BackendInput,
     created_by: String,
     events_wrapper: RoutingEventsWrapper<RoutingEvaluateRequest>,
-) -> RoutingResult<Vec<ConnectorInfo>> {
+) -> RoutingResult<Vec<RoutableConnectorChoice>> {
     logger::debug!("decision_engine_euclid: evaluate api call for euclid routing evaluation");
 
     let mut events_wrapper = events_wrapper;
@@ -413,47 +331,8 @@ pub async fn perform_decision_euclid_routing(
                 status_code: 500,
             })?;
 
-    let connector_info = euclid_response.evaluated_output.clone();
-    let mut routable_connectors = Vec::new();
-    for conn in &connector_info {
-        let connector = common_enums::RoutableConnectors::from_str(conn.gateway_name.as_str())
-            .change_context(errors::RoutingError::GenericConversionError {
-                from: "String".to_string(),
-                to: "RoutableConnectors".to_string(),
-            })
-            .attach_printable(
-                "decision_engine_euclid: unable to convert String to RoutableConnectors",
-            )
-            .ok();
-        let mca_id = conn
-            .gateway_id
-            .as_ref()
-            .map(|id| {
-                id_type::MerchantConnectorAccountId::wrap(id.to_string())
-        .change_context(errors::RoutingError::GenericConversionError {
-            from: "String".to_string(),
-            to: "MerchantConnectorAccountId".to_string(),
-        })
-        .attach_printable(
-            "decision_engine_euclid: unable to convert MerchantConnectorAccountId from string",
-        )
-            })
-            .transpose()
-            .ok()
-            .flatten();
-
-        if let Some(conn) = connector {
-            let connector = RoutableConnectorChoice {
-                choice_kind: api_routing::RoutableChoiceKind::FullStruct,
-                connector: conn,
-                merchant_connector_id: mca_id,
-            };
-            routable_connectors.push(connector);
-        }
-    }
-
     routing_event.set_routing_approach(RoutingApproach::StaticRouting.to_string());
-    routing_event.set_routable_connectors(routable_connectors);
+    routing_event.set_routable_connectors(euclid_response.evaluated_output.clone());
     state.event_handler.log_event(&routing_event);
 
     // Need to log euclid response event here
@@ -498,7 +377,7 @@ pub async fn link_de_euclid_routing_algorithm(
 ) -> RoutingResult<()> {
     logger::debug!("decision_engine_euclid: link api call for euclid routing algorithm");
 
-    EuclidApiClient::send_decision_engine_request_without_response_parsing(
+    EuclidApiClient::send_decision_engine_request::<_, String>(
         state,
         services::Method::Post,
         "routing/activate",
@@ -542,6 +421,31 @@ pub async fn list_de_euclid_routing_algorithms(
         .collect::<Vec<_>>())
 }
 
+pub async fn list_de_euclid_active_routing_algorithm(
+    state: &SessionState,
+    created_by: String,
+) -> RoutingResult<Vec<api_routing::RoutingDictionaryRecord>> {
+    logger::debug!("decision_engine_euclid: list api call for euclid active routing algorithm");
+    let response: Vec<RoutingAlgorithmRecord> = EuclidApiClient::send_decision_engine_request(
+        state,
+        services::Method::Post,
+        format!("routing/list/active/{created_by}").as_str(),
+        None::<()>,
+        Some(EUCLID_API_TIMEOUT),
+        None,
+    )
+    .await?
+    .response
+    .ok_or(errors::RoutingError::OpenRouterError(
+        "Response from decision engine API is empty".to_string(),
+    ))?;
+
+    Ok(response
+        .into_iter()
+        .map(|record| routing_algorithm::RoutingProfileMetadata::from(record).foreign_into())
+        .collect())
+}
+
 pub fn compare_and_log_result<T: RoutingEq<T> + Serialize>(
     de_result: Vec<T>,
     result: Vec<T>,
@@ -566,7 +470,8 @@ pub trait RoutingEq<T> {
 
 impl RoutingEq<Self> for api_routing::RoutingDictionaryRecord {
     fn is_equal(a: &Self, b: &Self) -> bool {
-        a.name == b.name
+        a.id == b.id
+            && a.name == b.name
             && a.profile_id == b.profile_id
             && a.description == b.description
             && a.kind == b.kind
@@ -580,6 +485,14 @@ impl RoutingEq<Self> for String {
     }
 }
 
+impl RoutingEq<Self> for RoutableConnectorChoice {
+    fn is_equal(a: &Self, b: &Self) -> bool {
+        a.connector.eq(&b.connector)
+            && a.choice_kind.eq(&b.choice_kind)
+            && a.merchant_connector_id.eq(&b.merchant_connector_id)
+    }
+}
+
 pub fn to_json_string<T: Serialize>(value: &T) -> String {
     serde_json::to_string(value)
         .map_err(|_| errors::RoutingError::GenericConversionError {
@@ -772,8 +685,30 @@ pub struct RoutingEvaluateRequest {
 pub struct RoutingEvaluateResponse {
     pub status: String,
     pub output: serde_json::Value,
-    pub evaluated_output: Vec<ConnectorInfo>,
-    pub eligible_connectors: Vec<ConnectorInfo>,
+    #[serde(deserialize_with = "deserialize_connector_choices")]
+    pub evaluated_output: Vec<RoutableConnectorChoice>,
+    #[serde(deserialize_with = "deserialize_connector_choices")]
+    pub eligible_connectors: Vec<RoutableConnectorChoice>,
+}
+
+/// Routable Connector chosen for a payment
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+pub struct DeRoutableConnectorChoice {
+    pub gateway_name: common_enums::RoutableConnectors,
+    pub gateway_id: Option<id_type::MerchantConnectorAccountId>,
+}
+
+fn deserialize_connector_choices<'de, D>(
+    deserializer: D,
+) -> Result<Vec<RoutableConnectorChoice>, D::Error>
+where
+    D: serde::Deserializer<'de>,
+{
+    let infos = Vec::<DeRoutableConnectorChoice>::deserialize(deserializer)?;
+    Ok(infos
+        .into_iter()
+        .map(RoutableConnectorChoice::from)
+        .collect())
 }
 
 #[derive(Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
@@ -794,12 +729,24 @@ pub enum ValueType {
     MetadataVariant(MetadataValue),
     /// Represents a arbitrary String value
     StrValue(String),
+    /// Represents a global reference, which is a reference to a global variable
     GlobalRef(String),
+    /// Represents an array of numbers. This is basically used for
+    /// "one of the given numbers" operations
+    /// eg: payment.method.amount = (1, 2, 3)
+    NumberArray(Vec<u64>),
+    /// Similar to NumberArray but for enum variants
+    /// eg: payment.method.cardtype = (debit, credit)
+    EnumVariantArray(Vec<String>),
+    /// Like a number array but can include comparisons. Useful for
+    /// conditions like "500 < amount < 1000"
+    /// eg: payment.amount = (> 500, < 1000)
+    NumberComparisonArray(Vec<NumberComparison>),
 }
 
 pub type Metadata = HashMap<String, serde_json::Value>;
 /// Represents a number comparison for "NumberComparisonArrayValue"
-#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
+#[derive(Clone, Debug, Eq, Hash, PartialEq, Serialize, Deserialize)]
 #[serde(rename_all = "snake_case")]
 pub struct NumberComparison {
     pub comparison_type: ComparisonType,
@@ -917,9 +864,20 @@ impl ConnectorInfo {
     }
 }
 
+impl From<DeRoutableConnectorChoice> for RoutableConnectorChoice {
+    fn from(choice: DeRoutableConnectorChoice) -> Self {
+        Self {
+            choice_kind: api_routing::RoutableChoiceKind::FullStruct,
+            connector: choice.gateway_name,
+            merchant_connector_id: choice.gateway_id,
+        }
+    }
+}
+
 #[derive(Clone, Debug, Serialize, Deserialize)]
 #[serde(rename_all = "snake_case")]
 pub enum Output {
+    Single(ConnectorInfo),
     Priority(Vec<ConnectorInfo>),
     VolumeSplit(Vec<VolumeSplit<ConnectorInfo>>),
     VolumeSplitPriority(Vec<VolumeSplit<Vec<ConnectorInfo>>>),
@@ -949,12 +907,73 @@ pub struct RoutingRule {
     pub description: Option<String>,
     pub metadata: Option<RoutingMetadata>,
     pub created_by: String,
+    #[serde(default)]
+    pub algorithm_for: AlgorithmType,
     pub algorithm: StaticRoutingAlgorithm,
 }
+#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize, strum::Display)]
+#[serde(rename_all = "snake_case")]
+#[strum(serialize_all = "snake_case")]
+pub enum AlgorithmType {
+    #[default]
+    Payment,
+    Payout,
+    ThreeDsAuthentication,
+}
+
+impl From<TransactionType> for AlgorithmType {
+    fn from(transaction_type: TransactionType) -> Self {
+        match transaction_type {
+            TransactionType::Payment => Self::Payment,
+            TransactionType::Payout => Self::Payout,
+            TransactionType::ThreeDsAuthentication => Self::ThreeDsAuthentication,
+        }
+    }
+}
+
+impl From<RoutableConnectorChoice> for ConnectorInfo {
+    fn from(c: RoutableConnectorChoice) -> Self {
+        Self {
+            gateway_name: c.connector.to_string(),
+            gateway_id: c
+                .merchant_connector_id
+                .map(|mca_id| mca_id.get_string_repr().to_string()),
+        }
+    }
+}
+
+impl From<Box<RoutableConnectorChoice>> for ConnectorInfo {
+    fn from(c: Box<RoutableConnectorChoice>) -> Self {
+        Self {
+            gateway_name: c.connector.to_string(),
+            gateway_id: c
+                .merchant_connector_id
+                .map(|mca_id| mca_id.get_string_repr().to_string()),
+        }
+    }
+}
+
+impl From<ConnectorVolumeSplit> for VolumeSplit<ConnectorInfo> {
+    fn from(v: ConnectorVolumeSplit) -> Self {
+        Self {
+            split: v.split,
+            output: ConnectorInfo {
+                gateway_name: v.connector.connector.to_string(),
+                gateway_id: v
+                    .connector
+                    .merchant_connector_id
+                    .map(|mca_id| mca_id.get_string_repr().to_string()),
+            },
+        }
+    }
+}
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[serde(tag = "type", content = "data", rename_all = "snake_case")]
 pub enum StaticRoutingAlgorithm {
+    Single(Box<ConnectorInfo>),
+    Priority(Vec<ConnectorInfo>),
+    VolumeSplit(Vec<VolumeSplit<ConnectorInfo>>),
     Advanced(Program),
 }
 
@@ -962,7 +981,6 @@ pub enum StaticRoutingAlgorithm {
 #[serde(rename_all = "snake_case")]
 pub struct RoutingMetadata {
     pub kind: enums::RoutingAlgorithmKind,
-    pub algorithm_for: enums::TransactionType,
 }
 
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
@@ -981,7 +999,8 @@ pub struct RoutingAlgorithmRecord {
     pub name: String,
     pub description: Option<String>,
     pub created_by: id_type::ProfileId,
-    pub algorithm_data: Program,
+    pub algorithm_data: StaticRoutingAlgorithm,
+    pub algorithm_for: TransactionType,
     pub metadata: Option<RoutingMetadata>,
     pub created_at: time::PrimitiveDateTime,
     pub modified_at: time::PrimitiveDateTime,
@@ -989,12 +1008,11 @@ pub struct RoutingAlgorithmRecord {
 
 impl From<RoutingAlgorithmRecord> for routing_algorithm::RoutingProfileMetadata {
     fn from(record: RoutingAlgorithmRecord) -> Self {
-        let (kind, algorithm_for) = match record.metadata {
-            Some(metadata) => (metadata.kind, metadata.algorithm_for),
-            None => (
-                enums::RoutingAlgorithmKind::Advanced,
-                enums::TransactionType::default(),
-            ),
+        let kind = match record.algorithm_data {
+            StaticRoutingAlgorithm::Single(_) => enums::RoutingAlgorithmKind::Single,
+            StaticRoutingAlgorithm::Priority(_) => enums::RoutingAlgorithmKind::Priority,
+            StaticRoutingAlgorithm::VolumeSplit(_) => enums::RoutingAlgorithmKind::VolumeSplit,
+            StaticRoutingAlgorithm::Advanced(_) => enums::RoutingAlgorithmKind::Advanced,
         };
         Self {
             profile_id: record.created_by,
@@ -1004,7 +1022,7 @@ impl From<RoutingAlgorithmRecord> for routing_algorithm::RoutingProfileMetadata
             kind,
             created_at: record.created_at,
             modified_at: record.modified_at,
-            algorithm_for,
+            algorithm_for: record.algorithm_for,
         }
     }
 }
@@ -1106,8 +1124,20 @@ fn convert_value(v: ast::ValueType) -> RoutingResult<ValueType> {
             value: m.value,
         })),
         StrValue(s) => Ok(ValueType::StrValue(s)),
-        _ => Err(error_stack::Report::new(
-            errors::RoutingError::InvalidRoutingAlgorithmStructure,
+
+        NumberArray(arr) => Ok(ValueType::NumberArray(
+            arr.into_iter()
+                .map(|n| n.get_amount_as_i64().try_into().unwrap_or_default())
+                .collect(),
+        )),
+        EnumVariantArray(arr) => Ok(ValueType::EnumVariantArray(arr)),
+        NumberComparisonArray(arr) => Ok(ValueType::NumberComparisonArray(
+            arr.into_iter()
+                .map(|nc| NumberComparison {
+                    comparison_type: convert_comparison_type(nc.comparison_type),
+                    number: nc.number.get_amount_as_i64().try_into().unwrap_or_default(),
+                })
+                .collect(),
         )),
     }
 }
@@ -1136,6 +1166,29 @@ fn stringify_choice(c: RoutableConnectorChoice) -> ConnectorInfo {
     )
 }
 
+pub async fn select_routing_result<T>(
+    state: &SessionState,
+    business_profile: &business_profile::Profile,
+    hyperswitch_result: T,
+    de_result: T,
+) -> T {
+    let routing_result_source: Option<api_routing::RoutingResultSource> = state
+        .store
+        .find_config_by_key(&format!(
+            "routing_result_source_{0}",
+            business_profile.get_id().get_string_repr()
+        ))
+        .await
+        .map(|c| c.config.parse_enum("RoutingResultSource").ok())
+        .unwrap_or(None); //Ignore errors so that we can use the hyperswitch result as a fallback
+    if let Some(api_routing::RoutingResultSource::DecisionEngine) = routing_result_source {
+        logger::debug!(business_profile_id=?business_profile.get_id(), "Using Decision Engine routing result");
+        de_result
+    } else {
+        logger::debug!(business_profile_id=?business_profile.get_id(), "Using Hyperswitch routing result");
+        hyperswitch_result
+    }
+}
 pub trait DecisionEngineErrorsInterface {
     fn get_error_message(&self) -> String;
     fn get_error_code(&self) -> String;
diff --git a/crates/router/src/core/routing.rs b/crates/router/src/core/routing.rs
index 1ab333ef139..b52fc966721 100644
--- a/crates/router/src/core/routing.rs
+++ b/crates/router/src/core/routing.rs
@@ -25,8 +25,6 @@ use external_services::grpc_client::dynamic_routing::{
 use helpers::update_decision_engine_dynamic_routing_setup;
 use hyperswitch_domain_models::{mandates, payment_address};
 use payment_methods::helpers::StorageErrorExt;
-#[cfg(all(feature = "v1", feature = "dynamic_routing"))]
-use router_env::logger;
 use rustc_hash::FxHashSet;
 #[cfg(all(feature = "v1", feature = "dynamic_routing"))]
 use storage_impl::redis::cache;
@@ -167,7 +165,7 @@ pub async fn retrieve_merchant_routing_dictionary(
         routing_metadata,
     );
 
-    let result = routing_metadata
+    let mut result = routing_metadata
         .into_iter()
         .map(ForeignInto::foreign_into)
         .collect::<Vec<_>>();
@@ -175,7 +173,7 @@ pub async fn retrieve_merchant_routing_dictionary(
     if let Some(profile_ids) = profile_id_list {
         let mut de_result: Vec<routing_types::RoutingDictionaryRecord> = vec![];
         // DE_TODO: need to replace this with batch API call to reduce the number of network calls
-        for profile_id in profile_ids {
+        for profile_id in &profile_ids {
             let list_request = ListRountingAlgorithmsRequest {
                 created_by: profile_id.get_string_repr().to_string(),
             };
@@ -186,8 +184,32 @@ pub async fn retrieve_merchant_routing_dictionary(
                 })
                 .ok() // Avoid throwing error if Decision Engine is not available or other errors
                 .map(|mut de_routing| de_result.append(&mut de_routing));
+            // filter de_result based on transaction type
+            de_result.retain(|record| record.algorithm_for == Some(transaction_type));
+            // append dynamic routing algorithms to de_result
+            de_result.append(
+                &mut result
+                    .clone()
+                    .into_iter()
+                    .filter(|record: &routing_types::RoutingDictionaryRecord| {
+                        record.kind == routing_types::RoutingAlgorithmKind::Dynamic
+                    })
+                    .collect::<Vec<_>>(),
+            );
         }
-        compare_and_log_result(de_result, result.clone(), "list_routing".to_string());
+        compare_and_log_result(
+            de_result.clone(),
+            result.clone(),
+            "list_routing".to_string(),
+        );
+        result = build_list_routing_result(
+            &state,
+            merchant_context,
+            &result,
+            &de_result,
+            profile_ids.clone(),
+        )
+        .await?;
     }
 
     metrics::ROUTING_MERCHANT_DICTIONARY_RETRIEVE_SUCCESS_RESPONSE.add(1, &[]);
@@ -196,6 +218,47 @@ pub async fn retrieve_merchant_routing_dictionary(
     ))
 }
 
+async fn build_list_routing_result(
+    state: &SessionState,
+    merchant_context: domain::MerchantContext,
+    hs_results: &[routing_types::RoutingDictionaryRecord],
+    de_results: &[routing_types::RoutingDictionaryRecord],
+    profile_ids: Vec<common_utils::id_type::ProfileId>,
+) -> RouterResult<Vec<routing_types::RoutingDictionaryRecord>> {
+    let db = state.store.as_ref();
+    let key_manager_state = &state.into();
+    let mut list_result: Vec<routing_types::RoutingDictionaryRecord> = vec![];
+    for profile_id in profile_ids.iter() {
+        let by_profile =
+            |rec: &&routing_types::RoutingDictionaryRecord| &rec.profile_id == profile_id;
+        let de_result_for_profile = de_results.iter().filter(by_profile).cloned().collect();
+        let hs_result_for_profile = hs_results.iter().filter(by_profile).cloned().collect();
+        let business_profile = core_utils::validate_and_get_business_profile(
+            db,
+            key_manager_state,
+            merchant_context.get_merchant_key_store(),
+            Some(profile_id),
+            merchant_context.get_merchant_account().get_id(),
+        )
+        .await?
+        .get_required_value("Profile")
+        .change_context(errors::ApiErrorResponse::ProfileNotFound {
+            id: profile_id.get_string_repr().to_owned(),
+        })?;
+
+        list_result.append(
+            &mut select_routing_result(
+                state,
+                &business_profile,
+                hs_result_for_profile,
+                de_result_for_profile,
+            )
+            .await,
+        );
+    }
+    Ok(list_result)
+}
+
 #[cfg(feature = "v2")]
 pub async fn create_routing_algorithm_under_profile(
     state: SessionState,
@@ -280,8 +343,6 @@ pub async fn create_routing_algorithm_under_profile(
 ) -> RouterResponse<routing_types::RoutingDictionaryRecord> {
     use api_models::routing::StaticRoutingAlgorithm as EuclidAlgorithm;
 
-    use crate::services::logger;
-
     metrics::ROUTING_CREATE_REQUEST_RECEIVED.add(1, &[]);
     let db = state.store.as_ref();
     let key_manager_state = &(&state).into();
@@ -344,61 +405,84 @@ pub async fn create_routing_algorithm_under_profile(
 
     let mut decision_engine_routing_id: Option<String> = None;
 
-    if let Some(EuclidAlgorithm::Advanced(program)) = request.algorithm.clone() {
-        match program.try_into() {
-            Ok(internal_program) => {
-                let routing_rule = RoutingRule {
-                    rule_id: None,
-                    name: name.clone(),
-                    description: Some(description.clone()),
-                    created_by: profile_id.get_string_repr().to_string(),
-                    algorithm: internal_program,
-                    metadata: Some(RoutingMetadata {
-                        kind: algorithm.get_kind().foreign_into(),
-                        algorithm_for: transaction_type.to_owned(),
-                    }),
-                };
-
-                match create_de_euclid_routing_algo(&state, &routing_rule).await {
-                    Ok(id) => {
-                        decision_engine_routing_id = Some(id);
-                    }
-                    Err(e)
-                        if matches!(
-                            e.current_context(),
-                            errors::RoutingError::DecisionEngineValidationError(_)
-                        ) =>
+    if let Some(euclid_algorithm) = request.algorithm.clone() {
+        let maybe_static_algorithm: Option<StaticRoutingAlgorithm> = match euclid_algorithm {
+            EuclidAlgorithm::Advanced(program) => match program.try_into() {
+                Ok(internal_program) => Some(StaticRoutingAlgorithm::Advanced(internal_program)),
+                Err(e) => {
+                    router_env::logger::error!(decision_engine_error = ?e, "decision_engine_euclid");
+                    None
+                }
+            },
+            EuclidAlgorithm::Single(conn) => {
+                Some(StaticRoutingAlgorithm::Single(Box::new(conn.into())))
+            }
+            EuclidAlgorithm::Priority(connectors) => {
+                let converted: Vec<ConnectorInfo> =
+                    connectors.into_iter().map(Into::into).collect();
+                Some(StaticRoutingAlgorithm::Priority(converted))
+            }
+            EuclidAlgorithm::VolumeSplit(splits) => {
+                let converted: Vec<VolumeSplit<ConnectorInfo>> =
+                    splits.into_iter().map(Into::into).collect();
+                Some(StaticRoutingAlgorithm::VolumeSplit(converted))
+            }
+            EuclidAlgorithm::ThreeDsDecisionRule(_) => {
+                router_env::logger::error!(
+                    "decision_engine_euclid: ThreeDsDecisionRules are not yet implemented"
+                );
+                None
+            }
+        };
+
+        if let Some(static_algorithm) = maybe_static_algorithm {
+            let routing_rule = RoutingRule {
+                rule_id: Some(algorithm_id.clone().get_string_repr().to_owned()),
+                name: name.clone(),
+                description: Some(description.clone()),
+                created_by: profile_id.get_string_repr().to_string(),
+                algorithm: static_algorithm,
+                algorithm_for: transaction_type.into(),
+                metadata: Some(RoutingMetadata {
+                    kind: algorithm.get_kind().foreign_into(),
+                }),
+            };
+
+            match create_de_euclid_routing_algo(&state, &routing_rule).await {
+                Ok(id) => {
+                    decision_engine_routing_id = Some(id);
+                }
+                Err(e)
+                    if matches!(
+                        e.current_context(),
+                        errors::RoutingError::DecisionEngineValidationError(_)
+                    ) =>
+                {
+                    if let errors::RoutingError::DecisionEngineValidationError(msg) =
+                        e.current_context()
                     {
-                        if let errors::RoutingError::DecisionEngineValidationError(msg) =
-                            e.current_context()
-                        {
-                            logger::error!(
-                                decision_engine_euclid_error = ?msg,
-                                decision_engine_euclid_request = ?routing_rule,
-                                "failed to create rule in decision_engine with validation error"
-                            );
-                        }
-                    }
-                    Err(e) => {
-                        logger::error!(
-                            decision_engine_euclid_error = ?e,
+                        router_env::logger::error!(
+                            decision_engine_euclid_error = ?msg,
                             decision_engine_euclid_request = ?routing_rule,
-                            "failed to create rule in decision_engine"
+                            "failed to create rule in decision_engine with validation error"
                         );
                     }
                 }
+                Err(e) => {
+                    router_env::logger::error!(
+                        decision_engine_euclid_error = ?e,
+                        decision_engine_euclid_request = ?routing_rule,
+                        "failed to create rule in decision_engine"
+                    );
+                }
             }
-            Err(e) => {
-                // errors are ignored as this is just for diff checking as of now (optional flow).
-                logger::error!(decision_engine_error=?e, "decision_engine_euclid");
-            }
-        };
+        }
     }
 
     if decision_engine_routing_id.is_some() {
-        logger::info!(routing_flow=?"create_euclid_routing_algorithm", is_equal=?"true", "decision_engine_euclid");
+        router_env::logger::info!(routing_flow=?"create_euclid_routing_algorithm", is_equal=?"true", "decision_engine_euclid");
     } else {
-        logger::info!(routing_flow=?"create_euclid_routing_algorithm", is_equal=?"false", "decision_engine_euclid");
+        router_env::logger::info!(routing_flow=?"create_euclid_routing_algorithm", is_equal=?"false", "decision_engine_euclid");
     }
 
     let timestamp = common_utils::date_time::now();
@@ -1276,7 +1360,23 @@ pub async fn retrieve_linked_routing_config(
                 )
                 .await
                 .to_not_found_response(errors::ApiErrorResponse::ResourceIdNotFound)?;
-            active_algorithms.push(record.foreign_into());
+            let hs_records: Vec<routing_types::RoutingDictionaryRecord> =
+                vec![record.foreign_into()];
+            let de_records = retrieve_decision_engine_active_rules(
+                &state,
+                &transaction_type,
+                profile_id.clone(),
+                hs_records.clone(),
+            )
+            .await;
+            compare_and_log_result(
+                de_records.clone(),
+                hs_records.clone(),
+                "list_active_routing".to_string(),
+            );
+            active_algorithms.append(
+                &mut select_routing_result(&state, &business_profile, hs_records, de_records).await,
+            );
         }
 
         // Handle dynamic routing algorithms
@@ -1319,7 +1419,9 @@ pub async fn retrieve_linked_routing_config(
                 )
                 .await
                 .to_not_found_response(errors::ApiErrorResponse::ResourceIdNotFound)?;
-            active_algorithms.push(record.foreign_into());
+            if record.algorithm_for == transaction_type {
+                active_algorithms.push(record.foreign_into());
+            }
         }
     }
 
@@ -1328,6 +1430,32 @@ pub async fn retrieve_linked_routing_config(
         routing_types::LinkedRoutingConfigRetrieveResponse::ProfileBased(active_algorithms),
     ))
 }
+
+pub async fn retrieve_decision_engine_active_rules(
+    state: &SessionState,
+    transaction_type: &enums::TransactionType,
+    profile_id: common_utils::id_type::ProfileId,
+    hs_records: Vec<routing_types::RoutingDictionaryRecord>,
+) -> Vec<routing_types::RoutingDictionaryRecord> {
+    let mut de_records =
+        list_de_euclid_active_routing_algorithm(state, profile_id.get_string_repr().to_owned())
+            .await
+            .map_err(|e| {
+                router_env::logger::error!(?e, "Failed to list DE Euclid active routing algorithm");
+            })
+            .ok() // Avoid throwing error if Decision Engine is not available or other errors thrown
+            .unwrap_or_default();
+    // Use Hs records to list the dynamic algorithms as DE is not supporting dynamic algorithms in HS standard
+    let mut dynamic_algos = hs_records
+        .into_iter()
+        .filter(|record| record.kind == routing_types::RoutingAlgorithmKind::Dynamic)
+        .collect::<Vec<_>>();
+    de_records.append(&mut dynamic_algos);
+    de_records
+        .into_iter()
+        .filter(|r| r.algorithm_for == Some(*transaction_type))
+        .collect::<Vec<_>>()
+}
 // List all the default fallback algorithms under all the profile under a merchant
 pub async fn retrieve_default_routing_config_for_profiles(
     state: SessionState,
@@ -1707,7 +1835,7 @@ pub async fn success_based_routing_update_configs(
         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:?}"));
+    .map_err(|e| router_env::logger::error!("unable to publish into the redact channel for evicting the success based routing config cache {e:?}"));
 
     let new_record = record.foreign_into();
 
@@ -1811,7 +1939,7 @@ pub async fn elimination_routing_update_configs(
         cache_entries_to_redact,
     )
     .await
-    .map_err(|e| logger::error!("unable to publish into the redact channel for evicting the elimination routing config cache {e:?}")).ok();
+    .map_err(|e| router_env::logger::error!("unable to publish into the redact channel for evicting the elimination routing config cache {e:?}")).ok();
 
     let new_record = record.foreign_into();
 
@@ -2148,7 +2276,7 @@ pub async fn contract_based_routing_update_configs(
         cache_entries_to_redact,
     )
     .await
-    .map_err(|e| logger::error!("unable to publish into the redact channel for evicting the contract based routing config cache {e:?}"));
+    .map_err(|e| router_env::logger::error!("unable to publish into the redact channel for evicting the contract based routing config cache {e:?}"));
 
     let new_record = record.foreign_into();
 
@@ -2324,15 +2452,13 @@ pub async fn migrate_rules_for_profile(
 ) -> RouterResult<routing_types::RuleMigrationResult> {
     use api_models::routing::StaticRoutingAlgorithm as EuclidAlgorithm;
 
-    use crate::services::logger;
-
     let profile_id = query_params.profile_id.clone();
     let db = state.store.as_ref();
     let key_manager_state = &(&state).into();
     let merchant_key_store = merchant_context.get_merchant_key_store();
     let merchant_id = merchant_context.get_merchant_account().get_id();
 
-    core_utils::validate_and_get_business_profile(
+    let business_profile = core_utils::validate_and_get_business_profile(
         db,
         key_manager_state,
         merchant_key_store,
@@ -2345,7 +2471,29 @@ pub async fn migrate_rules_for_profile(
         id: profile_id.get_string_repr().to_owned(),
     })?;
 
-    let routing_metadatas: Vec<diesel_models::routing_algorithm::RoutingProfileMetadata> = state
+    #[cfg(feature = "v1")]
+    let active_payment_routing_ids: Vec<Option<common_utils::id_type::RoutingId>> = vec![
+        business_profile
+            .get_payment_routing_algorithm()
+            .attach_printable("Failed to get payment routing algorithm")?
+            .unwrap_or_default()
+            .algorithm_id,
+        business_profile
+            .get_payout_routing_algorithm()
+            .attach_printable("Failed to get payout routing algorithm")?
+            .unwrap_or_default()
+            .algorithm_id,
+        business_profile
+            .get_frm_routing_algorithm()
+            .attach_printable("Failed to get frm routing algorithm")?
+            .unwrap_or_default()
+            .algorithm_id,
+    ];
+
+    #[cfg(feature = "v2")]
+    let active_payment_routing_ids = [business_profile.routing_algorithm_id.clone()];
+
+    let routing_metadatas = state
         .store
         .list_routing_algorithm_metadata_by_profile_id(
             &profile_id,
@@ -2358,111 +2506,119 @@ pub async fn migrate_rules_for_profile(
     let mut response_list = Vec::new();
     let mut error_list = Vec::new();
 
-    for routing_metadata in routing_metadatas
-        .into_iter()
-        .filter(|algo| algo.metadata_is_advanced_rule_for_payments())
-    {
-        match db
-            .find_routing_algorithm_by_profile_id_algorithm_id(
-                &profile_id,
-                &routing_metadata.algorithm_id,
-            )
+    let mut push_error = |algorithm_id, msg: String| {
+        error_list.push(RuleMigrationError {
+            profile_id: profile_id.clone(),
+            algorithm_id,
+            error: msg,
+        });
+    };
+
+    for routing_metadata in routing_metadatas {
+        let algorithm_id = routing_metadata.algorithm_id.clone();
+        let algorithm = match db
+            .find_routing_algorithm_by_profile_id_algorithm_id(&profile_id, &algorithm_id)
             .await
         {
-            Ok(algorithm) => {
-                let parsed_result = algorithm
-                    .algorithm_data
-                    .parse_value::<EuclidAlgorithm>("EuclidAlgorithm");
-
-                match parsed_result {
-                    Ok(EuclidAlgorithm::Advanced(program)) => match program.try_into() {
-                        Ok(internal_program) => {
-                            let routing_rule = RoutingRule {
-                                rule_id: Some(
-                                    algorithm.algorithm_id.clone().get_string_repr().to_string(),
-                                ),
-                                name: algorithm.name.clone(),
-                                description: algorithm.description.clone(),
-                                created_by: profile_id.get_string_repr().to_string(),
-                                algorithm: StaticRoutingAlgorithm::Advanced(internal_program),
-                                metadata: None,
-                            };
-
-                            let result = create_de_euclid_routing_algo(&state, &routing_rule).await;
-
-                            match result {
-                                Ok(decision_engine_routing_id) => {
-                                    let response = RuleMigrationResponse {
-                                        profile_id: profile_id.clone(),
-                                        euclid_algorithm_id: algorithm.algorithm_id.clone(),
-                                        decision_engine_algorithm_id: decision_engine_routing_id,
-                                    };
-                                    response_list.push(response);
-                                }
-                                Err(err) => {
-                                    logger::error!(
-                                        decision_engine_rule_migration_error = ?err,
-                                        algorithm_id = ?algorithm.algorithm_id,
-                                        "Failed to insert into decision engine"
-                                    );
-                                    error_list.push(RuleMigrationError {
-                                        profile_id: profile_id.clone(),
-                                        algorithm_id: algorithm.algorithm_id.clone(),
-                                        error: format!("Insertion error: {:?}", err),
-                                    });
-                                }
-                            }
-                        }
-                        Err(e) => {
-                            logger::error!(
-                                decision_engine_rule_migration_error = ?e,
-                                algorithm_id = ?algorithm.algorithm_id,
-                                "Failed to convert program"
-                            );
-                            error_list.push(RuleMigrationError {
-                                profile_id: profile_id.clone(),
-                                algorithm_id: algorithm.algorithm_id.clone(),
-                                error: format!("Program conversion error: {:?}", e),
-                            });
-                        }
-                    },
-                    Err(e) => {
-                        logger::error!(
-                            decision_engine_rule_migration_error = ?e,
-                            algorithm_id = ?algorithm.algorithm_id,
-                            "Failed to parse EuclidAlgorithm"
-                        );
-                        error_list.push(RuleMigrationError {
-                            profile_id: profile_id.clone(),
-                            algorithm_id: algorithm.algorithm_id.clone(),
-                            error: format!("JSON parse error: {:?}", e),
-                        });
-                    }
-                    _ => {
-                        logger::info!(
-                            "decision_engine_rule_migration_error: Skipping non-advanced algorithm {:?}",
-                            algorithm.algorithm_id
-                        );
-                        error_list.push(RuleMigrationError {
-                            profile_id: profile_id.clone(),
-                            algorithm_id: algorithm.algorithm_id.clone(),
-                            error: "Not an advanced algorithm".to_string(),
-                        });
-                    }
+            Ok(algo) => algo,
+            Err(e) => {
+                router_env::logger::error!(?e, ?algorithm_id, "Failed to fetch routing algorithm");
+                push_error(algorithm_id, format!("Fetch error: {:?}", e));
+                continue;
+            }
+        };
+
+        let parsed_result = algorithm
+            .algorithm_data
+            .parse_value::<EuclidAlgorithm>("EuclidAlgorithm");
+
+        let maybe_static_algorithm: Option<StaticRoutingAlgorithm> = match parsed_result {
+            Ok(EuclidAlgorithm::Advanced(program)) => match program.try_into() {
+                Ok(ip) => Some(StaticRoutingAlgorithm::Advanced(ip)),
+                Err(e) => {
+                    router_env::logger::error!(
+                        ?e,
+                        ?algorithm_id,
+                        "Failed to convert advanced program"
+                    );
+                    push_error(algorithm_id.clone(), format!("Conversion error: {:?}", e));
+                    None
                 }
+            },
+            Ok(EuclidAlgorithm::Single(conn)) => {
+                Some(StaticRoutingAlgorithm::Single(Box::new(conn.into())))
             }
-            Err(e) => {
-                logger::error!(
-                    decision_engine_rule_migration_error = ?e,
-                    algorithm_id = ?routing_metadata.algorithm_id,
-                    "Failed to fetch routing algorithm"
+            Ok(EuclidAlgorithm::Priority(connectors)) => Some(StaticRoutingAlgorithm::Priority(
+                connectors.into_iter().map(Into::into).collect(),
+            )),
+            Ok(EuclidAlgorithm::VolumeSplit(splits)) => Some(StaticRoutingAlgorithm::VolumeSplit(
+                splits.into_iter().map(Into::into).collect(),
+            )),
+            Ok(EuclidAlgorithm::ThreeDsDecisionRule(_)) => {
+                router_env::logger::info!(
+                    ?algorithm_id,
+                    "Skipping 3DS rule migration (not supported yet)"
                 );
-                error_list.push(RuleMigrationError {
+                push_error(algorithm_id.clone(), "3DS migration not implemented".into());
+                None
+            }
+            Err(e) => {
+                router_env::logger::error!(?e, ?algorithm_id, "Failed to parse algorithm");
+                push_error(algorithm_id.clone(), format!("Parse error: {:?}", e));
+                None
+            }
+        };
+
+        let Some(static_algorithm) = maybe_static_algorithm else {
+            continue;
+        };
+
+        let routing_rule = RoutingRule {
+            rule_id: Some(algorithm.algorithm_id.clone().get_string_repr().to_string()),
+            name: algorithm.name.clone(),
+            description: algorithm.description.clone(),
+            created_by: profile_id.get_string_repr().to_string(),
+            algorithm: static_algorithm,
+            algorithm_for: algorithm.algorithm_for.into(),
+            metadata: Some(RoutingMetadata {
+                kind: algorithm.kind,
+            }),
+        };
+
+        match create_de_euclid_routing_algo(&state, &routing_rule).await {
+            Ok(decision_engine_routing_id) => {
+                let mut is_active_rule = false;
+                if active_payment_routing_ids.contains(&Some(algorithm.algorithm_id.clone())) {
+                    link_de_euclid_routing_algorithm(
+                        &state,
+                        ActivateRoutingConfigRequest {
+                            created_by: profile_id.get_string_repr().to_string(),
+                            routing_algorithm_id: decision_engine_routing_id.clone(),
+                        },
+                    )
+                    .await
+                    .change_context(errors::ApiErrorResponse::InternalServerError)
+                    .attach_printable("unable to link active routing algorithm")?;
+                    is_active_rule = true;
+                }
+                response_list.push(RuleMigrationResponse {
                     profile_id: profile_id.clone(),
-                    algorithm_id: routing_metadata.algorithm_id.clone(),
-                    error: format!("Fetch error: {:?}", e),
+                    euclid_algorithm_id: algorithm.algorithm_id.clone(),
+                    decision_engine_algorithm_id: decision_engine_routing_id,
+                    is_active_rule,
                 });
             }
+            Err(err) => {
+                router_env::logger::error!(
+                    decision_engine_rule_migration_error = ?err,
+                    algorithm_id = ?algorithm.algorithm_id,
+                    "Failed to insert into decision engine"
+                );
+                push_error(
+                    algorithm.algorithm_id.clone(),
+                    format!("Insertion error: {:?}", err),
+                );
+            }
         }
     }
 
diff --git a/crates/router/src/core/routing/helpers.rs b/crates/router/src/core/routing/helpers.rs
index 666d0020260..e73c9c5bc75 100644
--- a/crates/router/src/core/routing/helpers.rs
+++ b/crates/router/src/core/routing/helpers.rs
@@ -72,6 +72,7 @@ pub const CONTRACT_BASED_DYNAMIC_ROUTING_ALGORITHM: &str =
 
 pub const DECISION_ENGINE_RULE_CREATE_ENDPOINT: &str = "rule/create";
 pub const DECISION_ENGINE_RULE_UPDATE_ENDPOINT: &str = "rule/update";
+pub const DECISION_ENGINE_RULE_GET_ENDPOINT: &str = "rule/get";
 pub const DECISION_ENGINE_RULE_DELETE_ENDPOINT: &str = "rule/delete";
 pub const DECISION_ENGINE_MERCHANT_BASE_ENDPOINT: &str = "merchant-account";
 pub const DECISION_ENGINE_MERCHANT_CREATE_ENDPOINT: &str = "merchant-account/create";
@@ -2433,6 +2434,37 @@ pub async fn update_decision_engine_dynamic_routing_setup(
     Ok(())
 }
 
+#[cfg(all(feature = "v1", feature = "dynamic_routing"))]
+pub async fn get_decision_engine_active_dynamic_routing_algorithm(
+    state: &SessionState,
+    profile_id: &id_type::ProfileId,
+    dynamic_routing_type: open_router::DecisionEngineDynamicAlgorithmType,
+) -> RouterResult<Option<open_router::DecisionEngineConfigSetupRequest>> {
+    logger::debug!(
+        "decision_engine_euclid: GET api call for decision active {:?} routing algorithm",
+        dynamic_routing_type
+    );
+    let request = open_router::GetDecisionEngineConfigRequest {
+        merchant_id: profile_id.get_string_repr().to_owned(),
+        config: dynamic_routing_type,
+    };
+    let response: Option<open_router::DecisionEngineConfigSetupRequest> =
+        routing_utils::ConfigApiClient::send_decision_engine_request(
+            state,
+            services::Method::Post,
+            DECISION_ENGINE_RULE_GET_ENDPOINT,
+            Some(request),
+            None,
+            None,
+        )
+        .await
+        .change_context(errors::ApiErrorResponse::InternalServerError)
+        .attach_printable("Failed to get active dynamic algorithm from decision engine")?
+        .response;
+
+    Ok(response)
+}
+
 #[cfg(all(feature = "dynamic_routing", feature = "v1"))]
 #[instrument(skip_all)]
 pub async fn disable_decision_engine_dynamic_routing_setup(
@@ -2545,11 +2577,11 @@ pub async fn delete_decision_engine_merchant(
         DECISION_ENGINE_MERCHANT_BASE_ENDPOINT,
         profile_id.get_string_repr()
     );
-    routing_utils::ConfigApiClient::send_decision_engine_request_without_response_parsing::<()>(
+    routing_utils::ConfigApiClient::send_decision_engine_request::<_, String>(
         state,
         services::Method::Delete,
         &path,
-        None,
+        None::<id_type::ProfileId>,
         None,
         None,
     )

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

## Description
<!-- Describe your changes in detail -->

This pull request introduces a new field, `routing_result_source`, across various structs and enums in the codebase to specify the source of routing decision results. This field allows for flexibility in choosing between Hyperswitch's inbuilt routing engine and an external decision engine. The changes span multiple modules, including API models, database models, and domain models.

Also, while creating the routing rule decision engine rule_id will be prefilled with the hyperswitch rule_id, so both systems will have the same id

### 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).
-->
Add profile-level config to choose between Hyperswitch routing result and Decision engine routing result

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

1. Create a merchant account, API key, profile, connector 

<details>
<summary>2. Create Routing Rule </summary>

Request

    curl --location 'http://localhost:8080/routing' \
    --header 'X-Profile-Id: pro_Wt98jyZeU8jaQ3cr9zDP' \
    --header 'api-key: dev_yUplJwLqvtAJiqOMmiamv78Xj9k1DkaEK3kxRExy4baMX4tsSVpEm55eTohlaifG' \
    --data '{
        "name": "Rule Based Routing-2025-05-07",
        "description": "This is a rule based routing created at Wed, 07 May 2025 13:03:39 GMT",
        "profile_id": "pro_Wt98jyZeU8jaQ3cr9zDP",
        "algorithm": {
            "type": "advanced",
            "data": {
                "defaultSelection": {
                    "type": "priority",
                    "data": []
                },
                "metadata": {},
                "rules": [
                    {
                        "name": "rule_1",
                        "connectorSelection": {
                            "type": "priority",
                            "data": [
                                {
                                    "connector": "cybersource",
                                    "merchant_connector_id": "mca_EU1XaLVDXZumYK8r3ft7"
                                }
                            ]
                        },
                        "statements": [
                            {
                                "condition": [
                                    {
                                        "lhs": "billing_country",
                                        "comparison": "equal",
                                        "value": {
                                            "type": "enum_variant",
                                            "value": "India"
                                        },
                                        "metadata": {}
                                    }
                                ],
                                "nested": null
                            }
                        ]
                    }
                ]
            }
        }
    }'
    
Response

    {
        "id": "routing_rFQyeEY3ujqyjtjJIlDB",
        "profile_id": "pro_Wt98jyZeU8jaQ3cr9zDP",
        "name": "Rule Based Routing-2025-05-07",
        "kind": "advanced",
        "description": "This is a rule based routing created at Wed, 07 May 2025 13:03:39 GMT",
        "created_at": 1750082560,
        "modified_at": 1750082560,
        "algorithm_for": "payment",
        "decision_engine_routing_id": "routing_rFQyeEY3ujqyjtjJIlDB"
    }

</details>

<details>
<summary>3. List Routing Rules </summary>

Request 

      curl --location 'http://localhost:8080/configs/' \
      --header 'Content-Type: application/json' \
      --header 'Accept: application/json' \
      --header 'api-key: test_admin' \
      --data '{
          "key": "routing_result_source_pro_Wt98jyZeU8jaQ3cr9zDP",
          "value": "decision_engine"
      }'
    
Response

     {
         key: "routing_result_source_pro_Wt98jyZeU8jaQ3cr9zDP",
         value: "decision_engine" 
     }
 
Rules are getting fetched from hyperswitch 
<img width="1324" alt="Screenshot 2025-06-16 at 7 35 55 PM" src="https://github.com/user-attachments/assets/2b655974-444f-4bf4-8b14-2f6f0b10f055" />

</details>
<details>
<summary>4. Change Profile config to use decision engine result directly </summary>

Request 


    curl --location 'http://localhost:8080/account/merchant_1750082519/business_profile/pro_Wt98jyZeU8jaQ3cr9zDP' \
    --header 'Content-Type: application/json' \
    --header 'Accept: application/json' \
    --header 'api-key: dev_j1154b6pVqBwC32eFEcQDZDIBKGGEBT3zh3VMi3cWUpFcurnQqBD2SDLXmFbaJem' \
    --data '{
            "dynamic_routing_algorithm": {
                "routing_result_source": "decision_engine"
            }
        }'

Response

    {
        "merchant_id": "merchant_1750082519",
        "profile_id": "pro_Wt98jyZeU8jaQ3cr9zDP",
        "profile_name": "1750082525",
        "return_url": "https://google.com/success",
        "enable_payment_response_hash": true,
        "payment_response_hash_key": "MkZkTPS7BIusBLHYEmQCKB0FaGiTkTsASUBTQCfuEuGwCRrN9zso4M0qIOKValbF",
        "redirect_to_merchant_with_http_post": false,
        "webhook_details": {
            "webhook_version": "1.0.1",
            "webhook_username": "ekart_retail",
            "webhook_password": "password_ekart@123",
            "webhook_url": null,
            "payment_created_enabled": true,
            "payment_succeeded_enabled": true,
            "payment_failed_enabled": true
        },
        "metadata": null,
        "routing_algorithm": null,
        "intent_fulfillment_time": 900,
        "frm_routing_algorithm": null,
        "payout_routing_algorithm": null,
        "applepay_verified_domains": null,
        "session_expiry": 900,
        "payment_link_config": null,
        "authentication_connector_details": null,
        "use_billing_as_payment_method_billing": true,
        "extended_card_info_config": null,
        "collect_shipping_details_from_wallet_connector": false,
        "collect_billing_details_from_wallet_connector": false,
        "always_collect_shipping_details_from_wallet_connector": false,
        "always_collect_billing_details_from_wallet_connector": false,
        "is_connector_agnostic_mit_enabled": false,
        "payout_link_config": null,
        "outgoing_webhook_custom_http_headers": null,
        "tax_connector_id": null,
        "is_tax_connector_enabled": false,
        "is_network_tokenization_enabled": false,
        "is_auto_retries_enabled": false,
        "max_auto_retries_enabled": null,
        "always_request_extended_authorization": null,
        "is_click_to_pay_enabled": false,
        "authentication_product_ids": null,
        "card_testing_guard_config": {
            "card_ip_blocking_status": "disabled",
            "card_ip_blocking_threshold": 3,
            "guest_user_card_blocking_status": "disabled",
            "guest_user_card_blocking_threshold": 10,
            "customer_id_blocking_status": "disabled",
            "customer_id_blocking_threshold": 5,
            "card_testing_guard_expiry": 3600
        },
        "is_clear_pan_retries_enabled": false,
        "force_3ds_challenge": false,
        "is_debit_routing_enabled": false,
        "merchant_business_country": null,
        "is_pre_network_tokenization_enabled": false,
        "acquirer_configs": null,
        "is_iframe_redirection_enabled": null,
        "merchant_category_code": null
    }

</details>
<details>
<summary>5. List Routing Rules again</summary>
This time result from the decision engine should be used to return results

Request 

    curl --location 'http://localhost:8080/routing/list/profile?limit=100' \
    --header 'sec-ch-ua-platform: "macOS"' \
    --header 'Referer: https://app.hyperswitch.io/dashboard/routing' \
    --header 'sec-ch-ua: "Chromium";v="136", "Google Chrome";v="136", "Not.A/Brand";v="99"' \
    --header 'X-Profile-Id: pro_Wt98jyZeU8jaQ3cr9zDP' \
    --header 'sec-ch-ua-mobile: ?0' \
    --header 'api-key: dev_yUplJwLqvtAJiqOMmiamv78Xj9k1DkaEK3kxRExy4baMX4tsSVpEm55eTohlaifG' \
    --header 'User-Agent: Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/136.0.0.0 Safari/537.36' \
    --header 'Content-Type: application/json'
    
Response

      [
          {
              "id": "routing_rFQyeEY3ujqyjtjJIlDB",
              "profile_id": "pro_Wt98jyZeU8jaQ3cr9zDP",
              "name": "Rule Based Routing-2025-05-07",
              "kind": "advanced",
              "description": "This is a rule based routing created at Wed, 07 May 2025 13:03:39 GMT",
              "created_at": 1750082560,
              "modified_at": 1750082560,
              "algorithm_for": "payment",
              "decision_engine_routing_id": null
          }
      ]


<img width="1320" alt="Screenshot 2025-06-16 at 7 37 57 PM" src="https://github.com/user-attachments/assets/df30369e-30b8-4565-ab29-c5289305409e" />
</details>

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