Bug: [BUG] connector's txn ID exceeding 512 characters are not being stored in DB



### Bug Description

Worldpay's payment flows generate a txn identifier which is quite long - as it's a JWE. This is stored in DB for querying or performing any other processor related operation on the payment. DB schema for storing such long identifiers has been 512 characters. Recently, we started to see a txn ID being returned exceeding 512 characters due to which the application started throwing errors on DB update operations.

### Expected Behavior

DB should respect and store Worldpay IDs exceeding 512 characters.

### Actual Behavior

DB is not storing Worldpay IDs.

### Steps To Reproduce

Route a payment through Worldpay connector.

### Possible solution

For keeping backwards compatibility in mind, and wanting to ensure that query plan related errors are not encountered, the straightforward approach for handling this is to add a new column with required type and migrate any existing data to this column.

### Environment

Are you using hyperswitch hosted version? Yes/No
If yes, please provide the value of the `x-request-id` response header to help 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`): `rustc 1.80.1 (3f5fd8dd4 2024-08-06)`
3. App version (output of `cargo r --features vergen -- --version`): ``


### Have you spent some time checking if this bug has been raised before?

- [X] I checked and didn't find a 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/common_utils/src/types.rs b/crates/common_utils/src/types.rs
index 4e88d5c0fde..0377a3e8183 100644
--- a/crates/common_utils/src/types.rs
+++ b/crates/common_utils/src/types.rs
@@ -1420,7 +1420,7 @@ crate::impl_to_sql_from_sql_json!(BrowserInformation);
 /// In case connector's use an identifier whose length exceeds 128 characters,
 /// the hash value of such identifiers will be stored as connector_transaction_id.
 /// The actual connector's identifier will be stored in a separate column -
-/// connector_transaction_data or something with a similar name.
+/// processor_transaction_data or something with a similar name.
 #[derive(Debug, Clone, PartialEq, Eq, serde::Deserialize, serde::Serialize, AsExpression)]
 #[diesel(sql_type = sql_types::Text)]
 pub enum ConnectorTransactionId {
@@ -1438,7 +1438,7 @@ impl ConnectorTransactionId {
         }
     }
 
-    /// Implementation for forming ConnectorTransactionId and an optional string to be used for connector_transaction_id and connector_transaction_data
+    /// Implementation for forming ConnectorTransactionId and an optional string to be used for connector_transaction_id and processor_transaction_data
     pub fn form_id_and_data(src: String) -> (Self, Option<String>) {
         let txn_id = Self::from(src.clone());
         match txn_id {
@@ -1456,10 +1456,10 @@ impl ConnectorTransactionId {
             (Self::TxnId(id), _) => Ok(id),
             (Self::HashedData(_), Some(id)) => Ok(id),
             (Self::HashedData(id), None) => Err(report!(ValidationError::InvalidValue {
-                message: "connector_transaction_data is empty for HashedData variant".to_string(),
+                message: "processor_transaction_data is empty for HashedData variant".to_string(),
             })
             .attach_printable(format!(
-                "connector_transaction_data is empty for connector_transaction_id {}",
+                "processor_transaction_data is empty for connector_transaction_id {}",
                 id
             ))),
         }
diff --git a/crates/diesel_models/src/capture.rs b/crates/diesel_models/src/capture.rs
index d6272c34060..367382b98bf 100644
--- a/crates/diesel_models/src/capture.rs
+++ b/crates/diesel_models/src/capture.rs
@@ -30,7 +30,9 @@ pub struct Capture {
     pub capture_sequence: i16,
     // reference to the capture at connector side
     pub connector_response_reference_id: Option<String>,
+    /// INFO: This field is deprecated and replaced by processor_capture_data
     pub connector_capture_data: Option<String>,
+    pub processor_capture_data: Option<String>,
 }
 
 #[derive(Clone, Debug, Insertable, router_derive::DebugAsDisplay, Serialize, Deserialize)]
@@ -55,7 +57,9 @@ pub struct CaptureNew {
     pub connector_capture_id: Option<ConnectorTransactionId>,
     pub capture_sequence: i16,
     pub connector_response_reference_id: Option<String>,
+    /// INFO: This field is deprecated and replaced by processor_capture_data
     pub connector_capture_data: Option<String>,
+    pub processor_capture_data: Option<String>,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -64,7 +68,7 @@ pub enum CaptureUpdate {
         status: storage_enums::CaptureStatus,
         connector_capture_id: Option<ConnectorTransactionId>,
         connector_response_reference_id: Option<String>,
-        connector_capture_data: Option<String>,
+        processor_capture_data: Option<String>,
     },
     ErrorUpdate {
         status: storage_enums::CaptureStatus,
@@ -84,7 +88,7 @@ pub struct CaptureUpdateInternal {
     pub modified_at: Option<PrimitiveDateTime>,
     pub connector_capture_id: Option<ConnectorTransactionId>,
     pub connector_response_reference_id: Option<String>,
-    pub connector_capture_data: Option<String>,
+    pub processor_capture_data: Option<String>,
 }
 
 impl CaptureUpdate {
@@ -97,7 +101,7 @@ impl CaptureUpdate {
             modified_at: _,
             connector_capture_id,
             connector_response_reference_id,
-            connector_capture_data,
+            processor_capture_data,
         } = self.into();
         Capture {
             status: status.unwrap_or(source.status),
@@ -108,7 +112,7 @@ impl CaptureUpdate {
             connector_capture_id: connector_capture_id.or(source.connector_capture_id),
             connector_response_reference_id: connector_response_reference_id
                 .or(source.connector_response_reference_id),
-            connector_capture_data: connector_capture_data.or(source.connector_capture_data),
+            processor_capture_data: processor_capture_data.or(source.processor_capture_data),
             ..source
         }
     }
@@ -122,13 +126,13 @@ impl From<CaptureUpdate> for CaptureUpdateInternal {
                 status,
                 connector_capture_id: connector_transaction_id,
                 connector_response_reference_id,
-                connector_capture_data,
+                processor_capture_data,
             } => Self {
                 status: Some(status),
                 connector_capture_id: connector_transaction_id,
                 modified_at: now,
                 connector_response_reference_id,
-                connector_capture_data,
+                processor_capture_data,
                 ..Self::default()
             },
             CaptureUpdate::ErrorUpdate {
diff --git a/crates/diesel_models/src/kv.rs b/crates/diesel_models/src/kv.rs
index 94dcf00f755..54e60891532 100644
--- a/crates/diesel_models/src/kv.rs
+++ b/crates/diesel_models/src/kv.rs
@@ -227,7 +227,7 @@ pub enum Insertable {
 pub enum Updateable {
     PaymentIntentUpdate(Box<PaymentIntentUpdateMems>),
     PaymentAttemptUpdate(Box<PaymentAttemptUpdateMems>),
-    RefundUpdate(RefundUpdateMems),
+    RefundUpdate(Box<RefundUpdateMems>),
     CustomerUpdate(CustomerUpdateMems),
     AddressUpdate(Box<AddressUpdateMems>),
     PayoutsUpdate(PayoutsUpdateMems),
diff --git a/crates/diesel_models/src/payment_attempt.rs b/crates/diesel_models/src/payment_attempt.rs
index eab95f81316..b3b3f44f3c9 100644
--- a/crates/diesel_models/src/payment_attempt.rs
+++ b/crates/diesel_models/src/payment_attempt.rs
@@ -171,8 +171,10 @@ pub struct PaymentAttempt {
     pub card_network: Option<String>,
     pub shipping_cost: Option<MinorUnit>,
     pub order_tax_amount: Option<MinorUnit>,
+    /// INFO: This field is deprecated and replaced by processor_transaction_data
     pub connector_transaction_data: Option<String>,
     pub connector_mandate_detail: Option<ConnectorMandateReferenceId>,
+    pub processor_transaction_data: Option<String>,
     pub card_discovery: Option<storage_enums::CardDiscovery>,
     pub charges: Option<common_types::payments::ConnectorChargeResponseData>,
 }
@@ -183,7 +185,7 @@ impl ConnectorTransactionIdTrait for PaymentAttempt {
         match self
             .connector_transaction_id
             .as_ref()
-            .map(|txn_id| txn_id.get_txn_id(self.connector_transaction_data.as_ref()))
+            .map(|txn_id| txn_id.get_txn_id(self.processor_transaction_data.as_ref()))
             .transpose()
         {
             Ok(txn_id) => txn_id,
@@ -871,8 +873,8 @@ pub struct PaymentAttemptUpdateInternal {
     pub card_network: Option<String>,
     pub shipping_cost: Option<MinorUnit>,
     pub order_tax_amount: Option<MinorUnit>,
-    pub connector_transaction_data: Option<String>,
     pub connector_mandate_detail: Option<ConnectorMandateReferenceId>,
+    pub processor_transaction_data: Option<String>,
     pub card_discovery: Option<common_enums::CardDiscovery>,
     pub charges: Option<common_types::payments::ConnectorChargeResponseData>,
 }
@@ -1055,7 +1057,7 @@ impl PaymentAttemptUpdate {
             card_network,
             shipping_cost,
             order_tax_amount,
-            connector_transaction_data,
+            processor_transaction_data,
             connector_mandate_detail,
             card_discovery,
             charges,
@@ -1113,8 +1115,8 @@ impl PaymentAttemptUpdate {
             card_network: card_network.or(source.card_network),
             shipping_cost: shipping_cost.or(source.shipping_cost),
             order_tax_amount: order_tax_amount.or(source.order_tax_amount),
-            connector_transaction_data: connector_transaction_data
-                .or(source.connector_transaction_data),
+            processor_transaction_data: processor_transaction_data
+                .or(source.processor_transaction_data),
             connector_mandate_detail: connector_mandate_detail.or(source.connector_mandate_detail),
             card_discovery: card_discovery.or(source.card_discovery),
             charges: charges.or(source.charges),
@@ -1483,7 +1485,7 @@ impl From<PaymentAttemptUpdate> for PaymentAttemptUpdateInternal {
         //         card_network: None,
         //         shipping_cost: None,
         //         order_tax_amount: None,
-        //         connector_transaction_data: None,
+        //         processor_transaction_data: None,
         //         connector_mandate_detail,
         //     },
         //     PaymentAttemptUpdate::ConnectorMandateDetailUpdate {
@@ -2167,7 +2169,7 @@ impl From<PaymentAttemptUpdate> for PaymentAttemptUpdateInternal {
                 card_network: None,
                 shipping_cost: None,
                 order_tax_amount: None,
-                connector_transaction_data: None,
+                processor_transaction_data: None,
                 connector_mandate_detail: None,
                 card_discovery: None,
                 charges: None,
@@ -2224,7 +2226,7 @@ impl From<PaymentAttemptUpdate> for PaymentAttemptUpdateInternal {
                 card_network: None,
                 shipping_cost: None,
                 order_tax_amount: None,
-                connector_transaction_data: None,
+                processor_transaction_data: None,
                 connector_mandate_detail: None,
                 card_discovery: None,
                 charges: None,
@@ -2313,7 +2315,7 @@ impl From<PaymentAttemptUpdate> for PaymentAttemptUpdateInternal {
                 card_network: None,
                 shipping_cost,
                 order_tax_amount,
-                connector_transaction_data: None,
+                processor_transaction_data: None,
                 connector_mandate_detail,
                 card_discovery,
                 charges: None,
@@ -2371,7 +2373,7 @@ impl From<PaymentAttemptUpdate> for PaymentAttemptUpdateInternal {
                 card_network: None,
                 shipping_cost: None,
                 order_tax_amount: None,
-                connector_transaction_data: None,
+                processor_transaction_data: None,
                 connector_mandate_detail: None,
                 card_discovery: None,
                 charges: None,
@@ -2430,7 +2432,7 @@ impl From<PaymentAttemptUpdate> for PaymentAttemptUpdateInternal {
                 card_network: None,
                 shipping_cost: None,
                 order_tax_amount: None,
-                connector_transaction_data: None,
+                processor_transaction_data: None,
                 connector_mandate_detail: None,
                 card_discovery: None,
                 charges: None,
@@ -2489,7 +2491,7 @@ impl From<PaymentAttemptUpdate> for PaymentAttemptUpdateInternal {
                 card_network: None,
                 shipping_cost: None,
                 order_tax_amount: None,
-                connector_transaction_data: None,
+                processor_transaction_data: None,
                 connector_mandate_detail: None,
                 card_discovery: None,
                 charges: None,
@@ -2546,7 +2548,7 @@ impl From<PaymentAttemptUpdate> for PaymentAttemptUpdateInternal {
                 card_network: None,
                 shipping_cost: None,
                 order_tax_amount: None,
-                connector_transaction_data: None,
+                processor_transaction_data: None,
                 connector_mandate_detail,
                 card_discovery: None,
                 charges: None,
@@ -2603,7 +2605,7 @@ impl From<PaymentAttemptUpdate> for PaymentAttemptUpdateInternal {
                 card_network: None,
                 shipping_cost: None,
                 order_tax_amount: None,
-                connector_transaction_data: None,
+                processor_transaction_data: None,
                 connector_mandate_detail: None,
                 card_discovery: None,
                 charges: None,
@@ -2631,7 +2633,7 @@ impl From<PaymentAttemptUpdate> for PaymentAttemptUpdateInternal {
                 connector_mandate_detail,
                 charges,
             } => {
-                let (connector_transaction_id, connector_transaction_data) =
+                let (connector_transaction_id, processor_transaction_data) =
                     connector_transaction_id
                         .map(ConnectorTransactionId::form_id_and_data)
                         .map(|(txn_id, txn_data)| (Some(txn_id), txn_data))
@@ -2657,7 +2659,9 @@ impl From<PaymentAttemptUpdate> for PaymentAttemptUpdateInternal {
                     unified_code,
                     unified_message,
                     payment_method_data,
-                    connector_transaction_data,
+                    processor_transaction_data,
+                    connector_mandate_detail,
+                    charges,
                     amount: None,
                     net_amount: None,
                     currency: None,
@@ -2686,9 +2690,7 @@ impl From<PaymentAttemptUpdate> for PaymentAttemptUpdateInternal {
                     card_network: None,
                     shipping_cost: None,
                     order_tax_amount: None,
-                    connector_mandate_detail,
                     card_discovery: None,
-                    charges,
                 }
             }
             PaymentAttemptUpdate::ErrorUpdate {
@@ -2705,7 +2707,7 @@ impl From<PaymentAttemptUpdate> for PaymentAttemptUpdateInternal {
                 payment_method_data,
                 authentication_type,
             } => {
-                let (connector_transaction_id, connector_transaction_data) =
+                let (connector_transaction_id, processor_transaction_data) =
                     connector_transaction_id
                         .map(ConnectorTransactionId::form_id_and_data)
                         .map(|(txn_id, txn_data)| (Some(txn_id), txn_data))
@@ -2724,7 +2726,7 @@ impl From<PaymentAttemptUpdate> for PaymentAttemptUpdateInternal {
                     connector_transaction_id,
                     payment_method_data,
                     authentication_type,
-                    connector_transaction_data,
+                    processor_transaction_data,
                     amount: None,
                     net_amount: None,
                     currency: None,
@@ -2814,7 +2816,7 @@ impl From<PaymentAttemptUpdate> for PaymentAttemptUpdateInternal {
                 card_network: None,
                 shipping_cost: None,
                 order_tax_amount: None,
-                connector_transaction_data: None,
+                processor_transaction_data: None,
                 connector_mandate_detail: None,
                 card_discovery: None,
                 charges: None,
@@ -2877,7 +2879,7 @@ impl From<PaymentAttemptUpdate> for PaymentAttemptUpdateInternal {
                 card_network: None,
                 shipping_cost: None,
                 order_tax_amount: None,
-                connector_transaction_data: None,
+                processor_transaction_data: None,
                 connector_mandate_detail: None,
                 card_discovery: None,
                 charges: None,
@@ -2893,7 +2895,7 @@ impl From<PaymentAttemptUpdate> for PaymentAttemptUpdateInternal {
                 connector_response_reference_id,
                 updated_by,
             } => {
-                let (connector_transaction_id, connector_transaction_data) =
+                let (connector_transaction_id, processor_transaction_data) =
                     connector_transaction_id
                         .map(ConnectorTransactionId::form_id_and_data)
                         .map(|(txn_id, txn_data)| (Some(txn_id), txn_data))
@@ -2909,7 +2911,7 @@ impl From<PaymentAttemptUpdate> for PaymentAttemptUpdateInternal {
                     error_reason,
                     connector_response_reference_id,
                     updated_by,
-                    connector_transaction_data,
+                    processor_transaction_data,
                     amount: None,
                     net_amount: None,
                     currency: None,
@@ -2962,7 +2964,7 @@ impl From<PaymentAttemptUpdate> for PaymentAttemptUpdateInternal {
                 connector_response_reference_id,
                 updated_by,
             } => {
-                let (connector_transaction_id, connector_transaction_data) =
+                let (connector_transaction_id, processor_transaction_data) =
                     connector_transaction_id
                         .map(ConnectorTransactionId::form_id_and_data)
                         .map(|(txn_id, txn_data)| (Some(txn_id), txn_data))
@@ -2976,7 +2978,7 @@ impl From<PaymentAttemptUpdate> for PaymentAttemptUpdateInternal {
                     connector_transaction_id,
                     connector_response_reference_id,
                     updated_by,
-                    connector_transaction_data,
+                    processor_transaction_data,
                     amount: None,
                     net_amount: None,
                     currency: None,
@@ -3075,7 +3077,7 @@ impl From<PaymentAttemptUpdate> for PaymentAttemptUpdateInternal {
                 card_network: None,
                 shipping_cost: None,
                 order_tax_amount: None,
-                connector_transaction_data: None,
+                processor_transaction_data: None,
                 connector_mandate_detail: None,
                 card_discovery: None,
                 charges: None,
@@ -3133,7 +3135,7 @@ impl From<PaymentAttemptUpdate> for PaymentAttemptUpdateInternal {
                 card_network: None,
                 shipping_cost: None,
                 order_tax_amount: None,
-                connector_transaction_data: None,
+                processor_transaction_data: None,
                 connector_mandate_detail: None,
                 card_discovery: None,
                 charges: None,
@@ -3146,7 +3148,7 @@ impl From<PaymentAttemptUpdate> for PaymentAttemptUpdateInternal {
                 updated_by,
                 charges,
             } => {
-                let (connector_transaction_id, connector_transaction_data) =
+                let (connector_transaction_id, processor_transaction_data) =
                     connector_transaction_id
                         .map(ConnectorTransactionId::form_id_and_data)
                         .map(|(txn_id, txn_data)| (Some(txn_id), txn_data))
@@ -3158,7 +3160,8 @@ impl From<PaymentAttemptUpdate> for PaymentAttemptUpdateInternal {
                     connector: connector.map(Some),
                     modified_at: common_utils::date_time::now(),
                     updated_by,
-                    connector_transaction_data,
+                    processor_transaction_data,
+                    charges,
                     amount: None,
                     net_amount: None,
                     currency: None,
@@ -3203,7 +3206,6 @@ impl From<PaymentAttemptUpdate> for PaymentAttemptUpdateInternal {
                     order_tax_amount: None,
                     connector_mandate_detail: None,
                     card_discovery: None,
-                    charges,
                 }
             }
             PaymentAttemptUpdate::IncrementalAuthorizationAmountUpdate {
@@ -3258,7 +3260,7 @@ impl From<PaymentAttemptUpdate> for PaymentAttemptUpdateInternal {
                 card_network: None,
                 shipping_cost: None,
                 order_tax_amount: None,
-                connector_transaction_data: None,
+                processor_transaction_data: None,
                 connector_mandate_detail: None,
                 card_discovery: None,
                 charges: None,
@@ -3318,7 +3320,7 @@ impl From<PaymentAttemptUpdate> for PaymentAttemptUpdateInternal {
                 card_network: None,
                 shipping_cost: None,
                 order_tax_amount: None,
-                connector_transaction_data: None,
+                processor_transaction_data: None,
                 connector_mandate_detail: None,
                 card_discovery: None,
                 charges: None,
@@ -3333,7 +3335,7 @@ impl From<PaymentAttemptUpdate> for PaymentAttemptUpdateInternal {
                 unified_message,
                 connector_transaction_id,
             } => {
-                let (connector_transaction_id, connector_transaction_data) =
+                let (connector_transaction_id, processor_transaction_data) =
                     connector_transaction_id
                         .map(ConnectorTransactionId::form_id_and_data)
                         .map(|(txn_id, txn_data)| (Some(txn_id), txn_data))
@@ -3348,7 +3350,7 @@ impl From<PaymentAttemptUpdate> for PaymentAttemptUpdateInternal {
                     unified_code: unified_code.map(Some),
                     unified_message: unified_message.map(Some),
                     connector_transaction_id,
-                    connector_transaction_data,
+                    processor_transaction_data,
                     amount: None,
                     net_amount: None,
                     currency: None,
@@ -3445,7 +3447,7 @@ impl From<PaymentAttemptUpdate> for PaymentAttemptUpdateInternal {
                 card_network: None,
                 shipping_cost: None,
                 order_tax_amount: None,
-                connector_transaction_data: None,
+                processor_transaction_data: None,
                 connector_mandate_detail: None,
                 card_discovery: None,
                 charges: None,
diff --git a/crates/diesel_models/src/query/capture.rs b/crates/diesel_models/src/query/capture.rs
index 7a32e410961..3f02948d2a4 100644
--- a/crates/diesel_models/src/query/capture.rs
+++ b/crates/diesel_models/src/query/capture.rs
@@ -74,7 +74,7 @@ impl ConnectorTransactionIdTrait for Capture {
         match self
             .connector_capture_id
             .as_ref()
-            .map(|capture_id| capture_id.get_txn_id(self.connector_capture_data.as_ref()))
+            .map(|capture_id| capture_id.get_txn_id(self.processor_capture_data.as_ref()))
             .transpose()
         {
             Ok(capture_id) => capture_id,
diff --git a/crates/diesel_models/src/refund.rs b/crates/diesel_models/src/refund.rs
index b929481d64d..f19c3252c60 100644
--- a/crates/diesel_models/src/refund.rs
+++ b/crates/diesel_models/src/refund.rs
@@ -51,11 +51,15 @@ pub struct Refund {
     pub merchant_connector_id: Option<common_utils::id_type::MerchantConnectorAccountId>,
     pub charges: Option<ChargeRefunds>,
     pub organization_id: common_utils::id_type::OrganizationId,
+    /// INFO: This field is deprecated and replaced by processor_refund_data
     pub connector_refund_data: Option<String>,
+    /// INFO: This field is deprecated and replaced by processor_transaction_data
     pub connector_transaction_data: Option<String>,
     pub split_refunds: Option<common_types::refunds::SplitRefund>,
     pub unified_code: Option<String>,
     pub unified_message: Option<String>,
+    pub processor_refund_data: Option<String>,
+    pub processor_transaction_data: Option<String>,
 }
 
 #[derive(
@@ -99,9 +103,9 @@ pub struct RefundNew {
     pub merchant_connector_id: Option<common_utils::id_type::MerchantConnectorAccountId>,
     pub charges: Option<ChargeRefunds>,
     pub organization_id: common_utils::id_type::OrganizationId,
-    pub connector_refund_data: Option<String>,
-    pub connector_transaction_data: Option<String>,
     pub split_refunds: Option<common_types::refunds::SplitRefund>,
+    pub processor_refund_data: Option<String>,
+    pub processor_transaction_data: Option<String>,
 }
 
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
@@ -113,7 +117,7 @@ pub enum RefundUpdate {
         refund_error_message: Option<String>,
         refund_arn: String,
         updated_by: String,
-        connector_refund_data: Option<String>,
+        processor_refund_data: Option<String>,
     },
     MetadataAndReasonUpdate {
         metadata: Option<pii::SecretSerdeValue>,
@@ -125,7 +129,7 @@ pub enum RefundUpdate {
         sent_to_gateway: bool,
         refund_status: storage_enums::RefundStatus,
         updated_by: String,
-        connector_refund_data: Option<String>,
+        processor_refund_data: Option<String>,
     },
     ErrorUpdate {
         refund_status: Option<storage_enums::RefundStatus>,
@@ -133,7 +137,7 @@ pub enum RefundUpdate {
         refund_error_code: Option<String>,
         updated_by: String,
         connector_refund_id: Option<ConnectorTransactionId>,
-        connector_refund_data: Option<String>,
+        processor_refund_data: Option<String>,
         unified_code: Option<String>,
         unified_message: Option<String>,
     },
@@ -158,7 +162,7 @@ pub struct RefundUpdateInternal {
     refund_error_code: Option<String>,
     updated_by: String,
     modified_at: PrimitiveDateTime,
-    connector_refund_data: Option<String>,
+    processor_refund_data: Option<String>,
     unified_code: Option<String>,
     unified_message: Option<String>,
 }
@@ -176,7 +180,7 @@ impl RefundUpdateInternal {
             refund_error_code: self.refund_error_code,
             updated_by: self.updated_by,
             modified_at: self.modified_at,
-            connector_refund_data: self.connector_refund_data,
+            processor_refund_data: self.processor_refund_data,
             unified_code: self.unified_code,
             unified_message: self.unified_message,
             ..source
@@ -194,7 +198,7 @@ impl From<RefundUpdate> for RefundUpdateInternal {
                 refund_error_message,
                 refund_arn,
                 updated_by,
-                connector_refund_data,
+                processor_refund_data,
             } => Self {
                 connector_refund_id: Some(connector_refund_id),
                 refund_status: Some(refund_status),
@@ -202,7 +206,7 @@ impl From<RefundUpdate> for RefundUpdateInternal {
                 refund_error_message,
                 refund_arn: Some(refund_arn),
                 updated_by,
-                connector_refund_data,
+                processor_refund_data,
                 metadata: None,
                 refund_reason: None,
                 refund_error_code: None,
@@ -225,7 +229,7 @@ impl From<RefundUpdate> for RefundUpdateInternal {
                 refund_arn: None,
                 refund_error_code: None,
                 modified_at: common_utils::date_time::now(),
-                connector_refund_data: None,
+                processor_refund_data: None,
                 unified_code: None,
                 unified_message: None,
             },
@@ -234,13 +238,13 @@ impl From<RefundUpdate> for RefundUpdateInternal {
                 sent_to_gateway,
                 refund_status,
                 updated_by,
-                connector_refund_data,
+                processor_refund_data,
             } => Self {
                 connector_refund_id,
                 sent_to_gateway: Some(sent_to_gateway),
                 refund_status: Some(refund_status),
                 updated_by,
-                connector_refund_data,
+                processor_refund_data,
                 refund_error_message: None,
                 refund_arn: None,
                 metadata: None,
@@ -258,14 +262,14 @@ impl From<RefundUpdate> for RefundUpdateInternal {
                 unified_message,
                 updated_by,
                 connector_refund_id,
-                connector_refund_data,
+                processor_refund_data,
             } => Self {
                 refund_status,
                 refund_error_message,
                 refund_error_code,
                 updated_by,
                 connector_refund_id,
-                connector_refund_data,
+                processor_refund_data,
                 sent_to_gateway: None,
                 refund_arn: None,
                 metadata: None,
@@ -290,7 +294,7 @@ impl From<RefundUpdate> for RefundUpdateInternal {
                 metadata: None,
                 refund_reason: None,
                 modified_at: common_utils::date_time::now(),
-                connector_refund_data: None,
+                processor_refund_data: None,
                 unified_code: None,
                 unified_message: None,
             },
@@ -311,7 +315,7 @@ impl RefundUpdate {
             refund_error_code,
             updated_by,
             modified_at: _,
-            connector_refund_data,
+            processor_refund_data,
             unified_code,
             unified_message,
         } = self.into();
@@ -326,7 +330,7 @@ impl RefundUpdate {
             refund_reason: refund_reason.or(source.refund_reason),
             updated_by,
             modified_at: common_utils::date_time::now(),
-            connector_refund_data: connector_refund_data.or(source.connector_refund_data),
+            processor_refund_data: processor_refund_data.or(source.processor_refund_data),
             unified_code: unified_code.or(source.unified_code),
             unified_message: unified_message.or(source.unified_message),
             ..source
@@ -340,7 +344,7 @@ pub struct RefundCoreWorkflow {
     pub connector_transaction_id: ConnectorTransactionId,
     pub merchant_id: common_utils::id_type::MerchantId,
     pub payment_id: common_utils::id_type::PaymentId,
-    pub connector_transaction_data: Option<String>,
+    pub processor_transaction_data: Option<String>,
 }
 
 #[cfg(feature = "v1")]
@@ -358,7 +362,7 @@ impl ConnectorTransactionIdTrait for Refund {
         match self
             .connector_refund_id
             .as_ref()
-            .map(|refund_id| refund_id.get_txn_id(self.connector_refund_data.as_ref()))
+            .map(|refund_id| refund_id.get_txn_id(self.processor_refund_data.as_ref()))
             .transpose()
         {
             Ok(refund_id) => refund_id,
@@ -374,7 +378,7 @@ impl ConnectorTransactionIdTrait for Refund {
     fn get_connector_transaction_id(&self) -> &String {
         match self
             .connector_transaction_id
-            .get_txn_id(self.connector_transaction_data.as_ref())
+            .get_txn_id(self.processor_transaction_data.as_ref())
         {
             Ok(txn_id) => txn_id,
 
@@ -418,6 +422,7 @@ mod tests {
     "connector_transaction_data": null
     "unified_code": null,
     "unified_message": null,
+    "processor_transaction_data": null,
 }"#;
         let deserialized = serde_json::from_str::<super::Refund>(serialized_refund);
 
diff --git a/crates/diesel_models/src/schema.rs b/crates/diesel_models/src/schema.rs
index bbde92ea040..2e8219c5ab4 100644
--- a/crates/diesel_models/src/schema.rs
+++ b/crates/diesel_models/src/schema.rs
@@ -271,6 +271,7 @@ diesel::table! {
         connector_response_reference_id -> Nullable<Varchar>,
         #[max_length = 512]
         connector_capture_data -> Nullable<Varchar>,
+        processor_capture_data -> Nullable<Text>,
     }
 }
 
@@ -909,6 +910,7 @@ diesel::table! {
         #[max_length = 512]
         connector_transaction_data -> Nullable<Varchar>,
         connector_mandate_detail -> Nullable<Jsonb>,
+        processor_transaction_data -> Nullable<Text>,
         card_discovery -> Nullable<CardDiscovery>,
         charges -> Nullable<Jsonb>,
     }
@@ -1261,6 +1263,8 @@ diesel::table! {
         unified_code -> Nullable<Varchar>,
         #[max_length = 1024]
         unified_message -> Nullable<Varchar>,
+        processor_refund_data -> Nullable<Text>,
+        processor_transaction_data -> Nullable<Text>,
     }
 }
 
diff --git a/crates/diesel_models/src/schema_v2.rs b/crates/diesel_models/src/schema_v2.rs
index d17d4f8478a..ab2195526f5 100644
--- a/crates/diesel_models/src/schema_v2.rs
+++ b/crates/diesel_models/src/schema_v2.rs
@@ -278,8 +278,7 @@ diesel::table! {
         capture_sequence -> Int2,
         #[max_length = 128]
         connector_response_reference_id -> Nullable<Varchar>,
-        #[max_length = 512]
-        connector_capture_data -> Nullable<Varchar>,
+        processor_capture_data -> Nullable<Text>,
     }
 }
 
@@ -872,8 +871,7 @@ diesel::table! {
         tax_on_surcharge -> Nullable<Int8>,
         payment_method_billing_address -> Nullable<Bytea>,
         redirection_data -> Nullable<Jsonb>,
-        #[max_length = 512]
-        connector_payment_data -> Nullable<Varchar>,
+        connector_payment_data -> Nullable<Text>,
         connector_token_details -> Nullable<Jsonb>,
         #[max_length = 64]
         id -> Varchar,
@@ -1198,15 +1196,13 @@ diesel::table! {
         charges -> Nullable<Jsonb>,
         #[max_length = 32]
         organization_id -> Varchar,
-        #[max_length = 512]
-        connector_refund_data -> Nullable<Varchar>,
-        #[max_length = 512]
-        connector_transaction_data -> Nullable<Varchar>,
         split_refunds -> Nullable<Jsonb>,
         #[max_length = 255]
         unified_code -> Nullable<Varchar>,
         #[max_length = 1024]
         unified_message -> Nullable<Varchar>,
+        processor_refund_data -> Nullable<Text>,
+        processor_transaction_data -> Nullable<Text>,
     }
 }
 
diff --git a/crates/diesel_models/src/user/sample_data.rs b/crates/diesel_models/src/user/sample_data.rs
index c063545d81f..27085424083 100644
--- a/crates/diesel_models/src/user/sample_data.rs
+++ b/crates/diesel_models/src/user/sample_data.rs
@@ -201,7 +201,7 @@ pub struct PaymentAttemptBatchNew {
     pub organization_id: common_utils::id_type::OrganizationId,
     pub shipping_cost: Option<MinorUnit>,
     pub order_tax_amount: Option<MinorUnit>,
-    pub connector_transaction_data: Option<String>,
+    pub processor_transaction_data: Option<String>,
     pub connector_mandate_detail: Option<ConnectorMandateReferenceId>,
     pub card_discovery: Option<common_enums::CardDiscovery>,
 }
diff --git a/crates/hyperswitch_connectors/src/connectors/worldpay/requests.rs b/crates/hyperswitch_connectors/src/connectors/worldpay/requests.rs
index 6e19c20edab..a43ffb7cfa1 100644
--- a/crates/hyperswitch_connectors/src/connectors/worldpay/requests.rs
+++ b/crates/hyperswitch_connectors/src/connectors/worldpay/requests.rs
@@ -149,14 +149,12 @@ pub struct ExpiryDate {
 #[derive(Clone, Debug, PartialEq, Default, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct BillingAddress {
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub address1: Option<Secret<String>>,
+    pub address1: Secret<String>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub address2: Option<Secret<String>>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub address3: Option<Secret<String>>,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub city: Option<String>,
+    pub city: String,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub state: Option<Secret<String>>,
     pub postal_code: Secret<String>,
diff --git a/crates/hyperswitch_connectors/src/connectors/worldpay/transformers.rs b/crates/hyperswitch_connectors/src/connectors/worldpay/transformers.rs
index 28cbca418a3..65c884601de 100644
--- a/crates/hyperswitch_connectors/src/connectors/worldpay/transformers.rs
+++ b/crates/hyperswitch_connectors/src/connectors/worldpay/transformers.rs
@@ -90,10 +90,16 @@ fn fetch_payment_instrument(
                 billing_address.and_then(|addr| addr.address.clone())
             {
                 Some(BillingAddress {
-                    address1: address.line1,
+                    address1: address.line1.get_required_value("line1").change_context(
+                        errors::ConnectorError::MissingRequiredField {
+                            field_name: "line1",
+                        },
+                    )?,
                     address2: address.line2,
                     address3: address.line3,
-                    city: address.city,
+                    city: address.city.get_required_value("city").change_context(
+                        errors::ConnectorError::MissingRequiredField { field_name: "city" },
+                    )?,
                     state: address.state,
                     postal_code: address.zip.get_required_value("zip").change_context(
                         errors::ConnectorError::MissingRequiredField { field_name: "zip" },
diff --git a/crates/hyperswitch_domain_models/src/payments/payment_attempt.rs b/crates/hyperswitch_domain_models/src/payments/payment_attempt.rs
index 361973daf94..d2eb8e44e87 100644
--- a/crates/hyperswitch_domain_models/src/payments/payment_attempt.rs
+++ b/crates/hyperswitch_domain_models/src/payments/payment_attempt.rs
@@ -1503,7 +1503,7 @@ impl behaviour::Conversion for PaymentAttempt {
             .and_then(|card| card.get("card_network"))
             .and_then(|network| network.as_str())
             .map(|network| network.to_string());
-        let (connector_transaction_id, connector_transaction_data) = self
+        let (connector_transaction_id, processor_transaction_data) = self
             .connector_transaction_id
             .map(ConnectorTransactionId::form_id_and_data)
             .map(|(txn_id, txn_data)| (Some(txn_id), txn_data))
@@ -1570,12 +1570,14 @@ impl behaviour::Conversion for PaymentAttempt {
             profile_id: self.profile_id,
             organization_id: self.organization_id,
             card_network,
-            connector_transaction_data,
             order_tax_amount: self.net_amount.get_order_tax_amount(),
             shipping_cost: self.net_amount.get_shipping_cost(),
             connector_mandate_detail: self.connector_mandate_detail,
+            processor_transaction_data,
             card_discovery: self.card_discovery,
             charges: self.charges,
+            // Below fields are deprecated. Please add any new fields above this line.
+            connector_transaction_data: None,
         })
     }
 
diff --git a/crates/router/src/configs/defaults/payment_connector_required_fields.rs b/crates/router/src/configs/defaults/payment_connector_required_fields.rs
index c5b9b9a0430..a9f5ae59493 100644
--- a/crates/router/src/configs/defaults/payment_connector_required_fields.rs
+++ b/crates/router/src/configs/defaults/payment_connector_required_fields.rs
@@ -13514,5 +13514,14 @@ pub fn get_worldpay_billing_required_fields() -> HashMap<String, RequiredFieldIn
                 value: None,
             },
         ),
+        (
+            "billing.address.zip".to_string(),
+            RequiredFieldInfo {
+                required_field: "payment_method_data.billing.address.zip".to_string(),
+                display_name: "zip".to_string(),
+                field_type: enums::FieldType::UserAddressPincode,
+                value: None,
+            },
+        ),
     ])
 }
diff --git a/crates/router/src/core/payments/operations/payment_response.rs b/crates/router/src/core/payments/operations/payment_response.rs
index 4616be81a5f..25ae8c84c39 100644
--- a/crates/router/src/core/payments/operations/payment_response.rs
+++ b/crates/router/src/core/payments/operations/payment_response.rs
@@ -1663,7 +1663,7 @@ async fn payment_response_update_tracker<F: Clone, T: types::Capturable>(
                                 .multiple_capture_data
                             {
                                 Some(multiple_capture_data) => {
-                                    let (connector_capture_id, connector_capture_data) =
+                                    let (connector_capture_id, processor_capture_data) =
                                         match resource_id {
                                             types::ResponseId::NoResponseId => (None, None),
                                             types::ResponseId::ConnectorTransactionId(id)
@@ -1679,7 +1679,7 @@ async fn payment_response_update_tracker<F: Clone, T: types::Capturable>(
                                         )?,
                                         connector_capture_id: connector_capture_id.clone(),
                                         connector_response_reference_id,
-                                        connector_capture_data: connector_capture_data.clone(),
+                                        processor_capture_data: processor_capture_data.clone(),
                                     };
                                     let capture_update_list = vec![(
                                         multiple_capture_data.get_latest_capture().clone(),
diff --git a/crates/router/src/core/payments/transformers.rs b/crates/router/src/core/payments/transformers.rs
index e2967e4ba59..5659fef6296 100644
--- a/crates/router/src/core/payments/transformers.rs
+++ b/crates/router/src/core/payments/transformers.rs
@@ -3839,7 +3839,7 @@ impl ForeignTryFrom<types::CaptureSyncResponse> for storage::CaptureUpdate {
                 connector_response_reference_id,
                 ..
             } => {
-                let (connector_capture_id, connector_capture_data) = match resource_id {
+                let (connector_capture_id, processor_capture_data) = match resource_id {
                     types::ResponseId::EncodedData(_) | types::ResponseId::NoResponseId => {
                         (None, None)
                     }
@@ -3853,7 +3853,7 @@ impl ForeignTryFrom<types::CaptureSyncResponse> for storage::CaptureUpdate {
                     status: enums::CaptureStatus::foreign_try_from(status)?,
                     connector_capture_id,
                     connector_response_reference_id,
-                    connector_capture_data,
+                    processor_capture_data,
                 })
             }
             types::CaptureSyncResponse::Error {
diff --git a/crates/router/src/core/refunds.rs b/crates/router/src/core/refunds.rs
index 4d8cc195693..938bf1a8369 100644
--- a/crates/router/src/core/refunds.rs
+++ b/crates/router/src/core/refunds.rs
@@ -235,7 +235,7 @@ pub async fn trigger_refund_to_gateway(
                             refund_error_code: Some("NOT_IMPLEMENTED".to_string()),
                             updated_by: storage_scheme.to_string(),
                             connector_refund_id: None,
-                            connector_refund_data: None,
+                            processor_refund_data: None,
                             unified_code: None,
                             unified_message: None,
                         })
@@ -249,7 +249,7 @@ pub async fn trigger_refund_to_gateway(
                             refund_error_code: Some("NOT_SUPPORTED".to_string()),
                             updated_by: storage_scheme.to_string(),
                             connector_refund_id: None,
-                            connector_refund_data: None,
+                            processor_refund_data: None,
                             unified_code: None,
                             unified_message: None,
                         })
@@ -332,7 +332,7 @@ pub async fn trigger_refund_to_gateway(
                 refund_error_code: Some(err.code),
                 updated_by: storage_scheme.to_string(),
                 connector_refund_id: None,
-                connector_refund_data: None,
+                processor_refund_data: None,
                 unified_code: Some(unified_code),
                 unified_message: Some(unified_message),
             }
@@ -341,7 +341,7 @@ pub async fn trigger_refund_to_gateway(
             // match on connector integrity checks
             match router_data_res.integrity_check.clone() {
                 Err(err) => {
-                    let (refund_connector_transaction_id, connector_refund_data) =
+                    let (refund_connector_transaction_id, processor_refund_data) =
                         err.connector_transaction_id.map_or((None, None), |txn_id| {
                             let (refund_id, refund_data) =
                                 ConnectorTransactionId::form_id_and_data(txn_id);
@@ -363,7 +363,7 @@ pub async fn trigger_refund_to_gateway(
                         refund_error_code: Some("IE".to_string()),
                         updated_by: storage_scheme.to_string(),
                         connector_refund_id: refund_connector_transaction_id,
-                        connector_refund_data,
+                        processor_refund_data,
                         unified_code: None,
                         unified_message: None,
                     }
@@ -378,7 +378,7 @@ pub async fn trigger_refund_to_gateway(
                             )),
                         )
                     }
-                    let (connector_refund_id, connector_refund_data) =
+                    let (connector_refund_id, processor_refund_data) =
                         ConnectorTransactionId::form_id_and_data(response.connector_refund_id);
                     storage::RefundUpdate::Update {
                         connector_refund_id,
@@ -387,7 +387,7 @@ pub async fn trigger_refund_to_gateway(
                         refund_error_message: None,
                         refund_arn: "".to_string(),
                         updated_by: storage_scheme.to_string(),
-                        connector_refund_data,
+                        processor_refund_data,
                     }
                 }
             }
@@ -677,7 +677,7 @@ pub async fn sync_refund_with_gateway(
                 refund_error_code: Some(error_message.code),
                 updated_by: storage_scheme.to_string(),
                 connector_refund_id: None,
-                connector_refund_data: None,
+                processor_refund_data: None,
                 unified_code: None,
                 unified_message: None,
             }
@@ -691,7 +691,7 @@ pub async fn sync_refund_with_gateway(
                         ("merchant_id", merchant_account.get_id().clone()),
                     ),
                 );
-                let (refund_connector_transaction_id, connector_refund_data) = err
+                let (refund_connector_transaction_id, processor_refund_data) = err
                     .connector_transaction_id
                     .map_or((None, None), |refund_id| {
                         let (refund_id, refund_data) =
@@ -707,13 +707,13 @@ pub async fn sync_refund_with_gateway(
                     refund_error_code: Some("IE".to_string()),
                     updated_by: storage_scheme.to_string(),
                     connector_refund_id: refund_connector_transaction_id,
-                    connector_refund_data,
+                    processor_refund_data,
                     unified_code: None,
                     unified_message: None,
                 }
             }
             Ok(()) => {
-                let (connector_refund_id, connector_refund_data) =
+                let (connector_refund_id, processor_refund_data) =
                     ConnectorTransactionId::form_id_and_data(response.connector_refund_id);
                 storage::RefundUpdate::Update {
                     connector_refund_id,
@@ -722,7 +722,7 @@ pub async fn sync_refund_with_gateway(
                     refund_error_message: None,
                     refund_arn: "".to_string(),
                     updated_by: storage_scheme.to_string(),
-                    connector_refund_data,
+                    processor_refund_data,
                 }
             }
         },
@@ -882,7 +882,7 @@ pub async fn validate_and_create_refund(
         .clone()
         .ok_or(errors::ApiErrorResponse::InternalServerError)
         .attach_printable("No connector populated in payment attempt")?;
-    let (connector_transaction_id, connector_transaction_data) =
+    let (connector_transaction_id, processor_transaction_data) =
         ConnectorTransactionId::form_id_and_data(connector_transaction_id);
     let refund_create_req = storage::RefundNew {
         refund_id: refund_id.to_string(),
@@ -912,8 +912,8 @@ pub async fn validate_and_create_refund(
         refund_arn: None,
         updated_by: Default::default(),
         organization_id: merchant_account.organization_id.clone(),
-        connector_refund_data: None,
-        connector_transaction_data,
+        processor_transaction_data,
+        processor_refund_data: None,
     };
 
     let refund = match db
@@ -1584,7 +1584,7 @@ pub fn refund_to_refund_core_workflow_model(
         connector_transaction_id: refund.connector_transaction_id.clone(),
         merchant_id: refund.merchant_id.clone(),
         payment_id: refund.payment_id.clone(),
-        connector_transaction_data: refund.connector_transaction_data.clone(),
+        processor_transaction_data: refund.processor_transaction_data.clone(),
     }
 }
 
diff --git a/crates/router/src/core/webhooks/incoming.rs b/crates/router/src/core/webhooks/incoming.rs
index 2c0d9b5b5c5..8b562978804 100644
--- a/crates/router/src/core/webhooks/incoming.rs
+++ b/crates/router/src/core/webhooks/incoming.rs
@@ -1023,7 +1023,7 @@ async fn refunds_incoming_webhook_flow(
                 .change_context(errors::ApiErrorResponse::WebhookProcessingFailure)
                 .attach_printable("failed refund status mapping from event type")?,
             updated_by: merchant_account.storage_scheme.to_string(),
-            connector_refund_data: None,
+            processor_refund_data: None,
         };
         db.update_refund(
             refund.to_owned(),
diff --git a/crates/router/src/db/capture.rs b/crates/router/src/db/capture.rs
index cbc27f918d4..a2e4a90d66d 100644
--- a/crates/router/src/db/capture.rs
+++ b/crates/router/src/db/capture.rs
@@ -198,7 +198,9 @@ impl CaptureInterface for MockDb {
             capture_sequence: capture.capture_sequence,
             connector_capture_id: capture.connector_capture_id,
             connector_response_reference_id: capture.connector_response_reference_id,
-            connector_capture_data: capture.connector_capture_data,
+            processor_capture_data: capture.processor_capture_data,
+            // Below fields are deprecated. Please add any new fields above this line.
+            connector_capture_data: None,
         };
         captures.push(capture.clone());
         Ok(capture)
diff --git a/crates/router/src/db/refund.rs b/crates/router/src/db/refund.rs
index e07821cd9a8..07bdfa3f77d 100644
--- a/crates/router/src/db/refund.rs
+++ b/crates/router/src/db/refund.rs
@@ -435,10 +435,13 @@ mod storage {
                         charges: new.charges.clone(),
                         split_refunds: new.split_refunds.clone(),
                         organization_id: new.organization_id.clone(),
-                        connector_refund_data: new.connector_refund_data.clone(),
-                        connector_transaction_data: new.connector_transaction_data.clone(),
                         unified_code: None,
                         unified_message: None,
+                        processor_refund_data: new.processor_refund_data.clone(),
+                        processor_transaction_data: new.processor_transaction_data.clone(),
+                        // Below fields are deprecated. Please add any new fields above this line.
+                        connector_refund_data: None,
+                        connector_transaction_data: None,
                     };
 
                     let field = format!(
@@ -620,12 +623,12 @@ mod storage {
 
                     let redis_entry = kv::TypedSql {
                         op: kv::DBOperation::Update {
-                            updatable: Box::new(kv::Updateable::RefundUpdate(
+                            updatable: Box::new(kv::Updateable::RefundUpdate(Box::new(
                                 kv::RefundUpdateMems {
                                     orig: this,
                                     update_data: refund,
                                 },
-                            )),
+                            ))),
                         },
                     };
 
@@ -932,10 +935,13 @@ impl RefundInterface for MockDb {
             charges: new.charges,
             split_refunds: new.split_refunds,
             organization_id: new.organization_id,
-            connector_refund_data: new.connector_refund_data,
-            connector_transaction_data: new.connector_transaction_data,
             unified_code: None,
             unified_message: None,
+            processor_refund_data: new.processor_refund_data.clone(),
+            processor_transaction_data: new.processor_transaction_data.clone(),
+            // Below fields are deprecated. Please add any new fields above this line.
+            connector_refund_data: None,
+            connector_transaction_data: None,
         };
         refunds.push(refund.clone());
         Ok(refund)
diff --git a/crates/router/src/types/storage/payment_attempt.rs b/crates/router/src/types/storage/payment_attempt.rs
index 1773d7f3a17..c82a921fe17 100644
--- a/crates/router/src/types/storage/payment_attempt.rs
+++ b/crates/router/src/types/storage/payment_attempt.rs
@@ -63,6 +63,8 @@ impl PaymentAttemptExt for PaymentAttempt {
             capture_sequence,
             connector_capture_id: None,
             connector_response_reference_id: None,
+            processor_capture_data: None,
+            // Below fields are deprecated. Please add any new fields above this line.
             connector_capture_data: None,
         })
     }
diff --git a/crates/router/src/utils/user/sample_data.rs b/crates/router/src/utils/user/sample_data.rs
index c5652e6cbfa..e189deb9487 100644
--- a/crates/router/src/utils/user/sample_data.rs
+++ b/crates/router/src/utils/user/sample_data.rs
@@ -277,7 +277,7 @@ pub async fn generate_sample_data(
             psd2_sca_exemption_type: None,
             platform_merchant_id: None,
         };
-        let (connector_transaction_id, connector_transaction_data) =
+        let (connector_transaction_id, processor_transaction_data) =
             ConnectorTransactionId::form_id_and_data(attempt_id.clone());
         let payment_attempt = PaymentAttemptBatchNew {
             attempt_id: attempt_id.clone(),
@@ -359,14 +359,14 @@ pub async fn generate_sample_data(
             organization_id: org_id.clone(),
             shipping_cost: None,
             order_tax_amount: None,
-            connector_transaction_data,
+            processor_transaction_data,
             connector_mandate_detail: None,
             card_discovery: None,
         };
 
         let refund = if refunds_count < number_of_refunds && !is_failed_payment {
             refunds_count += 1;
-            let (connector_transaction_id, connector_transaction_data) =
+            let (connector_transaction_id, processor_transaction_data) =
                 ConnectorTransactionId::form_id_and_data(attempt_id.clone());
             Some(RefundNew {
                 refund_id: common_utils::generate_id_with_default_len("test"),
@@ -401,8 +401,8 @@ pub async fn generate_sample_data(
                 charges: None,
                 split_refunds: None,
                 organization_id: org_id.clone(),
-                connector_refund_data: None,
-                connector_transaction_data,
+                processor_refund_data: None,
+                processor_transaction_data,
             })
         } else {
             None
diff --git a/crates/storage_impl/src/payments/payment_attempt.rs b/crates/storage_impl/src/payments/payment_attempt.rs
index 8a8e8d4e4a8..847e9c14da0 100644
--- a/crates/storage_impl/src/payments/payment_attempt.rs
+++ b/crates/storage_impl/src/payments/payment_attempt.rs
@@ -1434,7 +1434,7 @@ impl DataModelExt for PaymentAttempt {
     type StorageModel = DieselPaymentAttempt;
 
     fn to_storage_model(self) -> Self::StorageModel {
-        let (connector_transaction_id, connector_transaction_data) = self
+        let (connector_transaction_id, processor_transaction_data) = self
             .connector_transaction_id
             .map(ConnectorTransactionId::form_id_and_data)
             .map(|(txn_id, txn_data)| (Some(txn_id), txn_data))
@@ -1509,12 +1509,14 @@ impl DataModelExt for PaymentAttempt {
             customer_acceptance: self.customer_acceptance,
             organization_id: self.organization_id,
             profile_id: self.profile_id,
-            connector_transaction_data,
             shipping_cost: self.net_amount.get_shipping_cost(),
             order_tax_amount: self.net_amount.get_order_tax_amount(),
             connector_mandate_detail: self.connector_mandate_detail,
+            processor_transaction_data,
             card_discovery: self.card_discovery,
             charges: self.charges,
+            // Below fields are deprecated. Please add any new fields above this line.
+            connector_transaction_data: None,
         }
     }
 
diff --git a/migrations/2025-01-09-135057_add_processor_transaction_data/down.sql b/migrations/2025-01-09-135057_add_processor_transaction_data/down.sql
new file mode 100644
index 00000000000..bbba634e6e2
--- /dev/null
+++ b/migrations/2025-01-09-135057_add_processor_transaction_data/down.sql
@@ -0,0 +1,8 @@
+ALTER TABLE payment_attempt
+DROP COLUMN IF EXISTS processor_transaction_data;
+
+ALTER TABLE refund DROP COLUMN IF EXISTS processor_refund_data;
+
+ALTER TABLE refund DROP COLUMN IF EXISTS processor_transaction_data;
+
+ALTER TABLE captures DROP COLUMN IF EXISTS processor_capture_data;
\ No newline at end of file
diff --git a/migrations/2025-01-09-135057_add_processor_transaction_data/up.sql b/migrations/2025-01-09-135057_add_processor_transaction_data/up.sql
new file mode 100644
index 00000000000..81e9cdcd074
--- /dev/null
+++ b/migrations/2025-01-09-135057_add_processor_transaction_data/up.sql
@@ -0,0 +1,11 @@
+ALTER TABLE payment_attempt
+ADD COLUMN IF NOT EXISTS processor_transaction_data TEXT;
+
+ALTER TABLE refund
+ADD COLUMN IF NOT EXISTS processor_refund_data TEXT;
+
+ALTER TABLE refund
+ADD COLUMN IF NOT EXISTS processor_transaction_data TEXT;
+
+ALTER TABLE captures
+ADD COLUMN IF NOT EXISTS processor_capture_data TEXT;
\ No newline at end of file
diff --git a/v2_migrations/2024-08-28-081721_add_v2_columns/up.sql b/v2_migrations/2024-08-28-081721_add_v2_columns/up.sql
index 3926a02afa6..03059985875 100644
--- a/v2_migrations/2024-08-28-081721_add_v2_columns/up.sql
+++ b/v2_migrations/2024-08-28-081721_add_v2_columns/up.sql
@@ -51,7 +51,7 @@ ADD COLUMN payment_method_type_v2 VARCHAR,
     ADD COLUMN tax_on_surcharge BIGINT,
     ADD COLUMN payment_method_billing_address BYTEA,
     ADD COLUMN redirection_data JSONB,
-    ADD COLUMN connector_payment_data VARCHAR(512),
+    ADD COLUMN connector_payment_data TEXT,
     ADD COLUMN connector_token_details JSONB;
 
 -- Change the type of the column from JSON to JSONB
diff --git a/v2_migrations/2024-11-08-081847_drop_v1_columns/down.sql b/v2_migrations/2025-01-13-081847_drop_v1_columns/down.sql
similarity index 98%
rename from v2_migrations/2024-11-08-081847_drop_v1_columns/down.sql
rename to v2_migrations/2025-01-13-081847_drop_v1_columns/down.sql
index f67b9b2e7ac..8d0fedce089 100644
--- a/v2_migrations/2024-11-08-081847_drop_v1_columns/down.sql
+++ b/v2_migrations/2025-01-13-081847_drop_v1_columns/down.sql
@@ -76,7 +76,7 @@ ADD COLUMN IF NOT EXISTS attempt_id VARCHAR(64) NOT NULL,
     ADD COLUMN offer_amount bigint,
     ADD COLUMN payment_method VARCHAR,
     ADD COLUMN connector_transaction_id VARCHAR(64),
-    ADD COLUMN connector_transaction_data VARCHAR(512),
+    ADD COLUMN connector_transaction_data TEXT,
     ADD COLUMN capture_method "CaptureMethod",
     ADD COLUMN capture_on TIMESTAMP,
     ADD COLUMN mandate_id VARCHAR(64),
diff --git a/v2_migrations/2024-11-08-081847_drop_v1_columns/up.sql b/v2_migrations/2025-01-13-081847_drop_v1_columns/up.sql
similarity index 92%
rename from v2_migrations/2024-11-08-081847_drop_v1_columns/up.sql
rename to v2_migrations/2025-01-13-081847_drop_v1_columns/up.sql
index e19f8631f62..caac00abd0f 100644
--- a/v2_migrations/2024-11-08-081847_drop_v1_columns/up.sql
+++ b/v2_migrations/2025-01-13-081847_drop_v1_columns/up.sql
@@ -75,6 +75,7 @@ ALTER TABLE payment_attempt DROP COLUMN attempt_id,
     DROP COLUMN payment_method,
     DROP COLUMN connector_transaction_id,
     DROP COLUMN connector_transaction_data,
+    DROP COLUMN processor_transaction_data,
     DROP COLUMN capture_method,
     DROP COLUMN capture_on,
     DROP COLUMN mandate_id,
@@ -89,3 +90,10 @@ ALTER TABLE payment_attempt DROP COLUMN attempt_id,
     DROP COLUMN payment_method_billing_address_id,
     DROP COLUMN connector_mandate_detail,
     DROP COLUMN charge_id;
+
+-- Run below queries only when V1 is deprecated
+ALTER TABLE refund DROP COLUMN connector_refund_data,
+    DROP COLUMN connector_transaction_data;
+
+-- Run below queries only when V1 is deprecated
+ALTER TABLE captures DROP COLUMN connector_capture_data;

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

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

## Description
This PR introduces below changes around Worldpay connector

### Core changes
- add new columns (TEXT) for storing the JWE based connector transaction ID
    - `payment_attempt` - `processor_transaction_data`
    - `refund` - `processor_transaction_data` and `processor_refund_data`
    - `captures` - `processor_capture_data`

### Connector integration changes (Worldpay only)
- make `line1` and `city` non optional in payment request
- add `zip` in dynamic required fields for Worldpay cards

### QA
- add cypress test cases for dynamic fields for Worldpay

### Additional Changes

- [ ] This PR modifies the API contract
- [x] This PR modifies the database schema
- [ ] This PR modifies application configuration/environment variables


## Motivation and Context
This PR keeps Worldpay's payments integration intact.

## How did you test it?
<details>
    <summary>1. Route a payment through Worldpay</summary>

cURL

    curl --location --request POST 'http://localhost:8080/payments' \
        --header 'Content-Type: application/json' \
        --header 'Accept: application/json' \
        --header 'api-key: dev_Wc9lMS2KJ6l35ehmlSsR3dwwiOvGZaWLmkk3BqRLRe6vqa1rkiNJ3mVza3U1EdWh' \
        --data-raw '{"amount":100,"currency":"USD","confirm":true,"capture_method":"automatic","capture_on":"2022-09-10T10:11:12Z","customer_id":"cus_JgZZwioWal2CoKgsLXEb","email":"guest@example.com","name":"John Doe","phone":"999999999","profile_id":"pro_NyQDoa1SRAC4rZ5QjDv9","phone_country_code":"+65","description":"Its my first payment request","authentication_type":"no_three_ds","return_url":"https://hyperswitch.io","setup_future_usage":"off_session","customer_acceptance":{"acceptance_type":"online","accepted_at":"1963-05-03T04:07:52.723Z","online":{"ip_address":"127.0.0.1","user_agent":"amet irure esse"}},"connector":["worldpay"],"payment_method":"card","payment_method_data":{"card":{"card_number":"4917610000000000","card_exp_month":"03","card_exp_year":"2030","card_cvc":"737","nick_name":"JD"},"billing":{"email":"guest@example.com","address":{"first_name":"John","last_name":"US"}}},"billing":{"address":{"line1":"1467","line2":"Harrison Street","line3":"Harrison Street","city":"San Fransico","state":"California","zip":"94122","country":"US","first_name":"John","last_name":"US"}},"statement_descriptor_name":"joseph","statement_descriptor_suffix":"JS","metadata":{"udf1":"value1","new_customer":"true","login_date":"2019-09-10T10:11:12Z"},"browser_info":{"ip_address":"129.0.0.1","user_agent":"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36","accept_header":"text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8","language":"en-US","color_depth":32,"screen_height":1117,"screen_width":1728,"time_zone":-330,"java_enabled":true,"java_script_enabled":true}}'

Response

    {"payment_id":"pay_wY8cJE6MgvO13EE0vIyx","merchant_id":"merchant_1736497521","status":"succeeded","amount":100,"net_amount":100,"shipping_cost":null,"amount_capturable":0,"amount_received":100,"connector":"worldpay","client_secret":"pay_wY8cJE6MgvO13EE0vIyx_secret_42yawtMDmMbJf15nP8TP","created":"2025-01-10T09:35:51.576Z","currency":"USD","customer_id":"cus_JgZZwioWal2CoKgsLXEb","customer":{"id":"cus_JgZZwioWal2CoKgsLXEb","name":"John Doe","email":"guest@example.com","phone":"999999999","phone_country_code":"+65"},"description":"Its my first payment request","refunds":null,"disputes":null,"mandate_id":null,"mandate_data":null,"setup_future_usage":"off_session","off_session":null,"capture_on":null,"capture_method":"automatic","payment_method":"card","payment_method_data":{"card":{"last4":"0000","card_type":"CREDIT","card_network":"Visa","card_issuer":"BANKPOLSKAKASAOPIEKIS.A.(BANKPEKAOSA)","card_issuing_country":"POLAND","card_isin":"491761","card_extended_bin":null,"card_exp_month":"03","card_exp_year":"2030","card_holder_name":"John US","payment_checks":null,"authentication_data":null},"billing":{"address":{"city":null,"country":null,"line1":null,"line2":null,"line3":null,"zip":null,"state":null,"first_name":"John","last_name":"US"},"phone":null,"email":"guest@example.com"}},"payment_token":null,"shipping":null,"billing":{"address":{"city":"San Fransico","country":"US","line1":"1467","line2":"Harrison Street","line3":"Harrison Street","zip":"94122","state":"California","first_name":"John","last_name":"US"},"phone":null,"email":null},"order_details":null,"email":"guest@example.com","name":"John Doe","phone":"999999999","return_url":"https://hyperswitch.io/","authentication_type":"no_three_ds","statement_descriptor_name":"joseph","statement_descriptor_suffix":"JS","next_action":null,"cancellation_reason":null,"error_code":null,"error_message":null,"unified_code":null,"unified_message":null,"payment_experience":null,"payment_method_type":"credit","connector_label":null,"business_country":null,"business_label":"default","business_sub_label":null,"allowed_payment_method_types":null,"ephemeral_key":{"customer_id":"cus_JgZZwioWal2CoKgsLXEb","created_at":1736501751,"expires":1736505351,"secret":"epk_d47f022a1de246bebe54d6898c408836"},"manual_retry_allowed":false,"connector_transaction_id":"eyJrIjoiazNhYjYzMiIsImxpbmtWZXJzaW9uIjoiNi4wLjAifQ==.sN:g8wd64bwkbrp0md+bPxcanBnk2zLdsIqSa1pR99GGg8fCNQpPLoWNslSzWNPFBM5Tpa8tW7EFI5onKINsgChMHeJVoeH2lrBWCRyjZYT6h+lbqfJa+1BSoKFSY8HLJDIunO0DhjaUi3niPBfv98:VUQcO2wyJ4EGywR:l4ssCpZY8cZeCeMgt1ol:GABqQG6u+T:u3sfHv3ezSOJxioRTixsThPEhzFW4ZZ6mObj3CK0rnFndcM0swvZMqgQwSEj5tBsydfzM4XKX20O6Nme96ha9twqxTSQIfr1rtl9V3q7fw3w5O9UZT4vQi9BMyCcaHkWSD:RbCWCcmiQqa5EPu9d6p6Bt0lCYl:Np4aWZ0pJCEaV4g9uGTAlKDtig0jKl6MU3ArZWtDWm7LKjrd+hQ9RR:Z491q0vLPK+etFBNqgwjiceXMDUvG7lafYHHR5Ak5lI+Sa7YaesgGkXXzgiDTOQzNK9UH+W9rzqFt6t5VcedliH8B9MKI3IKaPyVavYo4YzMmVfRID7seWtakAYMzGq+YEqwioSKkuPXbo=","frm_message":null,"metadata":{"udf1":"value1","login_date":"2019-09-10T10:11:12Z","new_customer":"true"},"connector_metadata":null,"feature_metadata":null,"reference_id":"95f4583f-a8ee-426c-acd8-f475d85b0bef","payment_link":null,"profile_id":"pro_NyQDoa1SRAC4rZ5QjDv9","surcharge_details":null,"attempt_count":1,"merchant_decision":null,"merchant_connector_id":"mca_jJbWUjL9iGOEfQo8Uw8P","incremental_authorization_allowed":null,"authorization_count":null,"incremental_authorizations":null,"external_authentication_details":null,"external_3ds_authentication_attempted":false,"expires_on":"2025-01-10T09:50:51.576Z","fingerprint":null,"browser_info":{"language":"en-US","time_zone":-330,"ip_address":"129.0.0.1","user_agent":"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36","color_depth":32,"java_enabled":true,"screen_width":1728,"accept_header":"text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8","screen_height":1117,"java_script_enabled":true},"payment_method_id":"pm_xeVtmDykRQKg3pvt4EI5","payment_method_status":"active","updated":"2025-01-10T09:35:52.981Z","split_payments":null,"frm_metadata":null,"merchant_order_reference_id":null,"order_tax_amount":null,"connector_mandate_id":"https://try.access.worldpay.com/tokens/eyJrIjoxLCJkIjoiWndnc0NNQjJXNkZFTkl4TUZUZFBKdzhQdm9OS3JjYnFnT0haOThuOVE4RT0ifQ"}

</details>

<details>
    <summary>2. Create a payment link without passing in details</summary>

cURL

    curl --location --request POST 'http://localhost:8080/payments' \
        --header 'Content-Type: application/json' \
        --header 'Accept: application/json' \
        --header 'api-key: dev_Wc9lMS2KJ6l35ehmlSsR3dwwiOvGZaWLmkk3BqRLRe6vqa1rkiNJ3mVza3U1EdWh' \
        --data-raw '{"authentication_type":"three_ds","customer_id":"cus_JgZZwioWal2CoKgsLXEb","profile_id":"pro_NyQDoa1SRAC4rZ5QjDv9","amount":100,"currency":"USD","payment_link":true,"connector":["worldpay"],"description":"This is my description of why this payment was requested.","capture_method":"automatic","billing":{"address":{"line2":"Harrison Street","line3":"Harrison Street","state":"CA","first_name":"John","last_name":"Doe"},"phone":{"number":"8056594427","country_code":"+91"}},"email":"john.doe@example.co.in","session_expiry":100000,"return_url":"https://example.com","payment_link_config":{"theme":"#0E103D","logo":"https://hyperswitch.io/favicon.ico","seller_name":"Hyperswitch Inc.","enabled_saved_payment_method":true,"hide_card_nickname_field":true,"show_card_form_by_default":true,"transaction_details":[{"key":"Policy Number","value":"938478327648","ui_configuration":{"is_key_bold":false,"is_value_bold":false,"position":2}},{"key":"Business Country","value":"Germany","ui_configuration":{"is_key_bold":false,"is_value_bold":false,"position":1}},{"key":"Product Name","value":"Life Insurance Elite","ui_configuration":{"is_key_bold":false,"is_value_bold":false,"position":3}}]}}'


- Open payment link

<img width="599" alt="Screenshot 2025-01-10 at 3 08 50 PM" src="https://github.com/user-attachments/assets/8a3b0a75-ff52-4f5e-b889-460e5261556b" />


</details>

<details>
    <summary>3. Dynamic field tests</summary>

<img width="642" alt="Screenshot 2025-01-10 at 3 10 29 PM" src="https://github.com/user-attachments/assets/5929d89e-341c-4a86-a488-5a77d983a08e" />

</details>

## 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 the submitted code
- [ ] I added unit tests for my changes where possible
