Bug: refactor(dynamic_routing): add logic for creating merchant account in decision engine




diff --git a/crates/api_models/src/routing.rs b/crates/api_models/src/routing.rs
index 46cd37bc68a..a5448b563b1 100644
--- a/crates/api_models/src/routing.rs
+++ b/crates/api_models/src/routing.rs
@@ -608,6 +608,8 @@ pub struct DynamicRoutingAlgorithmRef {
     pub dynamic_routing_volume_split: Option<u8>,
     pub elimination_routing_algorithm: Option<EliminationRoutingAlgorithm>,
     pub contract_based_routing: Option<ContractRoutingAlgorithm>,
+    #[serde(default)]
+    pub is_merchant_created_in_decision_engine: bool,
 }
 
 pub trait DynamicRoutingAlgoAccessor {
@@ -717,6 +719,10 @@ impl DynamicRoutingAlgorithmRef {
         self.dynamic_routing_volume_split = volume
     }
 
+    pub fn update_merchant_creation_status_in_decision_engine(&mut self, is_created: bool) {
+        self.is_merchant_created_in_decision_engine = is_created;
+    }
+
     pub fn is_success_rate_routing_enabled(&self) -> bool {
         self.success_based_algorithm
             .as_ref()
diff --git a/crates/router/src/core/admin.rs b/crates/router/src/core/admin.rs
index 942a659dcc1..13c219e7eba 100644
--- a/crates/router/src/core/admin.rs
+++ b/crates/router/src/core/admin.rs
@@ -264,26 +264,6 @@ pub async fn create_merchant_account(
         .await
         .to_duplicate_response(errors::ApiErrorResponse::DuplicateMerchantAccount)?;
 
-    // Call to DE here
-    // Check if creation should be based on default profile
-    #[cfg(all(feature = "dynamic_routing", feature = "v1"))]
-    {
-        if state.conf.open_router.enabled {
-            merchant_account
-                .default_profile
-                .as_ref()
-                .async_map(|profile_id| {
-                    routing::helpers::create_decision_engine_merchant(&state, profile_id)
-                })
-                .await
-                .transpose()
-                .map_err(|err| {
-                    crate::logger::error!("Failed to create merchant in Decision Engine {err:?}");
-                })
-                .ok();
-        }
-    }
-
     let merchant_context = domain::MerchantContext::NormalMerchant(Box::new(domain::Context(
         merchant_account.clone(),
         key_store.clone(),
@@ -3904,6 +3884,22 @@ impl ProfileCreateBridge for api::ProfileCreate {
             .map(CardTestingGuardConfig::foreign_from)
             .or(Some(CardTestingGuardConfig::default()));
 
+        let mut dynamic_routing_algorithm_ref =
+            routing_types::DynamicRoutingAlgorithmRef::default();
+
+        if self.is_debit_routing_enabled == Some(true) {
+            routing::helpers::create_merchant_in_decision_engine_if_not_exists(
+                state,
+                &profile_id,
+                &mut dynamic_routing_algorithm_ref,
+            )
+            .await;
+        }
+
+        let dynamic_routing_algorithm = serde_json::to_value(dynamic_routing_algorithm_ref)
+            .change_context(errors::ApiErrorResponse::InternalServerError)
+            .attach_printable("error serializing dynamic_routing_algorithm_ref to JSON Value")?;
+
         Ok(domain::Profile::from(domain::ProfileSetter {
             profile_id,
             merchant_id: merchant_context.get_merchant_account().get_id().clone(),
@@ -3981,7 +3977,7 @@ impl ProfileCreateBridge for api::ProfileCreate {
                 .always_collect_billing_details_from_wallet_connector,
             always_collect_shipping_details_from_wallet_connector: self
                 .always_collect_shipping_details_from_wallet_connector,
-            dynamic_routing_algorithm: None,
+            dynamic_routing_algorithm: Some(dynamic_routing_algorithm),
             is_network_tokenization_enabled: self.is_network_tokenization_enabled,
             is_auto_retries_enabled: self.is_auto_retries_enabled.unwrap_or_default(),
             max_auto_retries_enabled: self.max_auto_retries_enabled.map(i16::from),
@@ -4414,6 +4410,37 @@ impl ProfileUpdateBridge for api::ProfileUpdate {
             }
         };
 
+        let dynamic_routing_algo_ref = if self.is_debit_routing_enabled == Some(true) {
+            let mut dynamic_routing_algo_ref: routing_types::DynamicRoutingAlgorithmRef =
+                business_profile
+                    .dynamic_routing_algorithm
+                    .clone()
+                    .map(|val| val.parse_value("DynamicRoutingAlgorithmRef"))
+                    .transpose()
+                    .change_context(errors::ApiErrorResponse::InternalServerError)
+                    .attach_printable(
+                        "unable to deserialize dynamic routing algorithm ref from business profile",
+                    )?
+                    .unwrap_or_default();
+
+            routing::helpers::create_merchant_in_decision_engine_if_not_exists(
+                state,
+                business_profile.get_id(),
+                &mut dynamic_routing_algo_ref,
+            )
+            .await;
+
+            let dynamic_routing_algo_ref_value = serde_json::to_value(dynamic_routing_algo_ref)
+                .change_context(errors::ApiErrorResponse::InternalServerError)
+                .attach_printable(
+                    "error serializing dynamic_routing_algorithm_ref to JSON Value",
+                )?;
+
+            Some(dynamic_routing_algo_ref_value)
+        } else {
+            self.dynamic_routing_algorithm
+        };
+
         Ok(domain::ProfileUpdate::Update(Box::new(
             domain::ProfileGeneralUpdate {
                 profile_name: self.profile_name,
@@ -4451,7 +4478,7 @@ impl ProfileUpdateBridge for api::ProfileUpdate {
                     .always_collect_shipping_details_from_wallet_connector,
                 tax_connector_id: self.tax_connector_id,
                 is_tax_connector_enabled: self.is_tax_connector_enabled,
-                dynamic_routing_algorithm: self.dynamic_routing_algorithm,
+                dynamic_routing_algorithm: dynamic_routing_algo_ref,
                 is_network_tokenization_enabled: self.is_network_tokenization_enabled,
                 is_auto_retries_enabled: self.is_auto_retries_enabled,
                 max_auto_retries_enabled: self.max_auto_retries_enabled.map(i16::from),
diff --git a/crates/router/src/core/routing.rs b/crates/router/src/core/routing.rs
index 702763ccafb..3a30559336e 100644
--- a/crates/router/src/core/routing.rs
+++ b/crates/router/src/core/routing.rs
@@ -578,6 +578,7 @@ pub async fn link_routing_config(
                             business_profile.get_id(),
                             routing_algorithm.algorithm_data.clone(),
                             routing_types::DynamicRoutingType::SuccessRateBasedRouting,
+                            &mut dynamic_routing_ref,
                         )
                         .await
                         .change_context(errors::ApiErrorResponse::InternalServerError)
@@ -608,6 +609,7 @@ pub async fn link_routing_config(
                             business_profile.get_id(),
                             routing_algorithm.algorithm_data.clone(),
                             routing_types::DynamicRoutingType::EliminationRouting,
+                            &mut dynamic_routing_ref,
                         )
                         .await
                         .change_context(errors::ApiErrorResponse::InternalServerError)
@@ -1929,6 +1931,9 @@ pub async fn contract_based_dynamic_routing_setup(
             elimination_routing_algorithm: None,
             dynamic_routing_volume_split: None,
             contract_based_routing: Some(contract_algo),
+            is_merchant_created_in_decision_engine: dynamic_routing_algo_ref
+                .as_ref()
+                .is_some_and(|algo| algo.is_merchant_created_in_decision_engine),
         }
     };
 
diff --git a/crates/router/src/core/routing/helpers.rs b/crates/router/src/core/routing/helpers.rs
index 4798cce9b02..c4b56e1a8cb 100644
--- a/crates/router/src/core/routing/helpers.rs
+++ b/crates/router/src/core/routing/helpers.rs
@@ -30,9 +30,9 @@ use external_services::grpc_client::dynamic_routing::{
 use hyperswitch_domain_models::api::ApplicationResponse;
 #[cfg(all(feature = "v1", feature = "dynamic_routing"))]
 use hyperswitch_interfaces::events::routing_api_logs as routing_events;
-#[cfg(all(feature = "dynamic_routing", feature = "v1"))]
+#[cfg(feature = "v1")]
 use router_env::logger;
-#[cfg(all(feature = "dynamic_routing", feature = "v1"))]
+#[cfg(feature = "v1")]
 use router_env::{instrument, tracing};
 use rustc_hash::FxHashSet;
 use storage_impl::redis::cache;
@@ -52,15 +52,15 @@ use crate::{
     types::{domain, storage},
     utils::StringExt,
 };
+#[cfg(feature = "v1")]
+use crate::{
+    core::payments::routing::utils::{self as routing_utils, DecisionEngineApiHandler},
+    services,
+};
 #[cfg(all(feature = "dynamic_routing", feature = "v1"))]
 use crate::{
-    core::{
-        metrics as core_metrics,
-        payments::routing::utils::{self as routing_utils, DecisionEngineApiHandler},
-        routing,
-    },
+    core::{metrics as core_metrics, routing},
     routes::app::SessionStateInfo,
-    services,
     types::transformers::ForeignInto,
 };
 pub const SUCCESS_BASED_DYNAMIC_ROUTING_ALGORITHM: &str =
@@ -1723,7 +1723,7 @@ pub async fn disable_dynamic_routing_algorithm(
     let db = state.store.as_ref();
     let key_manager_state = &state.into();
     let profile_id = business_profile.get_id().clone();
-    let (algorithm_id, dynamic_routing_algorithm, cache_entries_to_redact) =
+    let (algorithm_id, mut dynamic_routing_algorithm, cache_entries_to_redact) =
         match dynamic_routing_type {
             routing_types::DynamicRoutingType::SuccessRateBasedRouting => {
                 let Some(algorithm_ref) = dynamic_routing_algo_ref.success_based_algorithm else {
@@ -1760,6 +1760,8 @@ pub async fn disable_dynamic_routing_algorithm(
                         contract_based_routing: dynamic_routing_algo_ref.contract_based_routing,
                         dynamic_routing_volume_split: dynamic_routing_algo_ref
                             .dynamic_routing_volume_split,
+                        is_merchant_created_in_decision_engine: dynamic_routing_algo_ref
+                            .is_merchant_created_in_decision_engine,
                     },
                     cache_entries_to_redact,
                 )
@@ -1800,6 +1802,8 @@ pub async fn disable_dynamic_routing_algorithm(
                             },
                         ),
                         contract_based_routing: dynamic_routing_algo_ref.contract_based_routing,
+                        is_merchant_created_in_decision_engine: dynamic_routing_algo_ref
+                            .is_merchant_created_in_decision_engine,
                     },
                     cache_entries_to_redact,
                 )
@@ -1838,6 +1842,8 @@ pub async fn disable_dynamic_routing_algorithm(
                                 routing_types::DynamicAlgorithmWithTimestamp::new(None),
                             enabled_feature: routing_types::DynamicRoutingFeatures::None,
                         }),
+                        is_merchant_created_in_decision_engine: dynamic_routing_algo_ref
+                            .is_merchant_created_in_decision_engine,
                     },
                     cache_entries_to_redact,
                 )
@@ -1850,6 +1856,7 @@ pub async fn disable_dynamic_routing_algorithm(
             state,
             business_profile.get_id(),
             dynamic_routing_type,
+            &mut dynamic_routing_algorithm,
         )
         .await
         .change_context(errors::ApiErrorResponse::InternalServerError)
@@ -2088,6 +2095,7 @@ pub async fn default_specific_dynamic_routing_setup(
             state,
             business_profile.get_id(),
             dynamic_routing_type,
+            &mut dynamic_routing_algo_ref,
         )
         .await
         .change_context(errors::ApiErrorResponse::InternalServerError)
@@ -2203,6 +2211,7 @@ pub async fn enable_decision_engine_dynamic_routing_setup(
     state: &SessionState,
     profile_id: &id_type::ProfileId,
     dynamic_routing_type: routing_types::DynamicRoutingType,
+    dynamic_routing_algo_ref: &mut routing_types::DynamicRoutingAlgorithmRef,
 ) -> RouterResult<()> {
     logger::debug!(
         "performing call with open_router for profile {}",
@@ -2248,6 +2257,10 @@ pub async fn enable_decision_engine_dynamic_routing_setup(
         }
     };
 
+    // Create merchant in Decision Engine if it is not already created
+    create_merchant_in_decision_engine_if_not_exists(state, profile_id, dynamic_routing_algo_ref)
+        .await;
+
     routing_utils::ConfigApiClient::send_decision_engine_request::<_, String>(
         state,
         services::Method::Post,
@@ -2269,6 +2282,7 @@ pub async fn update_decision_engine_dynamic_routing_setup(
     profile_id: &id_type::ProfileId,
     request: serde_json::Value,
     dynamic_routing_type: routing_types::DynamicRoutingType,
+    dynamic_routing_algo_ref: &mut routing_types::DynamicRoutingAlgorithmRef,
 ) -> RouterResult<()> {
     logger::debug!(
         "performing call with open_router for profile {}",
@@ -2320,6 +2334,10 @@ pub async fn update_decision_engine_dynamic_routing_setup(
         }
     };
 
+    // Create merchant in Decision Engine if it is not already created
+    create_merchant_in_decision_engine_if_not_exists(state, profile_id, dynamic_routing_algo_ref)
+        .await;
+
     routing_utils::ConfigApiClient::send_decision_engine_request::<_, String>(
         state,
         services::Method::Post,
@@ -2340,6 +2358,7 @@ pub async fn disable_decision_engine_dynamic_routing_setup(
     state: &SessionState,
     profile_id: &id_type::ProfileId,
     dynamic_routing_type: routing_types::DynamicRoutingType,
+    dynamic_routing_algo_ref: &mut routing_types::DynamicRoutingAlgorithmRef,
 ) -> RouterResult<()> {
     logger::debug!(
         "performing call with open_router for profile {}",
@@ -2364,6 +2383,10 @@ pub async fn disable_decision_engine_dynamic_routing_setup(
         },
     };
 
+    // Create merchant in Decision Engine if it is not already created
+    create_merchant_in_decision_engine_if_not_exists(state, profile_id, dynamic_routing_algo_ref)
+        .await;
+
     routing_utils::ConfigApiClient::send_decision_engine_request::<_, String>(
         state,
         services::Method::Post,
@@ -2378,7 +2401,32 @@ pub async fn disable_decision_engine_dynamic_routing_setup(
     Ok(())
 }
 
-#[cfg(all(feature = "dynamic_routing", feature = "v1"))]
+#[cfg(feature = "v1")]
+#[instrument(skip_all)]
+pub async fn create_merchant_in_decision_engine_if_not_exists(
+    state: &SessionState,
+    profile_id: &id_type::ProfileId,
+    dynamic_routing_algo_ref: &mut routing_types::DynamicRoutingAlgorithmRef,
+) {
+    if !dynamic_routing_algo_ref.is_merchant_created_in_decision_engine {
+        logger::debug!(
+            "Creating merchant_account in decision engine for profile {}",
+            profile_id.get_string_repr()
+        );
+
+        create_decision_engine_merchant(state, profile_id)
+            .await
+            .map_err(|err| {
+                logger::warn!("Merchant creation error in decision_engine: {err:?}");
+            })
+            .ok();
+
+        // TODO: Update the status based on the status code or error message from the API call
+        dynamic_routing_algo_ref.update_merchant_creation_status_in_decision_engine(true);
+    }
+}
+
+#[cfg(feature = "v1")]
 #[instrument(skip_all)]
 pub async fn create_decision_engine_merchant(
     state: &SessionState,

## Type of Change
<!-- Put an `x` in the boxes that apply -->

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

## Description
<!-- Describe your changes in detail -->
Merchant account creation on decision-engine is pre-requisite for making success based routing calls to it. Since toggle SR endpoint on Hyperswitch is the place where rule creation API of DE is being called, we can check whether merchant_account exists in DE, if exists, toggle endpoint will go ahead with creating the rule. If not, it will first call DE to create a merchant_account and then create the rule. 
I have created the `is_merchant_created_in_decision_engine` flag inside the `dynamic_routing_algorithm` column which is of type JSON in profile table. I have added a `#[serde(default)]` attribute on top of `is_merchant_created_in_decision_engine`. Hence it shouldn't break for older profiles which doesn't have this field itself.

### Additional Changes

- [ ] This PR modifies the API contract
- [ ] This PR modifies the database schema
- [ ] This PR modifies application configuration/environment variables

<!--
Provide links to the files with corresponding changes.

Following are the paths where you can find config files:
1. `config`
2. `crates/router/src/configs`
3. `loadtest/config`
-->


## Motivation and Context
<!--
Why is this change required? What problem does it solve?
If it fixes an open issue, please link to the issue here.

If you don't have an issue, we'd recommend starting with one first so the PR
can focus on the implementation (unless it is an obvious bug or documentation fix
that will have little conversation).
-->


## How did you test it?
<!--
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 merchant_account and API key in Hyperswitch.
2. Toggle SR for the default profile created above.

```
curl --location --request POST 'http://localhost:8080/account/merchant_1748681084/business_profile/pro_3gHXWoq1Xh3AKi6tNZV1/dynamic_routing/success_based/toggle?enable=dynamic_connector_selection' \
--header 'Content-Type: application/json' \
--header 'Accept: application/json' \
--header 'api-key: xyz'
```

Check logs for toggle call which should have something like `Creating merchant_account in decision engine for profile <profile_id>`

3. Try hitting same API again by disabling the SR, it shouldn't have above merchant_account creation log anymore as it should be created once per profile.
4. Also hit toggle API for the profile which already has SR toggled. Even for it, the merchant_account should be created during first toggle call and not during further calls.

#### Create merchant in debit routing flow
-> Update business profile by setting is_debit_routing_enabled
```
curl --location 'http://localhost:8080/account/merchant_1748956040/business_profile/pro_6QFJX6m7xjQacyygMmBT' \
--header 'Content-Type: application/json' \
--header 'Accept: application/json' \
--header 'api-key: ' \
--data '{
    "is_debit_routing_enabled": true
}'
```
```
{
    "merchant_id": "merchant_1748956040",
    "profile_id": "pro_6QFJX6m7xjQacyygMmBT",
    "profile_name": "US_default",
    "return_url": "https://google.com/success",
    "enable_payment_response_hash": true,
    "payment_response_hash_key": "THd00F3c6WIXOXufmm6sFsUL0zRUFVL2llcMS1SG9rdco16PFty01OOkLcY8IbSS",
    "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": true,
    "merchant_business_country": null,
    "is_pre_network_tokenization_enabled": false
}
```
-> Db screenshot
![image](https://github.com/user-attachments/assets/24b05943-8d3c-4add-9931-6f6ce66af57a)
-> Db entry when business profile is created
![image](https://github.com/user-attachments/assets/dc6b4d26-549b-414c-9395-929c78303d07)

#### Test sr
<img width="1728" alt="image" src="https://github.com/user-attachments/assets/f7253d1c-ec55-45cd-93e5-ff0e4503fbee" />
<img width="1490" alt="image" src="https://github.com/user-attachments/assets/e1f01c8f-8f22-4612-83ed-01562bc1dc50" />

-> Update profile to enable debit routing
<img width="1480" alt="image" src="https://github.com/user-attachments/assets/adce71c4-179b-4990-b693-c75e8ecad080" />



## Checklist
<!-- Put an `x` in the boxes that apply -->

- [x] I formatted the code `cargo +nightly fmt --all`
- [x] I addressed lints thrown by `cargo clippy`
- [x] I reviewed the submitted code
- [ ] I added unit tests for my changes where possible
