feat(connector): add webhook support for worldline connector (#721)

crates/common_utils/src/ext_traits.rs

@@ -176,17 +176,17 @@ impl<T> BytesExt<T> for bytes::Bytes {
 ///
 /// Extending functionalities of `[u8]` for performing parsing
 ///
-pub trait ByteSliceExt<T> {
+pub trait ByteSliceExt {
     ///
     /// Convert `[u8]` into type `<T>` by using `serde::Deserialize`
     ///
-    fn parse_struct<'de>(&'de self, type_name: &str) -> CustomResult<T, errors::ParsingError>
+    fn parse_struct<'de, T>(&'de self, type_name: &str) -> CustomResult<T, errors::ParsingError>
     where
         T: Deserialize<'de>;
 }
 
-impl<T> ByteSliceExt<T> for [u8] {
+impl ByteSliceExt for [u8] {
-    fn parse_struct<'de>(&'de self, type_name: &str) -> CustomResult<T, errors::ParsingError>
+    fn parse_struct<'de, T>(&'de self, type_name: &str) -> CustomResult<T, errors::ParsingError>
     where
         T: Deserialize<'de>,
     {

crates/router/src/connector/worldline.rs

@@ -3,6 +3,7 @@ mod transformers;
 use std::fmt::Debug;
 
 use base64::Engine;
+use common_utils::ext_traits::ByteSliceExt;
 use error_stack::{IntoReport, ResultExt};
 use ring::hmac;
 use storage_models::enums;
@@ -12,8 +13,10 @@ use transformers as worldline;
 use super::utils::RefundsRequestData;
 use crate::{
     configs::settings::Connectors,
+    connector::utils as conn_utils,
     consts,
     core::errors::{self, CustomResult},
+    db::StorageInterface,
     headers, logger,
     services::{self, ConnectorIntegration},
     types::{
@@ -21,7 +24,7 @@ use crate::{
         api::{self, ConnectorCommon, ConnectorCommonExt},
         ErrorResponse,
     },
-    utils::{self, BytesExt},
+    utils::{self, crypto, BytesExt, OptionExt},
 };
 
 #[derive(Debug, Clone)]
@@ -651,27 +654,123 @@ impl ConnectorIntegration<api::RSync, types::RefundsData, types::RefundsResponse
     }
 }
 
+fn is_endpoint_verification(headers: &actix_web::http::header::HeaderMap) -> bool {
+    headers
+        .get("x-gcs-webhooks-endpoint-verification")
+        .is_some()
+}
+
 #[async_trait::async_trait]
 impl api::IncomingWebhook for Worldline {
-    fn get_webhook_object_reference_id(
+    fn get_webhook_source_verification_algorithm(
         &self,
         _request: &api::IncomingWebhookRequestDetails<'_>,
+    ) -> CustomResult<Box<dyn crypto::VerifySignature + Send>, errors::ConnectorError> {
+        Ok(Box::new(crypto::HmacSha256))
+    }
+
+    fn get_webhook_source_verification_signature(
+        &self,
+        request: &api::IncomingWebhookRequestDetails<'_>,
+    ) -> CustomResult<Vec<u8>, errors::ConnectorError> {
+        let header_value = conn_utils::get_header_key_value("X-GCS-Signature", request.headers)?;
+        let signature = consts::BASE64_ENGINE
+            .decode(header_value.as_bytes())
+            .into_report()
+            .change_context(errors::ConnectorError::WebhookSourceVerificationFailed)?;
+        Ok(signature)
+    }
+
+    fn get_webhook_source_verification_message(
+        &self,
+        request: &api::IncomingWebhookRequestDetails<'_>,
+        _merchant_id: &str,
+        _secret: &[u8],
+    ) -> CustomResult<Vec<u8>, errors::ConnectorError> {
+        Ok(request.body.to_vec())
+    }
+
+    async fn get_webhook_source_verification_merchant_secret(
+        &self,
+        db: &dyn StorageInterface,
+        merchant_id: &str,
+    ) -> CustomResult<Vec<u8>, errors::ConnectorError> {
+        let key = format!("whsec_verification_{}_{}", self.id(), merchant_id);
+        let secret = db
+            .get_key(&key)
+            .await
+            .change_context(errors::ConnectorError::WebhookVerificationSecretNotFound)?;
+
+        Ok(secret)
+    }
+
+    fn get_webhook_object_reference_id(
+        &self,
+        request: &api::IncomingWebhookRequestDetails<'_>,
     ) -> CustomResult<String, errors::ConnectorError> {
-        Err(errors::ConnectorError::WebhooksNotImplemented).into_report()
+        || -> _ {
+            Ok(request
+                .body
+                .parse_struct::<worldline::WebhookBody>("WorldlineWebhookEvent")?
+                .payment
+                .parse_value::<worldline::Payment>("WorldlineWebhookObjectId")?
+                .id)
+        }()
+        .change_context(errors::ConnectorError::WebhookReferenceIdNotFound)
     }
 
     fn get_webhook_event_type(
         &self,
-        _request: &api::IncomingWebhookRequestDetails<'_>,
+        request: &api::IncomingWebhookRequestDetails<'_>,
     ) -> CustomResult<api::IncomingWebhookEvent, errors::ConnectorError> {
-        Err(errors::ConnectorError::WebhooksNotImplemented).into_report()
+        if is_endpoint_verification(request.headers) {
+            Ok(api::IncomingWebhookEvent::EndpointVerification)
+        } else {
+            let details: worldline::WebhookBody = request
+                .body
+                .parse_struct("WorldlineWebhookObjectId")
+                .change_context(errors::ConnectorError::WebhookEventTypeNotFound)?;
+            let event = match details.event_type {
+                worldline::WebhookEvent::Paid => api::IncomingWebhookEvent::PaymentIntentSuccess,
+                worldline::WebhookEvent::Rejected | worldline::WebhookEvent::RejectedCapture => {
+                    api::IncomingWebhookEvent::PaymentIntentFailure
+                }
+            };
+            Ok(event)
+        }
     }
 
     fn get_webhook_resource_object(
         &self,
-        _request: &api::IncomingWebhookRequestDetails<'_>,
+        request: &api::IncomingWebhookRequestDetails<'_>,
     ) -> CustomResult<serde_json::Value, errors::ConnectorError> {
-        Err(errors::ConnectorError::WebhooksNotImplemented).into_report()
+        let details = request
+            .body
+            .parse_struct::<worldline::WebhookBody>("WorldlineWebhookObjectId")
+            .change_context(errors::ConnectorError::WebhookResourceObjectNotFound)?
+            .payment
+            .ok_or(errors::ConnectorError::WebhookResourceObjectNotFound)?;
+        Ok(details)
+    }
+
+    fn get_webhook_api_response(
+        &self,
+        request: &api::IncomingWebhookRequestDetails<'_>,
+    ) -> CustomResult<services::api::ApplicationResponse<serde_json::Value>, errors::ConnectorError>
+    {
+        let verification_header = request.headers.get("x-gcs-webhooks-endpoint-verification");
+        let response = match verification_header {
+            None => services::api::ApplicationResponse::StatusOk,
+            Some(header_value) => {
+                let verification_signature_value = header_value
+                    .to_str()
+                    .into_report()
+                    .change_context(errors::ConnectorError::WebhookResponseEncodingFailed)?
+                    .to_string();
+                services::api::ApplicationResponse::TextPlain(verification_signature_value)
+            }
+        };
+        Ok(response)
     }
 }
 

crates/router/src/connector/worldline/transformers.rs

@@ -297,6 +297,7 @@ pub enum PaymentStatus {
     CaptureRequested,
     #[default]
     Processing,
+    Created,
 }
 
 impl ForeignFrom<(PaymentStatus, enums::CaptureMethod)> for enums::AttemptStatus {
@@ -307,7 +308,8 @@ impl ForeignFrom<(PaymentStatus, enums::CaptureMethod)> for enums::AttemptStatus
             | PaymentStatus::Paid
             | PaymentStatus::ChargebackNotification => Self::Charged,
             PaymentStatus::Cancelled => Self::Voided,
-            PaymentStatus::Rejected | PaymentStatus::RejectedCapture => Self::Failure,
+            PaymentStatus::Rejected => Self::Failure,
+            PaymentStatus::RejectedCapture => Self::CaptureFailed,
             PaymentStatus::CaptureRequested => {
                 if capture_method == enums::CaptureMethod::Automatic {
                     Self::Pending
@@ -316,6 +318,7 @@ impl ForeignFrom<(PaymentStatus, enums::CaptureMethod)> for enums::AttemptStatus
                 }
             }
             PaymentStatus::PendingApproval => Self::Authorized,
+            PaymentStatus::Created => Self::Started,
             _ => Self::Pending,
         }
     }
@@ -326,8 +329,8 @@ impl ForeignFrom<(PaymentStatus, enums::CaptureMethod)> for enums::AttemptStatus
 /// To keep this try_from logic generic in case of AUTHORIZE, SYNC and CAPTURE flows capture_method will be set from RouterData request.
 #[derive(Default, Debug, Clone, Deserialize, PartialEq)]
 pub struct Payment {
-    id: String,
+    pub id: String,
-    status: PaymentStatus,
+    pub status: PaymentStatus,
     #[serde(skip_deserializing)]
     pub capture_method: enums::CaptureMethod,
 }
@@ -486,3 +489,28 @@ pub struct ErrorResponse {
     pub error_id: Option<String>,
     pub errors: Vec<Error>,
 }
+
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct WebhookBody {
+    pub api_version: Option<String>,
+    pub id: String,
+    pub created: String,
+    pub merchant_id: String,
+    #[serde(rename = "type")]
+    pub event_type: WebhookEvent,
+    pub payment: Option<serde_json::Value>,
+    pub refund: Option<serde_json::Value>,
+    pub payout: Option<serde_json::Value>,
+    pub token: Option<serde_json::Value>,
+}
+
+#[derive(Debug, Deserialize)]
+pub enum WebhookEvent {
+    #[serde(rename = "payment.rejected")]
+    Rejected,
+    #[serde(rename = "payment.rejected_capture")]
+    RejectedCapture,
+    #[serde(rename = "payment.paid")]
+    Paid,
+}

crates/router/src/core/errors.rs

@@ -281,6 +281,8 @@ pub enum ConnectorError {
     WebhookEventTypeNotFound,
     #[error("Incoming webhook event resource object not found")]
     WebhookResourceObjectNotFound,
+    #[error("Could not respond to the incoming webhook event")]
+    WebhookResponseEncodingFailed,
     #[error("Invalid Date/time format")]
     InvalidDateFormat,
     #[error("Invalid Data format")]

crates/router/src/core/webhooks.rs

@@ -298,16 +298,6 @@ pub async fn webhooks_core<W: api::OutgoingWebhookType>(
         body: &body,
     };
 
-    let source_verified = connector
-        .verify_webhook_source(
-            &*state.store,
-            &request_details,
-            &merchant_account.merchant_id,
-        )
-        .await
-        .change_context(errors::ApiErrorResponse::InternalServerError)
-        .attach_printable("There was an issue in incoming webhook source verification")?;
-
     let decoded_body = connector
         .decode_webhook_body(
             &*state.store,
@@ -325,63 +315,78 @@ pub async fn webhooks_core<W: api::OutgoingWebhookType>(
         .change_context(errors::ApiErrorResponse::InternalServerError)
         .attach_printable("Could not find event type in incoming webhook body")?;
 
-    let process_webhook_further = utils::lookup_webhook_event(
+    if !matches!(
-        &*state.store,
+        event_type,
-        connector_name,
+        api_models::webhooks::IncomingWebhookEvent::EndpointVerification
-        &merchant_account.merchant_id,
+    ) {
-        &event_type,
+        let source_verified = connector
-    )
+            .verify_webhook_source(
-    .await;
+                &*state.store,
-
+                &request_details,
-    if process_webhook_further {
+                &merchant_account.merchant_id,
-        let object_ref_id = connector
+            )
-            .get_webhook_object_reference_id(&request_details)
+            .await
             .change_context(errors::ApiErrorResponse::InternalServerError)
-            .attach_printable("Could not find object reference id in incoming webhook body")?;
+            .attach_printable("There was an issue in incoming webhook source verification")?;
 
-        let event_object = connector
+        let process_webhook_further = utils::lookup_webhook_event(
-            .get_webhook_resource_object(&request_details)
+            &*state.store,
-            .change_context(errors::ApiErrorResponse::InternalServerError)
+            connector_name,
-            .attach_printable("Could not find resource object in incoming webhook body")?;
+            &merchant_account.merchant_id,
+            &event_type,
+        )
+        .await;
 
-        let webhook_details = api::IncomingWebhookDetails {
+        if process_webhook_further {
-            object_reference_id: object_ref_id,
+            let object_ref_id = connector
-            resource_object: Encode::<serde_json::Value>::encode_to_vec(&event_object)
+                .get_webhook_object_reference_id(&request_details)
                 .change_context(errors::ApiErrorResponse::InternalServerError)
-                .attach_printable(
+                .attach_printable("Could not find object reference id in incoming webhook body")?;
-                    "There was an issue when encoding the incoming webhook body to bytes",
-                )?,
-        };
 
-        let flow_type: api::WebhookFlow = event_type.to_owned().into();
+            let event_object = connector
-        match flow_type {
+                .get_webhook_resource_object(&request_details)
-            api::WebhookFlow::Payment => payments_incoming_webhook_flow::<W>(
+                .change_context(errors::ApiErrorResponse::InternalServerError)
-                state.clone(),
+                .attach_printable("Could not find resource object in incoming webhook body")?;
-                merchant_account,
-                webhook_details,
-                source_verified,
-            )
-            .await
-            .change_context(errors::ApiErrorResponse::InternalServerError)
-            .attach_printable("Incoming webhook flow for payments failed")?,
 
-            api::WebhookFlow::Refund => refunds_incoming_webhook_flow::<W>(
+            let webhook_details = api::IncomingWebhookDetails {
-                state.clone(),
+                object_reference_id: object_ref_id,
-                merchant_account,
+                resource_object: Encode::<serde_json::Value>::encode_to_vec(&event_object)
-                webhook_details,
+                    .change_context(errors::ApiErrorResponse::InternalServerError)
-                connector_name,
+                    .attach_printable(
-                source_verified,
+                        "There was an issue when encoding the incoming webhook body to bytes",
-                event_type,
+                    )?,
-            )
+            };
-            .await
-            .change_context(errors::ApiErrorResponse::InternalServerError)
-            .attach_printable("Incoming webhook flow for refunds failed")?,
 
-            api::WebhookFlow::ReturnResponse => {}
+            let flow_type: api::WebhookFlow = event_type.to_owned().into();
+            match flow_type {
+                api::WebhookFlow::Payment => payments_incoming_webhook_flow::<W>(
+                    state.clone(),
+                    merchant_account,
+                    webhook_details,
+                    source_verified,
+                )
+                .await
+                .change_context(errors::ApiErrorResponse::InternalServerError)
+                .attach_printable("Incoming webhook flow for payments failed")?,
 
-            _ => Err(errors::ApiErrorResponse::InternalServerError)
+                api::WebhookFlow::Refund => refunds_incoming_webhook_flow::<W>(
-                .into_report()
+                    state.clone(),
-                .attach_printable("Unsupported Flow Type received in incoming webhooks")?,
+                    merchant_account,
+                    webhook_details,
+                    connector_name,
+                    source_verified,
+                    event_type,
+                )
+                .await
+                .change_context(errors::ApiErrorResponse::InternalServerError)
+                .attach_printable("Incoming webhook flow for refunds failed")?,
+
+                api::WebhookFlow::ReturnResponse => {}
+
+                _ => Err(errors::ApiErrorResponse::InternalServerError)
+                    .into_report()
+                    .attach_printable("Unsupported Flow Type received in incoming webhooks")?,
+            }
         }
     }