Issue:
Bug: [FEATURE] add support for filtering in `payments_session`



### Feature Description

After supporting multiple connector accounts through `business_label` , `business_country` and `business_sub_label`, PaymentsSession flow ( creating session tokens - wallets ) should filter based on this.

### Possible Implementation

Add a similar filter which is in `list_payment_methods`

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

- [X] I checked and didn't find 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 -->

- [x] Refactoring

## Description
<!-- Describe your changes in detail -->
- This PR refactors the get connector function of `PaymentsSession` operation to use idiomatic rust. 
- Add business country and business label filtering.

<!-- 
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 its an obvious bug or documentation fix
that will have little conversation).
-->
 #1091 

## 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)?
-->
Tested manually. 
- Create payments
- Create session token
  - With `wallets` fields as empty list
  
<img width="1195" alt="Screenshot 2023-05-11 at 6 27 48 PM" src="https://github.com/juspay/hyperswitch/assets/48803246/2bae4c34-92d5-49ef-9aa2-80c31800052e">

  - With `wallets` field as `google_pay` only
<img width="1200" alt="Screenshot 2023-05-11 at 6 28 09 PM" src="https://github.com/juspay/hyperswitch/assets/48803246/639eb9b7-8ec9-471a-a240-23d74aa6e074">

- With `wallets` field as `apple_pay` only
<img width="1185" alt="Screenshot 2023-05-11 at 6 28 52 PM" src="https://github.com/juspay/hyperswitch/assets/48803246/c1638fe8-9acf-4e97-8640-0ec5d64988aa">

## 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 submitted code


Testing:

Tested manually. 
- Create payments
- Create session token
  - With `wallets` fields as empty list
  
<img width="1195" alt="Screenshot 2023-05-11 at 6 27 48 PM" src="https://github.com/juspay/hyperswitch/assets/48803246/2bae4c34-92d5-49ef-9aa2-80c31800052e">

  - With `wallets` field as `google_pay` only
<img width="1200" alt="Screenshot 2023-05-11 at 6 28 09 PM" src="https://github.com/juspay/hyperswitch/assets/48803246/639eb9b7-8ec9-471a-a240-23d74aa6e074">

- With `wallets` field as `apple_pay` only
<img width="1185" alt="Screenshot 2023-05-11 at 6 28 52 PM" src="https://github.com/juspay/hyperswitch/assets/48803246/c1638fe8-9acf-4e97-8640-0ec5d64988aa">

Patch:
diff --git a/crates/router/src/core/payment_methods/cards.rs b/crates/router/src/core/payment_methods/cards.rs
index b3c7c031c5e..8e12da732bb 100644
--- a/crates/router/src/core/payment_methods/cards.rs
+++ b/crates/router/src/core/payment_methods/cards.rs
@@ -838,7 +838,8 @@ pub async fn list_payment_methods(
         .to_not_found_response(errors::ApiErrorResponse::MerchantAccountNotFound)?;
 
     // filter out connectors based on the business country
-    let filtered_mcas = filter_mca_based_on_business_details(all_mcas, payment_intent.as_ref());
+    let filtered_mcas =
+        helpers::filter_mca_based_on_business_details(all_mcas, payment_intent.as_ref());
 
     logger::debug!(mca_before_filtering=?filtered_mcas);
 
@@ -1214,25 +1215,6 @@ async fn filter_payment_methods(
     Ok(())
 }
 
-fn filter_mca_based_on_business_details(
-    merchant_connector_accounts: Vec<
-        storage_models::merchant_connector_account::MerchantConnectorAccount,
-    >,
-    payment_intent: Option<&storage_models::payment_intent::PaymentIntent>,
-) -> Vec<storage_models::merchant_connector_account::MerchantConnectorAccount> {
-    if let Some(payment_intent) = payment_intent {
-        merchant_connector_accounts
-            .into_iter()
-            .filter(|mca| {
-                mca.business_country == payment_intent.business_country
-                    && mca.business_label == payment_intent.business_label
-            })
-            .collect::<Vec<_>>()
-    } else {
-        merchant_connector_accounts
-    }
-}
-
 fn filter_pm_based_on_config<'a>(
     config: &'a crate::configs::settings::ConnectorFilters,
     connector: &'a str,
diff --git a/crates/router/src/core/payments.rs b/crates/router/src/core/payments.rs
index 26ba2e1065f..ee344989818 100644
--- a/crates/router/src/core/payments.rs
+++ b/crates/router/src/core/payments.rs
@@ -536,7 +536,7 @@ where
 pub async fn call_multiple_connectors_service<F, Op, Req>(
     state: &AppState,
     merchant_account: &storage::MerchantAccount,
-    connectors: Vec<api::ConnectorData>,
+    connectors: Vec<api::SessionConnectorData>,
     _operation: &Op,
     mut payment_data: PaymentData<F>,
     customer: &Option<storage::Customer>,
@@ -558,15 +558,16 @@ where
     let call_connectors_start_time = Instant::now();
     let mut join_handlers = Vec::with_capacity(connectors.len());
 
-    for connector in connectors.iter() {
-        let connector_id = connector.connector.id();
+    for session_connector_data in connectors.iter() {
+        let connector_id = session_connector_data.connector.connector.id();
+
         let router_data = payment_data
             .construct_router_data(state, connector_id, merchant_account, customer)
             .await?;
 
         let res = router_data.decide_flows(
             state,
-            connector,
+            &session_connector_data.connector,
             customer,
             CallConnectorAction::Trigger,
             merchant_account,
@@ -577,8 +578,8 @@ where
 
     let result = join_all(join_handlers).await;
 
-    for (connector_res, connector) in result.into_iter().zip(connectors) {
-        let connector_name = connector.connector_name.to_string();
+    for (connector_res, session_connector) in result.into_iter().zip(connectors) {
+        let connector_name = session_connector.connector.connector_name.to_string();
         match connector_res {
             Ok(connector_response) => {
                 if let Ok(types::PaymentsResponseData::SessionResponse { session_token }) =
@@ -1077,18 +1078,13 @@ where
 {
     let connector_choice = operation
         .to_domain()?
-        .get_connector(merchant_account, state, req)
+        .get_connector(merchant_account, state, req, &payment_data.payment_intent)
         .await?;
 
     let connector = if should_call_connector(operation, payment_data) {
         Some(match connector_choice {
             api::ConnectorChoice::SessionMultiple(session_connectors) => {
-                api::ConnectorCallType::Multiple(
-                    session_connectors
-                        .into_iter()
-                        .map(|c| c.connector)
-                        .collect(),
-                )
+                api::ConnectorCallType::Multiple(session_connectors)
             }
 
             api::ConnectorChoice::StraightThrough(straight_through) => connector_selection(
diff --git a/crates/router/src/core/payments/helpers.rs b/crates/router/src/core/payments/helpers.rs
index 8741ac0ca59..5c834a874d9 100644
--- a/crates/router/src/core/payments/helpers.rs
+++ b/crates/router/src/core/payments/helpers.rs
@@ -43,6 +43,25 @@ use crate::{
     },
 };
 
+pub fn filter_mca_based_on_business_details(
+    merchant_connector_accounts: Vec<
+        storage_models::merchant_connector_account::MerchantConnectorAccount,
+    >,
+    payment_intent: Option<&storage_models::payment_intent::PaymentIntent>,
+) -> Vec<storage_models::merchant_connector_account::MerchantConnectorAccount> {
+    if let Some(payment_intent) = payment_intent {
+        merchant_connector_accounts
+            .into_iter()
+            .filter(|mca| {
+                mca.business_country == payment_intent.business_country
+                    && mca.business_label == payment_intent.business_label
+            })
+            .collect::<Vec<_>>()
+    } else {
+        merchant_connector_accounts
+    }
+}
+
 pub async fn get_address_for_payment_request(
     db: &dyn StorageInterface,
     req_address: Option<&api::Address>,
diff --git a/crates/router/src/core/payments/operations.rs b/crates/router/src/core/payments/operations.rs
index b53e3307fb8..773f219cf04 100644
--- a/crates/router/src/core/payments/operations.rs
+++ b/crates/router/src/core/payments/operations.rs
@@ -127,6 +127,7 @@ pub trait Domain<F: Clone, R>: Send + Sync {
         merchant_account: &storage::MerchantAccount,
         state: &AppState,
         request: &R,
+        payment_intent: &storage::payment_intent::PaymentIntent,
     ) -> CustomResult<api::ConnectorChoice, errors::ApiErrorResponse>;
 }
 
@@ -196,6 +197,7 @@ where
         _merchant_account: &storage::MerchantAccount,
         state: &AppState,
         _request: &api::PaymentsRetrieveRequest,
+        _payment_intent: &storage::payment_intent::PaymentIntent,
     ) -> CustomResult<api::ConnectorChoice, errors::ApiErrorResponse> {
         helpers::get_connector_default(state, None).await
     }
@@ -263,6 +265,7 @@ where
         _merchant_account: &storage::MerchantAccount,
         state: &AppState,
         _request: &api::PaymentsCaptureRequest,
+        _payment_intent: &storage::payment_intent::PaymentIntent,
     ) -> CustomResult<api::ConnectorChoice, errors::ApiErrorResponse> {
         helpers::get_connector_default(state, None).await
     }
@@ -318,6 +321,7 @@ where
         _merchant_account: &storage::MerchantAccount,
         state: &AppState,
         _request: &api::PaymentsCancelRequest,
+        _payment_intent: &storage::payment_intent::PaymentIntent,
     ) -> CustomResult<api::ConnectorChoice, errors::ApiErrorResponse> {
         helpers::get_connector_default(state, None).await
     }
diff --git a/crates/router/src/core/payments/operations/payment_complete_authorize.rs b/crates/router/src/core/payments/operations/payment_complete_authorize.rs
index ec6d7f7a38f..ef0b3c215cf 100644
--- a/crates/router/src/core/payments/operations/payment_complete_authorize.rs
+++ b/crates/router/src/core/payments/operations/payment_complete_authorize.rs
@@ -266,6 +266,7 @@ impl<F: Clone + Send> Domain<F, api::PaymentsRequest> for CompleteAuthorize {
         _merchant_account: &storage::MerchantAccount,
         state: &AppState,
         request: &api::PaymentsRequest,
+        _payment_intent: &storage::payment_intent::PaymentIntent,
     ) -> CustomResult<api::ConnectorChoice, errors::ApiErrorResponse> {
         // Use a new connector in the confirm call or use the same one which was passed when
         // creating the payment or if none is passed then use the routing algorithm
diff --git a/crates/router/src/core/payments/operations/payment_confirm.rs b/crates/router/src/core/payments/operations/payment_confirm.rs
index 4813bbf624f..852935dfe5b 100644
--- a/crates/router/src/core/payments/operations/payment_confirm.rs
+++ b/crates/router/src/core/payments/operations/payment_confirm.rs
@@ -285,6 +285,7 @@ impl<F: Clone + Send> Domain<F, api::PaymentsRequest> for PaymentConfirm {
         _merchant_account: &storage::MerchantAccount,
         state: &AppState,
         request: &api::PaymentsRequest,
+        _payment_intent: &storage::payment_intent::PaymentIntent,
     ) -> CustomResult<api::ConnectorChoice, errors::ApiErrorResponse> {
         // Use a new connector in the confirm call or use the same one which was passed when
         // creating the payment or if none is passed then use the routing algorithm
diff --git a/crates/router/src/core/payments/operations/payment_create.rs b/crates/router/src/core/payments/operations/payment_create.rs
index 0a6b3bd488b..9746768c19c 100644
--- a/crates/router/src/core/payments/operations/payment_create.rs
+++ b/crates/router/src/core/payments/operations/payment_create.rs
@@ -303,6 +303,7 @@ impl<F: Clone + Send> Domain<F, api::PaymentsRequest> for PaymentCreate {
         _merchant_account: &storage::MerchantAccount,
         state: &AppState,
         request: &api::PaymentsRequest,
+        _payment_intent: &storage::payment_intent::PaymentIntent,
     ) -> CustomResult<api::ConnectorChoice, errors::ApiErrorResponse> {
         helpers::get_connector_default(state, request.routing.clone()).await
     }
diff --git a/crates/router/src/core/payments/operations/payment_method_validate.rs b/crates/router/src/core/payments/operations/payment_method_validate.rs
index d8df71ab8e8..b328a00945c 100644
--- a/crates/router/src/core/payments/operations/payment_method_validate.rs
+++ b/crates/router/src/core/payments/operations/payment_method_validate.rs
@@ -278,6 +278,7 @@ where
         _merchant_account: &storage::MerchantAccount,
         state: &AppState,
         _request: &api::VerifyRequest,
+        _payment_intent: &storage::payment_intent::PaymentIntent,
     ) -> CustomResult<api::ConnectorChoice, errors::ApiErrorResponse> {
         helpers::get_connector_default(state, None).await
     }
diff --git a/crates/router/src/core/payments/operations/payment_session.rs b/crates/router/src/core/payments/operations/payment_session.rs
index 0206ee91bac..de8bed67d5a 100644
--- a/crates/router/src/core/payments/operations/payment_session.rs
+++ b/crates/router/src/core/payments/operations/payment_session.rs
@@ -281,16 +281,26 @@ where
         Ok((Box::new(self), None))
     }
 
+    /// Returns `Vec<SessionConnectorData>`
+    /// Steps carried out in this function
+    /// Get all the `merchant_connector_accounts` which are not disabled
+    /// Filter out connectors which have `invoke_sdk_client` enabled in `payment_method_types`
+    /// If session token is requested for certain wallets only, then return them, else
+    /// return all eligible connectors
+    ///
+    /// `GetToken` parameter specifies whether to get the session token from connector integration
+    /// or from separate implementation ( for googlepay - from metadata and applepay - from metadata and call connector)
     async fn get_connector<'a>(
         &'a self,
         merchant_account: &storage::MerchantAccount,
         state: &AppState,
         request: &api::PaymentsSessionRequest,
+        payment_intent: &storage::payment_intent::PaymentIntent,
     ) -> RouterResult<api::ConnectorChoice> {
         let connectors = &state.conf.connectors;
         let db = &state.store;
 
-        let connector_accounts = db
+        let all_connector_accounts = db
             .find_merchant_connector_account_by_merchant_id_and_disabled_list(
                 &merchant_account.merchant_id,
                 false,
@@ -299,84 +309,88 @@ where
             .change_context(errors::ApiErrorResponse::InternalServerError)
             .attach_printable("Database error when querying for merchant connector accounts")?;
 
+        let filtered_connector_accounts = helpers::filter_mca_based_on_business_details(
+            all_connector_accounts,
+            Some(payment_intent),
+        );
+
+        let requested_payment_method_types = request.wallets.clone();
         let mut connector_and_supporting_payment_method_type = Vec::new();
 
-        for connector_account in connector_accounts {
-            let payment_methods = connector_account
-                .payment_methods_enabled
-                .unwrap_or_default();
-            for payment_method in payment_methods {
-                let parsed_payment_method_result: Result<
-                    PaymentMethodsEnabled,
-                    error_stack::Report<errors::ParsingError>,
-                > = payment_method.clone().parse_value("payment_method");
-
-                match parsed_payment_method_result {
-                    Ok(parsed_payment_method) => {
-                        let payment_method_types = parsed_payment_method
+        filtered_connector_accounts
+            .into_iter()
+            .for_each(|connector_account| {
+                let res = connector_account
+                    .payment_methods_enabled
+                    .unwrap_or_default()
+                    .into_iter()
+                    .map(|payment_methods_enabled| {
+                        payment_methods_enabled
+                            .parse_value::<PaymentMethodsEnabled>("payment_methods_enabled")
+                    })
+                    .filter_map(|parsed_payment_method_result| {
+                        let error = parsed_payment_method_result.as_ref().err();
+                        logger::error!(session_token_parsing_error=?error);
+                        parsed_payment_method_result.ok()
+                    })
+                    .flat_map(|parsed_payment_methods_enabled| {
+                        parsed_payment_methods_enabled
                             .payment_method_types
-                            .unwrap_or_default();
-                        for payment_method_type in payment_method_types {
-                            if matches!(
-                                payment_method_type.payment_experience,
-                                Some(api_models::enums::PaymentExperience::InvokeSdkClient)
-                            ) {
-                                let connector_and_wallet = (
+                            .unwrap_or_default()
+                            .into_iter()
+                            .filter(|payment_method_type| {
+                                let is_invoke_sdk_client = matches!(
+                                    payment_method_type.payment_experience,
+                                    Some(api_models::enums::PaymentExperience::InvokeSdkClient)
+                                );
+
+                                // If session token is requested for the payment method type,
+                                // filter it out
+                                // if not, then create all sessions tokens
+                                let is_sent_in_request = requested_payment_method_types
+                                    .contains(&payment_method_type.payment_method_type)
+                                    || requested_payment_method_types.is_empty();
+
+                                is_invoke_sdk_client && is_sent_in_request
+                            })
+                            .map(|payment_method_type| {
+                                (
                                     connector_account.connector_name.to_owned(),
                                     payment_method_type.payment_method_type,
-                                );
-                                connector_and_supporting_payment_method_type
-                                    .push(connector_and_wallet);
-                            }
-                        }
-                    }
-                    Err(parsing_error) => {
-                        logger::debug!(session_token_parsing_error=?parsing_error);
-                    }
+                                    connector_account.business_sub_label.to_owned(),
+                                )
+                            })
+                            .collect::<Vec<_>>()
+                    })
+                    .collect::<Vec<_>>();
+                connector_and_supporting_payment_method_type.extend(res);
+            });
+
+        let mut session_connector_data =
+            Vec::with_capacity(connector_and_supporting_payment_method_type.len());
+
+        for (connector, payment_method_type, business_sub_label) in
+            connector_and_supporting_payment_method_type
+        {
+            match api::ConnectorData::get_connector_by_name(
+                connectors,
+                &connector,
+                api::GetToken::from(payment_method_type),
+            ) {
+                Ok(connector_data) => session_connector_data.push(api::SessionConnectorData {
+                    payment_method_type,
+                    connector: connector_data,
+                    business_sub_label,
+                }),
+                Err(error) => {
+                    logger::error!(session_token_error=?error)
                 }
             }
         }
 
-        let requested_payment_method_types = request.wallets.clone();
-
-        let connectors_data = if !requested_payment_method_types.is_empty() {
-            let mut connectors_data = Vec::new();
-            for payment_method_type in requested_payment_method_types {
-                for connector_and_payment_method_type in
-                    &connector_and_supporting_payment_method_type
-                {
-                    if connector_and_payment_method_type.1 == payment_method_type {
-                        let connector_details = api::ConnectorData::get_connector_by_name(
-                            connectors,
-                            connector_and_payment_method_type.0.as_str(),
-                            api::GetToken::from(connector_and_payment_method_type.1),
-                        )?;
-                        connectors_data.push(api::SessionConnectorData {
-                            payment_method_type,
-                            connector: connector_details,
-                        });
-                    }
-                }
-            }
-            connectors_data
-        } else {
-            let mut connectors_data = Vec::new();
-
-            for connector_and_payment_method_type in connector_and_supporting_payment_method_type {
-                let connector_details = api::ConnectorData::get_connector_by_name(
-                    connectors,
-                    connector_and_payment_method_type.0.as_str(),
-                    api::GetToken::from(connector_and_payment_method_type.1),
-                )?;
-                connectors_data.push(api::SessionConnectorData {
-                    payment_method_type: connector_and_payment_method_type.1,
-                    connector: connector_details,
-                });
-            }
-            connectors_data
-        };
-
-        Ok(api::ConnectorChoice::SessionMultiple(connectors_data))
+        Ok(api::ConnectorChoice::SessionMultiple(
+            session_connector_data,
+        ))
     }
 }
 
diff --git a/crates/router/src/core/payments/operations/payment_start.rs b/crates/router/src/core/payments/operations/payment_start.rs
index be798bc6004..3149bfea9e7 100644
--- a/crates/router/src/core/payments/operations/payment_start.rs
+++ b/crates/router/src/core/payments/operations/payment_start.rs
@@ -249,6 +249,7 @@ where
         _merchant_account: &storage::MerchantAccount,
         state: &AppState,
         _request: &api::PaymentsStartRequest,
+        _payment_intent: &storage::payment_intent::PaymentIntent,
     ) -> CustomResult<api::ConnectorChoice, errors::ApiErrorResponse> {
         helpers::get_connector_default(state, None).await
     }
diff --git a/crates/router/src/core/payments/operations/payment_status.rs b/crates/router/src/core/payments/operations/payment_status.rs
index 22da256adf4..acc38bc82cc 100644
--- a/crates/router/src/core/payments/operations/payment_status.rs
+++ b/crates/router/src/core/payments/operations/payment_status.rs
@@ -102,6 +102,7 @@ impl<F: Clone + Send> Domain<F, api::PaymentsRequest> for PaymentStatus {
         _merchant_account: &storage::MerchantAccount,
         state: &AppState,
         request: &api::PaymentsRequest,
+        _payment_intent: &storage::payment_intent::PaymentIntent,
     ) -> CustomResult<api::ConnectorChoice, errors::ApiErrorResponse> {
         helpers::get_connector_default(state, request.routing.clone()).await
     }
diff --git a/crates/router/src/core/payments/operations/payment_update.rs b/crates/router/src/core/payments/operations/payment_update.rs
index 4f0be30e585..6f41bf646d4 100644
--- a/crates/router/src/core/payments/operations/payment_update.rs
+++ b/crates/router/src/core/payments/operations/payment_update.rs
@@ -359,6 +359,7 @@ impl<F: Clone + Send> Domain<F, api::PaymentsRequest> for PaymentUpdate {
         _merchant_account: &storage::MerchantAccount,
         state: &AppState,
         request: &api::PaymentsRequest,
+        _payment_intent: &storage::payment_intent::PaymentIntent,
     ) -> CustomResult<api::ConnectorChoice, errors::ApiErrorResponse> {
         helpers::get_connector_default(state, request.routing.clone()).await
     }
diff --git a/crates/router/src/types/api.rs b/crates/router/src/types/api.rs
index 4b4b2f803ed..8bcd3a056f3 100644
--- a/crates/router/src/types/api.rs
+++ b/crates/router/src/types/api.rs
@@ -154,6 +154,7 @@ pub struct ConnectorData {
 pub struct SessionConnectorData {
     pub payment_method_type: api_enums::PaymentMethodType,
     pub connector: ConnectorData,
+    pub business_sub_label: Option<String>,
 }
 
 pub enum ConnectorChoice {
@@ -164,7 +165,7 @@ pub enum ConnectorChoice {
 
 #[derive(Clone)]
 pub enum ConnectorCallType {
-    Multiple(Vec<ConnectorData>),
+    Multiple(Vec<SessionConnectorData>),
     Single(ConnectorData),
 }