feat(data-migration): add connector customer and mandate details support for multiple profiles (#8473)

Co-authored-by: hyperswitch-bot[bot] <148525504+hyperswitch-bot[bot]@users.noreply.github.com>

crates/payment_methods/src/core/migration.rs

@@ -24,20 +24,22 @@ pub async fn migrate_payment_methods(
     payment_methods: Vec<pm_api::PaymentMethodRecord>,
     merchant_id: &common_utils::id_type::MerchantId,
     merchant_context: &merchant_context::MerchantContext,
-    mca_id: Option<common_utils::id_type::MerchantConnectorAccountId>,
+    mca_ids: Option<Vec<common_utils::id_type::MerchantConnectorAccountId>>,
     controller: &dyn pm::PaymentMethodsController,
 ) -> PmMigrationResult<Vec<pm_api::PaymentMethodMigrationResponse>> {
-    let mut result = Vec::new();
+    let mut result = Vec::with_capacity(payment_methods.len());
+
     for record in payment_methods {
         let req = pm_api::PaymentMethodMigrate::try_from((
-            record.clone(),
+            &record,
             merchant_id.clone(),
-            mca_id.clone(),
+            mca_ids.as_ref(),
         ))
         .map_err(|err| errors::ApiErrorResponse::InvalidRequestData {
             message: format!("error: {:?}", err),
         })
         .attach_printable("record deserialization failed");
+
         let res = match req {
             Ok(migrate_request) => {
                 let res = migrate_payment_method(
@@ -56,6 +58,7 @@ pub async fn migrate_payment_methods(
             }
             Err(e) => Err(e.to_string()),
         };
+
         result.push(pm_api::PaymentMethodMigrationResponse::from((res, record)));
     }
     Ok(api::ApplicationResponse::Json(result))
@@ -69,7 +72,138 @@ pub struct PaymentMethodsMigrateForm {
     pub merchant_id: text::Text<common_utils::id_type::MerchantId>,
 
     pub merchant_connector_id:
-        text::Text<Option<common_utils::id_type::MerchantConnectorAccountId>>,
+        Option<text::Text<common_utils::id_type::MerchantConnectorAccountId>>,
+
+    pub merchant_connector_ids: Option<text::Text<String>>,
+}
+
+struct MerchantConnectorValidator;
+
+impl MerchantConnectorValidator {
+    fn parse_comma_separated_ids(
+        ids_string: &str,
+    ) -> Result<Vec<common_utils::id_type::MerchantConnectorAccountId>, errors::ApiErrorResponse>
+    {
+        // Estimate capacity based on comma count
+        let capacity = ids_string.matches(',').count() + 1;
+        let mut result = Vec::with_capacity(capacity);
+
+        for id in ids_string.split(',') {
+            let trimmed_id = id.trim();
+            if !trimmed_id.is_empty() {
+                let mca_id =
+                    common_utils::id_type::MerchantConnectorAccountId::wrap(trimmed_id.to_string())
+                        .map_err(|_| errors::ApiErrorResponse::InvalidRequestData {
+                            message: format!(
+                                "Invalid merchant_connector_account_id: {}",
+                                trimmed_id
+                            ),
+                        })?;
+                result.push(mca_id);
+            }
+        }
+
+        Ok(result)
+    }
+
+    fn validate_form_csv_conflicts(
+        records: &[pm_api::PaymentMethodRecord],
+        form_has_single_id: bool,
+        form_has_multiple_ids: bool,
+    ) -> Result<(), errors::ApiErrorResponse> {
+        if form_has_single_id {
+            // If form has merchant_connector_id, CSV records should not have merchant_connector_ids
+            for (index, record) in records.iter().enumerate() {
+                if record.merchant_connector_ids.is_some() {
+                    return Err(errors::ApiErrorResponse::InvalidRequestData {
+                        message: format!(
+                            "Record at line {} has merchant_connector_ids but form has merchant_connector_id. Only one should be provided",
+                            index + 1
+                        ),
+                    });
+                }
+            }
+        }
+
+        if form_has_multiple_ids {
+            // If form has merchant_connector_ids, CSV records should not have merchant_connector_id
+            for (index, record) in records.iter().enumerate() {
+                if record.merchant_connector_id.is_some() {
+                    return Err(errors::ApiErrorResponse::InvalidRequestData {
+                        message: format!(
+                            "Record at line {} has merchant_connector_id but form has merchant_connector_ids. Only one should be provided",
+                            index + 1
+                        ),
+                    });
+                }
+            }
+        }
+
+        Ok(())
+    }
+}
+
+type MigrationValidationResult = Result<
+    (
+        common_utils::id_type::MerchantId,
+        Vec<pm_api::PaymentMethodRecord>,
+        Option<Vec<common_utils::id_type::MerchantConnectorAccountId>>,
+    ),
+    errors::ApiErrorResponse,
+>;
+
+impl PaymentMethodsMigrateForm {
+    pub fn validate_and_get_payment_method_records(self) -> MigrationValidationResult {
+        // Step 1: Validate form-level conflicts
+        let form_has_single_id = self.merchant_connector_id.is_some();
+        let form_has_multiple_ids = self.merchant_connector_ids.is_some();
+
+        if form_has_single_id && form_has_multiple_ids {
+            return Err(errors::ApiErrorResponse::InvalidRequestData {
+                message: "Both merchant_connector_id and merchant_connector_ids cannot be provided"
+                    .to_string(),
+            });
+        }
+
+        // Ensure at least one is provided
+        if !form_has_single_id && !form_has_multiple_ids {
+            return Err(errors::ApiErrorResponse::InvalidRequestData {
+                message: "Either merchant_connector_id or merchant_connector_ids must be provided"
+                    .to_string(),
+            });
+        }
+
+        // Step 2: Parse CSV
+        let records = parse_csv(self.file.data.to_bytes()).map_err(|e| {
+            errors::ApiErrorResponse::PreconditionFailed {
+                message: e.to_string(),
+            }
+        })?;
+
+        // Step 3: Validate CSV vs Form conflicts
+        MerchantConnectorValidator::validate_form_csv_conflicts(
+            &records,
+            form_has_single_id,
+            form_has_multiple_ids,
+        )?;
+
+        // Step 4: Prepare the merchant connector account IDs for return
+        let mca_ids = if let Some(ref single_id) = self.merchant_connector_id {
+            Some(vec![(**single_id).clone()])
+        } else if let Some(ref ids_string) = self.merchant_connector_ids {
+            let parsed_ids = MerchantConnectorValidator::parse_comma_separated_ids(ids_string)?;
+            if parsed_ids.is_empty() {
+                None
+            } else {
+                Some(parsed_ids)
+            }
+        } else {
+            None
+        };
+
+        // Step 5: Return the updated structure
+        Ok((self.merchant_id.clone(), records, mca_ids))
+    }
 }
 
 fn parse_csv(data: &[u8]) -> csv::Result<Vec<pm_api::PaymentMethodRecord>> {
@@ -84,27 +218,6 @@ fn parse_csv(data: &[u8]) -> csv::Result<Vec<pm_api::PaymentMethodRecord>> {
     }
     Ok(records)
 }
-pub fn get_payment_method_records(
-    form: PaymentMethodsMigrateForm,
-) -> Result<
-    (
-        common_utils::id_type::MerchantId,
-        Vec<pm_api::PaymentMethodRecord>,
-        Option<common_utils::id_type::MerchantConnectorAccountId>,
-    ),
-    errors::ApiErrorResponse,
-> {
-    match parse_csv(form.file.data.to_bytes()) {
-        Ok(records) => {
-            let merchant_id = form.merchant_id.clone();
-            let mca_id = form.merchant_connector_id.clone();
-            Ok((merchant_id.clone(), records, mca_id))
-        }
-        Err(e) => Err(errors::ApiErrorResponse::PreconditionFailed {
-            message: e.to_string(),
-        }),
-    }
-}
 
 #[instrument(skip_all)]
 pub fn validate_card_expiry(