feat(customers): add customer list endpoint (#2564)

Co-authored-by: Bernard Eugine <114725419+bernard-eugine@users.noreply.github.com>

crates/diesel_models/src/query/customers.rs

@@ -73,6 +73,21 @@ impl Customer {
         .await
     }
 
+    #[instrument(skip(conn))]
+    pub async fn list_by_merchant_id(
+        conn: &PgPooledConn,
+        merchant_id: &str,
+    ) -> StorageResult<Vec<Self>> {
+        generics::generic_filter::<<Self as HasTable>::Table, _, _, _>(
+            conn,
+            dsl::merchant_id.eq(merchant_id.to_owned()),
+            None,
+            None,
+            Some(dsl::created_at),
+        )
+        .await
+    }
+
     #[instrument(skip(conn))]
     pub async fn find_optional_by_customer_id_merchant_id(
         conn: &PgPooledConn,

crates/router/src/compatibility/stripe/errors.rs

@@ -69,6 +69,9 @@ pub enum StripeErrorCode {
     #[error(error_type = StripeErrorType::InvalidRequestError, code = "customer_redacted", message = "Customer has redacted")]
     CustomerRedacted,
 
+    #[error(error_type = StripeErrorType::InvalidRequestError, code = "customer_already_exists", message = "Customer with the given customer_id already exists")]
+    DuplicateCustomer,
+
     #[error(error_type = StripeErrorType::InvalidRequestError, code = "resource_missing", message = "No such refund")]
     RefundNotFound,
 
@@ -652,6 +655,7 @@ impl actix_web::ResponseError for StripeErrorCode {
             | Self::FileNotAvailable
             | Self::FileProviderNotSupported
             | Self::CurrencyNotSupported { .. }
+            | Self::DuplicateCustomer
             | Self::PaymentMethodUnactivated => StatusCode::BAD_REQUEST,
             Self::RefundFailed
             | Self::PayoutFailed
@@ -730,6 +734,7 @@ impl ErrorSwitch<StripeErrorCode> for CustomersErrorResponse {
             Self::InternalServerError => SC::InternalServerError,
             Self::MandateActive => SC::MandateActive,
             Self::CustomerNotFound => SC::CustomerNotFound,
+            Self::CustomerAlreadyExists => SC::DuplicateCustomer,
         }
     }
 }

crates/router/src/core/customers.rs

@@ -2,13 +2,13 @@ use common_utils::{
     crypto::{Encryptable, GcmAes256},
     errors::ReportSwitchExt,
 };
-use error_stack::ResultExt;
+use error_stack::{IntoReport, ResultExt};
 use masking::ExposeInterface;
 use router_env::{instrument, tracing};
 
 use crate::{
     core::{
-        errors::{self},
+        errors::{self, StorageErrorExt},
         payment_methods::cards,
     },
     pii::PeekInterface,
@@ -39,6 +39,25 @@ pub async fn create_customer(
     let merchant_id = &merchant_account.merchant_id;
     customer_data.merchant_id = merchant_id.to_owned();
 
+    // We first need to validate whether the customer with the given customer id already exists
+    // this may seem like a redundant db call, as the insert_customer will anyway return this error
+    //
+    // Consider a scenerio where the address is inserted and then when inserting the customer,
+    // it errors out, now the address that was inserted is not deleted
+    match db
+        .find_customer_by_customer_id_merchant_id(customer_id, merchant_id, &key_store)
+        .await
+    {
+        Err(err) => {
+            if !err.current_context().is_db_not_found() {
+                Err(err).switch()
+            } else {
+                Ok(())
+            }
+        }
+        Ok(_) => Err(errors::CustomersErrorResponse::CustomerAlreadyExists).into_report(),
+    }?;
+
     let key = key_store.key.get_inner().peek();
     let address = if let Some(addr) = &customer_data.address {
         let customer_address: api_models::payments::AddressDetails = addr.clone();
@@ -89,23 +108,10 @@ pub async fn create_customer(
     .switch()
     .attach_printable("Failed while encrypting Customer")?;
 
-    let customer = match db.insert_customer(new_customer, &key_store).await {
-        Ok(customer) => customer,
-        Err(error) => {
-            if error.current_context().is_db_unique_violation() {
-                db.find_customer_by_customer_id_merchant_id(customer_id, merchant_id, &key_store)
+    let customer = db
+        .insert_customer(new_customer, &key_store)
         .await
-                    .switch()
-                    .attach_printable(format!(
-                        "Failed while fetching Customer, customer_id: {customer_id}",
-                    ))?
-            } else {
-                Err(error
-                    .change_context(errors::CustomersErrorResponse::InternalServerError)
-                    .attach_printable("Failed while inserting new customer"))?
-            }
-        }
-    };
+        .to_duplicate_response(errors::CustomersErrorResponse::CustomerAlreadyExists)?;
 
     let address_details = address.map(api_models::payments::AddressDetails::from);
 
@@ -143,6 +149,27 @@ pub async fn retrieve_customer(
     ))
 }
 
+#[instrument(skip(state))]
+pub async fn list_customers(
+    state: AppState,
+    merchant_id: String,
+    key_store: domain::MerchantKeyStore,
+) -> errors::CustomerResponse<Vec<customers::CustomerResponse>> {
+    let db = state.store.as_ref();
+
+    let domain_customers = db
+        .list_customers_by_merchant_id(&merchant_id, &key_store)
+        .await
+        .switch()?;
+
+    let customers = domain_customers
+        .into_iter()
+        .map(|domain_customer| customers::CustomerResponse::from((domain_customer, None)))
+        .collect();
+
+    Ok(services::ApplicationResponse::Json(customers))
+}
+
 #[instrument(skip_all)]
 pub async fn delete_customer(
     state: AppState,

crates/router/src/core/errors/customers_error_response.rs

@@ -13,6 +13,9 @@ pub enum CustomersErrorResponse {
 
     #[error("Customer does not exist in our records")]
     CustomerNotFound,
+
+    #[error("Customer with the given customer id already exists")]
+    CustomerAlreadyExists,
 }
 
 impl actix_web::ResponseError for CustomersErrorResponse {

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

@@ -314,6 +314,12 @@ impl ErrorSwitch<api_models::errors::types::ApiErrorResponse> for CustomersError
                 "Customer does not exist in our records",
                 None,
             )),
+            Self::CustomerAlreadyExists => AER::BadRequest(ApiError::new(
+                "IR",
+                12,
+                "Customer with the given `customer_id` already exists",
+                None,
+            )),
         }
     }
 }

crates/router/src/db/customers.rs

@@ -1,5 +1,6 @@
 use common_utils::ext_traits::AsyncExt;
 use error_stack::{IntoReport, ResultExt};
+use futures::future::try_join_all;
 use masking::PeekInterface;
 use router_env::{instrument, tracing};
 
@@ -52,6 +53,12 @@ where
         key_store: &domain::MerchantKeyStore,
     ) -> CustomResult<domain::Customer, errors::StorageError>;
 
+    async fn list_customers_by_merchant_id(
+        &self,
+        merchant_id: &str,
+        key_store: &domain::MerchantKeyStore,
+    ) -> CustomResult<Vec<domain::Customer>, errors::StorageError>;
+
     async fn insert_customer(
         &self,
         customer_data: domain::Customer,
@@ -148,6 +155,31 @@ impl CustomerInterface for Store {
         }
     }
 
+    async fn list_customers_by_merchant_id(
+        &self,
+        merchant_id: &str,
+        key_store: &domain::MerchantKeyStore,
+    ) -> CustomResult<Vec<domain::Customer>, errors::StorageError> {
+        let conn = connection::pg_connection_read(self).await?;
+
+        let encrypted_customers = storage::Customer::list_by_merchant_id(&conn, merchant_id)
+            .await
+            .map_err(Into::into)
+            .into_report()?;
+
+        let customers = try_join_all(encrypted_customers.into_iter().map(
+            |encrypted_customer| async {
+                encrypted_customer
+                    .convert(key_store.key.get_inner())
+                    .await
+                    .change_context(errors::StorageError::DecryptionError)
+            },
+        ))
+        .await?;
+
+        Ok(customers)
+    }
+
     async fn insert_customer(
         &self,
         customer_data: domain::Customer,
@@ -209,6 +241,30 @@ impl CustomerInterface for MockDb {
             .transpose()
     }
 
+    async fn list_customers_by_merchant_id(
+        &self,
+        merchant_id: &str,
+        key_store: &domain::MerchantKeyStore,
+    ) -> CustomResult<Vec<domain::Customer>, errors::StorageError> {
+        let customers = self.customers.lock().await;
+
+        let customers = try_join_all(
+            customers
+                .iter()
+                .filter(|customer| customer.merchant_id == merchant_id)
+                .map(|customer| async {
+                    customer
+                        .to_owned()
+                        .convert(key_store.key.get_inner())
+                        .await
+                        .change_context(errors::StorageError::DecryptionError)
+                }),
+        )
+        .await?;
+
+        Ok(customers)
+    }
+
     #[instrument(skip_all)]
     async fn update_customer_by_customer_id_merchant_id(
         &self,

crates/router/src/openapi.rs

@@ -101,6 +101,7 @@ Never share your secret api keys. Keep them guarded and secure.
         crate::routes::customers::customers_retrieve,
         crate::routes::customers::customers_update,
         crate::routes::customers::customers_delete,
+        crate::routes::customers::customers_list,
         // crate::routes::api_keys::api_key_create,
         // crate::routes::api_keys::api_key_retrieve,
         // crate::routes::api_keys::api_key_update,

crates/router/src/routes/app.rs

@@ -275,10 +275,12 @@ impl Customers {
 
         #[cfg(feature = "olap")]
         {
-            route = route.service(
+            route = route
+                .service(
                     web::resource("/{customer_id}/mandates")
                         .route(web::get().to(get_customer_mandates)),
-            );
+                )
+                .service(web::resource("/list").route(web::get().to(customers_list)))
         }
 
         #[cfg(feature = "oltp")]
@@ -300,6 +302,7 @@ impl Customers {
                         .route(web::delete().to(customers_delete)),
                 );
         }
+
         route
     }
 }

crates/router/src/routes/customers.rs

@@ -85,6 +85,37 @@ pub async fn customers_retrieve(
     )
     .await
 }
+
+/// List customers for a merchant
+///
+/// To filter and list the customers for a particular merchant id
+#[utoipa::path(
+    post,
+    path = "/customers/list",
+    responses(
+        (status = 200, description = "Customers retrieved", body = Vec<CustomerResponse>),
+        (status = 400, description = "Invalid Data"),
+    ),
+    tag = "Customers List",
+    operation_id = "List all Customers for a Merchant",
+    security(("api_key" = []))
+)]
+#[instrument(skip_all, fields(flow = ?Flow::CustomersList))]
+pub async fn customers_list(state: web::Data<AppState>, req: HttpRequest) -> HttpResponse {
+    let flow = Flow::CustomersList;
+
+    api::server_wrap(
+        flow,
+        state,
+        &req,
+        (),
+        |state, auth, _| list_customers(state, auth.merchant_account.merchant_id, auth.key_store),
+        &auth::ApiKeyAuth,
+        api_locking::LockAction::NotApplicable,
+    )
+    .await
+}
+
 /// Update Customer
 ///
 /// Updates the customer's details in a customer object.

crates/router/src/routes/lock_utils.rs

@@ -47,7 +47,8 @@ impl From<Flow> for ApiIdentifier {
             | Flow::CustomersRetrieve
             | Flow::CustomersUpdate
             | Flow::CustomersDelete
-            | Flow::CustomersGetMandates => Self::Customers,
+            | Flow::CustomersGetMandates
+            | Flow::CustomersList => Self::Customers,
 
             Flow::EphemeralKeyCreate | Flow::EphemeralKeyDelete => Self::Ephemeral,
 

crates/router_env/src/logger/types.rs

@@ -104,6 +104,8 @@ pub enum Flow {
     PaymentMethodsList,
     /// Customer payment methods list flow.
     CustomerPaymentMethodsList,
+    /// List Customers for a merchant
+    CustomersList,
     /// Payment methods retrieve flow.
     PaymentMethodsRetrieve,
     /// Payment methods update flow.

openapi/openapi_spec.json

@@ -169,6 +169,39 @@
         ]
       }
     },
+    "/customers/list": {
+      "post": {
+        "tags": [
+          "Customers List"
+        ],
+        "summary": "List customers for a merchant",
+        "description": "List customers for a merchant\n\nTo filter and list the customers for a particular merchant id",
+        "operationId": "List all Customers for a Merchant",
+        "responses": {
+          "200": {
+            "description": "Customers retrieved",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "type": "array",
+                  "items": {
+                    "$ref": "#/components/schemas/CustomerResponse"
+                  }
+                }
+              }
+            }
+          },
+          "400": {
+            "description": "Invalid Data"
+          }
+        },
+        "security": [
+          {
+            "api_key": []
+          }
+        ]
+      }
+    },
     "/customers/payment_methods": {
       "get": {
         "tags": [