Issue:
Bug: [REFACTOR] Refactor routes for creating and activating the routing configs for v2



### Feature Description

 Refactor routes for creating and activating the routing configs for v2

### Possible Implementation

 Refactor routes for creating and activating the routing configs for v2

### Have you spent some time checking if this feature request has been raised before?

- [X] I checked and didn't find a similar issue

### Have you read the Contributing Guidelines?

- [X] I have read the [Contributing Guidelines](https://github.com/juspay/hyperswitch/blob/main/docs/CONTRIBUTING.md)

### Are you willing to submit a PR?

Yes, I am willing to submit a PR!

Pull request description:
## Type of Change
<!-- Put an `x` in the boxes that apply -->

- [ ] Bugfix
- [ ] New feature
- [ ] Enhancement
- [x] Refactoring
- [ ] Dependency updates
- [ ] Documentation
- [ ] CI/CD

## Description
This PR adds a `Routing_v2` flag and renames all the endpoints, and refactors the create and activate endpoints


### 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?
 
 - Create routing 
 ```
 curl --location 'http://localhost:8080/routing_algorithm' \
--header 'Content-Type: application/json' \
--header 'api-key: dev_1ZYFaypqi4RKpkjTFbHHkdgAqvUsG6TYJXoH53D59BQ1uFMrBIcxrF4f7FiJkOWw' \
--data '{
    "name": "newalgo",
    "description": "It is my ADVANCED config",
    "algorithm": {
        "type": "advanced",
        "data": {
            "defaultSelection": {
                "type": "priority",
                "data": [
                   {"connector": "adyen",
                    "merchant_connector_id":"mca_XcrOMvDBAiq6ln7ESbu0"}
                ]
            },
            "rules": [
                {
                    "name": "adyen",
                    "connectorSelection": {
                        "type": "priority",
                        "data": [
                       { "connector": "adyen",
                    "merchant_connector_id":"mca_XcrOMvDBAiq6ln7ESbu0"},
                    {
                        "connector": "cybersource",
                    "merchant_connector_id":"mca_8uCqbT9Bt9bmKUVuR3IA"
                    }

                        ]
                    },
                    "statements": [
                        {
                            "condition": [
                                {
                                    "lhs": "payment_amount",
                                    "comparison": "greater_than",
                                    "value": {
                                        "type": "number",
                                        "value": 100077
                                    },
                                    "metadata": {}
                                }
                            ],
                            "nested": null
                        }
                    ]
                }
            ],
            "metadata": {}
        }
    },
    "profile_id":"pro_r6qEFW0XLe7UnQoaaQ67"
}'
```
- Activate Routing Algorihtm
```
curl --location --request PATCH 'http://localhost:8080/v2/profiles/pro_r6qEFW0XLe7UnQoaaQ67/activate_routing_algorithm' \
--header 'Content-Type: application/json' \
--header 'api-key: dev_1ZYFaypqi4RKpkjTFbHHkdgAqvUsG6TYJXoH53D59BQ1uFMrBIcxrF4f7FiJkOWw' \
--data '"routing_merchant_1721842459_WWHyZVY3oW"
'
```
- The activated algorithm is then store in Business Profile Against that profile_id
<img width="1728" alt="Screenshot 2024-07-29 at 2 33 56 PM" src="https://github.com/user-attachments/assets/f66170fe-9bfd-479e-a8b2-5d4b42f75e66">



## 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


Testing:
 - Create routing 
 ```
 curl --location 'http://localhost:8080/routing_algorithm' \
--header 'Content-Type: application/json' \
--header 'api-key: dev_1ZYFaypqi4RKpkjTFbHHkdgAqvUsG6TYJXoH53D59BQ1uFMrBIcxrF4f7FiJkOWw' \
--data '{
    "name": "newalgo",
    "description": "It is my ADVANCED config",
    "algorithm": {
        "type": "advanced",
        "data": {
            "defaultSelection": {
                "type": "priority",
                "data": [
                   {"connector": "adyen",
                    "merchant_connector_id":"mca_XcrOMvDBAiq6ln7ESbu0"}
                ]
            },
            "rules": [
                {
                    "name": "adyen",
                    "connectorSelection": {
                        "type": "priority",
                        "data": [
                       { "connector": "adyen",
                    "merchant_connector_id":"mca_XcrOMvDBAiq6ln7ESbu0"},
                    {
                        "connector": "cybersource",
                    "merchant_connector_id":"mca_8uCqbT9Bt9bmKUVuR3IA"
                    }

                        ]
                    },
                    "statements": [
                        {
                            "condition": [
                                {
                                    "lhs": "payment_amount",
                                    "comparison": "greater_than",
                                    "value": {
                                        "type": "number",
                                        "value": 100077
                                    },
                                    "metadata": {}
                                }
                            ],
                            "nested": null
                        }
                    ]
                }
            ],
            "metadata": {}
        }
    },
    "profile_id":"pro_r6qEFW0XLe7UnQoaaQ67"
}'
```
- Activate Routing Algorihtm
```
curl --location --request PATCH 'http://localhost:8080/v2/profiles/pro_r6qEFW0XLe7UnQoaaQ67/activate_routing_algorithm' \
--header 'Content-Type: application/json' \
--header 'api-key: dev_1ZYFaypqi4RKpkjTFbHHkdgAqvUsG6TYJXoH53D59BQ1uFMrBIcxrF4f7FiJkOWw' \
--data '"routing_merchant_1721842459_WWHyZVY3oW"
'
```
- The activated algorithm is then store in Business Profile Against that profile_id
<img width="1728" alt="Screenshot 2024-07-29 at 2 33 56 PM" src="https://github.com/user-attachments/assets/f66170fe-9bfd-479e-a8b2-5d4b42f75e66">

Patch:
diff --git a/crates/api_models/Cargo.toml b/crates/api_models/Cargo.toml
index 84f8d2f01e9..b6b6f219bd8 100644
--- a/crates/api_models/Cargo.toml
+++ b/crates/api_models/Cargo.toml
@@ -17,8 +17,9 @@ frm = []
 olap = []
 openapi = ["common_enums/openapi", "olap", "recon", "dummy_connector", "olap"]
 recon = []
+v1 =[]
 v2 = []
-v1 = []
+routing_v2 = []
 merchant_connector_account_v2 = []
 customer_v2 = []
 merchant_account_v2 = []
diff --git a/crates/api_models/src/events/routing.rs b/crates/api_models/src/events/routing.rs
index 1f068678c59..4c2c5ca2c0b 100644
--- a/crates/api_models/src/events/routing.rs
+++ b/crates/api_models/src/events/routing.rs
@@ -3,7 +3,7 @@ use common_utils::events::{ApiEventMetric, ApiEventsType};
 use crate::routing::{
     LinkedRoutingConfigRetrieveResponse, MerchantRoutingAlgorithm, ProfileDefaultRoutingConfig,
     RoutingAlgorithmId, RoutingConfigRequest, RoutingDictionaryRecord, RoutingKind,
-    RoutingPayloadWrapper, RoutingRetrieveLinkQuery, RoutingRetrieveQuery,
+    RoutingLinkWrapper, RoutingPayloadWrapper, RoutingRetrieveLinkQuery, RoutingRetrieveQuery,
 };
 
 impl ApiEventMetric for RoutingKind {
@@ -64,3 +64,9 @@ impl ApiEventMetric for RoutingRetrieveLinkQuery {
         Some(ApiEventsType::Routing)
     }
 }
+
+impl ApiEventMetric for RoutingLinkWrapper {
+    fn get_api_event_type(&self) -> Option<ApiEventsType> {
+        Some(ApiEventsType::Routing)
+    }
+}
diff --git a/crates/api_models/src/routing.rs b/crates/api_models/src/routing.rs
index 7d1bb5bdf30..265d0451117 100644
--- a/crates/api_models/src/routing.rs
+++ b/crates/api_models/src/routing.rs
@@ -1,6 +1,6 @@
 use std::fmt::Debug;
 
-use common_utils::errors::ParsingError;
+use common_utils::{errors::ParsingError, ext_traits::ValueExt, pii};
 pub use euclid::{
     dssa::types::EuclidAnalysable,
     frontend::{
@@ -427,6 +427,14 @@ impl RoutingAlgorithmRef {
         self.surcharge_config_algo_id = Some(ids);
         self.timestamp = common_utils::date_time::now_unix_timestamp();
     }
+
+    pub fn parse_routing_algorithm(
+        value: Option<pii::SecretSerdeValue>,
+    ) -> Result<Option<Self>, error_stack::Report<ParsingError>> {
+        value
+            .map(|val| val.parse_value::<Self>("RoutingAlgorithmRef"))
+            .transpose()
+    }
 }
 
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize, ToSchema)]
@@ -459,6 +467,12 @@ pub enum RoutingKind {
 }
 
 #[repr(transparent)]
-#[derive(serde::Serialize, serde::Deserialize, Debug)]
+#[derive(serde::Serialize, serde::Deserialize, Debug, Clone)]
 #[serde(transparent)]
 pub struct RoutingAlgorithmId(pub String);
+
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+pub struct RoutingLinkWrapper {
+    pub profile_id: String,
+    pub algorithm_id: RoutingAlgorithmId,
+}
diff --git a/crates/diesel_models/Cargo.toml b/crates/diesel_models/Cargo.toml
index 9024ffe79d7..c596cb2e2e1 100644
--- a/crates/diesel_models/Cargo.toml
+++ b/crates/diesel_models/Cargo.toml
@@ -10,7 +10,7 @@ license.workspace = true
 [features]
 default = ["kv_store", "v1"] 
 kv_store = []
-v1 = []
+v1 =[]
 v2 = []
 customer_v2 = []
 merchant_account_v2 = []
@@ -27,6 +27,7 @@ strum = { version = "0.26.2", features = ["derive"] }
 thiserror = "1.0.58"
 time = { version = "0.3.35", features = ["serde", "serde-well-known", "std"] }
 
+
 # First party crates
 common_enums = { version = "0.1.0", path = "../common_enums" }
 common_utils = { version = "0.1.0", path = "../common_utils" }
diff --git a/crates/hyperswitch_domain_models/src/lib.rs b/crates/hyperswitch_domain_models/src/lib.rs
index c48f721b210..450fdac8d8c 100644
--- a/crates/hyperswitch_domain_models/src/lib.rs
+++ b/crates/hyperswitch_domain_models/src/lib.rs
@@ -1,8 +1,10 @@
 pub mod api;
+pub mod behaviour;
 pub mod customer;
 pub mod errors;
 pub mod mandates;
 pub mod merchant_account;
+pub mod merchant_key_store;
 pub mod payment_address;
 pub mod payment_method_data;
 pub mod payments;
@@ -13,9 +15,6 @@ pub mod router_data_v2;
 pub mod router_flow_types;
 pub mod router_request_types;
 pub mod router_response_types;
-
-pub mod behaviour;
-pub mod merchant_key_store;
 pub mod type_encryption;
 pub mod types;
 
diff --git a/crates/router/Cargo.toml b/crates/router/Cargo.toml
index 54cdb068ec5..2c40911f666 100644
--- a/crates/router/Cargo.toml
+++ b/crates/router/Cargo.toml
@@ -36,6 +36,7 @@ v1 = ["api_models/v1", "diesel_models/v1", "hyperswitch_domain_models/v1", "stor
 customer_v2 = ["api_models/customer_v2", "diesel_models/customer_v2", "hyperswitch_domain_models/customer_v2"]
 merchant_account_v2 = ["api_models/merchant_account_v2", "diesel_models/merchant_account_v2", "hyperswitch_domain_models/merchant_account_v2"]
 payment_v2 = ["api_models/payment_v2", "diesel_models/payment_v2", "hyperswitch_domain_models/payment_v2"]
+routing_v2 = ["api_models/routing_v2"]
 merchant_connector_account_v2 = ["api_models/merchant_connector_account_v2", "kgraph_utils/merchant_connector_account_v2"]
 
 [dependencies]
diff --git a/crates/router/src/core/routing.rs b/crates/router/src/core/routing.rs
index c8ba5d6a0df..2451156e874 100644
--- a/crates/router/src/core/routing.rs
+++ b/crates/router/src/core/routing.rs
@@ -7,15 +7,23 @@ use api_models::{
         self as routing_types, RoutingAlgorithmId, RoutingRetrieveLinkQuery, RoutingRetrieveQuery,
     },
 };
+#[cfg(all(feature = "v2", feature = "routing_v2"))]
+use diesel_models::routing_algorithm::RoutingAlgorithm;
+#[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "routing_v2")))]
 use diesel_models::routing_algorithm::RoutingAlgorithm;
 use error_stack::ResultExt;
+#[cfg(all(feature = "v2", feature = "routing_v2"))]
+use masking::Secret;
 use rustc_hash::FxHashSet;
 
 use super::payments;
 #[cfg(feature = "payouts")]
 use super::payouts;
+#[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "routing_v2")))]
+use crate::consts;
+#[cfg(all(feature = "v2", feature = "routing_v2"))]
+use crate::{consts, core::errors::RouterResult, db::StorageInterface};
 use crate::{
-    consts,
     core::{
         errors::{self, RouterResponse, StorageErrorExt},
         metrics, utils as core_utils,
@@ -28,7 +36,6 @@ use crate::{
     },
     utils::{self, OptionExt, ValueExt},
 };
-
 pub enum TransactionData<'a, F>
 where
     F: Clone,
@@ -38,6 +45,77 @@ where
     Payout(&'a payouts::PayoutData),
 }
 
+#[cfg(all(feature = "v2", feature = "routing_v2"))]
+struct RoutingAlgorithmUpdate(RoutingAlgorithm);
+
+#[cfg(all(feature = "v2", feature = "routing_v2"))]
+impl RoutingAlgorithmUpdate {
+    pub fn create_new_routing_algorithm(
+        algorithm: routing_types::RoutingAlgorithm,
+        merchant_id: &common_utils::id_type::MerchantId,
+        name: String,
+        description: String,
+        profile_id: String,
+        transaction_type: &enums::TransactionType,
+    ) -> Self {
+        let algorithm_id = common_utils::generate_id(
+            consts::ROUTING_CONFIG_ID_LENGTH,
+            &format!("routing_{}", merchant_id.get_string_repr()),
+        );
+        let timestamp = common_utils::date_time::now();
+        let algo = RoutingAlgorithm {
+            algorithm_id,
+            profile_id,
+            merchant_id: merchant_id.clone(),
+            name,
+            description: Some(description),
+            kind: algorithm.get_kind().foreign_into(),
+            algorithm_data: serde_json::json!(algorithm),
+            created_at: timestamp,
+            modified_at: timestamp,
+            algorithm_for: transaction_type.to_owned(),
+        };
+        Self(algo)
+    }
+    pub async fn fetch_routing_algo(
+        merchant_id: &common_utils::id_type::MerchantId,
+        algorithm_id: &str,
+        db: &dyn StorageInterface,
+    ) -> RouterResult<Self> {
+        let routing_algo = db
+            .find_routing_algorithm_by_algorithm_id_merchant_id(algorithm_id, merchant_id)
+            .await
+            .change_context(errors::ApiErrorResponse::ResourceIdNotFound)?;
+        Ok(Self(routing_algo))
+    }
+
+    pub fn update_routing_ref_with_algorithm_id(
+        &self,
+        transaction_type: &enums::TransactionType,
+        routing_ref: &mut routing_types::RoutingAlgorithmRef,
+    ) -> RouterResult<()> {
+        utils::when(self.0.algorithm_for != *transaction_type, || {
+            Err(errors::ApiErrorResponse::PreconditionFailed {
+                message: format!(
+                    "Cannot use {}'s routing algorithm for {} operation",
+                    self.0.algorithm_for, transaction_type
+                ),
+            })
+        })?;
+
+        utils::when(
+            routing_ref.algorithm_id == Some(self.0.algorithm_id.clone()),
+            || {
+                Err(errors::ApiErrorResponse::PreconditionFailed {
+                    message: "Algorithm is already active".to_string(),
+                })
+            },
+        )?;
+        routing_ref.update_algorithm_id(self.0.algorithm_id.clone());
+        Ok(())
+    }
+}
+
 pub async fn retrieve_merchant_routing_dictionary(
     state: SessionState,
     merchant_account: domain::MerchantAccount,
@@ -67,6 +145,7 @@ pub async fn retrieve_merchant_routing_dictionary(
     ))
 }
 
+#[cfg(all(feature = "v2", feature = "routing_v2"))]
 pub async fn create_routing_config(
     state: SessionState,
     merchant_account: domain::MerchantAccount,
@@ -75,8 +154,94 @@ pub async fn create_routing_config(
     transaction_type: &enums::TransactionType,
 ) -> RouterResponse<routing_types::RoutingDictionaryRecord> {
     metrics::ROUTING_CREATE_REQUEST_RECEIVED.add(&metrics::CONTEXT, 1, &[]);
-    let db = state.store.as_ref();
+    let db = &*state.store;
+    let name = request
+        .name
+        .get_required_value("name")
+        .change_context(errors::ApiErrorResponse::MissingRequiredField { field_name: "name" })
+        .attach_printable("Name of config not given")?;
+
+    let description = request
+        .description
+        .get_required_value("description")
+        .change_context(errors::ApiErrorResponse::MissingRequiredField {
+            field_name: "description",
+        })
+        .attach_printable("Description of config not given")?;
+
+    let algorithm = request
+        .algorithm
+        .get_required_value("algorithm")
+        .change_context(errors::ApiErrorResponse::MissingRequiredField {
+            field_name: "algorithm",
+        })
+        .attach_printable("Algorithm of config not given")?;
+
+    let business_profile = core_utils::validate_and_get_business_profile(
+        db,
+        request.profile_id.as_ref(),
+        merchant_account.get_id(),
+    )
+    .await?
+    .get_required_value("BusinessProfile")?;
+
+    let all_mcas = helpers::MerchantConnectorAccounts::get_all_mcas(
+        merchant_account.get_id(),
+        &key_store,
+        &state,
+    )
+    .await?;
+
+    let name_mca_id_set = helpers::ConnectNameAndMCAIdForProfile(
+        all_mcas.filter_by_profile(&business_profile.profile_id, |mca| {
+            (&mca.connector_name, &mca.merchant_connector_id)
+        }),
+    );
+
+    let name_set = helpers::ConnectNameForProfile(
+        all_mcas.filter_by_profile(&business_profile.profile_id, |mca| &mca.connector_name),
+    );
+
+    let algorithm_helper = helpers::RoutingAlgorithmHelpers {
+        name_mca_id_set,
+        name_set,
+        routing_algorithm: &algorithm,
+    };
+
+    algorithm_helper.validate_connectors_in_routing_config()?;
+
+    let algo = RoutingAlgorithmUpdate::create_new_routing_algorithm(
+        algorithm,
+        merchant_account.get_id(),
+        name,
+        description,
+        business_profile.profile_id,
+        transaction_type,
+    );
+
+    let record = state
+        .store
+        .as_ref()
+        .insert_routing_algorithm(algo.0)
+        .await
+        .to_not_found_response(errors::ApiErrorResponse::ResourceIdNotFound)?;
+
+    let new_record = record.foreign_into();
 
+    metrics::ROUTING_CREATE_SUCCESS_RESPONSE.add(&metrics::CONTEXT, 1, &[]);
+    Ok(service_api::ApplicationResponse::Json(new_record))
+}
+
+#[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "routing_v2")))]
+pub async fn create_routing_config(
+    state: SessionState,
+    merchant_account: domain::MerchantAccount,
+    key_store: domain::MerchantKeyStore,
+    request: routing_types::RoutingConfigRequest,
+    transaction_type: &enums::TransactionType,
+) -> RouterResponse<routing_types::RoutingDictionaryRecord> {
+    metrics::ROUTING_CREATE_REQUEST_RECEIVED.add(&metrics::CONTEXT, 1, &[]);
+    let db = state.store.as_ref();
     let name = request
         .name
         .get_required_value("name")
@@ -148,6 +313,56 @@ pub async fn create_routing_config(
     Ok(service_api::ApplicationResponse::Json(new_record))
 }
 
+#[cfg(all(feature = "v2", feature = "routing_v2"))]
+pub async fn link_routing_config(
+    state: SessionState,
+    merchant_account: domain::MerchantAccount,
+    profile_id: String,
+    algorithm_id: String,
+    transaction_type: &enums::TransactionType,
+) -> RouterResponse<routing_types::RoutingDictionaryRecord> {
+    metrics::ROUTING_LINK_CONFIG.add(&metrics::CONTEXT, 1, &[]);
+    let db = state.store.as_ref();
+
+    let routing_algorithm =
+        RoutingAlgorithmUpdate::fetch_routing_algo(merchant_account.get_id(), &algorithm_id, db)
+            .await?;
+    utils::when(routing_algorithm.0.profile_id != profile_id, || {
+        Err(errors::ApiErrorResponse::PreconditionFailed {
+            message: "Profile Id is invalid for the routing config".to_string(),
+        })
+    })?;
+    let business_profile = core_utils::validate_and_get_business_profile(
+        db,
+        Some(&profile_id),
+        merchant_account.get_id(),
+    )
+    .await?
+    .get_required_value("BusinessProfile")?;
+
+    let mut routing_ref = routing_types::RoutingAlgorithmRef::parse_routing_algorithm(
+        business_profile.routing_algorithm.clone().map(Secret::new),
+    )
+    .change_context(errors::ApiErrorResponse::InternalServerError)
+    .attach_printable("unable to deserialize routing algorithm ref from merchant account")?
+    .unwrap_or_default();
+
+    routing_algorithm.update_routing_ref_with_algorithm_id(transaction_type, &mut routing_ref)?;
+    // TODO move to business profile
+    helpers::update_business_profile_active_algorithm_ref(
+        db,
+        business_profile,
+        routing_ref,
+        transaction_type,
+    )
+    .await?;
+    metrics::ROUTING_LINK_CONFIG_SUCCESS_RESPONSE.add(&metrics::CONTEXT, 1, &[]);
+    Ok(service_api::ApplicationResponse::Json(
+        routing_algorithm.0.foreign_into(),
+    ))
+}
+
+#[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "routing_v2")))]
 pub async fn link_routing_config(
     state: SessionState,
     merchant_account: domain::MerchantAccount,
@@ -249,6 +464,7 @@ pub async fn retrieve_routing_config(
     metrics::ROUTING_RETRIEVE_CONFIG_SUCCESS_RESPONSE.add(&metrics::CONTEXT, 1, &[]);
     Ok(service_api::ApplicationResponse::Json(response))
 }
+
 pub async fn unlink_routing_config(
     state: SessionState,
     merchant_account: domain::MerchantAccount,
@@ -326,6 +542,7 @@ pub async fn unlink_routing_config(
     }
 }
 
+//feature update
 pub async fn update_default_routing_config(
     state: SessionState,
     merchant_account: domain::MerchantAccount,
diff --git a/crates/router/src/core/routing/helpers.rs b/crates/router/src/core/routing/helpers.rs
index 5978ef521a8..d4729d1e629 100644
--- a/crates/router/src/core/routing/helpers.rs
+++ b/crates/router/src/core/routing/helpers.rs
@@ -12,6 +12,8 @@ use error_stack::ResultExt;
 use rustc_hash::FxHashSet;
 use storage_impl::redis::cache;
 
+#[cfg(all(feature = "v2", feature = "routing_v2"))]
+use crate::types::domain::MerchantConnectorAccount;
 use crate::{
     core::errors::{self, RouterResult},
     db::StorageInterface,
@@ -20,54 +22,6 @@ use crate::{
     utils::StringExt,
 };
 
-/// provides the complete merchant routing dictionary that is basically a list of all the routing
-/// configs a merchant configured with an active_id field that specifies the current active routing
-/// config
-// pub async fn get_merchant_routing_dictionary(
-//     db: &dyn StorageInterface,
-//     merchant_id: &str,
-// ) -> RouterResult<routing_types::RoutingDictionary> {
-//     let key = get_routing_dictionary_key(merchant_id);
-//     let maybe_dict = db.find_config_by_key(&key).await;
-
-//     match maybe_dict {
-//         Ok(config) => config
-//             .config
-//             .parse_struct("RoutingDictionary")
-//             .change_context(errors::ApiErrorResponse::InternalServerError)
-//             .attach_printable("Merchant routing dictionary has invalid structure"),
-
-//         Err(e) if e.current_context().is_db_not_found() => {
-//             let new_dictionary = routing_types::RoutingDictionary {
-//                 merchant_id: merchant_id.to_owned(),
-//                 active_id: None,
-//                 records: Vec::new(),
-//             };
-
-//             let serialized = new_dictionary
-//                 .encode_to_string_of_json()
-//                 .change_context(errors::ApiErrorResponse::InternalServerError)
-//                 .attach_printable("Error serializing newly created merchant dictionary")?;
-
-//             let new_config = configs::ConfigNew {
-//                 key,
-//                 config: serialized,
-//             };
-
-//             db.insert_config(new_config)
-//                 .await
-//                 .change_context(errors::ApiErrorResponse::InternalServerError)
-//                 .attach_printable("Error inserting new routing dictionary for merchant")?;
-
-//             Ok(new_dictionary)
-//         }
-
-//         Err(e) => Err(e)
-//             .change_context(errors::ApiErrorResponse::InternalServerError)
-//             .attach_printable("Error fetching routing dictionary for merchant"),
-//     }
-// }
-
 /// Provides us with all the configured configs of the Merchant in the ascending time configured
 /// manner and chooses the first of them
 pub async fn get_merchant_default_config(
@@ -163,28 +117,6 @@ pub async fn update_merchant_routing_dictionary(
     Ok(())
 }
 
-pub async fn update_routing_algorithm(
-    db: &dyn StorageInterface,
-    algorithm_id: String,
-    algorithm: routing_types::RoutingAlgorithm,
-) -> RouterResult<()> {
-    let algorithm_str = algorithm
-        .encode_to_string_of_json()
-        .change_context(errors::ApiErrorResponse::InternalServerError)
-        .attach_printable("Unable to serialize routing algorithm to string")?;
-
-    let config_update = configs::ConfigUpdate::Update {
-        config: Some(algorithm_str),
-    };
-
-    db.update_config_by_key(&algorithm_id, config_update)
-        .await
-        .change_context(errors::ApiErrorResponse::InternalServerError)
-        .attach_printable("Error updating the routing algorithm in DB")?;
-
-    Ok(())
-}
-
 /// This will help make one of all configured algorithms to be in active state for a particular
 /// merchant
 pub async fn update_merchant_active_algorithm_ref(
@@ -238,7 +170,7 @@ pub async fn update_merchant_active_algorithm_ref(
 
     Ok(())
 }
-
+// TODO: Move it to business_profile
 pub async fn update_business_profile_active_algorithm_ref(
     db: &dyn StorageInterface,
     current_business_profile: BusinessProfile,
@@ -307,6 +239,157 @@ pub async fn update_business_profile_active_algorithm_ref(
     Ok(())
 }
 
+#[cfg(all(feature = "v2", feature = "routing_v2"))]
+#[derive(Clone, Debug)]
+pub struct RoutingAlgorithmHelpers<'h> {
+    pub name_mca_id_set: ConnectNameAndMCAIdForProfile<'h>,
+    pub name_set: ConnectNameForProfile<'h>,
+    pub routing_algorithm: &'h routing_types::RoutingAlgorithm,
+}
+
+#[derive(Clone, Debug)]
+pub struct ConnectNameAndMCAIdForProfile<'a>(pub FxHashSet<(&'a String, &'a String)>);
+#[derive(Clone, Debug)]
+pub struct ConnectNameForProfile<'a>(pub FxHashSet<&'a String>);
+
+#[cfg(all(feature = "v2", feature = "routing_v2"))]
+#[derive(Clone, Debug)]
+pub struct MerchantConnectorAccounts(pub Vec<MerchantConnectorAccount>);
+
+#[cfg(all(feature = "v2", feature = "routing_v2"))]
+impl MerchantConnectorAccounts {
+    pub async fn get_all_mcas(
+        merchant_id: &common_utils::id_type::MerchantId,
+        key_store: &domain::MerchantKeyStore,
+        state: &SessionState,
+    ) -> RouterResult<Self> {
+        let db = &*state.store;
+        let key_manager_state = &state.into();
+        Ok(Self(
+            db.find_merchant_connector_account_by_merchant_id_and_disabled_list(
+                key_manager_state,
+                merchant_id,
+                true,
+                key_store,
+            )
+            .await
+            .change_context(
+                errors::ApiErrorResponse::MerchantConnectorAccountNotFound {
+                    id: merchant_id.get_string_repr().to_owned(),
+                },
+            )?,
+        ))
+    }
+
+    fn filter_and_map<'a, T>(
+        &'a self,
+        filter: impl Fn(&'a MerchantConnectorAccount) -> bool,
+        func: impl Fn(&'a MerchantConnectorAccount) -> T,
+    ) -> FxHashSet<T>
+    where
+        T: std::hash::Hash + Eq,
+    {
+        self.0
+            .iter()
+            .filter(|mca| filter(mca))
+            .map(func)
+            .collect::<FxHashSet<_>>()
+    }
+
+    pub fn filter_by_profile<'a, T>(
+        &'a self,
+        profile_id: &'a str,
+        func: impl Fn(&'a MerchantConnectorAccount) -> T,
+    ) -> FxHashSet<T>
+    where
+        T: std::hash::Hash + Eq,
+    {
+        self.filter_and_map(|mca| mca.profile_id.as_deref() == Some(profile_id), func)
+    }
+}
+
+#[cfg(all(feature = "v2", feature = "routing_v2"))]
+impl<'h> RoutingAlgorithmHelpers<'h> {
+    fn connector_choice(
+        &self,
+        choice: &routing_types::RoutableConnectorChoice,
+    ) -> RouterResult<()> {
+        if let Some(ref mca_id) = choice.merchant_connector_id {
+            error_stack::ensure!(
+                    self.name_mca_id_set.0.contains(&(&choice.connector.to_string(), mca_id)),
+                    errors::ApiErrorResponse::InvalidRequestData {
+                        message: format!(
+                            "connector with name '{}' and merchant connector account id '{}' not found for the given profile",
+                            choice.connector,
+                            mca_id,
+                        )
+                    }
+                );
+        } else {
+            error_stack::ensure!(
+                self.name_set.0.contains(&choice.connector.to_string()),
+                errors::ApiErrorResponse::InvalidRequestData {
+                    message: format!(
+                        "connector with name '{}' not found for the given profile",
+                        choice.connector,
+                    )
+                }
+            );
+        };
+        Ok(())
+    }
+
+    pub fn validate_connectors_in_routing_config(&self) -> RouterResult<()> {
+        match self.routing_algorithm {
+            routing_types::RoutingAlgorithm::Single(choice) => {
+                self.connector_choice(choice)?;
+            }
+
+            routing_types::RoutingAlgorithm::Priority(list) => {
+                for choice in list {
+                    self.connector_choice(choice)?;
+                }
+            }
+
+            routing_types::RoutingAlgorithm::VolumeSplit(splits) => {
+                for split in splits {
+                    self.connector_choice(&split.connector)?;
+                }
+            }
+
+            routing_types::RoutingAlgorithm::Advanced(program) => {
+                let check_connector_selection =
+                    |selection: &routing_types::ConnectorSelection| -> RouterResult<()> {
+                        match selection {
+                            routing_types::ConnectorSelection::VolumeSplit(splits) => {
+                                for split in splits {
+                                    self.connector_choice(&split.connector)?;
+                                }
+                            }
+
+                            routing_types::ConnectorSelection::Priority(list) => {
+                                for choice in list {
+                                    self.connector_choice(choice)?;
+                                }
+                            }
+                        }
+
+                        Ok(())
+                    };
+
+                check_connector_selection(&program.default_selection)?;
+
+                for rule in &program.rules {
+                    check_connector_selection(&rule.connector_selection)?;
+                }
+            }
+        }
+
+        Ok(())
+    }
+}
+
+#[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "routing_v2")))]
 pub async fn validate_connectors_in_routing_config(
     state: &SessionState,
     key_store: &domain::MerchantKeyStore,
@@ -326,7 +409,6 @@ pub async fn validate_connectors_in_routing_config(
         .change_context(errors::ApiErrorResponse::MerchantConnectorAccountNotFound {
             id: merchant_id.get_string_repr().to_owned(),
         })?;
-
     let name_mca_id_set = all_mcas
         .iter()
         .filter(|mca| mca.profile_id.as_deref() == Some(profile_id))
@@ -339,7 +421,7 @@ pub async fn validate_connectors_in_routing_config(
         .map(|mca| &mca.connector_name)
         .collect::<FxHashSet<_>>();
 
-    let check_connector_choice = |choice: &routing_types::RoutableConnectorChoice| {
+    let connector_choice = |choice: &routing_types::RoutableConnectorChoice| {
         if let Some(ref mca_id) = choice.merchant_connector_id {
             error_stack::ensure!(
                 name_mca_id_set.contains(&(&choice.connector.to_string(), mca_id)),
@@ -368,18 +450,18 @@ pub async fn validate_connectors_in_routing_config(
 
     match routing_algorithm {
         routing_types::RoutingAlgorithm::Single(choice) => {
-            check_connector_choice(choice)?;
+            connector_choice(choice)?;
         }
 
         routing_types::RoutingAlgorithm::Priority(list) => {
             for choice in list {
-                check_connector_choice(choice)?;
+                connector_choice(choice)?;
             }
         }
 
         routing_types::RoutingAlgorithm::VolumeSplit(splits) => {
             for split in splits {
-                check_connector_choice(&split.connector)?;
+                connector_choice(&split.connector)?;
             }
         }
 
@@ -389,13 +471,13 @@ pub async fn validate_connectors_in_routing_config(
                     match selection {
                         routing_types::ConnectorSelection::VolumeSplit(splits) => {
                             for split in splits {
-                                check_connector_choice(&split.connector)?;
+                                connector_choice(&split.connector)?;
                             }
                         }
 
                         routing_types::ConnectorSelection::Priority(list) => {
                             for choice in list {
-                                check_connector_choice(choice)?;
+                                connector_choice(choice)?;
                             }
                         }
                     }
diff --git a/crates/router/src/routes/app.rs b/crates/router/src/routes/app.rs
index 08d960e6693..709cdae6355 100644
--- a/crates/router/src/routes/app.rs
+++ b/crates/router/src/routes/app.rs
@@ -1,7 +1,11 @@
 use std::{collections::HashMap, sync::Arc};
 
 use actix_web::{web, Scope};
-#[cfg(feature = "olap")]
+#[cfg(all(
+    feature = "olap",
+    any(feature = "v1", feature = "v2"),
+    not(feature = "routing_v2")
+))]
 use api_models::routing::RoutingRetrieveQuery;
 #[cfg(feature = "olap")]
 use common_enums::TransactionType;
@@ -39,7 +43,7 @@ use super::pm_auth;
 #[cfg(feature = "oltp")]
 use super::poll::retrieve_poll_status;
 #[cfg(feature = "olap")]
-use super::routing as cloud_routing;
+use super::routing;
 #[cfg(feature = "olap")]
 use super::verification::{apple_pay_merchant_registration, retrieve_apple_pay_verified_domains};
 #[cfg(feature = "olap")]
@@ -62,7 +66,6 @@ use crate::routes::fraud_check as frm_routes;
 use crate::routes::recon as recon_routes;
 pub use crate::{
     configs::settings,
-    core::routing,
     db::{CommonStorageInterface, GlobalStorageInterface, StorageImpl, StorageInterface},
     events::EventsHandler,
     routes::cards_info::card_iin_info,
@@ -567,7 +570,27 @@ impl Forex {
 #[cfg(feature = "olap")]
 pub struct Routing;
 
-#[cfg(feature = "olap")]
+#[cfg(all(feature = "olap", feature = "v2", feature = "routing_v2"))]
+impl Routing {
+    pub fn server(state: AppState) -> Scope {
+        web::scope("/v2/routing_algorithm")
+            .app_data(web::Data::new(state.clone()))
+            .service(
+                web::resource("").route(web::post().to(|state, req, payload| {
+                    routing::routing_create_config(state, req, payload, &TransactionType::Payment)
+                })),
+            )
+            .service(
+                web::resource("/{algorithm_id}")
+                    .route(web::get().to(routing::routing_retrieve_config)),
+            )
+    }
+}
+#[cfg(all(
+    feature = "olap",
+    any(feature = "v1", feature = "v2"),
+    not(feature = "routing_v2")
+))]
 impl Routing {
     pub fn server(state: AppState) -> Scope {
         #[allow(unused_mut)]
@@ -575,7 +598,7 @@ impl Routing {
             .app_data(web::Data::new(state.clone()))
             .service(
                 web::resource("/active").route(web::get().to(|state, req, query_params| {
-                    cloud_routing::routing_retrieve_linked_config(
+                    routing::routing_retrieve_linked_config(
                         state,
                         req,
                         query_params,
@@ -587,7 +610,7 @@ impl Routing {
                 web::resource("")
                     .route(
                         web::get().to(|state, req, path: web::Query<RoutingRetrieveQuery>| {
-                            cloud_routing::list_routing_configs(
+                            routing::list_routing_configs(
                                 state,
                                 req,
                                 path,
@@ -596,7 +619,7 @@ impl Routing {
                         }),
                     )
                     .route(web::post().to(|state, req, payload| {
-                        cloud_routing::routing_create_config(
+                        routing::routing_create_config(
                             state,
                             req,
                             payload,
@@ -607,14 +630,14 @@ impl Routing {
             .service(
                 web::resource("/default")
                     .route(web::get().to(|state, req| {
-                        cloud_routing::routing_retrieve_default_config(
+                        routing::routing_retrieve_default_config(
                             state,
                             req,
                             &TransactionType::Payment,
                         )
                     }))
                     .route(web::post().to(|state, req, payload| {
-                        cloud_routing::routing_update_default_config(
+                        routing::routing_update_default_config(
                             state,
                             req,
                             payload,
@@ -624,32 +647,25 @@ impl Routing {
             )
             .service(
                 web::resource("/deactivate").route(web::post().to(|state, req, payload| {
-                    cloud_routing::routing_unlink_config(
-                        state,
-                        req,
-                        payload,
-                        &TransactionType::Payment,
-                    )
+                    routing::routing_unlink_config(state, req, payload, &TransactionType::Payment)
                 })),
             )
             .service(
                 web::resource("/decision")
-                    .route(web::put().to(cloud_routing::upsert_decision_manager_config))
-                    .route(web::get().to(cloud_routing::retrieve_decision_manager_config))
-                    .route(web::delete().to(cloud_routing::delete_decision_manager_config)),
+                    .route(web::put().to(routing::upsert_decision_manager_config))
+                    .route(web::get().to(routing::retrieve_decision_manager_config))
+                    .route(web::delete().to(routing::delete_decision_manager_config)),
             )
             .service(
                 web::resource("/decision/surcharge")
-                    .route(web::put().to(cloud_routing::upsert_surcharge_decision_manager_config))
-                    .route(web::get().to(cloud_routing::retrieve_surcharge_decision_manager_config))
-                    .route(
-                        web::delete().to(cloud_routing::delete_surcharge_decision_manager_config),
-                    ),
+                    .route(web::put().to(routing::upsert_surcharge_decision_manager_config))
+                    .route(web::get().to(routing::retrieve_surcharge_decision_manager_config))
+                    .route(web::delete().to(routing::delete_surcharge_decision_manager_config)),
             )
             .service(
                 web::resource("/default/profile/{profile_id}").route(web::post().to(
                     |state, req, path, payload| {
-                        cloud_routing::routing_update_default_config_for_profile(
+                        routing::routing_update_default_config_for_profile(
                             state,
                             req,
                             path,
@@ -661,7 +677,7 @@ impl Routing {
             )
             .service(
                 web::resource("/default/profile").route(web::get().to(|state, req| {
-                    cloud_routing::routing_retrieve_default_config_for_profiles(
+                    routing::routing_retrieve_default_config_for_profiles(
                         state,
                         req,
                         &TransactionType::Payment,
@@ -676,7 +692,7 @@ impl Routing {
                     web::resource("/payouts")
                         .route(web::get().to(
                             |state, req, path: web::Query<RoutingRetrieveQuery>| {
-                                cloud_routing::list_routing_configs(
+                                routing::list_routing_configs(
                                     state,
                                     req,
                                     path,
@@ -685,7 +701,7 @@ impl Routing {
                             },
                         ))
                         .route(web::post().to(|state, req, payload| {
-                            cloud_routing::routing_create_config(
+                            routing::routing_create_config(
                                 state,
                                 req,
                                 payload,
@@ -695,7 +711,7 @@ impl Routing {
                 )
                 .service(web::resource("/payouts/active").route(web::get().to(
                     |state, req, query_params| {
-                        cloud_routing::routing_retrieve_linked_config(
+                        routing::routing_retrieve_linked_config(
                             state,
                             req,
                             query_params,
@@ -706,14 +722,14 @@ impl Routing {
                 .service(
                     web::resource("/payouts/default")
                         .route(web::get().to(|state, req| {
-                            cloud_routing::routing_retrieve_default_config(
+                            routing::routing_retrieve_default_config(
                                 state,
                                 req,
                                 &TransactionType::Payout,
                             )
                         }))
                         .route(web::post().to(|state, req, payload| {
-                            cloud_routing::routing_update_default_config(
+                            routing::routing_update_default_config(
                                 state,
                                 req,
                                 payload,
@@ -724,18 +740,13 @@ impl Routing {
                 .service(
                     web::resource("/payouts/{algorithm_id}/activate").route(web::post().to(
                         |state, req, path| {
-                            cloud_routing::routing_link_config(
-                                state,
-                                req,
-                                path,
-                                &TransactionType::Payout,
-                            )
+                            routing::routing_link_config(state, req, path, &TransactionType::Payout)
                         },
                     )),
                 )
                 .service(web::resource("/payouts/deactivate").route(web::post().to(
                     |state, req, payload| {
-                        cloud_routing::routing_unlink_config(
+                        routing::routing_unlink_config(
                             state,
                             req,
                             payload,
@@ -746,7 +757,7 @@ impl Routing {
                 .service(
                     web::resource("/payouts/default/profile/{profile_id}").route(web::post().to(
                         |state, req, path, payload| {
-                            cloud_routing::routing_update_default_config_for_profile(
+                            routing::routing_update_default_config_for_profile(
                                 state,
                                 req,
                                 path,
@@ -758,7 +769,7 @@ impl Routing {
                 )
                 .service(
                     web::resource("/payouts/default/profile").route(web::get().to(|state, req| {
-                        cloud_routing::routing_retrieve_default_config_for_profiles(
+                        routing::routing_retrieve_default_config_for_profiles(
                             state,
                             req,
                             &TransactionType::Payout,
@@ -770,17 +781,12 @@ impl Routing {
         route = route
             .service(
                 web::resource("/{algorithm_id}")
-                    .route(web::get().to(cloud_routing::routing_retrieve_config)),
+                    .route(web::get().to(routing::routing_retrieve_config)),
             )
             .service(
                 web::resource("/{algorithm_id}/activate").route(web::post().to(
                     |state, req, path| {
-                        cloud_routing::routing_link_config(
-                            state,
-                            req,
-                            path,
-                            &TransactionType::Payment,
-                        )
+                        routing::routing_link_config(state, req, path, &TransactionType::Payment)
                     },
                 )),
             );
@@ -1365,8 +1371,64 @@ impl PayoutLink {
 }
 
 pub struct BusinessProfile;
-
-#[cfg(feature = "olap")]
+#[cfg(all(feature = "olap", feature = "v2", feature = "routing_v2"))]
+impl BusinessProfile {
+    pub fn server(state: AppState) -> Scope {
+        web::scope("/v2/profiles")
+            .app_data(web::Data::new(state))
+            .service(
+                web::scope("/{profile_id}")
+                    .service(
+                        web::resource("/fallback_routing")
+                            .route(web::get().to(|state, req| {
+                                routing::routing_retrieve_default_config(
+                                    state,
+                                    req,
+                                    &TransactionType::Payment,
+                                )
+                            }))
+                            .route(web::post().to(|state, req, payload| {
+                                routing::routing_update_default_config(
+                                    state,
+                                    req,
+                                    payload,
+                                    &TransactionType::Payment,
+                                )
+                            })),
+                    )
+                    .service(
+                        web::resource("/activate_routing_algorithm").route(web::patch().to(
+                            |state, req, path, payload| {
+                                routing::routing_link_config(
+                                    state,
+                                    req,
+                                    path,
+                                    payload,
+                                    &TransactionType::Payment,
+                                )
+                            },
+                        )),
+                    )
+                    .service(
+                        web::resource("/deactivate_routing_algorithm").route(web::post().to(
+                            |state, req, path| {
+                                routing::routing_unlink_config(
+                                    state,
+                                    req,
+                                    path,
+                                    &TransactionType::Payment,
+                                )
+                            },
+                        )),
+                    ),
+            )
+    }
+}
+#[cfg(all(
+    feature = "olap",
+    any(feature = "v1", feature = "v2"),
+    not(feature = "routing_v2")
+))]
 impl BusinessProfile {
     pub fn server(state: AppState) -> Scope {
         web::scope("/account/{account_id}/business_profile")
diff --git a/crates/router/src/routes/routing.rs b/crates/router/src/routes/routing.rs
index 788acd4ba2d..f410a79028e 100644
--- a/crates/router/src/routes/routing.rs
+++ b/crates/router/src/routes/routing.rs
@@ -53,7 +53,11 @@ pub async fn routing_create_config(
     .await
 }
 
-#[cfg(feature = "olap")]
+#[cfg(all(
+    feature = "olap",
+    any(feature = "v1", feature = "v2"),
+    not(feature = "routing_v2")
+))]
 #[instrument(skip_all)]
 pub async fn routing_link_config(
     state: web::Data<AppState>,
@@ -88,6 +92,48 @@ pub async fn routing_link_config(
     .await
 }
 
+#[cfg(all(feature = "olap", feature = "v2", feature = "routing_v2"))]
+#[instrument(skip_all)]
+pub async fn routing_link_config(
+    state: web::Data<AppState>,
+    req: HttpRequest,
+    path: web::Path<String>,
+    json_payload: web::Json<routing_types::RoutingAlgorithmId>,
+    transaction_type: &enums::TransactionType,
+) -> impl Responder {
+    let flow = Flow::RoutingLinkConfig;
+    let wrapper = routing_types::RoutingLinkWrapper {
+        profile_id: path.into_inner(),
+        algorithm_id: json_payload.into_inner(),
+    };
+
+    Box::pin(oss_api::server_wrap(
+        flow,
+        state,
+        &req,
+        wrapper,
+        |state, auth: auth::AuthenticationData, wrapper, _| {
+            routing::link_routing_config(
+                state,
+                auth.merchant_account,
+                wrapper.profile_id,
+                wrapper.algorithm_id.0,
+                transaction_type,
+            )
+        },
+        #[cfg(not(feature = "release"))]
+        auth::auth_type(
+            &auth::ApiKeyAuth,
+            &auth::JWTAuth(Permission::RoutingWrite),
+            req.headers(),
+        ),
+        #[cfg(feature = "release")]
+        &auth::JWTAuth(Permission::RoutingWrite),
+        api_locking::LockAction::NotApplicable,
+    ))
+    .await
+}
+
 #[cfg(feature = "olap")]
 #[instrument(skip_all)]
 pub async fn routing_retrieve_config(
diff --git a/justfile b/justfile
index b755e3d7de0..228653a007a 100644
--- a/justfile
+++ b/justfile
@@ -55,7 +55,7 @@ check_v2 *FLAGS:
         jq -r '
             [ ( .workspace_members | sort ) as $package_ids # Store workspace crate package IDs in `package_ids` array
             | .packages[] | select( IN(.id; $package_ids[]) ) | .features | keys[] ] | unique # Select all unique features from all workspace crates
-            | del( .[] | select( any( . ; . == ("v1", "merchant_account_v2", "payment_v2") ) ) ) # Exclude some features from features list
+            | del( .[] | select( any( . ; . == ("v1", "merchant_account_v2", "payment_v2","routing_v2") ) ) ) # Exclude some features from features list
             | join(",") # Construct a comma-separated string of features for passing to `cargo`
     ')"