Bug: audit trail - add webhook events



### Webhook Events Audit trail

Build a webhook events audit trail endpoint for fetching all events from `webhook_events_audit` table for ops audit trail.

![Image](https://github.com/juspay/hyperswitch/assets/21202349/3ae37c76-117e-46af-8e5c-3c9b74410efa)


can refer to the existing API Events audit trail implementation [here](https://github.com/juspay/hyperswitch/blob/main/crates/analytics/src/api_event/events.rs)




```
curl --location 'https://sandbox.hyperswitch.io/analytics/v1/api_event_logs?type=Payment&payment_id=pay_yEsFwSqi6Umv8LVGFFPD' \
--header 'Authorization: Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJ1c2VyX2lkIjoiNjBmZDAyNzAtNTE1MS00MWVhLWJiMWItMGY0ZWExZGNmYjQzIiwibWVyY2hhbnRfaWQiOiJtZXJjaGFudF8xNjg5OTI0ODQzIiwicm9sZV9pZCI6Im1lcmNoYW50X2FkbWluIiwiZXhwIjoxNzA0ODgyODg5LCJvcmdfaWQiOiJvcmdfRk5ZWjhQaE5CQjdtV1hodWxWQW0ifQ.LCgIW8uMGVo2WRFt2irzQ1D8sHuRBSbeRkTGt1qxsXM'
```


The API contract would involve a simple `GET analytics/v1/webhook_event_logs` path.
with similar response format as above curl request
diff --git a/crates/analytics/src/clickhouse.rs b/crates/analytics/src/clickhouse.rs
index 964486c9364..b8fd5e6a35d 100644
--- a/crates/analytics/src/clickhouse.rs
+++ b/crates/analytics/src/clickhouse.rs
@@ -21,6 +21,7 @@ use crate::{
         filters::ApiEventFilter,
         metrics::{latency::LatencyAvg, ApiEventMetricRow},
     },
+    outgoing_webhook_event::events::OutgoingWebhookLogsResult,
     sdk_events::events::SdkEventsResult,
     types::TableEngine,
 };
@@ -120,6 +121,7 @@ impl AnalyticsDataSource for ClickhouseClient {
             }
             AnalyticsCollection::SdkEvents => TableEngine::BasicTree,
             AnalyticsCollection::ApiEvents => TableEngine::BasicTree,
+            AnalyticsCollection::OutgoingWebhookEvent => TableEngine::BasicTree,
         }
     }
 }
@@ -145,6 +147,10 @@ impl super::sdk_events::events::SdkEventsFilterAnalytics for ClickhouseClient {}
 impl super::api_event::events::ApiLogsFilterAnalytics for ClickhouseClient {}
 impl super::api_event::filters::ApiEventFilterAnalytics for ClickhouseClient {}
 impl super::api_event::metrics::ApiEventMetricAnalytics for ClickhouseClient {}
+impl super::outgoing_webhook_event::events::OutgoingWebhookLogsFilterAnalytics
+    for ClickhouseClient
+{
+}
 
 #[derive(Debug, serde::Serialize)]
 struct CkhQuery {
@@ -302,6 +308,18 @@ impl TryInto<ApiEventFilter> for serde_json::Value {
     }
 }
 
+impl TryInto<OutgoingWebhookLogsResult> for serde_json::Value {
+    type Error = Report<ParsingError>;
+
+    fn try_into(self) -> Result<OutgoingWebhookLogsResult, Self::Error> {
+        serde_json::from_value(self)
+            .into_report()
+            .change_context(ParsingError::StructParseFailure(
+                "Failed to parse OutgoingWebhookLogsResult in clickhouse results",
+            ))
+    }
+}
+
 impl ToSql<ClickhouseClient> for PrimitiveDateTime {
     fn to_sql(&self, _table_engine: &TableEngine) -> error_stack::Result<String, ParsingError> {
         let format =
@@ -326,6 +344,7 @@ impl ToSql<ClickhouseClient> for AnalyticsCollection {
             Self::SdkEvents => Ok("sdk_events_dist".to_string()),
             Self::ApiEvents => Ok("api_audit_log".to_string()),
             Self::PaymentIntent => Ok("payment_intents_dist".to_string()),
+            Self::OutgoingWebhookEvent => Ok("outgoing_webhook_events_audit".to_string()),
         }
     }
 }
diff --git a/crates/analytics/src/lib.rs b/crates/analytics/src/lib.rs
index 24da77f84f2..8529807a1a1 100644
--- a/crates/analytics/src/lib.rs
+++ b/crates/analytics/src/lib.rs
@@ -7,6 +7,7 @@ mod query;
 pub mod refunds;
 
 pub mod api_event;
+pub mod outgoing_webhook_event;
 pub mod sdk_events;
 mod sqlx;
 mod types;
diff --git a/crates/analytics/src/outgoing_webhook_event.rs b/crates/analytics/src/outgoing_webhook_event.rs
new file mode 100644
index 00000000000..9919d8bbb0f
--- /dev/null
+++ b/crates/analytics/src/outgoing_webhook_event.rs
@@ -0,0 +1,6 @@
+mod core;
+pub mod events;
+
+pub trait OutgoingWebhookEventAnalytics: events::OutgoingWebhookLogsFilterAnalytics {}
+
+pub use self::core::outgoing_webhook_events_core;
diff --git a/crates/analytics/src/outgoing_webhook_event/core.rs b/crates/analytics/src/outgoing_webhook_event/core.rs
new file mode 100644
index 00000000000..5024cc70ec1
--- /dev/null
+++ b/crates/analytics/src/outgoing_webhook_event/core.rs
@@ -0,0 +1,27 @@
+use api_models::analytics::outgoing_webhook_event::OutgoingWebhookLogsRequest;
+use common_utils::errors::ReportSwitchExt;
+use error_stack::{IntoReport, ResultExt};
+
+use super::events::{get_outgoing_webhook_event, OutgoingWebhookLogsResult};
+use crate::{errors::AnalyticsResult, types::FiltersError, AnalyticsProvider};
+
+pub async fn outgoing_webhook_events_core(
+    pool: &AnalyticsProvider,
+    req: OutgoingWebhookLogsRequest,
+    merchant_id: String,
+) -> AnalyticsResult<Vec<OutgoingWebhookLogsResult>> {
+    let data = match pool {
+        AnalyticsProvider::Sqlx(_) => Err(FiltersError::NotImplemented(
+            "Outgoing Webhook Events Logs not implemented for SQLX",
+        ))
+        .into_report()
+        .attach_printable("SQL Analytics is not implemented for Outgoing Webhook Events"),
+        AnalyticsProvider::Clickhouse(ckh_pool)
+        | AnalyticsProvider::CombinedSqlx(_, ckh_pool)
+        | AnalyticsProvider::CombinedCkh(_, ckh_pool) => {
+            get_outgoing_webhook_event(&merchant_id, req, ckh_pool).await
+        }
+    }
+    .switch()?;
+    Ok(data)
+}
diff --git a/crates/analytics/src/outgoing_webhook_event/events.rs b/crates/analytics/src/outgoing_webhook_event/events.rs
new file mode 100644
index 00000000000..e742387e1eb
--- /dev/null
+++ b/crates/analytics/src/outgoing_webhook_event/events.rs
@@ -0,0 +1,90 @@
+use api_models::analytics::{outgoing_webhook_event::OutgoingWebhookLogsRequest, Granularity};
+use common_utils::errors::ReportSwitchExt;
+use error_stack::ResultExt;
+use time::PrimitiveDateTime;
+
+use crate::{
+    query::{Aggregate, GroupByClause, QueryBuilder, ToSql, Window},
+    types::{AnalyticsCollection, AnalyticsDataSource, FiltersError, FiltersResult, LoadRow},
+};
+pub trait OutgoingWebhookLogsFilterAnalytics: LoadRow<OutgoingWebhookLogsResult> {}
+
+pub async fn get_outgoing_webhook_event<T>(
+    merchant_id: &String,
+    query_param: OutgoingWebhookLogsRequest,
+    pool: &T,
+) -> FiltersResult<Vec<OutgoingWebhookLogsResult>>
+where
+    T: AnalyticsDataSource + OutgoingWebhookLogsFilterAnalytics,
+    PrimitiveDateTime: ToSql<T>,
+    AnalyticsCollection: ToSql<T>,
+    Granularity: GroupByClause<T>,
+    Aggregate<&'static str>: ToSql<T>,
+    Window<&'static str>: ToSql<T>,
+{
+    let mut query_builder: QueryBuilder<T> =
+        QueryBuilder::new(AnalyticsCollection::OutgoingWebhookEvent);
+    query_builder.add_select_column("*").switch()?;
+
+    query_builder
+        .add_filter_clause("merchant_id", merchant_id)
+        .switch()?;
+    query_builder
+        .add_filter_clause("payment_id", query_param.payment_id)
+        .switch()?;
+
+    if let Some(event_id) = query_param.event_id {
+        query_builder
+            .add_filter_clause("event_id", &event_id)
+            .switch()?;
+    }
+    if let Some(refund_id) = query_param.refund_id {
+        query_builder
+            .add_filter_clause("refund_id", &refund_id)
+            .switch()?;
+    }
+    if let Some(dispute_id) = query_param.dispute_id {
+        query_builder
+            .add_filter_clause("dispute_id", &dispute_id)
+            .switch()?;
+    }
+    if let Some(mandate_id) = query_param.mandate_id {
+        query_builder
+            .add_filter_clause("mandate_id", &mandate_id)
+            .switch()?;
+    }
+    if let Some(payment_method_id) = query_param.payment_method_id {
+        query_builder
+            .add_filter_clause("payment_method_id", &payment_method_id)
+            .switch()?;
+    }
+    if let Some(attempt_id) = query_param.attempt_id {
+        query_builder
+            .add_filter_clause("attempt_id", &attempt_id)
+            .switch()?;
+    }
+    //TODO!: update the execute_query function to return reports instead of plain errors...
+    query_builder
+        .execute_query::<OutgoingWebhookLogsResult, _>(pool)
+        .await
+        .change_context(FiltersError::QueryBuildingError)?
+        .change_context(FiltersError::QueryExecutionFailure)
+}
+#[derive(Debug, serde::Serialize, serde::Deserialize)]
+pub struct OutgoingWebhookLogsResult {
+    pub merchant_id: String,
+    pub event_id: String,
+    pub event_type: String,
+    pub outgoing_webhook_event_type: String,
+    pub payment_id: String,
+    pub refund_id: Option<String>,
+    pub attempt_id: Option<String>,
+    pub dispute_id: Option<String>,
+    pub payment_method_id: Option<String>,
+    pub mandate_id: Option<String>,
+    pub content: Option<String>,
+    pub is_error: bool,
+    pub error: Option<String>,
+    #[serde(with = "common_utils::custom_serde::iso8601")]
+    pub created_at: PrimitiveDateTime,
+}
diff --git a/crates/analytics/src/sqlx.rs b/crates/analytics/src/sqlx.rs
index cdd2647e4e7..e32b85a5367 100644
--- a/crates/analytics/src/sqlx.rs
+++ b/crates/analytics/src/sqlx.rs
@@ -429,6 +429,8 @@ impl ToSql<SqlxClient> for AnalyticsCollection {
             Self::ApiEvents => Err(error_stack::report!(ParsingError::UnknownError)
                 .attach_printable("ApiEvents table is not implemented for Sqlx"))?,
             Self::PaymentIntent => Ok("payment_intent".to_string()),
+            Self::OutgoingWebhookEvent => Err(error_stack::report!(ParsingError::UnknownError)
+                .attach_printable("OutgoingWebhookEvents table is not implemented for Sqlx"))?,
         }
     }
 }
diff --git a/crates/analytics/src/types.rs b/crates/analytics/src/types.rs
index 8b1bdbd1ab9..8da4655e255 100644
--- a/crates/analytics/src/types.rs
+++ b/crates/analytics/src/types.rs
@@ -26,6 +26,7 @@ pub enum AnalyticsCollection {
     SdkEvents,
     ApiEvents,
     PaymentIntent,
+    OutgoingWebhookEvent,
 }
 
 #[allow(dead_code)]
diff --git a/crates/api_models/src/analytics.rs b/crates/api_models/src/analytics.rs
index 0263427b0fd..e0d3fa671b6 100644
--- a/crates/api_models/src/analytics.rs
+++ b/crates/api_models/src/analytics.rs
@@ -12,6 +12,7 @@ use self::{
 pub use crate::payments::TimeRange;
 
 pub mod api_event;
+pub mod outgoing_webhook_event;
 pub mod payments;
 pub mod refunds;
 pub mod sdk_events;
diff --git a/crates/api_models/src/analytics/outgoing_webhook_event.rs b/crates/api_models/src/analytics/outgoing_webhook_event.rs
new file mode 100644
index 00000000000..b6f0aca056f
--- /dev/null
+++ b/crates/api_models/src/analytics/outgoing_webhook_event.rs
@@ -0,0 +1,10 @@
+#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
+pub struct OutgoingWebhookLogsRequest {
+    pub payment_id: String,
+    pub event_id: Option<String>,
+    pub refund_id: Option<String>,
+    pub dispute_id: Option<String>,
+    pub mandate_id: Option<String>,
+    pub payment_method_id: Option<String>,
+    pub attempt_id: Option<String>,
+}
diff --git a/crates/api_models/src/events.rs b/crates/api_models/src/events.rs
index 457d3fde05b..6d9bd5db342 100644
--- a/crates/api_models/src/events.rs
+++ b/crates/api_models/src/events.rs
@@ -17,7 +17,9 @@ use common_utils::{
 
 use crate::{
     admin::*,
-    analytics::{api_event::*, sdk_events::*, *},
+    analytics::{
+        api_event::*, outgoing_webhook_event::OutgoingWebhookLogsRequest, sdk_events::*, *,
+    },
     api_keys::*,
     cards_info::*,
     disputes::*,
@@ -89,7 +91,8 @@ impl_misc_api_event_type!(
     ApiLogsRequest,
     GetApiEventMetricRequest,
     SdkEventsRequest,
-    ReportRequest
+    ReportRequest,
+    OutgoingWebhookLogsRequest
 );
 
 #[cfg(feature = "stripe")]
diff --git a/crates/router/src/analytics.rs b/crates/router/src/analytics.rs
index f31e908e0dc..c62de5bd29a 100644
--- a/crates/router/src/analytics.rs
+++ b/crates/router/src/analytics.rs
@@ -4,7 +4,7 @@ pub mod routes {
     use actix_web::{web, Responder, Scope};
     use analytics::{
         api_event::api_events_core, errors::AnalyticsError, lambda_utils::invoke_lambda,
-        sdk_events::sdk_events_core,
+        outgoing_webhook_event::outgoing_webhook_events_core, sdk_events::sdk_events_core,
     };
     use api_models::analytics::{
         GenerateReportRequest, GetApiEventFiltersRequest, GetApiEventMetricRequest,
@@ -71,6 +71,10 @@ pub mod routes {
                     )
                     .service(web::resource("api_event_logs").route(web::get().to(get_api_events)))
                     .service(web::resource("sdk_event_logs").route(web::post().to(get_sdk_events)))
+                    .service(
+                        web::resource("outgoing_webhook_event_logs")
+                            .route(web::get().to(get_outgoing_webhook_events)),
+                    )
                     .service(
                         web::resource("filters/api_events")
                             .route(web::post().to(get_api_event_filters)),
@@ -314,6 +318,30 @@ pub mod routes {
         .await
     }
 
+    pub async fn get_outgoing_webhook_events(
+        state: web::Data<AppState>,
+        req: actix_web::HttpRequest,
+        json_payload: web::Query<
+            api_models::analytics::outgoing_webhook_event::OutgoingWebhookLogsRequest,
+        >,
+    ) -> impl Responder {
+        let flow = AnalyticsFlow::GetOutgoingWebhookEvents;
+        Box::pin(api::server_wrap(
+            flow,
+            state,
+            &req,
+            json_payload.into_inner(),
+            |state, auth: AuthenticationData, req| async move {
+                outgoing_webhook_events_core(&state.pool, req, auth.merchant_account.merchant_id)
+                    .await
+                    .map(ApplicationResponse::Json)
+            },
+            &auth::JWTAuth(Permission::Analytics),
+            api_locking::LockAction::NotApplicable,
+        ))
+        .await
+    }
+
     pub async fn get_sdk_events(
         state: web::Data<AppState>,
         req: actix_web::HttpRequest,
diff --git a/crates/router_env/src/lib.rs b/crates/router_env/src/lib.rs
index 3c7ba8b93df..0127d07170f 100644
--- a/crates/router_env/src/lib.rs
+++ b/crates/router_env/src/lib.rs
@@ -52,6 +52,7 @@ pub enum AnalyticsFlow {
     GenerateRefundReport,
     GetApiEventMetrics,
     GetApiEventFilters,
+    GetOutgoingWebhookEvents,
 }
 
 impl FlowMetric for AnalyticsFlow {}

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

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

## Description
adding GET api for query to fetch outgoing webhook events log by specified filters
**Filter:**
```payment_id: String```
```event_id: Option<String>```
```refund_id: Option<String>```
```dispute_id: Option<String>```
```mandate_id: Option<String>```
```payment_method_id: Option<String>```
```attempt_id: Option<String>```

### 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)?
-->
**Api Endpoint:** ```http://localhost:8080/analytics/v1/outgoing_webhook_event_logs```
curl attached below with sample response tested on local machine

## 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
- [ ] I added a [CHANGELOG](/CHANGELOG.md) entry if applicable
