rust/hyperswitch
1
0
Fork 0
mirror of https://github.com/juspay/hyperswitch.git synced 2025-10-29 00:49:42 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(kafka): add payment_intent payment_attempt and refund kafka events for v2 (#8328)

Browse Source 
Co-authored-by: Kashif <kashif.dev@protonmail.com>
This commit is contained in:
Hrithikesh
2025-06-19 12:06:07 +05:30
committed by GitHub
parent 5f7055fc8c
commit 305ca9bda9
12 changed files with 1138 additions and 332 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
crates/router/src/core/payments/transformers.rs
View File

@ -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![],
)))

306
crates/router/src/services/kafka/payment_attempt.rs
View File

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

307
crates/router/src/services/kafka/payment_attempt_event.rs
View File

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

291
crates/router/src/services/kafka/payment_intent.rs
View File

@ -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
}
}

259
crates/router/src/services/kafka/payment_intent_event.rs
View File

@ -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,
}
}
}

128
crates/router/src/services/kafka/refund.rs
View File

@ -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!(
"{}_{}_{}_{}",

121
crates/router/src/services/kafka/refund_event.rs
View File

@ -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(),
}
}
}