Bug: Adding support for Tenant ID in Kafka events apart from Transactional events ( API , Connector, Outgoing )



Need to add tenant-id for kafka events for the domains API events, Connector events and Outgoing Webhook Events
diff --git a/crates/hyperswitch_interfaces/src/events/connector_api_logs.rs b/crates/hyperswitch_interfaces/src/events/connector_api_logs.rs
index 5edd8d68c32..bf9eafe5aaf 100644
--- a/crates/hyperswitch_interfaces/src/events/connector_api_logs.rs
+++ b/crates/hyperswitch_interfaces/src/events/connector_api_logs.rs
@@ -9,6 +9,7 @@ use time::OffsetDateTime;
 /// struct ConnectorEvent
 #[derive(Debug, Serialize)]
 pub struct ConnectorEvent {
+    tenant_id: common_utils::id_type::TenantId,
     connector_name: String,
     flow: String,
     request: String,
@@ -31,6 +32,7 @@ impl ConnectorEvent {
     /// fn new ConnectorEvent
     #[allow(clippy::too_many_arguments)]
     pub fn new(
+        tenant_id: common_utils::id_type::TenantId,
         connector_name: String,
         flow: &str,
         request: serde_json::Value,
@@ -45,6 +47,7 @@ impl ConnectorEvent {
         status_code: u16,
     ) -> Self {
         Self {
+            tenant_id,
             connector_name,
             flow: flow
                 .rsplit_once("::")
diff --git a/crates/router/src/core/webhooks/incoming.rs b/crates/router/src/core/webhooks/incoming.rs
index fe844ef6d73..320283701ef 100644
--- a/crates/router/src/core/webhooks/incoming.rs
+++ b/crates/router/src/core/webhooks/incoming.rs
@@ -99,6 +99,7 @@ pub async fn incoming_webhooks_wrapper<W: types::OutgoingWebhookType>(
         .attach_printable("Could not convert webhook effect to string")?;
 
     let api_event = ApiEvent::new(
+        state.tenant.tenant_id.clone(),
         Some(merchant_account.get_id().clone()),
         flow,
         &request_id,
diff --git a/crates/router/src/core/webhooks/incoming_v2.rs b/crates/router/src/core/webhooks/incoming_v2.rs
index b91a6f0e9a0..2ec17246a6a 100644
--- a/crates/router/src/core/webhooks/incoming_v2.rs
+++ b/crates/router/src/core/webhooks/incoming_v2.rs
@@ -91,6 +91,7 @@ pub async fn incoming_webhooks_wrapper<W: types::OutgoingWebhookType>(
         .attach_printable("Could not convert webhook effect to string")?;
 
     let api_event = ApiEvent::new(
+        state.tenant.tenant_id.clone(),
         Some(merchant_account.get_id().clone()),
         flow,
         &request_id,
diff --git a/crates/router/src/core/webhooks/outgoing.rs b/crates/router/src/core/webhooks/outgoing.rs
index 0a9c17ed3a0..4a1f6cbc9a2 100644
--- a/crates/router/src/core/webhooks/outgoing.rs
+++ b/crates/router/src/core/webhooks/outgoing.rs
@@ -504,6 +504,7 @@ async fn raise_webhooks_analytics_event(
     });
 
     let webhook_event = OutgoingWebhookEvent::new(
+        state.tenant.tenant_id.clone(),
         merchant_id,
         event_id,
         event.event_type,
diff --git a/crates/router/src/events/api_logs.rs b/crates/router/src/events/api_logs.rs
index 76cbcb958c8..419164810c3 100644
--- a/crates/router/src/events/api_logs.rs
+++ b/crates/router/src/events/api_logs.rs
@@ -24,6 +24,7 @@ use crate::{
 #[derive(Clone, Debug, Eq, PartialEq, Serialize)]
 #[serde(rename_all = "snake_case")]
 pub struct ApiEvent {
+    tenant_id: common_utils::id_type::TenantId,
     merchant_id: Option<common_utils::id_type::MerchantId>,
     api_flow: String,
     created_at_timestamp: i128,
@@ -47,6 +48,7 @@ pub struct ApiEvent {
 impl ApiEvent {
     #[allow(clippy::too_many_arguments)]
     pub fn new(
+        tenant_id: common_utils::id_type::TenantId,
         merchant_id: Option<common_utils::id_type::MerchantId>,
         api_flow: &impl FlowMetric,
         request_id: &RequestId,
@@ -62,6 +64,7 @@ impl ApiEvent {
         http_method: &http::Method,
     ) -> Self {
         Self {
+            tenant_id,
             merchant_id,
             api_flow: api_flow.to_string(),
             created_at_timestamp: OffsetDateTime::now_utc().unix_timestamp_nanos() / 1_000_000,
diff --git a/crates/router/src/events/outgoing_webhook_logs.rs b/crates/router/src/events/outgoing_webhook_logs.rs
index 64126ec76d9..b56c473920d 100644
--- a/crates/router/src/events/outgoing_webhook_logs.rs
+++ b/crates/router/src/events/outgoing_webhook_logs.rs
@@ -10,6 +10,7 @@ use crate::services::kafka::KafkaMessage;
 #[derive(Clone, Debug, PartialEq, Serialize)]
 #[serde(rename_all = "snake_case")]
 pub struct OutgoingWebhookEvent {
+    tenant_id: common_utils::id_type::TenantId,
     merchant_id: common_utils::id_type::MerchantId,
     event_id: String,
     event_type: OutgoingWebhookEventType,
@@ -147,6 +148,7 @@ impl OutgoingWebhookEventMetric for OutgoingWebhookContent {
 impl OutgoingWebhookEvent {
     #[allow(clippy::too_many_arguments)]
     pub fn new(
+        tenant_id: common_utils::id_type::TenantId,
         merchant_id: common_utils::id_type::MerchantId,
         event_id: String,
         event_type: OutgoingWebhookEventType,
@@ -157,6 +159,7 @@ impl OutgoingWebhookEvent {
         delivery_attempt: Option<WebhookDeliveryAttempt>,
     ) -> Self {
         Self {
+            tenant_id,
             merchant_id,
             event_id,
             event_type,
diff --git a/crates/router/src/services/api.rs b/crates/router/src/services/api.rs
index 9416ff175a8..c808b29280b 100644
--- a/crates/router/src/services/api.rs
+++ b/crates/router/src/services/api.rs
@@ -228,6 +228,7 @@ where
                         })
                         .unwrap_or_default();
                     let mut connector_event = ConnectorEvent::new(
+                        state.tenant.tenant_id.clone(),
                         req.connector.clone(),
                         std::any::type_name::<T>(),
                         masked_request_body,
@@ -851,6 +852,7 @@ where
     };
 
     let api_event = ApiEvent::new(
+        tenant_id,
         Some(merchant_id.clone()),
         flow,
         &request_id,

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

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

## Description
<!-- Describe your changes in detail -->
The changes are required for pushing tenant ID for API events, Connector events and Outgoing Webhook events

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


Attaching screenshots 

<img width="917" alt="Screenshot 2024-12-09 at 3 20 34 PM" src="https://github.com/user-attachments/assets/b8918318-68b5-426c-aad3-9eed76cd5150">
<img width="925" alt="Screenshot 2024-12-09 at 3 20 08 PM" src="https://github.com/user-attachments/assets/b21c5b4b-359b-4d35-a875-c33154c06e66">
<img width="910" alt="Screenshot 2024-12-09 at 3 20 21 PM" src="https://github.com/user-attachments/assets/771b9b0d-c073-49c6-b711-0c5fa24df49e">

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

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