Bug: [FEATURE] Add v2 kafka events for payment_intent, payment_attempt and refund




diff --git a/crates/api_models/src/payments.rs b/crates/api_models/src/payments.rs
index 7b14351a49a..ca382fa1479 100644
--- a/crates/api_models/src/payments.rs
+++ b/crates/api_models/src/payments.rs
@@ -5560,25 +5560,24 @@ impl From<&PaymentsRequest> for PaymentsCreateIntentRequest {
             billing: request.billing.clone(),
             shipping: request.shipping.clone(),
             customer_id: request.customer_id.clone(),
-            customer_present: request.customer_present.clone(),
+            customer_present: request.customer_present,
             description: request.description.clone(),
             return_url: request.return_url.clone(),
             setup_future_usage: request.setup_future_usage,
-            apply_mit_exemption: request.apply_mit_exemption.clone(),
+            apply_mit_exemption: request.apply_mit_exemption,
             statement_descriptor: request.statement_descriptor.clone(),
             order_details: request.order_details.clone(),
             allowed_payment_method_types: request.allowed_payment_method_types.clone(),
             metadata: request.metadata.clone(),
             connector_metadata: request.connector_metadata.clone(),
             feature_metadata: request.feature_metadata.clone(),
-            payment_link_enabled: request.payment_link_enabled.clone(),
+            payment_link_enabled: request.payment_link_enabled,
             payment_link_config: request.payment_link_config.clone(),
             request_incremental_authorization: request.request_incremental_authorization,
             session_expiry: request.session_expiry,
             frm_metadata: request.frm_metadata.clone(),
             request_external_three_ds_authentication: request
-                .request_external_three_ds_authentication
-                .clone(),
+                .request_external_three_ds_authentication,
             force_3ds_challenge: request.force_3ds_challenge,
             merchant_connector_details: request.merchant_connector_details.clone(),
         }
diff --git a/crates/common_enums/src/enums.rs b/crates/common_enums/src/enums.rs
index 568fa5959eb..641dc83291c 100644
--- a/crates/common_enums/src/enums.rs
+++ b/crates/common_enums/src/enums.rs
@@ -1679,6 +1679,16 @@ pub enum FutureUsage {
     OnSession,
 }
 
+impl FutureUsage {
+    /// Indicates whether the payment method should be saved for future use or not
+    pub fn is_off_session(self) -> bool {
+        match self {
+            Self::OffSession => true,
+            Self::OnSession => false,
+        }
+    }
+}
+
 #[derive(
     Clone,
     Copy,
@@ -7975,7 +7985,9 @@ pub enum SuccessBasedRoutingConclusiveState {
 }
 
 /// Whether 3ds authentication is requested or not
-#[derive(Clone, Debug, PartialEq, serde::Serialize, serde::Deserialize, Default, ToSchema)]
+#[derive(
+    Clone, Copy, Debug, PartialEq, serde::Serialize, serde::Deserialize, Default, ToSchema,
+)]
 pub enum External3dsAuthenticationRequest {
     /// Request for 3ds authentication
     Enable,
@@ -7985,7 +7997,9 @@ pub enum External3dsAuthenticationRequest {
 }
 
 /// Whether payment link is requested to be enabled or not for this transaction
-#[derive(Clone, Debug, PartialEq, serde::Serialize, serde::Deserialize, Default, ToSchema)]
+#[derive(
+    Clone, Copy, Debug, PartialEq, serde::Serialize, serde::Deserialize, Default, ToSchema,
+)]
 pub enum EnablePaymentLinkRequest {
     /// Request for enabling payment link
     Enable,
@@ -7994,7 +8008,9 @@ pub enum EnablePaymentLinkRequest {
     Skip,
 }
 
-#[derive(Clone, Debug, PartialEq, serde::Serialize, serde::Deserialize, Default, ToSchema)]
+#[derive(
+    Clone, Copy, Debug, PartialEq, serde::Serialize, serde::Deserialize, Default, ToSchema,
+)]
 pub enum MitExemptionRequest {
     /// Request for applying MIT exemption
     Apply,
@@ -8004,7 +8020,9 @@ pub enum MitExemptionRequest {
 }
 
 /// Set to `present` to indicate that the customer is in your checkout flow during this payment, and therefore is able to authenticate. This parameter should be `absent` when merchant's doing merchant initiated payments and customer is not present while doing the payment.
-#[derive(Clone, Debug, PartialEq, serde::Serialize, serde::Deserialize, Default, ToSchema)]
+#[derive(
+    Clone, Copy, Debug, PartialEq, serde::Serialize, serde::Deserialize, Default, ToSchema,
+)]
 #[serde(rename_all = "snake_case")]
 pub enum PresenceOfCustomerDuringPayment {
     /// Customer is present during the payment. This is the default value
diff --git a/crates/hyperswitch_domain_models/src/payments.rs b/crates/hyperswitch_domain_models/src/payments.rs
index 8c892dcc08f..eb5c0a467c5 100644
--- a/crates/hyperswitch_domain_models/src/payments.rs
+++ b/crates/hyperswitch_domain_models/src/payments.rs
@@ -289,7 +289,7 @@ impl AmountDetails {
     }
 
     /// Get the action to whether calculate external tax or not as a boolean value
-    fn get_external_tax_action_as_bool(&self) -> bool {
+    pub fn get_external_tax_action_as_bool(&self) -> bool {
         self.skip_external_tax_calculation.as_bool()
     }
 
diff --git a/crates/masking/src/secret.rs b/crates/masking/src/secret.rs
index f2f519dd3ed..889c666f801 100644
--- a/crates/masking/src/secret.rs
+++ b/crates/masking/src/secret.rs
@@ -85,6 +85,14 @@ where
     {
         StrongSecret::new(self.inner_secret)
     }
+
+    /// Convert to [`Secret`] with a reference to the inner secret
+    pub fn as_ref(&self) -> Secret<&SecretValue, MaskingStrategy>
+    where
+        MaskingStrategy: for<'a> Strategy<&'a SecretValue>,
+    {
+        Secret::new(self.peek())
+    }
 }
 
 impl<SecretValue, MaskingStrategy> PeekInterface<SecretValue>
diff --git a/crates/masking/src/serde.rs b/crates/masking/src/serde.rs
index ecb92086a63..463c0ebc369 100644
--- a/crates/masking/src/serde.rs
+++ b/crates/masking/src/serde.rs
@@ -31,6 +31,8 @@ impl SerializableSecret for url::Url {}
 #[cfg(feature = "time")]
 impl SerializableSecret for time::Date {}
 
+impl<T: SerializableSecret> SerializableSecret for &T {}
+
 impl<'de, T, I> Deserialize<'de> for Secret<T, I>
 where
     T: Clone + de::DeserializeOwned + Sized,
diff --git a/crates/router/src/core/payments/transformers.rs b/crates/router/src/core/payments/transformers.rs
index 7094c73be5a..4ded67d612e 100644
--- a/crates/router/src/core/payments/transformers.rs
+++ b/crates/router/src/core/payments/transformers.rs
@@ -361,7 +361,7 @@ pub async fn construct_payment_router_data_for_authorize<'a>(
         // TODO: Create unified address
         address: payment_data.payment_address.clone(),
         auth_type: payment_data.payment_attempt.authentication_type,
-        connector_meta_data: None,
+        connector_meta_data: merchant_connector_account.get_metadata(),
         connector_wallets_details: None,
         request,
         response: Err(hyperswitch_domain_models::router_data::ErrorResponse::default()),
@@ -1791,11 +1791,11 @@ where
                     .map(|shipping| shipping.into_inner())
                     .map(From::from),
                 customer_id: payment_intent.customer_id.clone(),
-                customer_present: payment_intent.customer_present.clone(),
+                customer_present: payment_intent.customer_present,
                 description: payment_intent.description.clone(),
                 return_url: payment_intent.return_url.clone(),
                 setup_future_usage: payment_intent.setup_future_usage,
-                apply_mit_exemption: payment_intent.apply_mit_exemption.clone(),
+                apply_mit_exemption: payment_intent.apply_mit_exemption,
                 statement_descriptor: payment_intent.statement_descriptor.clone(),
                 order_details: payment_intent.order_details.clone().map(|order_details| {
                     order_details
@@ -1810,7 +1810,7 @@ where
                     .feature_metadata
                     .clone()
                     .map(|feature_metadata| feature_metadata.convert_back()),
-                payment_link_enabled: payment_intent.enable_payment_link.clone(),
+                payment_link_enabled: payment_intent.enable_payment_link,
                 payment_link_config: payment_intent
                     .payment_link_config
                     .clone()
@@ -1819,8 +1819,7 @@ where
                 expires_on: payment_intent.session_expiry,
                 frm_metadata: payment_intent.frm_metadata.clone(),
                 request_external_three_ds_authentication: payment_intent
-                    .request_external_three_ds_authentication
-                    .clone(),
+                    .request_external_three_ds_authentication,
             },
             vec![],
         )))
diff --git a/crates/router/src/services/kafka/payment_attempt.rs b/crates/router/src/services/kafka/payment_attempt.rs
index 6232bbcbf07..f2c18ae9302 100644
--- a/crates/router/src/services/kafka/payment_attempt.rs
+++ b/crates/router/src/services/kafka/payment_attempt.rs
@@ -1,11 +1,22 @@
-// use diesel_models::enums::MandateDetails;
+#[cfg(feature = "v2")]
+use common_types::payments;
+#[cfg(feature = "v2")]
+use common_utils::types;
 use common_utils::{id_type, types::MinorUnit};
 use diesel_models::enums as storage_enums;
+#[cfg(feature = "v2")]
+use diesel_models::payment_attempt;
+#[cfg(feature = "v2")]
+use hyperswitch_domain_models::{
+    address, payments::payment_attempt::PaymentAttemptFeatureMetadata,
+    router_response_types::RedirectForm,
+};
 use hyperswitch_domain_models::{
     mandates::MandateDetails, payments::payment_attempt::PaymentAttempt,
 };
 use time::OffsetDateTime;
 
+#[cfg(feature = "v1")]
 #[derive(serde::Serialize, Debug)]
 pub struct KafkaPaymentAttempt<'a> {
     pub payment_id: &'a id_type::PaymentId,
@@ -123,68 +134,242 @@ impl<'a> KafkaPaymentAttempt<'a> {
     }
 }
 
+#[cfg(feature = "v2")]
+#[derive(serde::Serialize, Debug)]
+pub struct KafkaPaymentAttempt<'a> {
+    pub payment_id: &'a id_type::GlobalPaymentId,
+    pub merchant_id: &'a id_type::MerchantId,
+    pub attempt_id: &'a id_type::GlobalAttemptId,
+    pub status: storage_enums::AttemptStatus,
+    pub amount: MinorUnit,
+    pub connector: Option<&'a String>,
+    pub error_message: Option<&'a String>,
+    pub surcharge_amount: Option<MinorUnit>,
+    pub tax_amount: Option<MinorUnit>,
+    pub payment_method_id: Option<&'a id_type::GlobalPaymentMethodId>,
+    pub payment_method: storage_enums::PaymentMethod,
+    pub connector_transaction_id: Option<&'a String>,
+    pub authentication_type: storage_enums::AuthenticationType,
+    #[serde(with = "time::serde::timestamp")]
+    pub created_at: OffsetDateTime,
+    #[serde(with = "time::serde::timestamp")]
+    pub modified_at: OffsetDateTime,
+    #[serde(default, with = "time::serde::timestamp::option")]
+    pub last_synced: Option<OffsetDateTime>,
+    pub cancellation_reason: Option<&'a String>,
+    pub amount_to_capture: Option<MinorUnit>,
+    pub browser_info: Option<&'a types::BrowserInformation>,
+    pub error_code: Option<&'a String>,
+    pub connector_metadata: Option<String>,
+    // TODO: These types should implement copy ideally
+    pub payment_experience: Option<&'a storage_enums::PaymentExperience>,
+    pub payment_method_type: &'a storage_enums::PaymentMethodType,
+    pub payment_method_data: Option<String>,
+    pub error_reason: Option<&'a String>,
+    pub multiple_capture_count: Option<i16>,
+    pub amount_capturable: MinorUnit,
+    pub merchant_connector_id: Option<&'a id_type::MerchantConnectorAccountId>,
+    pub net_amount: MinorUnit,
+    pub unified_code: Option<&'a String>,
+    pub unified_message: Option<&'a String>,
+    pub client_source: Option<&'a String>,
+    pub client_version: Option<&'a String>,
+    pub profile_id: &'a id_type::ProfileId,
+    pub organization_id: &'a id_type::OrganizationId,
+    pub card_network: Option<String>,
+    pub card_discovery: Option<String>,
+
+    pub connector_payment_id: Option<types::ConnectorTransactionId>,
+    pub payment_token: Option<String>,
+    pub preprocessing_step_id: Option<String>,
+    pub connector_response_reference_id: Option<String>,
+    pub updated_by: &'a String,
+    pub encoded_data: Option<&'a masking::Secret<String>>,
+    pub external_three_ds_authentication_attempted: Option<bool>,
+    pub authentication_connector: Option<String>,
+    pub authentication_id: Option<String>,
+    pub fingerprint_id: Option<String>,
+    pub customer_acceptance: Option<&'a masking::Secret<serde_json::Value>>,
+    pub shipping_cost: Option<MinorUnit>,
+    pub order_tax_amount: Option<MinorUnit>,
+    pub charges: Option<payments::ConnectorChargeResponseData>,
+    pub processor_merchant_id: &'a id_type::MerchantId,
+    pub created_by: Option<&'a types::CreatedBy>,
+    pub payment_method_type_v2: storage_enums::PaymentMethod,
+    pub payment_method_subtype: storage_enums::PaymentMethodType,
+    pub routing_result: Option<serde_json::Value>,
+    pub authentication_applied: Option<common_enums::AuthenticationType>,
+    pub external_reference_id: Option<String>,
+    pub tax_on_surcharge: Option<MinorUnit>,
+    pub payment_method_billing_address: Option<masking::Secret<&'a address::Address>>, // adjusted from Encryption
+    pub redirection_data: Option<&'a RedirectForm>,
+    pub connector_payment_data: Option<String>,
+    pub connector_token_details: Option<&'a payment_attempt::ConnectorTokenDetails>,
+    pub feature_metadata: Option<&'a PaymentAttemptFeatureMetadata>,
+    pub network_advice_code: Option<String>,
+    pub network_decline_code: Option<String>,
+    pub network_error_message: Option<String>,
+    pub connector_request_reference_id: Option<String>,
+}
+
 #[cfg(feature = "v2")]
 impl<'a> KafkaPaymentAttempt<'a> {
     pub fn from_storage(attempt: &'a PaymentAttempt) -> Self {
-        todo!()
-        // Self {
-        //     payment_id: &attempt.payment_id,
-        //     merchant_id: &attempt.merchant_id,
-        //     attempt_id: &attempt.attempt_id,
-        //     status: attempt.status,
-        //     amount: attempt.amount,
-        //     currency: attempt.currency,
-        //     save_to_locker: attempt.save_to_locker,
-        //     connector: attempt.connector.as_ref(),
-        //     error_message: attempt.error_message.as_ref(),
-        //     offer_amount: attempt.offer_amount,
-        //     surcharge_amount: attempt.surcharge_amount,
-        //     tax_amount: attempt.tax_amount,
-        //     payment_method_id: attempt.payment_method_id.as_ref(),
-        //     payment_method: attempt.payment_method,
-        //     connector_transaction_id: attempt.connector_transaction_id.as_ref(),
-        //     capture_method: attempt.capture_method,
-        //     capture_on: attempt.capture_on.map(|i| i.assume_utc()),
-        //     confirm: attempt.confirm,
-        //     authentication_type: attempt.authentication_type,
-        //     created_at: attempt.created_at.assume_utc(),
-        //     modified_at: attempt.modified_at.assume_utc(),
-        //     last_synced: attempt.last_synced.map(|i| i.assume_utc()),
-        //     cancellation_reason: attempt.cancellation_reason.as_ref(),
-        //     amount_to_capture: attempt.amount_to_capture,
-        //     mandate_id: attempt.mandate_id.as_ref(),
-        //     browser_info: attempt.browser_info.as_ref().map(|v| v.to_string()),
-        //     error_code: attempt.error_code.as_ref(),
-        //     connector_metadata: attempt.connector_metadata.as_ref().map(|v| v.to_string()),
-        //     payment_experience: attempt.payment_experience.as_ref(),
-        //     payment_method_type: attempt.payment_method_type.as_ref(),
-        //     payment_method_data: attempt.payment_method_data.as_ref().map(|v| v.to_string()),
-        //     error_reason: attempt.error_reason.as_ref(),
-        //     multiple_capture_count: attempt.multiple_capture_count,
-        //     amount_capturable: attempt.amount_capturable,
-        //     merchant_connector_id: attempt.merchant_connector_id.as_ref(),
-        //     net_amount: attempt.net_amount,
-        //     unified_code: attempt.unified_code.as_ref(),
-        //     unified_message: attempt.unified_message.as_ref(),
-        //     mandate_data: attempt.mandate_data.as_ref(),
-        //     client_source: attempt.client_source.as_ref(),
-        //     client_version: attempt.client_version.as_ref(),
-        //     profile_id: &attempt.profile_id,
-        //     organization_id: &attempt.organization_id,
-        //     card_network: attempt
-        //         .payment_method_data
-        //         .as_ref()
-        //         .and_then(|data| data.as_object())
-        //         .and_then(|pm| pm.get("card"))
-        //         .and_then(|data| data.as_object())
-        //         .and_then(|card| card.get("card_network"))
-        //         .and_then(|network| network.as_str())
-        //         .map(|network| network.to_string()),
-        // }
+        use masking::PeekInterface;
+        let PaymentAttempt {
+            payment_id,
+            merchant_id,
+            amount_details,
+            status,
+            connector,
+            error,
+            authentication_type,
+            created_at,
+            modified_at,
+            last_synced,
+            cancellation_reason,
+            browser_info,
+            payment_token,
+            connector_metadata,
+            payment_experience,
+            payment_method_data,
+            routing_result,
+            preprocessing_step_id,
+            multiple_capture_count,
+            connector_response_reference_id,
+            updated_by,
+            redirection_data,
+            encoded_data,
+            merchant_connector_id,
+            external_three_ds_authentication_attempted,
+            authentication_connector,
+            authentication_id,
+            fingerprint_id,
+            client_source,
+            client_version,
+            customer_acceptance,
+            profile_id,
+            organization_id,
+            payment_method_type,
+            payment_method_id,
+            connector_payment_id,
+            payment_method_subtype,
+            authentication_applied,
+            external_reference_id,
+            payment_method_billing_address,
+            id,
+            connector_token_details,
+            card_discovery,
+            charges,
+            feature_metadata,
+            processor_merchant_id,
+            created_by,
+            connector_request_reference_id,
+        } = attempt;
+
+        let (connector_payment_id, connector_payment_data) = connector_payment_id
+            .clone()
+            .map(types::ConnectorTransactionId::form_id_and_data)
+            .map(|(txn_id, txn_data)| (Some(txn_id), txn_data))
+            .unwrap_or((None, None));
+
+        Self {
+            payment_id,
+            merchant_id,
+            attempt_id: id,
+            status: *status,
+            amount: amount_details.get_net_amount(),
+            connector: connector.as_ref(),
+            error_message: error.as_ref().map(|error_details| &error_details.message),
+            surcharge_amount: amount_details.get_surcharge_amount(),
+            tax_amount: amount_details.get_tax_on_surcharge(),
+            payment_method_id: payment_method_id.as_ref(),
+            payment_method: *payment_method_type,
+            connector_transaction_id: connector_response_reference_id.as_ref(),
+            authentication_type: *authentication_type,
+            created_at: created_at.assume_utc(),
+            modified_at: modified_at.assume_utc(),
+            last_synced: last_synced.map(|i| i.assume_utc()),
+            cancellation_reason: cancellation_reason.as_ref(),
+            amount_to_capture: amount_details.get_amount_to_capture(),
+            browser_info: browser_info.as_ref(),
+            error_code: error.as_ref().map(|error_details| &error_details.code),
+            connector_metadata: connector_metadata.as_ref().map(|v| v.peek().to_string()),
+            payment_experience: payment_experience.as_ref(),
+            payment_method_type: payment_method_subtype,
+            payment_method_data: payment_method_data.as_ref().map(|v| v.peek().to_string()),
+            error_reason: error
+                .as_ref()
+                .and_then(|error_details| error_details.reason.as_ref()),
+            multiple_capture_count: *multiple_capture_count,
+            amount_capturable: amount_details.get_amount_capturable(),
+            merchant_connector_id: merchant_connector_id.as_ref(),
+            net_amount: amount_details.get_net_amount(),
+            unified_code: error
+                .as_ref()
+                .and_then(|error_details| error_details.unified_code.as_ref()),
+            unified_message: error
+                .as_ref()
+                .and_then(|error_details| error_details.unified_message.as_ref()),
+            client_source: client_source.as_ref(),
+            client_version: client_version.as_ref(),
+            profile_id,
+            organization_id,
+            card_network: payment_method_data
+                .as_ref()
+                .map(|data| data.peek())
+                .and_then(|data| data.as_object())
+                .and_then(|pm| pm.get("card"))
+                .and_then(|data| data.as_object())
+                .and_then(|card| card.get("card_network"))
+                .and_then(|network| network.as_str())
+                .map(|network| network.to_string()),
+            card_discovery: card_discovery.map(|discovery| discovery.to_string()),
+            payment_token: payment_token.clone(),
+            preprocessing_step_id: preprocessing_step_id.clone(),
+            connector_response_reference_id: connector_response_reference_id.clone(),
+            updated_by,
+            encoded_data: encoded_data.as_ref(),
+            external_three_ds_authentication_attempted: *external_three_ds_authentication_attempted,
+            authentication_connector: authentication_connector.clone(),
+            authentication_id: authentication_id.clone(),
+            fingerprint_id: fingerprint_id.clone(),
+            customer_acceptance: customer_acceptance.as_ref(),
+            shipping_cost: amount_details.get_shipping_cost(),
+            order_tax_amount: amount_details.get_order_tax_amount(),
+            charges: charges.clone(),
+            processor_merchant_id,
+            created_by: created_by.as_ref(),
+            payment_method_type_v2: *payment_method_type,
+            connector_payment_id: connector_payment_id.as_ref().cloned(),
+            payment_method_subtype: *payment_method_subtype,
+            routing_result: routing_result.clone(),
+            authentication_applied: *authentication_applied,
+            external_reference_id: external_reference_id.clone(),
+            tax_on_surcharge: amount_details.get_tax_on_surcharge(),
+            payment_method_billing_address: payment_method_billing_address
+                .as_ref()
+                .map(|v| masking::Secret::new(v.get_inner())),
+            redirection_data: redirection_data.as_ref(),
+            connector_payment_data,
+            connector_token_details: connector_token_details.as_ref(),
+            feature_metadata: feature_metadata.as_ref(),
+            network_advice_code: error
+                .as_ref()
+                .and_then(|details| details.network_advice_code.clone()),
+            network_decline_code: error
+                .as_ref()
+                .and_then(|details| details.network_decline_code.clone()),
+            network_error_message: error
+                .as_ref()
+                .and_then(|details| details.network_error_message.clone()),
+            connector_request_reference_id: connector_request_reference_id.clone(),
+        }
     }
 }
 
 impl super::KafkaMessage for KafkaPaymentAttempt<'_> {
+    #[cfg(feature = "v1")]
     fn key(&self) -> String {
         format!(
             "{}_{}_{}",
@@ -193,6 +378,15 @@ impl super::KafkaMessage for KafkaPaymentAttempt<'_> {
             self.attempt_id
         )
     }
+    #[cfg(feature = "v2")]
+    fn key(&self) -> String {
+        format!(
+            "{}_{}_{}",
+            self.merchant_id.get_string_repr(),
+            self.payment_id.get_string_repr(),
+            self.attempt_id.get_string_repr()
+        )
+    }
 
     fn event_type(&self) -> crate::events::EventType {
         crate::events::EventType::PaymentAttempt
diff --git a/crates/router/src/services/kafka/payment_attempt_event.rs b/crates/router/src/services/kafka/payment_attempt_event.rs
index f3fcd701846..e06f4e014dd 100644
--- a/crates/router/src/services/kafka/payment_attempt_event.rs
+++ b/crates/router/src/services/kafka/payment_attempt_event.rs
@@ -1,11 +1,22 @@
-// use diesel_models::enums::MandateDetails;
+#[cfg(feature = "v2")]
+use common_types::payments;
+#[cfg(feature = "v2")]
+use common_utils::types;
 use common_utils::{id_type, types::MinorUnit};
 use diesel_models::enums as storage_enums;
+#[cfg(feature = "v2")]
+use diesel_models::payment_attempt;
+#[cfg(feature = "v2")]
+use hyperswitch_domain_models::{
+    address, payments::payment_attempt::PaymentAttemptFeatureMetadata,
+    router_response_types::RedirectForm,
+};
 use hyperswitch_domain_models::{
     mandates::MandateDetails, payments::payment_attempt::PaymentAttempt,
 };
 use time::OffsetDateTime;
 
+#[cfg(feature = "v1")]
 #[serde_with::skip_serializing_none]
 #[derive(serde::Serialize, Debug)]
 pub struct KafkaPaymentAttemptEvent<'a> {
@@ -124,68 +135,243 @@ impl<'a> KafkaPaymentAttemptEvent<'a> {
     }
 }
 
+#[cfg(feature = "v2")]
+#[serde_with::skip_serializing_none]
+#[derive(serde::Serialize, Debug)]
+pub struct KafkaPaymentAttemptEvent<'a> {
+    pub payment_id: &'a id_type::GlobalPaymentId,
+    pub merchant_id: &'a id_type::MerchantId,
+    pub attempt_id: &'a id_type::GlobalAttemptId,
+    pub status: storage_enums::AttemptStatus,
+    pub amount: MinorUnit,
+    pub connector: Option<&'a String>,
+    pub error_message: Option<&'a String>,
+    pub surcharge_amount: Option<MinorUnit>,
+    pub tax_amount: Option<MinorUnit>,
+    pub payment_method_id: Option<&'a id_type::GlobalPaymentMethodId>,
+    pub payment_method: storage_enums::PaymentMethod,
+    pub connector_transaction_id: Option<&'a String>,
+    pub authentication_type: storage_enums::AuthenticationType,
+    #[serde(with = "time::serde::timestamp")]
+    pub created_at: OffsetDateTime,
+    #[serde(with = "time::serde::timestamp")]
+    pub modified_at: OffsetDateTime,
+    #[serde(default, with = "time::serde::timestamp::option")]
+    pub last_synced: Option<OffsetDateTime>,
+    pub cancellation_reason: Option<&'a String>,
+    pub amount_to_capture: Option<MinorUnit>,
+    pub browser_info: Option<&'a types::BrowserInformation>,
+    pub error_code: Option<&'a String>,
+    pub connector_metadata: Option<String>,
+    // TODO: These types should implement copy ideally
+    pub payment_experience: Option<&'a storage_enums::PaymentExperience>,
+    pub payment_method_type: &'a storage_enums::PaymentMethodType,
+    pub payment_method_data: Option<String>,
+    pub error_reason: Option<&'a String>,
+    pub multiple_capture_count: Option<i16>,
+    pub amount_capturable: MinorUnit,
+    pub merchant_connector_id: Option<&'a id_type::MerchantConnectorAccountId>,
+    pub net_amount: MinorUnit,
+    pub unified_code: Option<&'a String>,
+    pub unified_message: Option<&'a String>,
+    pub client_source: Option<&'a String>,
+    pub client_version: Option<&'a String>,
+    pub profile_id: &'a id_type::ProfileId,
+    pub organization_id: &'a id_type::OrganizationId,
+    pub card_network: Option<String>,
+    pub card_discovery: Option<String>,
+
+    pub connector_payment_id: Option<types::ConnectorTransactionId>,
+    pub payment_token: Option<String>,
+    pub preprocessing_step_id: Option<String>,
+    pub connector_response_reference_id: Option<String>,
+    pub updated_by: &'a String,
+    pub encoded_data: Option<&'a masking::Secret<String>>,
+    pub external_three_ds_authentication_attempted: Option<bool>,
+    pub authentication_connector: Option<String>,
+    pub authentication_id: Option<String>,
+    pub fingerprint_id: Option<String>,
+    pub customer_acceptance: Option<&'a masking::Secret<serde_json::Value>>,
+    pub shipping_cost: Option<MinorUnit>,
+    pub order_tax_amount: Option<MinorUnit>,
+    pub charges: Option<payments::ConnectorChargeResponseData>,
+    pub processor_merchant_id: &'a id_type::MerchantId,
+    pub created_by: Option<&'a types::CreatedBy>,
+    pub payment_method_type_v2: storage_enums::PaymentMethod,
+    pub payment_method_subtype: storage_enums::PaymentMethodType,
+    pub routing_result: Option<serde_json::Value>,
+    pub authentication_applied: Option<common_enums::AuthenticationType>,
+    pub external_reference_id: Option<String>,
+    pub tax_on_surcharge: Option<MinorUnit>,
+    pub payment_method_billing_address: Option<masking::Secret<&'a address::Address>>, // adjusted from Encryption
+    pub redirection_data: Option<&'a RedirectForm>,
+    pub connector_payment_data: Option<String>,
+    pub connector_token_details: Option<&'a payment_attempt::ConnectorTokenDetails>,
+    pub feature_metadata: Option<&'a PaymentAttemptFeatureMetadata>,
+    pub network_advice_code: Option<String>,
+    pub network_decline_code: Option<String>,
+    pub network_error_message: Option<String>,
+    pub connector_request_reference_id: Option<String>,
+}
+
 #[cfg(feature = "v2")]
 impl<'a> KafkaPaymentAttemptEvent<'a> {
     pub fn from_storage(attempt: &'a PaymentAttempt) -> Self {
-        todo!()
-        // Self {
-        //     payment_id: &attempt.payment_id,
-        //     merchant_id: &attempt.merchant_id,
-        //     attempt_id: &attempt.attempt_id,
-        //     status: attempt.status,
-        //     amount: attempt.amount,
-        //     currency: attempt.currency,
-        //     save_to_locker: attempt.save_to_locker,
-        //     connector: attempt.connector.as_ref(),
-        //     error_message: attempt.error_message.as_ref(),
-        //     offer_amount: attempt.offer_amount,
-        //     surcharge_amount: attempt.surcharge_amount,
-        //     tax_amount: attempt.tax_amount,
-        //     payment_method_id: attempt.payment_method_id.as_ref(),
-        //     payment_method: attempt.payment_method,
-        //     connector_transaction_id: attempt.connector_transaction_id.as_ref(),
-        //     capture_method: attempt.capture_method,
-        //     capture_on: attempt.capture_on.map(|i| i.assume_utc()),
-        //     confirm: attempt.confirm,
-        //     authentication_type: attempt.authentication_type,
-        //     created_at: attempt.created_at.assume_utc(),
-        //     modified_at: attempt.modified_at.assume_utc(),
-        //     last_synced: attempt.last_synced.map(|i| i.assume_utc()),
-        //     cancellation_reason: attempt.cancellation_reason.as_ref(),
-        //     amount_to_capture: attempt.amount_to_capture,
-        //     mandate_id: attempt.mandate_id.as_ref(),
-        //     browser_info: attempt.browser_info.as_ref().map(|v| v.to_string()),
-        //     error_code: attempt.error_code.as_ref(),
-        //     connector_metadata: attempt.connector_metadata.as_ref().map(|v| v.to_string()),
-        //     payment_experience: attempt.payment_experience.as_ref(),
-        //     payment_method_type: attempt.payment_method_type.as_ref(),
-        //     payment_method_data: attempt.payment_method_data.as_ref().map(|v| v.to_string()),
-        //     error_reason: attempt.error_reason.as_ref(),
-        //     multiple_capture_count: attempt.multiple_capture_count,
-        //     amount_capturable: attempt.amount_capturable,
-        //     merchant_connector_id: attempt.merchant_connector_id.as_ref(),
-        //     net_amount: attempt.net_amount,
-        //     unified_code: attempt.unified_code.as_ref(),
-        //     unified_message: attempt.unified_message.as_ref(),
-        //     mandate_data: attempt.mandate_data.as_ref(),
-        //     client_source: attempt.client_source.as_ref(),
-        //     client_version: attempt.client_version.as_ref(),
-        //     profile_id: &attempt.profile_id,
-        //     organization_id: &attempt.organization_id,
-        //     card_network: attempt
-        //         .payment_method_data
-        //         .as_ref()
-        //         .and_then(|data| data.as_object())
-        //         .and_then(|pm| pm.get("card"))
-        //         .and_then(|data| data.as_object())
-        //         .and_then(|card| card.get("card_network"))
-        //         .and_then(|network| network.as_str())
-        //         .map(|network| network.to_string()),
-        // }
+        use masking::PeekInterface;
+        let PaymentAttempt {
+            payment_id,
+            merchant_id,
+            amount_details,
+            status,
+            connector,
+            error,
+            authentication_type,
+            created_at,
+            modified_at,
+            last_synced,
+            cancellation_reason,
+            browser_info,
+            payment_token,
+            connector_metadata,
+            payment_experience,
+            payment_method_data,
+            routing_result,
+            preprocessing_step_id,
+            multiple_capture_count,
+            connector_response_reference_id,
+            updated_by,
+            redirection_data,
+            encoded_data,
+            merchant_connector_id,
+            external_three_ds_authentication_attempted,
+            authentication_connector,
+            authentication_id,
+            fingerprint_id,
+            client_source,
+            client_version,
+            customer_acceptance,
+            profile_id,
+            organization_id,
+            payment_method_type,
+            payment_method_id,
+            connector_payment_id,
+            payment_method_subtype,
+            authentication_applied,
+            external_reference_id,
+            payment_method_billing_address,
+            id,
+            connector_token_details,
+            card_discovery,
+            charges,
+            feature_metadata,
+            processor_merchant_id,
+            created_by,
+            connector_request_reference_id,
+        } = attempt;
+
+        let (connector_payment_id, connector_payment_data) = connector_payment_id
+            .clone()
+            .map(types::ConnectorTransactionId::form_id_and_data)
+            .map(|(txn_id, txn_data)| (Some(txn_id), txn_data))
+            .unwrap_or((None, None));
+
+        Self {
+            payment_id,
+            merchant_id,
+            attempt_id: id,
+            status: *status,
+            amount: amount_details.get_net_amount(),
+            connector: connector.as_ref(),
+            error_message: error.as_ref().map(|error_details| &error_details.message),
+            surcharge_amount: amount_details.get_surcharge_amount(),
+            tax_amount: amount_details.get_tax_on_surcharge(),
+            payment_method_id: payment_method_id.as_ref(),
+            payment_method: *payment_method_type,
+            connector_transaction_id: connector_response_reference_id.as_ref(),
+            authentication_type: *authentication_type,
+            created_at: created_at.assume_utc(),
+            modified_at: modified_at.assume_utc(),
+            last_synced: last_synced.map(|i| i.assume_utc()),
+            cancellation_reason: cancellation_reason.as_ref(),
+            amount_to_capture: amount_details.get_amount_to_capture(),
+            browser_info: browser_info.as_ref(),
+            error_code: error.as_ref().map(|error_details| &error_details.code),
+            connector_metadata: connector_metadata.as_ref().map(|v| v.peek().to_string()),
+            payment_experience: payment_experience.as_ref(),
+            payment_method_type: payment_method_subtype,
+            payment_method_data: payment_method_data.as_ref().map(|v| v.peek().to_string()),
+            error_reason: error
+                .as_ref()
+                .and_then(|error_details| error_details.reason.as_ref()),
+            multiple_capture_count: *multiple_capture_count,
+            amount_capturable: amount_details.get_amount_capturable(),
+            merchant_connector_id: merchant_connector_id.as_ref(),
+            net_amount: amount_details.get_net_amount(),
+            unified_code: error
+                .as_ref()
+                .and_then(|error_details| error_details.unified_code.as_ref()),
+            unified_message: error
+                .as_ref()
+                .and_then(|error_details| error_details.unified_message.as_ref()),
+            client_source: client_source.as_ref(),
+            client_version: client_version.as_ref(),
+            profile_id,
+            organization_id,
+            card_network: payment_method_data
+                .as_ref()
+                .map(|data| data.peek())
+                .and_then(|data| data.as_object())
+                .and_then(|pm| pm.get("card"))
+                .and_then(|data| data.as_object())
+                .and_then(|card| card.get("card_network"))
+                .and_then(|network| network.as_str())
+                .map(|network| network.to_string()),
+            card_discovery: card_discovery.map(|discovery| discovery.to_string()),
+            payment_token: payment_token.clone(),
+            preprocessing_step_id: preprocessing_step_id.clone(),
+            connector_response_reference_id: connector_response_reference_id.clone(),
+            updated_by,
+            encoded_data: encoded_data.as_ref(),
+            external_three_ds_authentication_attempted: *external_three_ds_authentication_attempted,
+            authentication_connector: authentication_connector.clone(),
+            authentication_id: authentication_id.clone(),
+            fingerprint_id: fingerprint_id.clone(),
+            customer_acceptance: customer_acceptance.as_ref(),
+            shipping_cost: amount_details.get_shipping_cost(),
+            order_tax_amount: amount_details.get_order_tax_amount(),
+            charges: charges.clone(),
+            processor_merchant_id,
+            created_by: created_by.as_ref(),
+            payment_method_type_v2: *payment_method_type,
+            connector_payment_id: connector_payment_id.as_ref().cloned(),
+            payment_method_subtype: *payment_method_subtype,
+            routing_result: routing_result.clone(),
+            authentication_applied: *authentication_applied,
+            external_reference_id: external_reference_id.clone(),
+            tax_on_surcharge: amount_details.get_tax_on_surcharge(),
+            payment_method_billing_address: payment_method_billing_address
+                .as_ref()
+                .map(|v| masking::Secret::new(v.get_inner())),
+            redirection_data: redirection_data.as_ref(),
+            connector_payment_data,
+            connector_token_details: connector_token_details.as_ref(),
+            feature_metadata: feature_metadata.as_ref(),
+            network_advice_code: error
+                .as_ref()
+                .and_then(|details| details.network_advice_code.clone()),
+            network_decline_code: error
+                .as_ref()
+                .and_then(|details| details.network_decline_code.clone()),
+            network_error_message: error
+                .as_ref()
+                .and_then(|details| details.network_error_message.clone()),
+            connector_request_reference_id: connector_request_reference_id.clone(),
+        }
     }
 }
 
 impl super::KafkaMessage for KafkaPaymentAttemptEvent<'_> {
+    #[cfg(feature = "v1")]
     fn key(&self) -> String {
         format!(
             "{}_{}_{}",
@@ -194,6 +380,15 @@ impl super::KafkaMessage for KafkaPaymentAttemptEvent<'_> {
             self.attempt_id
         )
     }
+    #[cfg(feature = "v2")]
+    fn key(&self) -> String {
+        format!(
+            "{}_{}_{}",
+            self.merchant_id.get_string_repr(),
+            self.payment_id.get_string_repr(),
+            self.attempt_id.get_string_repr()
+        )
+    }
 
     fn event_type(&self) -> crate::events::EventType {
         crate::events::EventType::PaymentAttempt
diff --git a/crates/router/src/services/kafka/payment_intent.rs b/crates/router/src/services/kafka/payment_intent.rs
index b438a07b9ca..f0b4438bd2e 100644
--- a/crates/router/src/services/kafka/payment_intent.rs
+++ b/crates/router/src/services/kafka/payment_intent.rs
@@ -1,6 +1,20 @@
-use common_utils::{crypto::Encryptable, hashing::HashedString, id_type, pii, types::MinorUnit};
+#[cfg(feature = "v2")]
+use ::common_types::{payments, primitive_wrappers::RequestExtendedAuthorizationBool};
+#[cfg(feature = "v2")]
+use common_enums;
+#[cfg(feature = "v2")]
+use common_enums::RequestIncrementalAuthorization;
+use common_utils::{
+    crypto::Encryptable, hashing::HashedString, id_type, pii, types as common_types,
+};
 use diesel_models::enums as storage_enums;
+#[cfg(feature = "v2")]
+use diesel_models::{types as diesel_types, PaymentLinkConfigRequestForPayments};
+#[cfg(feature = "v2")]
+use diesel_models::{types::OrderDetailsWithAmount, TaxDetails};
 use hyperswitch_domain_models::payments::PaymentIntent;
+#[cfg(feature = "v2")]
+use hyperswitch_domain_models::{address, routing};
 use masking::{PeekInterface, Secret};
 use serde_json::Value;
 use time::OffsetDateTime;
@@ -11,9 +25,9 @@ pub struct KafkaPaymentIntent<'a> {
     pub payment_id: &'a id_type::PaymentId,
     pub merchant_id: &'a id_type::MerchantId,
     pub status: storage_enums::IntentStatus,
-    pub amount: MinorUnit,
+    pub amount: common_types::MinorUnit,
     pub currency: Option<storage_enums::Currency>,
-    pub amount_captured: Option<MinorUnit>,
+    pub amount_captured: Option<common_types::MinorUnit>,
     pub customer_id: Option<&'a id_type::CustomerId>,
     pub description: Option<&'a String>,
     pub return_url: Option<&'a String>,
@@ -46,41 +60,6 @@ pub struct KafkaPaymentIntent<'a> {
     infra_values: Option<Value>,
 }
 
-#[cfg(feature = "v2")]
-#[derive(serde::Serialize, Debug)]
-pub struct KafkaPaymentIntent<'a> {
-    pub id: &'a id_type::PaymentId,
-    pub merchant_id: &'a id_type::MerchantId,
-    pub status: storage_enums::IntentStatus,
-    pub amount: MinorUnit,
-    pub currency: storage_enums::Currency,
-    pub amount_captured: Option<MinorUnit>,
-    pub customer_id: Option<&'a id_type::CustomerId>,
-    pub description: Option<&'a String>,
-    pub return_url: Option<&'a String>,
-    pub metadata: Option<String>,
-    pub statement_descriptor: Option<&'a String>,
-    #[serde(with = "time::serde::timestamp")]
-    pub created_at: OffsetDateTime,
-    #[serde(with = "time::serde::timestamp")]
-    pub modified_at: OffsetDateTime,
-    #[serde(default, with = "time::serde::timestamp::option")]
-    pub last_synced: Option<OffsetDateTime>,
-    pub setup_future_usage: Option<storage_enums::FutureUsage>,
-    pub off_session: Option<bool>,
-    pub client_secret: Option<&'a String>,
-    pub active_attempt_id: String,
-    pub attempt_count: i16,
-    pub profile_id: &'a id_type::ProfileId,
-    pub payment_confirm_source: Option<storage_enums::PaymentSource>,
-    pub billing_details: Option<Encryptable<Secret<Value>>>,
-    pub shipping_details: Option<Encryptable<Secret<Value>>>,
-    pub customer_email: Option<HashedString<pii::EmailStrategy>>,
-    pub feature_metadata: Option<&'a Value>,
-    pub merchant_order_reference_id: Option<&'a String>,
-    pub organization_id: &'a id_type::OrganizationId,
-}
-
 #[cfg(feature = "v1")]
 impl<'a> KafkaPaymentIntent<'a> {
     pub fn from_storage(intent: &'a PaymentIntent, infra_values: Option<Value>) -> Self {
@@ -128,46 +107,204 @@ impl<'a> KafkaPaymentIntent<'a> {
     }
 }
 
+#[cfg(feature = "v2")]
+#[derive(serde::Serialize, Debug)]
+pub struct KafkaPaymentIntent<'a> {
+    pub payment_id: &'a id_type::GlobalPaymentId,
+    pub merchant_id: &'a id_type::MerchantId,
+    pub status: storage_enums::IntentStatus,
+    pub amount: common_types::MinorUnit,
+    pub currency: storage_enums::Currency,
+    pub amount_captured: Option<common_types::MinorUnit>,
+    pub customer_id: Option<&'a id_type::GlobalCustomerId>,
+    pub description: Option<&'a common_types::Description>,
+    pub return_url: Option<&'a common_types::Url>,
+    pub metadata: Option<&'a Secret<Value>>,
+    pub statement_descriptor: Option<&'a common_types::StatementDescriptor>,
+    #[serde(with = "time::serde::timestamp")]
+    pub created_at: OffsetDateTime,
+    #[serde(with = "time::serde::timestamp")]
+    pub modified_at: OffsetDateTime,
+    #[serde(default, with = "time::serde::timestamp::option")]
+    pub last_synced: Option<OffsetDateTime>,
+    pub setup_future_usage: storage_enums::FutureUsage,
+    pub off_session: bool,
+    pub active_attempt_id: Option<&'a id_type::GlobalAttemptId>,
+    pub attempt_count: i16,
+    pub profile_id: &'a id_type::ProfileId,
+    pub customer_email: Option<HashedString<pii::EmailStrategy>>,
+    pub feature_metadata: Option<&'a diesel_types::FeatureMetadata>,
+    pub organization_id: &'a id_type::OrganizationId,
+    pub order_details: Option<&'a Vec<Secret<OrderDetailsWithAmount>>>,
+
+    pub allowed_payment_method_types: Option<&'a Vec<common_enums::PaymentMethodType>>,
+    pub connector_metadata: Option<&'a Secret<Value>>,
+    pub payment_link_id: Option<&'a String>,
+    pub updated_by: &'a String,
+    pub surcharge_applicable: Option<bool>,
+    pub request_incremental_authorization: RequestIncrementalAuthorization,
+    pub authorization_count: Option<i32>,
+    #[serde(with = "time::serde::timestamp")]
+    pub session_expiry: OffsetDateTime,
+    pub request_external_three_ds_authentication: common_enums::External3dsAuthenticationRequest,
+    pub frm_metadata: Option<Secret<&'a Value>>,
+    pub customer_details: Option<Secret<&'a Value>>,
+    pub shipping_cost: Option<common_types::MinorUnit>,
+    pub tax_details: Option<TaxDetails>,
+    pub skip_external_tax_calculation: bool,
+    pub request_extended_authorization: Option<RequestExtendedAuthorizationBool>,
+    pub psd2_sca_exemption_type: Option<storage_enums::ScaExemptionType>,
+    pub split_payments: Option<&'a payments::SplitPaymentsRequest>,
+    pub platform_merchant_id: Option<&'a id_type::MerchantId>,
+    pub force_3ds_challenge: Option<bool>,
+    pub force_3ds_challenge_trigger: Option<bool>,
+    pub processor_merchant_id: &'a id_type::MerchantId,
+    pub created_by: Option<&'a common_types::CreatedBy>,
+    pub is_iframe_redirection_enabled: Option<bool>,
+    pub merchant_reference_id: Option<&'a id_type::PaymentReferenceId>,
+    pub capture_method: storage_enums::CaptureMethod,
+    pub authentication_type: Option<common_enums::AuthenticationType>,
+    pub prerouting_algorithm: Option<&'a routing::PaymentRoutingInfo>,
+    pub surcharge_amount: Option<common_types::MinorUnit>,
+    pub billing_address: Option<Secret<&'a address::Address>>,
+    pub shipping_address: Option<Secret<&'a address::Address>>,
+    pub tax_on_surcharge: Option<common_types::MinorUnit>,
+    pub frm_merchant_decision: Option<common_enums::MerchantDecision>,
+    pub enable_payment_link: common_enums::EnablePaymentLinkRequest,
+    pub apply_mit_exemption: common_enums::MitExemptionRequest,
+    pub customer_present: common_enums::PresenceOfCustomerDuringPayment,
+    pub routing_algorithm_id: Option<&'a id_type::RoutingId>,
+    pub payment_link_config: Option<&'a PaymentLinkConfigRequestForPayments>,
+
+    #[serde(flatten)]
+    infra_values: Option<Value>,
+}
+
 #[cfg(feature = "v2")]
 impl<'a> KafkaPaymentIntent<'a> {
     pub fn from_storage(intent: &'a PaymentIntent, infra_values: Option<Value>) -> Self {
-        // Self {
-        //     id: &intent.id,
-        //     merchant_id: &intent.merchant_id,
-        //     status: intent.status,
-        //     amount: intent.amount,
-        //     currency: intent.currency,
-        //     amount_captured: intent.amount_captured,
-        //     customer_id: intent.customer_id.as_ref(),
-        //     description: intent.description.as_ref(),
-        //     return_url: intent.return_url.as_ref(),
-        //     metadata: intent.metadata.as_ref().map(|x| x.to_string()),
-        //     statement_descriptor: intent.statement_descriptor.as_ref(),
-        //     created_at: intent.created_at.assume_utc(),
-        //     modified_at: intent.modified_at.assume_utc(),
-        //     last_synced: intent.last_synced.map(|i| i.assume_utc()),
-        //     setup_future_usage: intent.setup_future_usage,
-        //     off_session: intent.off_session,
-        //     client_secret: intent.client_secret.as_ref(),
-        //     active_attempt_id: intent.active_attempt.get_id(),
-        //     attempt_count: intent.attempt_count,
-        //     profile_id: &intent.profile_id,
-        //     payment_confirm_source: intent.payment_confirm_source,
-        //     // TODO: use typed information here to avoid PII logging
-        //     billing_details: None,
-        //     shipping_details: None,
-        //     customer_email: intent
-        //         .customer_details
-        //         .as_ref()
-        //         .and_then(|value| value.get_inner().peek().as_object())
-        //         .and_then(|obj| obj.get("email"))
-        //         .and_then(|email| email.as_str())
-        //         .map(|email| HashedString::from(Secret::new(email.to_string()))),
-        //     feature_metadata: intent.feature_metadata.as_ref(),
-        //     merchant_order_reference_id: intent.merchant_order_reference_id.as_ref(),
-        //     organization_id: &intent.organization_id,
-        // }
-        todo!()
+        let PaymentIntent {
+            id,
+            merchant_id,
+            status,
+            amount_details,
+            amount_captured,
+            customer_id,
+            description,
+            return_url,
+            metadata,
+            statement_descriptor,
+            created_at,
+            modified_at,
+            last_synced,
+            setup_future_usage,
+            active_attempt_id,
+            order_details,
+            allowed_payment_method_types,
+            connector_metadata,
+            feature_metadata,
+            attempt_count,
+            profile_id,
+            payment_link_id,
+            frm_merchant_decision,
+            updated_by,
+            request_incremental_authorization,
+            authorization_count,
+            session_expiry,
+            request_external_three_ds_authentication,
+            frm_metadata,
+            customer_details,
+            merchant_reference_id,
+            billing_address,
+            shipping_address,
+            capture_method,
+            authentication_type,
+            prerouting_algorithm,
+            organization_id,
+            enable_payment_link,
+            apply_mit_exemption,
+            customer_present,
+            payment_link_config,
+            routing_algorithm_id,
+            split_payments,
+            force_3ds_challenge,
+            force_3ds_challenge_trigger,
+            processor_merchant_id,
+            created_by,
+            is_iframe_redirection_enabled,
+        } = intent;
+
+        Self {
+            payment_id: id,
+            merchant_id,
+            status: *status,
+            amount: amount_details.order_amount,
+            currency: amount_details.currency,
+            amount_captured: *amount_captured,
+            customer_id: customer_id.as_ref(),
+            description: description.as_ref(),
+            return_url: return_url.as_ref(),
+            metadata: metadata.as_ref(),
+            statement_descriptor: statement_descriptor.as_ref(),
+            created_at: created_at.assume_utc(),
+            modified_at: modified_at.assume_utc(),
+            last_synced: last_synced.map(|t| t.assume_utc()),
+            setup_future_usage: *setup_future_usage,
+            off_session: setup_future_usage.is_off_session(),
+            active_attempt_id: active_attempt_id.as_ref(),
+            attempt_count: *attempt_count,
+            profile_id,
+            customer_email: None,
+            feature_metadata: feature_metadata.as_ref(),
+            organization_id,
+            order_details: order_details.as_ref(),
+            allowed_payment_method_types: allowed_payment_method_types.as_ref(),
+            connector_metadata: connector_metadata.as_ref(),
+            payment_link_id: payment_link_id.as_ref(),
+            updated_by,
+            surcharge_applicable: None,
+            request_incremental_authorization: *request_incremental_authorization,
+            authorization_count: *authorization_count,
+            session_expiry: session_expiry.assume_utc(),
+            request_external_three_ds_authentication: *request_external_three_ds_authentication,
+            frm_metadata: frm_metadata
+                .as_ref()
+                .map(|frm_metadata| frm_metadata.as_ref()),
+            customer_details: customer_details
+                .as_ref()
+                .map(|customer_details| customer_details.get_inner().as_ref()),
+            shipping_cost: amount_details.shipping_cost,
+            tax_details: amount_details.tax_details.clone(),
+            skip_external_tax_calculation: amount_details.get_external_tax_action_as_bool(),
+            request_extended_authorization: None,
+            psd2_sca_exemption_type: None,
+            split_payments: split_payments.as_ref(),
+            platform_merchant_id: None,
+            force_3ds_challenge: *force_3ds_challenge,
+            force_3ds_challenge_trigger: *force_3ds_challenge_trigger,
+            processor_merchant_id,
+            created_by: created_by.as_ref(),
+            is_iframe_redirection_enabled: *is_iframe_redirection_enabled,
+            merchant_reference_id: merchant_reference_id.as_ref(),
+            billing_address: billing_address
+                .as_ref()
+                .map(|billing_address| Secret::new(billing_address.get_inner())),
+            shipping_address: shipping_address
+                .as_ref()
+                .map(|shipping_address| Secret::new(shipping_address.get_inner())),
+            capture_method: *capture_method,
+            authentication_type: *authentication_type,
+            prerouting_algorithm: prerouting_algorithm.as_ref(),
+            surcharge_amount: amount_details.surcharge_amount,
+            tax_on_surcharge: amount_details.tax_on_surcharge,
+            frm_merchant_decision: *frm_merchant_decision,
+            enable_payment_link: *enable_payment_link,
+            apply_mit_exemption: *apply_mit_exemption,
+            customer_present: *customer_present,
+            routing_algorithm_id: routing_algorithm_id.as_ref(),
+            payment_link_config: payment_link_config.as_ref(),
+            infra_values,
+        }
     }
 }
 
@@ -178,8 +315,8 @@ impl KafkaPaymentIntent<'_> {
     }
 
     #[cfg(feature = "v2")]
-    fn get_id(&self) -> &id_type::PaymentId {
-        self.id
+    fn get_id(&self) -> &id_type::GlobalPaymentId {
+        self.payment_id
     }
 }
 
diff --git a/crates/router/src/services/kafka/payment_intent_event.rs b/crates/router/src/services/kafka/payment_intent_event.rs
index b7715a591ef..f7d1bca5b32 100644
--- a/crates/router/src/services/kafka/payment_intent_event.rs
+++ b/crates/router/src/services/kafka/payment_intent_event.rs
@@ -1,6 +1,18 @@
-use common_utils::{crypto::Encryptable, hashing::HashedString, id_type, pii, types::MinorUnit};
+#[cfg(feature = "v2")]
+use ::common_types::{payments, primitive_wrappers::RequestExtendedAuthorizationBool};
+#[cfg(feature = "v2")]
+use common_enums::{self, RequestIncrementalAuthorization};
+use common_utils::{
+    crypto::Encryptable, hashing::HashedString, id_type, pii, types as common_types,
+};
 use diesel_models::enums as storage_enums;
+#[cfg(feature = "v2")]
+use diesel_models::{types as diesel_types, PaymentLinkConfigRequestForPayments};
+#[cfg(feature = "v2")]
+use diesel_models::{types::OrderDetailsWithAmount, TaxDetails};
 use hyperswitch_domain_models::payments::PaymentIntent;
+#[cfg(feature = "v2")]
+use hyperswitch_domain_models::{address, routing};
 use masking::{PeekInterface, Secret};
 use serde_json::Value;
 use time::OffsetDateTime;
@@ -12,9 +24,9 @@ pub struct KafkaPaymentIntentEvent<'a> {
     pub payment_id: &'a id_type::PaymentId,
     pub merchant_id: &'a id_type::MerchantId,
     pub status: storage_enums::IntentStatus,
-    pub amount: MinorUnit,
+    pub amount: common_types::MinorUnit,
     pub currency: Option<storage_enums::Currency>,
-    pub amount_captured: Option<MinorUnit>,
+    pub amount_captured: Option<common_types::MinorUnit>,
     pub customer_id: Option<&'a id_type::CustomerId>,
     pub description: Option<&'a String>,
     pub return_url: Option<&'a String>,
@@ -51,36 +63,74 @@ pub struct KafkaPaymentIntentEvent<'a> {
 #[serde_with::skip_serializing_none]
 #[derive(serde::Serialize, Debug)]
 pub struct KafkaPaymentIntentEvent<'a> {
-    pub id: &'a id_type::PaymentId,
+    pub payment_id: &'a id_type::GlobalPaymentId,
     pub merchant_id: &'a id_type::MerchantId,
     pub status: storage_enums::IntentStatus,
-    pub amount: MinorUnit,
+    pub amount: common_types::MinorUnit,
     pub currency: storage_enums::Currency,
-    pub amount_captured: Option<MinorUnit>,
-    pub customer_id: Option<&'a id_type::CustomerId>,
-    pub description: Option<&'a String>,
-    pub return_url: Option<&'a String>,
-    pub metadata: Option<String>,
-    pub statement_descriptor: Option<&'a String>,
-    #[serde(with = "time::serde::timestamp::nanoseconds")]
+    pub amount_captured: Option<common_types::MinorUnit>,
+    pub customer_id: Option<&'a id_type::GlobalCustomerId>,
+    pub description: Option<&'a common_types::Description>,
+    pub return_url: Option<&'a common_types::Url>,
+    pub metadata: Option<&'a Secret<Value>>,
+    pub statement_descriptor: Option<&'a common_types::StatementDescriptor>,
+    #[serde(with = "time::serde::timestamp")]
     pub created_at: OffsetDateTime,
-    #[serde(with = "time::serde::timestamp::nanoseconds")]
+    #[serde(with = "time::serde::timestamp")]
     pub modified_at: OffsetDateTime,
-    #[serde(default, with = "time::serde::timestamp::nanoseconds::option")]
+    #[serde(default, with = "time::serde::timestamp::option")]
     pub last_synced: Option<OffsetDateTime>,
-    pub setup_future_usage: Option<storage_enums::FutureUsage>,
-    pub off_session: Option<bool>,
-    pub client_secret: Option<&'a String>,
-    pub active_attempt_id: String,
+    pub setup_future_usage: storage_enums::FutureUsage,
+    pub off_session: bool,
+    pub active_attempt_id: Option<&'a id_type::GlobalAttemptId>,
     pub attempt_count: i16,
     pub profile_id: &'a id_type::ProfileId,
-    pub payment_confirm_source: Option<storage_enums::PaymentSource>,
-    pub billing_details: Option<Encryptable<Secret<Value>>>,
-    pub shipping_details: Option<Encryptable<Secret<Value>>>,
     pub customer_email: Option<HashedString<pii::EmailStrategy>>,
-    pub feature_metadata: Option<&'a Value>,
-    pub merchant_order_reference_id: Option<&'a String>,
+    pub feature_metadata: Option<&'a diesel_types::FeatureMetadata>,
     pub organization_id: &'a id_type::OrganizationId,
+    pub order_details: Option<&'a Vec<Secret<OrderDetailsWithAmount>>>,
+
+    pub allowed_payment_method_types: Option<&'a Vec<common_enums::PaymentMethodType>>,
+    pub connector_metadata: Option<&'a Secret<Value>>,
+    pub payment_link_id: Option<&'a String>,
+    pub updated_by: &'a String,
+    pub surcharge_applicable: Option<bool>,
+    pub request_incremental_authorization: RequestIncrementalAuthorization,
+    pub authorization_count: Option<i32>,
+    #[serde(with = "time::serde::timestamp")]
+    pub session_expiry: OffsetDateTime,
+    pub request_external_three_ds_authentication: common_enums::External3dsAuthenticationRequest,
+    pub frm_metadata: Option<Secret<&'a Value>>,
+    pub customer_details: Option<Secret<&'a Value>>,
+    pub shipping_cost: Option<common_types::MinorUnit>,
+    pub tax_details: Option<TaxDetails>,
+    pub skip_external_tax_calculation: bool,
+    pub request_extended_authorization: Option<RequestExtendedAuthorizationBool>,
+    pub psd2_sca_exemption_type: Option<storage_enums::ScaExemptionType>,
+    pub split_payments: Option<&'a payments::SplitPaymentsRequest>,
+    pub platform_merchant_id: Option<&'a id_type::MerchantId>,
+    pub force_3ds_challenge: Option<bool>,
+    pub force_3ds_challenge_trigger: Option<bool>,
+    pub processor_merchant_id: &'a id_type::MerchantId,
+    pub created_by: Option<&'a common_types::CreatedBy>,
+    pub is_iframe_redirection_enabled: Option<bool>,
+    pub merchant_reference_id: Option<&'a id_type::PaymentReferenceId>,
+    pub capture_method: storage_enums::CaptureMethod,
+    pub authentication_type: Option<common_enums::AuthenticationType>,
+    pub prerouting_algorithm: Option<&'a routing::PaymentRoutingInfo>,
+    pub surcharge_amount: Option<common_types::MinorUnit>,
+    pub billing_address: Option<Secret<&'a address::Address>>,
+    pub shipping_address: Option<Secret<&'a address::Address>>,
+    pub tax_on_surcharge: Option<common_types::MinorUnit>,
+    pub frm_merchant_decision: Option<common_enums::MerchantDecision>,
+    pub enable_payment_link: common_enums::EnablePaymentLinkRequest,
+    pub apply_mit_exemption: common_enums::MitExemptionRequest,
+    pub customer_present: common_enums::PresenceOfCustomerDuringPayment,
+    pub routing_algorithm_id: Option<&'a id_type::RoutingId>,
+    pub payment_link_config: Option<&'a PaymentLinkConfigRequestForPayments>,
+
+    #[serde(flatten)]
+    infra_values: Option<Value>,
 }
 
 impl KafkaPaymentIntentEvent<'_> {
@@ -90,8 +140,8 @@ impl KafkaPaymentIntentEvent<'_> {
     }
 
     #[cfg(feature = "v2")]
-    pub fn get_id(&self) -> &id_type::PaymentId {
-        self.id
+    pub fn get_id(&self) -> &id_type::GlobalPaymentId {
+        self.payment_id
     }
 }
 
@@ -145,43 +195,128 @@ impl<'a> KafkaPaymentIntentEvent<'a> {
 #[cfg(feature = "v2")]
 impl<'a> KafkaPaymentIntentEvent<'a> {
     pub fn from_storage(intent: &'a PaymentIntent, infra_values: Option<Value>) -> Self {
-        // Self {
-        //     id: &intent.id,
-        //     merchant_id: &intent.merchant_id,
-        //     status: intent.status,
-        //     amount: intent.amount,
-        //     currency: intent.currency,
-        //     amount_captured: intent.amount_captured,
-        //     customer_id: intent.customer_id.as_ref(),
-        //     description: intent.description.as_ref(),
-        //     return_url: intent.return_url.as_ref(),
-        //     metadata: intent.metadata.as_ref().map(|x| x.to_string()),
-        //     statement_descriptor: intent.statement_descriptor.as_ref(),
-        //     created_at: intent.created_at.assume_utc(),
-        //     modified_at: intent.modified_at.assume_utc(),
-        //     last_synced: intent.last_synced.map(|i| i.assume_utc()),
-        //     setup_future_usage: intent.setup_future_usage,
-        //     off_session: intent.off_session,
-        //     client_secret: intent.client_secret.as_ref(),
-        //     active_attempt_id: intent.active_attempt.get_id(),
-        //     attempt_count: intent.attempt_count,
-        //     profile_id: &intent.profile_id,
-        //     payment_confirm_source: intent.payment_confirm_source,
-        //     // TODO: use typed information here to avoid PII logging
-        //     billing_details: None,
-        //     shipping_details: None,
-        //     customer_email: intent
-        //         .customer_details
-        //         .as_ref()
-        //         .and_then(|value| value.get_inner().peek().as_object())
-        //         .and_then(|obj| obj.get("email"))
-        //         .and_then(|email| email.as_str())
-        //         .map(|email| HashedString::from(Secret::new(email.to_string()))),
-        //     feature_metadata: intent.feature_metadata.as_ref(),
-        //     merchant_order_reference_id: intent.merchant_order_reference_id.as_ref(),
-        //     organization_id: &intent.organization_id,
-        // }
-        todo!()
+        let PaymentIntent {
+            id,
+            merchant_id,
+            status,
+            amount_details,
+            amount_captured,
+            customer_id,
+            description,
+            return_url,
+            metadata,
+            statement_descriptor,
+            created_at,
+            modified_at,
+            last_synced,
+            setup_future_usage,
+            active_attempt_id,
+            order_details,
+            allowed_payment_method_types,
+            connector_metadata,
+            feature_metadata,
+            attempt_count,
+            profile_id,
+            payment_link_id,
+            frm_merchant_decision,
+            updated_by,
+            request_incremental_authorization,
+            authorization_count,
+            session_expiry,
+            request_external_three_ds_authentication,
+            frm_metadata,
+            customer_details,
+            merchant_reference_id,
+            billing_address,
+            shipping_address,
+            capture_method,
+            authentication_type,
+            prerouting_algorithm,
+            organization_id,
+            enable_payment_link,
+            apply_mit_exemption,
+            customer_present,
+            payment_link_config,
+            routing_algorithm_id,
+            split_payments,
+            force_3ds_challenge,
+            force_3ds_challenge_trigger,
+            processor_merchant_id,
+            created_by,
+            is_iframe_redirection_enabled,
+        } = intent;
+
+        Self {
+            payment_id: id,
+            merchant_id,
+            status: *status,
+            amount: amount_details.order_amount,
+            currency: amount_details.currency,
+            amount_captured: *amount_captured,
+            customer_id: customer_id.as_ref(),
+            description: description.as_ref(),
+            return_url: return_url.as_ref(),
+            metadata: metadata.as_ref(),
+            statement_descriptor: statement_descriptor.as_ref(),
+            created_at: created_at.assume_utc(),
+            modified_at: modified_at.assume_utc(),
+            last_synced: last_synced.map(|t| t.assume_utc()),
+            setup_future_usage: *setup_future_usage,
+            off_session: setup_future_usage.is_off_session(),
+            active_attempt_id: active_attempt_id.as_ref(),
+            attempt_count: *attempt_count,
+            profile_id,
+            customer_email: None,
+            feature_metadata: feature_metadata.as_ref(),
+            organization_id,
+            order_details: order_details.as_ref(),
+            allowed_payment_method_types: allowed_payment_method_types.as_ref(),
+            connector_metadata: connector_metadata.as_ref(),
+            payment_link_id: payment_link_id.as_ref(),
+            updated_by,
+            surcharge_applicable: None,
+            request_incremental_authorization: *request_incremental_authorization,
+            authorization_count: *authorization_count,
+            session_expiry: session_expiry.assume_utc(),
+            request_external_three_ds_authentication: *request_external_three_ds_authentication,
+            frm_metadata: frm_metadata
+                .as_ref()
+                .map(|frm_metadata| frm_metadata.as_ref()),
+            customer_details: customer_details
+                .as_ref()
+                .map(|customer_details| customer_details.get_inner().as_ref()),
+            shipping_cost: amount_details.shipping_cost,
+            tax_details: amount_details.tax_details.clone(),
+            skip_external_tax_calculation: amount_details.get_external_tax_action_as_bool(),
+            request_extended_authorization: None,
+            psd2_sca_exemption_type: None,
+            split_payments: split_payments.as_ref(),
+            platform_merchant_id: None,
+            force_3ds_challenge: *force_3ds_challenge,
+            force_3ds_challenge_trigger: *force_3ds_challenge_trigger,
+            processor_merchant_id,
+            created_by: created_by.as_ref(),
+            is_iframe_redirection_enabled: *is_iframe_redirection_enabled,
+            merchant_reference_id: merchant_reference_id.as_ref(),
+            billing_address: billing_address
+                .as_ref()
+                .map(|billing_address| Secret::new(billing_address.get_inner())),
+            shipping_address: shipping_address
+                .as_ref()
+                .map(|shipping_address| Secret::new(shipping_address.get_inner())),
+            capture_method: *capture_method,
+            authentication_type: *authentication_type,
+            prerouting_algorithm: prerouting_algorithm.as_ref(),
+            surcharge_amount: amount_details.surcharge_amount,
+            tax_on_surcharge: amount_details.tax_on_surcharge,
+            frm_merchant_decision: *frm_merchant_decision,
+            enable_payment_link: *enable_payment_link,
+            apply_mit_exemption: *apply_mit_exemption,
+            customer_present: *customer_present,
+            routing_algorithm_id: routing_algorithm_id.as_ref(),
+            payment_link_config: payment_link_config.as_ref(),
+            infra_values,
+        }
     }
 }
 
diff --git a/crates/router/src/services/kafka/refund.rs b/crates/router/src/services/kafka/refund.rs
index 19635b2d8c2..4e10ac4ee36 100644
--- a/crates/router/src/services/kafka/refund.rs
+++ b/crates/router/src/services/kafka/refund.rs
@@ -1,3 +1,7 @@
+#[cfg(feature = "v2")]
+use common_utils::pii;
+#[cfg(feature = "v2")]
+use common_utils::types::{self, ChargeRefunds};
 use common_utils::{
     id_type,
     types::{ConnectorTransactionIdTrait, MinorUnit},
@@ -73,13 +77,13 @@ impl<'a> KafkaRefund<'a> {
 #[cfg(feature = "v2")]
 #[derive(serde::Serialize, Debug)]
 pub struct KafkaRefund<'a> {
-    pub id: &'a id_type::GlobalRefundId,
+    pub refund_id: &'a id_type::GlobalRefundId,
     pub merchant_reference_id: &'a id_type::RefundReferenceId,
     pub payment_id: &'a id_type::GlobalPaymentId,
     pub merchant_id: &'a id_type::MerchantId,
-    pub connector_transaction_id: &'a String,
+    pub connector_transaction_id: &'a types::ConnectorTransactionId,
     pub connector: &'a String,
-    pub connector_refund_id: Option<&'a String>,
+    pub connector_refund_id: Option<&'a types::ConnectorTransactionId>,
     pub external_reference_id: Option<&'a String>,
     pub refund_type: &'a storage_enums::RefundType,
     pub total_amount: &'a MinorUnit,
@@ -99,42 +103,101 @@ pub struct KafkaRefund<'a> {
     pub refund_error_code: Option<&'a String>,
     pub profile_id: Option<&'a id_type::ProfileId>,
     pub organization_id: &'a id_type::OrganizationId,
+
+    pub metadata: Option<&'a pii::SecretSerdeValue>,
+    pub updated_by: &'a String,
+    pub merchant_connector_id: Option<&'a id_type::MerchantConnectorAccountId>,
+    pub charges: Option<&'a ChargeRefunds>,
+    pub connector_refund_data: Option<&'a String>,
+    pub connector_transaction_data: Option<&'a String>,
+    pub split_refunds: Option<&'a common_types::refunds::SplitRefund>,
+    pub unified_code: Option<&'a String>,
+    pub unified_message: Option<&'a String>,
+    pub processor_refund_data: Option<&'a String>,
+    pub processor_transaction_data: Option<&'a String>,
 }
 
 #[cfg(feature = "v2")]
 impl<'a> KafkaRefund<'a> {
     pub fn from_storage(refund: &'a Refund) -> Self {
+        let Refund {
+            payment_id,
+            merchant_id,
+            connector_transaction_id,
+            connector,
+            connector_refund_id,
+            external_reference_id,
+            refund_type,
+            total_amount,
+            currency,
+            refund_amount,
+            refund_status,
+            sent_to_gateway,
+            refund_error_message,
+            metadata,
+            refund_arn,
+            created_at,
+            modified_at,
+            description,
+            attempt_id,
+            refund_reason,
+            refund_error_code,
+            profile_id,
+            updated_by,
+            charges,
+            organization_id,
+            split_refunds,
+            unified_code,
+            unified_message,
+            processor_refund_data,
+            processor_transaction_data,
+            id,
+            merchant_reference_id,
+            connector_id,
+        } = refund;
+
         Self {
-            id: &refund.id,
-            merchant_reference_id: &refund.merchant_reference_id,
-            payment_id: &refund.payment_id,
-            merchant_id: &refund.merchant_id,
-            connector_transaction_id: refund.get_connector_transaction_id(),
-            connector: &refund.connector,
-            connector_refund_id: refund.get_optional_connector_refund_id(),
-            external_reference_id: refund.external_reference_id.as_ref(),
-            refund_type: &refund.refund_type,
-            total_amount: &refund.total_amount,
-            currency: &refund.currency,
-            refund_amount: &refund.refund_amount,
-            refund_status: &refund.refund_status,
-            sent_to_gateway: &refund.sent_to_gateway,
-            refund_error_message: refund.refund_error_message.as_ref(),
-            refund_arn: refund.refund_arn.as_ref(),
-            created_at: refund.created_at.assume_utc(),
-            modified_at: refund.modified_at.assume_utc(),
-            description: refund.description.as_ref(),
-            attempt_id: &refund.attempt_id,
-            refund_reason: refund.refund_reason.as_ref(),
-            refund_error_code: refund.refund_error_code.as_ref(),
-            profile_id: refund.profile_id.as_ref(),
-            organization_id: &refund.organization_id,
+            refund_id: id,
+            merchant_reference_id,
+            payment_id,
+            merchant_id,
+            connector_transaction_id,
+            connector,
+            connector_refund_id: connector_refund_id.as_ref(),
+            external_reference_id: external_reference_id.as_ref(),
+            refund_type,
+            total_amount,
+            currency,
+            refund_amount,
+            refund_status,
+            sent_to_gateway,
+            refund_error_message: refund_error_message.as_ref(),
+            refund_arn: refund_arn.as_ref(),
+            created_at: created_at.assume_utc(),
+            modified_at: modified_at.assume_utc(),
+            description: description.as_ref(),
+            attempt_id,
+            refund_reason: refund_reason.as_ref(),
+            refund_error_code: refund_error_code.as_ref(),
+            profile_id: profile_id.as_ref(),
+            organization_id,
+            metadata: metadata.as_ref(),
+            updated_by,
+            merchant_connector_id: connector_id.as_ref(),
+            charges: charges.as_ref(),
+            connector_refund_data: processor_refund_data.as_ref(),
+            connector_transaction_data: processor_transaction_data.as_ref(),
+            split_refunds: split_refunds.as_ref(),
+            unified_code: unified_code.as_ref(),
+            unified_message: unified_message.as_ref(),
+            processor_refund_data: processor_refund_data.as_ref(),
+            processor_transaction_data: processor_transaction_data.as_ref(),
         }
     }
 }
 
-#[cfg(feature = "v1")]
 impl super::KafkaMessage for KafkaRefund<'_> {
+    #[cfg(feature = "v1")]
     fn key(&self) -> String {
         format!(
             "{}_{}_{}_{}",
@@ -144,14 +207,7 @@ impl super::KafkaMessage for KafkaRefund<'_> {
             self.refund_id
         )
     }
-
-    fn event_type(&self) -> events::EventType {
-        events::EventType::Refund
-    }
-}
-
-#[cfg(feature = "v2")]
-impl super::KafkaMessage for KafkaRefund<'_> {
+    #[cfg(feature = "v2")]
     fn key(&self) -> String {
         format!(
             "{}_{}_{}_{}",
diff --git a/crates/router/src/services/kafka/refund_event.rs b/crates/router/src/services/kafka/refund_event.rs
index 34ed0b85c55..18cc0445cc9 100644
--- a/crates/router/src/services/kafka/refund_event.rs
+++ b/crates/router/src/services/kafka/refund_event.rs
@@ -1,3 +1,7 @@
+#[cfg(feature = "v2")]
+use common_utils::pii;
+#[cfg(feature = "v2")]
+use common_utils::types::{self, ChargeRefunds};
 use common_utils::{
     id_type,
     types::{ConnectorTransactionIdTrait, MinorUnit},
@@ -75,13 +79,13 @@ impl<'a> KafkaRefundEvent<'a> {
 #[serde_with::skip_serializing_none]
 #[derive(serde::Serialize, Debug)]
 pub struct KafkaRefundEvent<'a> {
-    pub id: &'a id_type::GlobalRefundId,
+    pub refund_id: &'a id_type::GlobalRefundId,
     pub merchant_reference_id: &'a id_type::RefundReferenceId,
     pub payment_id: &'a id_type::GlobalPaymentId,
     pub merchant_id: &'a id_type::MerchantId,
-    pub connector_transaction_id: &'a String,
+    pub connector_transaction_id: &'a types::ConnectorTransactionId,
     pub connector: &'a String,
-    pub connector_refund_id: Option<&'a String>,
+    pub connector_refund_id: Option<&'a types::ConnectorTransactionId>,
     pub external_reference_id: Option<&'a String>,
     pub refund_type: &'a storage_enums::RefundType,
     pub total_amount: &'a MinorUnit,
@@ -91,9 +95,9 @@ pub struct KafkaRefundEvent<'a> {
     pub sent_to_gateway: &'a bool,
     pub refund_error_message: Option<&'a String>,
     pub refund_arn: Option<&'a String>,
-    #[serde(default, with = "time::serde::timestamp::nanoseconds")]
+    #[serde(default, with = "time::serde::timestamp")]
     pub created_at: OffsetDateTime,
-    #[serde(default, with = "time::serde::timestamp::nanoseconds")]
+    #[serde(default, with = "time::serde::timestamp")]
     pub modified_at: OffsetDateTime,
     pub description: Option<&'a String>,
     pub attempt_id: &'a id_type::GlobalAttemptId,
@@ -101,36 +105,95 @@ pub struct KafkaRefundEvent<'a> {
     pub refund_error_code: Option<&'a String>,
     pub profile_id: Option<&'a id_type::ProfileId>,
     pub organization_id: &'a id_type::OrganizationId,
+
+    pub metadata: Option<&'a pii::SecretSerdeValue>,
+    pub updated_by: &'a String,
+    pub merchant_connector_id: Option<&'a id_type::MerchantConnectorAccountId>,
+    pub charges: Option<&'a ChargeRefunds>,
+    pub connector_refund_data: Option<&'a String>,
+    pub connector_transaction_data: Option<&'a String>,
+    pub split_refunds: Option<&'a common_types::refunds::SplitRefund>,
+    pub unified_code: Option<&'a String>,
+    pub unified_message: Option<&'a String>,
+    pub processor_refund_data: Option<&'a String>,
+    pub processor_transaction_data: Option<&'a String>,
 }
 
 #[cfg(feature = "v2")]
 impl<'a> KafkaRefundEvent<'a> {
     pub fn from_storage(refund: &'a Refund) -> Self {
+        let Refund {
+            payment_id,
+            merchant_id,
+            connector_transaction_id,
+            connector,
+            connector_refund_id,
+            external_reference_id,
+            refund_type,
+            total_amount,
+            currency,
+            refund_amount,
+            refund_status,
+            sent_to_gateway,
+            refund_error_message,
+            metadata,
+            refund_arn,
+            created_at,
+            modified_at,
+            description,
+            attempt_id,
+            refund_reason,
+            refund_error_code,
+            profile_id,
+            updated_by,
+            charges,
+            organization_id,
+            split_refunds,
+            unified_code,
+            unified_message,
+            processor_refund_data,
+            processor_transaction_data,
+            id,
+            merchant_reference_id,
+            connector_id,
+        } = refund;
+
         Self {
-            id: &refund.id,
-            merchant_reference_id: &refund.merchant_reference_id,
-            payment_id: &refund.payment_id,
-            merchant_id: &refund.merchant_id,
-            connector_transaction_id: refund.get_connector_transaction_id(),
-            connector: &refund.connector,
-            connector_refund_id: refund.get_optional_connector_refund_id(),
-            external_reference_id: refund.external_reference_id.as_ref(),
-            refund_type: &refund.refund_type,
-            total_amount: &refund.total_amount,
-            currency: &refund.currency,
-            refund_amount: &refund.refund_amount,
-            refund_status: &refund.refund_status,
-            sent_to_gateway: &refund.sent_to_gateway,
-            refund_error_message: refund.refund_error_message.as_ref(),
-            refund_arn: refund.refund_arn.as_ref(),
-            created_at: refund.created_at.assume_utc(),
-            modified_at: refund.modified_at.assume_utc(),
-            description: refund.description.as_ref(),
-            attempt_id: &refund.attempt_id,
-            refund_reason: refund.refund_reason.as_ref(),
-            refund_error_code: refund.refund_error_code.as_ref(),
-            profile_id: refund.profile_id.as_ref(),
-            organization_id: &refund.organization_id,
+            refund_id: id,
+            merchant_reference_id,
+            payment_id,
+            merchant_id,
+            connector_transaction_id,
+            connector,
+            connector_refund_id: connector_refund_id.as_ref(),
+            external_reference_id: external_reference_id.as_ref(),
+            refund_type,
+            total_amount,
+            currency,
+            refund_amount,
+            refund_status,
+            sent_to_gateway,
+            refund_error_message: refund_error_message.as_ref(),
+            refund_arn: refund_arn.as_ref(),
+            created_at: created_at.assume_utc(),
+            modified_at: modified_at.assume_utc(),
+            description: description.as_ref(),
+            attempt_id,
+            refund_reason: refund_reason.as_ref(),
+            refund_error_code: refund_error_code.as_ref(),
+            profile_id: profile_id.as_ref(),
+            organization_id,
+            metadata: metadata.as_ref(),
+            updated_by,
+            merchant_connector_id: connector_id.as_ref(),
+            charges: charges.as_ref(),
+            connector_refund_data: processor_refund_data.as_ref(),
+            connector_transaction_data: processor_transaction_data.as_ref(),
+            split_refunds: split_refunds.as_ref(),
+            unified_code: unified_code.as_ref(),
+            unified_message: unified_message.as_ref(),
+            processor_refund_data: processor_refund_data.as_ref(),
+            processor_transaction_data: processor_transaction_data.as_ref(),
         }
     }
 }

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

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

## Description
<!-- Describe your changes in detail -->
Added payment_intent payment_attempt and refund kafka events for v2

Other changes:
* populate connector_metadata in AuthorizaRouterData in v2.


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

Made a payment and refund and verified if proper events are logged into Kafka.

<img width="1728" alt="Screenshot 2025-06-12 at 9 48 33 AM" src="https://github.com/user-attachments/assets/91286bfd-8c83-4cb4-85cb-a672efb81daa" />
<img width="1728" alt="Screenshot 2025-06-12 at 9 48 18 AM" src="https://github.com/user-attachments/assets/2fe52584-a28d-44b7-acfb-936805775fab" />
<img width="1728" alt="Screenshot 2025-06-12 at 9 48 52 AM" src="https://github.com/user-attachments/assets/2c3b085a-7b87-41e1-80e1-28266a091fb6" />



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