Issue:
Bug: Feat(euclid): Hyperswitch integration with decision engine



## Description

This PR integrates the Euclid-based decision engine with Hyperswitch’s dynamic routing infrastructure. It enables dual routing logic: legacy DSL-based evaluation and decision-engine-backed configurations. The integration ensures routing algorithms can be authored, activated, and evaluated via Euclid and linked seamlessly to merchant profiles.

## Outcomes

- Adds support for storing and linking decision-engine routing IDs (`decision_engine_routing_id`) in the existing `routing_algorithm` table.
- Enables creation and linking of Euclid routing configurations during algorithm creation and configuration linking flows.
- Enhances observability with logging and tracing of decision engine interactions.
- Preserves backward compatibility with existing static/dynamic routing mechanisms.

## Diff Hunk Explanation

### `crates/api_models/src/routing.rs`
- Added `decision_engine_routing_id` to `RoutingDictionaryRecord` to expose the DE mapping in API responses.
- Updated `RoutingAlgorithmKind` with `PartialEq` to aid in logic branching.

### `crates/diesel_models/src/routing_algorithm.rs`
- Included new optional field `decision_engine_routing_id` to map to Euclid's algorithm record.

### `crates/diesel_models/src/schema.rs`
- Altered `routing_algorithm` schema to include `decision_engine_routing_id VARCHAR(64)`.

### `crates/router/src/core/payments/routing.rs`
- Imported `perform_decision_euclid_routing` into static routing flow for test execution and future use.

### `crates/router/src/core/routing.rs`
- On routing creation, checks if the algorithm is a Euclid (Advanced) type, then creates it in the decision engine and stores the returned ID.
- On routing configuration linking, activates the respective routing config in the decision engine using `ActivateRoutingConfigRequest`.
- Enhances `retrieve_merchant_routing_dictionary` to fetch and append decision engine routing configs by profile ID.

### `crates/router/src/core/routing/transformers.rs`
- Ensures `decision_engine_routing_id` is preserved and transformed when converting between internal and API models.

### `migrations/2025-05-08-102850_add_de_euclid_id_in_routing_algorithm_table/*.sql`
- Adds and removes the `decision_engine_routing_id` column in the `routing_algorithm` table via Diesel migrations.


## 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)?
-->
Here is the complete testing [guidelines](https://docs.google.com/document/d/11Xr_KWA5ire4qBXLcALSe7JeHeRjn8_dzvW7HBh91rY/edit?tab=t.0).


Pull request description:
## Type of Change
<!-- Put an `x` in the boxes that apply -->

- [ ] Bugfix
- [x] New feature
- [ ] Enhancement
- [ ] Refactoring
- [ ] Dependency updates
- [ ] Documentation
- [ ] CI/CD

## Description

This PR integrates the Euclid-based decision engine with Hyperswitch’s dynamic routing infrastructure. It enables dual routing logic: legacy DSL-based evaluation and decision-engine-backed configurations. The integration ensures routing algorithms can be authored, activated, and evaluated via Euclid and linked seamlessly to merchant profiles.

## Outcomes

- Adds support for storing and linking decision-engine routing IDs (`decision_engine_routing_id`) in the existing `routing_algorithm` table.
- Enables creation and linking of Euclid routing configurations during algorithm creation and configuration linking flows.
- Enhances observability with logging and tracing of decision engine interactions.
- Preserves backward compatibility with existing static/dynamic routing mechanisms.

## Diff Hunk Explanation

### `crates/api_models/src/routing.rs`
- Added `decision_engine_routing_id` to `RoutingDictionaryRecord` to expose the DE mapping in API responses.
- Updated `RoutingAlgorithmKind` with `PartialEq` to aid in logic branching.

### `crates/diesel_models/src/routing_algorithm.rs`
- Included new optional field `decision_engine_routing_id` to map to Euclid's algorithm record.

### `crates/diesel_models/src/schema.rs`
- Altered `routing_algorithm` schema to include `decision_engine_routing_id VARCHAR(64)`.

### `crates/router/src/core/payments/routing.rs`
- Imported `perform_decision_euclid_routing` into static routing flow for test execution and future use.

### `crates/router/src/core/routing.rs`
- On routing creation, checks if the algorithm is a Euclid (Advanced) type, then creates it in the decision engine and stores the returned ID.
- On routing configuration linking, activates the respective routing config in the decision engine using `ActivateRoutingConfigRequest`.
- Enhances `retrieve_merchant_routing_dictionary` to fetch and append decision engine routing configs by profile ID.

### `crates/router/src/core/routing/transformers.rs`
- Ensures `decision_engine_routing_id` is preserved and transformed when converting between internal and API models.

### `migrations/2025-05-08-102850_add_de_euclid_id_in_routing_algorithm_table/*.sql`
- Adds and removes the `decision_engine_routing_id` column in the `routing_algorithm` table via Diesel migrations.



### Additional Changes

- [ ] This PR modifies the API contract
- [x] This PR modifies the database schema
- [ ] This PR modifies application configuration/environment variables

<!--
Provide links to the files with corresponding changes.

Following are the paths where you can find config files:
1. `config`
2. `crates/router/src/configs`
3. `loadtest/config`
-->


## Motivation and Context
<!--
Why is this change required? What problem does it solve?
If it fixes an open issue, please link to the issue here.

If you don't have an issue, we'd recommend starting with one first so the PR
can focus on the implementation (unless it is an obvious bug or documentation fix
that will have little conversation).
-->


## How did you test it?
<!--
Did you write an integration/unit/API test to verify the code changes?
Or did you test this change manually (provide relevant screenshots)?
-->
Here is the complete testing [guidelines](https://docs.google.com/document/d/1xL3EJdMoCLXP5epOeqNsNYvqFqlJyhxss3LahQQ5xGU/edit?usp=sharing).

## Migrations
```
ALTER TABLE routing_algorithm
ADD COLUMN decision_engine_routing_id VARCHAR(64);
```

## 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`
- [ ] I reviewed the submitted code
- [ ] I added unit tests for my changes where possible


Testing:

Here is the complete testing [guidelines](https://docs.google.com/document/d/1xL3EJdMoCLXP5epOeqNsNYvqFqlJyhxss3LahQQ5xGU/edit?usp=sharing).

Patch:
diff --git a/api-reference-v2/openapi_spec.json b/api-reference-v2/openapi_spec.json
index 12c30cc61f0..197f9fdbda8 100644
--- a/api-reference-v2/openapi_spec.json
+++ b/api-reference-v2/openapi_spec.json
@@ -21668,6 +21668,10 @@
               }
             ],
             "nullable": true
+          },
+          "decision_engine_routing_id": {
+            "type": "string",
+            "nullable": true
           }
         }
       },
diff --git a/crates/api_models/src/routing.rs b/crates/api_models/src/routing.rs
index 90cce48574b..c929bb3ea5d 100644
--- a/crates/api_models/src/routing.rs
+++ b/crates/api_models/src/routing.rs
@@ -269,7 +269,9 @@ impl RoutableConnectorChoiceWithStatus {
     }
 }
 
-#[derive(Debug, Copy, Clone, serde::Serialize, serde::Deserialize, strum::Display, ToSchema)]
+#[derive(
+    Debug, Copy, Clone, PartialEq, serde::Serialize, serde::Deserialize, strum::Display, ToSchema,
+)]
 #[serde(rename_all = "snake_case")]
 #[strum(serialize_all = "snake_case")]
 pub enum RoutingAlgorithmKind {
@@ -484,6 +486,7 @@ pub struct RoutingDictionaryRecord {
     pub created_at: i64,
     pub modified_at: i64,
     pub algorithm_for: Option<TransactionType>,
+    pub decision_engine_routing_id: Option<String>,
 }
 
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize, ToSchema)]
diff --git a/crates/diesel_models/src/routing_algorithm.rs b/crates/diesel_models/src/routing_algorithm.rs
index 5f5a0560811..4e6e9212885 100644
--- a/crates/diesel_models/src/routing_algorithm.rs
+++ b/crates/diesel_models/src/routing_algorithm.rs
@@ -17,6 +17,7 @@ pub struct RoutingAlgorithm {
     pub created_at: time::PrimitiveDateTime,
     pub modified_at: time::PrimitiveDateTime,
     pub algorithm_for: enums::TransactionType,
+    pub decision_engine_routing_id: Option<String>,
 }
 
 pub struct RoutingAlgorithmMetadata {
diff --git a/crates/diesel_models/src/schema.rs b/crates/diesel_models/src/schema.rs
index 6f978a7afec..50474dd9491 100644
--- a/crates/diesel_models/src/schema.rs
+++ b/crates/diesel_models/src/schema.rs
@@ -1407,6 +1407,8 @@ diesel::table! {
         created_at -> Timestamp,
         modified_at -> Timestamp,
         algorithm_for -> TransactionType,
+        #[max_length = 64]
+        decision_engine_routing_id -> Nullable<Varchar>,
     }
 }
 
diff --git a/crates/router/src/core/payments/routing.rs b/crates/router/src/core/payments/routing.rs
index 0b5b7003c07..9280415412d 100644
--- a/crates/router/src/core/payments/routing.rs
+++ b/crates/router/src/core/payments/routing.rs
@@ -1,4 +1,5 @@
 mod transformers;
+pub mod utils;
 
 #[cfg(all(feature = "v1", feature = "dynamic_routing"))]
 use std::collections::hash_map;
@@ -50,6 +51,7 @@ use rand::SeedableRng;
 use router_env::{instrument, tracing};
 use rustc_hash::FxHashMap;
 use storage_impl::redis::cache::{CacheKey, CGRAPH_CACHE, ROUTING_CACHE};
+use utils::perform_decision_euclid_routing;
 
 #[cfg(feature = "v2")]
 use crate::core::admin;
@@ -466,7 +468,27 @@ pub async fn perform_static_routing_v1(
                 }
             };
 
-            execute_dsl_and_get_connector_v1(backend_input, interpreter)?
+            let de_euclid_connectors = perform_decision_euclid_routing(
+                state,
+                backend_input.clone(),
+                business_profile.get_id().get_string_repr().to_string(),
+            )
+            .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>>();
+            utils::compare_and_log_result(
+                de_euclid_connectors,
+                connectors,
+                "evaluate_routing".to_string(),
+            );
+            routable_connectors
         }
     })
 }
diff --git a/crates/router/src/core/payments/routing/utils.rs b/crates/router/src/core/payments/routing/utils.rs
new file mode 100644
index 00000000000..551694a0aaf
--- /dev/null
+++ b/crates/router/src/core/payments/routing/utils.rs
@@ -0,0 +1,736 @@
+use std::collections::{HashMap, HashSet};
+
+use api_models::routing as api_routing;
+use async_trait::async_trait;
+use common_utils::id_type;
+use diesel_models::{enums, routing_algorithm};
+use error_stack::ResultExt;
+use euclid::{backend::BackendInput, frontend::ast};
+use serde::{Deserialize, Serialize};
+
+use super::RoutingResult;
+use crate::{
+    core::errors,
+    routes::SessionState,
+    services::{self, logger},
+    types::transformers::ForeignInto,
+};
+
+// New Trait for handling Euclid API calls
+#[async_trait]
+pub trait EuclidApiHandler {
+    async fn send_euclid_request<Req, Res>(
+        state: &SessionState,
+        http_method: services::Method,
+        path: &str,
+        request_body: Option<Req>, // Option to handle GET/DELETE requests without body
+        timeout: Option<u64>,
+    ) -> RoutingResult<Res>
+    where
+        Req: Serialize + Send + Sync + 'static,
+        Res: serde::de::DeserializeOwned + Send + 'static + std::fmt::Debug;
+
+    async fn send_euclid_request_without_response_parsing<Req>(
+        state: &SessionState,
+        http_method: services::Method,
+        path: &str,
+        request_body: Option<Req>,
+        timeout: Option<u64>,
+    ) -> RoutingResult<()>
+    where
+        Req: Serialize + Send + Sync + 'static;
+}
+
+// Struct to implement the EuclidApiHandler trait
+pub struct EuclidApiClient;
+
+impl EuclidApiClient {
+    async fn build_and_send_euclid_http_request<Req>(
+        state: &SessionState,
+        http_method: services::Method,
+        path: &str,
+        request_body: Option<Req>,
+        timeout: Option<u64>,
+        context_message: &str,
+    ) -> RoutingResult<reqwest::Response>
+    where
+        Req: Serialize + Send + Sync + 'static,
+    {
+        let euclid_base_url = &state.conf.open_router.url;
+        let url = format!("{}/{}", euclid_base_url, path);
+        logger::debug!(euclid_api_call_url = %url, euclid_request_path = %path, http_method = ?http_method, "decision_engine_euclid: Initiating Euclid API call ({})", context_message);
+
+        let mut request_builder = services::RequestBuilder::new()
+            .method(http_method)
+            .url(&url);
+
+        if let Some(body_content) = request_body {
+            let body = common_utils::request::RequestContent::Json(Box::new(body_content));
+            request_builder = request_builder.set_body(body);
+        }
+
+        let http_request = request_builder.build();
+        logger::info!(?http_request, euclid_request_path = %path, "decision_engine_euclid: Constructed Euclid API request details ({})", context_message);
+
+        state
+            .api_client
+            .send_request(state, http_request, timeout, false)
+            .await
+            .change_context(errors::RoutingError::DslExecutionError)
+            .attach_printable_lazy(|| {
+                format!(
+                    "Euclid API call to path '{}' unresponsive ({})",
+                    path, context_message
+                )
+            })
+    }
+}
+
+#[async_trait]
+impl EuclidApiHandler for EuclidApiClient {
+    async fn send_euclid_request<Req, Res>(
+        state: &SessionState,
+        http_method: services::Method,
+        path: &str,
+        request_body: Option<Req>, // Option to handle GET/DELETE requests without body
+        timeout: Option<u64>,
+    ) -> RoutingResult<Res>
+    where
+        Req: Serialize + Send + Sync + 'static,
+        Res: serde::de::DeserializeOwned + Send + 'static + std::fmt::Debug,
+    {
+        let response = Self::build_and_send_euclid_http_request(
+            state,
+            http_method,
+            path,
+            request_body,
+            timeout,
+            "parsing response",
+        )
+        .await?;
+        logger::debug!(euclid_response = ?response, euclid_request_path = %path, "decision_engine_euclid: Received raw response from Euclid API");
+
+        let parsed_response = response
+            .json::<Res>()
+            .await
+            .change_context(errors::RoutingError::GenericConversionError {
+                from: "ApiResponse".to_string(),
+                to: std::any::type_name::<Res>().to_string(),
+            })
+            .attach_printable_lazy(|| {
+                format!(
+                    "Unable to parse response of type '{}' received from Euclid API path: {}",
+                    std::any::type_name::<Res>(),
+                    path
+                )
+            })?;
+        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(parsed_response)
+    }
+
+    async fn send_euclid_request_without_response_parsing<Req>(
+        state: &SessionState,
+        http_method: services::Method,
+        path: &str,
+        request_body: Option<Req>,
+        timeout: Option<u64>,
+    ) -> RoutingResult<()>
+    where
+        Req: Serialize + Send + Sync + 'static,
+    {
+        let response = Self::build_and_send_euclid_http_request(
+            state,
+            http_method,
+            path,
+            request_body,
+            timeout,
+            "not parsing response",
+        )
+        .await?;
+
+        logger::debug!(euclid_response = ?response, euclid_request_path = %path, "decision_engine_routing: Received raw response from Euclid API");
+        Ok(())
+    }
+}
+
+const EUCLID_API_TIMEOUT: u64 = 5;
+
+pub async fn perform_decision_euclid_routing(
+    state: &SessionState,
+    input: BackendInput,
+    created_by: String,
+) -> RoutingResult<Vec<String>> {
+    logger::debug!("decision_engine_euclid: evaluate api call for euclid routing evaluation");
+
+    let routing_request = convert_backend_input_to_routing_eval(created_by, input)?;
+
+    let euclid_response: RoutingEvaluateResponse = EuclidApiClient::send_euclid_request(
+        state,
+        services::Method::Post,
+        "routing/evaluate",
+        Some(routing_request),
+        Some(EUCLID_API_TIMEOUT),
+    )
+    .await?;
+
+    logger::debug!(decision_engine_euclid_response=?euclid_response,"decision_engine_euclid");
+    logger::debug!(decision_engine_euclid_selected_connector=?euclid_response.evaluated_output,"decision_engine_euclid");
+
+    Ok(euclid_response.evaluated_output)
+}
+
+pub async fn create_de_euclid_routing_algo(
+    state: &SessionState,
+    routing_request: &RoutingRule,
+) -> RoutingResult<String> {
+    logger::debug!("decision_engine_euclid: create api call for euclid routing rule creation");
+
+    logger::debug!(decision_engine_euclid_request=?routing_request,"decision_engine_euclid");
+    let euclid_response: RoutingDictionaryRecord = EuclidApiClient::send_euclid_request(
+        state,
+        services::Method::Post,
+        "routing/create",
+        Some(routing_request.clone()),
+        Some(EUCLID_API_TIMEOUT),
+    )
+    .await?;
+
+    logger::debug!(decision_engine_euclid_parsed_response=?euclid_response,"decision_engine_euclid");
+    Ok(euclid_response.rule_id)
+}
+
+pub async fn link_de_euclid_routing_algorithm(
+    state: &SessionState,
+    routing_request: ActivateRoutingConfigRequest,
+) -> RoutingResult<()> {
+    logger::debug!("decision_engine_euclid: link api call for euclid routing algorithm");
+
+    EuclidApiClient::send_euclid_request_without_response_parsing(
+        state,
+        services::Method::Post,
+        "routing/activate",
+        Some(routing_request.clone()),
+        Some(EUCLID_API_TIMEOUT),
+    )
+    .await?;
+
+    logger::debug!(decision_engine_euclid_activated=?routing_request, "decision_engine_euclid: link_de_euclid_routing_algorithm completed");
+    Ok(())
+}
+
+pub async fn list_de_euclid_routing_algorithms(
+    state: &SessionState,
+    routing_list_request: ListRountingAlgorithmsRequest,
+) -> RoutingResult<Vec<api_routing::RoutingDictionaryRecord>> {
+    logger::debug!("decision_engine_euclid: list api call for euclid routing algorithms");
+    let created_by = routing_list_request.created_by;
+    let response: Vec<RoutingAlgorithmRecord> = EuclidApiClient::send_euclid_request(
+        state,
+        services::Method::Post,
+        format!("routing/list/{created_by}").as_str(),
+        None::<()>,
+        Some(EUCLID_API_TIMEOUT),
+    )
+    .await?;
+
+    Ok(response
+        .into_iter()
+        .map(routing_algorithm::RoutingProfileMetadata::from)
+        .map(ForeignInto::foreign_into)
+        .collect::<Vec<_>>())
+}
+
+pub fn compare_and_log_result<T: RoutingEq<T> + Serialize>(
+    de_result: Vec<T>,
+    result: Vec<T>,
+    flow: String,
+) {
+    let is_equal = de_result.len() == result.len()
+        && de_result
+            .iter()
+            .zip(result.iter())
+            .all(|(a, b)| T::is_equal(a, b));
+
+    if is_equal {
+        router_env::logger::info!(routing_flow=?flow, is_equal=?is_equal, "decision_engine_euclid");
+    } else {
+        router_env::logger::debug!(routing_flow=?flow, is_equal=?is_equal, de_response=?to_json_string(&de_result), hs_response=?to_json_string(&result), "decision_engine_euclid");
+    }
+}
+
+pub trait RoutingEq<T> {
+    fn is_equal(a: &T, b: &T) -> bool;
+}
+
+impl RoutingEq<Self> for api_routing::RoutingDictionaryRecord {
+    fn is_equal(a: &Self, b: &Self) -> bool {
+        a.name == b.name
+            && a.profile_id == b.profile_id
+            && a.description == b.description
+            && a.kind == b.kind
+            && a.algorithm_for == b.algorithm_for
+    }
+}
+
+impl RoutingEq<Self> for String {
+    fn is_equal(a: &Self, b: &Self) -> bool {
+        a.to_lowercase() == b.to_lowercase()
+    }
+}
+
+pub fn to_json_string<T: Serialize>(value: &T) -> String {
+    serde_json::to_string(value)
+        .map_err(|_| errors::RoutingError::GenericConversionError {
+            from: "T".to_string(),
+            to: "JsonValue".to_string(),
+        })
+        .unwrap_or_default()
+}
+
+#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
+pub struct ActivateRoutingConfigRequest {
+    pub created_by: String,
+    pub routing_algorithm_id: String,
+}
+
+#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
+pub struct ListRountingAlgorithmsRequest {
+    pub created_by: String,
+}
+
+// Maps Hyperswitch `BackendInput` to a `RoutingEvaluateRequest` compatible with Decision Engine
+pub fn convert_backend_input_to_routing_eval(
+    created_by: String,
+    input: BackendInput,
+) -> RoutingResult<RoutingEvaluateRequest> {
+    let mut params: HashMap<String, Option<ValueType>> = HashMap::new();
+
+    // Payment
+    params.insert(
+        "amount".to_string(),
+        Some(ValueType::Number(
+            input
+                .payment
+                .amount
+                .get_amount_as_i64()
+                .try_into()
+                .unwrap_or_default(),
+        )),
+    );
+    params.insert(
+        "currency".to_string(),
+        Some(ValueType::EnumVariant(input.payment.currency.to_string())),
+    );
+
+    if let Some(auth_type) = input.payment.authentication_type {
+        params.insert(
+            "authentication_type".to_string(),
+            Some(ValueType::EnumVariant(auth_type.to_string())),
+        );
+    }
+    if let Some(bin) = input.payment.card_bin {
+        params.insert("card_bin".to_string(), Some(ValueType::StrValue(bin)));
+    }
+    if let Some(capture_method) = input.payment.capture_method {
+        params.insert(
+            "capture_method".to_string(),
+            Some(ValueType::EnumVariant(capture_method.to_string())),
+        );
+    }
+    if let Some(country) = input.payment.business_country {
+        params.insert(
+            "business_country".to_string(),
+            Some(ValueType::EnumVariant(country.to_string())),
+        );
+    }
+    if let Some(country) = input.payment.billing_country {
+        params.insert(
+            "billing_country".to_string(),
+            Some(ValueType::EnumVariant(country.to_string())),
+        );
+    }
+    if let Some(label) = input.payment.business_label {
+        params.insert(
+            "business_label".to_string(),
+            Some(ValueType::StrValue(label)),
+        );
+    }
+    if let Some(sfu) = input.payment.setup_future_usage {
+        params.insert(
+            "setup_future_usage".to_string(),
+            Some(ValueType::EnumVariant(sfu.to_string())),
+        );
+    }
+
+    // PaymentMethod
+    if let Some(pm) = input.payment_method.payment_method {
+        params.insert(
+            "payment_method".to_string(),
+            Some(ValueType::EnumVariant(pm.to_string())),
+        );
+    }
+    if let Some(pmt) = input.payment_method.payment_method_type {
+        params.insert(
+            "payment_method_type".to_string(),
+            Some(ValueType::EnumVariant(pmt.to_string())),
+        );
+    }
+    if let Some(network) = input.payment_method.card_network {
+        params.insert(
+            "card_network".to_string(),
+            Some(ValueType::EnumVariant(network.to_string())),
+        );
+    }
+
+    // Mandate
+    if let Some(pt) = input.mandate.payment_type {
+        params.insert(
+            "payment_type".to_string(),
+            Some(ValueType::EnumVariant(pt.to_string())),
+        );
+    }
+    if let Some(mt) = input.mandate.mandate_type {
+        params.insert(
+            "mandate_type".to_string(),
+            Some(ValueType::EnumVariant(mt.to_string())),
+        );
+    }
+    if let Some(mat) = input.mandate.mandate_acceptance_type {
+        params.insert(
+            "mandate_acceptance_type".to_string(),
+            Some(ValueType::EnumVariant(mat.to_string())),
+        );
+    }
+
+    // Metadata
+    if let Some(meta) = input.metadata {
+        for (k, v) in meta.into_iter() {
+            params.insert(
+                k.clone(),
+                Some(ValueType::MetadataVariant(MetadataValue {
+                    key: k,
+                    value: v,
+                })),
+            );
+        }
+    }
+
+    Ok(RoutingEvaluateRequest {
+        created_by,
+        parameters: params,
+    })
+}
+
+//TODO: temporary change will be refactored afterwards
+#[derive(Clone, Debug, serde::Serialize, serde::Deserialize, PartialEq)]
+pub struct RoutingEvaluateRequest {
+    pub created_by: String,
+    pub parameters: HashMap<String, Option<ValueType>>,
+}
+
+#[derive(Debug, serde::Serialize, serde::Deserialize)]
+pub struct RoutingEvaluateResponse {
+    pub status: String,
+    pub output: serde_json::Value,
+    pub evaluated_output: Vec<String>,
+    pub eligible_connectors: Vec<String>,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
+pub struct MetadataValue {
+    pub key: String,
+    pub value: String,
+}
+
+/// Represents a value in the DSL
+#[derive(Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
+#[serde(tag = "type", content = "value", rename_all = "snake_case")]
+pub enum ValueType {
+    /// Represents a number literal
+    Number(u64),
+    /// Represents an enum variant
+    EnumVariant(String),
+    /// Represents a Metadata variant
+    MetadataVariant(MetadataValue),
+    /// Represents a arbitrary String value
+    StrValue(String),
+    GlobalRef(String),
+}
+
+pub type Metadata = HashMap<String, serde_json::Value>;
+/// Represents a number comparison for "NumberComparisonArrayValue"
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct NumberComparison {
+    pub comparison_type: ComparisonType,
+    pub number: u64,
+}
+
+/// Conditional comparison type
+#[derive(Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum ComparisonType {
+    Equal,
+    NotEqual,
+    LessThan,
+    LessThanEqual,
+    GreaterThan,
+    GreaterThanEqual,
+}
+
+/// Represents a single comparison condition.
+#[derive(Clone, Debug, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub struct Comparison {
+    /// The left hand side which will always be a domain input identifier like "payment.method.cardtype"
+    pub lhs: String,
+    /// The comparison operator
+    pub comparison: ComparisonType,
+    /// The value to compare against
+    pub value: ValueType,
+    /// Additional metadata that the Static Analyzer and Backend does not touch.
+    /// This can be used to store useful information for the frontend and is required for communication
+    /// between the static analyzer and the frontend.
+    // #[schema(value_type=HashMap<String, serde_json::Value>)]
+    pub metadata: Metadata,
+}
+
+/// Represents all the conditions of an IF statement
+/// eg:
+///
+/// ```text
+/// payment.method = card & payment.method.cardtype = debit & payment.method.network = diners
+/// ```
+pub type IfCondition = Vec<Comparison>;
+
+/// Represents an IF statement with conditions and optional nested IF statements
+///
+/// ```text
+/// payment.method = card {
+///     payment.method.cardtype = (credit, debit) {
+///         payment.method.network = (amex, rupay, diners)
+///     }
+/// }
+/// ```
+#[derive(Clone, Debug, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct IfStatement {
+    // #[schema(value_type=Vec<Comparison>)]
+    pub condition: IfCondition,
+    pub nested: Option<Vec<IfStatement>>,
+}
+
+/// Represents a rule
+///
+/// ```text
+/// rule_name: [stripe, adyen, checkout]
+/// {
+///     payment.method = card {
+///         payment.method.cardtype = (credit, debit) {
+///             payment.method.network = (amex, rupay, diners)
+///         }
+///
+///         payment.method.cardtype = credit
+///     }
+/// }
+/// ```
+#[derive(Clone, Debug, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+// #[aliases(RuleConnectorSelection = Rule<ConnectorSelection>)]
+pub struct Rule {
+    pub name: String,
+    #[serde(alias = "routingType")]
+    pub routing_type: RoutingType,
+    #[serde(alias = "routingOutput")]
+    pub output: Output,
+    pub statements: Vec<IfStatement>,
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum RoutingType {
+    Priority,
+    VolumeSplit,
+    VolumeSplitPriority,
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct VolumeSplit<T> {
+    pub split: u8,
+    pub output: T,
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub enum Output {
+    Priority(Vec<String>),
+    VolumeSplit(Vec<VolumeSplit<String>>),
+    VolumeSplitPriority(Vec<VolumeSplit<Vec<String>>>),
+}
+
+pub type Globals = HashMap<String, HashSet<ValueType>>;
+
+/// The program, having a default connector selection and
+/// a bunch of rules. Also can hold arbitrary metadata.
+#[derive(Clone, Debug, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+// #[aliases(ProgramConnectorSelection = Program<ConnectorSelection>)]
+pub struct Program {
+    pub globals: Globals,
+    pub default_selection: Output,
+    // #[schema(value_type=RuleConnectorSelection)]
+    pub rules: Vec<Rule>,
+    // #[schema(value_type=HashMap<String, serde_json::Value>)]
+    pub metadata: Option<Metadata>,
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub struct RoutingRule {
+    pub name: String,
+    pub description: Option<String>,
+    pub metadata: Option<RoutingMetadata>,
+    pub created_by: String,
+    pub algorithm: Program,
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub struct RoutingMetadata {
+    pub kind: enums::RoutingAlgorithmKind,
+    pub algorithm_for: enums::TransactionType,
+}
+
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub struct RoutingDictionaryRecord {
+    pub rule_id: String,
+    pub name: String,
+    pub created_at: time::PrimitiveDateTime,
+    pub modified_at: time::PrimitiveDateTime,
+}
+
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub struct RoutingAlgorithmRecord {
+    pub id: id_type::RoutingId,
+    pub name: String,
+    pub description: Option<String>,
+    pub created_by: id_type::ProfileId,
+    pub algorithm_data: Program,
+    pub metadata: Option<RoutingMetadata>,
+    pub created_at: time::PrimitiveDateTime,
+    pub modified_at: time::PrimitiveDateTime,
+}
+
+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(),
+            ),
+        };
+        Self {
+            profile_id: record.created_by,
+            algorithm_id: record.id,
+            name: record.name,
+            description: record.description,
+            kind,
+            created_at: record.created_at,
+            modified_at: record.modified_at,
+            algorithm_for,
+        }
+    }
+}
+use api_models::routing::{ConnectorSelection, RoutableConnectorChoice};
+impl From<ast::Program<ConnectorSelection>> for Program {
+    fn from(p: ast::Program<ConnectorSelection>) -> Self {
+        Self {
+            globals: HashMap::new(),
+            default_selection: convert_output(p.default_selection),
+            rules: p.rules.into_iter().map(convert_rule).collect(),
+            metadata: Some(p.metadata),
+        }
+    }
+}
+
+fn convert_rule(rule: ast::Rule<ConnectorSelection>) -> Rule {
+    let routing_type = match &rule.connector_selection {
+        ConnectorSelection::Priority(_) => RoutingType::Priority,
+        ConnectorSelection::VolumeSplit(_) => RoutingType::VolumeSplit,
+    };
+
+    Rule {
+        name: rule.name,
+        routing_type,
+        output: convert_output(rule.connector_selection),
+        statements: rule.statements.into_iter().map(convert_if_stmt).collect(),
+    }
+}
+
+fn convert_if_stmt(stmt: ast::IfStatement) -> IfStatement {
+    IfStatement {
+        condition: stmt.condition.into_iter().map(convert_comparison).collect(),
+        nested: stmt
+            .nested
+            .map(|v| v.into_iter().map(convert_if_stmt).collect()),
+    }
+}
+
+fn convert_comparison(c: ast::Comparison) -> Comparison {
+    Comparison {
+        lhs: c.lhs,
+        comparison: convert_comparison_type(c.comparison),
+        value: convert_value(c.value),
+        metadata: c.metadata,
+    }
+}
+
+fn convert_comparison_type(ct: ast::ComparisonType) -> ComparisonType {
+    match ct {
+        ast::ComparisonType::Equal => ComparisonType::Equal,
+        ast::ComparisonType::NotEqual => ComparisonType::NotEqual,
+        ast::ComparisonType::LessThan => ComparisonType::LessThan,
+        ast::ComparisonType::LessThanEqual => ComparisonType::LessThanEqual,
+        ast::ComparisonType::GreaterThan => ComparisonType::GreaterThan,
+        ast::ComparisonType::GreaterThanEqual => ComparisonType::GreaterThanEqual,
+    }
+}
+
+#[allow(clippy::unimplemented)]
+fn convert_value(v: ast::ValueType) -> ValueType {
+    use ast::ValueType::*;
+    match v {
+        Number(n) => ValueType::Number(n.get_amount_as_i64().try_into().unwrap_or_default()),
+        EnumVariant(e) => ValueType::EnumVariant(e),
+        MetadataVariant(m) => ValueType::MetadataVariant(MetadataValue {
+            key: m.key,
+            value: m.value,
+        }),
+        StrValue(s) => ValueType::StrValue(s),
+        _ => unimplemented!(), // GlobalRef(r) => ValueType::GlobalRef(r),
+    }
+}
+
+fn convert_output(sel: ConnectorSelection) -> Output {
+    match sel {
+        ConnectorSelection::Priority(choices) => {
+            Output::Priority(choices.into_iter().map(stringify_choice).collect())
+        }
+        ConnectorSelection::VolumeSplit(vs) => Output::VolumeSplit(
+            vs.into_iter()
+                .map(|v| VolumeSplit {
+                    split: v.split,
+                    output: stringify_choice(v.connector),
+                })
+                .collect(),
+        ),
+    }
+}
+
+fn stringify_choice(c: RoutableConnectorChoice) -> String {
+    c.connector.to_string()
+}
diff --git a/crates/router/src/core/routing.rs b/crates/router/src/core/routing.rs
index 25a1838e95e..6b86bb83f54 100644
--- a/crates/router/src/core/routing.rs
+++ b/crates/router/src/core/routing.rs
@@ -31,7 +31,10 @@ use super::payouts;
 use super::{
     errors::RouterResult,
     payments::{
-        routing::{self as payments_routing},
+        routing::{
+            utils::*,
+            {self as payments_routing},
+        },
         OperationSessionGetters,
     },
 };
@@ -118,6 +121,7 @@ impl RoutingAlgorithmUpdate {
             created_at: timestamp,
             modified_at: timestamp,
             algorithm_for: transaction_type,
+            decision_engine_routing_id: None,
         };
         Self(algo)
     }
@@ -153,14 +157,34 @@ pub async fn retrieve_merchant_routing_dictionary(
         )
         .await
         .to_not_found_response(errors::ApiErrorResponse::ResourceIdNotFound)?;
-    let routing_metadata =
-        super::utils::filter_objects_based_on_profile_id_list(profile_id_list, routing_metadata);
+    let routing_metadata = super::utils::filter_objects_based_on_profile_id_list(
+        profile_id_list.clone(),
+        routing_metadata,
+    );
 
     let result = routing_metadata
         .into_iter()
         .map(ForeignInto::foreign_into)
         .collect::<Vec<_>>();
 
+    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 {
+            let list_request = ListRountingAlgorithmsRequest {
+                created_by: profile_id.get_string_repr().to_string(),
+            };
+            list_de_euclid_routing_algorithms(&state, list_request)
+                .await
+                .map_err(|e| {
+                    router_env::logger::error!(decision_engine_error=?e, "decision_engine_euclid");
+                })
+                .ok() // Avoid throwing error if Decision Engine is not available or other errors
+                .map(|mut de_routing| de_result.append(&mut de_routing));
+        }
+        compare_and_log_result(de_result, result.clone(), "list_routing".to_string());
+    }
+
     metrics::ROUTING_MERCHANT_DICTIONARY_RETRIEVE_SUCCESS_RESPONSE.add(1, &[]);
     Ok(service_api::ApplicationResponse::Json(
         routing_types::RoutingKind::RoutingAlgorithm(result),
@@ -249,6 +273,10 @@ pub async fn create_routing_algorithm_under_profile(
     request: routing_types::RoutingConfigRequest,
     transaction_type: enums::TransactionType,
 ) -> RouterResponse<routing_types::RoutingDictionaryRecord> {
+    use api_models::routing::RoutingAlgorithm 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();
@@ -269,6 +297,7 @@ pub async fn create_routing_algorithm_under_profile(
 
     let algorithm = request
         .algorithm
+        .clone()
         .get_required_value("algorithm")
         .change_context(errors::ApiErrorResponse::MissingRequiredField {
             field_name: "algorithm",
@@ -306,6 +335,37 @@ pub async fn create_routing_algorithm_under_profile(
     )
     .await?;
 
+    let mut decision_engine_routing_id: Option<String> = None;
+
+    if let Some(EuclidAlgorithm::Advanced(program)) = request.algorithm.clone() {
+        let internal_program: Program = program.into();
+        let routing_rule = RoutingRule {
+            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(),
+            }),
+        };
+
+        decision_engine_routing_id = create_de_euclid_routing_algo(&state, &routing_rule)
+            .await
+            .map_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");
+                logger::debug!(decision_engine_request=?routing_rule, "decision_engine_euclid");
+            })
+            .ok();
+    }
+
+    if decision_engine_routing_id.is_some() {
+        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");
+    }
+
     let timestamp = common_utils::date_time::now();
     let algo = RoutingAlgorithm {
         algorithm_id: algorithm_id.clone(),
@@ -318,6 +378,7 @@ pub async fn create_routing_algorithm_under_profile(
         created_at: timestamp,
         modified_at: timestamp,
         algorithm_for: transaction_type.to_owned(),
+        decision_engine_routing_id,
     };
     let record = db
         .insert_routing_algorithm(algo)
@@ -536,7 +597,7 @@ pub async fn link_routing_config(
                 db,
                 key_manager_state,
                 merchant_context.get_merchant_key_store(),
-                business_profile,
+                business_profile.clone(),
                 dynamic_routing_ref,
             )
             .await?;
@@ -578,14 +639,37 @@ pub async fn link_routing_config(
                 db,
                 key_manager_state,
                 merchant_context.get_merchant_key_store(),
-                business_profile,
+                business_profile.clone(),
                 routing_ref,
                 transaction_type,
             )
             .await?;
         }
     };
-
+    if let Some(euclid_routing_id) = routing_algorithm.decision_engine_routing_id.clone() {
+        let routing_algo = ActivateRoutingConfigRequest {
+            created_by: business_profile.get_id().get_string_repr().to_string(),
+            routing_algorithm_id: euclid_routing_id,
+        };
+        let link_result = link_de_euclid_routing_algorithm(&state, routing_algo).await;
+        match link_result {
+            Ok(_) => {
+                router_env::logger::info!(
+                    routing_flow=?"link_routing_algorithm",
+                    is_equal=?true,
+                    "decision_engine_euclid"
+                );
+            }
+            Err(e) => {
+                router_env::logger::info!(
+                    routing_flow=?"link_routing_algorithm",
+                    is_equal=?false,
+                    error=?e,
+                    "decision_engine_euclid"
+                );
+            }
+        }
+    }
     metrics::ROUTING_LINK_CONFIG_SUCCESS_RESPONSE.add(1, &[]);
     Ok(service_api::ApplicationResponse::Json(
         routing_algorithm.foreign_into(),
@@ -1436,6 +1520,7 @@ pub async fn success_based_routing_update_configs(
         created_at: timestamp,
         modified_at: timestamp,
         algorithm_for: dynamic_routing_algo_to_update.algorithm_for,
+        decision_engine_routing_id: None,
     };
     let record = db
         .insert_routing_algorithm(algo)
@@ -1535,6 +1620,7 @@ pub async fn elimination_routing_update_configs(
         created_at: timestamp,
         modified_at: timestamp,
         algorithm_for: dynamic_routing_algo_to_update.algorithm_for,
+        decision_engine_routing_id: None,
     };
 
     let record = db
@@ -1680,6 +1766,7 @@ pub async fn contract_based_dynamic_routing_setup(
         created_at: timestamp,
         modified_at: timestamp,
         algorithm_for: common_enums::TransactionType::Payment,
+        decision_engine_routing_id: None,
     };
 
     // 1. if dynamic_routing_algo_ref already present, insert contract based algo and disable success based
@@ -1867,6 +1954,7 @@ pub async fn contract_based_routing_update_configs(
         created_at: timestamp,
         modified_at: timestamp,
         algorithm_for: dynamic_routing_algo_to_update.algorithm_for,
+        decision_engine_routing_id: None,
     };
     let record = db
         .insert_routing_algorithm(algo)
diff --git a/crates/router/src/core/routing/helpers.rs b/crates/router/src/core/routing/helpers.rs
index dc594180e70..39b601ed296 100644
--- a/crates/router/src/core/routing/helpers.rs
+++ b/crates/router/src/core/routing/helpers.rs
@@ -1761,6 +1761,7 @@ pub async fn default_specific_dynamic_routing_setup(
                 created_at: timestamp,
                 modified_at: timestamp,
                 algorithm_for: common_enums::TransactionType::Payment,
+                decision_engine_routing_id: None,
             }
         }
         routing_types::DynamicRoutingType::EliminationRouting => {
@@ -1777,6 +1778,7 @@ pub async fn default_specific_dynamic_routing_setup(
                 created_at: timestamp,
                 modified_at: timestamp,
                 algorithm_for: common_enums::TransactionType::Payment,
+                decision_engine_routing_id: None,
             }
         }
 
diff --git a/crates/router/src/core/routing/transformers.rs b/crates/router/src/core/routing/transformers.rs
index 85b6e5832a8..e9088ded2e7 100644
--- a/crates/router/src/core/routing/transformers.rs
+++ b/crates/router/src/core/routing/transformers.rs
@@ -32,6 +32,7 @@ impl ForeignFrom<RoutingProfileMetadata> for RoutingDictionaryRecord {
             created_at: value.created_at.assume_utc().unix_timestamp(),
             modified_at: value.modified_at.assume_utc().unix_timestamp(),
             algorithm_for: Some(value.algorithm_for),
+            decision_engine_routing_id: None,
         }
     }
 }
@@ -48,6 +49,7 @@ impl ForeignFrom<RoutingAlgorithm> for RoutingDictionaryRecord {
             created_at: value.created_at.assume_utc().unix_timestamp(),
             modified_at: value.modified_at.assume_utc().unix_timestamp(),
             algorithm_for: Some(value.algorithm_for),
+            decision_engine_routing_id: value.decision_engine_routing_id,
         }
     }
 }
diff --git a/migrations/2025-05-08-102850_add_de_euclid_id_in_routing_algorithm_table/down.sql b/migrations/2025-05-08-102850_add_de_euclid_id_in_routing_algorithm_table/down.sql
new file mode 100644
index 00000000000..e0b2631d03b
--- /dev/null
+++ b/migrations/2025-05-08-102850_add_de_euclid_id_in_routing_algorithm_table/down.sql
@@ -0,0 +1,3 @@
+-- This file should undo anything in `up.sql`
+ALTER TABLE routing_algorithm
+DROP COLUMN IF EXISTS decision_engine_routing_id;
diff --git a/migrations/2025-05-08-102850_add_de_euclid_id_in_routing_algorithm_table/up.sql b/migrations/2025-05-08-102850_add_de_euclid_id_in_routing_algorithm_table/up.sql
new file mode 100644
index 00000000000..b768b2a3070
--- /dev/null
+++ b/migrations/2025-05-08-102850_add_de_euclid_id_in_routing_algorithm_table/up.sql
@@ -0,0 +1,3 @@
+-- Your SQL goes here
+ALTER TABLE routing_algorithm
+ADD COLUMN decision_engine_routing_id VARCHAR(64);
diff --git a/v2_migrations/2025-01-13-081847_drop_v1_columns/up.sql b/v2_migrations/2025-01-13-081847_drop_v1_columns/up.sql
index 1140974d7c7..0e086500cc1 100644
--- a/v2_migrations/2025-01-13-081847_drop_v1_columns/up.sql
+++ b/v2_migrations/2025-01-13-081847_drop_v1_columns/up.sql
@@ -128,3 +128,7 @@ ALTER TABLE refund
     DROP COLUMN IF EXISTS internal_reference_id,
     DROP COLUMN IF EXISTS refund_id,
     DROP COLUMN IF EXISTS merchant_connector_id;
+
+-- Run below queries only when V1 is deprecated
+ALTER TABLE routing_algorithm
+    DROP COLUMN IF EXISTS decision_engine_routing_id;