Bug: [FEATURE] Populate status code and delivery attempt information in outgoing webhooks ClickHouse events



### Description

As of opening this issue, the outgoing webhooks ClickHouse events contain a few fields:

https://github.com/juspay/hyperswitch/blob/ca61e47585071865cf7df5c05fdbe3f57818ca95/crates/router/src/events/outgoing_webhook_logs.rs#L11-L21

There are `error` and `is_error` fields that indicate whether an error occurred when delivering the webhook to the merchant server, but do not include any information on whether the merchant server even received the webhook.

To address this issue, we must include the following optional fields in the ClickHouse events:

- `status_code`
- `delivery_attempt`
- `initial_attempt_id`

The fields would have to be added in the `OutgoingWebhookEvent` struct, and the corresponding columns must be added in the ClickHouse table. Once the fields have been added in code, the fields must be populated from the [`Event` in the database](https://github.com/juspay/hyperswitch/blob/ca61e47585071865cf7df5c05fdbe3f57818ca95/crates/router/src/types/domain/event.rs#L15-L31):

- The `delivery_attempt` and `initial_attempt_id` are available directly in the `Event` struct.
- The `status_code` field is available within the `response` struct, if the webhook was delivered to the merchant server.
diff --git a/crates/analytics/docs/clickhouse/scripts/outgoing_webhook_events.sql b/crates/analytics/docs/clickhouse/scripts/outgoing_webhook_events.sql
index 30f3f293bfa..a104bc506fd 100644
--- a/crates/analytics/docs/clickhouse/scripts/outgoing_webhook_events.sql
+++ b/crates/analytics/docs/clickhouse/scripts/outgoing_webhook_events.sql
@@ -12,6 +12,9 @@ CREATE TABLE outgoing_webhook_events_queue (
     `content` Nullable(String),
     `is_error` Bool,
     `error` Nullable(String),
+    `initial_attempt_id` Nullable(String),
+    `status_code` Nullable(UInt16),
+    `delivery_attempt` LowCardinality(String),
     `created_at_timestamp` DateTime64(3)
 ) ENGINE = Kafka SETTINGS kafka_broker_list = 'kafka0:29092',
 kafka_topic_list = 'hyperswitch-outgoing-webhook-events',
@@ -33,6 +36,9 @@ CREATE TABLE outgoing_webhook_events (
     `content` Nullable(String),
     `is_error` Bool,
     `error` Nullable(String),
+    `initial_attempt_id` Nullable(String),
+    `status_code` Nullable(UInt16),
+    `delivery_attempt` LowCardinality(String),
     `created_at` DateTime64(3),
     `inserted_at` DateTime DEFAULT now() CODEC(T64, LZ4),
     INDEX eventIndex event_type TYPE bloom_filter GRANULARITY 1,
@@ -61,6 +67,9 @@ CREATE TABLE outgoing_webhook_events_audit (
     `content` Nullable(String),
     `is_error` Bool,
     `error` Nullable(String),
+    `initial_attempt_id` Nullable(String),
+    `status_code` Nullable(UInt16),
+    `delivery_attempt` LowCardinality(String),
     `created_at` DateTime64(3),
     `inserted_at` DateTime DEFAULT now() CODEC(T64, LZ4)
 ) ENGINE = MergeTree PARTITION BY merchant_id
@@ -81,6 +90,9 @@ CREATE MATERIALIZED VIEW outgoing_webhook_events_mv TO outgoing_webhook_events (
     `content` Nullable(String),
     `is_error` Bool,
     `error` Nullable(String),
+    `initial_attempt_id` Nullable(String),
+    `status_code` Nullable(UInt16),
+    `delivery_attempt` LowCardinality(String),
     `created_at` DateTime64(3),
     `inserted_at` DateTime DEFAULT now() CODEC(T64, LZ4)
 ) AS
@@ -98,6 +110,9 @@ SELECT
     content,
     is_error,
     error,
+    initial_attempt_id,
+    status_code,
+    delivery_attempt,
     created_at_timestamp AS created_at,
     now() AS inserted_at
 FROM
@@ -119,6 +134,9 @@ CREATE MATERIALIZED VIEW outgoing_webhook_events_audit_mv TO outgoing_webhook_ev
     `content` Nullable(String),
     `is_error` Bool,
     `error` Nullable(String),
+    `initial_attempt_id` Nullable(String),
+    `status_code` Nullable(UInt16),
+    `delivery_attempt` LowCardinality(String),
     `created_at` DateTime64(3),
     `inserted_at` DateTime DEFAULT now() CODEC(T64, LZ4)
 ) AS
@@ -136,6 +154,9 @@ SELECT
     content,
     is_error,
     error,
+    initial_attempt_id,
+    status_code,
+    delivery_attempt,
     created_at_timestamp AS created_at,
     now() AS inserted_at
 FROM
diff --git a/crates/router/src/core/webhooks/outgoing.rs b/crates/router/src/core/webhooks/outgoing.rs
index 25c93f8465e..79a23259794 100644
--- a/crates/router/src/core/webhooks/outgoing.rs
+++ b/crates/router/src/core/webhooks/outgoing.rs
@@ -5,7 +5,10 @@ use api_models::{
     webhooks,
 };
 use common_utils::{
-    ext_traits::Encode, request::RequestContent, type_name, types::keymanager::Identifier,
+    ext_traits::{Encode, StringExt},
+    request::RequestContent,
+    type_name,
+    types::keymanager::{Identifier, KeyManagerState},
 };
 use diesel_models::process_tracker::business_status;
 use error_stack::{report, ResultExt};
@@ -33,7 +36,8 @@ use crate::{
     routes::{app::SessionStateInfo, SessionState},
     services,
     types::{
-        api, domain,
+        api,
+        domain::{self},
         storage::{self, enums},
         transformers::ForeignFrom,
     },
@@ -226,7 +230,15 @@ pub(crate) async fn trigger_webhook_and_raise_event(
     )
     .await;
 
-    raise_webhooks_analytics_event(state, trigger_webhook_result, content, merchant_id, event);
+    let _ = raise_webhooks_analytics_event(
+        state,
+        trigger_webhook_result,
+        content,
+        merchant_id,
+        event,
+        merchant_key_store,
+    )
+    .await;
 }
 
 async fn trigger_webhook_to_merchant(
@@ -436,13 +448,17 @@ async fn trigger_webhook_to_merchant(
     Ok(())
 }
 
-fn raise_webhooks_analytics_event(
+async fn raise_webhooks_analytics_event(
     state: SessionState,
     trigger_webhook_result: CustomResult<(), errors::WebhooksFlowError>,
     content: Option<api::OutgoingWebhookContent>,
     merchant_id: common_utils::id_type::MerchantId,
     event: domain::Event,
+    merchant_key_store: &domain::MerchantKeyStore,
 ) {
+    let key_manager_state: &KeyManagerState = &(&state).into();
+    let event_id = event.event_id;
+
     let error = if let Err(error) = trigger_webhook_result {
         logger::error!(?error, "Failed to send webhook to merchant");
 
@@ -462,13 +478,47 @@ fn raise_webhooks_analytics_event(
         .and_then(api::OutgoingWebhookContent::get_outgoing_webhook_event_content)
         .or_else(|| get_outgoing_webhook_event_content_from_event_metadata(event.metadata));
 
+    // Fetch updated_event from db
+    let updated_event = state
+        .store
+        .find_event_by_merchant_id_event_id(
+            key_manager_state,
+            &merchant_id,
+            &event_id,
+            merchant_key_store,
+        )
+        .await
+        .attach_printable_lazy(|| format!("event not found for id: {}", &event_id))
+        .map_err(|error| {
+            logger::error!(?error);
+            error
+        })
+        .ok();
+
+    // Get status_code from webhook response
+    let status_code = updated_event.and_then(|updated_event| {
+        let webhook_response: Option<OutgoingWebhookResponseContent> =
+            updated_event.response.and_then(|res| {
+                res.peek()
+                    .parse_struct("OutgoingWebhookResponseContent")
+                    .map_err(|error| {
+                        logger::error!(?error, "Error deserializing webhook response");
+                        error
+                    })
+                    .ok()
+            });
+        webhook_response.and_then(|res| res.status_code)
+    });
+
     let webhook_event = OutgoingWebhookEvent::new(
         merchant_id,
-        event.event_id,
+        event_id,
         event.event_type,
         outgoing_webhook_event_content,
         error,
         event.initial_attempt_id,
+        status_code,
+        event.delivery_attempt,
     );
     state.event_handler().log_event(&webhook_event);
 }
diff --git a/crates/router/src/events/outgoing_webhook_logs.rs b/crates/router/src/events/outgoing_webhook_logs.rs
index 7ab100c2503..2a56f60ecac 100644
--- a/crates/router/src/events/outgoing_webhook_logs.rs
+++ b/crates/router/src/events/outgoing_webhook_logs.rs
@@ -1,4 +1,5 @@
 use api_models::{enums::EventType as OutgoingWebhookEventType, webhooks::OutgoingWebhookContent};
+use common_enums::WebhookDeliveryAttempt;
 use serde::Serialize;
 use serde_json::Value;
 use time::OffsetDateTime;
@@ -18,6 +19,8 @@ pub struct OutgoingWebhookEvent {
     error: Option<Value>,
     created_at_timestamp: i128,
     initial_attempt_id: Option<String>,
+    status_code: Option<u16>,
+    delivery_attempt: Option<WebhookDeliveryAttempt>,
 }
 
 #[derive(Clone, Debug, PartialEq, Serialize)]
@@ -89,6 +92,7 @@ impl OutgoingWebhookEventMetric for OutgoingWebhookContent {
 }
 
 impl OutgoingWebhookEvent {
+    #[allow(clippy::too_many_arguments)]
     pub fn new(
         merchant_id: common_utils::id_type::MerchantId,
         event_id: String,
@@ -96,6 +100,8 @@ impl OutgoingWebhookEvent {
         content: Option<OutgoingWebhookEventContent>,
         error: Option<Value>,
         initial_attempt_id: Option<String>,
+        status_code: Option<u16>,
+        delivery_attempt: Option<WebhookDeliveryAttempt>,
     ) -> Self {
         Self {
             merchant_id,
@@ -106,6 +112,8 @@ impl OutgoingWebhookEvent {
             error,
             created_at_timestamp: OffsetDateTime::now_utc().unix_timestamp_nanos() / 1_000_000,
             initial_attempt_id,
+            status_code,
+            delivery_attempt,
         }
     }
 }

## Type of Change

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

## Description

Closes #5070.

This PR enhances the outgoing_webhook_events and outgoing_webhook_events_audit tables by adding three new columns: initial_attempt_id, status_code, and delivery_attempt. These additions are essential for tracking the initial_attempt_id, status_code, and delivery_attempt status for outgoing webhook events, allowing for more detailed monitoring and troubleshooting of webhook deliveries to merchant servers.

### Database Schema Changes On Clickhouse

Please run the following `ALTER TABLE` commands to update the database schema:

#### `outgoing_webhook_events` Table
```sql
ALTER TABLE outgoing_webhook_events
ADD COLUMN initial_attempt_id Nullable(String),
ADD COLUMN status_code Nullable(UInt16),
ADD COLUMN delivery_attempt LowCardinality(String);
```

#### `outgoing_webhook_events_audit` Table
```sql
ALTER TABLE outgoing_webhook_events_audit
ADD COLUMN initial_attempt_id Nullable(String),
ADD COLUMN status_code Nullable(UInt16),
ADD COLUMN delivery_attempt LowCardinality(String);
```


### Additional Changes

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


## How did you test it?
Tested locally. 
<img width="2052" alt="outgoing_webhook_events" src="https://github.com/user-attachments/assets/6f04c804-bfc3-40b8-9d6c-49aca0abadc5">
<img width="2055" alt="outgoing_webhook_events_audit" src="https://github.com/user-attachments/assets/9c453d2f-3cbe-4a9a-88f3-446715115dd9">




## Checklist
- [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
