fix(router): make customer details None in the Psync flow if the cu…

…stomer is deleted (#5732)

Co-authored-by: hyperswitch-bot[bot] <148525504+hyperswitch-bot[bot]@users.noreply.github.com>

crates/router/src/core/payments/helpers.rs

@@ -1372,6 +1372,18 @@ pub async fn get_customer_from_details<F: Clone>(
     todo!()
 }
 
+#[cfg(all(feature = "v2", feature = "customer_v2"))]
+pub async fn get_customer_details_even_for_redacted_customer<F: Clone>(
+    _state: &SessionState,
+    _customer_id: Option<id_type::CustomerId>,
+    _merchant_id: &id_type::MerchantId,
+    _payment_data: &mut PaymentData<F>,
+    _merchant_key_store: &domain::MerchantKeyStore,
+    _storage_scheme: enums::MerchantStorageScheme,
+) -> CustomResult<Option<domain::Customer>, errors::StorageError> {
+    todo!()
+}
+
 #[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "customer_v2")))]
 pub async fn get_customer_from_details<F: Clone>(
     state: &SessionState,
@@ -1712,6 +1724,44 @@ pub async fn create_customer_if_not_exist<'a, F: Clone, R>(
     ))
 }
 
+// This function is to retrieve customer details. If the customer is deleted, it returns
+// customer details that contains the fields as Redacted
+#[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "customer_v2")))]
+pub async fn get_customer_details_even_for_redacted_customer<F: Clone>(
+    state: &SessionState,
+    customer_id: Option<id_type::CustomerId>,
+    merchant_id: &id_type::MerchantId,
+    payment_data: &mut PaymentData<F>,
+    merchant_key_store: &domain::MerchantKeyStore,
+    storage_scheme: enums::MerchantStorageScheme,
+) -> CustomResult<Option<domain::Customer>, errors::StorageError> {
+    match customer_id {
+        None => Ok(None),
+        Some(customer_id) => {
+            let db = &*state.store;
+            let customer_details = db
+                .find_customer_optional_with_redacted_customer_details_by_customer_id_merchant_id(
+                    &state.into(),
+                    &customer_id,
+                    merchant_id,
+                    merchant_key_store,
+                    storage_scheme,
+                )
+                .await?;
+
+            payment_data.email = payment_data.email.clone().or_else(|| {
+                customer_details.as_ref().and_then(|inner| {
+                    inner
+                        .email
+                        .clone()
+                        .map(|encrypted_value| encrypted_value.into())
+                })
+            });
+            Ok(customer_details)
+        }
+    }
+}
+
 pub async fn retrieve_payment_method_with_temporary_token(
     state: &SessionState,
     token: &str,

crates/router/src/core/payments/operations.rs

@@ -276,7 +276,7 @@ where
     > {
         Ok((
             Box::new(self),
-            helpers::get_customer_from_details(
+            helpers::get_customer_details_even_for_redacted_customer(
                 state,
                 payment_data.payment_intent.customer_id.clone(),
                 &merchant_key_store.merchant_id,

crates/router/src/core/payments/transformers.rs

@@ -601,36 +601,50 @@ where
     let customer_table_response: Option<CustomerDetailsResponse> =
         customer.as_ref().map(ForeignInto::foreign_into);
 
-    // If we have customer data in Payment Intent, We are populating the Retrieve response from the
-    // same
+    // If we have customer data in Payment Intent and if the customer is not deleted, We are populating the Retrieve response from the
+    // same. If the customer is deleted then we use the customer table to populate customer details
     let customer_details_response =
         if let Some(customer_details_raw) = payment_intent.customer_details.clone() {
             let customer_details_encrypted =
                 serde_json::from_value::<CustomerData>(customer_details_raw.into_inner().expose());
             if let Ok(customer_details_encrypted_data) = customer_details_encrypted {
                 Some(CustomerDetailsResponse {
-                    id: customer_table_response.and_then(|customer_data| customer_data.id),
-                    name: customer_details_encrypted_data
-                        .name
-                        .or(customer.as_ref().and_then(|customer| {
-                            customer.name.as_ref().map(|name| name.clone().into_inner())
-                        })),
-                    email: customer_details_encrypted_data.email.or(customer
+                    id: customer_table_response
                         .as_ref()
-                        .and_then(|customer| customer.email.clone().map(Email::from))),
-                    phone: customer_details_encrypted_data
-                        .phone
-                        .or(customer.as_ref().and_then(|customer| {
-                            customer
-                                .phone
-                                .as_ref()
-                                .map(|phone| phone.clone().into_inner())
-                        })),
-                    phone_country_code: customer_details_encrypted_data.phone_country_code.or(
-                        customer
+                        .and_then(|customer_data| customer_data.id.clone()),
+                    name: customer_table_response
+                        .as_ref()
+                        .and_then(|customer_data| customer_data.name.clone())
+                        .or(customer_details_encrypted_data
+                            .name
+                            .or(customer.as_ref().and_then(|customer| {
+                                customer.name.as_ref().map(|name| name.clone().into_inner())
+                            }))),
+                    email: customer_table_response
+                        .as_ref()
+                        .and_then(|customer_data| customer_data.email.clone())
+                        .or(customer_details_encrypted_data.email.or(customer
                             .as_ref()
-                            .and_then(|customer| customer.phone_country_code.clone()),
-                    ),
+                            .and_then(|customer| customer.email.clone().map(Email::from)))),
+                    phone: customer_table_response
+                        .as_ref()
+                        .and_then(|customer_data| customer_data.phone.clone())
+                        .or(customer_details_encrypted_data
+                            .phone
+                            .or(customer.as_ref().and_then(|customer| {
+                                customer
+                                    .phone
+                                    .as_ref()
+                                    .map(|phone| phone.clone().into_inner())
+                            }))),
+                    phone_country_code: customer_table_response
+                        .as_ref()
+                        .and_then(|customer_data| customer_data.phone_country_code.clone())
+                        .or(customer_details_encrypted_data
+                            .phone_country_code
+                            .or(customer
+                                .as_ref()
+                                .and_then(|customer| customer.phone_country_code.clone()))),
                 })
             } else {
                 customer_table_response

crates/router/src/db/customers.rs

@@ -55,6 +55,16 @@ where
         storage_scheme: MerchantStorageScheme,
     ) -> CustomResult<Option<customer::Customer>, errors::StorageError>;
 
+    #[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "customer_v2")))]
+    async fn find_customer_optional_with_redacted_customer_details_by_customer_id_merchant_id(
+        &self,
+        state: &KeyManagerState,
+        customer_id: &id_type::CustomerId,
+        merchant_id: &id_type::MerchantId,
+        key_store: &domain::MerchantKeyStore,
+        storage_scheme: MerchantStorageScheme,
+    ) -> CustomResult<Option<customer::Customer>, errors::StorageError>;
+
     #[cfg(all(feature = "v2", feature = "customer_v2"))]
     async fn find_optional_by_merchant_id_merchant_reference_id(
         &self,
@@ -242,6 +252,73 @@ mod storage {
             })
         }
 
+        #[instrument(skip_all)]
+        // check customer not found in kv and fallback to db
+        #[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "customer_v2")))]
+        async fn find_customer_optional_with_redacted_customer_details_by_customer_id_merchant_id(
+            &self,
+            state: &KeyManagerState,
+            customer_id: &id_type::CustomerId,
+            merchant_id: &id_type::MerchantId,
+            key_store: &domain::MerchantKeyStore,
+            storage_scheme: MerchantStorageScheme,
+        ) -> CustomResult<Option<customer::Customer>, errors::StorageError> {
+            let conn = connection::pg_connection_read(self).await?;
+            let database_call = || async {
+                storage_types::Customer::find_optional_by_customer_id_merchant_id(
+                    &conn,
+                    customer_id,
+                    merchant_id,
+                )
+                .await
+                .map_err(|err| report!(errors::StorageError::from(err)))
+            };
+            let storage_scheme =
+                decide_storage_scheme::<_, diesel_models::Customer>(self, storage_scheme, Op::Find)
+                    .await;
+            let maybe_customer = match storage_scheme {
+                MerchantStorageScheme::PostgresOnly => database_call().await,
+                MerchantStorageScheme::RedisKv => {
+                    let key = PartitionKey::MerchantIdCustomerId {
+                        merchant_id,
+                        customer_id: customer_id.get_string_repr(),
+                    };
+                    let field = format!("cust_{}", customer_id.get_string_repr());
+                    Box::pin(db_utils::try_redis_get_else_try_database_get(
+                        // check for ValueNotFound
+                        async {
+                            kv_wrapper(
+                                self,
+                                KvOperation::<diesel_models::Customer>::HGet(&field),
+                                key,
+                            )
+                            .await?
+                            .try_into_hget()
+                            .map(Some)
+                        },
+                        database_call,
+                    ))
+                    .await
+                }
+            }?;
+
+            let maybe_result = maybe_customer
+                .async_map(|customer| async {
+                    customer
+                        .convert(
+                            state,
+                            key_store.key.get_inner(),
+                            key_store.merchant_id.clone().into(),
+                        )
+                        .await
+                        .change_context(errors::StorageError::DecryptionError)
+                })
+                .await
+                .transpose()?;
+
+            Ok(maybe_result)
+        }
+
         #[cfg(all(feature = "v2", feature = "customer_v2"))]
         async fn find_optional_by_merchant_id_merchant_reference_id(
             &self,
@@ -938,6 +1015,35 @@ mod storage {
             })
         }
 
+        #[instrument(skip_all)]
+        #[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "customer_v2")))]
+        async fn find_customer_optional_with_redacted_customer_details_by_customer_id_merchant_id(
+            &self,
+            state: &KeyManagerState,
+            customer_id: &id_type::CustomerId,
+            merchant_id: &id_type::MerchantId,
+            key_store: &domain::MerchantKeyStore,
+            _storage_scheme: MerchantStorageScheme,
+        ) -> CustomResult<Option<customer::Customer>, errors::StorageError> {
+            let conn = connection::pg_connection_read(self).await?;
+            let maybe_customer: Option<customer::Customer> =
+                storage_types::Customer::find_optional_by_customer_id_merchant_id(
+                    &conn,
+                    customer_id,
+                    merchant_id,
+                )
+                .await
+                .map_err(|error| report!(errors::StorageError::from(error)))?
+                .async_map(|c| async {
+                    c.convert(state, key_store.key.get_inner(), merchant_id.clone().into())
+                        .await
+                        .change_context(errors::StorageError::DecryptionError)
+                })
+                .await
+                .transpose()?;
+            Ok(maybe_customer)
+        }
+
         #[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "customer_v2")))]
         #[instrument(skip_all)]
         #[cfg(all(feature = "v2", feature = "customer_v2"))]
@@ -1239,6 +1345,37 @@ impl CustomerInterface for MockDb {
             .transpose()
     }
 
+    #[allow(clippy::panic)]
+    #[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "customer_v2")))]
+    async fn find_customer_optional_with_redacted_customer_details_by_customer_id_merchant_id(
+        &self,
+        state: &KeyManagerState,
+        customer_id: &id_type::CustomerId,
+        merchant_id: &id_type::MerchantId,
+        key_store: &domain::MerchantKeyStore,
+        _storage_scheme: MerchantStorageScheme,
+    ) -> CustomResult<Option<customer::Customer>, errors::StorageError> {
+        let customers = self.customers.lock().await;
+        let customer = customers
+            .iter()
+            .find(|customer| {
+                customer.get_customer_id() == *customer_id && &customer.merchant_id == merchant_id
+            })
+            .cloned();
+        customer
+            .async_map(|c| async {
+                c.convert(
+                    state,
+                    key_store.key.get_inner(),
+                    key_store.merchant_id.clone().into(),
+                )
+                .await
+                .change_context(errors::StorageError::DecryptionError)
+            })
+            .await
+            .transpose()
+    }
+
     #[allow(clippy::panic)]
     #[cfg(all(feature = "v2", feature = "customer_v2"))]
     async fn find_optional_by_merchant_id_merchant_reference_id(

crates/router/src/db/kafka_store.rs

@@ -378,6 +378,26 @@ impl CustomerInterface for KafkaStore {
             .await
     }
 
+    #[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "customer_v2")))]
+    async fn find_customer_optional_with_redacted_customer_details_by_customer_id_merchant_id(
+        &self,
+        state: &KeyManagerState,
+        customer_id: &id_type::CustomerId,
+        merchant_id: &id_type::MerchantId,
+        key_store: &domain::MerchantKeyStore,
+        storage_scheme: MerchantStorageScheme,
+    ) -> CustomResult<Option<domain::Customer>, errors::StorageError> {
+        self.diesel_store
+            .find_customer_optional_with_redacted_customer_details_by_customer_id_merchant_id(
+                state,
+                customer_id,
+                merchant_id,
+                key_store,
+                storage_scheme,
+            )
+            .await
+    }
+
     #[cfg(all(feature = "v2", feature = "customer_v2"))]
     async fn find_optional_by_merchant_id_merchant_reference_id(
         &self,