Bug: [FEATURE] Add support for sending additional metadata in the `MessagingInterface`



### Feature Description

In the current implementation of message interface I can only send a Message & a timestamp associated with the message,
There is no accomodation to send additional metadata about the message which isn't part of the data but could be propogated to the downstream implementations,

These could be used downstream as kafka headers or redis hashes or partition keys.

### Possible Implementation

metadata can be a simplistic `HashMap<String, String>` and can be either accomodated as a new param in the [`MessagingInterface`](https://github.com/juspay/hyperswitch/blob/cc1051da99c1b4e007d7f730e2fe3cb08b078d80/crates/events/src/lib.rs#L239) or a new method on the [`Message`](https://github.com/juspay/hyperswitch/blob/cc1051da99c1b4e007d7f730e2fe3cb08b078d80/crates/events/src/lib.rs#L239) Interface.

The implementers of MessageInterface can handle it as follows
1. Kafka
Send the metadata as headers
2. EventLogger
Log the metadata as a separate key along with the raw data

### Have you spent some time checking if this feature request 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/events/src/lib.rs b/crates/events/src/lib.rs
index 0136b64a0f6..cef0c8894de 100644
--- a/crates/events/src/lib.rs
+++ b/crates/events/src/lib.rs
@@ -49,6 +49,11 @@ pub trait Event: EventInfo {
 
     /// The class/type of the event. This is used to group/categorize events together.
     fn class(&self) -> Self::EventType;
+
+    /// Metadata associated with the event
+    fn metadata(&self) -> HashMap<String, String> {
+        HashMap::new()
+    }
 }
 
 /// Hold the context information for any events
@@ -73,7 +78,8 @@ where
     event: E,
 }
 
-struct RawEvent<T, A: Event<EventType = T>>(HashMap<String, Value>, A);
+/// A flattened event that flattens the context provided to it along with the actual event.
+struct FlatMapEvent<T, A: Event<EventType = T>>(HashMap<String, Value>, A);
 
 impl<T, A, E, D> EventBuilder<T, A, E, D>
 where
@@ -109,12 +115,13 @@ where
     /// Emit the event.
     pub fn try_emit(self) -> Result<(), EventsError> {
         let ts = self.event.timestamp();
+        let metadata = self.event.metadata();
         self.message_sink
-            .send_message(RawEvent(self.metadata, self.event), ts)
+            .send_message(FlatMapEvent(self.metadata, self.event), metadata, ts)
     }
 }
 
-impl<T, A> Serialize for RawEvent<T, A>
+impl<T, A> Serialize for FlatMapEvent<T, A>
 where
     A: Event<EventType = T>,
 {
@@ -236,7 +243,12 @@ pub trait MessagingInterface {
     /// The type of the event used for categorization by the event publisher.
     type MessageClass;
     /// Send a message that follows the defined message class.
-    fn send_message<T>(&self, data: T, timestamp: PrimitiveDateTime) -> Result<(), EventsError>
+    fn send_message<T>(
+        &self,
+        data: T,
+        metadata: HashMap<String, String>,
+        timestamp: PrimitiveDateTime,
+    ) -> Result<(), EventsError>
     where
         T: Message<Class = Self::MessageClass> + ErasedMaskSerialize;
 }
@@ -252,7 +264,7 @@ pub trait Message {
     fn identifier(&self) -> String;
 }
 
-impl<T, A> Message for RawEvent<T, A>
+impl<T, A> Message for FlatMapEvent<T, A>
 where
     A: Event<EventType = T>,
 {
diff --git a/crates/router/src/events.rs b/crates/router/src/events.rs
index e1ecd09804c..1af515b8c22 100644
--- a/crates/router/src/events.rs
+++ b/crates/router/src/events.rs
@@ -1,3 +1,5 @@
+use std::collections::HashMap;
+
 use error_stack::ResultExt;
 use events::{EventsError, Message, MessagingInterface};
 use hyperswitch_domain_models::errors::{StorageError, StorageResult};
@@ -94,14 +96,15 @@ impl MessagingInterface for EventsHandler {
     fn send_message<T>(
         &self,
         data: T,
+        metadata: HashMap<String, String>,
         timestamp: PrimitiveDateTime,
     ) -> error_stack::Result<(), EventsError>
     where
         T: Message<Class = Self::MessageClass> + ErasedMaskSerialize,
     {
         match self {
-            Self::Kafka(a) => a.send_message(data, timestamp),
-            Self::Logs(a) => a.send_message(data, timestamp),
+            Self::Kafka(a) => a.send_message(data, metadata, timestamp),
+            Self::Logs(a) => a.send_message(data, metadata, timestamp),
         }
     }
 }
diff --git a/crates/router/src/events/event_logger.rs b/crates/router/src/events/event_logger.rs
index 235ee4a3e3c..004f94942d2 100644
--- a/crates/router/src/events/event_logger.rs
+++ b/crates/router/src/events/event_logger.rs
@@ -1,3 +1,5 @@
+use std::collections::HashMap;
+
 use events::{EventsError, Message, MessagingInterface};
 use masking::ErasedMaskSerialize;
 use time::PrimitiveDateTime;
@@ -21,12 +23,13 @@ impl MessagingInterface for EventLogger {
     fn send_message<T>(
         &self,
         data: T,
+        metadata: HashMap<String, String>,
         _timestamp: PrimitiveDateTime,
     ) -> error_stack::Result<(), EventsError>
     where
         T: Message<Class = Self::MessageClass> + ErasedMaskSerialize,
     {
-        logger::info!(event =? data.masked_serialize().unwrap_or_else(|e| serde_json::json!({"error": e.to_string()})), event_type =? data.get_message_class(), event_id =? data.identifier(), log_type =? "event");
+        logger::info!(event =? data.masked_serialize().unwrap_or_else(|e| serde_json::json!({"error": e.to_string()})), event_type =? data.get_message_class(), event_id =? data.identifier(), log_type =? "event", metadata = ?metadata);
         Ok(())
     }
 }
diff --git a/crates/router/src/middleware.rs b/crates/router/src/middleware.rs
index e0a4f80908d..e1601f8bbbe 100644
--- a/crates/router/src/middleware.rs
+++ b/crates/router/src/middleware.rs
@@ -157,7 +157,8 @@ where
                     payment_method = Empty,
                     status_code = Empty,
                     flow = "UNKNOWN",
-                    golden_log_line = Empty
+                    golden_log_line = Empty,
+                    tenant_id = "ta"
                 )
                 .or_current(),
             ),
diff --git a/crates/router/src/services/kafka.rs b/crates/router/src/services/kafka.rs
index 3523ab9261f..9c38917424e 100644
--- a/crates/router/src/services/kafka.rs
+++ b/crates/router/src/services/kafka.rs
@@ -1,4 +1,4 @@
-use std::sync::Arc;
+use std::{collections::HashMap, sync::Arc};
 
 use bigdecimal::ToPrimitive;
 use common_utils::errors::CustomResult;
@@ -6,6 +6,7 @@ use error_stack::{report, ResultExt};
 use events::{EventsError, Message, MessagingInterface};
 use rdkafka::{
     config::FromClientConfig,
+    message::{Header, OwnedHeaders},
     producer::{BaseRecord, DefaultProducerContext, Producer, ThreadedProducer},
 };
 #[cfg(feature = "payouts")]
@@ -528,6 +529,7 @@ impl MessagingInterface for KafkaProducer {
     fn send_message<T>(
         &self,
         data: T,
+        metadata: HashMap<String, String>,
         timestamp: PrimitiveDateTime,
     ) -> error_stack::Result<(), EventsError>
     where
@@ -538,12 +540,20 @@ impl MessagingInterface for KafkaProducer {
             .masked_serialize()
             .and_then(|i| serde_json::to_vec(&i))
             .change_context(EventsError::SerializationError)?;
+        let mut headers = OwnedHeaders::new();
+        for (k, v) in metadata.iter() {
+            headers = headers.insert(Header {
+                key: k.as_str(),
+                value: Some(v),
+            });
+        }
         self.producer
             .0
             .send(
                 BaseRecord::to(topic)
                     .key(&data.identifier())
                     .payload(&json_data)
+                    .headers(headers)
                     .timestamp(
                         (timestamp.assume_utc().unix_timestamp_nanos() / 1_000_000)
                             .to_i64()

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

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

## Description
<!-- Describe your changes in detail -->
- Allow setting kafka headers in message

### 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)?
-->
<img width="1597" alt="image" src="https://github.com/juspay/hyperswitch/assets/21202349/67529c15-c762-465d-a02a-fab090676c3a">


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