Issue:
Bug: Facilitating dispute management



Dispute occurs when a customer files a chargeback claim to the bank stating that the payment was not intended. While cardholders often have extended dispute filing periods of 60-120 days after the transaction or order delivery, merchants, on the other hand, must deal with shorter turnaround windows.

Faciitating disputes involves:
1. Notifying merchants upon disputes using webhooks
2. Ability to upload evidences against dispute using API/ dashboard
3. Ability to accept/ challenge dispute using the dashboard
4. Integration of dispute management APIs with those connectors which has exposed APIs

Pull request description:
## Type of Change
<!-- Put an `x` in the boxes that apply -->

- [ ] Bugfix
- [x] New feature
- [ ] Enhancement
- [ ] Refactoring
- [ ] Dependency updates

## Description
Added support for incoming dispute webhooks.


### Additional Changes

- [x] 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 its an obvious bug or documentation fix
that will have little conversation).
-->


## How did you test it?
Manually Tested
Webhook Request
<img width="1256" alt="image" src="https://user-images.githubusercontent.com/99009240/226270250-5f8926dd-cba5-4a52-81c7-48c679ec9110.png">

DB Entry
<img width="763" alt="image" src="https://user-images.githubusercontent.com/99009240/226270204-d9c4403d-d20c-4085-bd0b-6a434bd9c4ba.png">

Outgoing webhook to merchant
<img width="780" alt="image" src="https://user-images.githubusercontent.com/99009240/226270349-bc958318-a2e5-4b8e-9d8f-dc9bc3fa494e.png">

## 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 submitted code
- [ ] I added unit tests for my changes where possible
- [ ] I added a [CHANGELOG](/CHANGELOG.md) entry if applicable


Testing:
Manually Tested
Webhook Request
<img width="1256" alt="image" src="https://user-images.githubusercontent.com/99009240/226270250-5f8926dd-cba5-4a52-81c7-48c679ec9110.png">

DB Entry
<img width="763" alt="image" src="https://user-images.githubusercontent.com/99009240/226270204-d9c4403d-d20c-4085-bd0b-6a434bd9c4ba.png">

Outgoing webhook to merchant
<img width="780" alt="image" src="https://user-images.githubusercontent.com/99009240/226270349-bc958318-a2e5-4b8e-9d8f-dc9bc3fa494e.png">

Patch:
diff --git a/crates/api_models/src/disputes.rs b/crates/api_models/src/disputes.rs
index 8b137891791..e60a6a44fde 100644
--- a/crates/api_models/src/disputes.rs
+++ b/crates/api_models/src/disputes.rs
@@ -1 +1,38 @@
+use masking::Serialize;
+use utoipa::ToSchema;
 
+use super::enums::{DisputeStage, DisputeStatus};
+
+#[derive(Default, Clone, Debug, Serialize, ToSchema)]
+pub struct DisputeResponse {
+    /// The identifier for dispute
+    pub dispute_id: String,
+    /// The identifier for payment_intent
+    pub payment_id: String,
+    /// The identifier for payment_attempt
+    pub attempt_id: String,
+    /// The dispute amount
+    pub amount: String,
+    /// The three-letter ISO currency code
+    pub currency: String,
+    /// Stage of the dispute
+    pub dispute_stage: DisputeStage,
+    /// Status of the dispute
+    pub dispute_status: DisputeStatus,
+    /// Status of the dispute sent by connector
+    pub connector_status: String,
+    /// Dispute id sent by connector
+    pub connector_dispute_id: String,
+    /// Reason of dispute sent by connector
+    pub connector_reason: Option<String>,
+    /// Reason code of dispute sent by connector
+    pub connector_reason_code: Option<String>,
+    /// Evidence deadline of dispute sent by connector
+    pub challenge_required_by: Option<String>,
+    /// Dispute created time sent by connector
+    pub created_at: Option<String>,
+    /// Dispute updated time sent by connector
+    pub updated_at: Option<String>,
+    /// Time at which dispute is received
+    pub received_at: String,
+}
diff --git a/crates/api_models/src/enums.rs b/crates/api_models/src/enums.rs
index ccdac373ac0..00706727f34 100644
--- a/crates/api_models/src/enums.rs
+++ b/crates/api_models/src/enums.rs
@@ -267,6 +267,13 @@ pub enum EventType {
     PaymentSucceeded,
     RefundSucceeded,
     RefundFailed,
+    DisputeOpened,
+    DisputeExpired,
+    DisputeAccepted,
+    DisputeCancelled,
+    DisputeChallenged,
+    DisputeWon,
+    DisputeLost,
 }
 
 #[derive(
@@ -767,3 +774,51 @@ impl From<AttemptStatus> for IntentStatus {
         }
     }
 }
+
+#[derive(
+    Clone,
+    Default,
+    Debug,
+    Eq,
+    Hash,
+    PartialEq,
+    serde::Deserialize,
+    serde::Serialize,
+    strum::Display,
+    strum::EnumString,
+    frunk::LabelledGeneric,
+    ToSchema,
+)]
+pub enum DisputeStage {
+    PreDispute,
+    #[default]
+    Dispute,
+    PreArbitration,
+}
+
+#[derive(
+    Clone,
+    Debug,
+    Default,
+    Eq,
+    Hash,
+    PartialEq,
+    serde::Deserialize,
+    serde::Serialize,
+    strum::Display,
+    strum::EnumString,
+    frunk::LabelledGeneric,
+    ToSchema,
+)]
+pub enum DisputeStatus {
+    #[default]
+    DisputeOpened,
+    DisputeExpired,
+    DisputeAccepted,
+    DisputeCancelled,
+    DisputeChallenged,
+    // dispute has been successfully challenged by the merchant
+    DisputeWon,
+    // dispute has been unsuccessfully challenged
+    DisputeLost,
+}
diff --git a/crates/api_models/src/webhooks.rs b/crates/api_models/src/webhooks.rs
index 98e9e0c3e34..43b69fc413b 100644
--- a/crates/api_models/src/webhooks.rs
+++ b/crates/api_models/src/webhooks.rs
@@ -2,7 +2,7 @@ use common_utils::custom_serde;
 use serde::{Deserialize, Serialize};
 use time::PrimitiveDateTime;
 
-use crate::{enums as api_enums, payments, refunds};
+use crate::{disputes, enums as api_enums, payments, refunds};
 
 #[derive(Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
 #[serde(rename_all = "snake_case")]
@@ -11,12 +11,22 @@ pub enum IncomingWebhookEvent {
     PaymentIntentSuccess,
     RefundFailure,
     RefundSuccess,
+    DisputeOpened,
+    DisputeExpired,
+    DisputeAccepted,
+    DisputeCancelled,
+    DisputeChallenged,
+    // dispute has been successfully challenged by the merchant
+    DisputeWon,
+    // dispute has been unsuccessfully challenged
+    DisputeLost,
     EndpointVerification,
 }
 
 pub enum WebhookFlow {
     Payment,
     Refund,
+    Dispute,
     Subscription,
     ReturnResponse,
 }
@@ -28,6 +38,13 @@ impl From<IncomingWebhookEvent> for WebhookFlow {
             IncomingWebhookEvent::PaymentIntentSuccess => Self::Payment,
             IncomingWebhookEvent::RefundSuccess => Self::Refund,
             IncomingWebhookEvent::RefundFailure => Self::Refund,
+            IncomingWebhookEvent::DisputeOpened => Self::Dispute,
+            IncomingWebhookEvent::DisputeAccepted => Self::Dispute,
+            IncomingWebhookEvent::DisputeExpired => Self::Dispute,
+            IncomingWebhookEvent::DisputeCancelled => Self::Dispute,
+            IncomingWebhookEvent::DisputeChallenged => Self::Dispute,
+            IncomingWebhookEvent::DisputeWon => Self::Dispute,
+            IncomingWebhookEvent::DisputeLost => Self::Dispute,
             IncomingWebhookEvent::EndpointVerification => Self::ReturnResponse,
         }
     }
@@ -73,6 +90,7 @@ pub struct OutgoingWebhook {
 pub enum OutgoingWebhookContent {
     PaymentDetails(payments::PaymentsResponse),
     RefundDetails(refunds::RefundResponse),
+    DisputeDetails(Box<disputes::DisputeResponse>),
 }
 
 pub trait OutgoingWebhookType: Serialize + From<OutgoingWebhook> + Sync + Send {}
diff --git a/crates/router/src/compatibility/stripe/webhooks.rs b/crates/router/src/compatibility/stripe/webhooks.rs
index 71bd2682347..23952f2aef2 100644
--- a/crates/router/src/compatibility/stripe/webhooks.rs
+++ b/crates/router/src/compatibility/stripe/webhooks.rs
@@ -1,4 +1,7 @@
-use api_models::webhooks::{self as api};
+use api_models::{
+    enums::DisputeStatus,
+    webhooks::{self as api},
+};
 use serde::Serialize;
 
 use super::{
@@ -20,6 +23,57 @@ impl api::OutgoingWebhookType for StripeOutgoingWebhook {}
 pub enum StripeWebhookObject {
     PaymentIntent(StripePaymentIntentResponse),
     Refund(StripeRefundResponse),
+    Dispute(StripeDisputeResponse),
+}
+
+#[derive(Serialize)]
+pub struct StripeDisputeResponse {
+    pub id: String,
+    pub amount: String,
+    pub currency: String,
+    pub payment_intent: String,
+    pub reason: Option<String>,
+    pub status: StripeDisputeStatus,
+}
+
+#[derive(Serialize)]
+#[serde(rename_all = "snake_case")]
+pub enum StripeDisputeStatus {
+    WarningNeedsResponse,
+    WarningUnderReview,
+    WarningClosed,
+    NeedsResponse,
+    UnderReview,
+    ChargeRefunded,
+    Won,
+    Lost,
+}
+
+impl From<api_models::disputes::DisputeResponse> for StripeDisputeResponse {
+    fn from(res: api_models::disputes::DisputeResponse) -> Self {
+        Self {
+            id: res.dispute_id,
+            amount: res.amount,
+            currency: res.currency,
+            payment_intent: res.payment_id,
+            reason: res.connector_reason,
+            status: StripeDisputeStatus::from(res.dispute_status),
+        }
+    }
+}
+
+impl From<DisputeStatus> for StripeDisputeStatus {
+    fn from(status: DisputeStatus) -> Self {
+        match status {
+            DisputeStatus::DisputeOpened => Self::WarningNeedsResponse,
+            DisputeStatus::DisputeExpired => Self::Lost,
+            DisputeStatus::DisputeAccepted => Self::Lost,
+            DisputeStatus::DisputeCancelled => Self::WarningClosed,
+            DisputeStatus::DisputeChallenged => Self::WarningUnderReview,
+            DisputeStatus::DisputeWon => Self::Won,
+            DisputeStatus::DisputeLost => Self::Lost,
+        }
+    }
 }
 
 impl From<api::OutgoingWebhook> for StripeOutgoingWebhook {
@@ -40,6 +94,9 @@ impl From<api::OutgoingWebhookContent> for StripeWebhookObject {
                 Self::PaymentIntent(payment.into())
             }
             api::OutgoingWebhookContent::RefundDetails(refund) => Self::Refund(refund.into()),
+            api::OutgoingWebhookContent::DisputeDetails(dispute) => {
+                Self::Dispute((*dispute).into())
+            }
         }
     }
 }
@@ -49,6 +106,7 @@ impl StripeWebhookObject {
         match self {
             Self::PaymentIntent(p) => p.id.to_owned(),
             Self::Refund(r) => Some(r.id.to_owned()),
+            Self::Dispute(d) => Some(d.id.to_owned()),
         }
     }
 }
diff --git a/crates/router/src/connector/adyen.rs b/crates/router/src/connector/adyen.rs
index 75bcc42b423..813708b3fe6 100644
--- a/crates/router/src/connector/adyen.rs
+++ b/crates/router/src/connector/adyen.rs
@@ -2,6 +2,7 @@ mod transformers;
 
 use std::fmt::Debug;
 
+use api_models::webhooks::IncomingWebhookEvent;
 use base64::Engine;
 use error_stack::{IntoReport, ResultExt};
 use router_env::{instrument, tracing};
@@ -17,6 +18,7 @@ use crate::{
     types::{
         self,
         api::{self, ConnectorCommon},
+        transformers::ForeignFrom,
     },
     utils::{self, crypto, ByteSliceExt, BytesExt, OptionExt},
 };
@@ -719,27 +721,38 @@ impl api::IncomingWebhook for Adyen {
     ) -> CustomResult<api_models::webhooks::ObjectReferenceId, errors::ConnectorError> {
         let notif = get_webhook_object_from_body(request.body)
             .change_context(errors::ConnectorError::WebhookReferenceIdNotFound)?;
-        match notif.event_code {
-            adyen::WebhookEventCode::Authorisation => {
-                Ok(api_models::webhooks::ObjectReferenceId::PaymentId(
-                    api_models::payments::PaymentIdType::ConnectorTransactionId(
-                        notif.psp_reference,
-                    ),
-                ))
-            }
-            _ => Ok(api_models::webhooks::ObjectReferenceId::RefundId(
+        if adyen::is_transaction_event(&notif.event_code) {
+            return Ok(api_models::webhooks::ObjectReferenceId::PaymentId(
+                api_models::payments::PaymentIdType::ConnectorTransactionId(notif.psp_reference),
+            ));
+        }
+        if adyen::is_refund_event(&notif.event_code) {
+            return Ok(api_models::webhooks::ObjectReferenceId::RefundId(
                 api_models::webhooks::RefundIdType::ConnectorRefundId(notif.psp_reference),
-            )),
+            ));
         }
+        if adyen::is_chargeback_event(&notif.event_code) {
+            return Ok(api_models::webhooks::ObjectReferenceId::PaymentId(
+                api_models::payments::PaymentIdType::ConnectorTransactionId(
+                    notif
+                        .original_reference
+                        .ok_or(errors::ConnectorError::WebhookReferenceIdNotFound)?,
+                ),
+            ));
+        }
+        Err(errors::ConnectorError::WebhookReferenceIdNotFound).into_report()
     }
 
     fn get_webhook_event_type(
         &self,
         request: &api::IncomingWebhookRequestDetails<'_>,
-    ) -> CustomResult<api::IncomingWebhookEvent, errors::ConnectorError> {
+    ) -> CustomResult<IncomingWebhookEvent, errors::ConnectorError> {
         let notif = get_webhook_object_from_body(request.body)
             .change_context(errors::ConnectorError::WebhookEventTypeNotFound)?;
-        Ok(notif.event_code.into())
+        Ok(IncomingWebhookEvent::foreign_from((
+            notif.event_code,
+            notif.additional_data.dispute_status,
+        )))
     }
 
     fn get_webhook_resource_object(
@@ -767,4 +780,24 @@ impl api::IncomingWebhook for Adyen {
             "[accepted]".to_string(),
         ))
     }
+
+    fn get_dispute_details(
+        &self,
+        request: &api_models::webhooks::IncomingWebhookRequestDetails<'_>,
+    ) -> CustomResult<api::disputes::DisputePayload, errors::ConnectorError> {
+        let notif = get_webhook_object_from_body(request.body)
+            .change_context(errors::ConnectorError::WebhookBodyDecodingFailed)?;
+        Ok(api::disputes::DisputePayload {
+            amount: notif.amount.value.to_string(),
+            currency: notif.amount.currency,
+            dispute_stage: api_models::enums::DisputeStage::from(notif.event_code.clone()),
+            connector_dispute_id: notif.psp_reference,
+            connector_reason: notif.reason,
+            connector_reason_code: notif.additional_data.chargeback_reason_code,
+            challenge_required_by: notif.additional_data.defense_period_ends_at,
+            connector_status: notif.event_code.to_string(),
+            created_at: notif.event_date.clone(),
+            updated_at: notif.event_date,
+        })
+    }
 }
diff --git a/crates/router/src/connector/adyen/transformers.rs b/crates/router/src/connector/adyen/transformers.rs
index 71123ac1f65..46a1fc69f9a 100644
--- a/crates/router/src/connector/adyen/transformers.rs
+++ b/crates/router/src/connector/adyen/transformers.rs
@@ -1,4 +1,4 @@
-use api_models::webhooks::IncomingWebhookEvent;
+use api_models::{enums::DisputeStage, webhooks::IncomingWebhookEvent};
 use masking::PeekInterface;
 use reqwest::Url;
 use serde::{Deserialize, Serialize};
@@ -13,6 +13,7 @@ use crate::{
         self,
         api::{self, enums as api_enums},
         storage::enums as storage_enums,
+        transformers::ForeignFrom,
     },
 };
 
@@ -1148,10 +1149,22 @@ pub struct ErrorResponse {
 //     }
 // }
 
+#[derive(Debug, Deserialize)]
+pub enum DisputeStatus {
+    Undefended,
+    Pending,
+    Lost,
+    Accepted,
+    Won,
+}
+
 #[derive(Debug, Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct AdyenAdditionalDataWH {
     pub hmac_signature: String,
+    pub dispute_status: Option<DisputeStatus>,
+    pub chargeback_reason_code: Option<String>,
+    pub defense_period_ends_at: Option<String>,
 }
 
 #[derive(Debug, Deserialize)]
@@ -1160,7 +1173,7 @@ pub struct AdyenAmountWH {
     pub currency: String,
 }
 
-#[derive(Debug, Deserialize, strum::Display)]
+#[derive(Clone, Debug, Deserialize, strum::Display)]
 #[serde(rename_all = "SCREAMING_SNAKE_CASE")]
 #[strum(serialize_all = "SCREAMING_SNAKE_CASE")]
 pub enum WebhookEventCode {
@@ -1168,15 +1181,73 @@ pub enum WebhookEventCode {
     Refund,
     CancelOrRefund,
     RefundFailed,
+    NotificationOfChargeback,
+    Chargeback,
+    ChargebackReversed,
+    SecondChargeback,
+    PrearbitrationWon,
+    PrearbitrationLost,
+}
+
+pub fn is_transaction_event(event_code: &WebhookEventCode) -> bool {
+    matches!(event_code, WebhookEventCode::Authorisation)
+}
+
+pub fn is_refund_event(event_code: &WebhookEventCode) -> bool {
+    matches!(
+        event_code,
+        WebhookEventCode::Refund
+            | WebhookEventCode::CancelOrRefund
+            | WebhookEventCode::RefundFailed
+    )
+}
+
+pub fn is_chargeback_event(event_code: &WebhookEventCode) -> bool {
+    matches!(
+        event_code,
+        WebhookEventCode::NotificationOfChargeback
+            | WebhookEventCode::Chargeback
+            | WebhookEventCode::ChargebackReversed
+            | WebhookEventCode::SecondChargeback
+            | WebhookEventCode::PrearbitrationWon
+            | WebhookEventCode::PrearbitrationLost
+    )
+}
+
+impl ForeignFrom<(WebhookEventCode, Option<DisputeStatus>)> for IncomingWebhookEvent {
+    fn foreign_from((code, status): (WebhookEventCode, Option<DisputeStatus>)) -> Self {
+        match (code, status) {
+            (WebhookEventCode::Authorisation, _) => Self::PaymentIntentSuccess,
+            (WebhookEventCode::Refund, _) => Self::RefundSuccess,
+            (WebhookEventCode::CancelOrRefund, _) => Self::RefundSuccess,
+            (WebhookEventCode::RefundFailed, _) => Self::RefundFailure,
+            (WebhookEventCode::NotificationOfChargeback, _) => Self::DisputeOpened,
+            (WebhookEventCode::Chargeback, None) => Self::DisputeLost,
+            (WebhookEventCode::Chargeback, Some(DisputeStatus::Won)) => Self::DisputeWon,
+            (WebhookEventCode::Chargeback, Some(DisputeStatus::Lost)) => Self::DisputeLost,
+            (WebhookEventCode::Chargeback, Some(_)) => Self::DisputeOpened,
+            (WebhookEventCode::ChargebackReversed, Some(DisputeStatus::Pending)) => {
+                Self::DisputeChallenged
+            }
+            (WebhookEventCode::ChargebackReversed, _) => Self::DisputeWon,
+            (WebhookEventCode::SecondChargeback, _) => Self::DisputeLost,
+            (WebhookEventCode::PrearbitrationWon, Some(DisputeStatus::Pending)) => {
+                Self::DisputeOpened
+            }
+            (WebhookEventCode::PrearbitrationWon, _) => Self::DisputeWon,
+            (WebhookEventCode::PrearbitrationLost, _) => Self::DisputeLost,
+        }
+    }
 }
 
-impl From<WebhookEventCode> for IncomingWebhookEvent {
+impl From<WebhookEventCode> for DisputeStage {
     fn from(code: WebhookEventCode) -> Self {
         match code {
-            WebhookEventCode::Authorisation => Self::PaymentIntentSuccess,
-            WebhookEventCode::Refund => Self::RefundSuccess,
-            WebhookEventCode::CancelOrRefund => Self::RefundSuccess,
-            WebhookEventCode::RefundFailed => Self::RefundFailure,
+            WebhookEventCode::NotificationOfChargeback => Self::PreDispute,
+            WebhookEventCode::SecondChargeback => Self::PreArbitration,
+            WebhookEventCode::PrearbitrationWon => Self::PreArbitration,
+            WebhookEventCode::PrearbitrationLost => Self::PreArbitration,
+            _ => Self::Dispute,
         }
     }
 }
@@ -1192,6 +1263,8 @@ pub struct AdyenNotificationRequestItemWH {
     pub merchant_account_code: String,
     pub merchant_reference: String,
     pub success: String,
+    pub reason: Option<String>,
+    pub event_date: Option<String>,
 }
 
 #[derive(Debug, Deserialize)]
diff --git a/crates/router/src/connector/trustpay.rs b/crates/router/src/connector/trustpay.rs
index 8736a17062f..9df8fb33903 100644
--- a/crates/router/src/connector/trustpay.rs
+++ b/crates/router/src/connector/trustpay.rs
@@ -598,7 +598,7 @@ impl api::IncomingWebhook for Trustpay {
         match details.payment_information.credit_debit_indicator {
             trustpay::CreditDebitIndicator::Crdt => {
                 Ok(api_models::webhooks::ObjectReferenceId::PaymentId(
-                    api_models::payments::PaymentIdType::PaymentIntentId(
+                    api_models::payments::PaymentIdType::PaymentAttemptId(
                         details.payment_information.references.merchant_reference,
                     ),
                 ))
@@ -606,7 +606,7 @@ impl api::IncomingWebhook for Trustpay {
             trustpay::CreditDebitIndicator::Dbit => {
                 if details.payment_information.status == trustpay::WebhookStatus::Chargebacked {
                     Ok(api_models::webhooks::ObjectReferenceId::PaymentId(
-                        api_models::payments::PaymentIdType::PaymentIntentId(
+                        api_models::payments::PaymentIdType::PaymentAttemptId(
                             details.payment_information.references.merchant_reference,
                         ),
                     ))
@@ -648,6 +648,9 @@ impl api::IncomingWebhook for Trustpay {
             (trustpay::CreditDebitIndicator::Dbit, trustpay::WebhookStatus::Rejected) => {
                 Ok(api_models::webhooks::IncomingWebhookEvent::RefundFailure)
             }
+            (trustpay::CreditDebitIndicator::Dbit, trustpay::WebhookStatus::Chargebacked) => {
+                Ok(api_models::webhooks::IncomingWebhookEvent::DisputeLost)
+            }
             _ => Err(errors::ConnectorError::WebhookEventTypeNotFound).into_report()?,
         }
     }
@@ -716,6 +719,30 @@ impl api::IncomingWebhook for Trustpay {
             .change_context(errors::ConnectorError::WebhookVerificationSecretNotFound)?;
         Ok(secret)
     }
+
+    fn get_dispute_details(
+        &self,
+        request: &api_models::webhooks::IncomingWebhookRequestDetails<'_>,
+    ) -> CustomResult<api::disputes::DisputePayload, errors::ConnectorError> {
+        let trustpay_response: trustpay::TrustpayWebhookResponse = request
+            .body
+            .parse_struct("TrustpayWebhookResponse")
+            .switch()?;
+        let payment_info = trustpay_response.payment_information;
+        let reason = payment_info.status_reason_information.unwrap_or_default();
+        Ok(api::disputes::DisputePayload {
+            amount: payment_info.amount.amount.to_string(),
+            currency: payment_info.amount.currency,
+            dispute_stage: api_models::enums::DisputeStage::Dispute,
+            connector_dispute_id: payment_info.references.payment_id,
+            connector_reason: reason.reason.reject_reason,
+            connector_reason_code: Some(reason.reason.code),
+            challenge_required_by: None,
+            connector_status: payment_info.status.to_string(),
+            created_at: None,
+            updated_at: None,
+        })
+    }
 }
 
 impl services::ConnectorRedirectResponse for Trustpay {
diff --git a/crates/router/src/connector/trustpay/transformers.rs b/crates/router/src/connector/trustpay/transformers.rs
index 07c0b9d1d80..f06b0385a63 100644
--- a/crates/router/src/connector/trustpay/transformers.rs
+++ b/crates/router/src/connector/trustpay/transformers.rs
@@ -219,7 +219,7 @@ fn get_card_request_data(
         cvv: ccard.card_cvc.clone(),
         expiry_date: ccard.get_card_expiry_month_year_2_digit_with_delimiter("/".to_owned()),
         cardholder: ccard.card_holder_name.clone(),
-        reference: item.payment_id.clone(),
+        reference: item.attempt_id.clone(),
         redirect_url: return_url,
         billing_city: params.billing_city,
         billing_country: params.billing_country,
@@ -260,7 +260,7 @@ fn get_bank_redirection_request_data(
                 currency: item.request.currency.to_string(),
             },
             references: References {
-                merchant_reference: item.payment_id.clone(),
+                merchant_reference: item.attempt_id.clone(),
             },
         },
         callback_urls: CallbackURLs {
@@ -1115,12 +1115,11 @@ pub enum CreditDebitIndicator {
     Dbit,
 }
 
-#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+#[derive(strum::Display, Debug, Clone, Serialize, Deserialize, PartialEq)]
 pub enum WebhookStatus {
     Paid,
     Rejected,
     Refunded,
-    // TODO (Handle Chargebacks)
     Chargebacked,
 }
 
@@ -1151,15 +1150,25 @@ impl TryFrom<WebhookStatus> for storage_models::enums::RefundStatus {
 #[serde(rename_all = "PascalCase")]
 pub struct WebhookReferences {
     pub merchant_reference: String,
+    pub payment_id: String,
     pub payment_request_id: Option<String>,
 }
 
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+#[serde(rename_all = "PascalCase")]
+pub struct WebhookAmount {
+    pub amount: f64,
+    pub currency: String,
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
 #[serde(rename_all = "PascalCase")]
 pub struct WebhookPaymentInformation {
     pub credit_debit_indicator: CreditDebitIndicator,
     pub references: WebhookReferences,
     pub status: WebhookStatus,
+    pub amount: WebhookAmount,
+    pub status_reason_information: Option<StatusReasonInformation>,
 }
 
 #[derive(Debug, Serialize, Deserialize, PartialEq)]
diff --git a/crates/router/src/core.rs b/crates/router/src/core.rs
index 0ae37d567c0..58690e8c449 100644
--- a/crates/router/src/core.rs
+++ b/crates/router/src/core.rs
@@ -5,6 +5,7 @@ pub mod configs;
 pub mod customers;
 pub mod errors;
 pub mod mandate;
+pub mod metrics;
 pub mod payment_methods;
 pub mod payments;
 pub mod refunds;
diff --git a/crates/router/src/core/errors.rs b/crates/router/src/core/errors.rs
index ea7b85e8804..00462971d4d 100644
--- a/crates/router/src/core/errors.rs
+++ b/crates/router/src/core/errors.rs
@@ -428,6 +428,8 @@ pub enum WebhooksFlowError {
     PaymentsCoreFailed,
     #[error("Refunds core flow failed")]
     RefundsCoreFailed,
+    #[error("Dispuste core flow failed")]
+    DisputeCoreFailed,
     #[error("Webhook event creation failed")]
     WebhookEventCreationFailed,
     #[error("Unable to fork webhooks flow for outgoing webhooks")]
@@ -438,4 +440,12 @@ pub enum WebhooksFlowError {
     NotReceivedByMerchant,
     #[error("Resource not found")]
     ResourceNotFound,
+    #[error("Webhook source verification failed")]
+    WebhookSourceVerificationFailed,
+    #[error("Webhook event object creation failed")]
+    WebhookEventObjectCreationFailed,
+    #[error("Not implemented")]
+    NotImplemented,
+    #[error("Dispute webhook status validation failed")]
+    DisputeWebhookValidationFailed,
 }
diff --git a/crates/router/src/core/metrics.rs b/crates/router/src/core/metrics.rs
new file mode 100644
index 00000000000..fe172049652
--- /dev/null
+++ b/crates/router/src/core/metrics.rs
@@ -0,0 +1,20 @@
+use router_env::{counter_metric, global_meter, metrics_context};
+
+metrics_context!(CONTEXT);
+global_meter!(GLOBAL_METER, "ROUTER_API");
+
+counter_metric!(INCOMING_DISPUTE_WEBHOOK_METRIC, GLOBAL_METER); // No. of incoming dispute webhooks
+counter_metric!(
+    INCOMING_DISPUTE_WEBHOOK_SIGNATURE_FAILURE_METRIC,
+    GLOBAL_METER
+); // No. of incoming dispute webhooks for which signature verification failed
+counter_metric!(
+    INCOMING_DISPUTE_WEBHOOK_VALIDATION_FAILURE_METRIC,
+    GLOBAL_METER
+); // No. of incoming dispute webhooks for which validation failed
+counter_metric!(INCOMING_DISPUTE_WEBHOOK_NEW_RECORD_METRIC, GLOBAL_METER); // No. of incoming dispute webhooks for which new record is created in our db
+counter_metric!(INCOMING_DISPUTE_WEBHOOK_UPDATE_RECORD_METRIC, GLOBAL_METER); // No. of incoming dispute webhooks for which we have updated the details to existing record
+counter_metric!(
+    INCOMING_DISPUTE_WEBHOOK_MERCHANT_NOTIFIED_METRIC,
+    GLOBAL_METER
+); // No. of incoming dispute webhooks which are notified to merchant
diff --git a/crates/router/src/core/utils.rs b/crates/router/src/core/utils.rs
index 5908fcb285a..75df85f7974 100644
--- a/crates/router/src/core/utils.rs
+++ b/crates/router/src/core/utils.rs
@@ -1,5 +1,7 @@
 use std::marker::PhantomData;
 
+use api_models::enums::{DisputeStage, DisputeStatus};
+use common_utils::errors::CustomResult;
 use error_stack::ResultExt;
 use router_env::{instrument, tracing};
 
@@ -149,3 +151,67 @@ mod tests {
         assert_eq!(generated_id.len(), consts::ID_LENGTH + 4)
     }
 }
+
+// Dispute Stage can move linearly from PreDispute -> Dispute -> PreArbitration
+pub fn validate_dispute_stage(
+    prev_dispute_stage: &DisputeStage,
+    dispute_stage: &DisputeStage,
+) -> bool {
+    match prev_dispute_stage {
+        DisputeStage::PreDispute => true,
+        DisputeStage::Dispute => !matches!(dispute_stage, DisputeStage::PreDispute),
+        DisputeStage::PreArbitration => matches!(dispute_stage, DisputeStage::PreArbitration),
+    }
+}
+
+//Dispute status can go from Opened -> (Expired | Accepted | Cancelled | Challenged -> (Won | Lost))
+pub fn validate_dispute_status(
+    prev_dispute_status: DisputeStatus,
+    dispute_status: DisputeStatus,
+) -> bool {
+    match prev_dispute_status {
+        DisputeStatus::DisputeOpened => true,
+        DisputeStatus::DisputeExpired => {
+            matches!(dispute_status, DisputeStatus::DisputeExpired)
+        }
+        DisputeStatus::DisputeAccepted => {
+            matches!(dispute_status, DisputeStatus::DisputeAccepted)
+        }
+        DisputeStatus::DisputeCancelled => {
+            matches!(dispute_status, DisputeStatus::DisputeCancelled)
+        }
+        DisputeStatus::DisputeChallenged => matches!(
+            dispute_status,
+            DisputeStatus::DisputeChallenged
+                | DisputeStatus::DisputeWon
+                | DisputeStatus::DisputeLost
+        ),
+        DisputeStatus::DisputeWon => matches!(dispute_status, DisputeStatus::DisputeWon),
+        DisputeStatus::DisputeLost => matches!(dispute_status, DisputeStatus::DisputeLost),
+    }
+}
+
+pub fn validate_dispute_stage_and_dispute_status(
+    prev_dispute_stage: DisputeStage,
+    prev_dispute_status: DisputeStatus,
+    dispute_stage: DisputeStage,
+    dispute_status: DisputeStatus,
+) -> CustomResult<(), errors::WebhooksFlowError> {
+    let dispute_stage_validation = validate_dispute_stage(&prev_dispute_stage, &dispute_stage);
+    let dispute_status_validation = if dispute_stage == prev_dispute_stage {
+        validate_dispute_status(prev_dispute_status, dispute_status)
+    } else {
+        true
+    };
+    common_utils::fp_utils::when(
+        !(dispute_stage_validation && dispute_status_validation),
+        || {
+            super::metrics::INCOMING_DISPUTE_WEBHOOK_VALIDATION_FAILURE_METRIC.add(
+                &super::metrics::CONTEXT,
+                1,
+                &[],
+            );
+            Err(errors::WebhooksFlowError::DisputeWebhookValidationFailed)?
+        },
+    )
+}
diff --git a/crates/router/src/core/webhooks.rs b/crates/router/src/core/webhooks.rs
index 72bda6d1bbf..5c8b1717d4e 100644
--- a/crates/router/src/core/webhooks.rs
+++ b/crates/router/src/core/webhooks.rs
@@ -7,6 +7,7 @@ use error_stack::{IntoReport, ResultExt};
 use masking::ExposeInterface;
 use router_env::{instrument, tracing};
 
+use super::metrics;
 use crate::{
     consts,
     core::{
@@ -197,6 +198,173 @@ async fn refunds_incoming_webhook_flow<W: api::OutgoingWebhookType>(
     Ok(())
 }
 
+async fn get_payment_attempt_from_object_reference_id(
+    state: AppState,
+    object_reference_id: api_models::webhooks::ObjectReferenceId,
+    merchant_account: &storage::MerchantAccount,
+) -> CustomResult<storage_models::payment_attempt::PaymentAttempt, errors::WebhooksFlowError> {
+    let db = &*state.store;
+    match object_reference_id {
+        api::ObjectReferenceId::PaymentId(api::PaymentIdType::ConnectorTransactionId(ref id)) => db
+            .find_payment_attempt_by_merchant_id_connector_txn_id(
+                &merchant_account.merchant_id,
+                id,
+                merchant_account.storage_scheme,
+            )
+            .await
+            .change_context(errors::WebhooksFlowError::ResourceNotFound),
+        api::ObjectReferenceId::PaymentId(api::PaymentIdType::PaymentAttemptId(ref id)) => db
+            .find_payment_attempt_by_merchant_id_attempt_id(
+                &merchant_account.merchant_id,
+                id,
+                merchant_account.storage_scheme,
+            )
+            .await
+            .change_context(errors::WebhooksFlowError::ResourceNotFound),
+        _ => Err(errors::WebhooksFlowError::ResourceNotFound).into_report(),
+    }
+}
+
+async fn get_or_update_dispute_object(
+    state: AppState,
+    option_dispute: Option<storage_models::dispute::Dispute>,
+    dispute_details: api::disputes::DisputePayload,
+    merchant_id: &str,
+    payment_id: &str,
+    attempt_id: &str,
+    event_type: api_models::webhooks::IncomingWebhookEvent,
+) -> CustomResult<storage_models::dispute::Dispute, errors::WebhooksFlowError> {
+    let db = &*state.store;
+    match option_dispute {
+        None => {
+            metrics::INCOMING_DISPUTE_WEBHOOK_NEW_RECORD_METRIC.add(&metrics::CONTEXT, 1, &[]);
+            let dispute_id = generate_id(consts::ID_LENGTH, "dp");
+            let new_dispute = storage_models::dispute::DisputeNew {
+                dispute_id,
+                amount: dispute_details.amount,
+                currency: dispute_details.currency,
+                dispute_stage: dispute_details.dispute_stage.foreign_into(),
+                dispute_status: event_type
+                    .foreign_try_into()
+                    .into_report()
+                    .change_context(errors::WebhooksFlowError::DisputeCoreFailed)?,
+                payment_id: payment_id.to_owned(),
+                attempt_id: attempt_id.to_owned(),
+                merchant_id: merchant_id.to_owned(),
+                connector_status: dispute_details.connector_status,
+                connector_dispute_id: dispute_details.connector_dispute_id,
+                connector_reason: dispute_details.connector_reason,
+                connector_reason_code: dispute_details.connector_reason_code,
+                challenge_required_by: dispute_details.challenge_required_by,
+                dispute_created_at: dispute_details.created_at,
+                updated_at: dispute_details.updated_at,
+            };
+            state
+                .store
+                .insert_dispute(new_dispute.clone())
+                .await
+                .change_context(errors::WebhooksFlowError::WebhookEventCreationFailed)
+        }
+        Some(dispute) => {
+            logger::info!("Dispute Already exists, Updating the dispute details");
+            metrics::INCOMING_DISPUTE_WEBHOOK_UPDATE_RECORD_METRIC.add(&metrics::CONTEXT, 1, &[]);
+            let dispute_status: storage_models::enums::DisputeStatus = event_type
+                .foreign_try_into()
+                .into_report()
+                .change_context(errors::WebhooksFlowError::DisputeCoreFailed)?;
+            crate::core::utils::validate_dispute_stage_and_dispute_status(
+                dispute.dispute_stage.foreign_into(),
+                dispute.dispute_status.foreign_into(),
+                dispute_details.dispute_stage.clone(),
+                dispute_status.foreign_into(),
+            )?;
+            let update_dispute = storage_models::dispute::DisputeUpdate::Update {
+                dispute_stage: dispute_details.dispute_stage.foreign_into(),
+                dispute_status,
+                connector_status: dispute_details.connector_status,
+                connector_reason: dispute_details.connector_reason,
+                connector_reason_code: dispute_details.connector_reason_code,
+                challenge_required_by: dispute_details.challenge_required_by,
+                updated_at: dispute_details.updated_at,
+            };
+            db.update_dispute(dispute, update_dispute)
+                .await
+                .change_context(errors::WebhooksFlowError::ResourceNotFound)
+        }
+    }
+}
+
+#[instrument(skip_all)]
+async fn disputes_incoming_webhook_flow<W: api::OutgoingWebhookType>(
+    state: AppState,
+    merchant_account: storage::MerchantAccount,
+    webhook_details: api::IncomingWebhookDetails,
+    source_verified: bool,
+    connector: &(dyn api::Connector + Sync),
+    request_details: &api_models::webhooks::IncomingWebhookRequestDetails<'_>,
+    event_type: api_models::webhooks::IncomingWebhookEvent,
+) -> CustomResult<(), errors::WebhooksFlowError> {
+    metrics::INCOMING_DISPUTE_WEBHOOK_METRIC.add(&metrics::CONTEXT, 1, &[]);
+    if source_verified {
+        let db = &*state.store;
+        let dispute_details = connector
+            .get_dispute_details(request_details)
+            .change_context(errors::WebhooksFlowError::WebhookEventObjectCreationFailed)?;
+        let payment_attempt = get_payment_attempt_from_object_reference_id(
+            state.clone(),
+            webhook_details.object_reference_id,
+            &merchant_account,
+        )
+        .await?;
+        let option_dispute = db
+            .find_by_merchant_id_payment_id_connector_dispute_id(
+                &merchant_account.merchant_id,
+                &payment_attempt.payment_id,
+                &dispute_details.connector_dispute_id,
+            )
+            .await
+            .change_context(errors::WebhooksFlowError::ResourceNotFound)?;
+        let dispute_object = get_or_update_dispute_object(
+            state.clone(),
+            option_dispute,
+            dispute_details,
+            &merchant_account.merchant_id,
+            &payment_attempt.payment_id,
+            &payment_attempt.attempt_id,
+            event_type.clone(),
+        )
+        .await?;
+        let disputes_response = Box::new(
+            dispute_object
+                .clone()
+                .foreign_try_into()
+                .into_report()
+                .change_context(errors::WebhooksFlowError::DisputeCoreFailed)?,
+        );
+        let event_type: enums::EventType = dispute_object
+            .dispute_status
+            .foreign_try_into()
+            .into_report()
+            .change_context(errors::WebhooksFlowError::DisputeCoreFailed)?;
+        create_event_and_trigger_outgoing_webhook::<W>(
+            state,
+            merchant_account,
+            event_type,
+            enums::EventClass::Disputes,
+            None,
+            dispute_object.dispute_id,
+            enums::EventObjectType::DisputeDetails,
+            api::OutgoingWebhookContent::DisputeDetails(disputes_response),
+        )
+        .await?;
+        metrics::INCOMING_DISPUTE_WEBHOOK_MERCHANT_NOTIFIED_METRIC.add(&metrics::CONTEXT, 1, &[]);
+        Ok(())
+    } else {
+        metrics::INCOMING_DISPUTE_WEBHOOK_SIGNATURE_FAILURE_METRIC.add(&metrics::CONTEXT, 1, &[]);
+        Err(errors::WebhooksFlowError::WebhookSourceVerificationFailed).into_report()
+    }
+}
+
 #[allow(clippy::too_many_arguments)]
 #[instrument(skip_all)]
 async fn create_event_and_trigger_outgoing_webhook<W: api::OutgoingWebhookType>(
@@ -329,7 +497,7 @@ pub async fn webhooks_core<W: api::OutgoingWebhookType>(
         .change_context(errors::ApiErrorResponse::InternalServerError)
         .attach_printable("There was an error in parsing the query params")?;
 
-    let mut request_details = api::IncomingWebhookRequestDetails {
+    let mut request_details = api_models::webhooks::IncomingWebhookRequestDetails {
         method: req.method().clone(),
         headers: req.headers(),
         query_params: req.query_string().to_string(),
@@ -420,6 +588,19 @@ pub async fn webhooks_core<W: api::OutgoingWebhookType>(
                 .change_context(errors::ApiErrorResponse::InternalServerError)
                 .attach_printable("Incoming webhook flow for refunds failed")?,
 
+                api::WebhookFlow::Dispute => disputes_incoming_webhook_flow::<W>(
+                    state.clone(),
+                    merchant_account,
+                    webhook_details,
+                    source_verified,
+                    *connector,
+                    &request_details,
+                    event_type,
+                )
+                .await
+                .change_context(errors::ApiErrorResponse::InternalServerError)
+                .attach_printable("Incoming webhook flow for disputes failed")?,
+
                 api::WebhookFlow::ReturnResponse => {}
 
                 _ => Err(errors::ApiErrorResponse::InternalServerError)
diff --git a/crates/router/src/db.rs b/crates/router/src/db.rs
index 6653167e442..9e8c08f1d2e 100644
--- a/crates/router/src/db.rs
+++ b/crates/router/src/db.rs
@@ -5,6 +5,7 @@ pub mod cards_info;
 pub mod configs;
 pub mod connector_response;
 pub mod customers;
+pub mod dispute;
 pub mod ephemeral_key;
 pub mod events;
 pub mod locker_mock_up;
@@ -42,6 +43,7 @@ pub trait StorageInterface:
     + configs::ConfigInterface
     + connector_response::ConnectorResponseInterface
     + customers::CustomerInterface
+    + dispute::DisputeInterface
     + ephemeral_key::EphemeralKeyInterface
     + events::EventInterface
     + locker_mock_up::LockerMockUpInterface
diff --git a/crates/router/src/db/dispute.rs b/crates/router/src/db/dispute.rs
new file mode 100644
index 00000000000..d4ea14d135d
--- /dev/null
+++ b/crates/router/src/db/dispute.rs
@@ -0,0 +1,103 @@
+use error_stack::IntoReport;
+
+use super::{MockDb, Store};
+use crate::{
+    connection,
+    core::errors::{self, CustomResult},
+    types::storage,
+};
+
+#[async_trait::async_trait]
+pub trait DisputeInterface {
+    async fn insert_dispute(
+        &self,
+        dispute: storage::DisputeNew,
+    ) -> CustomResult<storage::Dispute, errors::StorageError>;
+
+    async fn find_by_merchant_id_payment_id_connector_dispute_id(
+        &self,
+        merchant_id: &str,
+        payment_id: &str,
+        connector_dispute_id: &str,
+    ) -> CustomResult<Option<storage::Dispute>, errors::StorageError>;
+
+    async fn update_dispute(
+        &self,
+        this: storage::Dispute,
+        dispute: storage::DisputeUpdate,
+    ) -> CustomResult<storage::Dispute, errors::StorageError>;
+}
+
+#[async_trait::async_trait]
+impl DisputeInterface for Store {
+    async fn insert_dispute(
+        &self,
+        dispute: storage::DisputeNew,
+    ) -> CustomResult<storage::Dispute, errors::StorageError> {
+        let conn = connection::pg_connection_write(self).await?;
+        dispute
+            .insert(&conn)
+            .await
+            .map_err(Into::into)
+            .into_report()
+    }
+
+    async fn find_by_merchant_id_payment_id_connector_dispute_id(
+        &self,
+        merchant_id: &str,
+        payment_id: &str,
+        connector_dispute_id: &str,
+    ) -> CustomResult<Option<storage::Dispute>, errors::StorageError> {
+        let conn = connection::pg_connection_read(self).await?;
+        storage::Dispute::find_by_merchant_id_payment_id_connector_dispute_id(
+            &conn,
+            merchant_id,
+            payment_id,
+            connector_dispute_id,
+        )
+        .await
+        .map_err(Into::into)
+        .into_report()
+    }
+
+    async fn update_dispute(
+        &self,
+        this: storage::Dispute,
+        dispute: storage::DisputeUpdate,
+    ) -> CustomResult<storage::Dispute, errors::StorageError> {
+        let conn = connection::pg_connection_write(self).await?;
+        this.update(&conn, dispute)
+            .await
+            .map_err(Into::into)
+            .into_report()
+    }
+}
+
+#[async_trait::async_trait]
+impl DisputeInterface for MockDb {
+    async fn insert_dispute(
+        &self,
+        _dispute: storage::DisputeNew,
+    ) -> CustomResult<storage::Dispute, errors::StorageError> {
+        // TODO: Implement function for `MockDb`
+        Err(errors::StorageError::MockDbError)?
+    }
+    async fn find_by_merchant_id_payment_id_connector_dispute_id(
+        &self,
+        _merchant_id: &str,
+        _payment_id: &str,
+        _connector_dispute_id: &str,
+    ) -> CustomResult<Option<storage::Dispute>, errors::StorageError> {
+        // TODO: Implement function for `MockDb`
+        Err(errors::StorageError::MockDbError)?
+    }
+
+    async fn update_dispute(
+        &self,
+        _this: storage::Dispute,
+        _dispute: storage::DisputeUpdate,
+    ) -> CustomResult<storage::Dispute, errors::StorageError> {
+        // TODO: Implement function for `MockDb`
+        Err(errors::StorageError::MockDbError)?
+    }
+}
diff --git a/crates/router/src/types/api.rs b/crates/router/src/types/api.rs
index 2e9147ebf16..654800c3c87 100644
--- a/crates/router/src/types/api.rs
+++ b/crates/router/src/types/api.rs
@@ -2,6 +2,7 @@ pub mod admin;
 pub mod api_keys;
 pub mod configs;
 pub mod customers;
+pub mod disputes;
 pub mod enums;
 pub mod mandates;
 pub mod payment_methods;
diff --git a/crates/router/src/types/api/disputes.rs b/crates/router/src/types/api/disputes.rs
index e69de29bb2d..aba9a915fd9 100644
--- a/crates/router/src/types/api/disputes.rs
+++ b/crates/router/src/types/api/disputes.rs
@@ -0,0 +1,15 @@
+use masking::Deserialize;
+
+#[derive(Default, Debug, Deserialize)]
+pub struct DisputePayload {
+    pub amount: String,
+    pub currency: String,
+    pub dispute_stage: api_models::enums::DisputeStage,
+    pub connector_status: String,
+    pub connector_dispute_id: String,
+    pub connector_reason: Option<String>,
+    pub connector_reason_code: Option<String>,
+    pub challenge_required_by: Option<String>,
+    pub created_at: Option<String>,
+    pub updated_at: Option<String>,
+}
diff --git a/crates/router/src/types/api/webhooks.rs b/crates/router/src/types/api/webhooks.rs
index df8b8089635..0647ee0e923 100644
--- a/crates/router/src/types/api/webhooks.rs
+++ b/crates/router/src/types/api/webhooks.rs
@@ -137,4 +137,11 @@ pub trait IncomingWebhook: ConnectorCommon + Sync {
     {
         Ok(services::api::ApplicationResponse::StatusOk)
     }
+
+    fn get_dispute_details(
+        &self,
+        _request: &IncomingWebhookRequestDetails<'_>,
+    ) -> CustomResult<super::disputes::DisputePayload, errors::ConnectorError> {
+        Err(errors::ConnectorError::NotImplemented("get_dispute_details method".to_string()).into())
+    }
 }
diff --git a/crates/router/src/types/storage.rs b/crates/router/src/types/storage.rs
index 59067cdaec3..55bed000732 100644
--- a/crates/router/src/types/storage.rs
+++ b/crates/router/src/types/storage.rs
@@ -4,6 +4,7 @@ pub mod cards_info;
 pub mod configs;
 pub mod connector_response;
 pub mod customers;
+pub mod dispute;
 pub mod enums;
 pub mod ephemeral_key;
 pub mod events;
@@ -25,7 +26,7 @@ pub mod kv;
 
 pub use self::{
     address::*, api_keys::*, cards_info::*, configs::*, connector_response::*, customers::*,
-    events::*, locker_mock_up::*, mandate::*, merchant_account::*, merchant_connector_account::*,
-    payment_attempt::*, payment_intent::*, payment_method::*, process_tracker::*, refund::*,
-    reverse_lookup::*,
+    dispute::*, events::*, locker_mock_up::*, mandate::*, merchant_account::*,
+    merchant_connector_account::*, payment_attempt::*, payment_intent::*, payment_method::*,
+    process_tracker::*, refund::*, reverse_lookup::*,
 };
diff --git a/crates/router/src/types/storage/dispute.rs b/crates/router/src/types/storage/dispute.rs
index e69de29bb2d..5052feaa276 100644
--- a/crates/router/src/types/storage/dispute.rs
+++ b/crates/router/src/types/storage/dispute.rs
@@ -0,0 +1 @@
+pub use storage_models::dispute::{Dispute, DisputeNew, DisputeUpdate};
diff --git a/crates/router/src/types/transformers.rs b/crates/router/src/types/transformers.rs
index ea8150b9cd0..876dac7a7a6 100644
--- a/crates/router/src/types/transformers.rs
+++ b/crates/router/src/types/transformers.rs
@@ -177,6 +177,22 @@ impl ForeignTryFrom<storage_enums::RefundStatus> for storage_enums::EventType {
     }
 }
 
+impl ForeignTryFrom<storage_enums::DisputeStatus> for storage_enums::EventType {
+    type Error = errors::ValidationError;
+
+    fn foreign_try_from(value: storage_enums::DisputeStatus) -> Result<Self, Self::Error> {
+        match value {
+            storage_enums::DisputeStatus::DisputeOpened => Ok(Self::DisputeOpened),
+            storage_enums::DisputeStatus::DisputeExpired => Ok(Self::DisputeExpired),
+            storage_enums::DisputeStatus::DisputeAccepted => Ok(Self::DisputeAccepted),
+            storage_enums::DisputeStatus::DisputeCancelled => Ok(Self::DisputeCancelled),
+            storage_enums::DisputeStatus::DisputeChallenged => Ok(Self::DisputeChallenged),
+            storage_enums::DisputeStatus::DisputeWon => Ok(Self::DisputeWon),
+            storage_enums::DisputeStatus::DisputeLost => Ok(Self::DisputeLost),
+        }
+    }
+}
+
 impl ForeignTryFrom<api_models::webhooks::IncomingWebhookEvent> for storage_enums::RefundStatus {
     type Error = errors::ValidationError;
 
@@ -436,6 +452,81 @@ impl ForeignFrom<storage_enums::AttemptStatus> for api_enums::AttemptStatus {
     }
 }
 
+impl ForeignFrom<api_enums::DisputeStage> for storage_enums::DisputeStage {
+    fn foreign_from(status: api_enums::DisputeStage) -> Self {
+        frunk::labelled_convert_from(status)
+    }
+}
+
+impl ForeignFrom<api_enums::DisputeStatus> for storage_enums::DisputeStatus {
+    fn foreign_from(status: api_enums::DisputeStatus) -> Self {
+        frunk::labelled_convert_from(status)
+    }
+}
+
+impl ForeignFrom<storage_enums::DisputeStage> for api_enums::DisputeStage {
+    fn foreign_from(status: storage_enums::DisputeStage) -> Self {
+        frunk::labelled_convert_from(status)
+    }
+}
+
+impl ForeignFrom<storage_enums::DisputeStatus> for api_enums::DisputeStatus {
+    fn foreign_from(status: storage_enums::DisputeStatus) -> Self {
+        frunk::labelled_convert_from(status)
+    }
+}
+
+impl ForeignTryFrom<api_models::webhooks::IncomingWebhookEvent> for storage_enums::DisputeStatus {
+    type Error = errors::ValidationError;
+
+    fn foreign_try_from(
+        value: api_models::webhooks::IncomingWebhookEvent,
+    ) -> Result<Self, Self::Error> {
+        match value {
+            api_models::webhooks::IncomingWebhookEvent::DisputeOpened => Ok(Self::DisputeOpened),
+            api_models::webhooks::IncomingWebhookEvent::DisputeExpired => Ok(Self::DisputeExpired),
+            api_models::webhooks::IncomingWebhookEvent::DisputeAccepted => {
+                Ok(Self::DisputeAccepted)
+            }
+            api_models::webhooks::IncomingWebhookEvent::DisputeCancelled => {
+                Ok(Self::DisputeCancelled)
+            }
+            api_models::webhooks::IncomingWebhookEvent::DisputeChallenged => {
+                Ok(Self::DisputeChallenged)
+            }
+            api_models::webhooks::IncomingWebhookEvent::DisputeWon => Ok(Self::DisputeWon),
+            api_models::webhooks::IncomingWebhookEvent::DisputeLost => Ok(Self::DisputeLost),
+            _ => Err(errors::ValidationError::IncorrectValueProvided {
+                field_name: "incoming_webhook_event",
+            }),
+        }
+    }
+}
+
+impl ForeignTryFrom<storage::Dispute> for api_models::disputes::DisputeResponse {
+    type Error = errors::ValidationError;
+
+    fn foreign_try_from(dispute: storage::Dispute) -> Result<Self, Self::Error> {
+        Ok(Self {
+            dispute_id: dispute.dispute_id,
+            payment_id: dispute.payment_id,
+            attempt_id: dispute.attempt_id,
+            amount: dispute.amount,
+            currency: dispute.currency,
+            dispute_stage: dispute.dispute_stage.foreign_into(),
+            dispute_status: dispute.dispute_status.foreign_into(),
+            connector_status: dispute.connector_status,
+            connector_dispute_id: dispute.connector_dispute_id,
+            connector_reason: dispute.connector_reason,
+            connector_reason_code: dispute.connector_reason_code,
+            challenge_required_by: dispute.challenge_required_by,
+            created_at: dispute.dispute_created_at,
+            updated_at: dispute.updated_at,
+            received_at: dispute.created_at.to_string(),
+        })
+    }
+}
+
 impl ForeignFrom<storage_models::cards_info::CardInfo>
     for api_models::cards_info::CardInfoResponse
 {
diff --git a/crates/storage_models/src/dispute.rs b/crates/storage_models/src/dispute.rs
index 8b137891791..ec70ca24263 100644
--- a/crates/storage_models/src/dispute.rs
+++ b/crates/storage_models/src/dispute.rs
@@ -1 +1,104 @@
+use common_utils::custom_serde;
+use diesel::{AsChangeset, Identifiable, Insertable, Queryable};
+use serde::{Deserialize, Serialize};
+use time::PrimitiveDateTime;
 
+use crate::{enums as storage_enums, schema::dispute};
+
+#[derive(Clone, Debug, Deserialize, Insertable, Serialize, router_derive::DebugAsDisplay)]
+#[diesel(table_name = dispute)]
+#[serde(deny_unknown_fields)]
+pub struct DisputeNew {
+    pub dispute_id: String,
+    pub amount: String,
+    pub currency: String,
+    pub dispute_stage: storage_enums::DisputeStage,
+    pub dispute_status: storage_enums::DisputeStatus,
+    pub payment_id: String,
+    pub attempt_id: String,
+    pub merchant_id: String,
+    pub connector_status: String,
+    pub connector_dispute_id: String,
+    pub connector_reason: Option<String>,
+    pub connector_reason_code: Option<String>,
+    pub challenge_required_by: Option<String>,
+    pub dispute_created_at: Option<String>,
+    pub updated_at: Option<String>,
+}
+
+#[derive(Clone, Debug, Deserialize, Serialize, Identifiable, Queryable)]
+#[diesel(table_name = dispute)]
+pub struct Dispute {
+    #[serde(skip_serializing)]
+    pub id: i32,
+    pub dispute_id: String,
+    pub amount: String,
+    pub currency: String,
+    pub dispute_stage: storage_enums::DisputeStage,
+    pub dispute_status: storage_enums::DisputeStatus,
+    pub payment_id: String,
+    pub attempt_id: String,
+    pub merchant_id: String,
+    pub connector_status: String,
+    pub connector_dispute_id: String,
+    pub connector_reason: Option<String>,
+    pub connector_reason_code: Option<String>,
+    pub challenge_required_by: Option<String>,
+    pub dispute_created_at: Option<String>,
+    pub updated_at: Option<String>,
+    #[serde(with = "custom_serde::iso8601")]
+    pub created_at: PrimitiveDateTime,
+    #[serde(with = "custom_serde::iso8601")]
+    pub modified_at: PrimitiveDateTime,
+}
+
+#[derive(Debug)]
+pub enum DisputeUpdate {
+    Update {
+        dispute_stage: storage_enums::DisputeStage,
+        dispute_status: storage_enums::DisputeStatus,
+        connector_status: String,
+        connector_reason: Option<String>,
+        connector_reason_code: Option<String>,
+        challenge_required_by: Option<String>,
+        updated_at: Option<String>,
+    },
+}
+
+#[derive(Clone, Debug, Default, AsChangeset, router_derive::DebugAsDisplay)]
+#[diesel(table_name = dispute)]
+pub struct DisputeUpdateInternal {
+    dispute_stage: storage_enums::DisputeStage,
+    dispute_status: storage_enums::DisputeStatus,
+    connector_status: String,
+    connector_reason: Option<String>,
+    connector_reason_code: Option<String>,
+    challenge_required_by: Option<String>,
+    updated_at: Option<String>,
+    modified_at: Option<PrimitiveDateTime>,
+}
+
+impl From<DisputeUpdate> for DisputeUpdateInternal {
+    fn from(merchant_account_update: DisputeUpdate) -> Self {
+        match merchant_account_update {
+            DisputeUpdate::Update {
+                dispute_stage,
+                dispute_status,
+                connector_status,
+                connector_reason,
+                connector_reason_code,
+                challenge_required_by,
+                updated_at,
+            } => Self {
+                dispute_stage,
+                dispute_status,
+                connector_status,
+                connector_reason,
+                connector_reason_code,
+                challenge_required_by,
+                updated_at,
+                modified_at: Some(common_utils::date_time::now()),
+            },
+        }
+    }
+}
diff --git a/crates/storage_models/src/enums.rs b/crates/storage_models/src/enums.rs
index c7068fe9e36..30b5276bca5 100644
--- a/crates/storage_models/src/enums.rs
+++ b/crates/storage_models/src/enums.rs
@@ -3,6 +3,7 @@ pub mod diesel_exports {
     pub use super::{
         DbAttemptStatus as AttemptStatus, DbAuthenticationType as AuthenticationType,
         DbCaptureMethod as CaptureMethod, DbConnectorType as ConnectorType, DbCurrency as Currency,
+        DbDisputeStage as DisputeStage, DbDisputeStatus as DisputeStatus,
         DbEventClass as EventClass, DbEventObjectType as EventObjectType, DbEventType as EventType,
         DbFutureUsage as FutureUsage, DbIntentStatus as IntentStatus,
         DbMandateStatus as MandateStatus, DbMandateType as MandateType,
@@ -273,6 +274,7 @@ pub enum Currency {
 pub enum EventClass {
     Payments,
     Refunds,
+    Disputes,
 }
 
 #[derive(
@@ -292,6 +294,7 @@ pub enum EventClass {
 pub enum EventObjectType {
     PaymentDetails,
     RefundDetails,
+    DisputeDetails,
 }
 
 #[derive(
@@ -313,6 +316,13 @@ pub enum EventType {
     PaymentSucceeded,
     RefundSucceeded,
     RefundFailed,
+    DisputeOpened,
+    DisputeExpired,
+    DisputeAccepted,
+    DisputeCancelled,
+    DisputeChallenged,
+    DisputeWon,
+    DisputeLost,
 }
 
 #[derive(
@@ -706,3 +716,53 @@ pub enum PaymentExperience {
     LinkWallet,
     InvokePaymentApp,
 }
+
+#[derive(
+    Clone,
+    Copy,
+    Debug,
+    Eq,
+    PartialEq,
+    Default,
+    serde::Deserialize,
+    serde::Serialize,
+    strum::Display,
+    strum::EnumString,
+    frunk::LabelledGeneric,
+)]
+#[router_derive::diesel_enum(storage_type = "pg_enum")]
+#[serde(rename_all = "snake_case")]
+#[strum(serialize_all = "snake_case")]
+pub enum DisputeStage {
+    PreDispute,
+    #[default]
+    Dispute,
+    PreArbitration,
+}
+
+#[derive(
+    Clone,
+    Copy,
+    Debug,
+    Eq,
+    PartialEq,
+    Default,
+    serde::Deserialize,
+    serde::Serialize,
+    strum::Display,
+    strum::EnumString,
+    frunk::LabelledGeneric,
+)]
+#[router_derive::diesel_enum(storage_type = "pg_enum")]
+#[serde(rename_all = "snake_case")]
+#[strum(serialize_all = "snake_case")]
+pub enum DisputeStatus {
+    #[default]
+    DisputeOpened,
+    DisputeExpired,
+    DisputeAccepted,
+    DisputeCancelled,
+    DisputeChallenged,
+    DisputeWon,
+    DisputeLost,
+}
diff --git a/crates/storage_models/src/query.rs b/crates/storage_models/src/query.rs
index 8f3ea34c81b..83baa4502c2 100644
--- a/crates/storage_models/src/query.rs
+++ b/crates/storage_models/src/query.rs
@@ -4,6 +4,7 @@ pub mod cards_info;
 pub mod configs;
 pub mod connector_response;
 pub mod customers;
+pub mod dispute;
 pub mod events;
 pub mod generics;
 pub mod locker_mock_up;
diff --git a/crates/storage_models/src/query/dispute.rs b/crates/storage_models/src/query/dispute.rs
new file mode 100644
index 00000000000..a61a58bc8e6
--- /dev/null
+++ b/crates/storage_models/src/query/dispute.rs
@@ -0,0 +1,58 @@
+use diesel::{associations::HasTable, BoolExpressionMethods, ExpressionMethods};
+use router_env::{instrument, tracing};
+
+use super::generics;
+use crate::{
+    dispute::{Dispute, DisputeNew, DisputeUpdate, DisputeUpdateInternal},
+    errors,
+    schema::dispute::dsl,
+    PgPooledConn, StorageResult,
+};
+
+impl DisputeNew {
+    #[instrument(skip(conn))]
+    pub async fn insert(self, conn: &PgPooledConn) -> StorageResult<Dispute> {
+        generics::generic_insert(conn, self).await
+    }
+}
+
+impl Dispute {
+    #[instrument(skip(conn))]
+    pub async fn find_by_merchant_id_payment_id_connector_dispute_id(
+        conn: &PgPooledConn,
+        merchant_id: &str,
+        payment_id: &str,
+        connector_dispute_id: &str,
+    ) -> StorageResult<Option<Self>> {
+        generics::generic_find_one_optional::<<Self as HasTable>::Table, _, _>(
+            conn,
+            dsl::merchant_id
+                .eq(merchant_id.to_owned())
+                .and(dsl::payment_id.eq(payment_id.to_owned()))
+                .and(dsl::connector_dispute_id.eq(connector_dispute_id.to_owned())),
+        )
+        .await
+    }
+
+    #[instrument(skip(conn))]
+    pub async fn update(self, conn: &PgPooledConn, dispute: DisputeUpdate) -> StorageResult<Self> {
+        match generics::generic_update_with_unique_predicate_get_result::<
+            <Self as HasTable>::Table,
+            _,
+            _,
+            _,
+        >(
+            conn,
+            dsl::dispute_id.eq(self.dispute_id.to_owned()),
+            DisputeUpdateInternal::from(dispute),
+        )
+        .await
+        {
+            Err(error) => match error.current_context() {
+                errors::DatabaseError::NoFieldsToUpdate => Ok(self),
+                _ => Err(error),
+            },
+            result => result,
+        }
+    }
+}
diff --git a/crates/storage_models/src/schema.rs b/crates/storage_models/src/schema.rs
index 50f313ec8d1..677a293b258 100644
--- a/crates/storage_models/src/schema.rs
+++ b/crates/storage_models/src/schema.rs
@@ -106,6 +106,32 @@ diesel::table! {
     }
 }
 
+diesel::table! {
+    use diesel::sql_types::*;
+    use crate::enums::diesel_exports::*;
+
+    dispute (id) {
+        id -> Int4,
+        dispute_id -> Varchar,
+        amount -> Varchar,
+        currency -> Varchar,
+        dispute_stage -> DisputeStage,
+        dispute_status -> DisputeStatus,
+        payment_id -> Varchar,
+        attempt_id -> Varchar,
+        merchant_id -> Varchar,
+        connector_status -> Varchar,
+        connector_dispute_id -> Varchar,
+        connector_reason -> Nullable<Varchar>,
+        connector_reason_code -> Nullable<Varchar>,
+        challenge_required_by -> Nullable<Varchar>,
+        dispute_created_at -> Nullable<Varchar>,
+        updated_at -> Nullable<Varchar>,
+        created_at -> Timestamp,
+        modified_at -> Timestamp,
+    }
+}
+
 diesel::table! {
     use diesel::sql_types::*;
     use crate::enums::diesel_exports::*;
@@ -387,6 +413,7 @@ diesel::allow_tables_to_appear_in_same_query!(
     configs,
     connector_response,
     customers,
+    dispute,
     events,
     locker_mock_up,
     mandate,
diff --git a/migrations/2023-03-15-185959_add_dispute_table/down.sql b/migrations/2023-03-15-185959_add_dispute_table/down.sql
new file mode 100644
index 00000000000..61ef6e480fe
--- /dev/null
+++ b/migrations/2023-03-15-185959_add_dispute_table/down.sql
@@ -0,0 +1,5 @@
+DROP TABLE dispute;
+
+DROP TYPE "DisputeStage";
+
+DROP TYPE "DisputeStatus";
diff --git a/migrations/2023-03-15-185959_add_dispute_table/up.sql b/migrations/2023-03-15-185959_add_dispute_table/up.sql
new file mode 100644
index 00000000000..046186d753d
--- /dev/null
+++ b/migrations/2023-03-15-185959_add_dispute_table/up.sql
@@ -0,0 +1,44 @@
+CREATE TYPE "DisputeStage" AS ENUM ('pre_dispute', 'dispute', 'pre_arbitration');
+
+CREATE TYPE "DisputeStatus" AS ENUM ('dispute_opened', 'dispute_expired', 'dispute_accepted', 'dispute_cancelled', 'dispute_challenged', 'dispute_won', 'dispute_lost');
+
+CREATE TABLE dispute (
+    id SERIAL PRIMARY KEY,
+    dispute_id VARCHAR(64) NOT NULL,
+    amount VARCHAR(255) NOT NULL,
+    currency VARCHAR(255) NOT NULL,
+    dispute_stage "DisputeStage" NOT NULL,
+    dispute_status "DisputeStatus" NOT NULL,
+    payment_id VARCHAR(255) NOT NULL,
+    attempt_id VARCHAR(64) NOT NULL,
+    merchant_id VARCHAR(255) NOT NULL,
+    connector_status VARCHAR(255) NOT NULL,
+    connector_dispute_id VARCHAR(255) NOT NULL,
+    connector_reason VARCHAR(255),
+    connector_reason_code VARCHAR(255),
+    challenge_required_by VARCHAR(255),
+    dispute_created_at VARCHAR(255),
+    updated_at VARCHAR(255),
+    created_at TIMESTAMP NOT NULL DEFAULT now()::TIMESTAMP,
+    modified_at TIMESTAMP NOT NULL DEFAULT now()::TIMESTAMP
+);
+
+CREATE UNIQUE INDEX dispute_id_index ON dispute (dispute_id);
+
+CREATE UNIQUE INDEX merchant_id_payment_id_connector_dispute_id_index ON dispute (merchant_id, payment_id, connector_dispute_id);
+
+CREATE INDEX dispute_status_index ON dispute (dispute_status);
+
+CREATE INDEX dispute_stage_index ON dispute (dispute_stage);
+
+ALTER TYPE "EventClass" ADD VALUE 'disputes';
+
+ALTER TYPE "EventObjectType" ADD VALUE 'dispute_details';
+
+ALTER TYPE "EventType" ADD VALUE 'dispute_opened';
+ALTER TYPE "EventType" ADD VALUE 'dispute_expired';
+ALTER TYPE "EventType" ADD VALUE 'dispute_accepted';
+ALTER TYPE "EventType" ADD VALUE 'dispute_cancelled';
+ALTER TYPE "EventType" ADD VALUE 'dispute_challenged';
+ALTER TYPE "EventType" ADD VALUE 'dispute_won';
+ALTER TYPE "EventType" ADD VALUE 'dispute_lost';