Bug: [FEATURE] Extend the BatchSampleDataInterface Trait for Disputes



### Feature Description
The issue is sub part of issue: #5991 


We currently have a feature for generating sample data, which includes batch insertion of payments and refunds. The feature inserts payments and refunds in batches using the BatchSampleDataInterface trait. We need to extend this functionality to support batch insertion and deletion of disputes as well.

Steps to Implement Batch Support for Disputes

**Extend the BatchSampleDataInterface Trait**
We need to extend the existing BatchSampleDataInterface to include methods for batch insertion and deletion of disputes. These methods will be similar to the ones already implemented for PaymentIntent, PaymentAttempt, and Refund.

**New Methods for Disputes:**
`insert_disputes_batch_for_sample_data`: This method will handle batch insertion of disputes.
`delete_disputes_for_sample_data`: This method will handle (batch) deletion of disputes.

### Possible Implementation

Add Functions for Dispute Insert and Delete in `BatchSampleDataInterface`
- File: `crates/router/src/db/user/sample_data.rs`
- Extend the existing BatchSampleDataInterface to include the methods for inserting and deleting disputes.

Complete these interface functions for Store, Mock Db and Kafka like we did for other functions.

Write Batch Insert and Delete Queries for Disputes
- File: `crates/diesel_model/src/query/user/sample_data.rs`
 - Define the queries (using diesel) to handle batch insertion and deletion of disputes, similar to how they are handled for payments and refunds.
 
 
To test we can hard code some dummy dispute and try inserting using the function we made in core. We can check Db and see the logs for the query that is getting printed.


Steps for Testing Dispute Batch Insertion and Deletion
- To verify the batch insertion and deletion of disputes using the functions implemented in the BatchSampleDataInterface, we can perform a simple test by hardcoding some dummy dispute data. 
- Once we have the dummy dispute data, we can use the `generate_sample_data_for_user`  method from the core function to insert the disputes into the database. This should invoke the appropriate batch insertion logic. (Or we can make an api to test the behaviour when the function is invoked)
Curl to invoke this core function:
```
curl --location 'http://localhost:8080/user/sample_data' \
--header 'Content-Type: application/json' \
--header 'Authorization: Bearer JWT' \
--data '{

}'
```
The JWT token we can get from signup and then skipping the 2FA:
```
curl --location 'http://localhost:8080/user/signup' \
--header 'Content-Type: application/json' \
--data-raw '{
    "email": "JohnTest@test.com",
    "password": "Test@321"
}'
```
This will give an intermediate token in response, we need to use this token to skip 2FA
```
curl --location 'http://localhost:8080/user/2fa/terminate?skip_two_factor_auth=true' \
--header 'Authorization: Bearer Intermediatetoken'
```
This will give us JWT Token that we can use to hit the sample_data api
Curl to delete sample data (The core function for sample dispute delete can be used to our advantage):
```
curl --location --request DELETE 'http://localhost:8080/user/sample_data' \
--header 'Content-Type: application/json' \
--header 'Authorization: Bearer JWT' \
--data '{

}'
```
- We can check Database Logs and Printed Queries whether the appropriate queries are getting printed
- We can also manually check DB 

Note: Omit the Code for Testing part in the PR raised, have only Insert and Delete logic related to the DB, include the relevant logs and screenshots after test.
The Sample disputes generation part, we will be handling in the separate PR. (#6117 )



### Have you spent some time checking if this feature request has been raised before?

- [X] I checked and didn't find a similar issue

### Have you read the Contributing Guidelines?

- [X] I have read the [Contributing Guidelines](https://github.com/juspay/hyperswitch/blob/main/docs/CONTRIBUTING.md)

### Are you willing to submit a PR?

None
diff --git a/crates/diesel_models/src/query/user/sample_data.rs b/crates/diesel_models/src/query/user/sample_data.rs
index b5400f570a3..27e674f7d7b 100644
--- a/crates/diesel_models/src/query/user/sample_data.rs
+++ b/crates/diesel_models/src/query/user/sample_data.rs
@@ -12,9 +12,11 @@ use crate::schema_v2::{
     payment_attempt::dsl as payment_attempt_dsl, payment_intent::dsl as payment_intent_dsl,
 };
 use crate::{
-    errors, schema::refund::dsl as refund_dsl, user::sample_data::PaymentAttemptBatchNew,
-    PaymentAttempt, PaymentIntent, PaymentIntentNew, PgPooledConn, Refund, RefundNew,
-    StorageResult,
+    errors,
+    schema::{dispute::dsl as dispute_dsl, refund::dsl as refund_dsl},
+    user::sample_data::PaymentAttemptBatchNew,
+    Dispute, DisputeNew, PaymentAttempt, PaymentIntent, PaymentIntentNew, PgPooledConn, Refund,
+    RefundNew, StorageResult,
 };
 
 pub async fn insert_payment_intents(
@@ -61,6 +63,21 @@ pub async fn insert_refunds(
         .attach_printable("Error while inserting refunds")
 }
 
+pub async fn insert_disputes(
+    conn: &PgPooledConn,
+    batch: Vec<DisputeNew>,
+) -> StorageResult<Vec<Dispute>> {
+    let query = diesel::insert_into(<Dispute>::table()).values(batch);
+
+    logger::debug!(query = %debug_query::<diesel::pg::Pg,_>(&query).to_string());
+
+    query
+        .get_results_async(conn)
+        .await
+        .change_context(errors::DatabaseError::Others)
+        .attach_printable("Error while inserting disputes")
+}
+
 #[cfg(feature = "v1")]
 pub async fn delete_payment_intents(
     conn: &PgPooledConn,
@@ -165,3 +182,29 @@ pub async fn delete_refunds(
             _ => Ok(result),
         })
 }
+
+pub async fn delete_disputes(
+    conn: &PgPooledConn,
+    merchant_id: &common_utils::id_type::MerchantId,
+) -> StorageResult<Vec<Dispute>> {
+    let query = diesel::delete(<Dispute>::table())
+        .filter(dispute_dsl::merchant_id.eq(merchant_id.to_owned()))
+        .filter(dispute_dsl::dispute_id.like("test_%"));
+
+    logger::debug!(query = %debug_query::<diesel::pg::Pg,_>(&query).to_string());
+
+    query
+        .get_results_async(conn)
+        .await
+        .change_context(errors::DatabaseError::Others)
+        .attach_printable("Error while deleting disputes")
+        .and_then(|result| match result.len() {
+            n if n > 0 => {
+                logger::debug!("{n} records deleted");
+                Ok(result)
+            }
+            0 => Err(error_stack::report!(errors::DatabaseError::NotFound)
+                .attach_printable("No records deleted")),
+            _ => Ok(result),
+        })
+}
diff --git a/crates/router/src/db/kafka_store.rs b/crates/router/src/db/kafka_store.rs
index 9f237c226b6..6e9daeedc59 100644
--- a/crates/router/src/db/kafka_store.rs
+++ b/crates/router/src/db/kafka_store.rs
@@ -3239,6 +3239,26 @@ impl BatchSampleDataInterface for KafkaStore {
         Ok(refunds_list)
     }
 
+    #[cfg(feature = "v1")]
+    async fn insert_disputes_batch_for_sample_data(
+        &self,
+        batch: Vec<diesel_models::DisputeNew>,
+    ) -> CustomResult<Vec<diesel_models::Dispute>, hyperswitch_domain_models::errors::StorageError>
+    {
+        let disputes_list = self
+            .diesel_store
+            .insert_disputes_batch_for_sample_data(batch)
+            .await?;
+
+        for dispute in disputes_list.iter() {
+            let _ = self
+                .kafka_producer
+                .log_dispute(dispute, None, self.tenant_id.clone())
+                .await;
+        }
+        Ok(disputes_list)
+    }
+
     #[cfg(feature = "v1")]
     async fn delete_payment_intents_for_sample_data(
         &self,
@@ -3306,6 +3326,27 @@ impl BatchSampleDataInterface for KafkaStore {
 
         Ok(refunds_list)
     }
+
+    #[cfg(feature = "v1")]
+    async fn delete_disputes_for_sample_data(
+        &self,
+        merchant_id: &id_type::MerchantId,
+    ) -> CustomResult<Vec<diesel_models::Dispute>, hyperswitch_domain_models::errors::StorageError>
+    {
+        let disputes_list = self
+            .diesel_store
+            .delete_disputes_for_sample_data(merchant_id)
+            .await?;
+
+        for dispute in disputes_list.iter() {
+            let _ = self
+                .kafka_producer
+                .log_dispute_delete(dispute, self.tenant_id.clone())
+                .await;
+        }
+
+        Ok(disputes_list)
+    }
 }
 
 #[async_trait::async_trait]
diff --git a/crates/router/src/db/user/sample_data.rs b/crates/router/src/db/user/sample_data.rs
index 709ff4a358a..3add7b744d0 100644
--- a/crates/router/src/db/user/sample_data.rs
+++ b/crates/router/src/db/user/sample_data.rs
@@ -1,5 +1,6 @@
 use common_utils::types::keymanager::KeyManagerState;
 use diesel_models::{
+    dispute::{Dispute, DisputeNew},
     errors::DatabaseError,
     query::user::sample_data as sample_data_queries,
     refund::{Refund, RefundNew},
@@ -39,6 +40,12 @@ pub trait BatchSampleDataInterface {
         batch: Vec<RefundNew>,
     ) -> CustomResult<Vec<Refund>, StorageError>;
 
+    #[cfg(feature = "v1")]
+    async fn insert_disputes_batch_for_sample_data(
+        &self,
+        batch: Vec<DisputeNew>,
+    ) -> CustomResult<Vec<Dispute>, StorageError>;
+
     #[cfg(feature = "v1")]
     async fn delete_payment_intents_for_sample_data(
         &self,
@@ -58,6 +65,12 @@ pub trait BatchSampleDataInterface {
         &self,
         merchant_id: &common_utils::id_type::MerchantId,
     ) -> CustomResult<Vec<Refund>, StorageError>;
+
+    #[cfg(feature = "v1")]
+    async fn delete_disputes_for_sample_data(
+        &self,
+        merchant_id: &common_utils::id_type::MerchantId,
+    ) -> CustomResult<Vec<Dispute>, StorageError>;
 }
 
 #[async_trait::async_trait]
@@ -127,6 +140,19 @@ impl BatchSampleDataInterface for Store {
             .map_err(diesel_error_to_data_error)
     }
 
+    #[cfg(feature = "v1")]
+    async fn insert_disputes_batch_for_sample_data(
+        &self,
+        batch: Vec<DisputeNew>,
+    ) -> CustomResult<Vec<Dispute>, StorageError> {
+        let conn = pg_connection_write(self)
+            .await
+            .change_context(StorageError::DatabaseConnectionError)?;
+        sample_data_queries::insert_disputes(&conn, batch)
+            .await
+            .map_err(diesel_error_to_data_error)
+    }
+
     #[cfg(feature = "v1")]
     async fn delete_payment_intents_for_sample_data(
         &self,
@@ -184,6 +210,19 @@ impl BatchSampleDataInterface for Store {
             .await
             .map_err(diesel_error_to_data_error)
     }
+
+    #[cfg(feature = "v1")]
+    async fn delete_disputes_for_sample_data(
+        &self,
+        merchant_id: &common_utils::id_type::MerchantId,
+    ) -> CustomResult<Vec<Dispute>, StorageError> {
+        let conn = pg_connection_write(self)
+            .await
+            .change_context(StorageError::DatabaseConnectionError)?;
+        sample_data_queries::delete_disputes(&conn, merchant_id)
+            .await
+            .map_err(diesel_error_to_data_error)
+    }
 }
 
 #[async_trait::async_trait]
@@ -214,6 +253,14 @@ impl BatchSampleDataInterface for storage_impl::MockDb {
         Err(StorageError::MockDbError)?
     }
 
+    #[cfg(feature = "v1")]
+    async fn insert_disputes_batch_for_sample_data(
+        &self,
+        _batch: Vec<DisputeNew>,
+    ) -> CustomResult<Vec<Dispute>, StorageError> {
+        Err(StorageError::MockDbError)?
+    }
+
     #[cfg(feature = "v1")]
     async fn delete_payment_intents_for_sample_data(
         &self,
@@ -239,6 +286,14 @@ impl BatchSampleDataInterface for storage_impl::MockDb {
     ) -> CustomResult<Vec<Refund>, StorageError> {
         Err(StorageError::MockDbError)?
     }
+
+    #[cfg(feature = "v1")]
+    async fn delete_disputes_for_sample_data(
+        &self,
+        _merchant_id: &common_utils::id_type::MerchantId,
+    ) -> CustomResult<Vec<Dispute>, StorageError> {
+        Err(StorageError::MockDbError)?
+    }
 }
 
 // TODO: This error conversion is re-used from storage_impl and is not DRY when it should be
diff --git a/crates/router/src/services/kafka.rs b/crates/router/src/services/kafka.rs
index 4fd692e1166..51ae9456ecc 100644
--- a/crates/router/src/services/kafka.rs
+++ b/crates/router/src/services/kafka.rs
@@ -582,6 +582,21 @@ impl KafkaProducer {
         .attach_printable_lazy(|| format!("Failed to add consolidated dispute event {dispute:?}"))
     }
 
+    pub async fn log_dispute_delete(
+        &self,
+        delete_old_dispute: &Dispute,
+        tenant_id: TenantID,
+    ) -> MQResult<()> {
+        self.log_event(&KafkaEvent::old(
+            &KafkaDispute::from_storage(delete_old_dispute),
+            tenant_id.clone(),
+            self.ckh_database_name.clone(),
+        ))
+        .attach_printable_lazy(|| {
+            format!("Failed to add negative dispute event {delete_old_dispute:?}")
+        })
+    }
+
     #[cfg(feature = "payouts")]
     pub async fn log_payout(
         &self,

## Type of Change
<!-- Put an `x` in the boxes that apply -->

- [ ] Bugfix
- [x] New feature
- [x] Enhancement
- [ ] Refactoring
- [ ] Dependency updates
- [ ] Documentation
- [ ] CI/CD

## Description
<!-- Describe your changes in detail -->
I've extended the interface for `BatchSampleDataInterface` to with `insert_disputes_batch_for_sample_data` and `delete_disputes_for_sample_data`, and added implementation to all structs implementing this trait. I've also modified `generate_sample_data` function in `utils/user/sample_data.rs` to generate dispute data, and `delete_sample_data_for_user` function in `core/user/sample_data.rs` to delete the generated dispute data.

The overall result is the `/user/sample_data` endpoint now supports dispute data generation and deletion.

### Additional Changes

- [x] This PR modifies the API contract
- [ ] This PR modifies the database schema
- [ ] This PR modifies application configuration/environment variables

<!--
Provide links to the files with corresponding changes.

Following are the paths where you can find config files:
1. `config`
2. `crates/router/src/configs`
3. `loadtest/config`
-->


## Motivation and Context
<!--
Why is this change required? What problem does it solve?
If it fixes an open issue, please link to the issue here.

If you don't have an issue, we'd recommend starting with one first so the PR
can focus on the implementation (unless it is an obvious bug or documentation fix
that will have little conversation).
-->
resolve #6111 


## How did you test it?
<!--
Did you write an integration/unit/API test to verify the code changes?
Or did you test this change manually (provide relevant screenshots)?
-->
Manual test is done, screenshots are provided in the comment section of this issue: https://github.com/juspay/hyperswitch/issues/6111


## Checklist
<!-- Put an `x` in the boxes that apply -->

- [x] I formatted the code `cargo +nightly fmt --all`
- [x] I addressed lints thrown by `cargo clippy`
- [x] I reviewed the submitted code
- [ ] I added unit tests for my changes where possible (I did not find where to add unit tests for this issue)
