refactor(customers): add offset and limit to customers list (#5741)

crates/api_models/src/customers.rs

@@ -48,6 +48,16 @@ pub struct CustomerRequest {
     pub metadata: Option<pii::SecretSerdeValue>,
 }
 
+#[derive(Debug, Default, Clone, Deserialize, Serialize, ToSchema)]
+pub struct CustomerListRequest {
+    /// Offset
+    #[schema(example = 32)]
+    pub offset: Option<u32>,
+    /// Limit
+    #[schema(example = 32)]
+    pub limit: Option<u16>,
+}
+
 #[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "customer_v2")))]
 impl CustomerRequest {
     pub fn get_merchant_reference_id(&self) -> Option<id_type::CustomerId> {

crates/api_models/src/events.rs

@@ -19,6 +19,7 @@ use common_utils::{
     impl_api_event_type,
 };
 
+use crate::customers::CustomerListRequest;
 #[allow(unused_imports)]
 use crate::{
     admin::*,
@@ -131,7 +132,8 @@ impl_api_event_type!(
         GetDisputeMetricRequest,
         OrganizationResponse,
         OrganizationRequest,
-        OrganizationId
+        OrganizationId,
+        CustomerListRequest
     )
 );
 

crates/diesel_models/src/query/customers.rs

@@ -23,6 +23,11 @@ impl CustomerNew {
     }
 }
 
+pub struct CustomerListConstraints {
+    pub limit: i64,
+    pub offset: Option<i64>,
+}
+
 // #[cfg(all(feature = "v2", feature = "customer_v2"))]
 impl Customer {
     #[cfg(all(feature = "v2", feature = "customer_v2"))]
@@ -57,12 +62,13 @@ impl Customer {
     pub async fn list_by_merchant_id(
         conn: &PgPooledConn,
         merchant_id: &id_type::MerchantId,
+        constraints: CustomerListConstraints,
     ) -> StorageResult<Vec<Self>> {
         generics::generic_filter::<<Self as HasTable>::Table, _, _, _>(
             conn,
             dsl::merchant_id.eq(merchant_id.to_owned()),
-            None,
-            None,
+            Some(constraints.limit),
+            constraints.offset,
             Some(dsl::created_at),
         )
         .await

crates/router/src/consts.rs

@@ -32,6 +32,8 @@ pub const DEFAULT_SESSION_EXPIRY: i64 = 15 * 60;
 /// The length of a merchant fingerprint secret
 pub const FINGERPRINT_SECRET_LENGTH: usize = 64;
 
+pub const DEFAULT_LIST_API_LIMIT: u16 = 10;
+
 // String literals
 pub(crate) const UNSUPPORTED_ERROR_MESSAGE: &str = "Unsupported response type";
 pub(crate) const LOW_BALANCE_ERROR_MESSAGE: &str = "Insufficient balance in the payment method";

crates/router/src/core/customers.rs

@@ -497,11 +497,24 @@ pub async fn list_customers(
     merchant_id: id_type::MerchantId,
     _profile_id_list: Option<Vec<String>>,
     key_store: domain::MerchantKeyStore,
+    request: customers::CustomerListRequest,
 ) -> errors::CustomerResponse<Vec<customers::CustomerResponse>> {
     let db = state.store.as_ref();
 
+    let customer_list_constraints = crate::db::customers::CustomerListConstraints {
+        limit: request
+            .limit
+            .unwrap_or(crate::consts::DEFAULT_LIST_API_LIMIT),
+        offset: request.offset,
+    };
+
     let domain_customers = db
-        .list_customers_by_merchant_id(&(&state).into(), &merchant_id, &key_store)
+        .list_customers_by_merchant_id(
+            &(&state).into(),
+            &merchant_id,
+            &key_store,
+            customer_list_constraints,
+        )
         .await
         .switch()?;
 

crates/router/src/core/locker_migration.rs

@@ -30,6 +30,8 @@ pub async fn rust_locker_migration(
     state: SessionState,
     merchant_id: &id_type::MerchantId,
 ) -> CustomResult<services::ApplicationResponse<MigrateCardResponse>, errors::ApiErrorResponse> {
+    use crate::db::customers::CustomerListConstraints;
+
     let db = state.store.as_ref();
     let key_manager_state = &(&state).into();
     let key_store = state
@@ -48,8 +50,14 @@ pub async fn rust_locker_migration(
         .to_not_found_response(errors::ApiErrorResponse::MerchantAccountNotFound)
         .change_context(errors::ApiErrorResponse::InternalServerError)?;
 
+    // Handle cases where the number of customers is greater than the limit
+    let constraints = CustomerListConstraints {
+        limit: u16::MAX,
+        offset: None,
+    };
+
     let domain_customers = db
-        .list_customers_by_merchant_id(key_manager_state, merchant_id, &key_store)
+        .list_customers_by_merchant_id(key_manager_state, merchant_id, &key_store, constraints)
         .await
         .change_context(errors::ApiErrorResponse::InternalServerError)?;
 

crates/router/src/db/customers.rs

@@ -1,4 +1,5 @@
 use common_utils::{ext_traits::AsyncExt, id_type, types::keymanager::KeyManagerState};
+use diesel_models::query::customers::CustomerListConstraints as DieselCustomerListConstraints;
 use error_stack::ResultExt;
 #[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "customer_v2")))]
 use futures::future::try_join_all;
@@ -18,6 +19,20 @@ use crate::{
     },
 };
 
+pub struct CustomerListConstraints {
+    pub limit: u16,
+    pub offset: Option<u32>,
+}
+
+impl From<CustomerListConstraints> for DieselCustomerListConstraints {
+    fn from(value: CustomerListConstraints) -> Self {
+        Self {
+            limit: i64::from(value.limit),
+            offset: value.offset.map(i64::from),
+        }
+    }
+}
+
 #[async_trait::async_trait]
 pub trait CustomerInterface
 where
@@ -90,6 +105,7 @@ where
         state: &KeyManagerState,
         merchant_id: &id_type::MerchantId,
         key_store: &domain::MerchantKeyStore,
+        constraints: CustomerListConstraints,
     ) -> CustomResult<Vec<customer::Customer>, errors::StorageError>;
 
     async fn insert_customer(
@@ -518,13 +534,20 @@ mod storage {
             state: &KeyManagerState,
             merchant_id: &id_type::MerchantId,
             key_store: &domain::MerchantKeyStore,
+            constraints: super::CustomerListConstraints,
         ) -> CustomResult<Vec<customer::Customer>, errors::StorageError> {
             let conn = connection::pg_connection_read(self).await?;
 
-            let encrypted_customers =
-                storage_types::Customer::list_by_merchant_id(&conn, merchant_id)
-                    .await
-                    .map_err(|error| report!(errors::StorageError::from(error)))?;
+            let customer_list_constraints =
+                diesel_models::query::customers::CustomerListConstraints::from(constraints);
+
+            let encrypted_customers = storage_types::Customer::list_by_merchant_id(
+                &conn,
+                merchant_id,
+                customer_list_constraints,
+            )
+            .await
+            .map_err(|error| report!(errors::StorageError::from(error)))?;
 
             let customers = try_join_all(encrypted_customers.into_iter().map(
                 |encrypted_customer| async {
@@ -1059,13 +1082,20 @@ mod storage {
             state: &KeyManagerState,
             merchant_id: &id_type::MerchantId,
             key_store: &domain::MerchantKeyStore,
+            constraints: super::CustomerListConstraints,
         ) -> CustomResult<Vec<customer::Customer>, errors::StorageError> {
             let conn = connection::pg_connection_read(self).await?;
 
-            let encrypted_customers =
-                storage_types::Customer::list_by_merchant_id(&conn, merchant_id)
-                    .await
-                    .map_err(|error| report!(errors::StorageError::from(error)))?;
+            let customer_list_constraints =
+                diesel_models::query::customers::CustomerListConstraints::from(constraints);
+
+            let encrypted_customers = storage_types::Customer::list_by_merchant_id(
+                &conn,
+                merchant_id,
+                customer_list_constraints,
+            )
+            .await
+            .map_err(|error| report!(errors::StorageError::from(error)))?;
 
             let customers = try_join_all(encrypted_customers.into_iter().map(
                 |encrypted_customer| async {
@@ -1250,13 +1280,16 @@ impl CustomerInterface for MockDb {
         state: &KeyManagerState,
         merchant_id: &id_type::MerchantId,
         key_store: &domain::MerchantKeyStore,
+        constraints: CustomerListConstraints,
     ) -> CustomResult<Vec<customer::Customer>, errors::StorageError> {
         let customers = self.customers.lock().await;
 
         let customers = try_join_all(
             customers
                 .iter()
                 .filter(|customer| customer.merchant_id == *merchant_id)
+                .take(usize::from(constraints.limit))
+                .skip(usize::try_from(constraints.offset.unwrap_or(0)).unwrap_or(0))
                 .map(|customer| async {
                     customer
                         .to_owned()

crates/router/src/db/kafka_store.rs

@@ -449,9 +449,10 @@ impl CustomerInterface for KafkaStore {
         state: &KeyManagerState,
         merchant_id: &id_type::MerchantId,
         key_store: &domain::MerchantKeyStore,
+        constraints: super::customers::CustomerListConstraints,
     ) -> CustomResult<Vec<domain::Customer>, errors::StorageError> {
         self.diesel_store
-            .list_customers_by_merchant_id(state, merchant_id, key_store)
+            .list_customers_by_merchant_id(state, merchant_id, key_store, constraints)
             .await
     }
 

crates/router/src/routes/customers.rs

@@ -112,20 +112,26 @@ pub async fn customers_retrieve(
 
 #[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "customer_v2")))]
 #[instrument(skip_all, fields(flow = ?Flow::CustomersList))]
-pub async fn customers_list(state: web::Data<AppState>, req: HttpRequest) -> HttpResponse {
+pub async fn customers_list(
+    state: web::Data<AppState>,
+    req: HttpRequest,
+    query: web::Query<customers::CustomerListRequest>,
+) -> HttpResponse {
     let flow = Flow::CustomersList;
+    let payload = query.into_inner();
 
     api::server_wrap(
         flow,
         state,
         &req,
-        (),
-        |state, auth, _, _| {
+        payload,
+        |state, auth, request, _| {
             list_customers(
                 state,
                 auth.merchant_account.get_id().to_owned(),
                 None,
                 auth.key_store,
+                request,
             )
         },
         auth::auth_type(

crates/router/src/types/api/customers.rs

@@ -2,7 +2,8 @@ use api_models::customers;
 #[cfg(all(feature = "v2", feature = "customer_v2"))]
 pub use api_models::customers::GlobalId;
 pub use api_models::customers::{
-    CustomerDeleteResponse, CustomerId, CustomerRequest, CustomerUpdateRequest, UpdateCustomerId,
+    CustomerDeleteResponse, CustomerId, CustomerListRequest, CustomerRequest,
+    CustomerUpdateRequest, UpdateCustomerId,
 };
 #[cfg(all(feature = "v2", feature = "customer_v2"))]
 use hyperswitch_domain_models::customer;