feat(multitenancy): add support for multitenancy and handle the same in router, producer, consumer, drainer and analytics

config/config.example.toml

@@ -629,3 +629,9 @@ disputes = "hyperswitch-dispute-events"
 
 [saved_payment_methods]
 sdk_eligible_payment_methods = "card"
+
+[multitenancy]
+enabled = false
+
+[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 

config/deployments/env_specific.toml

@@ -230,7 +230,6 @@ recon_admin_api_key = "recon_test_admin" # recon_admin API key for recon authent
 
 # Server configuration
 [server]
-base_url = "https://server_base_url"
 workers = 8
 port = 8080
 host = "127.0.0.1"
@@ -253,3 +252,9 @@ encryption_manager = "aws_kms" # Encryption manager client to be used
 [encryption_management.aws_kms]
 key_id = "kms_key_id" # The AWS key ID used by the KMS SDK for decrypting data.
 region = "kms_region" # The AWS region used by the KMS SDK for decrypting data.
+
+[multitenancy]
+enabled = false
+
+[multitenancy.tenants]
+public = { name = "hyperswitch", base_url = "http://localhost:8080", schema = "public", redis_key_prefix = ""}

config/development.toml

@@ -628,3 +628,9 @@ disputes = "hyperswitch-dispute-events"
 
 [saved_payment_methods]
 sdk_eligible_payment_methods = "card"
+
+[multitenancy]
+enabled = false
+
+[multitenancy.tenants]
+public = { name = "hyperswitch", base_url = "http://localhost:8080", schema = "public", redis_key_prefix = ""}

config/docker_compose.toml

@@ -492,3 +492,9 @@ disputes = "hyperswitch-dispute-events"
 
 [saved_payment_methods]
 sdk_eligible_payment_methods = "card"
+
+[multitenancy]
+enabled = false
+
+[multitenancy.tenants]
+public = { name = "hyperswitch", base_url = "http://localhost:8080", schema = "public", redis_key_prefix = ""}

crates/analytics/src/lib.rs

@@ -601,20 +601,20 @@ impl AnalyticsProvider {
         }
     }
 
-    pub async fn from_conf(config: &AnalyticsConfig) -> Self {
+    pub async fn from_conf(config: &AnalyticsConfig, tenant: &str) -> Self {
         match config {
-            AnalyticsConfig::Sqlx { sqlx } => Self::Sqlx(SqlxClient::from_conf(sqlx).await),
+            AnalyticsConfig::Sqlx { sqlx } => Self::Sqlx(SqlxClient::from_conf(sqlx, tenant).await),
             AnalyticsConfig::Clickhouse { clickhouse } => Self::Clickhouse(ClickhouseClient {
                 config: Arc::new(clickhouse.clone()),
             }),
             AnalyticsConfig::CombinedCkh { sqlx, clickhouse } => Self::CombinedCkh(
-                SqlxClient::from_conf(sqlx).await,
+                SqlxClient::from_conf(sqlx, tenant).await,
                 ClickhouseClient {
                     config: Arc::new(clickhouse.clone()),
                 },
             ),
             AnalyticsConfig::CombinedSqlx { sqlx, clickhouse } => Self::CombinedSqlx(
-                SqlxClient::from_conf(sqlx).await,
+                SqlxClient::from_conf(sqlx, tenant).await,
                 ClickhouseClient {
                     config: Arc::new(clickhouse.clone()),
                 },

crates/analytics/src/sqlx.rs

@@ -4,12 +4,14 @@ use api_models::{
     analytics::refunds::RefundType,
     enums::{DisputeStage, DisputeStatus},
 };
-use common_utils::errors::{CustomResult, ParsingError};
+use common_utils::{
+    errors::{CustomResult, ParsingError},
+    DbConnectionParams,
+};
 use diesel_models::enums::{
     AttemptStatus, AuthenticationType, Currency, PaymentMethod, RefundStatus,
 };
 use error_stack::ResultExt;
-use masking::PeekInterface;
 use sqlx::{
     postgres::{PgArgumentBuffer, PgPoolOptions, PgRow, PgTypeInfo, PgValueRef},
     Decode, Encode,
@@ -49,12 +51,8 @@ impl Default for SqlxClient {
 }
 
 impl SqlxClient {
-    pub async fn from_conf(conf: &Database) -> Self {
-        let password = &conf.password.peek();
-        let database_url = format!(
-            "postgres://{}:{}@{}:{}/{}",
-            conf.username, password, conf.host, conf.port, conf.dbname
-        );
+    pub async fn from_conf(conf: &Database, schema: &str) -> Self {
+        let database_url = conf.get_database_url(schema);
         #[allow(clippy::expect_used)]
         let pool = PgPoolOptions::new()
             .max_connections(conf.pool_size)

crates/common_utils/src/consts.rs

@@ -84,6 +84,9 @@ pub const DEFAULT_TTL_FOR_EXTENDED_CARD_INFO: u16 = 15 * 60;
 /// Max ttl for Extended card info in redis (in seconds)
 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";
+
 /// Max Length for MerchantReferenceId
 pub const MAX_ALLOWED_MERCHANT_REFERENCE_ID_LENGTH: u8 = 64;
 

crates/common_utils/src/lib.rs

@@ -2,6 +2,8 @@
 #![warn(missing_docs, missing_debug_implementations)]
 #![doc = include_str!(concat!(env!("CARGO_MANIFEST_DIR" ), "/", "README.md"))]
 
+use masking::{PeekInterface, Secret};
+
 use crate::{
     consts::ID_LENGTH,
     id_type::{CustomerId, MerchantReferenceId},
@@ -224,6 +226,27 @@ pub fn generate_time_ordered_id(prefix: &str) -> String {
     format!("{prefix}_{}", uuid::Uuid::now_v7().as_simple())
 }
 
+#[allow(missing_docs)]
+pub trait DbConnectionParams {
+    fn get_username(&self) -> &str;
+    fn get_password(&self) -> Secret<String>;
+    fn get_host(&self) -> &str;
+    fn get_port(&self) -> u16;
+    fn get_dbname(&self) -> &str;
+    fn get_database_url(&self, schema: &str) -> String {
+        format!(
+            "postgres://{}:{}@{}:{}/{}?application_name={}&options=-c search_path%3D{}",
+            self.get_username(),
+            self.get_password().peek(),
+            self.get_host(),
+            self.get_port(),
+            self.get_dbname(),
+            schema,
+            schema,
+        )
+    }
+}
+
 #[cfg(test)]
 mod nanoid_tests {
     #![allow(clippy::unwrap_used)]

crates/drainer/src/connection.rs

@@ -1,6 +1,6 @@
 use bb8::PooledConnection;
+use common_utils::DbConnectionParams;
 use diesel::PgConnection;
-use masking::PeekInterface;
 
 use crate::{settings::Database, Settings};
 
@@ -18,15 +18,12 @@ pub async fn redis_connection(conf: &Settings) -> redis_interface::RedisConnecti
 ///
 /// Will panic if could not create a db pool
 #[allow(clippy::expect_used)]
-pub async fn diesel_make_pg_pool(database: &Database, _test_transaction: bool) -> PgPool {
-    let database_url = format!(
-        "postgres://{}:{}@{}:{}/{}",
-        database.username,
-        database.password.peek(),
-        database.host,
-        database.port,
-        database.dbname
-    );
+pub async fn diesel_make_pg_pool(
+    database: &Database,
+    _test_transaction: bool,
+    schema: &str,
+) -> PgPool {
+    let database_url = database.get_database_url(schema);
     let manager = async_bb8_diesel::ConnectionManager::<PgConnection>::new(database_url);
     let pool = bb8::Pool::builder()
         .max_size(database.pool_size)

crates/drainer/src/handler.rs

@@ -1,4 +1,7 @@
-use std::sync::{atomic, Arc};
+use std::{
+    collections::HashMap,
+    sync::{atomic, Arc},
+};
 
 use router_env::tracing::Instrument;
 use tokio::{
@@ -31,12 +34,12 @@ pub struct HandlerInner {
     loop_interval: Duration,
     active_tasks: Arc<atomic::AtomicU64>,
     conf: DrainerSettings,
-    store: Arc<Store>,
+    stores: HashMap<String, Arc<Store>>,
     running: Arc<atomic::AtomicBool>,
 }
 
 impl Handler {
-    pub fn from_conf(conf: DrainerSettings, store: Arc<Store>) -> Self {
+    pub fn from_conf(conf: DrainerSettings, stores: HashMap<String, Arc<Store>>) -> Self {
         let shutdown_interval = Duration::from_millis(conf.shutdown_interval.into());
         let loop_interval = Duration::from_millis(conf.loop_interval.into());
 
@@ -49,7 +52,7 @@ impl Handler {
             loop_interval,
             active_tasks,
             conf,
-            store,
+            stores,
             running,
         };
 
@@ -68,21 +71,23 @@ impl Handler {
 
         while self.running.load(atomic::Ordering::SeqCst) {
             metrics::DRAINER_HEALTH.add(&metrics::CONTEXT, 1, &[]);
-            if self.store.is_stream_available(stream_index).await {
-                let _task_handle = tokio::spawn(
-                    drainer_handler(
-                        self.store.clone(),
-                        stream_index,
-                        self.conf.max_read_count,
-                        self.active_tasks.clone(),
-                        jobs_picked.clone(),
-                    )
-                    .in_current_span(),
-                );
+            for store in self.stores.values() {
+                if store.is_stream_available(stream_index).await {
+                    let _task_handle = tokio::spawn(
+                        drainer_handler(
+                            store.clone(),
+                            stream_index,
+                            self.conf.max_read_count,
+                            self.active_tasks.clone(),
+                            jobs_picked.clone(),
+                        )
+                        .in_current_span(),
+                    );
+                }
             }
             stream_index = utils::increment_stream_index(
                 (stream_index, jobs_picked.clone()),
-                self.store.config.drainer_num_partitions,
+                self.conf.num_partitions,
             )
             .await;
             time::sleep(self.loop_interval).await;
@@ -116,18 +121,33 @@ impl Handler {
 
     pub fn spawn_error_handlers(&self, tx: mpsc::Sender<()>) -> errors::DrainerResult<()> {
         let (redis_error_tx, redis_error_rx) = oneshot::channel();
+        let redis_conn_clone = self
+            .stores
+            .values()
+            .next()
+            .map(|store| store.redis_conn.clone());
+        match redis_conn_clone {
+            None => {
+                logger::error!("No redis connection found");
+                Err(
+                    errors::DrainerError::UnexpectedError("No redis connection found".to_string())
+                        .into(),
+                )
+            }
+            Some(redis_conn_clone) => {
+                // Spawn a task to monitor if redis is down or not
+                let _task_handle = tokio::spawn(
+                    async move { redis_conn_clone.on_error(redis_error_tx).await }
+                        .in_current_span(),
+                );
 
-        let redis_conn_clone = self.store.redis_conn.clone();
-
-        // Spawn a task to monitor if redis is down or not
-        let _task_handle = tokio::spawn(
-            async move { redis_conn_clone.on_error(redis_error_tx).await }.in_current_span(),
-        );
-
-        //Spawns a task to send shutdown signal if redis goes down
-        let _task_handle = tokio::spawn(redis_error_receiver(redis_error_rx, tx).in_current_span());
+                //Spawns a task to send shutdown signal if redis goes down
+                let _task_handle =
+                    tokio::spawn(redis_error_receiver(redis_error_rx, tx).in_current_span());
 
-        Ok(())
+                Ok(())
+            }
+        }
     }
 }
 
@@ -208,7 +228,10 @@ async fn drainer(
     };
 
     // parse_stream_entries returns error if no entries is found, handle it
-    let entries = utils::parse_stream_entries(&stream_read, stream_name)?;
+    let entries = utils::parse_stream_entries(
+        &stream_read,
+        store.redis_conn.add_prefix(stream_name).as_str(),
+    )?;
     let read_count = entries.len();
 
     metrics::JOBS_PICKED_PER_STREAM.add(