Issue:
Bug: [REFACTOR] Incorporate `hyperswitch_interface` into drainer



Refactor drainer to incorporate the newly added crate `hyperswitch_interface` for secrets decryption 

Pull request description:
## Type of Change
<!-- Put an `x` in the boxes that apply -->

- [ ] Bugfix
- [ ] New feature
- [ ] Enhancement
- [x] Refactoring
- [ ] Dependency updates
- [ ] Documentation
- [ ] CI/CD

## Description
<!-- Describe your changes in detail -->
Added `AppState` in drainer which now contains `encryption_client` for runtime encryption and decryption of values. This PR also incorporates the newly added `hyperswitch_interface` crate for decrypting the secrets during application startup.

### Additional Changes

- [ ] This PR modifies the API contract
- [ ] This PR modifies the database schema
- [x] 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)?
-->
This PR refactors the way decryption of secrets work in drainer. So this cannot be tested

## Checklist
<!-- Put an `x` in the boxes that apply -->

- [x] I formatted the code `cargo +nightly fmt --all`
- [x] I addressed lints thrown by `cargo clippy`
- [x] I reviewed the submitted code
- [ ] I added unit tests for my changes where possible
- [ ] I added a [CHANGELOG](/CHANGELOG.md) entry if applicable


Testing:

This PR refactors the way decryption of secrets work in drainer. So this cannot be tested

Patch:
diff --git a/Cargo.lock b/Cargo.lock
index 4bee1804f4c..dcde5456d7d 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2342,6 +2342,7 @@ dependencies = [
  "diesel_models",
  "error-stack",
  "external_services",
+ "hyperswitch_interfaces",
  "masking",
  "mime",
  "once_cell",
diff --git a/config/deployments/drainer.toml b/config/deployments/drainer.toml
index 42c89cbfd58..f8bf2826768 100644
--- a/config/deployments/drainer.toml
+++ b/config/deployments/drainer.toml
@@ -5,7 +5,17 @@ num_partitions = 64
 shutdown_interval = 1000
 stream_name = "drainer_stream"
 
-[kms]
+[secrets_management]
+secrets_manager = "aws_kms"
+
+[secrets_management.aws_kms]
+key_id = "kms_key_id"
+region = "kms_region"
+
+[encryption_management]
+encryption_manager = "aws_kms"
+
+[encryption_management.aws_kms]
 key_id = "kms_key_id"
 region = "kms_region"
 
diff --git a/crates/drainer/Cargo.toml b/crates/drainer/Cargo.toml
index a8972e279a6..f45c1979827 100644
--- a/crates/drainer/Cargo.toml
+++ b/crates/drainer/Cargo.toml
@@ -35,6 +35,7 @@ async-trait = "0.1.74"
 common_utils = { version = "0.1.0", path = "../common_utils", features = ["signals"] }
 diesel_models = { version = "0.1.0", path = "../diesel_models", features = ["kv_store"] }
 external_services = { version = "0.1.0", path = "../external_services" }
+hyperswitch_interfaces = { version = "0.1.0", path = "../hyperswitch_interfaces" }
 masking = { version = "0.1.0", path = "../masking" }
 redis_interface = { version = "0.1.0", path = "../redis_interface" }
 router_env = { version = "0.1.0", path = "../router_env", features = ["log_extra_implicit_fields", "log_custom_entries_to_extra"] }
diff --git a/crates/drainer/src/connection.rs b/crates/drainer/src/connection.rs
index e01b72150c5..c539e9a734c 100644
--- a/crates/drainer/src/connection.rs
+++ b/crates/drainer/src/connection.rs
@@ -1,23 +1,13 @@
 use bb8::PooledConnection;
 use diesel::PgConnection;
-#[cfg(feature = "aws_kms")]
-use external_services::aws_kms::{self, decrypt::AwsKmsDecrypt};
-#[cfg(feature = "hashicorp-vault")]
-use external_services::hashicorp_vault::{
-    core::{HashiCorpVault, Kv2},
-    decrypt::VaultFetch,
-};
-#[cfg(not(feature = "aws_kms"))]
 use masking::PeekInterface;
 
-use crate::settings::Database;
+use crate::{settings::Database, Settings};
 
 pub type PgPool = bb8::Pool<async_bb8_diesel::ConnectionManager<PgConnection>>;
 
 #[allow(clippy::expect_used)]
-pub async fn redis_connection(
-    conf: &crate::settings::Settings,
-) -> redis_interface::RedisConnectionPool {
+pub async fn redis_connection(conf: &Settings) -> redis_interface::RedisConnectionPool {
     redis_interface::RedisConnectionPool::new(&conf.redis)
         .await
         .expect("Failed to create Redis connection Pool")
@@ -28,31 +18,14 @@ pub async fn redis_connection(
 ///
 /// 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,
-    #[cfg(feature = "aws_kms")] aws_kms_client: &'static aws_kms::core::AwsKmsClient,
-    #[cfg(feature = "hashicorp-vault")] hashicorp_client: &'static HashiCorpVault,
-) -> PgPool {
-    let password = database.password.clone();
-    #[cfg(feature = "hashicorp-vault")]
-    let password = password
-        .fetch_inner::<Kv2>(hashicorp_client)
-        .await
-        .expect("Failed while fetching db password");
-
-    #[cfg(feature = "aws_kms")]
-    let password = password
-        .decrypt_inner(aws_kms_client)
-        .await
-        .expect("Failed to decrypt password");
-
-    #[cfg(not(feature = "aws_kms"))]
-    let password = &password.peek();
-
+pub async fn diesel_make_pg_pool(database: &Database, _test_transaction: bool) -> PgPool {
     let database_url = format!(
         "postgres://{}:{}@{}:{}/{}",
-        database.username, password, database.host, database.port, database.dbname
+        database.username,
+        database.password.peek(),
+        database.host,
+        database.port,
+        database.dbname
     );
     let manager = async_bb8_diesel::ConnectionManager::<PgConnection>::new(database_url);
     let pool = bb8::Pool::builder()
diff --git a/crates/drainer/src/health_check.rs b/crates/drainer/src/health_check.rs
index 33b4a1395a8..443cb5b6f3f 100644
--- a/crates/drainer/src/health_check.rs
+++ b/crates/drainer/src/health_check.rs
@@ -11,7 +11,7 @@ use crate::{
     connection::{pg_connection, redis_connection},
     errors::HealthCheckError,
     services::{self, Store},
-    settings::Settings,
+    Settings,
 };
 
 pub const TEST_STREAM_NAME: &str = "TEST_STREAM_0";
diff --git a/crates/drainer/src/lib.rs b/crates/drainer/src/lib.rs
index 909ae065e26..0ed6183faef 100644
--- a/crates/drainer/src/lib.rs
+++ b/crates/drainer/src/lib.rs
@@ -11,19 +11,20 @@ mod stream;
 mod types;
 mod utils;
 use std::sync::Arc;
+mod secrets_transformers;
 
 use actix_web::dev::Server;
 use common_utils::signals::get_allowed_signals;
 use diesel_models::kv;
 use error_stack::{IntoReport, ResultExt};
+use hyperswitch_interfaces::secrets_interface::secret_state::RawSecret;
 use router_env::{instrument, tracing};
 use tokio::sync::mpsc;
 
+pub(crate) type Settings = crate::settings::Settings<RawSecret>;
+
 use crate::{
-    connection::pg_connection,
-    services::Store,
-    settings::{DrainerSettings, Settings},
-    types::StreamData,
+    connection::pg_connection, services::Store, settings::DrainerSettings, types::StreamData,
 };
 
 pub async fn start_drainer(store: Arc<Store>, conf: DrainerSettings) -> errors::DrainerResult<()> {
diff --git a/crates/drainer/src/main.rs b/crates/drainer/src/main.rs
index 943a66f6791..377fdcd1569 100644
--- a/crates/drainer/src/main.rs
+++ b/crates/drainer/src/main.rs
@@ -15,7 +15,9 @@ async fn main() -> DrainerResult<()> {
     conf.validate()
         .expect("Failed to validate drainer configuration");
 
-    let store = services::Store::new(&conf, false).await;
+    let state = settings::AppState::new(conf.clone()).await;
+
+    let store = services::Store::new(&state.conf, false).await;
     let store = std::sync::Arc::new(store);
 
     #[cfg(feature = "vergen")]
@@ -28,7 +30,7 @@ async fn main() -> DrainerResult<()> {
     );
 
     #[allow(clippy::expect_used)]
-    let web_server = Box::pin(start_web_server(conf.clone(), store.clone()))
+    let web_server = Box::pin(start_web_server(state.conf.as_ref().clone(), store.clone()))
         .await
         .expect("Failed to create the server");
 
diff --git a/crates/drainer/src/secrets_transformers.rs b/crates/drainer/src/secrets_transformers.rs
new file mode 100644
index 00000000000..c0447725a79
--- /dev/null
+++ b/crates/drainer/src/secrets_transformers.rs
@@ -0,0 +1,50 @@
+use common_utils::errors::CustomResult;
+use hyperswitch_interfaces::secrets_interface::{
+    secret_handler::SecretsHandler,
+    secret_state::{RawSecret, SecretStateContainer, SecuredSecret},
+    SecretManagementInterface, SecretsManagementError,
+};
+
+use crate::settings::{Database, Settings};
+
+#[async_trait::async_trait]
+impl SecretsHandler for Database {
+    async fn convert_to_raw_secret(
+        value: SecretStateContainer<Self, SecuredSecret>,
+        secret_management_client: Box<dyn SecretManagementInterface>,
+    ) -> CustomResult<SecretStateContainer<Self, RawSecret>, SecretsManagementError> {
+        let secured_db_config = value.get_inner();
+        let raw_db_password = secret_management_client
+            .get_secret(secured_db_config.password.clone())
+            .await?;
+
+        Ok(value.transition_state(|db| Self {
+            password: raw_db_password,
+            ..db
+        }))
+    }
+}
+
+/// # Panics
+///
+/// Will panic even if fetching raw secret fails for at least one config value
+#[allow(clippy::unwrap_used)]
+pub async fn fetch_raw_secrets(
+    conf: Settings<SecuredSecret>,
+    secret_management_client: Box<dyn SecretManagementInterface>,
+) -> Settings<RawSecret> {
+    #[allow(clippy::expect_used)]
+    let database = Database::convert_to_raw_secret(conf.master_database, secret_management_client)
+        .await
+        .expect("Failed to decrypt database password");
+
+    Settings {
+        server: conf.server,
+        master_database: database,
+        redis: conf.redis,
+        log: conf.log,
+        drainer: conf.drainer,
+        encryption_management: conf.encryption_management,
+        secrets_management: conf.secrets_management,
+    }
+}
diff --git a/crates/drainer/src/services.rs b/crates/drainer/src/services.rs
index 3918c756c10..9e0165548a0 100644
--- a/crates/drainer/src/services.rs
+++ b/crates/drainer/src/services.rs
@@ -28,20 +28,10 @@ impl Store {
     /// Panics if there is a failure while obtaining the HashiCorp client using the provided configuration.
     /// This panic indicates a critical failure in setting up external services, and the application cannot proceed without a valid HashiCorp client.
     ///
-    pub async fn new(config: &crate::settings::Settings, test_transaction: bool) -> Self {
+    pub async fn new(config: &crate::Settings, test_transaction: bool) -> Self {
         Self {
-            master_pool: diesel_make_pg_pool(
-                &config.master_database,
-                test_transaction,
-                #[cfg(feature = "aws_kms")]
-                external_services::aws_kms::core::get_aws_kms_client(&config.kms).await,
-                #[cfg(feature = "hashicorp-vault")]
-                #[allow(clippy::expect_used)]
-                external_services::hashicorp_vault::core::get_hashicorp_client(&config.hc_vault)
-                    .await
-                    .expect("Failed while getting hashicorp client"),
-            )
-            .await,
+            master_pool: diesel_make_pg_pool(config.master_database.get_inner(), test_transaction)
+                .await,
             redis_conn: Arc::new(crate::connection::redis_connection(config).await),
             config: StoreConfig {
                 drainer_stream_name: config.drainer.stream_name.clone(),
diff --git a/crates/drainer/src/settings.rs b/crates/drainer/src/settings.rs
index de5af654540..ec24c472af2 100644
--- a/crates/drainer/src/settings.rs
+++ b/crates/drainer/src/settings.rs
@@ -1,22 +1,23 @@
-use std::path::PathBuf;
+use std::{path::PathBuf, sync::Arc};
 
 use common_utils::ext_traits::ConfigExt;
 use config::{Environment, File};
-#[cfg(feature = "aws_kms")]
-use external_services::aws_kms;
-#[cfg(feature = "hashicorp-vault")]
-use external_services::hashicorp_vault;
+use external_services::managers::{
+    encryption_management::EncryptionManagementConfig, secrets_management::SecretsManagementConfig,
+};
+use hyperswitch_interfaces::{
+    encryption_interface::EncryptionManagementInterface,
+    secrets_interface::secret_state::{
+        RawSecret, SecretState, SecretStateContainer, SecuredSecret,
+    },
+};
+use masking::Secret;
 use redis_interface as redis;
 pub use router_env::config::{Log, LogConsole, LogFile, LogTelemetry};
 use router_env::{env, logger};
 use serde::Deserialize;
 
-use crate::errors;
-
-#[cfg(feature = "aws_kms")]
-pub type Password = aws_kms::core::AwsKmsValue;
-#[cfg(not(feature = "aws_kms"))]
-pub type Password = masking::Secret<String>;
+use crate::{errors, secrets_transformers};
 
 #[derive(clap::Parser, Default)]
 #[cfg_attr(feature = "vergen", command(version = router_env::version!()))]
@@ -27,25 +28,58 @@ pub struct CmdLineConf {
     pub config_path: Option<PathBuf>,
 }
 
+#[derive(Clone)]
+pub struct AppState {
+    pub conf: Arc<Settings<RawSecret>>,
+    pub encryption_client: Box<dyn EncryptionManagementInterface>,
+}
+
+impl AppState {
+    /// # Panics
+    ///
+    /// Panics if secret or encryption management client cannot be initiated
+    pub async fn new(conf: Settings<SecuredSecret>) -> Self {
+        #[allow(clippy::expect_used)]
+        let secret_management_client = conf
+            .secrets_management
+            .get_secret_management_client()
+            .await
+            .expect("Failed to create secret management client");
+
+        let raw_conf =
+            secrets_transformers::fetch_raw_secrets(conf, secret_management_client).await;
+
+        #[allow(clippy::expect_used)]
+        let encryption_client = raw_conf
+            .encryption_management
+            .get_encryption_management_client()
+            .await
+            .expect("Failed to create encryption management client");
+
+        Self {
+            conf: Arc::new(raw_conf),
+            encryption_client,
+        }
+    }
+}
+
 #[derive(Debug, Deserialize, Clone, Default)]
 #[serde(default)]
-pub struct Settings {
+pub struct Settings<S: SecretState> {
     pub server: Server,
-    pub master_database: Database,
+    pub master_database: SecretStateContainer<Database, S>,
     pub redis: redis::RedisSettings,
     pub log: Log,
     pub drainer: DrainerSettings,
-    #[cfg(feature = "aws_kms")]
-    pub kms: aws_kms::core::AwsKmsConfig,
-    #[cfg(feature = "hashicorp-vault")]
-    pub hc_vault: hashicorp_vault::core::HashiCorpVaultConfig,
+    pub encryption_management: EncryptionManagementConfig,
+    pub secrets_management: SecretsManagementConfig,
 }
 
 #[derive(Debug, Deserialize, Clone)]
 #[serde(default)]
 pub struct Database {
     pub username: String,
-    pub password: Password,
+    pub password: Secret<String>,
     pub host: String,
     pub port: u16,
     pub dbname: String,
@@ -85,7 +119,7 @@ impl Default for Database {
     fn default() -> Self {
         Self {
             username: String::new(),
-            password: Password::default(),
+            password: String::new().into(),
             host: "localhost".into(),
             port: 5432,
             dbname: String::new(),
@@ -157,7 +191,7 @@ impl DrainerSettings {
     }
 }
 
-impl Settings {
+impl Settings<SecuredSecret> {
     pub fn new() -> Result<Self, errors::DrainerError> {
         Self::with_config_path(None)
     }
@@ -199,12 +233,25 @@ impl Settings {
 
     pub fn validate(&self) -> Result<(), errors::DrainerError> {
         self.server.validate()?;
-        self.master_database.validate()?;
+        self.master_database.get_inner().validate()?;
         self.redis.validate().map_err(|error| {
             println!("{error}");
             errors::DrainerError::ConfigParsingError("invalid Redis configuration".into())
         })?;
         self.drainer.validate()?;
+        self.secrets_management.validate().map_err(|error| {
+            println!("{error}");
+            errors::DrainerError::ConfigParsingError(
+                "invalid secrets management configuration".into(),
+            )
+        })?;
+
+        self.encryption_management.validate().map_err(|error| {
+            println!("{error}");
+            errors::DrainerError::ConfigParsingError(
+                "invalid encryption management configuration".into(),
+            )
+        })?;
 
         Ok(())
     }
diff --git a/crates/hyperswitch_interfaces/src/secrets_interface/secret_state.rs b/crates/hyperswitch_interfaces/src/secrets_interface/secret_state.rs
index 6d3f75500af..d1da6a8c8b6 100644
--- a/crates/hyperswitch_interfaces/src/secrets_interface/secret_state.rs
+++ b/crates/hyperswitch_interfaces/src/secrets_interface/secret_state.rs
@@ -8,12 +8,12 @@ use serde::{Deserialize, Deserializer};
 pub trait SecretState {}
 
 /// Decrypted state
-#[derive(Debug, Clone, Deserialize)]
-pub enum RawSecret {}
+#[derive(Debug, Clone, Deserialize, Default)]
+pub struct RawSecret {}
 
 /// Encrypted state
-#[derive(Debug, Clone, Deserialize)]
-pub enum SecuredSecret {}
+#[derive(Debug, Clone, Deserialize, Default)]
+pub struct SecuredSecret {}
 
 impl SecretState for RawSecret {}
 impl SecretState for SecuredSecret {}