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

feat(webhooks): store request and response payloads in events table (#4029)

Browse Source
This commit is contained in:
Sanchith Hegde
2024-03-14 12:30:37 +05:30
committed by GitHub
parent ad17cc7383
commit fd67a6c225
32 changed files with 1267 additions and 644 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
crates/router/Cargo.toml
View File

@ -92,7 +92,7 @@ tokio = { version = "1.36.0", features = ["macros", "rt-multi-thread"] }
unicode-segmentation = "1.10.1"
url = { version = "2.4.0", features = ["serde"] }
utoipa = { version = "3.3.0", features = ["preserve_order", "time"] }
uuid = { version = "1.3.3", features = ["serde", "v4"] }
uuid = { version = "1.7.0", features = ["v4"] }
validator = "0.16.0"
x509-parser = "0.15.0"
tracing-futures = { version = "0.2.5", features = ["tokio"] }

18
crates/router/src/compatibility/stripe/webhooks.rs
View File

@ -11,7 +11,10 @@ use super::{
payment_intents::types::StripePaymentIntentResponse, refunds::types::StripeRefundResponse,
};
use crate::{
core::{errors, webhooks::types::OutgoingWebhookType},
core::{
errors,
webhooks::types::{OutgoingWebhookPayloadWithSignature, OutgoingWebhookType},
},
headers,
services::request::Maskable,
};
@ -30,8 +33,8 @@ pub struct StripeOutgoingWebhook {
impl OutgoingWebhookType for StripeOutgoingWebhook {
fn get_outgoing_webhooks_signature(
&self,
payment_response_hash_key: Option<String>,
) -> errors::CustomResult<Option<String>, errors::WebhooksFlowError> {
payment_response_hash_key: Option<impl AsRef<[u8]>>,
) -> errors::CustomResult<OutgoingWebhookPayloadWithSignature, errors::WebhooksFlowError> {
let timestamp = self.created;
let payment_response_hash_key = payment_response_hash_key
@ -48,7 +51,7 @@ impl OutgoingWebhookType for StripeOutgoingWebhook {
let v1 = hex::encode(
common_utils::crypto::HmacSha256::sign_message(
&common_utils::crypto::HmacSha256,
payment_response_hash_key.as_bytes(),
payment_response_hash_key.as_ref(),
new_signature_payload.as_bytes(),
)
.change_context(errors::WebhooksFlowError::OutgoingWebhookSigningFailed)
@ -56,7 +59,12 @@ impl OutgoingWebhookType for StripeOutgoingWebhook {
);
let t = timestamp;
Ok(Some(format!("t={t},v1={v1}")))
let signature = Some(format!("t={t},v1={v1}"));
Ok(OutgoingWebhookPayloadWithSignature {
payload: webhook_signature_payload.into(),
signature,
})
}
fn add_webhook_header(header: &mut Vec<(String, Maskable<String>)>, signature: String) {

5
crates/router/src/core/errors.rs
View File

@ -276,6 +276,8 @@ pub enum WebhooksFlowError {
OutgoingWebhookProcessTrackerTaskUpdateFailed,
#[error("Failed to schedule retry attempt for outgoing webhook")]
OutgoingWebhookRetrySchedulingFailed,
#[error("Outgoing webhook response encoding failed")]
OutgoingWebhookResponseEncodingFailed,
}
impl WebhooksFlowError {
@ -283,7 +285,8 @@ impl WebhooksFlowError {
match self {
Self::MerchantConfigNotFound
| Self::MerchantWebhookDetailsNotFound
| Self::MerchantWebhookUrlNotConfigured => false,
| Self::MerchantWebhookUrlNotConfigured
| Self::OutgoingWebhookResponseEncodingFailed => false,
Self::WebhookEventUpdationFailed
| Self::OutgoingWebhookSigningFailed

3
crates/router/src/core/payments.rs
View File

@ -422,7 +422,7 @@ where
.into_report()
.attach_printable("Frm configs label not found")?,
&customer,
key_store,
key_store.clone(),
))
.await?;
}
@ -482,6 +482,7 @@ where
crate::utils::trigger_payments_webhook(
merchant_account,
business_profile,
&key_store,
cloned_payment_data,
Some(cloned_request),
cloned_customer,

395
crates/router/src/core/webhooks.rs
View File

@ -8,9 +8,11 @@ use api_models::{
payments::HeaderPayload,
webhooks::{self, WebhookResponseTracker},
};
use common_utils::{errors::ReportSwitchExt, events::ApiEventsType, request::RequestContent};
use common_utils::{
errors::ReportSwitchExt, events::ApiEventsType, ext_traits::Encode, request::RequestContent,
};
use error_stack::{report, IntoReport, ResultExt};
use masking::ExposeInterface;
use masking::{ExposeInterface, Mask, PeekInterface, Secret};
use router_env::{
instrument,
tracing::{self, Instrument},
@ -39,7 +41,7 @@ use crate::{
services::{self, authentication as auth},
types::{
api::{self, mandates::MandateResponseExt},
domain,
domain::{self, types as domain_types},
storage::{self, enums},
transformers::{ForeignInto, ForeignTryInto},
},
@ -96,7 +98,7 @@ pub async fn payments_incoming_webhook_flow<Ctx: PaymentMethodRetrieve>(
>(
state.clone(),
merchant_account.clone(),
key_store,
key_store.clone(),
payments::operations::PaymentStatus,
api::PaymentsRetrieveRequest {
resource_id: id,
@ -168,16 +170,18 @@ pub async fn payments_incoming_webhook_flow<Ctx: PaymentMethodRetrieve>(
// If event is NOT an UnsupportedEvent, trigger Outgoing Webhook
if let Some(outgoing_event_type) = event_type {
let primary_object_created_at = payments_response.created;
create_event_and_trigger_outgoing_webhook(
state,
merchant_account,
business_profile,
&key_store,
outgoing_event_type,
enums::EventClass::Payments,
None,
payment_id.clone(),
enums::EventObjectType::PaymentDetails,
api::OutgoingWebhookContent::PaymentDetails(payments_response),
primary_object_created_at,
)
.await?;
};
@ -258,7 +262,7 @@ pub async fn refunds_incoming_webhook_flow(
Box::pin(refunds::refund_retrieve_core(
state.clone(),
merchant_account.clone(),
key_store,
key_store.clone(),
api_models::refunds::RefundsRetrieveRequest {
refund_id: refund_id.to_owned(),
force_sync: Some(true),
@ -278,12 +282,13 @@ pub async fn refunds_incoming_webhook_flow(
state,
merchant_account,
business_profile,
&key_store,
outgoing_event_type,
enums::EventClass::Refunds,
None,
refund_id,
enums::EventObjectType::RefundDetails,
api::OutgoingWebhookContent::RefundDetails(refund_response),
Some(updated_refund.created_at),
)
.await?;
}
@ -463,7 +468,7 @@ pub async fn mandates_incoming_webhook_flow(
let mandates_response = Box::new(
api::mandates::MandateResponse::from_db_mandate(
&state,
key_store,
key_store.clone(),
updated_mandate.clone(),
)
.await?,
@ -474,12 +479,13 @@ pub async fn mandates_incoming_webhook_flow(
state,
merchant_account,
business_profile,
&key_store,
outgoing_event_type,
enums::EventClass::Mandates,
None,
updated_mandate.mandate_id.clone(),
enums::EventObjectType::MandateDetails,
api::OutgoingWebhookContent::MandateDetails(mandates_response),
Some(updated_mandate.created_at),
)
.await?;
}
@ -499,6 +505,7 @@ pub async fn disputes_incoming_webhook_flow(
state: AppState,
merchant_account: domain::MerchantAccount,
business_profile: diesel_models::business_profile::BusinessProfile,
key_store: domain::MerchantKeyStore,
webhook_details: api::IncomingWebhookDetails,
source_verified: bool,
connector: &(dyn api::Connector + Sync),
@ -541,12 +548,13 @@ pub async fn disputes_incoming_webhook_flow(
state,
merchant_account,
business_profile,
&key_store,
event_type,
enums::EventClass::Disputes,
None,
dispute_object.dispute_id.clone(),
enums::EventObjectType::DisputeDetails,
api::OutgoingWebhookContent::DisputeDetails(disputes_response),
Some(dispute_object.created_at),
)
.await?;
metrics::INCOMING_DISPUTE_WEBHOOK_MERCHANT_NOTIFIED_METRIC.add(&metrics::CONTEXT, 1, &[]);
@ -594,7 +602,7 @@ async fn bank_transfer_webhook_flow<Ctx: PaymentMethodRetrieve>(
>(
state.clone(),
merchant_account.to_owned(),
key_store,
key_store.clone(),
payments::PaymentConfirm,
request,
services::api::AuthFlow::Merchant,
@ -623,16 +631,18 @@ async fn bank_transfer_webhook_flow<Ctx: PaymentMethodRetrieve>(
// If event is NOT an UnsupportedEvent, trigger Outgoing Webhook
if let Some(outgoing_event_type) = event_type {
let primary_object_created_at = payments_response.created;
create_event_and_trigger_outgoing_webhook(
state,
merchant_account,
business_profile,
&key_store,
outgoing_event_type,
enums::EventClass::Payments,
None,
payment_id.clone(),
enums::EventObjectType::PaymentDetails,
api::OutgoingWebhookContent::PaymentDetails(payments_response),
primary_object_created_at,
)
.await?;
}
@ -652,45 +662,91 @@ pub(crate) async fn create_event_and_trigger_outgoing_webhook(
state: AppState,
merchant_account: domain::MerchantAccount,
business_profile: diesel_models::business_profile::BusinessProfile,
merchant_key_store: &domain::MerchantKeyStore,
event_type: enums::EventType,
event_class: enums::EventClass,
intent_reference_id: Option<String>,
primary_object_id: String,
primary_object_type: enums::EventObjectType,
content: api::OutgoingWebhookContent,
primary_object_created_at: Option<time::PrimitiveDateTime>,
) -> CustomResult<(), errors::ApiErrorResponse> {
let event_id = format!("{primary_object_id}_{event_type}");
let delivery_attempt = types::WebhookDeliveryAttempt::InitialAttempt;
let idempotent_event_id =
utils::get_idempotent_event_id(&primary_object_id, event_type, delivery_attempt);
let webhook_url_result = get_webhook_url_from_business_profile(&business_profile);
if !state.conf.webhooks.outgoing_enabled
|| get_webhook_url_from_business_profile(&business_profile).is_err()
|| webhook_url_result.is_err()
|| webhook_url_result.as_ref().is_ok_and(String::is_empty)
{
logger::debug!(
business_profile_id=%business_profile.profile_id,
%event_id,
%idempotent_event_id,
"Outgoing webhooks are disabled in application configuration, or merchant webhook URL \
could not be obtained; skipping outgoing webhooks for event"
);
return Ok(());
}
let event_id = utils::generate_event_id();
let merchant_id = business_profile.merchant_id.clone();
let new_event = storage::EventNew {
let now = common_utils::date_time::now();
let outgoing_webhook = api::OutgoingWebhook {
merchant_id: merchant_id.clone(),
event_id: event_id.clone(),
event_type,
content: content.clone(),
timestamp: now,
};
let request_content = get_outgoing_webhook_request(
&merchant_account,
outgoing_webhook,
business_profile.payment_response_hash_key.as_deref(),
)
.change_context(errors::ApiErrorResponse::WebhookProcessingFailure)
.attach_printable("Failed to construct outgoing webhook request content")?;
let new_event = domain::Event {
event_id: event_id.clone(),
event_type,
event_class,
is_webhook_notified: false,
intent_reference_id,
primary_object_id,
primary_object_type,
created_at: now,
merchant_id: Some(business_profile.merchant_id.clone()),
business_profile_id: Some(business_profile.profile_id.clone()),
primary_object_created_at,
idempotent_event_id: Some(idempotent_event_id.clone()),
initial_attempt_id: Some(event_id.clone()),
request: Some(
domain_types::encrypt(
request_content
.encode_to_string_of_json()
.change_context(errors::ApiErrorResponse::WebhookProcessingFailure)
.attach_printable("Failed to encode outgoing webhook request content")
.map(Secret::new)?,
merchant_key_store.key.get_inner().peek(),
)
.await
.change_context(errors::ApiErrorResponse::WebhookProcessingFailure)
.attach_printable("Failed to encrypt outgoing webhook request content")?,
),
response: None,
};
let event_insert_result = state.store.insert_event(new_event).await;
let event_insert_result = state
.store
.insert_event(new_event, merchant_key_store)
.await;
let event = match event_insert_result {
Ok(event) => Ok(event),
Err(error) => {
if error.current_context().is_db_unique_violation() {
logger::debug!("Event `{event_id}` already exists in the database");
logger::debug!("Event with idempotent ID `{idempotent_event_id}` already exists in the database");
return Ok(());
} else {
logger::error!(event_insertion_failure=?error);
@ -701,14 +757,6 @@ pub(crate) async fn create_event_and_trigger_outgoing_webhook(
}
}?;
let outgoing_webhook = api::OutgoingWebhook {
merchant_id: merchant_id.clone(),
event_id: event.event_id.clone(),
event_type: event.event_type,
content: content.clone(),
timestamp: event.created_at,
};
let process_tracker = add_outgoing_webhook_retry_task_to_process_tracker(
&*state.store,
&business_profile,
@ -724,19 +772,19 @@ pub(crate) async fn create_event_and_trigger_outgoing_webhook(
})
.ok();
let cloned_key_store = merchant_key_store.clone();
// Using a tokio spawn here and not arbiter because not all caller of this function
// may have an actix arbiter
tokio::spawn(
async move {
trigger_appropriate_webhook_and_raise_event(
trigger_webhook_and_raise_event(
state,
merchant_account,
business_profile,
outgoing_webhook,
types::WebhookDeliveryAttempt::InitialAttempt,
content,
event.event_id,
event_type,
&cloned_key_store,
event,
request_content,
delivery_attempt,
Some(content),
process_tracker,
)
.await;
@ -748,83 +796,45 @@ pub(crate) async fn create_event_and_trigger_outgoing_webhook(
}
#[allow(clippy::too_many_arguments)]
pub(crate) async fn trigger_appropriate_webhook_and_raise_event(
#[instrument(skip_all)]
pub(crate) async fn trigger_webhook_and_raise_event(
state: AppState,
merchant_account: domain::MerchantAccount,
business_profile: diesel_models::business_profile::BusinessProfile,
outgoing_webhook: api::OutgoingWebhook,
merchant_key_store: &domain::MerchantKeyStore,
event: domain::Event,
request_content: types::OutgoingWebhookRequestContent,
delivery_attempt: types::WebhookDeliveryAttempt,
content: api::OutgoingWebhookContent,
event_id: String,
event_type: enums::EventType,
content: Option<api::OutgoingWebhookContent>,
process_tracker: Option<storage::ProcessTracker>,
) {
match merchant_account.get_compatible_connector() {
#[cfg(feature = "stripe")]
Some(api_models::enums::Connector::Stripe) => {
trigger_webhook_and_raise_event::<stripe_webhooks::StripeOutgoingWebhook>(
state,
business_profile,
outgoing_webhook,
delivery_attempt,
content,
event_id,
event_type,
process_tracker,
)
.await
}
_ => {
trigger_webhook_and_raise_event::<api_models::webhooks::OutgoingWebhook>(
state,
business_profile,
outgoing_webhook,
delivery_attempt,
content,
event_id,
event_type,
process_tracker,
)
.await
}
}
}
logger::debug!(
event_id=%event.event_id,
idempotent_event_id=?event.idempotent_event_id,
initial_attempt_id=?event.initial_attempt_id,
"Attempting to send webhook"
);
#[allow(clippy::too_many_arguments)]
async fn trigger_webhook_and_raise_event<W: types::OutgoingWebhookType>(
state: AppState,
business_profile: diesel_models::business_profile::BusinessProfile,
outgoing_webhook: api::OutgoingWebhook,
delivery_attempt: types::WebhookDeliveryAttempt,
content: api::OutgoingWebhookContent,
event_id: String,
event_type: enums::EventType,
process_tracker: Option<storage::ProcessTracker>,
) {
let merchant_id = business_profile.merchant_id.clone();
let trigger_webhook_result = trigger_webhook_to_merchant::<W>(
let trigger_webhook_result = trigger_webhook_to_merchant(
state.clone(),
business_profile,
outgoing_webhook,
merchant_key_store,
event.clone(),
request_content,
delivery_attempt,
process_tracker,
)
.await;
raise_webhooks_analytics_event(
state,
trigger_webhook_result,
content,
&merchant_id,
&event_id,
event_type,
);
raise_webhooks_analytics_event(state, trigger_webhook_result, content, merchant_id, event);
}
async fn trigger_webhook_to_merchant<W: types::OutgoingWebhookType>(
async fn trigger_webhook_to_merchant(
state: AppState,
business_profile: diesel_models::business_profile::BusinessProfile,
webhook: api::OutgoingWebhook,
merchant_key_store: &domain::MerchantKeyStore,
event: domain::Event,
request_content: types::OutgoingWebhookRequestContent,
delivery_attempt: types::WebhookDeliveryAttempt,
process_tracker: Option<storage::ProcessTracker>,
) -> CustomResult<(), errors::WebhooksFlowError> {
@ -853,28 +863,21 @@ async fn trigger_webhook_to_merchant<W: types::OutgoingWebhookType>(
(Err(error), None) => Err(error),
}?;
let outgoing_webhook_event_id = webhook.event_id.clone();
let transformed_outgoing_webhook = W::from(webhook);
let outgoing_webhooks_signature = transformed_outgoing_webhook
.get_outgoing_webhooks_signature(business_profile.payment_response_hash_key.clone())?;
let mut header = vec![(
reqwest::header::CONTENT_TYPE.to_string(),
mime::APPLICATION_JSON.essence_str().into(),
)];
if let Some(signature) = outgoing_webhooks_signature {
W::add_webhook_header(&mut header, signature)
}
let event_id = event.event_id;
let headers = request_content
.headers
.into_iter()
.map(|(name, value)| (name, value.into_masked()))
.collect();
let request = services::RequestBuilder::new()
.method(services::Method::Post)
.url(&webhook_url)
.attach_default_headers()
.headers(header)
.set_body(RequestContent::Json(Box::new(transformed_outgoing_webhook)))
.headers(headers)
.set_body(RequestContent::RawBytes(
request_content.payload.expose().into_bytes(),
))
.build();
let response = state
@ -903,10 +906,72 @@ async fn trigger_webhook_to_merchant<W: types::OutgoingWebhookType>(
"An error occurred when sending webhook to merchant"
);
};
let update_event_in_storage = |state: AppState,
merchant_key_store: domain::MerchantKeyStore,
event_id: String,
response: reqwest::Response| async move {
let status_code = response.status();
let is_webhook_notified = status_code.is_success();
let response_headers = response
.headers()
.iter()
.map(|(name, value)| {
(
name.as_str().to_owned(),
value
.to_str()
.map(|s| Secret::from(String::from(s)))
.unwrap_or_else(|error| {
logger::warn!(
"Response header {} contains non-UTF-8 characters: {error:?}",
name.as_str()
);
Secret::from(String::from("Non-UTF-8 header value"))
}),
)
})
.collect::<Vec<_>>();
let response_payload = response
.text()
.await
.map(Secret::from)
.unwrap_or_else(|error| {
logger::warn!("Response contains non-UTF-8 characters: {error:?}");
Secret::from(String::from("Non-UTF-8 response body"))
});
let response_to_store = types::OutgoingWebhookResponseContent {
payload: response_payload,
headers: response_headers,
status_code: status_code.as_u16(),
};
let event_update = domain::EventUpdate::UpdateResponse {
is_webhook_notified,
response: Some(
domain_types::encrypt(
response_to_store
.encode_to_string_of_json()
.change_context(
errors::WebhooksFlowError::OutgoingWebhookResponseEncodingFailed,
)
.map(Secret::new)?,
merchant_key_store.key.get_inner().peek(),
)
.await
.change_context(errors::WebhooksFlowError::WebhookEventUpdationFailed)
.attach_printable("Failed to encrypt outgoing webhook request content")?,
),
};
state
.store
.update_event(event_id, event_update, &merchant_key_store)
.await
.change_context(errors::WebhooksFlowError::WebhookEventUpdationFailed)
};
let success_response_handler =
|state: AppState,
merchant_id: String,
outgoing_webhook_event_id: String,
process_tracker: Option<storage::ProcessTracker>,
business_status: &'static str| async move {
metrics::WEBHOOK_OUTGOING_RECEIVED_COUNT.add(
@ -915,15 +980,6 @@ async fn trigger_webhook_to_merchant<W: types::OutgoingWebhookType>(
&[metrics::KeyValue::new(MERCHANT_ID, merchant_id)],
);
let update_event = storage::EventUpdate::UpdateWebhookNotified {
is_webhook_notified: Some(true),
};
state
.store
.update_event(outgoing_webhook_event_id, update_event)
.await
.change_context(errors::WebhooksFlowError::WebhookEventUpdationFailed)?;
match process_tracker {
Some(process_tracker) => state
.store
@ -954,11 +1010,19 @@ async fn trigger_webhook_to_merchant<W: types::OutgoingWebhookType>(
types::WebhookDeliveryAttempt::InitialAttempt => match response {
Err(client_error) => api_client_error_handler(client_error, delivery_attempt),
Ok(response) => {
if response.status().is_success() {
let status_code = response.status();
let _updated_event = update_event_in_storage(
state.clone(),
merchant_key_store.clone(),
event_id.clone(),
response,
)
.await?;
if status_code.is_success() {
success_response_handler(
state.clone(),
business_profile.merchant_id,
outgoing_webhook_event_id,
process_tracker,
"INITIAL_DELIVERY_ATTEMPT_SUCCESSFUL",
)
@ -967,7 +1031,7 @@ async fn trigger_webhook_to_merchant<W: types::OutgoingWebhookType>(
error_response_handler(
business_profile.merchant_id,
delivery_attempt,
response.status().as_u16(),
status_code.as_u16(),
"Ignoring error when sending webhook to merchant",
);
}
@ -993,11 +1057,19 @@ async fn trigger_webhook_to_merchant<W: types::OutgoingWebhookType>(
)?;
}
Ok(response) => {
if response.status().is_success() {
let status_code = response.status();
let _updated_event = update_event_in_storage(
state.clone(),
merchant_key_store.clone(),
event_id.clone(),
response,
)
.await?;
if status_code.is_success() {
success_response_handler(
state.clone(),
business_profile.merchant_id,
outgoing_webhook_event_id,
Some(process_tracker),
"COMPLETED_BY_PT",
)
@ -1006,7 +1078,7 @@ async fn trigger_webhook_to_merchant<W: types::OutgoingWebhookType>(
error_response_handler(
business_profile.merchant_id.clone(),
delivery_attempt,
response.status().as_u16(),
status_code.as_u16(),
"An error occurred when sending webhook to merchant",
);
// Schedule a retry attempt for webhook delivery
@ -1031,10 +1103,9 @@ async fn trigger_webhook_to_merchant<W: types::OutgoingWebhookType>(
fn raise_webhooks_analytics_event(
state: AppState,
trigger_webhook_result: CustomResult<(), errors::WebhooksFlowError>,
content: api::OutgoingWebhookContent,
merchant_id: &str,
event_id: &str,
event_type: enums::EventType,
content: Option<api::OutgoingWebhookContent>,
merchant_id: String,
event: domain::Event,
) {
let error = if let Err(error) = trigger_webhook_result {
logger::error!(?error, "Failed to send webhook to merchant");
@ -1051,13 +1122,16 @@ fn raise_webhooks_analytics_event(
None
};
let outgoing_webhook_event_content = content.get_outgoing_webhook_event_content();
let outgoing_webhook_event_content = content
.as_ref()
.and_then(api::OutgoingWebhookContent::get_outgoing_webhook_event_content);
let webhook_event = OutgoingWebhookEvent::new(
merchant_id.to_owned(),
event_id.to_owned(),
event_type,
merchant_id,
event.event_id,
event.event_type,
outgoing_webhook_event_content,
error,
event.initial_attempt_id,
);
match RawEvent::try_from(webhook_event.clone()) {
@ -1410,6 +1484,7 @@ pub async fn webhooks_core<W: types::OutgoingWebhookType, Ctx: PaymentMethodRetr
state.clone(),
merchant_account,
business_profile,
key_store,
webhook_details,
source_verified,
*connector,
@ -1570,7 +1645,7 @@ async fn fetch_optional_mca_and_connector(
pub async fn add_outgoing_webhook_retry_task_to_process_tracker(
db: &dyn StorageInterface,
business_profile: &diesel_models::business_profile::BusinessProfile,
event: &storage::Event,
event: &domain::Event,
) -> CustomResult<storage::ProcessTracker, errors::StorageError> {
let schedule_time = outgoing_webhook_retry::get_webhook_delivery_retry_schedule_time(
db,
@ -1591,6 +1666,7 @@ pub async fn add_outgoing_webhook_retry_task_to_process_tracker(
event_class: event.event_class,
primary_object_id: event.primary_object_id.clone(),
primary_object_type: event.primary_object_type,
initial_attempt_id: event.initial_attempt_id.clone(),
};
let runner = storage::ProcessTrackerRunner::OutgoingWebhookRetryWorkflow;
@ -1652,3 +1728,50 @@ fn get_webhook_url_from_business_profile(
.change_context(errors::WebhooksFlowError::MerchantWebhookUrlNotConfigured)
.map(ExposeInterface::expose)
}
pub(crate) fn get_outgoing_webhook_request(
merchant_account: &domain::MerchantAccount,
outgoing_webhook: api::OutgoingWebhook,
payment_response_hash_key: Option<&str>,
) -> CustomResult<types::OutgoingWebhookRequestContent, errors::WebhooksFlowError> {
#[inline]
fn get_outgoing_webhook_request_inner<WebhookType: types::OutgoingWebhookType>(
outgoing_webhook: api::OutgoingWebhook,
payment_response_hash_key: Option<&str>,
) -> CustomResult<types::OutgoingWebhookRequestContent, errors::WebhooksFlowError> {
let mut headers = vec![(
reqwest::header::CONTENT_TYPE.to_string(),
mime::APPLICATION_JSON.essence_str().into(),
)];
let transformed_outgoing_webhook = WebhookType::from(outgoing_webhook);
let outgoing_webhooks_signature = transformed_outgoing_webhook
.get_outgoing_webhooks_signature(payment_response_hash_key)?;
if let Some(signature) = outgoing_webhooks_signature.signature {
WebhookType::add_webhook_header(&mut headers, signature)
}
Ok(types::OutgoingWebhookRequestContent {
payload: outgoing_webhooks_signature.payload,
headers: headers
.into_iter()
.map(|(name, value)| (name, Secret::new(value.into_inner())))
.collect(),
})
}
match merchant_account.get_compatible_connector() {
#[cfg(feature = "stripe")]
Some(api_models::enums::Connector::Stripe) => get_outgoing_webhook_request_inner::<
stripe_webhooks::StripeOutgoingWebhook,
>(
outgoing_webhook, payment_response_hash_key
),
_ => get_outgoing_webhook_request_inner::<api_models::webhooks::OutgoingWebhook>(
outgoing_webhook,
payment_response_hash_key,
),
}
}

40
crates/router/src/core/webhooks/types.rs
View File

@ -1,17 +1,23 @@
use api_models::webhooks;
use common_utils::{crypto::SignMessage, ext_traits::Encode};
use error_stack::ResultExt;
use masking::Secret;
use serde::Serialize;
use crate::{core::errors, headers, services::request::Maskable, types::storage::enums};
pub struct OutgoingWebhookPayloadWithSignature {
pub payload: Secret<String>,
pub signature: Option<String>,
}
pub trait OutgoingWebhookType:
Serialize + From<webhooks::OutgoingWebhook> + Sync + Send + std::fmt::Debug + 'static
{
fn get_outgoing_webhooks_signature(
&self,
payment_response_hash_key: Option<String>,
) -> errors::CustomResult<Option<String>, errors::WebhooksFlowError>;
payment_response_hash_key: Option<impl AsRef<[u8]>>,
) -> errors::CustomResult<OutgoingWebhookPayloadWithSignature, errors::WebhooksFlowError>;
fn add_webhook_header(header: &mut Vec<(String, Maskable<String>)>, signature: String);
}
@ -19,26 +25,32 @@ pub trait OutgoingWebhookType:
impl OutgoingWebhookType for webhooks::OutgoingWebhook {
fn get_outgoing_webhooks_signature(
&self,
payment_response_hash_key: Option<String>,
) -> errors::CustomResult<Option<String>, errors::WebhooksFlowError> {
payment_response_hash_key: Option<impl AsRef<[u8]>>,
) -> errors::CustomResult<OutgoingWebhookPayloadWithSignature, errors::WebhooksFlowError> {
let webhook_signature_payload = self
.encode_to_string_of_json()
.change_context(errors::WebhooksFlowError::OutgoingWebhookEncodingFailed)
.attach_printable("failed encoding outgoing webhook payload")?;
Ok(payment_response_hash_key
let signature = payment_response_hash_key
.map(|key| {
common_utils::crypto::HmacSha512::sign_message(
&common_utils::crypto::HmacSha512,
key.as_bytes(),
key.as_ref(),
webhook_signature_payload.as_bytes(),
)
})
.transpose()
.change_context(errors::WebhooksFlowError::OutgoingWebhookSigningFailed)
.attach_printable("Failed to sign the message")?
.map(hex::encode))
.map(hex::encode);
Ok(OutgoingWebhookPayloadWithSignature {
payload: webhook_signature_payload.into(),
signature,
})
}
fn add_webhook_header(header: &mut Vec<(String, Maskable<String>)>, signature: String) {
header.push((headers::X_WEBHOOK_SIGNATURE.to_string(), signature.into()))
}
@ -58,4 +70,18 @@ pub(crate) struct OutgoingWebhookTrackingData {
pub(crate) event_class: enums::EventClass,
pub(crate) primary_object_id: String,
pub(crate) primary_object_type: enums::EventObjectType,
pub(crate) initial_attempt_id: Option<String>,
}
#[derive(Debug, serde::Serialize, serde::Deserialize)]
pub(crate) struct OutgoingWebhookRequestContent {
pub(crate) payload: Secret<String>,
pub(crate) headers: Vec<(String, Secret<String>)>,
}
#[derive(Debug, serde::Serialize, serde::Deserialize)]
pub(crate) struct OutgoingWebhookResponseContent {
pub(crate) payload: Secret<String>,
pub(crate) headers: Vec<(String, Secret<String>)>,
pub(crate) status_code: u16,
}

24
crates/router/src/core/webhooks/utils.rs
View File

@ -120,3 +120,27 @@ pub async fn construct_webhook_router_data<'a>(
};
Ok(router_data)
}
#[inline]
pub(crate) fn get_idempotent_event_id(
primary_object_id: &str,
event_type: crate::types::storage::enums::EventType,
delivery_attempt: super::types::WebhookDeliveryAttempt,
) -> String {
use super::types::WebhookDeliveryAttempt;
const EVENT_ID_SUFFIX_LENGTH: usize = 8;
let common_prefix = format!("{primary_object_id}_{event_type}");
match delivery_attempt {
WebhookDeliveryAttempt::InitialAttempt => common_prefix,
WebhookDeliveryAttempt::AutomaticRetry => {
common_utils::generate_id(EVENT_ID_SUFFIX_LENGTH, &common_prefix)
}
}
}
#[inline]
pub(crate) fn generate_event_id() -> String {
common_utils::generate_time_ordered_id("evt")
}

200
crates/router/src/db/events.rs
View File

@ -1,3 +1,4 @@
use common_utils::ext_traits::AsyncExt;
use error_stack::{IntoReport, ResultExt};
use router_env::{instrument, tracing};
@ -5,26 +6,39 @@ use super::{MockDb, Store};
use crate::{
connection,
core::errors::{self, CustomResult},
types::storage,
types::{
domain::{
self,
behaviour::{Conversion, ReverseConversion},
},
storage,
},
};
#[async_trait::async_trait]
pub trait EventInterface {
pub trait EventInterface
where
domain::Event:
Conversion<DstType = storage::events::Event, NewDstType = storage::events::EventNew>,
{
async fn insert_event(
&self,
event: storage::EventNew,
) -> CustomResult<storage::Event, errors::StorageError>;
event: domain::Event,
merchant_key_store: &domain::MerchantKeyStore,
) -> CustomResult<domain::Event, errors::StorageError>;
async fn find_event_by_event_id(
&self,
event_id: &str,
) -> CustomResult<storage::Event, errors::StorageError>;
merchant_key_store: &domain::MerchantKeyStore,
) -> CustomResult<domain::Event, errors::StorageError>;
async fn update_event(
&self,
event_id: String,
event: storage::EventUpdate,
) -> CustomResult<storage::Event, errors::StorageError>;
event: domain::EventUpdate,
merchant_key_store: &domain::MerchantKeyStore,
) -> CustomResult<domain::Event, errors::StorageError>;
}
#[async_trait::async_trait]
@ -32,35 +46,54 @@ impl EventInterface for Store {
#[instrument(skip_all)]
async fn insert_event(
&self,
event: storage::EventNew,
) -> CustomResult<storage::Event, errors::StorageError> {
event: domain::Event,
merchant_key_store: &domain::MerchantKeyStore,
) -> CustomResult<domain::Event, errors::StorageError> {
let conn = connection::pg_connection_write(self).await?;
event.insert(&conn).await.map_err(Into::into).into_report()
event
.construct_new()
.await
.change_context(errors::StorageError::EncryptionError)?
.insert(&conn)
.await
.map_err(Into::into)
.into_report()?
.convert(merchant_key_store.key.get_inner())
.await
.change_context(errors::StorageError::DecryptionError)
}
#[instrument(skip_all)]
async fn find_event_by_event_id(
&self,
event_id: &str,
) -> CustomResult<storage::Event, errors::StorageError> {
merchant_key_store: &domain::MerchantKeyStore,
) -> CustomResult<domain::Event, errors::StorageError> {
let conn = connection::pg_connection_read(self).await?;
storage::Event::find_by_event_id(&conn, event_id)
.await
.map_err(Into::into)
.into_report()
.into_report()?
.convert(merchant_key_store.key.get_inner())
.await
.change_context(errors::StorageError::DecryptionError)
}
#[instrument(skip_all)]
async fn update_event(
&self,
event_id: String,
event: storage::EventUpdate,
) -> CustomResult<storage::Event, errors::StorageError> {
event: domain::EventUpdate,
merchant_key_store: &domain::MerchantKeyStore,
) -> CustomResult<domain::Event, errors::StorageError> {
let conn = connection::pg_connection_write(self).await?;
storage::Event::update(&conn, &event_id, event)
storage::Event::update(&conn, &event_id, event.into())
.await
.map_err(Into::into)
.into_report()
.into_report()?
.convert(merchant_key_store.key.get_inner())
.await
.change_context(errors::StorageError::DecryptionError)
}
}
@ -68,41 +101,41 @@ impl EventInterface for Store {
impl EventInterface for MockDb {
async fn insert_event(
&self,
event: storage::EventNew,
) -> CustomResult<storage::Event, errors::StorageError> {
event: domain::Event,
merchant_key_store: &domain::MerchantKeyStore,
) -> CustomResult<domain::Event, errors::StorageError> {
let mut locked_events = self.events.lock().await;
let now = common_utils::date_time::now();
let stored_event = storage::Event {
id: locked_events
.len()
.try_into()
.into_report()
.change_context(errors::StorageError::MockDbError)?,
event_id: event.event_id,
event_type: event.event_type,
event_class: event.event_class,
is_webhook_notified: event.is_webhook_notified,
intent_reference_id: event.intent_reference_id,
primary_object_id: event.primary_object_id,
primary_object_type: event.primary_object_type,
created_at: now,
};
let stored_event = Conversion::convert(event)
.await
.change_context(errors::StorageError::EncryptionError)?;
locked_events.push(stored_event.clone());
Ok(stored_event)
stored_event
.convert(merchant_key_store.key.get_inner())
.await
.change_context(errors::StorageError::DecryptionError)
}
async fn find_event_by_event_id(
&self,
event_id: &str,
) -> CustomResult<storage::Event, errors::StorageError> {
merchant_key_store: &domain::MerchantKeyStore,
) -> CustomResult<domain::Event, errors::StorageError> {
let locked_events = self.events.lock().await;
locked_events
.iter()
.find(|event| event.event_id == event_id)
.cloned()
.async_map(|event| async {
event
.convert(merchant_key_store.key.get_inner())
.await
.change_context(errors::StorageError::DecryptionError)
})
.await
.transpose()?
.ok_or(
errors::StorageError::ValueNotFound(format!(
"No event available with event_id = {event_id}"
@ -114,8 +147,9 @@ impl EventInterface for MockDb {
async fn update_event(
&self,
event_id: String,
event: storage::EventUpdate,
) -> CustomResult<storage::Event, errors::StorageError> {
event: domain::EventUpdate,
merchant_key_store: &domain::MerchantKeyStore,
) -> CustomResult<domain::Event, errors::StorageError> {
let mut locked_events = self.events.lock().await;
let event_to_update = locked_events
.iter_mut()
@ -123,26 +157,35 @@ impl EventInterface for MockDb {
.ok_or(errors::StorageError::MockDbError)?;
match event {
storage::EventUpdate::UpdateWebhookNotified {
domain::EventUpdate::UpdateResponse {
is_webhook_notified,
response,
} => {
if let Some(is_webhook_notified) = is_webhook_notified {
event_to_update.is_webhook_notified = is_webhook_notified;
}
event_to_update.is_webhook_notified = is_webhook_notified;
event_to_update.response = response.map(Into::into);
}
}
Ok(event_to_update.clone())
event_to_update
.clone()
.convert(merchant_key_store.key.get_inner())
.await
.change_context(errors::StorageError::DecryptionError)
}
}
#[cfg(test)]
mod tests {
use diesel_models::enums;
use time::macros::datetime;
use crate::{
db::{events::EventInterface, MockDb},
types::storage,
db::{
events::EventInterface, merchant_key_store::MerchantKeyStoreInterface,
MasterKeyInterface, MockDb,
},
services,
types::domain,
};
#[allow(clippy::unwrap_used)]
@ -152,34 +195,71 @@ mod tests {
let mockdb = MockDb::new(&redis_interface::RedisSettings::default())
.await
.expect("Failed to create Mock store");
let event_id = "test_event_id";
let merchant_id = "merchant1";
let business_profile_id = "profile1";
let event1 = mockdb
.insert_event(storage::EventNew {
event_id: "test_event_id".into(),
event_type: enums::EventType::PaymentSucceeded,
event_class: enums::EventClass::Payments,
is_webhook_notified: false,
intent_reference_id: Some("test".into()),
primary_object_id: "primary_object_tet".into(),
primary_object_type: enums::EventObjectType::PaymentDetails,
})
let master_key = mockdb.get_master_key();
mockdb
.insert_merchant_key_store(
domain::MerchantKeyStore {
merchant_id: merchant_id.into(),
key: domain::types::encrypt(
services::generate_aes256_key().unwrap().to_vec().into(),
master_key,
)
.await
.unwrap(),
created_at: datetime!(2023-02-01 0:00),
},
&master_key.to_vec().into(),
)
.await
.unwrap();
let merchant_key_store = mockdb
.get_merchant_key_store_by_merchant_id(merchant_id, &master_key.to_vec().into())
.await
.unwrap();
assert_eq!(event1.id, 0);
let event1 = mockdb
.insert_event(
domain::Event {
event_id: event_id.into(),
event_type: enums::EventType::PaymentSucceeded,
event_class: enums::EventClass::Payments,
is_webhook_notified: false,
primary_object_id: "primary_object_tet".into(),
primary_object_type: enums::EventObjectType::PaymentDetails,
created_at: common_utils::date_time::now(),
merchant_id: Some(merchant_id.to_owned()),
business_profile_id: Some(business_profile_id.to_owned()),
primary_object_created_at: Some(common_utils::date_time::now()),
idempotent_event_id: Some(event_id.into()),
initial_attempt_id: Some(event_id.into()),
request: None,
response: None,
},
&merchant_key_store,
)
.await
.unwrap();
assert_eq!(event1.event_id, event_id);
let updated_event = mockdb
.update_event(
"test_event_id".into(),
storage::EventUpdate::UpdateWebhookNotified {
is_webhook_notified: Some(true),
event_id.into(),
domain::EventUpdate::UpdateResponse {
is_webhook_notified: true,
response: None,
},
&merchant_key_store,
)
.await
.unwrap();
assert!(updated_event.is_webhook_notified);
assert_eq!(updated_event.primary_object_id, "primary_object_tet");
assert_eq!(updated_event.id, 0);
assert_eq!(updated_event.event_id, event_id);
}
}

25
crates/router/src/db/kafka_store.rs
View File

@ -483,24 +483,33 @@ impl EphemeralKeyInterface for KafkaStore {
impl EventInterface for KafkaStore {
async fn insert_event(
&self,
event: storage::EventNew,
) -> CustomResult<storage::Event, errors::StorageError> {
self.diesel_store.insert_event(event).await
event: domain::Event,
merchant_key_store: &domain::MerchantKeyStore,
) -> CustomResult<domain::Event, errors::StorageError> {
self.diesel_store
.insert_event(event, merchant_key_store)
.await
}
async fn find_event_by_event_id(
&self,
event_id: &str,
) -> CustomResult<storage::Event, errors::StorageError> {
self.diesel_store.find_event_by_event_id(event_id).await
merchant_key_store: &domain::MerchantKeyStore,
) -> CustomResult<domain::Event, errors::StorageError> {
self.diesel_store
.find_event_by_event_id(event_id, merchant_key_store)
.await
}
async fn update_event(
&self,
event_id: String,
event: storage::EventUpdate,
) -> CustomResult<storage::Event, errors::StorageError> {
self.diesel_store.update_event(event_id, event).await
event: domain::EventUpdate,
merchant_key_store: &domain::MerchantKeyStore,
) -> CustomResult<domain::Event, errors::StorageError> {
self.diesel_store
.update_event(event_id, event, merchant_key_store)
.await
}
}

3
crates/router/src/events/outgoing_webhook_logs.rs
View File

@ -16,6 +16,7 @@ pub struct OutgoingWebhookEvent {
is_error: bool,
error: Option<Value>,
created_at_timestamp: i128,
initial_attempt_id: Option<String>,
}
#[derive(Clone, Debug, PartialEq, Serialize)]
@ -83,6 +84,7 @@ impl OutgoingWebhookEvent {
event_type: OutgoingWebhookEventType,
content: Option<OutgoingWebhookEventContent>,
error: Option<Value>,
initial_attempt_id: Option<String>,
) -> Self {
Self {
merchant_id,
@ -92,6 +94,7 @@ impl OutgoingWebhookEvent {
is_error: error.is_some(),
error,
created_at_timestamp: OffsetDateTime::now_utc().unix_timestamp_nanos() / 1_000_000,
initial_attempt_id,
}
}
}

4
crates/router/src/services/api.rs
View File

@ -374,6 +374,7 @@ where
.masked_serialize()
.unwrap_or(json!({ "error": "failed to mask serialize"})),
RequestContent::FormData(_) => json!({"request_type": "FORM_DATA"}),
RequestContent::RawBytes(_) => json!({"request_type": "RAW_BYTES"}),
},
None => serde_json::Value::Null,
};
@ -643,6 +644,7 @@ pub async fn send_request(
.change_context(errors::ApiClientError::BodySerializationFailed)?;
client.body(body).header("Content-Type", "application/xml")
}
Some(RequestContent::RawBytes(payload)) => client.body(payload),
None => client,
}
}
@ -658,6 +660,7 @@ pub async fn send_request(
.change_context(errors::ApiClientError::BodySerializationFailed)?;
client.body(body).header("Content-Type", "application/xml")
}
Some(RequestContent::RawBytes(payload)) => client.body(payload),
None => client,
}
}
@ -673,6 +676,7 @@ pub async fn send_request(
.change_context(errors::ApiClientError::BodySerializationFailed)?;
client.body(body).header("Content-Type", "application/xml")
}
Some(RequestContent::RawBytes(payload)) => client.body(payload),
None => client,
}
}

2
crates/router/src/types/domain.rs
View File

@ -1,6 +1,7 @@
mod address;
pub mod behaviour;
mod customer;
mod event;
mod merchant_account;
mod merchant_connector_account;
mod merchant_key_store;
@ -10,6 +11,7 @@ pub mod user;
pub use address::*;
pub use customer::*;
pub use event::*;
pub use merchant_account::*;
pub use merchant_connector_account::*;
pub use merchant_key_store::*;

133
crates/router/src/types/domain/event.rs Normal file
View File

@ -0,0 +1,133 @@
use common_utils::crypto::OptionalEncryptableSecretString;
use diesel_models::{
enums::{EventClass, EventObjectType, EventType},
events::EventUpdateInternal,
};
use error_stack::ResultExt;
use masking::{PeekInterface, Secret};
use crate::{
errors::{CustomResult, ValidationError},
types::domain::types::{self, AsyncLift},
};
#[derive(Clone, Debug)]
pub struct Event {
pub event_id: String,
pub event_type: EventType,
pub event_class: EventClass,
pub is_webhook_notified: bool,
pub primary_object_id: String,
pub primary_object_type: EventObjectType,
pub created_at: time::PrimitiveDateTime,
pub merchant_id: Option<String>,
pub business_profile_id: Option<String>,
pub primary_object_created_at: Option<time::PrimitiveDateTime>,
pub idempotent_event_id: Option<String>,
pub initial_attempt_id: Option<String>,
pub request: OptionalEncryptableSecretString,
pub response: OptionalEncryptableSecretString,
}
#[derive(Debug)]
pub enum EventUpdate {
UpdateResponse {
is_webhook_notified: bool,
response: OptionalEncryptableSecretString,
},
}
impl From<EventUpdate> for EventUpdateInternal {
fn from(event_update: EventUpdate) -> Self {
match event_update {
EventUpdate::UpdateResponse {
is_webhook_notified,
response,
} => Self {
is_webhook_notified: Some(is_webhook_notified),
response: response.map(Into::into),
},
}
}
}
#[async_trait::async_trait]
impl super::behaviour::Conversion for Event {
type DstType = diesel_models::events::Event;
type NewDstType = diesel_models::events::EventNew;
async fn convert(self) -> CustomResult<Self::DstType, ValidationError> {
Ok(diesel_models::events::Event {
event_id: self.event_id,
event_type: self.event_type,
event_class: self.event_class,
is_webhook_notified: self.is_webhook_notified,
primary_object_id: self.primary_object_id,
primary_object_type: self.primary_object_type,
created_at: self.created_at,
merchant_id: self.merchant_id,
business_profile_id: self.business_profile_id,
primary_object_created_at: self.primary_object_created_at,
idempotent_event_id: self.idempotent_event_id,
initial_attempt_id: self.initial_attempt_id,
request: self.request.map(Into::into),
response: self.response.map(Into::into),
})
}
async fn convert_back(
item: Self::DstType,
key: &Secret<Vec<u8>>,
) -> CustomResult<Self, ValidationError>
where
Self: Sized,
{
async {
Ok(Self {
event_id: item.event_id,
event_type: item.event_type,
event_class: item.event_class,
is_webhook_notified: item.is_webhook_notified,
primary_object_id: item.primary_object_id,
primary_object_type: item.primary_object_type,
created_at: item.created_at,
merchant_id: item.merchant_id,
business_profile_id: item.business_profile_id,
primary_object_created_at: item.primary_object_created_at,
idempotent_event_id: item.idempotent_event_id,
initial_attempt_id: item.initial_attempt_id,
request: item
.request
.async_lift(|inner| types::decrypt(inner, key.peek()))
.await?,
response: item
.response
.async_lift(|inner| types::decrypt(inner, key.peek()))
.await?,
})
}
.await
.change_context(ValidationError::InvalidValue {
message: "Failed while decrypting event data".to_string(),
})
}
async fn construct_new(self) -> CustomResult<Self::NewDstType, ValidationError> {
Ok(diesel_models::events::EventNew {
event_id: self.event_id,
event_type: self.event_type,
event_class: self.event_class,
is_webhook_notified: self.is_webhook_notified,
primary_object_id: self.primary_object_id,
primary_object_type: self.primary_object_type,
created_at: self.created_at,
merchant_id: self.merchant_id,
business_profile_id: self.business_profile_id,
primary_object_created_at: self.primary_object_created_at,
idempotent_event_id: self.idempotent_event_id,
initial_attempt_id: self.initial_attempt_id,
request: self.request.map(Into::into),
response: self.response.map(Into::into),
})
}
}

2
crates/router/src/types/storage/events.rs
View File

@ -1 +1 @@
pub use diesel_models::events::{Event, EventNew, EventUpdate};
pub use diesel_models::events::{Event, EventNew};

11
crates/router/src/utils.rs
View File

@ -743,9 +743,11 @@ pub fn add_apple_pay_payment_status_metrics(
}
}
#[allow(clippy::too_many_arguments)]
pub async fn trigger_payments_webhook<F, Req, Op>(
merchant_account: domain::MerchantAccount,
business_profile: diesel_models::business_profile::BusinessProfile,
key_store: &domain::MerchantKeyStore,
payment_data: crate::core::payments::PaymentData<F>,
req: Option<Req>,
customer: Option<domain::Customer>,
@ -794,7 +796,8 @@ where
if let services::ApplicationResponse::JsonWithHeaders((payments_response_json, _)) =
payments_response
{
let m_state = state.clone();
let cloned_state = state.clone();
let cloned_key_store = key_store.clone();
// This spawns this futures in a background thread, the exception inside this future won't affect
// the current thread and the lifecycle of spawn thread is not handled by runtime.
// So when server shutdown won't wait for this thread's completion.
@ -802,18 +805,20 @@ where
if let Some(event_type) = event_type {
tokio::spawn(
async move {
let primary_object_created_at = payments_response_json.created;
Box::pin(webhooks_core::create_event_and_trigger_outgoing_webhook(
m_state,
cloned_state,
merchant_account,
business_profile,
&cloned_key_store,
event_type,
diesel_models::enums::EventClass::Payments,
None,
payment_id,
diesel_models::enums::EventObjectType::PaymentDetails,
webhooks::OutgoingWebhookContent::PaymentDetails(
payments_response_json,
),
primary_object_created_at,
))
.await
}

180
crates/router/src/workflows/outgoing_webhook_retry.rs
View File

@ -4,6 +4,7 @@ use api_models::{
};
use common_utils::ext_traits::{StringExt, ValueExt};
use error_stack::ResultExt;
use masking::PeekInterface;
use router_env::tracing::{self, instrument};
use scheduler::{
consumer::{self, workflows::ProcessTrackerWorkflow},
@ -12,7 +13,10 @@ use scheduler::{
};
use crate::{
core::webhooks::{self as webhooks_core, types::OutgoingWebhookTrackingData},
core::webhooks::{
self as webhooks_core,
types::{OutgoingWebhookRequestContent, OutgoingWebhookTrackingData},
},
db::StorageInterface,
errors, logger,
routes::AppState,
@ -29,6 +33,7 @@ impl ProcessTrackerWorkflow<AppState> for OutgoingWebhookRetryWorkflow {
state: &'a AppState,
process: storage::ProcessTracker,
) -> Result<(), errors::ProcessTrackerError> {
let delivery_attempt = webhooks_core::types::WebhookDeliveryAttempt::AutomaticRetry;
let tracking_data: OutgoingWebhookTrackingData = process
.tracking_data
.clone()
@ -41,69 +46,150 @@ impl ProcessTrackerWorkflow<AppState> for OutgoingWebhookRetryWorkflow {
&db.get_master_key().to_vec().into(),
)
.await?;
let merchant_account = db
.find_merchant_account_by_merchant_id(&tracking_data.merchant_id, &key_store)
.await?;
let business_profile = db
.find_business_profile_by_profile_id(&tracking_data.business_profile_id)
.await?;
let event_id = format!(
"{}_{}",
tracking_data.primary_object_id, tracking_data.event_type
let event_id = webhooks_core::utils::generate_event_id();
let idempotent_event_id = webhooks_core::utils::get_idempotent_event_id(
&tracking_data.primary_object_id,
tracking_data.event_type,
delivery_attempt,
);
let event = db.find_event_by_event_id(&event_id).await?;
let (content, event_type) = get_outgoing_webhook_content_and_event_type(
state.clone(),
merchant_account.clone(),
key_store,
&tracking_data,
)
.await?;
let initial_event = match &tracking_data.initial_attempt_id {
Some(initial_attempt_id) => {
db.find_event_by_event_id(initial_attempt_id, &key_store)
.await?
}
// Tracking data inserted by old version of application, fetch event using old event ID
// format
None => {
let old_event_id = format!(
"{}_{}",
tracking_data.primary_object_id, tracking_data.event_type
);
db.find_event_by_event_id(&old_event_id, &key_store).await?
}
};
match event_type {
// Resource status is same as the event type of the current event
Some(event_type) if event_type == tracking_data.event_type => {
let outgoing_webhook = OutgoingWebhook {
merchant_id: tracking_data.merchant_id.clone(),
event_id: event_id.clone(),
event_type,
content: content.clone(),
timestamp: event.created_at,
};
let now = common_utils::date_time::now();
let new_event = domain::Event {
event_id,
event_type: initial_event.event_type,
event_class: initial_event.event_class,
is_webhook_notified: false,
primary_object_id: initial_event.primary_object_id,
primary_object_type: initial_event.primary_object_type,
created_at: now,
merchant_id: Some(business_profile.merchant_id.clone()),
business_profile_id: Some(business_profile.profile_id.clone()),
primary_object_created_at: initial_event.primary_object_created_at,
idempotent_event_id: Some(idempotent_event_id),
initial_attempt_id: Some(initial_event.event_id.clone()),
request: initial_event.request,
response: None,
};
webhooks_core::trigger_appropriate_webhook_and_raise_event(
let event = db
.insert_event(new_event, &key_store)
.await
.map_err(|error| {
logger::error!(?error, "Failed to insert event in events table");
error
})?;
match &event.request {
Some(request) => {
let request_content: OutgoingWebhookRequestContent = request
.get_inner()
.peek()
.parse_struct("OutgoingWebhookRequestContent")?;
webhooks_core::trigger_webhook_and_raise_event(
state.clone(),
merchant_account,
business_profile,
outgoing_webhook,
&key_store,
event,
request_content,
webhooks_core::types::WebhookDeliveryAttempt::AutomaticRetry,
content,
event_id,
event_type,
None,
Some(process),
)
.await;
}
// Resource status has changed since the event was created, finish task
_ => {
logger::warn!(
%event_id,
"The current status of the resource `{}` (event type: {:?}) and the status of \
the resource when the event was created (event type: {:?}) differ, finishing task",
tracking_data.primary_object_id,
event_type,
tracking_data.event_type
);
db.as_scheduler()
.finish_process_with_business_status(
process.clone(),
"RESOURCE_STATUS_MISMATCH".to_string(),
)
// Event inserted by old version of application, fetch current information about
// resource
None => {
let merchant_account = db
.find_merchant_account_by_merchant_id(&tracking_data.merchant_id, &key_store)
.await?;
let (content, event_type) = get_outgoing_webhook_content_and_event_type(
state.clone(),
merchant_account.clone(),
key_store.clone(),
&tracking_data,
)
.await?;
match event_type {
// Resource status is same as the event type of the current event
Some(event_type) if event_type == tracking_data.event_type => {
let outgoing_webhook = OutgoingWebhook {
merchant_id: tracking_data.merchant_id.clone(),
event_id: event.event_id.clone(),
event_type,
content: content.clone(),
timestamp: event.created_at,
};
let request_content = webhooks_core::get_outgoing_webhook_request(
&merchant_account,
outgoing_webhook,
business_profile.payment_response_hash_key.as_deref(),
)
.map_err(|error| {
logger::error!(
?error,
"Failed to obtain outgoing webhook request content"
);
errors::ProcessTrackerError::EApiErrorResponse
})?;
webhooks_core::trigger_webhook_and_raise_event(
state.clone(),
business_profile,
&key_store,
event,
request_content,
webhooks_core::types::WebhookDeliveryAttempt::AutomaticRetry,
Some(content),
Some(process),
)
.await;
}
// Resource status has changed since the event was created, finish task
_ => {
logger::warn!(
%event.event_id,
"The current status of the resource `{}` (event type: {:?}) and the status of \
the resource when the event was created (event type: {:?}) differ, finishing task",
tracking_data.primary_object_id,
event_type,
tracking_data.event_type
);
db.as_scheduler()
.finish_process_with_business_status(
process.clone(),
"RESOURCE_STATUS_MISMATCH".to_string(),
)
.await?;
}
}
}
}
};
Ok(())
}

3
crates/router/src/workflows/payment_sync.rs
View File

@ -69,7 +69,7 @@ impl ProcessTrackerWorkflow<AppState> for PaymentsSyncWorkflow {
>(
state,
merchant_account.clone(),
key_store,
key_store.clone(),
operations::PaymentStatus,
tracking_data.clone(),
payment_flows::CallConnectorAction::Trigger,
@ -182,6 +182,7 @@ impl ProcessTrackerWorkflow<AppState> for PaymentsSyncWorkflow {
>(
merchant_account,
business_profile,
&key_store,
payment_data,
None,
customer,