Bug: Add batch api to migrate multiple payment methods at a time



#5242 In addition to this API, when merchants have multiple payment methods to migrate we are adding support to upload them as CSV and do the migration including below 2 steps

1. Create the customer if it is not already present 
2. Create the payment method and add the connector details if passed 
Sample payload:
```
merchant_id,merchant_connector_id,name,email,phone,phone_country_code,customer_id,subscription_id,payment_method,payment_method_type,nick_name,payment_instrument_id,card_number_masked,card_expiry_month,card_expiry_year,card_scheme,original_transaction_id,billing_address_zip,billing_address_state,billing_address_first_name,billing_address_last_name,billing_address_city,billing_address_country,billing_address_line1,billing_address_line2,raw_card_number,original_transaction_amount,original_transaction_currency
merchant_1720684811,mca_qqgmyxXRtsEjRLAUHBD4,noreal name,test@gmail.com,523698741,33,958709111eda444cdfdb2e8b7a3,900036959,card,credit,,4CBB27C05DE063AF598E,433026XXXXXX4675,03,2027,1,16153570198,94043,CA,NOREAL,NAME,Mountain View,US,1295 Charleston Rd,,,1000,USD
merchant_1720684811,mca_qqgmyxXRtsEjRLAUHBD4,john doe,test1@gmail.com,9159999999,1,dbc00f611eea04f024251790ff4,9000369592,card,credit,,F7A6049ADFE063AF598,378282XXXXX0005,03,2024,3,52127621321,700000,AK,NOREAL,NAME,hcm,US,hcm,,,,
```

diff --git a/Cargo.lock b/Cargo.lock
index 68792185b1b..c0e6b05c845 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2397,6 +2397,27 @@ dependencies = [
  "typenum",
 ]
 
+[[package]]
+name = "csv"
+version = "1.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ac574ff4d437a7b5ad237ef331c17ccca63c46479e5b5453eb8e10bb99a759fe"
+dependencies = [
+ "csv-core",
+ "itoa",
+ "ryu",
+ "serde",
+]
+
+[[package]]
+name = "csv-core"
+version = "0.1.11"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5efa2b3d7902f4b634a20cae3c9c4e6209dc4779feb6863329607560143efa70"
+dependencies = [
+ "memchr",
+]
+
 [[package]]
 name = "currency_conversion"
 version = "0.1.0"
@@ -6093,6 +6114,7 @@ dependencies = [
  "common_utils",
  "config",
  "cookie 0.18.1",
+ "csv",
  "currency_conversion",
  "derive_deref",
  "diesel",
diff --git a/crates/api_models/src/payment_methods.rs b/crates/api_models/src/payment_methods.rs
index 7b33e3dfeab..7799cc24df2 100644
--- a/crates/api_models/src/payment_methods.rs
+++ b/crates/api_models/src/payment_methods.rs
@@ -7,13 +7,14 @@ use common_utils::{
     id_type, link_utils, pii,
     types::{MinorUnit, Percentage, Surcharge},
 };
+use masking::PeekInterface;
 use serde::de;
 use utoipa::{schema, ToSchema};
 
 #[cfg(feature = "payouts")]
 use crate::payouts;
 use crate::{
-    admin, enums as api_enums,
+    admin, customers, enums as api_enums,
     payments::{self, BankCodeResponse},
 };
 
@@ -1335,3 +1336,177 @@ pub struct TokenizedBankRedirectValue1 {
 pub struct TokenizedBankRedirectValue2 {
     pub customer_id: Option<id_type::CustomerId>,
 }
+
+#[derive(Debug, Clone, serde::Deserialize, serde::Serialize)]
+pub struct PaymentMethodRecord {
+    pub customer_id: id_type::CustomerId,
+    pub name: Option<masking::Secret<String>>,
+    pub email: Option<pii::Email>,
+    pub phone: Option<masking::Secret<String>>,
+    pub phone_country_code: Option<String>,
+    pub merchant_id: String,
+    pub payment_method: Option<api_enums::PaymentMethod>,
+    pub payment_method_type: Option<api_enums::PaymentMethodType>,
+    pub nick_name: masking::Secret<String>,
+    pub payment_instrument_id: masking::Secret<String>,
+    pub card_number_masked: masking::Secret<String>,
+    pub card_expiry_month: masking::Secret<String>,
+    pub card_expiry_year: masking::Secret<String>,
+    pub card_scheme: Option<String>,
+    pub original_transaction_id: String,
+    pub billing_address_zip: masking::Secret<String>,
+    pub billing_address_state: masking::Secret<String>,
+    pub billing_address_first_name: masking::Secret<String>,
+    pub billing_address_last_name: masking::Secret<String>,
+    pub billing_address_city: String,
+    pub billing_address_country: Option<api_enums::CountryAlpha2>,
+    pub billing_address_line1: masking::Secret<String>,
+    pub billing_address_line2: Option<masking::Secret<String>>,
+    pub billing_address_line3: Option<masking::Secret<String>>,
+    pub raw_card_number: Option<masking::Secret<String>>,
+    pub merchant_connector_id: String,
+    pub original_transaction_amount: Option<i64>,
+    pub original_transaction_currency: Option<common_enums::Currency>,
+    pub line_number: Option<i64>,
+}
+
+#[derive(Debug, Default, serde::Serialize)]
+pub struct PaymentMethodMigrationResponse {
+    pub line_number: Option<i64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub payment_method_id: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub payment_method: Option<api_enums::PaymentMethod>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub payment_method_type: Option<api_enums::PaymentMethodType>,
+    pub customer_id: Option<id_type::CustomerId>,
+    pub migration_status: MigrationStatus,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub migration_error: Option<String>,
+    pub card_number_masked: Option<masking::Secret<String>>,
+}
+
+#[derive(Debug, Default, serde::Serialize)]
+pub enum MigrationStatus {
+    Success,
+    #[default]
+    Failed,
+}
+
+type PaymentMethodMigrationResponseType =
+    (Result<PaymentMethodResponse, String>, PaymentMethodRecord);
+impl From<PaymentMethodMigrationResponseType> for PaymentMethodMigrationResponse {
+    fn from((response, record): PaymentMethodMigrationResponseType) -> Self {
+        match response {
+            Ok(res) => Self {
+                payment_method_id: Some(res.payment_method_id),
+                payment_method: res.payment_method,
+                payment_method_type: res.payment_method_type,
+                customer_id: res.customer_id,
+                migration_status: MigrationStatus::Success,
+                migration_error: None,
+                card_number_masked: Some(record.card_number_masked),
+                line_number: record.line_number,
+            },
+            Err(e) => Self {
+                customer_id: Some(record.customer_id),
+                migration_status: MigrationStatus::Failed,
+                migration_error: Some(e),
+                card_number_masked: Some(record.card_number_masked),
+                line_number: record.line_number,
+                ..Self::default()
+            },
+        }
+    }
+}
+
+impl From<PaymentMethodRecord> for PaymentMethodMigrate {
+    fn from(record: PaymentMethodRecord) -> Self {
+        let mut mandate_reference = HashMap::new();
+        mandate_reference.insert(
+            record.merchant_connector_id,
+            PaymentsMandateReferenceRecord {
+                connector_mandate_id: record.payment_instrument_id.peek().to_string(),
+                payment_method_type: record.payment_method_type,
+                original_payment_authorized_amount: record
+                    .original_transaction_amount
+                    .or(Some(1000)),
+                original_payment_authorized_currency: record
+                    .original_transaction_currency
+                    .or(Some(common_enums::Currency::USD)),
+            },
+        );
+        Self {
+            merchant_id: record.merchant_id,
+            customer_id: Some(record.customer_id),
+            card: Some(MigrateCardDetail {
+                card_number: record.raw_card_number.unwrap_or(record.card_number_masked),
+                card_exp_month: record.card_expiry_month,
+                card_exp_year: record.card_expiry_year,
+                card_holder_name: record.name,
+                card_network: None,
+                card_type: None,
+                card_issuer: None,
+                card_issuing_country: None,
+                nick_name: Some(record.nick_name),
+            }),
+            payment_method: record.payment_method,
+            payment_method_type: record.payment_method_type,
+            payment_method_issuer: None,
+            billing: Some(payments::Address {
+                address: Some(payments::AddressDetails {
+                    city: Some(record.billing_address_city),
+                    country: record.billing_address_country,
+                    line1: Some(record.billing_address_line1),
+                    line2: record.billing_address_line2,
+                    state: Some(record.billing_address_state),
+                    line3: record.billing_address_line3,
+                    zip: Some(record.billing_address_zip),
+                    first_name: Some(record.billing_address_first_name),
+                    last_name: Some(record.billing_address_last_name),
+                }),
+                phone: Some(payments::PhoneDetails {
+                    number: record.phone,
+                    country_code: record.phone_country_code,
+                }),
+                email: record.email,
+            }),
+            connector_mandate_details: Some(PaymentsMandateReference(mandate_reference)),
+            metadata: None,
+            payment_method_issuer_code: None,
+            card_network: None,
+            #[cfg(feature = "payouts")]
+            bank_transfer: None,
+            #[cfg(feature = "payouts")]
+            wallet: None,
+            payment_method_data: None,
+            network_transaction_id: record.original_transaction_id.into(),
+        }
+    }
+}
+
+impl From<PaymentMethodRecord> for customers::CustomerRequest {
+    fn from(record: PaymentMethodRecord) -> Self {
+        Self {
+            customer_id: Some(record.customer_id),
+            merchant_id: record.merchant_id,
+            name: record.name,
+            email: record.email,
+            phone: record.phone,
+            description: None,
+            phone_country_code: record.phone_country_code,
+            address: Some(payments::AddressDetails {
+                city: Some(record.billing_address_city),
+                country: record.billing_address_country,
+                line1: Some(record.billing_address_line1),
+                line2: record.billing_address_line2,
+                state: Some(record.billing_address_state),
+                line3: record.billing_address_line3,
+                zip: Some(record.billing_address_zip),
+                first_name: Some(record.billing_address_first_name),
+                last_name: Some(record.billing_address_last_name),
+            }),
+            metadata: None,
+        }
+    }
+}
diff --git a/crates/router/Cargo.toml b/crates/router/Cargo.toml
index 4373c7fb169..108b035ec0f 100644
--- a/crates/router/Cargo.toml
+++ b/crates/router/Cargo.toml
@@ -51,6 +51,7 @@ bytes = "1.6.0"
 clap = { version = "4.4.18", default-features = false, features = ["std", "derive", "help", "usage"] }
 config = { version = "0.14.0", features = ["toml"] }
 cookie = "0.18.1"
+csv = "1.3.0"
 diesel = { version = "2.1.5", features = ["postgres"] }
 digest = "0.10.7"
 dyn-clone = "1.0.17"
diff --git a/crates/router/src/core/customers.rs b/crates/router/src/core/customers.rs
index ad1a292df09..c0e0291b8c4 100644
--- a/crates/router/src/core/customers.rs
+++ b/crates/router/src/core/customers.rs
@@ -470,3 +470,28 @@ pub async fn update_customer(
         customers::CustomerResponse::from((response, update_customer.address)),
     ))
 }
+
+pub async fn migrate_customers(
+    state: SessionState,
+    customers: Vec<customers::CustomerRequest>,
+    merchant_account: domain::MerchantAccount,
+    key_store: domain::MerchantKeyStore,
+) -> errors::CustomerResponse<()> {
+    for customer in customers {
+        match create_customer(
+            state.clone(),
+            merchant_account.clone(),
+            key_store.clone(),
+            customer,
+        )
+        .await
+        {
+            Ok(_) => (),
+            Err(e) => match e.current_context() {
+                errors::CustomersErrorResponse::CustomerAlreadyExists => (),
+                _ => return Err(e),
+            },
+        }
+    }
+    Ok(services::ApplicationResponse::Json(()))
+}
diff --git a/crates/router/src/core/payment_methods.rs b/crates/router/src/core/payment_methods.rs
index 6fcd27fefd0..a7a87b6d96e 100644
--- a/crates/router/src/core/payment_methods.rs
+++ b/crates/router/src/core/payment_methods.rs
@@ -1,4 +1,5 @@
 pub mod cards;
+pub mod migration;
 pub mod surcharge_decision_configs;
 pub mod transformers;
 pub mod utils;
diff --git a/crates/router/src/core/payment_methods/cards.rs b/crates/router/src/core/payment_methods/cards.rs
index 33e188b8ca0..db95b5c838c 100644
--- a/crates/router/src/core/payment_methods/cards.rs
+++ b/crates/router/src/core/payment_methods/cards.rs
@@ -269,23 +269,10 @@ pub async fn get_or_insert_payment_method(
 pub async fn migrate_payment_method(
     state: routes::SessionState,
     req: api::PaymentMethodMigrate,
+    merchant_id: &str,
+    merchant_account: &domain::MerchantAccount,
+    key_store: &domain::MerchantKeyStore,
 ) -> errors::RouterResponse<api::PaymentMethodResponse> {
-    let merchant_id = &req.merchant_id;
-    let key_store = state
-        .store
-        .get_merchant_key_store_by_merchant_id(
-            merchant_id,
-            &state.store.get_master_key().to_vec().into(),
-        )
-        .await
-        .to_not_found_response(errors::ApiErrorResponse::MerchantAccountNotFound)?;
-
-    let merchant_account = state
-        .store
-        .find_merchant_account_by_merchant_id(merchant_id, &key_store)
-        .await
-        .to_not_found_response(errors::ApiErrorResponse::MerchantAccountNotFound)?;
-
     let card_details = req.card.as_ref().get_required_value("card")?;
 
     let card_number_validation_result =
@@ -294,7 +281,7 @@ pub async fn migrate_payment_method(
     if let Some(connector_mandate_details) = &req.connector_mandate_details {
         helpers::validate_merchant_connector_ids_in_connector_mandate_details(
             &*state.store,
-            &key_store,
+            key_store,
             connector_mandate_details,
             merchant_id,
         )
@@ -311,8 +298,8 @@ pub async fn migrate_payment_method(
             get_client_secret_or_add_payment_method(
                 state,
                 payment_method_create_request,
-                &merchant_account,
-                &key_store,
+                merchant_account,
+                key_store,
             )
             .await
         }
@@ -322,8 +309,8 @@ pub async fn migrate_payment_method(
                 state,
                 &req,
                 merchant_id.into(),
-                &key_store,
-                &merchant_account,
+                key_store,
+                merchant_account,
             )
             .await
         }
@@ -513,7 +500,10 @@ pub async fn skip_locker_call_and_migrate_payment_method(
                 payment_method: req.payment_method,
                 payment_method_type: req.payment_method_type,
                 payment_method_issuer: req.payment_method_issuer.clone(),
-                scheme: req.card_network.clone(),
+                scheme: req
+                    .card_network
+                    .clone()
+                    .or(card.clone().and_then(|card| card.scheme.clone())),
                 metadata: payment_method_metadata.map(Secret::new),
                 payment_method_data: payment_method_data_encrypted.map(Into::into),
                 connector_mandate_details: Some(connector_mandate_details),
diff --git a/crates/router/src/core/payment_methods/migration.rs b/crates/router/src/core/payment_methods/migration.rs
new file mode 100644
index 00000000000..83e9993478b
--- /dev/null
+++ b/crates/router/src/core/payment_methods/migration.rs
@@ -0,0 +1,85 @@
+use actix_multipart::form::{bytes::Bytes, MultipartForm};
+use api_models::payment_methods::{PaymentMethodMigrationResponse, PaymentMethodRecord};
+use csv::Reader;
+use rdkafka::message::ToBytes;
+
+use crate::{
+    core::{errors, payment_methods::cards::migrate_payment_method},
+    routes, services,
+    types::{api, domain},
+};
+
+pub async fn migrate_payment_methods(
+    state: routes::SessionState,
+    payment_methods: Vec<PaymentMethodRecord>,
+    merchant_id: &str,
+    merchant_account: &domain::MerchantAccount,
+    key_store: &domain::MerchantKeyStore,
+) -> errors::RouterResponse<Vec<PaymentMethodMigrationResponse>> {
+    let mut result = Vec::new();
+    for record in payment_methods {
+        let res = migrate_payment_method(
+            state.clone(),
+            api::PaymentMethodMigrate::from(record.clone()),
+            merchant_id,
+            merchant_account,
+            key_store,
+        )
+        .await;
+        result.push(PaymentMethodMigrationResponse::from((
+            match res {
+                Ok(services::api::ApplicationResponse::Json(response)) => Ok(response),
+                Err(e) => Err(e.to_string()),
+                _ => Err("Failed to migrate payment method".to_string()),
+            },
+            record,
+        )));
+    }
+    Ok(services::api::ApplicationResponse::Json(result))
+}
+
+#[derive(Debug, MultipartForm)]
+pub struct PaymentMethodsMigrateForm {
+    #[multipart(limit = "1MB")]
+    pub file: Bytes,
+}
+
+fn parse_csv(data: &[u8]) -> csv::Result<Vec<PaymentMethodRecord>> {
+    let mut csv_reader = Reader::from_reader(data);
+    let mut records = Vec::new();
+    let mut id_counter = 0;
+    for result in csv_reader.deserialize() {
+        let mut record: PaymentMethodRecord = result?;
+        id_counter += 1;
+        record.line_number = Some(id_counter);
+        records.push(record);
+    }
+    Ok(records)
+}
+pub fn get_payment_method_records(
+    form: PaymentMethodsMigrateForm,
+) -> Result<(String, Vec<PaymentMethodRecord>), errors::ApiErrorResponse> {
+    match parse_csv(form.file.data.to_bytes()) {
+        Ok(records) => {
+            if let Some(first_record) = records.first() {
+                if records
+                    .iter()
+                    .all(|merchant_id| merchant_id.merchant_id == first_record.merchant_id)
+                {
+                    Ok((first_record.merchant_id.clone(), records))
+                } else {
+                    Err(errors::ApiErrorResponse::PreconditionFailed {
+                        message: "Only one merchant id can be updated at a time".to_string(),
+                    })
+                }
+            } else {
+                Err(errors::ApiErrorResponse::PreconditionFailed {
+                    message: "No records found".to_string(),
+                })
+            }
+        }
+        Err(e) => Err(errors::ApiErrorResponse::PreconditionFailed {
+            message: e.to_string(),
+        }),
+    }
+}
diff --git a/crates/router/src/routes/app.rs b/crates/router/src/routes/app.rs
index d82f98f295d..b8b329a3acb 100644
--- a/crates/router/src/routes/app.rs
+++ b/crates/router/src/routes/app.rs
@@ -934,6 +934,9 @@ impl PaymentMethods {
                 .service(
                     web::resource("/migrate").route(web::post().to(migrate_payment_method_api)),
                 )
+                .service(
+                    web::resource("/migrate-batch").route(web::post().to(migrate_payment_methods)),
+                )
                 .service(
                     web::resource("/collect").route(web::post().to(initiate_pm_collect_link_flow)),
                 )
diff --git a/crates/router/src/routes/payment_methods.rs b/crates/router/src/routes/payment_methods.rs
index 8ff6e48237c..003cd71c00d 100644
--- a/crates/router/src/routes/payment_methods.rs
+++ b/crates/router/src/routes/payment_methods.rs
@@ -1,18 +1,25 @@
+use actix_multipart::form::MultipartForm;
 use actix_web::{web, HttpRequest, HttpResponse};
 use common_utils::{errors::CustomResult, id_type};
 use diesel_models::enums::IntentStatus;
 use error_stack::ResultExt;
+use hyperswitch_domain_models::merchant_key_store::MerchantKeyStore;
 use router_env::{instrument, logger, tracing, Flow};
 
 use super::app::{AppState, SessionState};
 use crate::{
     core::{
-        api_locking, errors,
-        payment_methods::{self as payment_methods_routes, cards},
+        api_locking, customers, errors,
+        errors::utils::StorageErrorExt,
+        payment_methods::{self as payment_methods_routes, cards, migration},
     },
     services::{api, authentication as auth, authorization::permissions::Permission},
     types::{
-        api::payment_methods::{self, PaymentMethodId},
+        api::{
+            customers::CustomerRequest,
+            payment_methods::{self, PaymentMethodId},
+        },
+        domain,
         storage::payment_method::PaymentTokenData,
     },
     utils::Encode,
@@ -59,7 +66,84 @@ pub async fn migrate_payment_method_api(
         state,
         &req,
         json_payload.into_inner(),
-        |state, _, req, _| async move { Box::pin(cards::migrate_payment_method(state, req)).await },
+        |state, _, req, _| async move {
+            let merchant_id = req.merchant_id.clone();
+            let (key_store, merchant_account) = get_merchant_account(&state, &merchant_id).await?;
+            Box::pin(cards::migrate_payment_method(
+                state,
+                req,
+                &merchant_id,
+                &merchant_account,
+                &key_store,
+            ))
+            .await
+        },
+        &auth::AdminApiAuth,
+        api_locking::LockAction::NotApplicable,
+    ))
+    .await
+}
+
+async fn get_merchant_account(
+    state: &SessionState,
+    merchant_id: &str,
+) -> CustomResult<(MerchantKeyStore, domain::MerchantAccount), errors::ApiErrorResponse> {
+    let key_store = state
+        .store
+        .get_merchant_key_store_by_merchant_id(
+            merchant_id,
+            &state.store.get_master_key().to_vec().into(),
+        )
+        .await
+        .to_not_found_response(errors::ApiErrorResponse::MerchantAccountNotFound)?;
+
+    let merchant_account = state
+        .store
+        .find_merchant_account_by_merchant_id(merchant_id, &key_store)
+        .await
+        .to_not_found_response(errors::ApiErrorResponse::MerchantAccountNotFound)?;
+    Ok((key_store, merchant_account))
+}
+
+#[instrument(skip_all, fields(flow = ?Flow::PaymentMethodsMigrate))]
+pub async fn migrate_payment_methods(
+    state: web::Data<AppState>,
+    req: HttpRequest,
+    MultipartForm(form): MultipartForm<migration::PaymentMethodsMigrateForm>,
+) -> HttpResponse {
+    let flow = Flow::PaymentMethodsMigrate;
+    let (merchant_id, records) = match migration::get_payment_method_records(form) {
+        Ok((merchant_id, records)) => (merchant_id, records),
+        Err(e) => return api::log_and_return_error_response(e.into()),
+    };
+    let merchant_id = merchant_id.as_str();
+    Box::pin(api::server_wrap(
+        flow,
+        state,
+        &req,
+        records,
+        |state, _, req, _| async move {
+            let (key_store, merchant_account) = get_merchant_account(&state, merchant_id).await?;
+            // Create customers if they are not already present
+            customers::migrate_customers(
+                state.clone(),
+                req.iter()
+                    .map(|e| CustomerRequest::from(e.clone()))
+                    .collect(),
+                merchant_account.clone(),
+                key_store.clone(),
+            )
+            .await
+            .change_context(errors::ApiErrorResponse::InternalServerError)?;
+            Box::pin(migration::migrate_payment_methods(
+                state,
+                req,
+                merchant_id,
+                &merchant_account,
+                &key_store,
+            ))
+            .await
+        },
         &auth::AdminApiAuth,
         api_locking::LockAction::NotApplicable,
     ))
diff --git a/crates/router/src/services/api.rs b/crates/router/src/services/api.rs
index 540c67c5762..1a488721905 100644
--- a/crates/router/src/services/api.rs
+++ b/crates/router/src/services/api.rs
@@ -740,7 +740,15 @@ pub enum AuthFlow {
 
 #[allow(clippy::too_many_arguments)]
 #[instrument(
-    skip(request, payload, state, func, api_auth, request_state),
+    skip(
+        request,
+        payload,
+        state,
+        func,
+        api_auth,
+        request_state,
+        incoming_request_header
+    ),
     fields(merchant_id)
 )]
 pub async fn server_wrap_util<'a, 'b, U, T, Q, F, Fut, E, OErr>(

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

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

## Description
<!-- Describe your changes in detail -->
Merchants who has existing customer payment methods in payment processor can upload a csv file to migrate their customers and their payment methods using this `/migrate-bulk` API

### 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).
-->
To provide seamless experience to customers when merchant moves from payment processor to hyperswitch

## How did you test it?
<!--
Did you write an integration/unit/API test to verify the code changes?
Or did you test this change manually (provide relevant screenshots)?
-->
1. Add below records and see if the new customer and payment methods are created. try making a recurring payment with the payment method saved
```
merchant_id,merchant_connector_id,name,email,phone,phone_country_code,customer_id,subscription_id,payment_method,payment_method_type,nick_name,payment_instrument_id,card_number_masked,card_expiry_month,card_expiry_year,card_scheme,original_transaction_id,billing_address_zip,billing_address_state,billing_address_first_name,billing_address_last_name,billing_address_city,billing_address_country,billing_address_line1,billing_address_line2,raw_card_number,original_transaction_amount,original_transaction_currency
merchant_1720684811,mca_qqgmyxXRtsEjRLAUHBD4,noreal name,test@gmail.com,523698741,33,958709111eda444cdfdb2e8b7a3,900036959,card,credit,,4CBB27C05DE063AF598E,433026XXXXXX4675,03,2027,1,16153570198,94043,CA,NOREAL,NAME,Mountain View,US,1295 Charleston Rd,,,1000,USD
merchant_1720684811,mca_qqgmyxXRtsEjRLAUHBD4,john doe,test1@gmail.com,9159999999,1,dbc00f611eea04f024251790ff4,9000369592,card,credit,,F7A6049ADFE063AF598,378282XXXXX0005,03,2024,3,52127621321,700000,AK,NOREAL,NAME,hcm,US,hcm,,,,
```
```
curl --location 'http://localhost:8080/payment_methods/migrate-batch' \
--header 'api-key: test_admin' \
--header 'Content-Type: application/json' \
--form 'file=@"/Users/jagan.elavarasan/Downloads/sample.csv"'
```
<img width="1000" alt="image" src="https://github.com/user-attachments/assets/ba2e3ccc-ccf8-48d9-b148-931aea4cfe3f">

```
curl --location 'http://localhost:8080/customers/edca1edee0e411ec961c506b8dc695e7/payment_methods' \
--header 'Accept: application/json' \
--header 'x-feature: router-custom-be' \
--header 'api-key: dev_hRgXc8VdzglIywDYBmC7MBpw7dNGIAniCGRdqrF7cDdkWLoGi2iMfEI83pVGz2zz'
```
<img width="694" alt="image" src="https://github.com/user-attachments/assets/d96af05c-1b40-4fce-831d-eb220b4cb8d0">

```
curl --location 'http://localhost:8080/payments/pay_xAY8tQGtAijAhFX4a4pq/confirm' \
--header 'Content-Type: application/json' \
--header 'Accept: application/json' \
--header 'api-key: pk_dev_3a60113d6c90405d99a563e78961880b' \
--data '{
    "payment_method": "card",
    "client_secret": "pay_xAY8tQGtAijAhFX4a4pq_secret_AgGaWFPSW3inNKcxkHcg",
    "payment_token": "token_3VdfjykAKGS1ukxixQoM"
}'
```
<img width="627" alt="image" src="https://github.com/user-attachments/assets/45ad03c2-1947-491c-b841-b55bea3ad20c">

2. Try uploading big file > 1MB, it should throw appropriate error

```
curl --location 'http://localhost:8080/payment_methods/migrate-batch' \
--header 'api-key: test_admin' \
--header 'Content-Type: application/json' \
--form 'file=@"/Users/jagan.elavarasan/Downloads/sample.csv"'
```
<img width="595" alt="image" src="https://github.com/user-attachments/assets/47adf4a9-3c28-40a8-b715-e400238bed78">

3. Try uploading mulitple merchant at a time, it should not be allowed
<img width="555" alt="image" src="https://github.com/user-attachments/assets/a1af5dcb-168f-43b1-846c-af782333bcd7">

4. Should not allow merchant or merchant_connector_account if not created before
 <img width="567" alt="Screenshot 2024-07-12 at 15 06 50" src="https://github.com/user-attachments/assets/e3992488-b965-457d-810b-56c85db8dab6">
<img width="1012" alt="image" src="https://github.com/user-attachments/assets/95b8186a-06e8-4d1e-ad04-702e94703482">


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