Issue:
Bug: Deep health check for Hyperswitch Server



Components to be verified in the check:
- [x] RDS (PostgreSQL)
  - Connection to Database
    - Read check
    - Write check (v2)
  - State of Migration (v2)
 - [x] ElastiCache (Redis)
   - Connection to Redis
   - Stream insertions
- [x] Locker Connection (/health (deep health check))
- [x] #3518
  - Verifying how the pod is able to connect to internet
- [x] Clickhouse

Pull request description:
## Type of Change
<!-- Put an `x` in the boxes that apply -->

- [ ] Bugfix
- [x] New feature
- [ ] Enhancement
- [ ] Refactoring
- [ ] Dependency updates
- [ ] Documentation
- [ ] CI/CD

## Description
<!-- Describe your changes in detail -->
This PR adds support for performing a deeper health check of the application which includes:
* Database connection check with read, write and delete queries (Added support for database transaction too)
* Redis connection check with set, get and delete key (Set key with expiry of 30sec)
* Locker health call from Hyperswitch application

The deep health check API returns a 200 status code if all the components are in good health, but returns 500 if at least one component's health is down

### 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)?
-->
1. Hit `/health` for getting just the application health.
2. Hit `/health/deep_check` for getting deep report of components health

* Test case 1 (Application not able to connect to database because postgresql is down)

![image](https://github.com/juspay/hyperswitch/assets/70657455/5e8af307-0b01-4008-a8e7-30d1aa8d3ed4)

* Test case 2 (Database Insert query failing because of duplication)

![image](https://github.com/juspay/hyperswitch/assets/70657455/0775c8db-bf43-4cf1-b197-33dea3b51058)

* Test case 3 (Locker is down)

![image](https://github.com/juspay/hyperswitch/assets/70657455/2bc6d734-8da5-4b87-8138-d22ba76a0bc7)

* Test case 4 (Locker is up but key custodian locked)

![image](https://github.com/juspay/hyperswitch/assets/70657455/35b83c7a-388f-4128-8913-66f1a8e23bde)

* Test case 5 (Locker is up with key custodian unlocked)

![image](https://github.com/juspay/hyperswitch/assets/70657455/ed64322e-08d0-458e-af69-fe8010d21a61)

## 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:

1. Hit `/health` for getting just the application health.
2. Hit `/health/deep_check` for getting deep report of components health

* Test case 1 (Application not able to connect to database because postgresql is down)

![image](https://github.com/juspay/hyperswitch/assets/70657455/5e8af307-0b01-4008-a8e7-30d1aa8d3ed4)

* Test case 2 (Database Insert query failing because of duplication)

![image](https://github.com/juspay/hyperswitch/assets/70657455/0775c8db-bf43-4cf1-b197-33dea3b51058)

* Test case 3 (Locker is down)

![image](https://github.com/juspay/hyperswitch/assets/70657455/2bc6d734-8da5-4b87-8138-d22ba76a0bc7)

* Test case 4 (Locker is up but key custodian locked)

![image](https://github.com/juspay/hyperswitch/assets/70657455/35b83c7a-388f-4128-8913-66f1a8e23bde)

* Test case 5 (Locker is up with key custodian unlocked)

![image](https://github.com/juspay/hyperswitch/assets/70657455/ed64322e-08d0-458e-af69-fe8010d21a61)

Patch:
diff --git a/crates/api_models/src/health_check.rs b/crates/api_models/src/health_check.rs
new file mode 100644
index 00000000000..d7bb120d017
--- /dev/null
+++ b/crates/api_models/src/health_check.rs
@@ -0,0 +1,6 @@
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+pub struct RouterHealthCheckResponse {
+    pub database: String,
+    pub redis: String,
+    pub locker: String,
+}
diff --git a/crates/api_models/src/lib.rs b/crates/api_models/src/lib.rs
index 935944cf74c..459443747e3 100644
--- a/crates/api_models/src/lib.rs
+++ b/crates/api_models/src/lib.rs
@@ -16,6 +16,7 @@ pub mod errors;
 pub mod events;
 pub mod files;
 pub mod gsm;
+pub mod health_check;
 pub mod locker_migration;
 pub mod mandates;
 pub mod organization;
diff --git a/crates/router/src/consts.rs b/crates/router/src/consts.rs
index 4a2d2831d10..eff42c0cd7c 100644
--- a/crates/router/src/consts.rs
+++ b/crates/router/src/consts.rs
@@ -70,3 +70,5 @@ pub const EMAIL_TOKEN_TIME_IN_SECS: u64 = 60 * 60 * 24; // 1 day
 pub const VERIFY_CONNECTOR_ID_PREFIX: &str = "conn_verify";
 #[cfg(feature = "olap")]
 pub const VERIFY_CONNECTOR_MERCHANT_ID: &str = "test_merchant";
+
+pub const LOCKER_HEALTH_CALL_PATH: &str = "/health";
diff --git a/crates/router/src/db.rs b/crates/router/src/db.rs
index 0cd4cb21881..5beace9cbb8 100644
--- a/crates/router/src/db.rs
+++ b/crates/router/src/db.rs
@@ -14,6 +14,7 @@ pub mod events;
 pub mod file;
 pub mod fraud_check;
 pub mod gsm;
+pub mod health_check;
 mod kafka_store;
 pub mod locker_mock_up;
 pub mod mandate;
@@ -103,6 +104,7 @@ pub trait StorageInterface:
     + user_role::UserRoleInterface
     + authorization::AuthorizationInterface
     + user::sample_data::BatchSampleDataInterface
+    + health_check::HealthCheckInterface
     + 'static
 {
     fn get_scheduler_db(&self) -> Box<dyn scheduler::SchedulerInterface>;
diff --git a/crates/router/src/db/health_check.rs b/crates/router/src/db/health_check.rs
new file mode 100644
index 00000000000..73bc2a4321d
--- /dev/null
+++ b/crates/router/src/db/health_check.rs
@@ -0,0 +1,147 @@
+use async_bb8_diesel::{AsyncConnection, AsyncRunQueryDsl};
+use diesel_models::ConfigNew;
+use error_stack::ResultExt;
+use router_env::logger;
+
+use super::{MockDb, StorageInterface, Store};
+use crate::{
+    connection,
+    consts::LOCKER_HEALTH_CALL_PATH,
+    core::errors::{self, CustomResult},
+    routes,
+    services::api as services,
+    types::storage,
+};
+
+#[async_trait::async_trait]
+pub trait HealthCheckInterface {
+    async fn health_check_db(&self) -> CustomResult<(), errors::HealthCheckDBError>;
+    async fn health_check_redis(
+        &self,
+        db: &dyn StorageInterface,
+    ) -> CustomResult<(), errors::HealthCheckRedisError>;
+    async fn health_check_locker(
+        &self,
+        state: &routes::AppState,
+    ) -> CustomResult<(), errors::HealthCheckLockerError>;
+}
+
+#[async_trait::async_trait]
+impl HealthCheckInterface for Store {
+    async fn health_check_db(&self) -> CustomResult<(), errors::HealthCheckDBError> {
+        let conn = connection::pg_connection_write(self)
+            .await
+            .change_context(errors::HealthCheckDBError::DBError)?;
+
+        let _data = conn
+            .transaction_async(|conn| {
+                Box::pin(async move {
+                    let query =
+                        diesel::select(diesel::dsl::sql::<diesel::sql_types::Integer>("1 + 1"));
+                    let _x: i32 = query.get_result_async(&conn).await.map_err(|err| {
+                        logger::error!(read_err=?err,"Error while reading element in the database");
+                        errors::HealthCheckDBError::DBReadError
+                    })?;
+
+                    logger::debug!("Database read was successful");
+
+                    let config = ConfigNew {
+                        key: "test_key".to_string(),
+                        config: "test_value".to_string(),
+                    };
+
+                    config.insert(&conn).await.map_err(|err| {
+                        logger::error!(write_err=?err,"Error while writing to database");
+                        errors::HealthCheckDBError::DBWriteError
+                    })?;
+
+                    logger::debug!("Database write was successful");
+
+                    storage::Config::delete_by_key(&conn, "test_key").await.map_err(|err| {
+                        logger::error!(delete_err=?err,"Error while deleting element in the database");
+                        errors::HealthCheckDBError::DBDeleteError
+                    })?;
+
+                    logger::debug!("Database delete was successful");
+
+                    Ok::<_, errors::HealthCheckDBError>(())
+                })
+            })
+            .await?;
+
+        Ok(())
+    }
+
+    async fn health_check_redis(
+        &self,
+        db: &dyn StorageInterface,
+    ) -> CustomResult<(), errors::HealthCheckRedisError> {
+        let redis_conn = db
+            .get_redis_conn()
+            .change_context(errors::HealthCheckRedisError::RedisConnectionError)?;
+
+        redis_conn
+            .serialize_and_set_key_with_expiry("test_key", "test_value", 30)
+            .await
+            .change_context(errors::HealthCheckRedisError::SetFailed)?;
+
+        logger::debug!("Redis set_key was successful");
+
+        redis_conn
+            .get_key("test_key")
+            .await
+            .change_context(errors::HealthCheckRedisError::GetFailed)?;
+
+        logger::debug!("Redis get_key was successful");
+
+        redis_conn
+            .delete_key("test_key")
+            .await
+            .change_context(errors::HealthCheckRedisError::DeleteFailed)?;
+
+        logger::debug!("Redis delete_key was successful");
+
+        Ok(())
+    }
+
+    async fn health_check_locker(
+        &self,
+        state: &routes::AppState,
+    ) -> CustomResult<(), errors::HealthCheckLockerError> {
+        let locker = &state.conf.locker;
+        if !locker.mock_locker {
+            let mut url = locker.host_rs.to_owned();
+            url.push_str(LOCKER_HEALTH_CALL_PATH);
+            let request = services::Request::new(services::Method::Get, &url);
+            services::call_connector_api(state, request)
+                .await
+                .change_context(errors::HealthCheckLockerError::FailedToCallLocker)?
+                .ok();
+        }
+
+        logger::debug!("Locker call was successful");
+
+        Ok(())
+    }
+}
+
+#[async_trait::async_trait]
+impl HealthCheckInterface for MockDb {
+    async fn health_check_db(&self) -> CustomResult<(), errors::HealthCheckDBError> {
+        Ok(())
+    }
+
+    async fn health_check_redis(
+        &self,
+        _: &dyn StorageInterface,
+    ) -> CustomResult<(), errors::HealthCheckRedisError> {
+        Ok(())
+    }
+
+    async fn health_check_locker(
+        &self,
+        _: &routes::AppState,
+    ) -> CustomResult<(), errors::HealthCheckLockerError> {
+        Ok(())
+    }
+}
diff --git a/crates/router/src/db/kafka_store.rs b/crates/router/src/db/kafka_store.rs
index db94c1bcbca..1184992a8f7 100644
--- a/crates/router/src/db/kafka_store.rs
+++ b/crates/router/src/db/kafka_store.rs
@@ -43,6 +43,7 @@ use crate::{
         events::EventInterface,
         file::FileMetadataInterface,
         gsm::GsmInterface,
+        health_check::HealthCheckInterface,
         locker_mock_up::LockerMockUpInterface,
         mandate::MandateInterface,
         merchant_account::MerchantAccountInterface,
@@ -57,6 +58,7 @@ use crate::{
         routing_algorithm::RoutingAlgorithmInterface,
         MasterKeyInterface, StorageInterface,
     },
+    routes,
     services::{authentication, kafka::KafkaProducer, Store},
     types::{
         domain,
@@ -2131,3 +2133,24 @@ impl AuthorizationInterface for KafkaStore {
             .await
     }
 }
+
+#[async_trait::async_trait]
+impl HealthCheckInterface for KafkaStore {
+    async fn health_check_db(&self) -> CustomResult<(), errors::HealthCheckDBError> {
+        self.diesel_store.health_check_db().await
+    }
+
+    async fn health_check_redis(
+        &self,
+        db: &dyn StorageInterface,
+    ) -> CustomResult<(), errors::HealthCheckRedisError> {
+        self.diesel_store.health_check_redis(db).await
+    }
+
+    async fn health_check_locker(
+        &self,
+        state: &routes::AppState,
+    ) -> CustomResult<(), errors::HealthCheckLockerError> {
+        self.diesel_store.health_check_locker(state).await
+    }
+}
diff --git a/crates/router/src/routes/app.rs b/crates/router/src/routes/app.rs
index 0357cedd443..6625a206be2 100644
--- a/crates/router/src/routes/app.rs
+++ b/crates/router/src/routes/app.rs
@@ -253,9 +253,10 @@ pub struct Health;
 
 impl Health {
     pub fn server(state: AppState) -> Scope {
-        web::scope("")
+        web::scope("health")
             .app_data(web::Data::new(state))
-            .service(web::resource("/health").route(web::get().to(health)))
+            .service(web::resource("").route(web::get().to(health)))
+            .service(web::resource("/deep_check").route(web::post().to(deep_health_check)))
     }
 }
 
diff --git a/crates/router/src/routes/health.rs b/crates/router/src/routes/health.rs
index 7c7f29bd181..f07b744f7f5 100644
--- a/crates/router/src/routes/health.rs
+++ b/crates/router/src/routes/health.rs
@@ -1,7 +1,9 @@
+use actix_web::web;
+use api_models::health_check::RouterHealthCheckResponse;
 use router_env::{instrument, logger, tracing};
 
-use crate::routes::metrics;
-
+use super::app;
+use crate::{routes::metrics, services};
 /// .
 // #[logger::instrument(skip_all, name = "name1", level = "warn", fields( key1 = "val1" ))]
 #[instrument(skip_all)]
@@ -11,3 +13,59 @@ pub async fn health() -> impl actix_web::Responder {
     logger::info!("Health was called");
     actix_web::HttpResponse::Ok().body("health is good")
 }
+
+#[instrument(skip_all)]
+pub async fn deep_health_check(state: web::Data<app::AppState>) -> impl actix_web::Responder {
+    metrics::HEALTH_METRIC.add(&metrics::CONTEXT, 1, &[]);
+    let db = &*state.store;
+    let mut status_code = 200;
+    logger::info!("Deep health check was called");
+
+    logger::debug!("Database health check begin");
+
+    let db_status = match db.health_check_db().await {
+        Ok(_) => "Health is good".to_string(),
+        Err(err) => {
+            status_code = 500;
+            err.to_string()
+        }
+    };
+    logger::debug!("Database health check end");
+
+    logger::debug!("Redis health check begin");
+
+    let redis_status = match db.health_check_redis(db).await {
+        Ok(_) => "Health is good".to_string(),
+        Err(err) => {
+            status_code = 500;
+            err.to_string()
+        }
+    };
+
+    logger::debug!("Redis health check end");
+
+    logger::debug!("Locker health check begin");
+
+    let locker_status = match db.health_check_locker(&state).await {
+        Ok(_) => "Health is good".to_string(),
+        Err(err) => {
+            status_code = 500;
+            err.to_string()
+        }
+    };
+
+    logger::debug!("Locker health check end");
+
+    let response = serde_json::to_string(&RouterHealthCheckResponse {
+        database: db_status,
+        redis: redis_status,
+        locker: locker_status,
+    })
+    .unwrap_or_default();
+
+    if status_code == 200 {
+        services::http_response_json(response)
+    } else {
+        services::http_server_error_json_response(response)
+    }
+}
diff --git a/crates/router/src/services/api.rs b/crates/router/src/services/api.rs
index 92fda578727..71687e02c12 100644
--- a/crates/router/src/services/api.rs
+++ b/crates/router/src/services/api.rs
@@ -1138,6 +1138,14 @@ pub fn http_response_json<T: body::MessageBody + 'static>(response: T) -> HttpRe
         .body(response)
 }
 
+pub fn http_server_error_json_response<T: body::MessageBody + 'static>(
+    response: T,
+) -> HttpResponse {
+    HttpResponse::InternalServerError()
+        .content_type(mime::APPLICATION_JSON)
+        .body(response)
+}
+
 pub fn http_response_json_with_headers<T: body::MessageBody + 'static>(
     response: T,
     mut headers: Vec<(String, String)>,
diff --git a/crates/router/src/services/api/client.rs b/crates/router/src/services/api/client.rs
index cc7353dcda6..fca85c41699 100644
--- a/crates/router/src/services/api/client.rs
+++ b/crates/router/src/services/api/client.rs
@@ -10,6 +10,7 @@ use router_env::tracing_actix_web::RequestId;
 use super::{request::Maskable, Request};
 use crate::{
     configs::settings::{Locker, Proxy},
+    consts::LOCKER_HEALTH_CALL_PATH,
     core::{
         errors::{ApiClientError, CustomResult},
         payments,
@@ -119,6 +120,7 @@ pub fn proxy_bypass_urls(locker: &Locker) -> Vec<String> {
         format!("{locker_host_rs}/cards/add"),
         format!("{locker_host_rs}/cards/retrieve"),
         format!("{locker_host_rs}/cards/delete"),
+        format!("{locker_host_rs}{}", LOCKER_HEALTH_CALL_PATH),
         format!("{locker_host}/card/addCard"),
         format!("{locker_host}/card/getCard"),
         format!("{locker_host}/card/deleteCard"),
diff --git a/crates/storage_impl/src/errors.rs b/crates/storage_impl/src/errors.rs
index f0cbebf78c5..50173bb1c73 100644
--- a/crates/storage_impl/src/errors.rs
+++ b/crates/storage_impl/src/errors.rs
@@ -376,3 +376,53 @@ pub enum ConnectorError {
     #[error("Missing 3DS redirection payload: {field_name}")]
     MissingConnectorRedirectionPayload { field_name: &'static str },
 }
+
+#[derive(Debug, thiserror::Error)]
+pub enum HealthCheckDBError {
+    #[error("Error while connecting to database")]
+    DBError,
+    #[error("Error while writing to database")]
+    DBWriteError,
+    #[error("Error while reading element in the database")]
+    DBReadError,
+    #[error("Error while deleting element in the database")]
+    DBDeleteError,
+    #[error("Unpredictable error occurred")]
+    UnknownError,
+    #[error("Error in database transaction")]
+    TransactionError,
+}
+
+impl From<diesel::result::Error> for HealthCheckDBError {
+    fn from(error: diesel::result::Error) -> Self {
+        match error {
+            diesel::result::Error::DatabaseError(_, _) => Self::DBError,
+
+            diesel::result::Error::RollbackErrorOnCommit { .. }
+            | diesel::result::Error::RollbackTransaction
+            | diesel::result::Error::AlreadyInTransaction
+            | diesel::result::Error::NotInTransaction
+            | diesel::result::Error::BrokenTransactionManager => Self::TransactionError,
+
+            _ => Self::UnknownError,
+        }
+    }
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum HealthCheckRedisError {
+    #[error("Failed to establish Redis connection")]
+    RedisConnectionError,
+    #[error("Failed to set key value in Redis")]
+    SetFailed,
+    #[error("Failed to get key value in Redis")]
+    GetFailed,
+    #[error("Failed to delete key value in Redis")]
+    DeleteFailed,
+}
+
+#[derive(Debug, Clone, thiserror::Error)]
+pub enum HealthCheckLockerError {
+    #[error("Failed to establish Locker connection")]
+    FailedToCallLocker,
+}