feat(core): Adding integration for webhooks through UCS (#8814)

Author: AmitsinghTanwar007

crates/external_services/src/grpc_client/unified_connector_service.rs

@@ -9,7 +9,8 @@ use tonic::{
 };
 use unified_connector_service_client::payments::{
     self as payments_grpc, payment_service_client::PaymentServiceClient,
-    PaymentServiceAuthorizeResponse,
+    PaymentServiceAuthorizeResponse, PaymentServiceTransformRequest,
+    PaymentServiceTransformResponse,
 };
 
 use crate::{
@@ -96,6 +97,10 @@ pub enum UnifiedConnectorServiceError {
     /// Failed to perform Payment Repeat Payment from gRPC Server
     #[error("Failed to perform Repeat Payment from gRPC Server")]
     PaymentRepeatEverythingFailure,
+
+    /// Failed to transform incoming webhook from gRPC Server
+    #[error("Failed to transform incoming webhook from gRPC Server")]
+    WebhookTransformFailure,
 }
 
 /// Result type for Dynamic Routing
@@ -194,6 +199,7 @@ impl UnifiedConnectorServiceClient {
     ) -> UnifiedConnectorServiceResult<tonic::Response<PaymentServiceAuthorizeResponse>> {
         let mut request = tonic::Request::new(payment_authorize_request);
 
+        let connector_name = connector_auth_metadata.connector_name.clone();
         let metadata =
             build_unified_connector_service_grpc_headers(connector_auth_metadata, grpc_headers)?;
         *request.metadata_mut() = metadata;
@@ -203,7 +209,14 @@ impl UnifiedConnectorServiceClient {
             .authorize(request)
             .await
             .change_context(UnifiedConnectorServiceError::PaymentAuthorizeFailure)
-            .inspect_err(|error| logger::error!(?error))
+            .inspect_err(|error| {
+                logger::error!(
+                    grpc_error=?error,
+                    method="payment_authorize",
+                    connector_name=?connector_name,
+                    "UCS payment authorize gRPC call failed"
+                )
+            })
     }
 
     /// Performs Payment Sync/Get
@@ -216,6 +229,7 @@ impl UnifiedConnectorServiceClient {
     {
         let mut request = tonic::Request::new(payment_get_request);
 
+        let connector_name = connector_auth_metadata.connector_name.clone();
         let metadata =
             build_unified_connector_service_grpc_headers(connector_auth_metadata, grpc_headers)?;
         *request.metadata_mut() = metadata;
@@ -225,7 +239,14 @@ impl UnifiedConnectorServiceClient {
             .get(request)
             .await
             .change_context(UnifiedConnectorServiceError::PaymentGetFailure)
-            .inspect_err(|error| logger::error!(?error))
+            .inspect_err(|error| {
+                logger::error!(
+                    grpc_error=?error,
+                    method="payment_get",
+                    connector_name=?connector_name,
+                    "UCS payment get/sync gRPC call failed"
+                )
+            })
     }
 
     /// Performs Payment Setup Mandate
@@ -238,6 +259,7 @@ impl UnifiedConnectorServiceClient {
     {
         let mut request = tonic::Request::new(payment_register_request);
 
+        let connector_name = connector_auth_metadata.connector_name.clone();
         let metadata =
             build_unified_connector_service_grpc_headers(connector_auth_metadata, grpc_headers)?;
         *request.metadata_mut() = metadata;
@@ -247,7 +269,14 @@ impl UnifiedConnectorServiceClient {
             .register(request)
             .await
             .change_context(UnifiedConnectorServiceError::PaymentRegisterFailure)
-            .inspect_err(|error| logger::error!(?error))
+            .inspect_err(|error| {
+                logger::error!(
+                    grpc_error=?error,
+                    method="payment_setup_mandate",
+                    connector_name=?connector_name,
+                    "UCS payment setup mandate gRPC call failed"
+                )
+            })
     }
 
     /// Performs Payment repeat (MIT - Merchant Initiated Transaction).
@@ -261,6 +290,7 @@ impl UnifiedConnectorServiceClient {
     > {
         let mut request = tonic::Request::new(payment_repeat_request);
 
+        let connector_name = connector_auth_metadata.connector_name.clone();
         let metadata =
             build_unified_connector_service_grpc_headers(connector_auth_metadata, grpc_headers)?;
         *request.metadata_mut() = metadata;
@@ -270,7 +300,43 @@ impl UnifiedConnectorServiceClient {
             .repeat_everything(request)
             .await
             .change_context(UnifiedConnectorServiceError::PaymentRepeatEverythingFailure)
-            .inspect_err(|error| logger::error!(?error))
+            .inspect_err(|error| {
+                logger::error!(
+                    grpc_error=?error,
+                    method="payment_repeat",
+                    connector_name=?connector_name,
+                    "UCS payment repeat gRPC call failed"
+                )
+            })
+    }
+
+    /// Transforms incoming webhook through UCS
+    pub async fn transform_incoming_webhook(
+        &self,
+        webhook_transform_request: PaymentServiceTransformRequest,
+        connector_auth_metadata: ConnectorAuthMetadata,
+        grpc_headers: GrpcHeaders,
+    ) -> UnifiedConnectorServiceResult<tonic::Response<PaymentServiceTransformResponse>> {
+        let mut request = tonic::Request::new(webhook_transform_request);
+
+        let connector_name = connector_auth_metadata.connector_name.clone();
+        let metadata =
+            build_unified_connector_service_grpc_headers(connector_auth_metadata, grpc_headers)?;
+        *request.metadata_mut() = metadata;
+
+        self.client
+            .clone()
+            .transform(request)
+            .await
+            .change_context(UnifiedConnectorServiceError::WebhookTransformFailure)
+            .inspect_err(|error| {
+                logger::error!(
+                    grpc_error=?error,
+                    method="transform_incoming_webhook",
+                    connector_name=?connector_name,
+                    "UCS webhook transform gRPC call failed"
+                )
+            })
     }
 }
 
@@ -318,17 +384,18 @@ pub fn build_unified_connector_service_grpc_headers(
         parse(common_utils_consts::X_MERCHANT_ID, meta.merchant_id.peek())?,
     );
 
-    grpc_headers.tenant_id
-            .parse()
-            .map(|tenant_id| {
-                metadata.append(
-                    common_utils_consts::TENANT_HEADER,
-                    tenant_id)
-            })
-            .inspect_err(
-                |err| logger::warn!(header_parse_error=?err,"invalid {} received",common_utils_consts::TENANT_HEADER),
-            )
-            .ok();
+    if let Err(err) = grpc_headers
+        .tenant_id
+        .parse()
+        .map(|tenant_id| metadata.append(common_utils_consts::TENANT_HEADER, tenant_id))
+    {
+        logger::error!(
+            header_parse_error=?err,
+            tenant_id=?grpc_headers.tenant_id,
+            "Failed to parse tenant_id header for UCS gRPC request: {}",
+            common_utils_consts::TENANT_HEADER
+        );
+    }
 
     Ok(metadata)
 }

crates/router/src/core/payments/helpers.rs

@@ -4188,6 +4188,15 @@ impl MerchantConnectorAccountType {
             Self::CacheVal(_) => None,
         }
     }
+
+    pub fn get_webhook_details(
+        &self,
+    ) -> CustomResult<Option<&masking::Secret<serde_json::Value>>, errors::ApiErrorResponse> {
+        match self {
+            Self::DbVal(db_val) => Ok(db_val.connector_webhook_details.as_ref()),
+            Self::CacheVal(_) => Ok(None),
+        }
+    }
 }
 
 /// Query for merchant connector account either by business label or profile id