Bug: [BUG] incorrect mapping of connector customer 



### Bug Description

It bugs out when the merchant connector account credentials are changed, or when multiple connector accounts with the same name ( stripe, adyen ) are created in different countries or with different labels. When creating payment it will be using the older customer that is already created with older stipe account. 

This is because in the database, the mapping is `connector` -> `connector_customer_id`.
![Screenshot 2023-05-22 at 1 08 08 PM](https://github.com/juspay/hyperswitch/assets/48803246/39bed2c0-f31a-4ba4-85b0-ba2f8750cf41)


### Expected Behavior

If a new connector account is used in a different country, then new customer id should be created.

### Actual Behavior

same customer id is being used for connector account in different countries.

### Steps To Reproduce

Provide an unambiguous set of steps to reproduce this bug. Include code or configuration to reproduce, if relevant.
1. Create a stripe account in US region.
2. Make a payment using US as the business region.
3. Create a stripe account in AU region.
4. Make a payment using AU as the business region. 
5. The payment will fail with error `No such customer`


### Context For The Bug

_No response_

### Environment

Are you using hyperswitch hosted version? No
If yes, please provide the value of the `x-request-id` response header for helping us debug your issue.

If not (or if building/running locally), please provide the following details:
1. Operating System or Linux distribution: 
2. Rust version (output of `rustc --version`): ``
3. App version (output of `cargo r -- --version`): ``


### Have you spent some time to check if this bug 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!
diff --git a/crates/router/src/core/payments.rs b/crates/router/src/core/payments.rs
index 196d6b4c484..978f89a9f8a 100644
--- a/crates/router/src/core/payments.rs
+++ b/crates/router/src/core/payments.rs
@@ -122,7 +122,6 @@ where
 
     let updated_customer = call_create_connector_customer_if_required(
         state,
-        &payment_data.payment_attempt.connector.clone(),
         &customer,
         &merchant_account,
         &mut payment_data,
@@ -639,7 +638,6 @@ where
 
 pub async fn call_create_connector_customer_if_required<F, Req>(
     state: &AppState,
-    connector_name: &Option<String>,
     customer: &Option<domain::Customer>,
     merchant_account: &domain::MerchantAccount,
     payment_data: &mut PaymentData<F>,
@@ -658,17 +656,32 @@ where
     // To perform router related operation for PaymentResponse
     PaymentResponse: Operation<F, Req>,
 {
+    let connector_name = payment_data.payment_attempt.connector.clone();
+
     match connector_name {
         Some(connector_name) => {
             let connector = api::ConnectorData::get_connector_by_name(
                 &state.conf.connectors,
-                connector_name,
+                &connector_name,
                 api::GetToken::Connector,
             )?;
-            let (is_eligible, connector_customer_id, connector_customer_map) =
-                customers::should_call_connector_create_customer(state, &connector, customer)?;
 
-            if is_eligible {
+            let connector_label = helpers::get_connector_label(
+                payment_data.payment_intent.business_country,
+                &payment_data.payment_intent.business_label,
+                payment_data.payment_attempt.business_sub_label.as_ref(),
+                &connector_name,
+            );
+
+            let (should_call_connector, existing_connector_customer_id) =
+                customers::should_call_connector_create_customer(
+                    state,
+                    &connector,
+                    customer,
+                    &connector_label,
+                );
+
+            if should_call_connector {
                 // Create customer at connector and update the customer table to store this data
                 let router_data = payment_data
                     .construct_router_data(
@@ -679,15 +692,23 @@ where
                     )
                     .await?;
 
-                let (connector_customer, customer_update) = router_data
-                    .create_connector_customer(state, &connector, connector_customer_map)
+                let connector_customer_id = router_data
+                    .create_connector_customer(state, &connector)
                     .await?;
 
-                payment_data.connector_customer_id = connector_customer;
+                let customer_update = customers::update_connector_customer_in_customers(
+                    &connector_label,
+                    customer.as_ref(),
+                    &connector_customer_id,
+                )
+                .await;
+
+                payment_data.connector_customer_id = connector_customer_id;
                 Ok(customer_update)
             } else {
                 // Customer already created in previous calls use the same value, no need to update
-                payment_data.connector_customer_id = connector_customer_id;
+                payment_data.connector_customer_id =
+                    existing_connector_customer_id.map(ToOwned::to_owned);
                 Ok(None)
             }
         }
diff --git a/crates/router/src/core/payments/customers.rs b/crates/router/src/core/payments/customers.rs
index 9b23271f2dc..f443cc7302e 100644
--- a/crates/router/src/core/payments/customers.rs
+++ b/crates/router/src/core/payments/customers.rs
@@ -1,9 +1,8 @@
-use common_utils::ext_traits::ValueExt;
-use error_stack::{self, ResultExt};
+use router_env::{instrument, tracing};
 
 use crate::{
     core::{
-        errors::{self, ConnectorErrorExt, RouterResult},
+        errors::{ConnectorErrorExt, RouterResult},
         payments,
     },
     logger,
@@ -12,13 +11,13 @@ use crate::{
     types::{self, api, domain, storage},
 };
 
+#[instrument(skip_all)]
 pub async fn create_connector_customer<F: Clone, T: Clone>(
     state: &AppState,
     connector: &api::ConnectorData,
     router_data: &types::RouterData<F, T, types::PaymentsResponseData>,
     customer_request_data: types::ConnectorCustomerData,
-    connector_customer_map: Option<serde_json::Map<String, serde_json::Value>>,
-) -> RouterResult<(Option<String>, Option<storage::CustomerUpdate>)> {
+) -> RouterResult<Option<String>> {
     let connector_integration: services::BoxedConnectorIntegration<
         '_,
         api::CreateConnectorCustomer,
@@ -68,79 +67,75 @@ pub async fn create_connector_customer<F: Clone, T: Clone>(
             _ => None,
         },
         Err(err) => {
-            logger::debug!(payment_method_tokenization_error=?err);
+            logger::error!(create_connector_customer_error=?err);
             None
         }
     };
 
-    let update_customer = update_connector_customer_in_customers(
-        connector,
-        connector_customer_map,
-        &connector_customer_id,
-    )
-    .await?;
-    Ok((connector_customer_id, update_customer))
+    Ok(connector_customer_id)
+}
+
+pub fn get_connector_customer_details_if_present<'a>(
+    customer: &'a domain::Customer,
+    connector_name: &str,
+) -> Option<&'a str> {
+    customer
+        .connector_customer
+        .as_ref()
+        .and_then(|connector_customer_value| connector_customer_value.get(connector_name))
+        .and_then(|connector_customer| connector_customer.as_str())
 }
 
-type CreateCustomerCheck = (
-    bool,
-    Option<String>,
-    Option<serde_json::Map<String, serde_json::Value>>,
-);
-pub fn should_call_connector_create_customer(
+pub fn should_call_connector_create_customer<'a>(
     state: &AppState,
     connector: &api::ConnectorData,
-    customer: &Option<domain::Customer>,
-) -> RouterResult<CreateCustomerCheck> {
-    let connector_name = connector.connector_name.to_string();
-    //Check if create customer is required for the connector
-    let connector_customer_filter = state
+    customer: &'a Option<domain::Customer>,
+    connector_label: &str,
+) -> (bool, Option<&'a str>) {
+    // Check if create customer is required for the connector
+    let connector_needs_customer = state
         .conf
         .connector_customer
         .connector_list
         .contains(&connector.connector_name);
 
-    if connector_customer_filter {
-        match customer {
-            Some(customer) => match &customer.connector_customer {
-                Some(connector_customer) => {
-                    let connector_customer_map: serde_json::Map<String, serde_json::Value> =
-                        connector_customer
-                            .clone()
-                            .parse_value("Map<String, Value>")
-                            .change_context(errors::ApiErrorResponse::InternalServerError)
-                            .attach_printable("Failed to deserialize Value to CustomerConnector")?;
-                    let value = connector_customer_map.get(&connector_name); //Check if customer already created for this customer and for this connector
-                    Ok((
-                        value.is_none(),
-                        value.and_then(|val| val.as_str().map(|cust| cust.to_string())),
-                        Some(connector_customer_map),
-                    ))
-                }
-                None => Ok((true, None, None)),
-            },
-            None => Ok((false, None, None)),
-        }
+    if connector_needs_customer {
+        let connector_customer_details = customer.as_ref().and_then(|customer| {
+            get_connector_customer_details_if_present(customer, connector_label)
+        });
+        let should_call_connector = connector_customer_details.is_none();
+        (should_call_connector, connector_customer_details)
     } else {
-        Ok((false, None, None))
+        (false, None)
     }
 }
+
+#[instrument]
 pub async fn update_connector_customer_in_customers(
-    connector: &api::ConnectorData,
-    connector_customer_map: Option<serde_json::Map<String, serde_json::Value>>,
-    connector_cust_id: &Option<String>,
-) -> RouterResult<Option<storage::CustomerUpdate>> {
-    let mut connector_customer = match connector_customer_map {
-        Some(cc) => cc,
-        None => serde_json::Map::new(),
-    };
-    connector_cust_id.clone().map(|cc| {
-        connector_customer.insert(
-            connector.connector_name.to_string(),
-            serde_json::Value::String(cc),
+    connector_label: &str,
+    customer: Option<&domain::Customer>,
+    connector_customer_id: &Option<String>,
+) -> Option<storage::CustomerUpdate> {
+    let connector_customer_map = customer
+        .and_then(|customer| customer.connector_customer.as_ref())
+        .and_then(|connector_customer| connector_customer.as_object())
+        .map(ToOwned::to_owned)
+        .unwrap_or(serde_json::Map::new());
+
+    let updated_connector_customer_map =
+        connector_customer_id.as_ref().map(|connector_customer_id| {
+            let mut connector_customer_map = connector_customer_map;
+            let connector_customer_value =
+                serde_json::Value::String(connector_customer_id.to_string());
+            connector_customer_map.insert(connector_label.to_string(), connector_customer_value);
+            connector_customer_map
+        });
+
+    updated_connector_customer_map
+        .map(serde_json::Value::Object)
+        .map(
+            |connector_customer_value| storage::CustomerUpdate::ConnectorCustomer {
+                connector_customer: Some(connector_customer_value),
+            },
         )
-    });
-    Ok(Some(storage::CustomerUpdate::ConnectorCustomer {
-        connector_customer: Some(serde_json::Value::Object(connector_customer)),
-    }))
 }
diff --git a/crates/router/src/core/payments/flows.rs b/crates/router/src/core/payments/flows.rs
index 4bf7ece15ae..3a4b93c833f 100644
--- a/crates/router/src/core/payments/flows.rs
+++ b/crates/router/src/core/payments/flows.rs
@@ -16,7 +16,7 @@ use crate::{
     },
     routes::AppState,
     services,
-    types::{self, api, domain, storage},
+    types::{self, api, domain},
 };
 
 #[async_trait]
@@ -87,14 +87,13 @@ pub trait Feature<F, T> {
         &self,
         _state: &AppState,
         _connector: &api::ConnectorData,
-        _connector_customer_map: Option<serde_json::Map<String, serde_json::Value>>,
-    ) -> RouterResult<(Option<String>, Option<storage::CustomerUpdate>)>
+    ) -> RouterResult<Option<String>>
     where
         F: Clone,
         Self: Sized,
         dyn api::Connector: services::ConnectorIntegration<F, T, types::PaymentsResponseData>,
     {
-        Ok((None, None))
+        Ok(None)
     }
 }
 
diff --git a/crates/router/src/core/payments/flows/authorize_flow.rs b/crates/router/src/core/payments/flows/authorize_flow.rs
index abf3ec081dd..fed9ae15358 100644
--- a/crates/router/src/core/payments/flows/authorize_flow.rs
+++ b/crates/router/src/core/payments/flows/authorize_flow.rs
@@ -11,7 +11,7 @@ use crate::{
     logger,
     routes::{metrics, AppState},
     services,
-    types::{self, api, domain, storage},
+    types::{self, api, domain},
 };
 
 #[async_trait]
@@ -108,14 +108,12 @@ impl Feature<api::Authorize, types::PaymentsAuthorizeData> for types::PaymentsAu
         &self,
         state: &AppState,
         connector: &api::ConnectorData,
-        connector_customer_map: Option<serde_json::Map<String, serde_json::Value>>,
-    ) -> RouterResult<(Option<String>, Option<storage::CustomerUpdate>)> {
+    ) -> RouterResult<Option<String>> {
         customers::create_connector_customer(
             state,
             connector,
             self,
             types::ConnectorCustomerData::try_from(self)?,
-            connector_customer_map,
         )
         .await
     }
diff --git a/crates/router/src/core/payments/flows/verfiy_flow.rs b/crates/router/src/core/payments/flows/verfiy_flow.rs
index 2ff61f4baff..8026ddb2ac3 100644
--- a/crates/router/src/core/payments/flows/verfiy_flow.rs
+++ b/crates/router/src/core/payments/flows/verfiy_flow.rs
@@ -9,7 +9,7 @@ use crate::{
     },
     routes::AppState,
     services,
-    types::{self, api, domain, storage},
+    types::{self, api, domain},
 };
 
 #[async_trait]
@@ -84,14 +84,12 @@ impl Feature<api::Verify, types::VerifyRequestData> for types::VerifyRouterData
         &self,
         state: &AppState,
         connector: &api::ConnectorData,
-        connector_customer_map: Option<serde_json::Map<String, serde_json::Value>>,
-    ) -> RouterResult<(Option<String>, Option<storage::CustomerUpdate>)> {
+    ) -> RouterResult<Option<String>> {
         customers::create_connector_customer(
             state,
             connector,
             self,
             types::ConnectorCustomerData::try_from(self.request.to_owned())?,
-            connector_customer_map,
         )
         .await
     }

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

- [x] Refactoring

## Description
<!-- Describe your changes in detail -->
This PR adds support for multiple country and business label support in the connector customer. Previously the mapping to connector customer was of the form `connector_name -> connector_customer_id`, This has been changed to `connector_label -> connector_customer_id`

This Pr also includes some changes to the connector customer code to use idiomatic rust.

<!-- 
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).
-->
Fixes #1229 

## 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)?
-->
Create a payment through stripe, and check whether connector customer is generated.

![Screenshot 2023-05-25 at 4 37 03 PM](https://github.com/juspay/hyperswitch/assets/48803246/adc303e2-8cba-4909-8a5d-825b83dd350d)

Create a mandate transaction in confirm, without passing the customer id.

<img width="679" alt="Screenshot 2023-05-25 at 4 43 35 PM" src="https://github.com/juspay/hyperswitch/assets/48803246/3dc9363c-9031-44d7-ab19-550d518bf3f2">


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