Issue:
Bug: Add support to update card exp in update payment methods api



1. Need support to update card expiry in payments method data of payment methods table using batch pm update API.
2. Need support to pass multiple mca ids for updating payment methods entries.



Pull request description:
## Type of Change
<!-- Put an `x` in the boxes that apply -->

- [ ] Bugfix
- [ ] New feature
- [x] Enhancement
- [ ] Refactoring
- [ ] Dependency updates
- [ ] Documentation
- [ ] CI/CD

## Description
<!-- Describe your changes in detail -->
We can update card expiry in payments method data of payment methods table using batch pm update API>
Added support to pass multiple mca ids for updating payment methods entry.

### 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?
<!--
Did you write an integration/unit/API test to verify the code changes?
Or did you test this change manually (provide relevant screenshots)?
-->
Curls:
```
curl --location 'http://localhost:8080/payment_methods/update-batch' \
--header 'api-key: test_admin' \
--form 'file=@"/Users/mrudul.vajpayee/Downloads/update_check.csv"' \
--form 'merchant_id="merchant_1759727351"'
```
Response:
```
[{"payment_method_id":"pm_lGbP1ZvFLblFT3DafY5L","status":"active","network_transaction_id":null,"connector_mandate_details":{"payouts":{"mca_5hGHfrn1BYGHjse3VT6f":{"transfer_method_id":"yPNWMjhW4h5LmWJYYY"}},"mca_pwD3i1oyHLxzcu9cq1xR":{"mandate_metadata":null,"payment_method_type":"credit","connector_mandate_id":"yPNWMjhW4h5LmWJYYY","connector_mandate_status":"active","original_payment_authorized_amount":3545,"original_payment_authorized_currency":"EUR","connector_mandate_request_reference_id":"LtkxiLx5mDw1p8uLmD"}},"update_status":"Success","updated_payment_method_data":true,"line_number":1}]
```
File used for above:
[update_check.csv](https://github.com/user-attachments/files/22723044/update_check.csv)


## Checklist
<!-- Put an `x` in the boxes that apply -->

- [x] 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:

Curls:
```
curl --location 'http://localhost:8080/payment_methods/update-batch' \
--header 'api-key: test_admin' \
--form 'file=@"/Users/mrudul.vajpayee/Downloads/update_check.csv"' \
--form 'merchant_id="merchant_1759727351"'
```
Response:
```
[{"payment_method_id":"pm_lGbP1ZvFLblFT3DafY5L","status":"active","network_transaction_id":null,"connector_mandate_details":{"payouts":{"mca_5hGHfrn1BYGHjse3VT6f":{"transfer_method_id":"yPNWMjhW4h5LmWJYYY"}},"mca_pwD3i1oyHLxzcu9cq1xR":{"mandate_metadata":null,"payment_method_type":"credit","connector_mandate_id":"yPNWMjhW4h5LmWJYYY","connector_mandate_status":"active","original_payment_authorized_amount":3545,"original_payment_authorized_currency":"EUR","connector_mandate_request_reference_id":"LtkxiLx5mDw1p8uLmD"}},"update_status":"Success","updated_payment_method_data":true,"line_number":1}]
```
File used for above:
[update_check.csv](https://github.com/user-attachments/files/22723044/update_check.csv)

Patch:
diff --git a/crates/api_models/src/payment_methods.rs b/crates/api_models/src/payment_methods.rs
index 591932c5deb..cfb45ef045c 100644
--- a/crates/api_models/src/payment_methods.rs
+++ b/crates/api_models/src/payment_methods.rs
@@ -294,6 +294,7 @@ pub struct PaymentMethodRecordUpdateResponse {
     pub status: common_enums::PaymentMethodStatus,
     pub network_transaction_id: Option<String>,
     pub connector_mandate_details: Option<pii::SecretSerdeValue>,
+    pub updated_payment_method_data: Option<bool>,
 }
 
 #[derive(Debug, Default, Clone, serde::Serialize, serde::Deserialize)]
@@ -2689,7 +2690,9 @@ pub struct UpdatePaymentMethodRecord {
     pub network_transaction_id: Option<String>,
     pub line_number: Option<i64>,
     pub payment_instrument_id: Option<masking::Secret<String>>,
-    pub merchant_connector_id: Option<id_type::MerchantConnectorAccountId>,
+    pub merchant_connector_ids: Option<String>,
+    pub card_expiry_month: Option<masking::Secret<String>>,
+    pub card_expiry_year: Option<masking::Secret<String>>,
 }
 
 #[derive(Debug, serde::Serialize)]
@@ -2701,6 +2704,7 @@ pub struct PaymentMethodUpdateResponse {
     pub update_status: UpdateStatus,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub update_error: Option<String>,
+    pub updated_payment_method_data: Option<bool>,
     pub line_number: Option<i64>,
 }
 
@@ -2841,6 +2845,7 @@ impl From<PaymentMethodUpdateResponseType> for PaymentMethodUpdateResponse {
                 status: Some(res.status),
                 network_transaction_id: res.network_transaction_id,
                 connector_mandate_details: res.connector_mandate_details,
+                updated_payment_method_data: res.updated_payment_method_data,
                 update_status: UpdateStatus::Success,
                 update_error: None,
                 line_number: record.line_number,
@@ -2850,6 +2855,7 @@ impl From<PaymentMethodUpdateResponseType> for PaymentMethodUpdateResponse {
                 status: record.status,
                 network_transaction_id: record.network_transaction_id,
                 connector_mandate_details: None,
+                updated_payment_method_data: None,
                 update_status: UpdateStatus::Failed,
                 update_error: Some(e),
                 line_number: record.line_number,
diff --git a/crates/diesel_models/src/payment_method.rs b/crates/diesel_models/src/payment_method.rs
index 747c6fdbd39..edadeea7990 100644
--- a/crates/diesel_models/src/payment_method.rs
+++ b/crates/diesel_models/src/payment_method.rs
@@ -251,10 +251,11 @@ pub enum PaymentMethodUpdate {
         connector_mandate_details: Option<pii::SecretSerdeValue>,
         network_transaction_id: Option<Secret<String>>,
     },
-    ConnectorNetworkTransactionIdStatusAndMandateDetailsUpdate {
+    PaymentMethodBatchUpdate {
         connector_mandate_details: Option<pii::SecretSerdeValue>,
         network_transaction_id: Option<String>,
         status: Option<storage_enums::PaymentMethodStatus>,
+        payment_method_data: Option<Encryption>,
     },
 }
 
@@ -687,13 +688,13 @@ impl From<PaymentMethodUpdate> for PaymentMethodUpdateInternal {
                 network_token_payment_method_data: None,
                 scheme: None,
             },
-            PaymentMethodUpdate::ConnectorNetworkTransactionIdStatusAndMandateDetailsUpdate {
+            PaymentMethodUpdate::PaymentMethodBatchUpdate {
                 connector_mandate_details,
                 network_transaction_id,
                 status,
+                payment_method_data,
             } => Self {
                 metadata: None,
-                payment_method_data: None,
                 last_used_at: None,
                 status,
                 locker_id: None,
@@ -709,6 +710,7 @@ impl From<PaymentMethodUpdate> for PaymentMethodUpdateInternal {
                 network_token_locker_id: None,
                 network_token_payment_method_data: None,
                 scheme: None,
+                payment_method_data,
             },
         }
     }
diff --git a/crates/payment_methods/src/core/migration.rs b/crates/payment_methods/src/core/migration.rs
index ea9b6db1036..c077ddb2cfa 100644
--- a/crates/payment_methods/src/core/migration.rs
+++ b/crates/payment_methods/src/core/migration.rs
@@ -77,10 +77,10 @@ pub struct PaymentMethodsMigrateForm {
     pub merchant_connector_ids: Option<text::Text<String>>,
 }
 
-struct MerchantConnectorValidator;
+pub struct MerchantConnectorValidator;
 
 impl MerchantConnectorValidator {
-    fn parse_comma_separated_ids(
+    pub fn parse_comma_separated_ids(
         ids_string: &str,
     ) -> Result<Vec<common_utils::id_type::MerchantConnectorAccountId>, errors::ApiErrorResponse>
     {
diff --git a/crates/router/src/core/payment_methods/migration.rs b/crates/router/src/core/payment_methods/migration.rs
index d802e432ec1..26a2d1737cb 100644
--- a/crates/router/src/core/payment_methods/migration.rs
+++ b/crates/router/src/core/payment_methods/migration.rs
@@ -7,12 +7,15 @@ use hyperswitch_domain_models::{
     api::ApplicationResponse, errors::api_error_response as errors, merchant_context,
     payment_methods::PaymentMethodUpdate,
 };
-use masking::PeekInterface;
+use masking::{ExposeInterface, PeekInterface};
+use payment_methods::core::migration::MerchantConnectorValidator;
 use rdkafka::message::ToBytes;
 use router_env::logger;
 
-use crate::{core::errors::StorageErrorExt, routes::SessionState};
-
+use crate::{
+    core::{errors::StorageErrorExt, payment_methods::cards::create_encrypted_data},
+    routes::SessionState,
+};
 type PmMigrationResult<T> = CustomResult<ApplicationResponse<T>, errors::ApiErrorResponse>;
 
 #[cfg(feature = "v1")]
@@ -62,6 +65,8 @@ pub async fn update_payment_method_record(
     let payment_method_id = req.payment_method_id.clone();
     let network_transaction_id = req.network_transaction_id.clone();
     let status = req.status;
+    let key_manager_state = state.into();
+    let mut updated_card_expiry = false;
 
     let payment_method = db
         .find_payment_method(
@@ -79,94 +84,143 @@ pub async fn update_payment_method_record(
                 }.into());
     }
 
-    // Process mandate details when both payment_instrument_id and merchant_connector_id are present
-    let pm_update = match (&req.payment_instrument_id, &req.merchant_connector_id) {
-        (Some(payment_instrument_id), Some(merchant_connector_id)) => {
+    let updated_payment_method_data = match payment_method.payment_method_data.as_ref() {
+        Some(data) => {
+            match serde_json::from_value::<pm_api::PaymentMethodsData>(
+                data.clone().into_inner().expose(),
+            ) {
+                Ok(pm_api::PaymentMethodsData::Card(mut card_data)) => {
+                    if let Some(new_month) = &req.card_expiry_month {
+                        card_data.expiry_month = Some(new_month.clone());
+                        updated_card_expiry = true;
+                    }
+
+                    if let Some(new_year) = &req.card_expiry_year {
+                        card_data.expiry_year = Some(new_year.clone());
+                        updated_card_expiry = true;
+                    }
+
+                    if updated_card_expiry {
+                        Some(
+                            create_encrypted_data(
+                                &key_manager_state,
+                                merchant_context.get_merchant_key_store(),
+                                pm_api::PaymentMethodsData::Card(card_data),
+                            )
+                            .await
+                            .change_context(errors::ApiErrorResponse::InternalServerError)
+                            .attach_printable("Unable to encrypt payment method data")
+                            .map(Into::into),
+                        )
+                    } else {
+                        None
+                    }
+                }
+                _ => None,
+            }
+        }
+        None => None,
+    }
+    .transpose()?;
+
+    // Process mandate details when both payment_instrument_id and merchant_connector_ids are present
+    let pm_update = match (
+        &req.payment_instrument_id,
+        &req.merchant_connector_ids.clone(),
+    ) {
+        (Some(payment_instrument_id), Some(merchant_connector_ids)) => {
+            let parsed_mca_ids =
+                MerchantConnectorValidator::parse_comma_separated_ids(merchant_connector_ids)?;
             let mandate_details = payment_method
                 .get_common_mandate_reference()
                 .change_context(errors::ApiErrorResponse::InternalServerError)
                 .attach_printable("Failed to deserialize to Payment Mandate Reference ")?;
 
-            let mca = db
-                .find_by_merchant_connector_account_merchant_id_merchant_connector_id(
-                    &state.into(),
-                    merchant_context.get_merchant_account().get_id(),
-                    merchant_connector_id,
-                    merchant_context.get_merchant_key_store(),
-                )
-                .await
-                .to_not_found_response(
-                    errors::ApiErrorResponse::MerchantConnectorAccountNotFound {
-                        id: merchant_connector_id.get_string_repr().to_string(),
-                    },
-                )?;
+            let mut existing_payments_mandate = mandate_details
+                .payments
+                .clone()
+                .unwrap_or(PaymentsMandateReference(HashMap::new()));
+            let mut existing_payouts_mandate = mandate_details
+                .payouts
+                .clone()
+                .unwrap_or(PayoutsMandateReference(HashMap::new()));
 
-            let updated_connector_mandate_details = match mca.connector_type {
-                enums::ConnectorType::PayoutProcessor => {
-                    // Handle PayoutsMandateReference
-                    let mut existing_payouts_mandate = mandate_details
-                        .payouts
-                        .unwrap_or_else(|| PayoutsMandateReference(HashMap::new()));
+            for merchant_connector_id in parsed_mca_ids {
+                let mca = db
+                    .find_by_merchant_connector_account_merchant_id_merchant_connector_id(
+                        &state.into(),
+                        merchant_context.get_merchant_account().get_id(),
+                        &merchant_connector_id,
+                        merchant_context.get_merchant_key_store(),
+                    )
+                    .await
+                    .to_not_found_response(
+                        errors::ApiErrorResponse::MerchantConnectorAccountNotFound {
+                            id: merchant_connector_id.get_string_repr().to_string(),
+                        },
+                    )?;
 
-                    // Create new payout mandate record
-                    let new_payout_record = PayoutsMandateReferenceRecord {
-                        transfer_method_id: Some(payment_instrument_id.peek().to_string()),
-                    };
+                match mca.connector_type {
+                    enums::ConnectorType::PayoutProcessor => {
+                        // Handle PayoutsMandateReference
+                        let new_payout_record = PayoutsMandateReferenceRecord {
+                            transfer_method_id: Some(payment_instrument_id.peek().to_string()),
+                        };
 
-                    // Check if record exists for this merchant_connector_id
-                    if let Some(existing_record) =
-                        existing_payouts_mandate.0.get_mut(merchant_connector_id)
-                    {
-                        if let Some(transfer_method_id) = &new_payout_record.transfer_method_id {
-                            existing_record.transfer_method_id = Some(transfer_method_id.clone());
+                        // Check if record exists for this merchant_connector_id
+                        if let Some(existing_record) =
+                            existing_payouts_mandate.0.get_mut(&merchant_connector_id)
+                        {
+                            if let Some(transfer_method_id) = &new_payout_record.transfer_method_id
+                            {
+                                existing_record.transfer_method_id =
+                                    Some(transfer_method_id.clone());
+                            }
+                        } else {
+                            // Insert new record in connector_mandate_details
+                            existing_payouts_mandate
+                                .0
+                                .insert(merchant_connector_id.clone(), new_payout_record);
                         }
-                    } else {
-                        // Insert new record in connector_mandate_details
-                        existing_payouts_mandate
-                            .0
-                            .insert(merchant_connector_id.clone(), new_payout_record);
                     }
-
-                    // Create updated CommonMandateReference preserving payments section
-                    CommonMandateReference {
-                        payments: mandate_details.payments,
-                        payouts: Some(existing_payouts_mandate),
+                    _ => {
+                        // Handle PaymentsMandateReference
+                        // Check if record exists for this merchant_connector_id
+                        if let Some(existing_record) =
+                            existing_payments_mandate.0.get_mut(&merchant_connector_id)
+                        {
+                            existing_record.connector_mandate_id =
+                                payment_instrument_id.peek().to_string();
+                        } else {
+                            // Insert new record in connector_mandate_details
+                            existing_payments_mandate.0.insert(
+                                merchant_connector_id.clone(),
+                                PaymentsMandateReferenceRecord {
+                                    connector_mandate_id: payment_instrument_id.peek().to_string(),
+                                    payment_method_type: None,
+                                    original_payment_authorized_amount: None,
+                                    original_payment_authorized_currency: None,
+                                    mandate_metadata: None,
+                                    connector_mandate_status: None,
+                                    connector_mandate_request_reference_id: None,
+                                },
+                            );
+                        }
                     }
                 }
-                _ => {
-                    // Handle PaymentsMandateReference
-                    let mut existing_payments_mandate = mandate_details
-                        .payments
-                        .unwrap_or_else(|| PaymentsMandateReference(HashMap::new()));
-
-                    // Check if record exists for this merchant_connector_id
-                    if let Some(existing_record) =
-                        existing_payments_mandate.0.get_mut(merchant_connector_id)
-                    {
-                        existing_record.connector_mandate_id =
-                            payment_instrument_id.peek().to_string();
-                    } else {
-                        // Insert new record in connector_mandate_details
-                        existing_payments_mandate.0.insert(
-                            merchant_connector_id.clone(),
-                            PaymentsMandateReferenceRecord {
-                                connector_mandate_id: payment_instrument_id.peek().to_string(),
-                                payment_method_type: None,
-                                original_payment_authorized_amount: None,
-                                original_payment_authorized_currency: None,
-                                mandate_metadata: None,
-                                connector_mandate_status: None,
-                                connector_mandate_request_reference_id: None,
-                            },
-                        );
-                    }
+            }
 
-                    // Create updated CommonMandateReference preserving payouts section
-                    CommonMandateReference {
-                        payments: Some(existing_payments_mandate),
-                        payouts: mandate_details.payouts,
-                    }
-                }
+            let updated_connector_mandate_details = CommonMandateReference {
+                payments: if !existing_payments_mandate.0.is_empty() {
+                    Some(existing_payments_mandate)
+                } else {
+                    mandate_details.payments
+                },
+                payouts: if !existing_payouts_mandate.0.is_empty() {
+                    Some(existing_payouts_mandate)
+                } else {
+                    mandate_details.payouts
+                },
             };
 
             let connector_mandate_details_value = updated_connector_mandate_details
@@ -176,19 +230,25 @@ pub async fn update_payment_method_record(
                     errors::ApiErrorResponse::MandateUpdateFailed
                 })?;
 
-            PaymentMethodUpdate::ConnectorNetworkTransactionIdStatusAndMandateDetailsUpdate {
+            PaymentMethodUpdate::PaymentMethodBatchUpdate {
                 connector_mandate_details: Some(pii::SecretSerdeValue::new(
                     connector_mandate_details_value,
                 )),
                 network_transaction_id,
                 status,
+                payment_method_data: updated_payment_method_data.clone(),
             }
         }
         _ => {
-            // Update only network_transaction_id and status
-            PaymentMethodUpdate::NetworkTransactionIdAndStatusUpdate {
-                network_transaction_id,
-                status,
+            if updated_payment_method_data.is_some() {
+                PaymentMethodUpdate::PaymentMethodDataUpdate {
+                    payment_method_data: updated_payment_method_data,
+                }
+            } else {
+                PaymentMethodUpdate::NetworkTransactionIdAndStatusUpdate {
+                    network_transaction_id,
+                    status,
+                }
             }
         }
     };
@@ -217,6 +277,7 @@ pub async fn update_payment_method_record(
             connector_mandate_details: response
                 .connector_mandate_details
                 .map(pii::SecretSerdeValue::new),
+            updated_payment_method_data: Some(updated_card_expiry),
         },
     ))
 }