Bug: [REFACTOR] Update KafkaStore to hold variables for tenant ID



Update the [`KafkaStore`](https://github.com/juspay/hyperswitch/blob/c20ecb855aa3c4b3ce1798dcc19910fb38345b46/crates/router/src/db/kafka_store.rs#L75-L79) to hold a variable for tenant-ID.

This can be achieved by simply adding a new field in the `KafkaStore` struct.
Update the KafkaStore constructor method to accept a Tenant-ID.

Tenant-ID can be a new type field backed by a string;

```rust
pub struct TenantID(String);
```

For now the KafkaStore can be initialized with a fixed tenantID `"default"`
diff --git a/crates/router/src/db.rs b/crates/router/src/db.rs
index 9f9e61a5ffc..3ac9b3cc371 100644
--- a/crates/router/src/db.rs
+++ b/crates/router/src/db.rs
@@ -19,7 +19,7 @@ pub mod file;
 pub mod fraud_check;
 pub mod gsm;
 pub mod health_check;
-mod kafka_store;
+pub mod kafka_store;
 pub mod locker_mock_up;
 pub mod mandate;
 pub mod merchant_account;
diff --git a/crates/router/src/db/kafka_store.rs b/crates/router/src/db/kafka_store.rs
index 7da40e505e8..f4e079489c3 100644
--- a/crates/router/src/db/kafka_store.rs
+++ b/crates/router/src/db/kafka_store.rs
@@ -23,6 +23,7 @@ use scheduler::{
     db::{process_tracker::ProcessTrackerInterface, queue::QueueInterface},
     SchedulerInterface,
 };
+use serde::Serialize;
 use storage_impl::redis::kv_store::RedisConnInterface;
 use time::PrimitiveDateTime;
 
@@ -72,17 +73,22 @@ use crate::{
     },
 };
 
+#[derive(Clone, Serialize)]
+pub struct TenantID(pub String);
+
 #[derive(Clone)]
 pub struct KafkaStore {
     kafka_producer: KafkaProducer,
     pub diesel_store: Store,
+    pub tenant_id: TenantID,
 }
 
 impl KafkaStore {
-    pub async fn new(store: Store, kafka_producer: KafkaProducer) -> Self {
+    pub async fn new(store: Store, kafka_producer: KafkaProducer, tenant_id: TenantID) -> Self {
         Self {
             kafka_producer,
             diesel_store: store,
+            tenant_id,
         }
     }
 }
diff --git a/crates/router/src/routes/app.rs b/crates/router/src/routes/app.rs
index 2771ee42947..c04d30836f8 100644
--- a/crates/router/src/routes/app.rs
+++ b/crates/router/src/routes/app.rs
@@ -186,6 +186,7 @@ impl AppState {
                                 .await
                                 .expect("Failed to create store"),
                             kafka_client.clone(),
+                            crate::db::kafka_store::TenantID("default".to_string()),
                         )
                         .await,
                     ),

## Type of Change
<!-- Put an `x` in the boxes that apply -->
Add TenantId field to the KafkaStore struct
#4496 

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

## Description
<!-- Describe your changes in detail -->
- Add TenantId field to the KafkaStore struct
- Add the tenant_id parameter to the constructor of Kafkastore 


### 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)?
-->
- no testable changes except CI checks

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