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



Update the [AnalyticsProvider](https://github.com/juspay/hyperswitch/blob/be44447c09ea8814dc8b88aa971e08cc749db5f3/crates/analytics/src/lib.rs#L67-L72) (or [`AnalyticsConfig`](https://github.com/juspay/hyperswitch/blob/be44447c09ea8814dc8b88aa971e08cc749db5f3/crates/analytics/src/lib.rs#L593-L608)) to hold a variable for tenant-ID.

This can be achieved by simply adding a new field in the underlying structs for SqlxClient & ClickhouseClient.
Update the AnalyticsPool constructor method to accept a Tenant-ID.

Tenant-ID can be a new type field backed by a string;

pub struct TenantID(String);

For now the AnalyticsPool can be initialized with a fixed tenantID "default"

Pull request description:
## 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 -->
Add Tenant_id as a field for all logs
Add Tenant_id action for Kafka messages as well , headers to be a combination of ( tenant_id, merchant_id, payment_id )
Make logical databases and physical tables in clickhouse for different tenants

### 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).
-->
To establish tenancy in our application and start using the default tenant as 'Hyperswitch' and also offer a 'test' tenant in all environments

https://github.com/juspay/hyperswitch-product/issues/50

## 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)?
-->
Since there is no new changes involved, sanity testing is enough, Check if the payments are going fine and able to see analytics in dashboard


## 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:

Since there is no new changes involved, sanity testing is enough, Check if the payments are going fine and able to see analytics in dashboard

Patch:
diff --git a/config/config.example.toml b/config/config.example.toml
index 81516ee8b6b..30f33430378 100644
--- a/config/config.example.toml
+++ b/config/config.example.toml
@@ -639,6 +639,7 @@ sdk_eligible_payment_methods = "card"
 
 [multitenancy]
 enabled = false
+global_tenant = { schema = "public", redis_key_prefix = "" }
 
 [multitenancy.tenants]
-public = { name = "hyperswitch", base_url = "http://localhost:8080", schema = "public", redis_key_prefix = ""} # schema -> Postgres db schema, redis_key_prefix -> redis key distinguisher, base_url -> url of the tenant 
\ No newline at end of file
+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 
\ No newline at end of file
diff --git a/config/deployments/env_specific.toml b/config/deployments/env_specific.toml
index 9ab790b8ee8..162444c9098 100644
--- a/config/deployments/env_specific.toml
+++ b/config/deployments/env_specific.toml
@@ -256,6 +256,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 = "" }
 
 [multitenancy.tenants]
-public = { name = "hyperswitch", base_url = "http://localhost:8080", schema = "public", redis_key_prefix = ""}
\ No newline at end of file
+public = { name = "hyperswitch", base_url = "http://localhost:8080", schema = "public", redis_key_prefix = "", clickhouse_database = "default"}
\ No newline at end of file
diff --git a/config/development.toml b/config/development.toml
index 9a40a99cfce..b986a66d845 100644
--- a/config/development.toml
+++ b/config/development.toml
@@ -648,6 +648,7 @@ sdk_eligible_payment_methods = "card"
 
 [multitenancy]
 enabled = false
+global_tenant = { schema = "public", redis_key_prefix = "" }
 
 [multitenancy.tenants]
-public = { name = "hyperswitch", base_url = "http://localhost:8080", schema = "public", redis_key_prefix = ""}
+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 2cd93eade01..a9d37995728 100644
--- a/config/docker_compose.toml
+++ b/config/docker_compose.toml
@@ -502,6 +502,7 @@ sdk_eligible_payment_methods = "card"
 
 [multitenancy]
 enabled = false
+global_tenant = { schema = "public", redis_key_prefix = "" }
 
 [multitenancy.tenants]
-public = { name = "hyperswitch", base_url = "http://localhost:8080", schema = "public", redis_key_prefix = ""}
\ No newline at end of file
+public = { name = "hyperswitch", base_url = "http://localhost:8080", schema = "public", redis_key_prefix = "", clickhouse_database = "default"}
\ No newline at end of file
diff --git a/crates/analytics/src/clickhouse.rs b/crates/analytics/src/clickhouse.rs
index 64064c09c88..fd1746c4b9e 100644
--- a/crates/analytics/src/clickhouse.rs
+++ b/crates/analytics/src/clickhouse.rs
@@ -35,6 +35,7 @@ pub type ClickhouseResult<T> = error_stack::Result<T, ClickhouseError>;
 #[derive(Clone, Debug)]
 pub struct ClickhouseClient {
     pub config: Arc<ClickhouseConfig>,
+    pub database: String,
 }
 
 #[derive(Clone, Debug, serde::Deserialize)]
@@ -42,7 +43,6 @@ pub struct ClickhouseConfig {
     username: String,
     password: Option<String>,
     host: String,
-    database_name: String,
 }
 
 impl Default for ClickhouseConfig {
@@ -51,7 +51,6 @@ impl Default for ClickhouseConfig {
             username: "default".to_string(),
             password: None,
             host: "http://localhost:8123".to_string(),
-            database_name: "default".to_string(),
         }
     }
 }
@@ -63,7 +62,7 @@ impl ClickhouseClient {
         let params = CkhQuery {
             date_time_output_format: String::from("iso"),
             output_format_json_quote_64bit_integers: 0,
-            database: self.config.database_name.clone(),
+            database: self.database.clone(),
         };
         let response = client
             .post(&self.config.host)
diff --git a/crates/analytics/src/lib.rs b/crates/analytics/src/lib.rs
index f658f679097..5c269a4bb18 100644
--- a/crates/analytics/src/lib.rs
+++ b/crates/analytics/src/lib.rs
@@ -601,22 +601,30 @@ impl AnalyticsProvider {
         }
     }
 
-    pub async fn from_conf(config: &AnalyticsConfig, tenant: &str) -> Self {
+    pub async fn from_conf(
+        config: &AnalyticsConfig,
+        tenant: &dyn storage_impl::config::ClickHouseConfig,
+    ) -> Self {
         match config {
-            AnalyticsConfig::Sqlx { sqlx } => Self::Sqlx(SqlxClient::from_conf(sqlx, tenant).await),
+            AnalyticsConfig::Sqlx { sqlx } => {
+                Self::Sqlx(SqlxClient::from_conf(sqlx, tenant.get_schema()).await)
+            }
             AnalyticsConfig::Clickhouse { clickhouse } => Self::Clickhouse(ClickhouseClient {
                 config: Arc::new(clickhouse.clone()),
+                database: tenant.get_clickhouse_database().to_string(),
             }),
             AnalyticsConfig::CombinedCkh { sqlx, clickhouse } => Self::CombinedCkh(
-                SqlxClient::from_conf(sqlx, tenant).await,
+                SqlxClient::from_conf(sqlx, tenant.get_schema()).await,
                 ClickhouseClient {
                     config: Arc::new(clickhouse.clone()),
+                    database: tenant.get_clickhouse_database().to_string(),
                 },
             ),
             AnalyticsConfig::CombinedSqlx { sqlx, clickhouse } => Self::CombinedSqlx(
-                SqlxClient::from_conf(sqlx, tenant).await,
+                SqlxClient::from_conf(sqlx, tenant.get_schema()).await,
                 ClickhouseClient {
                     config: Arc::new(clickhouse.clone()),
+                    database: tenant.get_clickhouse_database().to_string(),
                 },
             ),
         }
diff --git a/crates/common_utils/src/consts.rs b/crates/common_utils/src/consts.rs
index 968ef22bf97..38c8997358d 100644
--- a/crates/common_utils/src/consts.rs
+++ b/crates/common_utils/src/consts.rs
@@ -87,8 +87,8 @@ pub const MAX_TTL_FOR_EXTENDED_CARD_INFO: u16 = 60 * 60 * 2;
 /// Default tenant to be used when multitenancy is disabled
 pub const DEFAULT_TENANT: &str = "public";
 
-/// Global tenant to be used when multitenancy is enabled
-pub const GLOBAL_TENANT: &str = "global";
+/// Default tenant to be used when multitenancy is disabled
+pub const TENANT_HEADER: &str = "x-tenant-id";
 
 /// Max Length for MerchantReferenceId
 pub const MAX_ALLOWED_MERCHANT_REFERENCE_ID_LENGTH: u8 = 64;
diff --git a/crates/diesel_models/src/query/user.rs b/crates/diesel_models/src/query/user.rs
index dac515cb279..2bd403a847b 100644
--- a/crates/diesel_models/src/query/user.rs
+++ b/crates/diesel_models/src/query/user.rs
@@ -1,23 +1,9 @@
-use async_bb8_diesel::AsyncRunQueryDsl;
 use common_utils::pii;
-use diesel::{
-    associations::HasTable, debug_query, result::Error as DieselError, ExpressionMethods,
-    JoinOnDsl, QueryDsl,
-};
-use error_stack::report;
-use router_env::logger;
+use diesel::{associations::HasTable, ExpressionMethods};
 pub mod sample_data;
 
 use crate::{
-    errors::{self},
-    query::generics,
-    schema::{
-        user_roles::{self, dsl as user_roles_dsl},
-        users::dsl as users_dsl,
-    },
-    user::*,
-    user_role::UserRole,
-    PgPooledConn, StorageResult,
+    query::generics, schema::users::dsl as users_dsl, user::*, PgPooledConn, StorageResult,
 };
 
 impl UserNew {
@@ -90,27 +76,6 @@ impl User {
         .await
     }
 
-    pub async fn find_joined_users_and_roles_by_merchant_id(
-        conn: &PgPooledConn,
-        mid: &str,
-    ) -> StorageResult<Vec<(Self, UserRole)>> {
-        let query = Self::table()
-            .inner_join(user_roles::table.on(user_roles_dsl::user_id.eq(users_dsl::user_id)))
-            .filter(user_roles_dsl::merchant_id.eq(mid.to_owned()));
-
-        logger::debug!(query = %debug_query::<diesel::pg::Pg,_>(&query).to_string());
-
-        query
-            .get_results_async::<(Self, UserRole)>(conn)
-            .await
-            .map_err(|err| match err {
-                DieselError::NotFound => {
-                    report!(err).change_context(errors::DatabaseError::NotFound)
-                }
-                _ => report!(err).change_context(errors::DatabaseError::Others),
-            })
-    }
-
     pub async fn find_users_by_user_ids(
         conn: &PgPooledConn,
         user_ids: Vec<String>,
diff --git a/crates/drainer/src/settings.rs b/crates/drainer/src/settings.rs
index e64b5dbd4ad..57c17f577b1 100644
--- a/crates/drainer/src/settings.rs
+++ b/crates/drainer/src/settings.rs
@@ -143,6 +143,7 @@ pub struct Tenant {
     pub base_url: String,
     pub schema: String,
     pub redis_key_prefix: String,
+    pub clickhouse_database: String,
 }
 
 #[derive(Debug, Deserialize, Clone)]
diff --git a/crates/router/src/configs/settings.rs b/crates/router/src/configs/settings.rs
index e99e216463a..f3f70cf42bd 100644
--- a/crates/router/src/configs/settings.rs
+++ b/crates/router/src/configs/settings.rs
@@ -128,6 +128,7 @@ pub struct Settings<S: SecretState> {
 pub struct Multitenancy {
     pub tenants: TenantConfig,
     pub enabled: bool,
+    pub global_tenant: GlobalTenant,
 }
 
 impl Multitenancy {
@@ -152,6 +153,7 @@ pub struct Tenant {
     pub base_url: String,
     pub schema: String,
     pub redis_key_prefix: String,
+    pub clickhouse_database: String,
 }
 
 impl storage_impl::config::TenantConfig for Tenant {
@@ -163,6 +165,12 @@ impl storage_impl::config::TenantConfig for Tenant {
     }
 }
 
+impl storage_impl::config::ClickHouseConfig for Tenant {
+    fn get_clickhouse_database(&self) -> &str {
+        self.clickhouse_database.as_str()
+    }
+}
+
 #[derive(Debug, Deserialize, Clone, Default)]
 pub struct GlobalTenant {
     pub schema: String,
diff --git a/crates/router/src/middleware.rs b/crates/router/src/middleware.rs
index e1601f8bbbe..7b9919d3afa 100644
--- a/crates/router/src/middleware.rs
+++ b/crates/router/src/middleware.rs
@@ -1,3 +1,4 @@
+use common_utils::consts::TENANT_HEADER;
 use futures::StreamExt;
 use router_env::{
     logger,
@@ -140,10 +141,17 @@ where
     // TODO: have a common source of truth for the list of top level fields
     // /crates/router_env/src/logger/storage.rs also has a list of fields  called PERSISTENT_KEYS
     fn call(&self, req: actix_web::dev::ServiceRequest) -> Self::Future {
+        let tenant_id = req
+            .headers()
+            .get(TENANT_HEADER)
+            .and_then(|i| i.to_str().ok())
+            .map(|s| s.to_owned());
         let response_fut = self.service.call(req);
-
         Box::pin(
             async move {
+                if let Some(tenant_id) = tenant_id {
+                    router_env::tracing::Span::current().record("tenant_id", &tenant_id);
+                }
                 let response = response_fut.await;
                 router_env::tracing::Span::current().record("golden_log_line", true);
                 response
diff --git a/crates/router/src/routes/app.rs b/crates/router/src/routes/app.rs
index 92ca3c264fe..2e7a6c4f61e 100644
--- a/crates/router/src/routes/app.rs
+++ b/crates/router/src/routes/app.rs
@@ -5,7 +5,6 @@ use actix_web::{web, Scope};
 use api_models::routing::RoutingRetrieveQuery;
 #[cfg(feature = "olap")]
 use common_enums::TransactionType;
-use common_utils::consts::{DEFAULT_TENANT, GLOBAL_TENANT};
 #[cfg(feature = "email")]
 use external_services::email::{ses::AwsSes, EmailService};
 use external_services::file_storage::FileStorageInterface;
@@ -257,19 +256,11 @@ impl AppState {
             let cache_store = get_cache_store(&conf.clone(), shut_down_signal, testable)
                 .await
                 .expect("Failed to create store");
-            let global_tenant = if conf.multitenancy.enabled {
-                GLOBAL_TENANT
-            } else {
-                DEFAULT_TENANT
-            };
             let global_store: Box<dyn GlobalStorageInterface> = Self::get_store_interface(
                 &storage_impl,
                 &event_handler,
                 &conf,
-                &settings::GlobalTenant {
-                    schema: global_tenant.to_string(),
-                    redis_key_prefix: String::default(),
-                },
+                &conf.multitenancy.global_tenant,
                 Arc::clone(&cache_store),
                 testable,
             )
@@ -288,9 +279,7 @@ impl AppState {
                 .get_storage_interface();
                 stores.insert(tenant_name.clone(), store);
                 #[cfg(feature = "olap")]
-                let pool =
-                    AnalyticsProvider::from_conf(conf.analytics.get_inner(), tenant_name.as_str())
-                        .await;
+                let pool = AnalyticsProvider::from_conf(conf.analytics.get_inner(), tenant).await;
                 #[cfg(feature = "olap")]
                 pools.insert(tenant_name.clone(), pool);
             }
diff --git a/crates/router/src/services/api.rs b/crates/router/src/services/api.rs
index baa824ced3f..7408f379489 100644
--- a/crates/router/src/services/api.rs
+++ b/crates/router/src/services/api.rs
@@ -19,7 +19,7 @@ use api_models::enums::{CaptureMethod, PaymentMethodType};
 pub use client::{proxy_bypass_urls, ApiClient, MockApiClient, ProxyClient};
 pub use common_utils::request::{ContentType, Method, Request, RequestBuilder};
 use common_utils::{
-    consts::X_HS_LATENCY,
+    consts::{DEFAULT_TENANT, TENANT_HEADER, X_HS_LATENCY},
     errors::{ErrorSwitch, ReportSwitchExt},
     request::RequestContent,
 };
@@ -941,10 +941,10 @@ where
         .into_iter()
         .collect();
     let tenant_id = if !state.conf.multitenancy.enabled {
-        common_utils::consts::DEFAULT_TENANT.to_string()
+        DEFAULT_TENANT.to_string()
     } else {
         incoming_request_header
-            .get("x-tenant-id")
+            .get(TENANT_HEADER)
             .and_then(|value| value.to_str().ok())
             .ok_or_else(|| errors::ApiErrorResponse::MissingTenantId.switch())
             .map(|req_tenant_id| {
diff --git a/crates/storage_impl/src/config.rs b/crates/storage_impl/src/config.rs
index e371936a04d..bb006b6a9e5 100644
--- a/crates/storage_impl/src/config.rs
+++ b/crates/storage_impl/src/config.rs
@@ -38,6 +38,10 @@ pub trait TenantConfig: Send + Sync {
     fn get_redis_key_prefix(&self) -> &str;
 }
 
+pub trait ClickHouseConfig: TenantConfig + Send + Sync {
+    fn get_clickhouse_database(&self) -> &str;
+}
+
 #[derive(Debug, serde::Deserialize, Clone, Copy, Default)]
 #[serde(rename_all = "PascalCase")]
 pub enum QueueStrategy {
diff --git a/loadtest/config/development.toml b/loadtest/config/development.toml
index d008ff3231c..ad97ef6d404 100644
--- a/loadtest/config/development.toml
+++ b/loadtest/config/development.toml
@@ -328,6 +328,7 @@ keys = "user-agent"
 
 [multitenancy]
 enabled = false
+global_tenant = { schema = "public", redis_key_prefix = "" }
 
 [multitenancy.tenants]
-public = { name = "hyperswitch", base_url = "http://localhost:8080", schema = "public", redis_key_prefix = ""}
\ No newline at end of file
+public = { name = "hyperswitch", base_url = "http://localhost:8080", schema = "public", redis_key_prefix = "", clickhouse_database = "default"}
\ No newline at end of file