Issue:
Bug: Deep health check for Hyperswitch Card Vault



Components to be verified in this check

- [x] RDS (PostgreSQL)

  - Connection to Database

  - State of Migration (v2)

- [x] Key Custodian State

Pull request description:
## Type of Change
<!-- Put an `x` in the boxes that apply -->

- [ ] Bugfix
- [x] New feature
- [ ] Enhancement
- [ ] Refactoring
- [ ] Dependency updates

## Description
GDPR payment regulations state that all businesses that collect/process personal data of customers from Europe need to comply with the regulatory requirements. GDPR protects customers’ privacy by empowering the customer with the ownership of their personal data and thereby enabling them to  dictate how their personal data should be handled by those who collected it
customer delete call also delete all the data associated with that customer

### Additional Changes

- [ ] 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 its an obvious bug or documentation fix
that will have little conversation).
-->


## How did you test it?
Not tested yet


## Checklist
<!-- Put an `x` in the boxes that apply -->

- [x] I formatted the code `cargo +nightly fmt`
- [x] I addressed lints thrown by `cargo clippy`
- [x] I reviewed submitted code
- [x] I added unit tests for my changes where possible
- [x] I added a [CHANGELOG](/CHANGELOG.md) entry if applicable


Testing:
Not tested yet

Patch:
diff --git a/crates/api_models/src/customers.rs b/crates/api_models/src/customers.rs
index 27611334fe0..e7c131a4a49 100644
--- a/crates/api_models/src/customers.rs
+++ b/crates/api_models/src/customers.rs
@@ -39,7 +39,9 @@ pub struct CustomerId {
 #[derive(Default, Debug, Deserialize, Serialize)]
 pub struct CustomerDeleteResponse {
     pub customer_id: String,
-    pub deleted: bool,
+    pub customer_deleted: bool,
+    pub address_deleted: bool,
+    pub payment_methods_deleted: bool,
 }
 
 pub fn generate_customer_id() -> String {
diff --git a/crates/router/src/compatibility/stripe/customers.rs b/crates/router/src/compatibility/stripe/customers.rs
index 590f7995925..2dec79707d4 100644
--- a/crates/router/src/compatibility/stripe/customers.rs
+++ b/crates/router/src/compatibility/stripe/customers.rs
@@ -111,9 +111,7 @@ pub async fn customer_delete(
         &state,
         &req,
         payload,
-        |state, merchant_account, req| {
-            customers::delete_customer(&*state.store, merchant_account, req)
-        },
+        customers::delete_customer,
         api::MerchantAuthentication::ApiKey,
     )
     .await
diff --git a/crates/router/src/compatibility/stripe/customers/types.rs b/crates/router/src/compatibility/stripe/customers/types.rs
index 8b1716d7e74..015bd516cb2 100644
--- a/crates/router/src/compatibility/stripe/customers/types.rs
+++ b/crates/router/src/compatibility/stripe/customers/types.rs
@@ -104,7 +104,7 @@ impl From<api::CustomerDeleteResponse> for CustomerDeleteResponse {
     fn from(cust: api::CustomerDeleteResponse) -> Self {
         Self {
             id: cust.customer_id,
-            deleted: cust.deleted,
+            deleted: cust.customer_deleted,
         }
     }
 }
diff --git a/crates/router/src/compatibility/stripe/errors.rs b/crates/router/src/compatibility/stripe/errors.rs
index f2f38ee997f..0ee8c8a3d2a 100644
--- a/crates/router/src/compatibility/stripe/errors.rs
+++ b/crates/router/src/compatibility/stripe/errors.rs
@@ -55,6 +55,12 @@ pub(crate) enum ErrorCode {
     #[error(error_type = StripeErrorType::ApiError, code = "internal_server_error", message = "Server is down")]
     DuplicateRefundRequest,
 
+    #[error(error_type = StripeErrorType::InvalidRequestError, code = "active_mandate", message = "Customer has active mandate")]
+    MandateActive,
+
+    #[error(error_type = StripeErrorType::InvalidRequestError, code = "customer_redacted", message = "Customer has redacted")]
+    CustomerRedacted,
+
     #[error(error_type = StripeErrorType::InvalidRequestError, code = "resource_missing", message = "No such refund")]
     RefundNotFound,
 
@@ -340,6 +346,8 @@ impl From<ApiErrorResponse> for ErrorCode {
             ApiErrorResponse::RefundFailed { data } => ErrorCode::RefundFailed, // Nothing at stripe to map
 
             ApiErrorResponse::InternalServerError => ErrorCode::InternalServerError, // not a stripe code
+            ApiErrorResponse::MandateActive => ErrorCode::MandateActive, //not a stripe code
+            ApiErrorResponse::CustomerRedacted => ErrorCode::CustomerRedacted, //not a stripe code
             ApiErrorResponse::DuplicateRefundRequest => ErrorCode::DuplicateRefundRequest,
             ApiErrorResponse::RefundNotFound => ErrorCode::RefundNotFound,
             ApiErrorResponse::CustomerNotFound => ErrorCode::CustomerNotFound,
@@ -433,9 +441,10 @@ impl actix_web::ResponseError for ErrorCode {
             | ErrorCode::ResourceIdNotFound
             | ErrorCode::PaymentIntentMandateInvalid { .. }
             | ErrorCode::PaymentIntentUnexpectedState { .. } => StatusCode::BAD_REQUEST,
-            ErrorCode::RefundFailed | ErrorCode::InternalServerError => {
-                StatusCode::INTERNAL_SERVER_ERROR
-            }
+            ErrorCode::RefundFailed
+            | ErrorCode::InternalServerError
+            | ErrorCode::MandateActive
+            | ErrorCode::CustomerRedacted => StatusCode::INTERNAL_SERVER_ERROR,
             ErrorCode::ReturnUrlUnavailable => StatusCode::SERVICE_UNAVAILABLE,
         }
     }
diff --git a/crates/router/src/core/customers.rs b/crates/router/src/core/customers.rs
index e0981bca4ff..fdc7e375a8d 100644
--- a/crates/router/src/core/customers.rs
+++ b/crates/router/src/core/customers.rs
@@ -2,12 +2,16 @@ use error_stack::ResultExt;
 use router_env::{tracing, tracing::instrument};
 
 use crate::{
-    core::errors::{self, RouterResponse, StorageErrorExt},
+    core::{
+        errors::{self, RouterResponse, StorageErrorExt},
+        payment_methods::cards,
+    },
     db::StorageInterface,
+    routes::AppState,
     services,
     types::{
         api::customers::{self, CustomerRequestExt},
-        storage,
+        storage::{self, enums},
     },
 };
 
@@ -63,20 +67,99 @@ pub async fn retrieve_customer(
     Ok(services::BachResponse::Json(response.into()))
 }
 
-#[instrument(skip(db))]
+#[instrument(skip_all)]
 pub async fn delete_customer(
-    db: &dyn StorageInterface,
+    state: &AppState,
     merchant_account: storage::MerchantAccount,
     req: customers::CustomerId,
 ) -> RouterResponse<customers::CustomerDeleteResponse> {
-    let response = db
-        .delete_customer_by_customer_id_merchant_id(&req.customer_id, &merchant_account.merchant_id)
+    let db = &state.store;
+
+    let cust = db
+        .find_customer_by_customer_id_merchant_id(&req.customer_id, &merchant_account.merchant_id)
         .await
-        .map(|response| customers::CustomerDeleteResponse {
-            customer_id: req.customer_id,
-            deleted: response,
-        })
-        .map_err(|error| error.to_not_found_response(errors::ApiErrorResponse::CustomerNotFound))?;
+        .map_err(|err| err.to_not_found_response(errors::ApiErrorResponse::CustomerNotFound))?;
+    if cust.name == Some("Redacted".to_string()) {
+        Err(errors::ApiErrorResponse::CustomerRedacted)?
+    }
+
+    let customer_mandates = db
+        .find_mandate_by_merchant_id_customer_id(&merchant_account.merchant_id, &req.customer_id)
+        .await
+        .map_err(|err| err.to_not_found_response(errors::ApiErrorResponse::MandateNotFound))?;
+
+    for mandate in customer_mandates.into_iter() {
+        if mandate.mandate_status == enums::MandateStatus::Active {
+            Err(errors::ApiErrorResponse::MandateActive)?
+        }
+    }
+
+    let customer_payment_methods = db
+        .find_payment_method_by_customer_id_merchant_id_list(
+            &req.customer_id,
+            &merchant_account.merchant_id,
+        )
+        .await
+        .map_err(|err| {
+            err.to_not_found_response(errors::ApiErrorResponse::PaymentMethodNotFound)
+        })?;
+    for pm in customer_payment_methods.into_iter() {
+        if pm.payment_method == enums::PaymentMethodType::Card {
+            cards::delete_card(state, &merchant_account.merchant_id, &pm.payment_method_id).await?;
+        }
+        db.delete_payment_method_by_merchant_id_payment_method_id(
+            &merchant_account.merchant_id,
+            &pm.payment_method_id,
+        )
+        .await
+        .map_err(|error| {
+            error.to_not_found_response(errors::ApiErrorResponse::PaymentMethodNotFound)
+        })?;
+    }
+
+    let update_address = storage::AddressUpdate::Update {
+        city: Some("Redacted".to_string()),
+        country: Some("Redacted".to_string()),
+        line1: Some("Redacted".to_string().into()),
+        line2: Some("Redacted".to_string().into()),
+        line3: Some("Redacted".to_string().into()),
+        state: Some("Redacted".to_string().into()),
+        zip: Some("Redacted".to_string().into()),
+        first_name: Some("Redacted".to_string().into()),
+        last_name: Some("Redacted".to_string().into()),
+        phone_number: Some("Redacted".to_string().into()),
+        country_code: Some("Redacted".to_string()),
+    };
+    db.update_address_by_merchant_id_customer_id(
+        &req.customer_id,
+        &merchant_account.merchant_id,
+        update_address,
+    )
+    .await
+    .change_context(errors::ApiErrorResponse::AddressNotFound)?;
+
+    let updated_customer = storage::CustomerUpdate::Update {
+        name: Some("Redacted".to_string()),
+        email: Some("Redacted".to_string().into()),
+        phone: Some("Redacted".to_string().into()),
+        description: Some("Redacted".to_string()),
+        phone_country_code: Some("Redacted".to_string()),
+        metadata: None,
+    };
+    db.update_customer_by_customer_id_merchant_id(
+        req.customer_id.clone(),
+        merchant_account.merchant_id,
+        updated_customer,
+    )
+    .await
+    .change_context(errors::ApiErrorResponse::CustomerNotFound)?;
+
+    let response = customers::CustomerDeleteResponse {
+        customer_id: req.customer_id,
+        customer_deleted: true,
+        address_deleted: true,
+        payment_methods_deleted: true,
+    };
     Ok(services::BachResponse::Json(response))
 }
 
diff --git a/crates/router/src/core/errors/api_error_response.rs b/crates/router/src/core/errors/api_error_response.rs
index 88dbb652cb7..12cc827e702 100644
--- a/crates/router/src/core/errors/api_error_response.rs
+++ b/crates/router/src/core/errors/api_error_response.rs
@@ -28,8 +28,6 @@ pub enum ApiErrorResponse {
                     the Dashboard Settings section."
     )]
     BadCredentials,
-    #[error(error_type = ErrorType::InvalidRequestError, code = "IR_10", message = "Invalid Ephemeral Key for the customer")]
-    InvalidEphermeralKey,
     #[error(error_type = ErrorType::InvalidRequestError, code = "IR_03", message = "Unrecognized request URL.")]
     InvalidRequestUrl,
     #[error(error_type = ErrorType::InvalidRequestError, code = "IR_04", message = "The HTTP method is not applicable for this API.")]
@@ -46,14 +44,20 @@ pub enum ApiErrorResponse {
     },
     #[error(error_type = ErrorType::InvalidRequestError, code = "IR_07", message = "{message}")]
     InvalidRequestData { message: String },
-    #[error(error_type = ErrorType::InvalidRequestError, code = "IR_08", message = "Refund amount exceeds the payment amount.")]
-    RefundAmountExceedsPaymentAmount,
     /// Typically used when a field has invalid value, or deserialization of the value contained in
     /// a field fails.
     #[error(error_type = ErrorType::InvalidRequestError, code = "IR_07", message = "Invalid value provided: {field_name}.")]
     InvalidDataValue { field_name: &'static str },
+    #[error(error_type = ErrorType::InvalidRequestError, code = "IR_07", message = "The client_secret provided does not match the client_secret associated with the Payment.")]
+    ClientSecretInvalid,
+    #[error(error_type = ErrorType::InvalidRequestError, code = "IR_07", message = "Customer has existing mandate/subsciption.")]
+    MandateActive,
+    #[error(error_type = ErrorType::InvalidRequestError, code = "IR_07", message = "Customer has already redacted.")]
+    CustomerRedacted,
     #[error(error_type = ErrorType::InvalidRequestError, code = "IR_08", message = "Reached maximum refund attempts")]
     MaximumRefundCount,
+    #[error(error_type = ErrorType::InvalidRequestError, code = "IR_08", message = "Refund amount exceeds the payment amount.")]
+    RefundAmountExceedsPaymentAmount,
     #[error(error_type = ErrorType::InvalidRequestError, code = "IR_09", message = "This PaymentIntent could not be {current_flow} because it has a {field_name} of {current_value}. The expected state is {states}.")]
     PaymentUnexpectedState {
         current_flow: String,
@@ -61,14 +65,13 @@ pub enum ApiErrorResponse {
         current_value: String,
         states: String,
     },
+    #[error(error_type = ErrorType::InvalidRequestError, code = "IR_10", message = "Invalid Ephemeral Key for the customer")]
+    InvalidEphermeralKey,
     /// Typically used when information involving multiple fields or previously provided
     /// information doesn't satisfy a condition.
     #[error(error_type = ErrorType::InvalidRequestError, code = "IR_10", message = "{message}")]
     PreconditionFailed { message: String },
 
-    #[error(error_type = ErrorType::InvalidRequestError, code = "IR_07", message = "The client_secret provided does not match the client_secret associated with the Payment.")]
-    ClientSecretInvalid,
-
     #[error(error_type = ErrorType::ProcessingError, code = "CE_01", message = "Payment failed while processing with connector. Retry payment.")]
     PaymentAuthorizationFailed { data: Option<serde_json::Value> },
     #[error(error_type = ErrorType::ProcessingError, code = "CE_02", message = "Payment failed while processing with connector. Retry payment.")]
@@ -168,6 +171,8 @@ impl actix_web::ResponseError for ApiErrorResponse {
             ApiErrorResponse::DuplicateRefundRequest => StatusCode::BAD_REQUEST,        // 400
             ApiErrorResponse::RefundNotFound
             | ApiErrorResponse::CustomerNotFound
+            | ApiErrorResponse::MandateActive
+            | ApiErrorResponse::CustomerRedacted
             | ApiErrorResponse::PaymentNotFound
             | ApiErrorResponse::PaymentMethodNotFound
             | ApiErrorResponse::MerchantAccountNotFound
diff --git a/crates/router/src/core/payment_methods/cards.rs b/crates/router/src/core/payment_methods/cards.rs
index 862c075346f..20fa547987e 100644
--- a/crates/router/src/core/payment_methods/cards.rs
+++ b/crates/router/src/core/payment_methods/cards.rs
@@ -198,6 +198,23 @@ pub async fn mock_get_card<'a>(
     })
 }
 
+#[instrument(skip_all)]
+pub async fn mock_delete_card<'a>(
+    db: &dyn db::StorageInterface,
+    card_id: &'a str,
+) -> errors::CustomResult<payment_methods::DeleteCardResponse, errors::CardVaultError> {
+    let locker_mock_up = db
+        .delete_locker_mock_up(card_id)
+        .await
+        .change_context(errors::CardVaultError::FetchCardFailed)?;
+    Ok(payment_methods::DeleteCardResponse {
+        card_id: locker_mock_up.card_id,
+        external_id: locker_mock_up.external_id,
+        card_isin: None,
+        status: "SUCCESS".to_string(),
+    })
+}
+
 #[instrument(skip_all)]
 pub async fn get_card_from_legacy_locker<'a>(
     state: &'a routes::AppState,
@@ -238,17 +255,25 @@ pub async fn delete_card<'a>(
     merchant_id: &'a str,
     card_id: &'a str,
 ) -> errors::RouterResult<payment_methods::DeleteCardResponse> {
+    let locker = &state.conf.locker;
     let request = payment_methods::mk_delete_card_request(&state.conf.locker, merchant_id, card_id)
         .change_context(errors::ApiErrorResponse::InternalServerError)
         .attach_printable("Making Delete card request Failed")?;
     // FIXME use call_api 2. Serde's handle should be inside the generic function
-    let delete_card_resp = services::call_connector_api(state, request)
-        .await
-        .change_context(errors::ApiErrorResponse::InternalServerError)?
-        .map_err(|_x| errors::ApiErrorResponse::InternalServerError)?
-        .response
-        .parse_struct("DeleteCardResponse")
-        .change_context(errors::ApiErrorResponse::InternalServerError)?;
+    let delete_card_resp = if !locker.mock_locker {
+        services::call_connector_api(state, request)
+            .await
+            .change_context(errors::ApiErrorResponse::InternalServerError)?
+            .map_err(|_x| errors::ApiErrorResponse::InternalServerError)?
+            .response
+            .parse_struct("DeleteCardResponse")
+            .change_context(errors::ApiErrorResponse::InternalServerError)?
+    } else {
+        mock_delete_card(&*state.store, card_id)
+            .await
+            .change_context(errors::ApiErrorResponse::InternalServerError)?
+    };
+
     Ok(delete_card_resp)
 }
 
diff --git a/crates/router/src/db/address.rs b/crates/router/src/db/address.rs
index 38a33feaa25..9a763044a20 100644
--- a/crates/router/src/db/address.rs
+++ b/crates/router/src/db/address.rs
@@ -14,14 +14,23 @@ pub trait AddressInterface {
         address_id: String,
         address: storage::AddressUpdate,
     ) -> CustomResult<storage::Address, errors::StorageError>;
+
     async fn insert_address(
         &self,
         address: storage::AddressNew,
     ) -> CustomResult<storage::Address, errors::StorageError>;
+
     async fn find_address(
         &self,
         address_id: &str,
     ) -> CustomResult<storage::Address, errors::StorageError>;
+
+    async fn update_address_by_merchant_id_customer_id(
+        &self,
+        customer_id: &str,
+        merchant_id: &str,
+        address: storage::AddressUpdate,
+    ) -> CustomResult<Vec<storage::Address>, errors::StorageError>;
 }
 
 #[async_trait::async_trait]
@@ -60,6 +69,24 @@ impl AddressInterface for Store {
             .map_err(Into::into)
             .into_report()
     }
+
+    async fn update_address_by_merchant_id_customer_id(
+        &self,
+        customer_id: &str,
+        merchant_id: &str,
+        address: storage::AddressUpdate,
+    ) -> CustomResult<Vec<storage::Address>, errors::StorageError> {
+        let conn = pg_connection(&self.master_pool).await;
+        storage::Address::update_by_merchant_id_customer_id(
+            &conn,
+            customer_id,
+            merchant_id,
+            address,
+        )
+        .await
+        .map_err(Into::into)
+        .into_report()
+    }
 }
 
 #[async_trait::async_trait]
@@ -85,4 +112,13 @@ impl AddressInterface for MockDb {
     ) -> CustomResult<storage::Address, errors::StorageError> {
         todo!()
     }
+
+    async fn update_address_by_merchant_id_customer_id(
+        &self,
+        _customer_id: &str,
+        _merchant_id: &str,
+        _address: storage::AddressUpdate,
+    ) -> CustomResult<Vec<storage::Address>, errors::StorageError> {
+        todo!()
+    }
 }
diff --git a/crates/router/src/db/locker_mock_up.rs b/crates/router/src/db/locker_mock_up.rs
index 20da7085452..cc023726348 100644
--- a/crates/router/src/db/locker_mock_up.rs
+++ b/crates/router/src/db/locker_mock_up.rs
@@ -18,6 +18,11 @@ pub trait LockerMockUpInterface {
         &self,
         new: storage::LockerMockUpNew,
     ) -> CustomResult<storage::LockerMockUp, errors::StorageError>;
+
+    async fn delete_locker_mock_up(
+        &self,
+        card_id: &str,
+    ) -> CustomResult<storage::LockerMockUp, errors::StorageError>;
 }
 
 #[async_trait::async_trait]
@@ -40,6 +45,17 @@ impl LockerMockUpInterface for Store {
         let conn = pg_connection(&self.master_pool).await;
         new.insert(&conn).await.map_err(Into::into).into_report()
     }
+
+    async fn delete_locker_mock_up(
+        &self,
+        card_id: &str,
+    ) -> CustomResult<storage::LockerMockUp, errors::StorageError> {
+        let conn = pg_connection(&self.master_pool).await;
+        storage::LockerMockUp::delete_by_card_id(&conn, card_id)
+            .await
+            .map_err(Into::into)
+            .into_report()
+    }
 }
 
 #[async_trait::async_trait]
@@ -57,4 +73,11 @@ impl LockerMockUpInterface for MockDb {
     ) -> CustomResult<storage::LockerMockUp, errors::StorageError> {
         todo!()
     }
+
+    async fn delete_locker_mock_up(
+        &self,
+        _card_id: &str,
+    ) -> CustomResult<storage::LockerMockUp, errors::StorageError> {
+        todo!()
+    }
 }
diff --git a/crates/router/src/routes/customers.rs b/crates/router/src/routes/customers.rs
index b60cb007120..419e4672e62 100644
--- a/crates/router/src/routes/customers.rs
+++ b/crates/router/src/routes/customers.rs
@@ -94,7 +94,7 @@ pub async fn customers_delete(
         &state,
         &req,
         payload,
-        |state, merchant_account, req| delete_customer(&*state.store, merchant_account, req),
+        delete_customer,
         api::MerchantAuthentication::ApiKey,
     )
     .await
diff --git a/crates/storage_models/src/query/address.rs b/crates/storage_models/src/query/address.rs
index 4b9b74cc9fd..b472290b51d 100644
--- a/crates/storage_models/src/query/address.rs
+++ b/crates/storage_models/src/query/address.rs
@@ -1,4 +1,4 @@
-use diesel::{associations::HasTable, ExpressionMethods};
+use diesel::{associations::HasTable, BoolExpressionMethods, ExpressionMethods};
 use router_env::{tracing, tracing::instrument};
 
 use super::generics::{self, ExecuteQuery};
@@ -61,6 +61,23 @@ impl Address {
         .await
     }
 
+    pub async fn update_by_merchant_id_customer_id(
+        conn: &PgPooledConn,
+        customer_id: &str,
+        merchant_id: &str,
+        address: AddressUpdate,
+    ) -> CustomResult<Vec<Self>, errors::DatabaseError> {
+        generics::generic_update_with_results::<<Self as HasTable>::Table, _, _, Self, _>(
+            conn,
+            dsl::merchant_id
+                .eq(merchant_id.to_owned())
+                .and(dsl::customer_id.eq(customer_id.to_owned())),
+            AddressUpdateInternal::from(address),
+            ExecuteQuery::new(),
+        )
+        .await
+    }
+
     #[instrument(skip(conn))]
     pub async fn find_by_address_id<'a>(
         conn: &PgPooledConn,
diff --git a/crates/storage_models/src/query/locker_mock_up.rs b/crates/storage_models/src/query/locker_mock_up.rs
index af0cf1d46b4..a634110bca7 100644
--- a/crates/storage_models/src/query/locker_mock_up.rs
+++ b/crates/storage_models/src/query/locker_mock_up.rs
@@ -31,4 +31,17 @@ impl LockerMockUp {
         )
         .await
     }
+
+    #[instrument(skip(conn))]
+    pub async fn delete_by_card_id(
+        conn: &PgPooledConn,
+        card_id: &str,
+    ) -> CustomResult<Self, errors::DatabaseError> {
+        generics::generic_delete_one_with_result::<<Self as HasTable>::Table, _, Self, _>(
+            conn,
+            dsl::card_id.eq(card_id.to_owned()),
+            ExecuteQuery::new(),
+        )
+        .await
+    }
 }