feat(user): generate and delete sample data (#2987)

Co-authored-by: Rachit Naithani <rachit.naithani@juspay.in> Co-authored-by: Mani Chandra Dulam <mani.dchandra@juspay.in>
2025-11-03 21:37:41 +08:00 · 2023-12-01 16:00:25 +05:30
parent c4bd47eca9
commit 092ec73b3c
22 changed files with 1152 additions and 19 deletions
--- a/crates/router/src/db/kafka_store.rs
+++ b/crates/router/src/db/kafka_store.rs
@ -23,7 +23,8 @@ use storage_impl::redis::kv_store::RedisConnInterface;
 use time::PrimitiveDateTime;

 use super::{
-    dashboard_metadata::DashboardMetadataInterface, user::UserInterface,
+    dashboard_metadata::DashboardMetadataInterface,
+    user::{sample_data::BatchSampleDataInterface, UserInterface},
    user_role::UserRoleInterface,
 };
 use crate::{
@ -1951,3 +1952,118 @@ impl DashboardMetadataInterface for KafkaStore {
            .await
    }
 }
+
+#[async_trait::async_trait]
+impl BatchSampleDataInterface for KafkaStore {
+    async fn insert_payment_intents_batch_for_sample_data(
+        &self,
+        batch: Vec<data_models::payments::payment_intent::PaymentIntentNew>,
+    ) -> CustomResult<Vec<data_models::payments::PaymentIntent>, data_models::errors::StorageError>
+    {
+        let payment_intents_list = self
+            .diesel_store
+            .insert_payment_intents_batch_for_sample_data(batch)
+            .await?;
+
+        for payment_intent in payment_intents_list.iter() {
+            let _ = self
+                .kafka_producer
+                .log_payment_intent(payment_intent, None)
+                .await;
+        }
+        Ok(payment_intents_list)
+    }
+
+    async fn insert_payment_attempts_batch_for_sample_data(
+        &self,
+        batch: Vec<diesel_models::user::sample_data::PaymentAttemptBatchNew>,
+    ) -> CustomResult<
+        Vec<data_models::payments::payment_attempt::PaymentAttempt>,
+        data_models::errors::StorageError,
+    > {
+        let payment_attempts_list = self
+            .diesel_store
+            .insert_payment_attempts_batch_for_sample_data(batch)
+            .await?;
+
+        for payment_attempt in payment_attempts_list.iter() {
+            let _ = self
+                .kafka_producer
+                .log_payment_attempt(payment_attempt, None)
+                .await;
+        }
+        Ok(payment_attempts_list)
+    }
+
+    async fn insert_refunds_batch_for_sample_data(
+        &self,
+        batch: Vec<diesel_models::RefundNew>,
+    ) -> CustomResult<Vec<diesel_models::Refund>, data_models::errors::StorageError> {
+        let refunds_list = self
+            .diesel_store
+            .insert_refunds_batch_for_sample_data(batch)
+            .await?;
+
+        for refund in refunds_list.iter() {
+            let _ = self.kafka_producer.log_refund(refund, None).await;
+        }
+        Ok(refunds_list)
+    }
+
+    async fn delete_payment_intents_for_sample_data(
+        &self,
+        merchant_id: &str,
+    ) -> CustomResult<Vec<data_models::payments::PaymentIntent>, data_models::errors::StorageError>
+    {
+        let payment_intents_list = self
+            .diesel_store
+            .delete_payment_intents_for_sample_data(merchant_id)
+            .await?;
+
+        for payment_intent in payment_intents_list.iter() {
+            let _ = self
+                .kafka_producer
+                .log_payment_intent_delete(payment_intent)
+                .await;
+        }
+        Ok(payment_intents_list)
+    }
+
+    async fn delete_payment_attempts_for_sample_data(
+        &self,
+        merchant_id: &str,
+    ) -> CustomResult<
+        Vec<data_models::payments::payment_attempt::PaymentAttempt>,
+        data_models::errors::StorageError,
+    > {
+        let payment_attempts_list = self
+            .diesel_store
+            .delete_payment_attempts_for_sample_data(merchant_id)
+            .await?;
+
+        for payment_attempt in payment_attempts_list.iter() {
+            let _ = self
+                .kafka_producer
+                .log_payment_attempt_delete(payment_attempt)
+                .await;
+        }
+
+        Ok(payment_attempts_list)
+    }
+
+    async fn delete_refunds_for_sample_data(
+        &self,
+        merchant_id: &str,
+    ) -> CustomResult<Vec<diesel_models::Refund>, data_models::errors::StorageError> {
+        let refunds_list = self
+            .diesel_store
+            .delete_refunds_for_sample_data(merchant_id)
+            .await?;
+
+        for refund in refunds_list.iter() {
+            let _ = self.kafka_producer.log_refund_delete(refund).await;
+        }
+
+        Ok(refunds_list)
+    }
+}
--- a/crates/router/src/db/user.rs
+++ b/crates/router/src/db/user.rs
@ -8,6 +8,7 @@ use crate::{
    core::errors::{self, CustomResult},
    services::Store,
 };
+pub mod sample_data;

 #[async_trait::async_trait]
 pub trait UserInterface {
--- a/crates/router/src/db/user/sample_data.rs
+++ b/crates/router/src/db/user/sample_data.rs
@ -0,0 +1,205 @@
+use data_models::{
+    errors::StorageError,
+    payments::{payment_attempt::PaymentAttempt, payment_intent::PaymentIntentNew, PaymentIntent},
+};
+use diesel_models::{
+    errors::DatabaseError,
+    query::user::sample_data as sample_data_queries,
+    refund::{Refund, RefundNew},
+    user::sample_data::PaymentAttemptBatchNew,
+};
+use error_stack::{Report, ResultExt};
+use storage_impl::DataModelExt;
+
+use crate::{connection::pg_connection_write, core::errors::CustomResult, services::Store};
+
+#[async_trait::async_trait]
+pub trait BatchSampleDataInterface {
+    async fn insert_payment_intents_batch_for_sample_data(
+        &self,
+        batch: Vec<PaymentIntentNew>,
+    ) -> CustomResult<Vec<PaymentIntent>, StorageError>;
+
+    async fn insert_payment_attempts_batch_for_sample_data(
+        &self,
+        batch: Vec<PaymentAttemptBatchNew>,
+    ) -> CustomResult<Vec<PaymentAttempt>, StorageError>;
+
+    async fn insert_refunds_batch_for_sample_data(
+        &self,
+        batch: Vec<RefundNew>,
+    ) -> CustomResult<Vec<Refund>, StorageError>;
+
+    async fn delete_payment_intents_for_sample_data(
+        &self,
+        merchant_id: &str,
+    ) -> CustomResult<Vec<PaymentIntent>, StorageError>;
+
+    async fn delete_payment_attempts_for_sample_data(
+        &self,
+        merchant_id: &str,
+    ) -> CustomResult<Vec<PaymentAttempt>, StorageError>;
+
+    async fn delete_refunds_for_sample_data(
+        &self,
+        merchant_id: &str,
+    ) -> CustomResult<Vec<Refund>, StorageError>;
+}
+
+#[async_trait::async_trait]
+impl BatchSampleDataInterface for Store {
+    async fn insert_payment_intents_batch_for_sample_data(
+        &self,
+        batch: Vec<PaymentIntentNew>,
+    ) -> CustomResult<Vec<PaymentIntent>, StorageError> {
+        let conn = pg_connection_write(self)
+            .await
+            .change_context(StorageError::DatabaseConnectionError)?;
+        let new_intents = batch.into_iter().map(|i| i.to_storage_model()).collect();
+        sample_data_queries::insert_payment_intents(&conn, new_intents)
+            .await
+            .map_err(diesel_error_to_data_error)
+            .map(|v| {
+                v.into_iter()
+                    .map(PaymentIntent::from_storage_model)
+                    .collect()
+            })
+    }
+
+    async fn insert_payment_attempts_batch_for_sample_data(
+        &self,
+        batch: Vec<PaymentAttemptBatchNew>,
+    ) -> CustomResult<Vec<PaymentAttempt>, StorageError> {
+        let conn = pg_connection_write(self)
+            .await
+            .change_context(StorageError::DatabaseConnectionError)?;
+        sample_data_queries::insert_payment_attempts(&conn, batch)
+            .await
+            .map_err(diesel_error_to_data_error)
+            .map(|res| {
+                res.into_iter()
+                    .map(PaymentAttempt::from_storage_model)
+                    .collect()
+            })
+    }
+    async fn insert_refunds_batch_for_sample_data(
+        &self,
+        batch: Vec<RefundNew>,
+    ) -> CustomResult<Vec<Refund>, StorageError> {
+        let conn = pg_connection_write(self)
+            .await
+            .change_context(StorageError::DatabaseConnectionError)?;
+        sample_data_queries::insert_refunds(&conn, batch)
+            .await
+            .map_err(diesel_error_to_data_error)
+    }
+
+    async fn delete_payment_intents_for_sample_data(
+        &self,
+        merchant_id: &str,
+    ) -> CustomResult<Vec<PaymentIntent>, StorageError> {
+        let conn = pg_connection_write(self)
+            .await
+            .change_context(StorageError::DatabaseConnectionError)?;
+        sample_data_queries::delete_payment_intents(&conn, merchant_id)
+            .await
+            .map_err(diesel_error_to_data_error)
+            .map(|v| {
+                v.into_iter()
+                    .map(PaymentIntent::from_storage_model)
+                    .collect()
+            })
+    }
+
+    async fn delete_payment_attempts_for_sample_data(
+        &self,
+        merchant_id: &str,
+    ) -> CustomResult<Vec<PaymentAttempt>, StorageError> {
+        let conn = pg_connection_write(self)
+            .await
+            .change_context(StorageError::DatabaseConnectionError)?;
+        sample_data_queries::delete_payment_attempts(&conn, merchant_id)
+            .await
+            .map_err(diesel_error_to_data_error)
+            .map(|res| {
+                res.into_iter()
+                    .map(PaymentAttempt::from_storage_model)
+                    .collect()
+            })
+    }
+    async fn delete_refunds_for_sample_data(
+        &self,
+        merchant_id: &str,
+    ) -> CustomResult<Vec<Refund>, StorageError> {
+        let conn = pg_connection_write(self)
+            .await
+            .change_context(StorageError::DatabaseConnectionError)?;
+        sample_data_queries::delete_refunds(&conn, merchant_id)
+            .await
+            .map_err(diesel_error_to_data_error)
+    }
+}
+
+#[async_trait::async_trait]
+impl BatchSampleDataInterface for storage_impl::MockDb {
+    async fn insert_payment_intents_batch_for_sample_data(
+        &self,
+        _batch: Vec<PaymentIntentNew>,
+    ) -> CustomResult<Vec<PaymentIntent>, StorageError> {
+        Err(StorageError::MockDbError)?
+    }
+
+    async fn insert_payment_attempts_batch_for_sample_data(
+        &self,
+        _batch: Vec<PaymentAttemptBatchNew>,
+    ) -> CustomResult<Vec<PaymentAttempt>, StorageError> {
+        Err(StorageError::MockDbError)?
+    }
+
+    async fn insert_refunds_batch_for_sample_data(
+        &self,
+        _batch: Vec<RefundNew>,
+    ) -> CustomResult<Vec<Refund>, StorageError> {
+        Err(StorageError::MockDbError)?
+    }
+
+    async fn delete_payment_intents_for_sample_data(
+        &self,
+        _merchant_id: &str,
+    ) -> CustomResult<Vec<PaymentIntent>, StorageError> {
+        Err(StorageError::MockDbError)?
+    }
+    async fn delete_payment_attempts_for_sample_data(
+        &self,
+        _merchant_id: &str,
+    ) -> CustomResult<Vec<PaymentAttempt>, StorageError> {
+        Err(StorageError::MockDbError)?
+    }
+    async fn delete_refunds_for_sample_data(
+        &self,
+        _merchant_id: &str,
+    ) -> CustomResult<Vec<Refund>, StorageError> {
+        Err(StorageError::MockDbError)?
+    }
+}
+
+// TODO: This error conversion is re-used from storage_impl and is not DRY when it should be
+// Ideally the impl's here should be defined in that crate avoiding this re-definition
+fn diesel_error_to_data_error(diesel_error: Report<DatabaseError>) -> Report<StorageError> {
+    let new_err = match diesel_error.current_context() {
+        DatabaseError::DatabaseConnectionError => StorageError::DatabaseConnectionError,
+        DatabaseError::NotFound => StorageError::ValueNotFound("Value not found".to_string()),
+        DatabaseError::UniqueViolation => StorageError::DuplicateValue {
+            entity: "entity ",
+            key: None,
+        },
+        DatabaseError::NoFieldsToUpdate => {
+            StorageError::DatabaseError("No fields to update".to_string())
+        }
+        DatabaseError::QueryGenerationFailed => {
+            StorageError::DatabaseError("Query generation failed".to_string())
+        }
+        DatabaseError::Others => StorageError::DatabaseError("Others".to_string()),
+    };
+    diesel_error.change_context(new_err)
+}