feat(core): [Payouts] Add billing address to payout list (#5004)

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

crates/hyperswitch_domain_models/src/payouts/payouts.rs

@@ -54,7 +54,12 @@ pub trait PayoutsInterface {
         _filters: &PayoutFetchConstraints,
         _storage_scheme: MerchantStorageScheme,
     ) -> error_stack::Result<
-        Vec<(Payouts, PayoutAttempt, Option<diesel_models::Customer>)>,
+        Vec<(
+            Payouts,
+            PayoutAttempt,
+            Option<diesel_models::Customer>,
+            Option<diesel_models::Address>,
+        )>,
         errors::StorageError,
     >;
 

crates/router/src/core/payouts.rs

@@ -6,6 +6,8 @@ pub mod transformers;
 pub mod validator;
 use std::{collections::HashSet, vec::IntoIter};
 
+#[cfg(feature = "olap")]
+use api_models::payments as payment_enums;
 use api_models::{self, enums as api_enums, payouts::PayoutLinkResponse};
 #[cfg(feature = "payout_retry")]
 use common_enums::PayoutRetryType;
@@ -33,6 +35,8 @@ use time::Duration;
 
 #[cfg(feature = "olap")]
 use crate::types::domain::behaviour::Conversion;
+#[cfg(feature = "olap")]
+use crate::types::PayoutActionData;
 use crate::{
     core::{
         errors::{
@@ -770,7 +774,9 @@ pub async fn payouts_list_core(
     .to_not_found_response(errors::ApiErrorResponse::PayoutNotFound)?;
     let payouts = core_utils::filter_objects_based_on_profile_id_list(profile_id_list, payouts);
 
-    let collected_futures = payouts.into_iter().map(|payout| async {
+    let mut pi_pa_tuple_vec = PayoutActionData::new();
+
+    for payout in payouts {
         match db
             .find_payout_attempt_by_merchant_id_payout_attempt_id(
                 merchant_id,
@@ -778,73 +784,80 @@ pub async fn payouts_list_core(
                 storage_enums::MerchantStorageScheme::PostgresOnly,
             )
             .await
+            .change_context(errors::ApiErrorResponse::InternalServerError)
         {
-            Ok(ref payout_attempt) => match payout.customer_id.clone() {
+            Ok(payout_attempt) => {
-                Some(ref customer_id) => {
+                let domain_customer = match payout.customer_id.clone() {
                     #[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "customer_v2")))]
-                    match db
+                    Some(customer_id) => db
                         .find_customer_by_customer_id_merchant_id(
                             &(&state).into(),
-                            customer_id,
+                            &customer_id,
                             merchant_id,
                             &key_store,
                             merchant_account.storage_scheme,
                         )
                         .await
-                    {
+                        .map_err(|err| {
-                        Ok(customer) => Ok((payout, payout_attempt.to_owned(), Some(customer))),
-                        Err(err) => {
                             let err_msg = format!(
                                 "failed while fetching customer for customer_id - {:?}",
                                 customer_id
                             );
                             logger::warn!(?err, err_msg);
-                            if err.current_context().is_db_not_found() {
+                        })
-                                Ok((payout, payout_attempt.to_owned(), None))
+                        .ok(),
-                            } else {
+                    _ => None,
-                                Err(err
+                };
-                                    .change_context(errors::ApiErrorResponse::InternalServerError)
+
-                                    .attach_printable(err_msg))
+                let payout_id_as_payment_id_type =
-                            }
+                    common_utils::id_type::PaymentId::wrap(payout.payout_id.clone())
-                        }
+                        .change_context(errors::ApiErrorResponse::InvalidRequestData {
-                    }
+                            message: "payout_id contains invalid data".to_string(),
-                }
+                        })
-                None => Ok((payout.to_owned(), payout_attempt.to_owned(), None)),
+                        .attach_printable("Error converting payout_id to PaymentId type")?;
-            },
+
+                let payment_addr = payment_helpers::create_or_find_address_for_payment_by_request(
+                    &state,
+                    None,
+                    payout.address_id.as_deref(),
+                    merchant_id,
+                    payout.customer_id.as_ref(),
+                    &key_store,
+                    &payout_id_as_payment_id_type,
+                    merchant_account.storage_scheme,
+                )
+                .await
+                .transpose()
+                .and_then(|addr| {
+                    addr.map_err(|err| {
+                        let err_msg = format!(
+                            "billing_address missing for address_id : {:?}",
+                            payout.address_id
+                        );
+                        logger::warn!(?err, err_msg);
+                    })
+                    .ok()
+                    .map(payment_enums::Address::foreign_from)
+                });
+
+                pi_pa_tuple_vec.push((
+                    payout.to_owned(),
+                    payout_attempt.to_owned(),
+                    domain_customer,
+                    payment_addr,
+                ));
+            }
             Err(err) => {
                 let err_msg = format!(
-                    "failed while fetching payout_attempt for payout_id - {}",
+                    "failed while fetching payout_attempt for payout_id - {:?}",
-                    payout.payout_id.clone(),
+                    payout.payout_id
                 );
                 logger::warn!(?err, err_msg);
-                Err(err
-                    .change_context(errors::ApiErrorResponse::InternalServerError)
-                    .attach_printable(err_msg))
             }
         }
-    });
+    }
-
-    let pi_pa_tuple_vec: Result<
-        Vec<(
-            storage::Payouts,
-            storage::PayoutAttempt,
-            Option<domain::Customer>,
-        )>,
-        _,
-    > = join_all(collected_futures)
-        .await
-        .into_iter()
-        .collect::<Result<
-            Vec<(
-                storage::Payouts,
-                storage::PayoutAttempt,
-                Option<domain::Customer>,
-            )>,
-            _,
-        >>();
 
     let data: Vec<api::PayoutCreateResponse> = pi_pa_tuple_vec
-        .change_context(errors::ApiErrorResponse::InternalServerError)?
         .into_iter()
         .map(ForeignFrom::foreign_from)
         .collect();
@@ -874,6 +887,7 @@ pub async fn payouts_filtered_list_core(
         storage::Payouts,
         storage::PayoutAttempt,
         Option<diesel_models::Customer>,
+        Option<diesel_models::Address>,
     )> = db
         .filter_payouts_and_attempts(
             merchant_account.get_id(),
@@ -883,30 +897,50 @@ pub async fn payouts_filtered_list_core(
         .await
         .to_not_found_response(errors::ApiErrorResponse::PayoutNotFound)?;
     let list = core_utils::filter_objects_based_on_profile_id_list(profile_id_list, list);
-    let data: Vec<api::PayoutCreateResponse> = join_all(list.into_iter().map(|(p, pa, c)| async {
+    let data: Vec<api::PayoutCreateResponse> =
-        let domain_cust = c
+        join_all(list.into_iter().map(|(p, pa, customer, address)| async {
-            .async_and_then(|cust| async {
+            let customer: Option<domain::Customer> = customer
-                domain::Customer::convert_back(
+                .async_and_then(|cust| async {
-                    &(&state).into(),
+                    domain::Customer::convert_back(
-                    cust,
+                        &(&state).into(),
-                    &key_store.key,
+                        cust,
-                    key_store.merchant_id.clone().into(),
+                        &key_store.key,
-                )
+                        key_store.merchant_id.clone().into(),
-                .await
+                    )
-                .map_err(|err| {
+                    .await
-                    let msg = format!("failed to convert customer for id: {:?}", p.customer_id);
+                    .map_err(|err| {
-                    logger::warn!(?err, msg);
+                        let msg = format!("failed to convert customer for id: {:?}", p.customer_id);
+                        logger::warn!(?err, msg);
+                    })
+                    .ok()
                 })
-                .ok()
+                .await;
-            })
+
-            .await;
+            let payout_addr: Option<payment_enums::Address> = address
-        Some((p, pa, domain_cust))
+                .async_and_then(|addr| async {
-    }))
+                    domain::Address::convert_back(
-    .await
+                        &(&state).into(),
-    .into_iter()
+                        addr,
-    .flatten()
+                        &key_store.key,
-    .map(ForeignFrom::foreign_from)
+                        key_store.merchant_id.clone().into(),
-    .collect();
+                    )
+                    .await
+                    .map(ForeignFrom::foreign_from)
+                    .map_err(|err| {
+                        let msg = format!("failed to convert address for id: {:?}", p.address_id);
+                        logger::warn!(?err, msg);
+                    })
+                    .ok()
+                })
+                .await;
+
+            Some((p, pa, customer, payout_addr))
+        }))
+        .await
+        .into_iter()
+        .flatten()
+        .map(ForeignFrom::foreign_from)
+        .collect();
 
     let active_payout_ids = db
         .filter_active_payout_ids_by_constraints(merchant_account.get_id(), &constraints)

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

@@ -9,7 +9,7 @@ use common_utils::link_utils::EnabledPaymentMethod;
 ))]
 use crate::types::transformers::ForeignInto;
 #[cfg(feature = "olap")]
-use crate::types::{domain, storage};
+use crate::types::{api::payments, domain, storage};
 use crate::{
     settings::PayoutRequiredFields,
     types::{api, transformers::ForeignFrom},
@@ -21,6 +21,7 @@ impl
         storage::Payouts,
         storage::PayoutAttempt,
         Option<domain::Customer>,
+        Option<payments::Address>,
     )> for api::PayoutCreateResponse
 {
     fn foreign_from(
@@ -28,6 +29,7 @@ impl
             storage::Payouts,
             storage::PayoutAttempt,
             Option<domain::Customer>,
+            Option<payments::Address>,
         ),
     ) -> Self {
         todo!()
@@ -44,6 +46,7 @@ impl
         storage::Payouts,
         storage::PayoutAttempt,
         Option<domain::Customer>,
+        Option<payments::Address>,
     )> for api::PayoutCreateResponse
 {
     fn foreign_from(
@@ -51,9 +54,10 @@ impl
             storage::Payouts,
             storage::PayoutAttempt,
             Option<domain::Customer>,
+            Option<payments::Address>,
         ),
     ) -> Self {
-        let (payout, payout_attempt, customer) = item;
+        let (payout, payout_attempt, customer, address) = item;
         let attempt = api::PayoutAttemptResponse {
             attempt_id: payout_attempt.payout_attempt_id,
             status: payout_attempt.status,
@@ -95,7 +99,7 @@ impl
             connector_transaction_id: attempt.connector_transaction_id.clone(),
             priority: payout.priority,
             attempts: Some(vec![attempt]),
-            billing: None,
+            billing: address,
             client_secret: None,
             payout_link: None,
             email: customer

crates/router/src/core/utils.rs

@@ -1521,7 +1521,7 @@ impl GetProfileId for storage::Payouts {
     }
 }
 #[cfg(feature = "payouts")]
-impl<T, F> GetProfileId for (storage::Payouts, T, F) {
+impl<T, F, R> GetProfileId for (storage::Payouts, T, F, R) {
     fn get_profile_id(&self) -> Option<&common_utils::id_type::ProfileId> {
         self.0.get_profile_id()
     }

crates/router/src/db/kafka_store.rs

@@ -2073,6 +2073,7 @@ impl PayoutsInterface for KafkaStore {
             storage::Payouts,
             storage::PayoutAttempt,
             Option<diesel_models::Customer>,
+            Option<diesel_models::Address>,
         )>,
         errors::DataStorageError,
     > {

crates/router/src/types.rs

@@ -207,6 +207,14 @@ pub type PayoutsRouterData<F> = RouterData<F, PayoutsData, PayoutsResponseData>;
 pub type PayoutsResponseRouterData<F, R> =
     ResponseRouterData<F, R, PayoutsData, PayoutsResponseData>;
 
+#[cfg(feature = "payouts")]
+pub type PayoutActionData = Vec<(
+    storage::Payouts,
+    storage::PayoutAttempt,
+    Option<domain::Customer>,
+    Option<api_models::payments::Address>,
+)>;
+
 #[cfg(feature = "payouts")]
 pub trait PayoutIndividualDetailsExt {
     type Error;

crates/router/src/types/transformers.rs

@@ -789,6 +789,52 @@ impl<'a> From<&'a domain::Address> for api_types::Address {
     }
 }
 
+impl ForeignFrom<domain::Address> for api_types::Address {
+    fn foreign_from(address: domain::Address) -> Self {
+        // If all the fields of address are none, then pass the address as None
+        let address_details = if address.city.is_none()
+            && address.line1.is_none()
+            && address.line2.is_none()
+            && address.line3.is_none()
+            && address.state.is_none()
+            && address.country.is_none()
+            && address.zip.is_none()
+            && address.first_name.is_none()
+            && address.last_name.is_none()
+        {
+            None
+        } else {
+            Some(api_types::AddressDetails {
+                city: address.city.clone(),
+                country: address.country,
+                line1: address.line1.clone().map(Encryptable::into_inner),
+                line2: address.line2.clone().map(Encryptable::into_inner),
+                line3: address.line3.clone().map(Encryptable::into_inner),
+                state: address.state.clone().map(Encryptable::into_inner),
+                zip: address.zip.clone().map(Encryptable::into_inner),
+                first_name: address.first_name.clone().map(Encryptable::into_inner),
+                last_name: address.last_name.clone().map(Encryptable::into_inner),
+            })
+        };
+
+        // If all the fields of phone are none, then pass the phone as None
+        let phone_details = if address.phone_number.is_none() && address.country_code.is_none() {
+            None
+        } else {
+            Some(api_types::PhoneDetails {
+                number: address.phone_number.clone().map(Encryptable::into_inner),
+                country_code: address.country_code.clone(),
+            })
+        };
+
+        Self {
+            address: address_details,
+            phone: phone_details,
+            email: address.email.clone().map(pii::Email::from),
+        }
+    }
+}
+
 impl
     ForeignFrom<(
         diesel_models::api_keys::ApiKey,

crates/storage_impl/src/mock_db/payouts.rs

@@ -69,8 +69,15 @@ impl PayoutsInterface for MockDb {
         _merchant_id: &common_utils::id_type::MerchantId,
         _filters: &hyperswitch_domain_models::payouts::PayoutFetchConstraints,
         _storage_scheme: storage_enums::MerchantStorageScheme,
-    ) -> CustomResult<Vec<(Payouts, PayoutAttempt, Option<diesel_models::Customer>)>, StorageError>
+    ) -> CustomResult<
-    {
+        Vec<(
+            Payouts,
+            PayoutAttempt,
+            Option<diesel_models::Customer>,
+            Option<diesel_models::Address>,
+        )>,
+        StorageError,
+    > {
         // TODO: Implement function for `MockDb`
         Err(StorageError::MockDbError)?
     }

crates/storage_impl/src/payouts/payouts.rs

@@ -19,12 +19,18 @@ use diesel::{associations::HasTable, ExpressionMethods, NullableExpressionMethod
     not(feature = "customer_v2")
 ))]
 use diesel_models::payout_attempt::PayoutAttempt as DieselPayoutAttempt;
+#[cfg(all(
+    feature = "olap",
+    any(feature = "v1", feature = "v2"),
+    not(feature = "customer_v2")
+))]
+use diesel_models::schema::{
+    address::dsl as add_dsl, customers::dsl as cust_dsl, payout_attempt::dsl as poa_dsl,
+};
 #[cfg(feature = "olap")]
 use diesel_models::{
-    customers::Customer as DieselCustomer,
+    address::Address as DieselAddress, customers::Customer as DieselCustomer,
-    enums as storage_enums,
+    enums as storage_enums, query::generics::db_metrics, schema::payouts::dsl as po_dsl,
-    query::generics::db_metrics,
-    schema::{customers::dsl as cust_dsl, payout_attempt::dsl as poa_dsl, payouts::dsl as po_dsl},
 };
 use diesel_models::{
     enums::MerchantStorageScheme,
@@ -57,8 +63,8 @@ use crate::connection;
     not(feature = "customer_v2")
 ))]
 use crate::store::schema::{
-    customers::all_columns as cust_all_columns, payout_attempt::all_columns as poa_all_columns,
+    address::all_columns as addr_all_columns, customers::all_columns as cust_all_columns,
-    payouts::all_columns as po_all_columns,
+    payout_attempt::all_columns as poa_all_columns, payouts::all_columns as po_all_columns,
 };
 use crate::{
     diesel_error_to_data_error,
@@ -331,8 +337,15 @@ impl<T: DatabaseStore> PayoutsInterface for KVRouterStore<T> {
         merchant_id: &common_utils::id_type::MerchantId,
         filters: &PayoutFetchConstraints,
         storage_scheme: MerchantStorageScheme,
-    ) -> error_stack::Result<Vec<(Payouts, PayoutAttempt, Option<DieselCustomer>)>, StorageError>
+    ) -> error_stack::Result<
-    {
+        Vec<(
+            Payouts,
+            PayoutAttempt,
+            Option<DieselCustomer>,
+            Option<DieselAddress>,
+        )>,
+        StorageError,
+    > {
         self.router_store
             .filter_payouts_and_attempts(merchant_id, filters, storage_scheme)
             .await
@@ -571,8 +584,17 @@ impl<T: DatabaseStore> PayoutsInterface for crate::RouterStore<T> {
         merchant_id: &common_utils::id_type::MerchantId,
         filters: &PayoutFetchConstraints,
         storage_scheme: MerchantStorageScheme,
-    ) -> error_stack::Result<Vec<(Payouts, PayoutAttempt, Option<DieselCustomer>)>, StorageError>
+    ) -> error_stack::Result<
-    {
+        Vec<(
+            Payouts,
+            PayoutAttempt,
+            Option<DieselCustomer>,
+            Option<DieselAddress>,
+        )>,
+        StorageError,
+    > {
+        use common_utils::errors::ReportSwitchExt;
+
         let conn = connection::pg_connection_read(self).await.switch()?;
         let conn = async_bb8_diesel::Connection::as_async_conn(&conn);
         let mut query = DieselPayouts::table()
@@ -585,6 +607,10 @@ impl<T: DatabaseStore> PayoutsInterface for crate::RouterStore<T> {
                     .on(cust_dsl::customer_id.nullable().eq(po_dsl::customer_id)),
             )
             .filter(cust_dsl::merchant_id.eq(merchant_id.to_owned()))
+            .left_outer_join(
+                diesel_models::schema::address::table
+                    .on(add_dsl::address_id.nullable().eq(po_dsl::address_id)),
+            )
             .filter(po_dsl::merchant_id.eq(merchant_id.to_owned()))
             .order(po_dsl::created_at.desc())
             .into_boxed();
@@ -675,17 +701,28 @@ impl<T: DatabaseStore> PayoutsInterface for crate::RouterStore<T> {
         logger::debug!(filter = %diesel::debug_query::<diesel::pg::Pg,_>(&query).to_string());
 
         query
-            .select((po_all_columns, poa_all_columns, cust_all_columns.nullable()))
+            .select((
-            .get_results_async::<(DieselPayouts, DieselPayoutAttempt, Option<DieselCustomer>)>(conn)
+                po_all_columns,
+                poa_all_columns,
+                cust_all_columns.nullable(),
+                addr_all_columns.nullable(),
+            ))
+            .get_results_async::<(
+                DieselPayouts,
+                DieselPayoutAttempt,
+                Option<DieselCustomer>,
+                Option<DieselAddress>,
+            )>(conn)
             .await
             .map(|results| {
                 results
                     .into_iter()
-                    .map(|(pi, pa, c)| {
+                    .map(|(pi, pa, c, add)| {
                         (
                             Payouts::from_storage_model(pi),
                             PayoutAttempt::from_storage_model(pa),
                             c,
+                            add,
                         )
                     })
                     .collect()
@@ -706,8 +743,15 @@ impl<T: DatabaseStore> PayoutsInterface for crate::RouterStore<T> {
         _merchant_id: &common_utils::id_type::MerchantId,
         _filters: &PayoutFetchConstraints,
         _storage_scheme: MerchantStorageScheme,
-    ) -> error_stack::Result<Vec<(Payouts, PayoutAttempt, Option<DieselCustomer>)>, StorageError>
+    ) -> error_stack::Result<
-    {
+        Vec<(
+            Payouts,
+            PayoutAttempt,
+            Option<DieselCustomer>,
+            Option<DieselAddress>,
+        )>,
+        StorageError,
+    > {
         todo!()
     }