Bug: audit trail - add connector events



### Connector Events Audit trail

Build a connector events audit trail endpoint for fetching all events from `connector_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/connector_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 b8fd5e6a35d..f81c29c801c 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},
     },
+    connector_events::events::ConnectorEventsResult,
     outgoing_webhook_event::events::OutgoingWebhookLogsResult,
     sdk_events::events::SdkEventsResult,
     types::TableEngine,
@@ -121,6 +122,7 @@ impl AnalyticsDataSource for ClickhouseClient {
             }
             AnalyticsCollection::SdkEvents => TableEngine::BasicTree,
             AnalyticsCollection::ApiEvents => TableEngine::BasicTree,
+            AnalyticsCollection::ConnectorEvents => TableEngine::BasicTree,
             AnalyticsCollection::OutgoingWebhookEvent => TableEngine::BasicTree,
         }
     }
@@ -147,6 +149,7 @@ 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::connector_events::events::ConnectorEventLogAnalytics for ClickhouseClient {}
 impl super::outgoing_webhook_event::events::OutgoingWebhookLogsFilterAnalytics
     for ClickhouseClient
 {
@@ -188,6 +191,18 @@ impl TryInto<SdkEventsResult> for serde_json::Value {
     }
 }
 
+impl TryInto<ConnectorEventsResult> for serde_json::Value {
+    type Error = Report<ParsingError>;
+
+    fn try_into(self) -> Result<ConnectorEventsResult, Self::Error> {
+        serde_json::from_value(self)
+            .into_report()
+            .change_context(ParsingError::StructParseFailure(
+                "Failed to parse ConnectorEventsResult in clickhouse results",
+            ))
+    }
+}
+
 impl TryInto<PaymentMetricRow> for serde_json::Value {
     type Error = Report<ParsingError>;
 
@@ -344,6 +359,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::ConnectorEvents => Ok("connector_events_audit".to_string()),
             Self::OutgoingWebhookEvent => Ok("outgoing_webhook_events_audit".to_string()),
         }
     }
diff --git a/crates/analytics/src/connector_events.rs b/crates/analytics/src/connector_events.rs
new file mode 100644
index 00000000000..c7c31306a2c
--- /dev/null
+++ b/crates/analytics/src/connector_events.rs
@@ -0,0 +1,5 @@
+mod core;
+pub mod events;
+pub trait ConnectorEventAnalytics: events::ConnectorEventLogAnalytics {}
+
+pub use self::core::connector_events_core;
diff --git a/crates/analytics/src/connector_events/core.rs b/crates/analytics/src/connector_events/core.rs
new file mode 100644
index 00000000000..15f841af5f8
--- /dev/null
+++ b/crates/analytics/src/connector_events/core.rs
@@ -0,0 +1,27 @@
+use api_models::analytics::connector_events::ConnectorEventsRequest;
+use common_utils::errors::ReportSwitchExt;
+use error_stack::{IntoReport, ResultExt};
+
+use super::events::{get_connector_events, ConnectorEventsResult};
+use crate::{errors::AnalyticsResult, types::FiltersError, AnalyticsProvider};
+
+pub async fn connector_events_core(
+    pool: &AnalyticsProvider,
+    req: ConnectorEventsRequest,
+    merchant_id: String,
+) -> AnalyticsResult<Vec<ConnectorEventsResult>> {
+    let data = match pool {
+        AnalyticsProvider::Sqlx(_) => Err(FiltersError::NotImplemented(
+            "Connector Events not implemented for SQLX",
+        ))
+        .into_report()
+        .attach_printable("SQL Analytics is not implemented for Connector Events"),
+        AnalyticsProvider::Clickhouse(ckh_pool)
+        | AnalyticsProvider::CombinedSqlx(_, ckh_pool)
+        | AnalyticsProvider::CombinedCkh(_, ckh_pool) => {
+            get_connector_events(&merchant_id, req, ckh_pool).await
+        }
+    }
+    .switch()?;
+    Ok(data)
+}
diff --git a/crates/analytics/src/connector_events/events.rs b/crates/analytics/src/connector_events/events.rs
new file mode 100644
index 00000000000..096520777ee
--- /dev/null
+++ b/crates/analytics/src/connector_events/events.rs
@@ -0,0 +1,63 @@
+use api_models::analytics::{
+    connector_events::{ConnectorEventsRequest, QueryType},
+    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 ConnectorEventLogAnalytics: LoadRow<ConnectorEventsResult> {}
+
+pub async fn get_connector_events<T>(
+    merchant_id: &String,
+    query_param: ConnectorEventsRequest,
+    pool: &T,
+) -> FiltersResult<Vec<ConnectorEventsResult>>
+where
+    T: AnalyticsDataSource + ConnectorEventLogAnalytics,
+    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::ConnectorEvents);
+    query_builder.add_select_column("*").switch()?;
+
+    query_builder
+        .add_filter_clause("merchant_id", merchant_id)
+        .switch()?;
+    match query_param.query_param {
+        QueryType::Payment { payment_id } => query_builder
+            .add_filter_clause("payment_id", payment_id)
+            .switch()?,
+    }
+    //TODO!: update the execute_query function to return reports instead of plain errors...
+    query_builder
+        .execute_query::<ConnectorEventsResult, _>(pool)
+        .await
+        .change_context(FiltersError::QueryBuildingError)?
+        .change_context(FiltersError::QueryExecutionFailure)
+}
+
+#[derive(Debug, serde::Serialize, serde::Deserialize)]
+pub struct ConnectorEventsResult {
+    pub merchant_id: String,
+    pub payment_id: String,
+    pub connector_name: Option<String>,
+    pub request_id: Option<String>,
+    pub flow: String,
+    pub request: String,
+    pub response: Option<String>,
+    pub error: Option<String>,
+    pub status_code: u16,
+    pub latency: Option<u128>,
+    #[serde(with = "common_utils::custom_serde::iso8601")]
+    pub created_at: PrimitiveDateTime,
+    pub method: Option<String>,
+}
diff --git a/crates/analytics/src/lib.rs b/crates/analytics/src/lib.rs
index 8529807a1a1..501bd58527c 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 connector_events;
 pub mod outgoing_webhook_event;
 pub mod sdk_events;
 mod sqlx;
diff --git a/crates/analytics/src/sqlx.rs b/crates/analytics/src/sqlx.rs
index e32b85a5367..7ab8a2aa4bc 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::ConnectorEvents => Err(error_stack::report!(ParsingError::UnknownError)
+                .attach_printable("ConnectorEvents table is not implemented for Sqlx"))?,
             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 8da4655e255..18e9e9f4334 100644
--- a/crates/analytics/src/types.rs
+++ b/crates/analytics/src/types.rs
@@ -26,6 +26,7 @@ pub enum AnalyticsCollection {
     SdkEvents,
     ApiEvents,
     PaymentIntent,
+    ConnectorEvents,
     OutgoingWebhookEvent,
 }
 
diff --git a/crates/api_models/src/analytics.rs b/crates/api_models/src/analytics.rs
index e0d3fa671b6..c6ca215f9f7 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 connector_events;
 pub mod outgoing_webhook_event;
 pub mod payments;
 pub mod refunds;
diff --git a/crates/api_models/src/analytics/connector_events.rs b/crates/api_models/src/analytics/connector_events.rs
new file mode 100644
index 00000000000..b2974b0a339
--- /dev/null
+++ b/crates/api_models/src/analytics/connector_events.rs
@@ -0,0 +1,11 @@
+#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
+#[serde(tag = "type")]
+pub enum QueryType {
+    Payment { payment_id: String },
+}
+
+#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
+pub struct ConnectorEventsRequest {
+    #[serde(flatten)]
+    pub query_param: QueryType,
+}
diff --git a/crates/api_models/src/events.rs b/crates/api_models/src/events.rs
index 26a9d222d6b..43a72b7e392 100644
--- a/crates/api_models/src/events.rs
+++ b/crates/api_models/src/events.rs
@@ -17,10 +17,12 @@ use common_utils::{
     impl_misc_api_event_type,
 };
 
+#[allow(unused_imports)]
 use crate::{
     admin::*,
     analytics::{
-        api_event::*, outgoing_webhook_event::OutgoingWebhookLogsRequest, sdk_events::*, *,
+        api_event::*, connector_events::ConnectorEventsRequest,
+        outgoing_webhook_event::OutgoingWebhookLogsRequest, sdk_events::*, *,
     },
     api_keys::*,
     cards_info::*,
@@ -94,6 +96,7 @@ impl_misc_api_event_type!(
     GetApiEventMetricRequest,
     SdkEventsRequest,
     ReportRequest,
+    ConnectorEventsRequest,
     OutgoingWebhookLogsRequest
 );
 
diff --git a/crates/router/src/analytics.rs b/crates/router/src/analytics.rs
index c62de5bd29a..3f0febcc592 100644
--- a/crates/router/src/analytics.rs
+++ b/crates/router/src/analytics.rs
@@ -3,7 +3,8 @@ pub use analytics::*;
 pub mod routes {
     use actix_web::{web, Responder, Scope};
     use analytics::{
-        api_event::api_events_core, errors::AnalyticsError, lambda_utils::invoke_lambda,
+        api_event::api_events_core, connector_events::connector_events_core,
+        errors::AnalyticsError, lambda_utils::invoke_lambda,
         outgoing_webhook_event::outgoing_webhook_events_core, sdk_events::sdk_events_core,
     };
     use api_models::analytics::{
@@ -71,6 +72,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("connector_event_logs")
+                            .route(web::get().to(get_connector_events)),
+                    )
                     .service(
                         web::resource("outgoing_webhook_event_logs")
                             .route(web::get().to(get_outgoing_webhook_events)),
@@ -585,4 +590,26 @@ pub mod routes {
         ))
         .await
     }
+
+    pub async fn get_connector_events(
+        state: web::Data<AppState>,
+        req: actix_web::HttpRequest,
+        json_payload: web::Query<api_models::analytics::connector_events::ConnectorEventsRequest>,
+    ) -> impl Responder {
+        let flow = AnalyticsFlow::GetConnectorEvents;
+        Box::pin(api::server_wrap(
+            flow,
+            state,
+            &req,
+            json_payload.into_inner(),
+            |state, auth: AuthenticationData, req| async move {
+                connector_events_core(&state.pool, req, auth.merchant_account.merchant_id)
+                    .await
+                    .map(ApplicationResponse::Json)
+            },
+            &auth::JWTAuth(Permission::Analytics),
+            api_locking::LockAction::NotApplicable,
+        ))
+        .await
+    }
 }
diff --git a/crates/router_env/src/lib.rs b/crates/router_env/src/lib.rs
index 0127d07170f..9139b5eed41 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,
+    GetConnectorEvents,
     GetOutgoingWebhookEvents,
 }
 

## 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 connector events log by specified `payment_id`


### Additional Changes

- [x] 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?
```
curl 'http://localhost:8080/analytics/v1/connector_event_logs?type=Payment&payment_id=pay_yfkCogrWxVbB8O0biPkh' \
  -H 'Accept: */*' \
  -H 'Accept-Language: en-GB,en' \
  -H 'Authorization: Bearer *' \
  -H 'Connection: keep-alive' \
  -H 'Content-Type: application/json' \
  -H 'Origin: http://localhost:9000' \
  -H 'Referer: http://localhost:9000/' \
  -H 'Sec-Fetch-Dest: empty' \
  -H 'Sec-Fetch-Mode: cors' \
  -H 'Sec-Fetch-Site: same-site' \
  -H 'Sec-GPC: 1' \
  -H 'User-Agent: Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/112.0.0.0 Safari/537.36' \
  -H 'api-key: hyperswitch' \
  -H 'sec-ch-ua: "Chromium";v="112", "Brave";v="112", "Not:A-Brand";v="99"' \
  -H 'sec-ch-ua-mobile: ?0' \
  -H 'sec-ch-ua-platform: "macOS"' \
  --compressed
```

`[{"merchant_id":"Allconnector123","payment_id":"pay_yfkCogrWxVbB8O0biPkh","connector_name":"stripe","request_id":"018cf773-0564-77e8-9e89-6870acfff2db","flow":"Authorize","request":"{\"amount\":6500,\"currency\":\"USD\",\"statement_descriptor_suffix\":null,\"statement_descriptor\":null,\"metadata[order_id]\":\"pay_yfkCogrWxVbB8O0biPkh_1\",\"metadata[is_refund_id_as_reference]\":null,\"return_url\":\"https://sandbox.hyperswitch.io/payments/pay_yfkCogrWxVbB8O0biPkh/Allconnector123/redirect/response/stripe\",\"confirm\":true,\"mandate\":null,\"payment_method\":null,\"customer\":\"*** alloc::string::String ***\",\"description\":\"Hello this is description\",\"shipping[address][city]\":\"Banglore\",\"shipping[address][country]\":\"US\",\"shipping[address][line1]\":\"*** alloc::string::String ***\",\"shipping[address][line2]\":\"*** alloc::string::String ***\",\"shipping[address][postal_code]\":\"*** alloc::string::String ***\",\"shipping[address][state]\":\"*** alloc::string::String ***\",\"shipping[name]\":\"*** alloc::string::String ***\",\"shipping[phone]\":\"*** alloc::string::String ***\",\"payment_method_data[billing_details][email]\":null,\"payment_method_data[billing_details][address][country]\":null,\"payment_method_data[billing_details][name]\":null,\"payment_method_data[billing_details][address][city]\":null,\"payment_method_data[billing_details][address][line1]\":null,\"payment_method_data[billing_details][address][line2]\":null,\"payment_method_data[billing_details][address][postal_code]\":null,\"payment_method_data[type]\":\"card\",\"payment_method_data[card][number]\":\"400000**********\",\"payment_method_data[card][exp_month]\":\"*** alloc::string::String ***\",\"payment_method_data[card][exp_year]\":\"*** alloc::string::String ***\",\"payment_method_data[card][cvc]\":\"*** alloc::string::String ***\",\"payment_method_options[card][request_three_d_secure]\":\"any\",\"capture_method\":\"automatic\",\"payment_method_options\":null,\"setup_future_usage\":null,\"off_session\":null,\"payment_method_types[0]\":\"card\"}","response":"{\\n  \\\"id\\\": \\\"pi_3OXInzD5R7gDAGff1cGAxxHo\\\",\\n  \\\"object\\\": \\\"payment_intent\\\",\\n  \\\"amount\\\": 6500,\\n  \\\"amount_capturable\\\": 0,\\n  \\\"amount_details\\\": {\\n    \\\"tip\\\": {}\\n  },\\n  \\\"amount_received\\\": 0,\\n  \\\"application\\\": null,\\n  \\\"application_fee_amount\\\": null,\\n  \\\"automatic_payment_methods\\\": null,\\n  \\\"canceled_at\\\": null,\\n  \\\"cancellation_reason\\\": null,\\n  \\\"capture_method\\\": \\\"automatic\\\",\\n  \\\"client_secret\\\": \\\"pi_3OXInzD5R7gDAGff1cGAxxHo_secret_E2nuQt3TQz1PbCGT8B6MiWLGu\\\",\\n  \\\"confirmation_method\\\": \\\"automatic\\\",\\n  \\\"created\\\": 1704958559,\\n  \\\"currency\\\": \\\"usd\\\",\\n  \\\"customer\\\": \\\"cus_OuQL3sWtKEn0Cs\\\",\\n  \\\"description\\\": \\\"Hello this is description\\\",\\n  \\\"invoice\\\": null,\\n  \\\"last_payment_error\\\": null,\\n  \\\"latest_charge\\\": null,\\n  \\\"livemode\\\": false,\\n  \\\"metadata\\\": {\\n    \\\"order_id\\\": \\\"pay_yfkCogrWxVbB8O0biPkh_1\\\"\\n  },\\n  \\\"next_action\\\": {\\n    \\\"redirect_to_url\\\": {\\n      \\\"return_url\\\": \\\"https://sandbox.hyperswitch.io/payments/pay_yfkCogrWxVbB8O0biPkh/Allconnector123/redirect/response/stripe\\\",\\n      \\\"url\\\": \\\"https://hooks.stripe.com/redirect/authenticate/src_1OXIo0D5R7gDAGffxkofsikB?client_secret=src_client_secret_1P7sv3nHuAc684vjLZlUhwT1&source_redirect_slug=test_YWNjdF8xTTdmVGFENVI3Z0RBR2ZmLF9QTTBwNkJKN01LOWIzeFA3bkMxNjhCd2tra01oUnZ50100WMdcNcbT\\\"\\n    },\\n    \\\"type\\\": \\\"redirect_to_url\\\"\\n  },\\n  \\\"on_behalf_of\\\": null,\\n  \\\"payment_method\\\": \\\"pm_1OXInzD5R7gDAGffpXWPgsBW\\\",\\n  \\\"payment_method_configuration_details\\\": null,\\n  \\\"payment_method_options\\\": {\\n    \\\"card\\\": {\\n      \\\"installments\\\": null,\\n      \\\"mandate_options\\\": null,\\n      \\\"network\\\": null,\\n      \\\"request_three_d_secure\\\": \\\"any\\\"\\n    }\\n  },\\n  \\\"payment_method_types\\\": [\\n    \\\"card\\\"\\n  ],\\n  \\\"processing\\\": null,\\n  \\\"receipt_email\\\": null,\\n  \\\"review\\\": null,\\n  \\\"setup_future_usage\\\": null,\\n  \\\"shipping\\\": {\\n    \\\"address\\\": {\\n      \\\"city\\\": \\\"Banglore\\\",\\n      \\\"country\\\": \\\"US\\\",\\n      \\\"line1\\\": \\\"sdsdfsdf\\\",\\n      \\\"line2\\\": \\\"hsgdbhd\\\",\\n      \\\"postal_code\\\": \\\"571201\\\",\\n      \\\"state\\\": \\\"zsaasdas\\\"\\n    },\\n    \\\"carrier\\\": null,\\n    \\\"name\\\": \\\"Bopanna MJ\\\",\\n    \\\"phone\\\": \\\"+1123456789\\\",\\n    \\\"tracking_number\\\": null\\n  },\\n  \\\"source\\\": null,\\n  \\\"statement_descriptor\\\": null,\\n  \\\"statement_descriptor_suffix\\\": null,\\n  \\\"status\\\": \\\"requires_action\\\",\\n  \\\"transfer_data\\\": null,\\n  \\\"transfer_group\\\": null\\n}","error":null,"status_code":0,"latency":891,"created_at":"2024-01-11T07:36:00.507Z","method":"POST"}]`


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