Bug: [BUG] There is a bug when the webhook response comes from ucs



### Bug Description

There is a bug when the webhook response comes from ucs,
the code was not able to deserialize the response

### Expected Behavior

webhook response form ucs should be handle using ucs logic

### Actual Behavior

a webhook failed

### Steps To Reproduce

Provide an unambiguous set of steps to reproduce this bug. Include code or configuration to reproduce, if relevant.
1. Go to '...'
2. Click on '....'
3. Scroll down to '....'


### Context For The Bug

_No response_

### Environment

Are you using hyperswitch hosted version? Yes/No
If yes, please provide the value of the `x-request-id` response header to help us debug your issue.

If not (or if building/running locally), please provide the following details:
1. Operating System or Linux distribution:
2. Rust version (output of `rustc --version`): ``
3. App version (output of `cargo r --features vergen -- --version`): ``


### Have you spent some time checking if this bug has been raised before?

- [x] I checked and didn't find a similar issue

### Have you read the Contributing Guidelines?

- [x] I have read the [Contributing Guidelines](https://github.com/juspay/hyperswitch/blob/main/docs/CONTRIBUTING.md)

### Are you willing to submit a PR?

None
diff --git a/crates/api_models/src/webhooks.rs b/crates/api_models/src/webhooks.rs
index a01ca3eeeea..bb58a0dc440 100644
--- a/crates/api_models/src/webhooks.rs
+++ b/crates/api_models/src/webhooks.rs
@@ -68,6 +68,67 @@ pub enum IncomingWebhookEvent {
     RecoveryInvoiceCancel,
 }
 
+impl IncomingWebhookEvent {
+    /// Convert UCS event type integer to IncomingWebhookEvent
+    /// Maps from proto WebhookEventType enum values to IncomingWebhookEvent variants
+    pub fn from_ucs_event_type(event_type: i32) -> Self {
+        match event_type {
+            0 => Self::EventNotSupported,
+            // Payment intent events
+            1 => Self::PaymentIntentFailure,
+            2 => Self::PaymentIntentSuccess,
+            3 => Self::PaymentIntentProcessing,
+            4 => Self::PaymentIntentPartiallyFunded,
+            5 => Self::PaymentIntentCancelled,
+            6 => Self::PaymentIntentCancelFailure,
+            7 => Self::PaymentIntentAuthorizationSuccess,
+            8 => Self::PaymentIntentAuthorizationFailure,
+            9 => Self::PaymentIntentCaptureSuccess,
+            10 => Self::PaymentIntentCaptureFailure,
+            11 => Self::PaymentIntentExpired,
+            12 => Self::PaymentActionRequired,
+            // Source events
+            13 => Self::SourceChargeable,
+            14 => Self::SourceTransactionCreated,
+            // Refund events
+            15 => Self::RefundFailure,
+            16 => Self::RefundSuccess,
+            // Dispute events
+            17 => Self::DisputeOpened,
+            18 => Self::DisputeExpired,
+            19 => Self::DisputeAccepted,
+            20 => Self::DisputeCancelled,
+            21 => Self::DisputeChallenged,
+            22 => Self::DisputeWon,
+            23 => Self::DisputeLost,
+            // Mandate events
+            24 => Self::MandateActive,
+            25 => Self::MandateRevoked,
+            // Miscellaneous events
+            26 => Self::EndpointVerification,
+            27 => Self::ExternalAuthenticationARes,
+            28 => Self::FrmApproved,
+            29 => Self::FrmRejected,
+            // Payout events
+            #[cfg(feature = "payouts")]
+            30 => Self::PayoutSuccess,
+            #[cfg(feature = "payouts")]
+            31 => Self::PayoutFailure,
+            #[cfg(feature = "payouts")]
+            32 => Self::PayoutProcessing,
+            #[cfg(feature = "payouts")]
+            33 => Self::PayoutCancelled,
+            #[cfg(feature = "payouts")]
+            34 => Self::PayoutCreated,
+            #[cfg(feature = "payouts")]
+            35 => Self::PayoutExpired,
+            #[cfg(feature = "payouts")]
+            36 => Self::PayoutReversed,
+            _ => Self::EventNotSupported,
+        }
+    }
+}
+
 pub enum WebhookFlow {
     Payment,
     #[cfg(feature = "payouts")]
diff --git a/crates/common_enums/src/enums.rs b/crates/common_enums/src/enums.rs
index 07916214269..806e84ebb9c 100644
--- a/crates/common_enums/src/enums.rs
+++ b/crates/common_enums/src/enums.rs
@@ -573,6 +573,7 @@ pub enum CallConnectorAction {
         error_message: Option<String>,
     },
     HandleResponse(Vec<u8>),
+    UCSHandleResponse(Vec<u8>),
 }
 
 /// The three-letter ISO 4217 currency code (e.g., "USD", "EUR") for the payment amount. This field is mandatory for creating a payment.
diff --git a/crates/router/src/core/payments.rs b/crates/router/src/core/payments.rs
index 5008b3a68b7..da8d9321f22 100644
--- a/crates/router/src/core/payments.rs
+++ b/crates/router/src/core/payments.rs
@@ -4263,7 +4263,13 @@ where
         services::api::ConnectorIntegration<F, RouterDReq, router_types::PaymentsResponseData>,
 {
     record_time_taken_with(|| async {
-        if should_call_unified_connector_service(
+        if !matches!(
+            call_connector_action,
+            CallConnectorAction::UCSHandleResponse(_)
+        ) && !matches!(
+            call_connector_action,
+            CallConnectorAction::HandleResponse(_),
+        ) && should_call_unified_connector_service(
             state,
             merchant_context,
             &router_data,
diff --git a/crates/router/src/core/unified_connector_service/transformers.rs b/crates/router/src/core/unified_connector_service/transformers.rs
index 40422e82542..dec4281e7a9 100644
--- a/crates/router/src/core/unified_connector_service/transformers.rs
+++ b/crates/router/src/core/unified_connector_service/transformers.rs
@@ -1220,6 +1220,7 @@ impl ForeignTryFrom<&hyperswitch_interfaces::webhooks::IncomingWebhookRequestDet
 }
 
 /// Webhook transform data structure containing UCS response information
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 pub struct WebhookTransformData {
     pub event_type: api_models::webhooks::IncomingWebhookEvent,
     pub source_verified: bool,
@@ -1231,16 +1232,8 @@ pub struct WebhookTransformData {
 pub fn transform_ucs_webhook_response(
     response: PaymentServiceTransformResponse,
 ) -> Result<WebhookTransformData, error_stack::Report<errors::ApiErrorResponse>> {
-    let event_type = match response.event_type {
-        0 => api_models::webhooks::IncomingWebhookEvent::PaymentIntentSuccess,
-        1 => api_models::webhooks::IncomingWebhookEvent::PaymentIntentFailure,
-        2 => api_models::webhooks::IncomingWebhookEvent::PaymentIntentProcessing,
-        3 => api_models::webhooks::IncomingWebhookEvent::PaymentIntentCancelled,
-        4 => api_models::webhooks::IncomingWebhookEvent::RefundSuccess,
-        5 => api_models::webhooks::IncomingWebhookEvent::RefundFailure,
-        6 => api_models::webhooks::IncomingWebhookEvent::MandateRevoked,
-        _ => api_models::webhooks::IncomingWebhookEvent::EventNotSupported,
-    };
+    let event_type =
+        api_models::webhooks::IncomingWebhookEvent::from_ucs_event_type(response.event_type);
 
     Ok(WebhookTransformData {
         event_type,
diff --git a/crates/router/src/core/webhooks/incoming.rs b/crates/router/src/core/webhooks/incoming.rs
index 29a6c6c8e26..4d38ce2ed02 100644
--- a/crates/router/src/core/webhooks/incoming.rs
+++ b/crates/router/src/core/webhooks/incoming.rs
@@ -695,6 +695,7 @@ async fn process_webhook_business_logic(
                 connector,
                 request_details,
                 event_type,
+                webhook_transform_data,
             ))
             .await
             .attach_printable("Incoming webhook flow for payments failed"),
@@ -931,9 +932,22 @@ async fn payments_incoming_webhook_flow(
     connector: &ConnectorEnum,
     request_details: &IncomingWebhookRequestDetails<'_>,
     event_type: webhooks::IncomingWebhookEvent,
+    webhook_transform_data: &Option<Box<unified_connector_service::WebhookTransformData>>,
 ) -> CustomResult<WebhookResponseTracker, errors::ApiErrorResponse> {
     let consume_or_trigger_flow = if source_verified {
-        payments::CallConnectorAction::HandleResponse(webhook_details.resource_object)
+        // Determine the appropriate action based on UCS availability
+        let resource_object = webhook_details.resource_object;
+
+        match webhook_transform_data.as_ref() {
+            Some(transform_data) => {
+                // Serialize the transform data to pass to UCS handler
+                let transform_data_bytes = serde_json::to_vec(transform_data.as_ref())
+                    .change_context(errors::ApiErrorResponse::InternalServerError)
+                    .attach_printable("Failed to serialize UCS webhook transform data")?;
+                payments::CallConnectorAction::UCSHandleResponse(transform_data_bytes)
+            }
+            None => payments::CallConnectorAction::HandleResponse(resource_object),
+        }
     } else {
         payments::CallConnectorAction::Trigger
     };
diff --git a/crates/router/src/services/api.rs b/crates/router/src/services/api.rs
index 41a12f35576..fa8e840fdea 100644
--- a/crates/router/src/services/api.rs
+++ b/crates/router/src/services/api.rs
@@ -61,7 +61,7 @@ use crate::{
     core::{
         api_locking,
         errors::{self, CustomResult},
-        payments,
+        payments, unified_connector_service,
     },
     events::{
         api_logs::{ApiEvent, ApiEventMetric, ApiEventsType},
@@ -127,6 +127,62 @@ pub type BoxedBillingConnectorPaymentsSyncIntegrationInterface<T, Req, Res> =
 pub type BoxedVaultConnectorIntegrationInterface<T, Req, Res> =
     BoxedConnectorIntegrationInterface<T, common_types::VaultConnectorFlowData, Req, Res>;
 
+/// Handle UCS webhook response processing
+fn handle_ucs_response<T, Req, Resp>(
+    router_data: types::RouterData<T, Req, Resp>,
+    transform_data_bytes: Vec<u8>,
+) -> CustomResult<types::RouterData<T, Req, Resp>, errors::ConnectorError>
+where
+    T: Clone + Debug + 'static,
+    Req: Debug + Clone + 'static,
+    Resp: Debug + Clone + 'static,
+{
+    let webhook_transform_data: unified_connector_service::WebhookTransformData =
+        serde_json::from_slice(&transform_data_bytes)
+            .change_context(errors::ConnectorError::ResponseDeserializationFailed)
+            .attach_printable("Failed to deserialize UCS webhook transform data")?;
+
+    let webhook_content = webhook_transform_data
+        .webhook_content
+        .ok_or(errors::ConnectorError::ResponseDeserializationFailed)
+        .attach_printable("UCS webhook transform data missing webhook_content")?;
+
+    let payment_get_response = match webhook_content.content {
+        Some(unified_connector_service_client::payments::webhook_response_content::Content::PaymentsResponse(payments_response)) => {
+            Ok(payments_response)
+        },
+        Some(unified_connector_service_client::payments::webhook_response_content::Content::RefundsResponse(_)) => {
+            Err(errors::ConnectorError::ProcessingStepFailed(Some("UCS webhook contains refund response but payment processing was expected".to_string().into())).into())
+        },
+        Some(unified_connector_service_client::payments::webhook_response_content::Content::DisputesResponse(_)) => {
+            Err(errors::ConnectorError::ProcessingStepFailed(Some("UCS webhook contains dispute response but payment processing was expected".to_string().into())).into())
+        },
+        None => {
+            Err(errors::ConnectorError::ResponseDeserializationFailed)
+                .attach_printable("UCS webhook content missing payments_response")
+        }
+    }?;
+
+    let (status, router_data_response, status_code) =
+        unified_connector_service::handle_unified_connector_service_response_for_payment_get(
+            payment_get_response.clone(),
+        )
+        .change_context(errors::ConnectorError::ProcessingStepFailed(None))
+        .attach_printable("Failed to process UCS webhook response using PSync handler")?;
+
+    let mut updated_router_data = router_data;
+    updated_router_data.status = status;
+
+    let _ = router_data_response.map_err(|error_response| {
+        updated_router_data.response = Err(error_response);
+    });
+    updated_router_data.raw_connector_response =
+        payment_get_response.raw_connector_response.map(Secret::new);
+    updated_router_data.connector_http_status_code = Some(status_code);
+
+    Ok(updated_router_data)
+}
+
 fn store_raw_connector_response_if_required<T, Req, Resp>(
     return_raw_connector_response: Option<bool>,
     router_data: &mut types::RouterData<T, Req, Resp>,
@@ -186,6 +242,9 @@ where
             };
             connector_integration.handle_response(req, None, response)
         }
+        payments::CallConnectorAction::UCSHandleResponse(transform_data_bytes) => {
+            handle_ucs_response(router_data, transform_data_bytes)
+        }
         payments::CallConnectorAction::Avoid => Ok(router_data),
         payments::CallConnectorAction::StatusUpdate {
             status,

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

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

## Description
while handling the response of UCS for webhooks Hyperswitch used to use the existing connector code (Psync's Handle Respnse) which gave error for UCS response so fixed this issue.
Closes #9027 

### Additional Changes

- [ ] This PR modifies the API contract
- [ ] This PR modifies the database schema
- [ ] This PR modifies application configuration/environment variables

<!--
Provide links to the files with corresponding changes.

Following are the paths where you can find config files:
1. `config`
2. `crates/router/src/configs`
3. `loadtest/config`
-->


## Motivation and Context
<!--
Why is this change required? What problem does it solve?
If it fixes an open issue, please link to the issue here.

If you don't have an issue, we'd recommend starting with one first so the PR
can focus on the implementation (unless it is an obvious bug or documentation fix
that will have little conversation).
-->


## How did you test it?
<!--
Did you write an integration/unit/API test to verify the code changes?
Or did you test this change manually (provide relevant screenshots)?
-->
testing of the webhook can be done by the step in this merged PR:https://github.com/juspay/hyperswitch/pull/8814


## Checklist
<!-- Put an `x` in the boxes that apply -->

- [x] I formatted the code `cargo +nightly fmt --all`
- [x] I addressed lints thrown by `cargo clippy`
- [x] I reviewed the submitted code
- [ ] I added unit tests for my changes where possible
