refactor(storage): Add a separate crate to represent store implementations (#1853)

crates/router/src/configs/kms.rs

@@ -44,3 +44,23 @@ impl KmsDecrypt for settings::ActiveKmsSecrets {
         Ok(self)
     }
 }
+
+#[async_trait::async_trait]
+impl KmsDecrypt for settings::Database {
+    type Output = storage_impl::config::Database;
+
+    async fn decrypt_inner(
+        mut self,
+        kms_client: &KmsClient,
+    ) -> CustomResult<Self::Output, KmsError> {
+        Ok(storage_impl::config::Database {
+            host: self.host,
+            port: self.port,
+            dbname: self.dbname,
+            username: self.username,
+            password: self.password.decrypt_inner(kms_client).await?.into(),
+            pool_size: self.pool_size,
+            connection_timeout: self.connection_timeout,
+        })
+    }
+}

crates/router/src/configs/settings.rs

@@ -432,6 +432,21 @@ pub struct Database {
     pub connection_timeout: u64,
 }
 
+#[cfg(not(feature = "kms"))]
+impl Into<storage_impl::config::Database> for Database {
+    fn into(self) -> storage_impl::config::Database {
+        storage_impl::config::Database {
+            username: self.username,
+            password: self.password,
+            host: self.host,
+            port: self.port,
+            dbname: self.dbname,
+            pool_size: self.pool_size,
+            connection_timeout: self.connection_timeout,
+        }
+    }
+}
+
 #[derive(Debug, Deserialize, Clone)]
 #[serde(default)]
 pub struct SupportedConnectors {

crates/router/src/connection.rs

@@ -84,7 +84,7 @@ pub async fn pg_connection_read(
 > {
     // If only OLAP is enabled get replica pool.
     #[cfg(all(feature = "olap", not(feature = "oltp")))]
-    let pool = &store.replica_pool;
+    let pool = &store.diesel_store.replica_pool;
 
     // If either one of these are true we need to get master pool.
     //  1. Only OLTP is enabled.
@@ -95,7 +95,7 @@ pub async fn pg_connection_read(
         all(feature = "olap", feature = "oltp"),
         all(not(feature = "olap"), not(feature = "oltp"))
     ))]
-    let pool = &store.master_pool;
+    let pool = &store.diesel_store.master_pool;
 
     pool.get()
         .await
@@ -110,7 +110,7 @@ pub async fn pg_connection_write(
     errors::StorageError,
 > {
     // Since all writes should happen to master DB only choose master DB.
-    let pool = &store.master_pool;
+    let pool = &store.diesel_store.master_pool;
 
     pool.get()
         .await

crates/router/src/services.rs

@@ -11,6 +11,7 @@ use external_services::kms::{self, decrypt::KmsDecrypt};
 #[cfg(not(feature = "kms"))]
 use masking::PeekInterface;
 use redis_interface::{errors as redis_errors, PubsubInterface, RedisValue};
+use storage_impl::diesel as diesel_impl;
 use tokio::sync::oneshot;
 
 pub use self::{api::*, encryption::*};
@@ -18,7 +19,6 @@ use crate::{
     async_spawn,
     cache::{CacheKind, ACCOUNTS_CACHE, CONFIG_CACHE},
     configs::settings,
-    connection::{diesel_make_pg_pool, PgPool},
     consts,
     core::errors,
 };
@@ -129,9 +129,10 @@ impl RedisConnInterface for Store {
 
 #[derive(Clone)]
 pub struct Store {
-    pub master_pool: PgPool,
+    #[cfg(not(feature = "olap"))]
+    pub diesel_store: diesel_impl::store::Store,
     #[cfg(feature = "olap")]
-    pub replica_pool: PgPool,
+    pub diesel_store: diesel_impl::store::ReplicaStore,
     pub redis_conn: Arc<redis_interface::RedisConnectionPool>,
     #[cfg(feature = "kv_store")]
     pub(crate) config: StoreConfig,
@@ -178,20 +179,43 @@ impl Store {
         )
         .await;
 
+        #[allow(clippy::expect_used)]
         Self {
-            master_pool: diesel_make_pg_pool(
-                &config.master_database,
-                test_transaction,
+            #[cfg(not(feature = "olap"))]
+            diesel_store: diesel_impl::store::Store::new(
+                #[cfg(not(feature = "kms"))]
+                &config.master_database.clone().into(),
                 #[cfg(feature = "kms")]
-                kms_client,
+                &config
+                    .master_database
+                    .clone()
+                    .decrypt_inner(kms_client)
+                    .await
+                    .expect("Failed to decrypt master database"),
+                test_transaction,
             )
             .await,
             #[cfg(feature = "olap")]
-            replica_pool: diesel_make_pg_pool(
-                &config.replica_database,
-                test_transaction,
+            diesel_store: diesel_impl::store::ReplicaStore::new(
+                #[cfg(not(feature = "kms"))]
+                &config.master_database.clone().into(),
                 #[cfg(feature = "kms")]
-                kms_client,
+                &config
+                    .master_database
+                    .clone()
+                    .decrypt_inner(kms_client)
+                    .await
+                    .expect("Failed to decrypt master database"),
+                #[cfg(not(feature = "kms"))]
+                &config.replica_database.clone().into(),
+                #[cfg(feature = "kms")]
+                &config
+                    .replica_database
+                    .clone()
+                    .decrypt_inner(kms_client)
+                    .await
+                    .expect("Failed to decrypt replica database"),
+                test_transaction,
             )
             .await,
             redis_conn,