Issue:
Bug: feat(events): forward the tenant configuration as part of the kafka message



In a previous [issue](https://github.com/juspay/hyperswitch/issues/4595) we've added tenant id to kafka events so that it can be used in downstream clickhouse,

But due to recent changes the tenant_id won't be used as a unique identifier on clickhouse and instead we need to pass in the entire `Tenant` object to kafka store,

with this the underlying `KafkaProducer` will accept a `Tenant` as well and send the clickhouse_database property from it as a field.

Pull request description:
## 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 -->
Enhance the KafkaProducer struct to include a clickhouse_db_name field. This field is needed for generating audit events that do not use KafkaStore. Additionally, we will implement a method to add tenant details within KafkaProducer and propagate this functionality through EventsHandler.

### 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).
-->
This is to support multitenancy.

## 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)?
-->
After making a payment, you can observe a new field added at the end: `clickhouse_database = default` for `[Topics] 
  hyperswitch-payment-attempt-events`  for sandbox the `clickhouse_database` should have the `hyperswitch_sandbox` or the current clickhouse db in use.
<img width="949" alt="image" src="https://github.com/juspay/hyperswitch/assets/89402434/417e6de2-4d68-4616-85f6-7d6114bd3177">


## 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


Testing:

After making a payment, you can observe a new field added at the end: `clickhouse_database = default` for `[Topics] 
  hyperswitch-payment-attempt-events`  for sandbox the `clickhouse_database` should have the `hyperswitch_sandbox` or the current clickhouse db in use.
<img width="949" alt="image" src="https://github.com/juspay/hyperswitch/assets/89402434/417e6de2-4d68-4616-85f6-7d6114bd3177">

Patch:
diff --git a/config/config.example.toml b/config/config.example.toml
index aa98c0533e3..83c05c90119 100644
--- a/config/config.example.toml
+++ b/config/config.example.toml
@@ -698,7 +698,7 @@ sdk_eligible_payment_methods = "card"
 
 [multitenancy]
 enabled = false
-global_tenant = { schema = "public", redis_key_prefix = "" }
+global_tenant = { schema = "public", redis_key_prefix = "", clickhouse_database = "default"}
 
 [multitenancy.tenants]
 public = { name = "hyperswitch", base_url = "http://localhost:8080", schema = "public", redis_key_prefix = "", clickhouse_database = "default" } # schema -> Postgres db schema, redis_key_prefix -> redis key distinguisher, base_url -> url of the tenant 
diff --git a/config/deployments/env_specific.toml b/config/deployments/env_specific.toml
index 1cf8699f695..a7bd116a322 100644
--- a/config/deployments/env_specific.toml
+++ b/config/deployments/env_specific.toml
@@ -293,7 +293,7 @@ region = "kms_region" # The AWS region used by the KMS SDK for decrypting data.
 
 [multitenancy]
 enabled = false
-global_tenant = { schema = "public", redis_key_prefix = "" }
+global_tenant = { schema = "public", redis_key_prefix = "", clickhouse_database = "default"}
 
 [multitenancy.tenants]
 public = { name = "hyperswitch", base_url = "http://localhost:8080", schema = "public", redis_key_prefix = "", clickhouse_database = "default" }
diff --git a/config/development.toml b/config/development.toml
index 49bc7756cc0..fb09efc74d7 100644
--- a/config/development.toml
+++ b/config/development.toml
@@ -696,7 +696,7 @@ sdk_eligible_payment_methods = "card"
 
 [multitenancy]
 enabled = false
-global_tenant = { schema = "public", redis_key_prefix = "" }
+global_tenant = { schema = "public", redis_key_prefix = "", clickhouse_database = "default"}
 
 [multitenancy.tenants]
 public = { name = "hyperswitch", base_url = "http://localhost:8080", schema = "public", redis_key_prefix = "", clickhouse_database = "default"}
diff --git a/config/docker_compose.toml b/config/docker_compose.toml
index 7f89acd01e2..42256f47121 100644
--- a/config/docker_compose.toml
+++ b/config/docker_compose.toml
@@ -524,7 +524,7 @@ sdk_eligible_payment_methods = "card"
 
 [multitenancy]
 enabled = false
-global_tenant = { schema = "public", redis_key_prefix = "" }
+global_tenant = { schema = "public", redis_key_prefix = "", clickhouse_database = "default"}
 
 [multitenancy.tenants]
 public = { name = "hyperswitch", base_url = "http://localhost:8080", schema = "public", redis_key_prefix = "", clickhouse_database = "default" }
diff --git a/crates/analytics/src/lib.rs b/crates/analytics/src/lib.rs
index 34dcc35d9ee..67fc43875a6 100644
--- a/crates/analytics/src/lib.rs
+++ b/crates/analytics/src/lib.rs
@@ -854,7 +854,7 @@ impl AnalyticsProvider {
 
     pub async fn from_conf(
         config: &AnalyticsConfig,
-        tenant: &dyn storage_impl::config::ClickHouseConfig,
+        tenant: &dyn storage_impl::config::TenantConfig,
     ) -> Self {
         match config {
             AnalyticsConfig::Sqlx { sqlx } => {
diff --git a/crates/router/src/compatibility/wrap.rs b/crates/router/src/compatibility/wrap.rs
index 9ed03c6d2ce..ec43259ad2f 100644
--- a/crates/router/src/compatibility/wrap.rs
+++ b/crates/router/src/compatibility/wrap.rs
@@ -43,14 +43,12 @@ where
 
     let start_instant = Instant::now();
     logger::info!(tag = ?Tag::BeginRequest, payload = ?payload);
-    let req_state = state.get_req_state();
 
     let server_wrap_util_res = metrics::request::record_request_time_metric(
         api::server_wrap_util(
             &flow,
             state.clone().into(),
             request.headers(),
-            req_state,
             request,
             payload,
             func,
diff --git a/crates/router/src/configs/settings.rs b/crates/router/src/configs/settings.rs
index adab685ea09..1b8ca2b148d 100644
--- a/crates/router/src/configs/settings.rs
+++ b/crates/router/src/configs/settings.rs
@@ -176,9 +176,6 @@ impl storage_impl::config::TenantConfig for Tenant {
     fn get_redis_key_prefix(&self) -> &str {
         self.redis_key_prefix.as_str()
     }
-}
-
-impl storage_impl::config::ClickHouseConfig for Tenant {
     fn get_clickhouse_database(&self) -> &str {
         self.clickhouse_database.as_str()
     }
@@ -188,6 +185,7 @@ impl storage_impl::config::ClickHouseConfig for Tenant {
 pub struct GlobalTenant {
     pub schema: String,
     pub redis_key_prefix: String,
+    pub clickhouse_database: String,
 }
 
 impl storage_impl::config::TenantConfig for GlobalTenant {
@@ -197,6 +195,9 @@ impl storage_impl::config::TenantConfig for GlobalTenant {
     fn get_redis_key_prefix(&self) -> &str {
         self.redis_key_prefix.as_str()
     }
+    fn get_clickhouse_database(&self) -> &str {
+        self.clickhouse_database.as_str()
+    }
 }
 
 #[derive(Debug, Deserialize, Clone, Default)]
diff --git a/crates/router/src/db/kafka_store.rs b/crates/router/src/db/kafka_store.rs
index b9d2f0907a2..5f44b941fd2 100644
--- a/crates/router/src/db/kafka_store.rs
+++ b/crates/router/src/db/kafka_store.rs
@@ -26,7 +26,7 @@ use scheduler::{
     SchedulerInterface,
 };
 use serde::Serialize;
-use storage_impl::redis::kv_store::RedisConnInterface;
+use storage_impl::{config::TenantConfig, redis::kv_store::RedisConnInterface};
 use time::PrimitiveDateTime;
 
 use super::{
@@ -89,7 +89,13 @@ pub struct KafkaStore {
 }
 
 impl KafkaStore {
-    pub async fn new(store: Store, kafka_producer: KafkaProducer, tenant_id: TenantID) -> Self {
+    pub async fn new(
+        store: Store,
+        mut kafka_producer: KafkaProducer,
+        tenant_id: TenantID,
+        tenant_config: &dyn TenantConfig,
+    ) -> Self {
+        kafka_producer.set_tenancy(tenant_config);
         Self {
             kafka_producer,
             diesel_store: store,
diff --git a/crates/router/src/events.rs b/crates/router/src/events.rs
index b7ab4ebaa1e..e9a1a17638c 100644
--- a/crates/router/src/events.rs
+++ b/crates/router/src/events.rs
@@ -6,7 +6,7 @@ use hyperswitch_domain_models::errors::{StorageError, StorageResult};
 use masking::ErasedMaskSerialize;
 use router_env::logger;
 use serde::{Deserialize, Serialize};
-use storage_impl::errors::ApplicationError;
+use storage_impl::{config::TenantConfig, errors::ApplicationError};
 use time::PrimitiveDateTime;
 
 use crate::{
@@ -90,6 +90,11 @@ impl EventsHandler {
             Self::Logs(logger) => logger.log_event(event),
         };
     }
+    pub fn add_tenant(&mut self, tenant_config: &dyn TenantConfig) {
+        if let Self::Kafka(kafka_producer) = self {
+            kafka_producer.set_tenancy(tenant_config);
+        }
+    }
 }
 
 impl MessagingInterface for EventsHandler {
diff --git a/crates/router/src/routes/app.rs b/crates/router/src/routes/app.rs
index 4f79670cac0..94f5511ab79 100644
--- a/crates/router/src/routes/app.rs
+++ b/crates/router/src/routes/app.rs
@@ -344,6 +344,7 @@ impl AppState {
                             .expect("Failed to create store"),
                         kafka_client.clone(),
                         TenantID(tenant.get_schema().to_string()),
+                        tenant,
                     )
                     .await,
                 ),
@@ -377,22 +378,19 @@ impl AppState {
         .await
     }
 
-    pub fn get_req_state(&self) -> ReqState {
-        ReqState {
-            event_context: events::EventContext::new(self.event_handler.clone()),
-        }
-    }
     pub fn get_session_state<E, F>(self: Arc<Self>, tenant: &str, err: F) -> Result<SessionState, E>
     where
         F: FnOnce() -> E + Copy,
     {
         let tenant_conf = self.conf.multitenancy.get_tenant(tenant).ok_or_else(err)?;
+        let mut event_handler = self.event_handler.clone();
+        event_handler.add_tenant(tenant_conf);
         Ok(SessionState {
             store: self.stores.get(tenant).ok_or_else(err)?.clone(),
             global_store: self.global_store.clone(),
             conf: Arc::clone(&self.conf),
             api_client: self.api_client.clone(),
-            event_handler: self.event_handler.clone(),
+            event_handler,
             #[cfg(feature = "olap")]
             pool: self.pools.get(tenant).ok_or_else(err)?.clone(),
             file_storage_client: self.file_storage_client.clone(),
diff --git a/crates/router/src/services/api.rs b/crates/router/src/services/api.rs
index ea9ce177439..cba859c07c8 100644
--- a/crates/router/src/services/api.rs
+++ b/crates/router/src/services/api.rs
@@ -751,22 +751,13 @@ pub enum AuthFlow {
 
 #[allow(clippy::too_many_arguments)]
 #[instrument(
-    skip(
-        request,
-        payload,
-        state,
-        func,
-        api_auth,
-        request_state,
-        incoming_request_header
-    ),
+    skip(request, payload, state, func, api_auth, incoming_request_header),
     fields(merchant_id)
 )]
 pub async fn server_wrap_util<'a, 'b, U, T, Q, F, Fut, E, OErr>(
     flow: &'a impl router_env::types::FlowMetric,
     state: web::Data<AppState>,
     incoming_request_header: &HeaderMap,
-    mut request_state: ReqState,
     request: &'a HttpRequest,
     payload: T,
     func: F,
@@ -788,12 +779,6 @@ where
         .attach_printable("Unable to extract request id from request")
         .change_context(errors::ApiErrorResponse::InternalServerError.switch())?;
 
-    request_state.event_context.record_info(request_id);
-    request_state
-        .event_context
-        .record_info(("flow".to_string(), flow.to_string()));
-    // request_state.event_context.record_info(request.clone());
-
     let mut app_state = state.get_ref().clone();
 
     let start_instant = Instant::now();
@@ -826,9 +811,6 @@ where
                 }
             })??
     };
-    request_state
-        .event_context
-        .record_info(("tenant_id".to_string(), tenant_id.to_string()));
     // let tenant_id = "public".to_string();
     let mut session_state =
         Arc::new(app_state.clone()).get_session_state(tenant_id.as_str(), || {
@@ -838,6 +820,16 @@ where
             .switch()
         })?;
     session_state.add_request_id(request_id);
+    let mut request_state = session_state.get_req_state();
+
+    request_state.event_context.record_info(request_id);
+    request_state
+        .event_context
+        .record_info(("flow".to_string(), flow.to_string()));
+
+    request_state
+        .event_context
+        .record_info(("tenant_id".to_string(), tenant_id.to_string()));
 
     // Currently auth failures are not recorded as API events
     let (auth_out, auth_type) = api_auth
@@ -965,7 +957,6 @@ where
     ApplicationResponse<Q>: Debug,
     E: ErrorSwitch<api_models::errors::types::ApiErrorResponse> + error_stack::Context,
 {
-    let req_state = state.get_req_state();
     let request_method = request.method().as_str();
     let url_path = request.path();
 
@@ -1000,7 +991,6 @@ where
             &flow,
             state.clone(),
             incoming_request_header,
-            req_state,
             request,
             payload,
             func,
diff --git a/crates/router/src/services/kafka.rs b/crates/router/src/services/kafka.rs
index baa78339d91..78e08c1c60b 100644
--- a/crates/router/src/services/kafka.rs
+++ b/crates/router/src/services/kafka.rs
@@ -9,6 +9,8 @@ use rdkafka::{
     message::{Header, OwnedHeaders},
     producer::{BaseRecord, DefaultProducerContext, Producer, ThreadedProducer},
 };
+use serde_json::Value;
+use storage_impl::config::TenantConfig;
 #[cfg(feature = "payouts")]
 pub mod payout;
 use diesel_models::fraud_check::FraudCheck;
@@ -70,21 +72,24 @@ struct KafkaEvent<'a, T: KafkaMessage> {
     event: &'a T,
     sign_flag: i32,
     tenant_id: TenantID,
+    clickhouse_database: Option<String>,
 }
 
 impl<'a, T: KafkaMessage> KafkaEvent<'a, T> {
-    fn new(event: &'a T, tenant_id: TenantID) -> Self {
+    fn new(event: &'a T, tenant_id: TenantID, clickhouse_database: Option<String>) -> Self {
         Self {
             event,
             sign_flag: 1,
             tenant_id,
+            clickhouse_database,
         }
     }
-    fn old(event: &'a T, tenant_id: TenantID) -> Self {
+    fn old(event: &'a T, tenant_id: TenantID, clickhouse_database: Option<String>) -> Self {
         Self {
             event,
             sign_flag: -1,
             tenant_id,
+            clickhouse_database,
         }
     }
 }
@@ -263,6 +268,7 @@ pub struct KafkaProducer {
     payout_analytics_topic: String,
     consolidated_events_topic: String,
     authentication_analytics_topic: String,
+    ckh_database_name: Option<String>,
 }
 
 struct RdKafkaProducer(ThreadedProducer<DefaultProducerContext>);
@@ -285,6 +291,10 @@ pub enum KafkaError {
 
 #[allow(unused)]
 impl KafkaProducer {
+    pub fn set_tenancy(&mut self, tenant_config: &dyn TenantConfig) {
+        self.ckh_database_name = Some(tenant_config.get_clickhouse_database().to_string());
+    }
+
     pub async fn create(conf: &KafkaSettings) -> MQResult<Self> {
         Ok(Self {
             producer: Arc::new(RdKafkaProducer(
@@ -307,6 +317,7 @@ impl KafkaProducer {
             payout_analytics_topic: conf.payout_analytics_topic.clone(),
             consolidated_events_topic: conf.consolidated_events_topic.clone(),
             authentication_analytics_topic: conf.authentication_analytics_topic.clone(),
+            ckh_database_name: None,
         })
     }
 
@@ -342,6 +353,7 @@ impl KafkaProducer {
             self.log_event(&KafkaEvent::old(
                 &KafkaFraudCheck::from_storage(&negative_event),
                 tenant_id.clone(),
+                self.ckh_database_name.clone(),
             ))
             .attach_printable_lazy(|| {
                 format!("Failed to add negative fraud check event {negative_event:?}")
@@ -351,6 +363,7 @@ impl KafkaProducer {
         self.log_event(&KafkaEvent::new(
             &KafkaFraudCheck::from_storage(attempt),
             tenant_id.clone(),
+            self.ckh_database_name.clone(),
         ))
         .attach_printable_lazy(|| {
             format!("Failed to add positive fraud check event {attempt:?}")
@@ -375,6 +388,7 @@ impl KafkaProducer {
             self.log_event(&KafkaEvent::old(
                 &KafkaPaymentAttempt::from_storage(&negative_event),
                 tenant_id.clone(),
+                self.ckh_database_name.clone(),
             ))
             .attach_printable_lazy(|| {
                 format!("Failed to add negative attempt event {negative_event:?}")
@@ -384,6 +398,7 @@ impl KafkaProducer {
         self.log_event(&KafkaEvent::new(
             &KafkaPaymentAttempt::from_storage(attempt),
             tenant_id.clone(),
+            self.ckh_database_name.clone(),
         ))
         .attach_printable_lazy(|| format!("Failed to add positive attempt event {attempt:?}"))?;
 
@@ -402,6 +417,7 @@ impl KafkaProducer {
         self.log_event(&KafkaEvent::old(
             &KafkaPaymentAttempt::from_storage(delete_old_attempt),
             tenant_id.clone(),
+            self.ckh_database_name.clone(),
         ))
         .attach_printable_lazy(|| {
             format!("Failed to add negative attempt event {delete_old_attempt:?}")
@@ -418,6 +434,7 @@ impl KafkaProducer {
             self.log_event(&KafkaEvent::old(
                 &KafkaAuthentication::from_storage(&negative_event),
                 tenant_id.clone(),
+                self.ckh_database_name.clone(),
             ))
             .attach_printable_lazy(|| {
                 format!("Failed to add negative authentication event {negative_event:?}")
@@ -427,6 +444,7 @@ impl KafkaProducer {
         self.log_event(&KafkaEvent::new(
             &KafkaAuthentication::from_storage(authentication),
             tenant_id.clone(),
+            self.ckh_database_name.clone(),
         ))
         .attach_printable_lazy(|| {
             format!("Failed to add positive authentication event {authentication:?}")
@@ -451,6 +469,7 @@ impl KafkaProducer {
             self.log_event(&KafkaEvent::old(
                 &KafkaPaymentIntent::from_storage(&negative_event),
                 tenant_id.clone(),
+                self.ckh_database_name.clone(),
             ))
             .attach_printable_lazy(|| {
                 format!("Failed to add negative intent event {negative_event:?}")
@@ -460,6 +479,7 @@ impl KafkaProducer {
         self.log_event(&KafkaEvent::new(
             &KafkaPaymentIntent::from_storage(intent),
             tenant_id.clone(),
+            self.ckh_database_name.clone(),
         ))
         .attach_printable_lazy(|| format!("Failed to add positive intent event {intent:?}"))?;
 
@@ -478,6 +498,7 @@ impl KafkaProducer {
         self.log_event(&KafkaEvent::old(
             &KafkaPaymentIntent::from_storage(delete_old_intent),
             tenant_id.clone(),
+            self.ckh_database_name.clone(),
         ))
         .attach_printable_lazy(|| {
             format!("Failed to add negative intent event {delete_old_intent:?}")
@@ -494,6 +515,7 @@ impl KafkaProducer {
             self.log_event(&KafkaEvent::old(
                 &KafkaRefund::from_storage(&negative_event),
                 tenant_id.clone(),
+                self.ckh_database_name.clone(),
             ))
             .attach_printable_lazy(|| {
                 format!("Failed to add negative refund event {negative_event:?}")
@@ -503,6 +525,7 @@ impl KafkaProducer {
         self.log_event(&KafkaEvent::new(
             &KafkaRefund::from_storage(refund),
             tenant_id.clone(),
+            self.ckh_database_name.clone(),
         ))
         .attach_printable_lazy(|| format!("Failed to add positive refund event {refund:?}"))?;
 
@@ -521,6 +544,7 @@ impl KafkaProducer {
         self.log_event(&KafkaEvent::old(
             &KafkaRefund::from_storage(delete_old_refund),
             tenant_id.clone(),
+            self.ckh_database_name.clone(),
         ))
         .attach_printable_lazy(|| {
             format!("Failed to add negative refund event {delete_old_refund:?}")
@@ -537,6 +561,7 @@ impl KafkaProducer {
             self.log_event(&KafkaEvent::old(
                 &KafkaDispute::from_storage(&negative_event),
                 tenant_id.clone(),
+                self.ckh_database_name.clone(),
             ))
             .attach_printable_lazy(|| {
                 format!("Failed to add negative dispute event {negative_event:?}")
@@ -546,6 +571,7 @@ impl KafkaProducer {
         self.log_event(&KafkaEvent::new(
             &KafkaDispute::from_storage(dispute),
             tenant_id.clone(),
+            self.ckh_database_name.clone(),
         ))
         .attach_printable_lazy(|| format!("Failed to add positive dispute event {dispute:?}"))?;
 
@@ -564,13 +590,21 @@ impl KafkaProducer {
         tenant_id: TenantID,
     ) -> MQResult<()> {
         if let Some(negative_event) = old_payout {
-            self.log_event(&KafkaEvent::old(&negative_event, tenant_id.clone()))
-                .attach_printable_lazy(|| {
-                    format!("Failed to add negative payout event {negative_event:?}")
-                })?;
+            self.log_event(&KafkaEvent::old(
+                &negative_event,
+                tenant_id.clone(),
+                self.ckh_database_name.clone(),
+            ))
+            .attach_printable_lazy(|| {
+                format!("Failed to add negative payout event {negative_event:?}")
+            })?;
         };
-        self.log_event(&KafkaEvent::new(payout, tenant_id.clone()))
-            .attach_printable_lazy(|| format!("Failed to add positive payout event {payout:?}"))
+        self.log_event(&KafkaEvent::new(
+            payout,
+            tenant_id.clone(),
+            self.ckh_database_name.clone(),
+        ))
+        .attach_printable_lazy(|| format!("Failed to add positive payout event {payout:?}"))
     }
 
     #[cfg(feature = "payouts")]
@@ -579,10 +613,14 @@ impl KafkaProducer {
         delete_old_payout: &KafkaPayout<'_>,
         tenant_id: TenantID,
     ) -> MQResult<()> {
-        self.log_event(&KafkaEvent::old(delete_old_payout, tenant_id.clone()))
-            .attach_printable_lazy(|| {
-                format!("Failed to add negative payout event {delete_old_payout:?}")
-            })
+        self.log_event(&KafkaEvent::old(
+            delete_old_payout,
+            tenant_id.clone(),
+            self.ckh_database_name.clone(),
+        ))
+        .attach_printable_lazy(|| {
+            format!("Failed to add negative payout event {delete_old_payout:?}")
+        })
     }
 
     pub fn get_topic(&self, event: EventType) -> &str {
@@ -631,7 +669,14 @@ impl MessagingInterface for KafkaProducer {
         let topic = self.get_topic(data.get_message_class());
         let json_data = data
             .masked_serialize()
-            .and_then(|i| serde_json::to_vec(&i))
+            .and_then(|mut value| {
+                if let Value::Object(ref mut map) = value {
+                    if let Some(db_name) = self.ckh_database_name.clone() {
+                        map.insert("clickhouse_database".to_string(), Value::String(db_name));
+                    }
+                }
+                serde_json::to_vec(&value)
+            })
             .change_context(EventsError::SerializationError)?;
         let mut headers = OwnedHeaders::new();
         for (k, v) in metadata.iter() {
@@ -640,6 +685,10 @@ impl MessagingInterface for KafkaProducer {
                 value: Some(v),
             });
         }
+        headers = headers.insert(Header {
+            key: "clickhouse_database",
+            value: self.ckh_database_name.as_ref(),
+        });
         self.producer
             .0
             .send(
diff --git a/crates/storage_impl/src/config.rs b/crates/storage_impl/src/config.rs
index bb006b6a9e5..ac68d1718a9 100644
--- a/crates/storage_impl/src/config.rs
+++ b/crates/storage_impl/src/config.rs
@@ -36,9 +36,6 @@ impl DbConnectionParams for Database {
 pub trait TenantConfig: Send + Sync {
     fn get_schema(&self) -> &str;
     fn get_redis_key_prefix(&self) -> &str;
-}
-
-pub trait ClickHouseConfig: TenantConfig + Send + Sync {
     fn get_clickhouse_database(&self) -> &str;
 }