Issue:
Bug: [FEATURE]: Add diesel model support as a part of refunds implementation for v2 apis.



Add diesel model and storage impl support as a part of refunds implementation for v2 apis.

Pull request description:
## Type of Change
<!-- Put an `x` in the boxes that apply -->

- [ ] Bugfix
- [x] New feature
- [ ] Enhancement
- [ ] Refactoring
- [ ] Dependency updates
- [ ] Documentation
- [ ] CI/CD

## Description
This PR is a part of v2 implementation of refunds. This PR contains the necessary database schema refactoring as well as handling this in the storage_impl crate.


### Additional Changes

- [ ] This PR modifies the API contract
- [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 it is an obvious bug or documentation fix
that will have little conversation).
-->
Merging this PR would add db support for refunds in v2 apis.


## 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)?
-->
Cannot be tested because the core flow is not implemented for the refunds in V2 APIs, will be taking that in upcoming PRs.

## 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


Testing:

Cannot be tested because the core flow is not implemented for the refunds in V2 APIs, will be taking that in upcoming PRs.

Patch:
diff --git a/crates/diesel_models/Cargo.toml b/crates/diesel_models/Cargo.toml
index 4f02cc4aea8..94913571396 100644
--- a/crates/diesel_models/Cargo.toml
+++ b/crates/diesel_models/Cargo.toml
@@ -14,6 +14,7 @@ v1 = ["common_utils/v1"]
 v2 = ["common_utils/v2"]
 customer_v2 = []
 payment_methods_v2 = []
+refunds_v2 = []
 
 [dependencies]
 async-bb8-diesel = { git = "https://github.com/jarnura/async-bb8-diesel", rev = "53b4ab901aab7635c8215fd1c2d542c8db443094" }
diff --git a/crates/diesel_models/src/kv.rs b/crates/diesel_models/src/kv.rs
index 54e60891532..260b0e4c86a 100644
--- a/crates/diesel_models/src/kv.rs
+++ b/crates/diesel_models/src/kv.rs
@@ -133,9 +133,14 @@ impl DBOperation {
                         )
                         .await?,
                 )),
+                #[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "refunds_v2")))]
                 Updateable::RefundUpdate(a) => {
                     DBResult::Refund(Box::new(a.orig.update(conn, a.update_data).await?))
                 }
+                #[cfg(all(feature = "v2", feature = "refunds_v2"))]
+                Updateable::RefundUpdate(a) => {
+                    DBResult::Refund(Box::new(a.orig.update_with_id(conn, a.update_data).await?))
+                }
                 Updateable::AddressUpdate(a) => {
                     DBResult::Address(Box::new(a.orig.update(conn, a.update_data).await?))
                 }
diff --git a/crates/diesel_models/src/query/refund.rs b/crates/diesel_models/src/query/refund.rs
index 902916bf98f..adae21c413f 100644
--- a/crates/diesel_models/src/query/refund.rs
+++ b/crates/diesel_models/src/query/refund.rs
@@ -3,10 +3,19 @@ use diesel::{associations::HasTable, BoolExpressionMethods, ExpressionMethods, T
 use super::generics;
 use crate::{
     errors,
-    refund::{Refund, RefundNew, RefundUpdate, RefundUpdateInternal},
-    schema::refund::dsl,
+    refund::{RefundUpdate, RefundUpdateInternal},
     PgPooledConn, StorageResult,
 };
+#[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "refunds_v2")))]
+use crate::{
+    refund::{Refund, RefundNew},
+    schema::refund::dsl,
+};
+#[cfg(all(feature = "v2", feature = "refunds_v2"))]
+use crate::{
+    refund::{Refund, RefundNew},
+    schema_v2::refund::dsl,
+};
 
 impl RefundNew {
     pub async fn insert(self, conn: &PgPooledConn) -> StorageResult<Refund> {
@@ -14,6 +23,7 @@ impl RefundNew {
     }
 }
 
+#[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "refunds_v2")))]
 impl Refund {
     pub async fn update(self, conn: &PgPooledConn, refund: RefundUpdate) -> StorageResult<Self> {
         match generics::generic_update_with_unique_predicate_get_result::<
@@ -127,3 +137,37 @@ impl Refund {
         .await
     }
 }
+
+#[cfg(all(feature = "v2", feature = "refunds_v2"))]
+impl Refund {
+    pub async fn update_with_id(
+        self,
+        conn: &PgPooledConn,
+        refund: RefundUpdate,
+    ) -> StorageResult<Self> {
+        match generics::generic_update_by_id::<<Self as HasTable>::Table, _, _, _>(
+            conn,
+            self.id.to_owned(),
+            RefundUpdateInternal::from(refund),
+        )
+        .await
+        {
+            Err(error) => match error.current_context() {
+                errors::DatabaseError::NoFieldsToUpdate => Ok(self),
+                _ => Err(error),
+            },
+            result => result,
+        }
+    }
+
+    pub async fn find_by_global_id(
+        conn: &PgPooledConn,
+        id: &common_utils::id_type::GlobalRefundId,
+    ) -> StorageResult<Self> {
+        generics::generic_find_one::<<Self as HasTable>::Table, _, _>(
+            conn,
+            dsl::id.eq(id.to_owned()),
+        )
+        .await
+    }
+}
diff --git a/crates/diesel_models/src/query/user/sample_data.rs b/crates/diesel_models/src/query/user/sample_data.rs
index b97ea39b82d..8ef4b1e9303 100644
--- a/crates/diesel_models/src/query/user/sample_data.rs
+++ b/crates/diesel_models/src/query/user/sample_data.rs
@@ -6,16 +6,16 @@ use router_env::logger;
 #[cfg(feature = "v1")]
 use crate::schema::{
     payment_attempt::dsl as payment_attempt_dsl, payment_intent::dsl as payment_intent_dsl,
+    refund::dsl as refund_dsl,
 };
 #[cfg(feature = "v2")]
 use crate::schema_v2::{
     payment_attempt::dsl as payment_attempt_dsl, payment_intent::dsl as payment_intent_dsl,
+    refund::dsl as refund_dsl,
 };
 use crate::{
-    errors,
-    schema::{dispute::dsl as dispute_dsl, refund::dsl as refund_dsl},
-    user, Dispute, DisputeNew, PaymentAttempt, PaymentIntent, PaymentIntentNew, PgPooledConn,
-    Refund, RefundNew, StorageResult,
+    errors, schema::dispute::dsl as dispute_dsl, user, Dispute, DisputeNew, PaymentAttempt,
+    PaymentIntent, PaymentIntentNew, PgPooledConn, Refund, RefundNew, StorageResult,
 };
 
 #[cfg(feature = "v1")]
diff --git a/crates/diesel_models/src/refund.rs b/crates/diesel_models/src/refund.rs
index ee598bebd2f..6a40baba9dd 100644
--- a/crates/diesel_models/src/refund.rs
+++ b/crates/diesel_models/src/refund.rs
@@ -6,8 +6,13 @@ use diesel::{AsChangeset, Identifiable, Insertable, Queryable, Selectable};
 use serde::{Deserialize, Serialize};
 use time::PrimitiveDateTime;
 
-use crate::{enums as storage_enums, schema::refund};
+use crate::enums as storage_enums;
+#[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "refunds_v2")))]
+use crate::schema::refund;
+#[cfg(all(feature = "v2", feature = "refunds_v2"))]
+use crate::schema_v2::refund;
 
+#[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "refunds_v2")))]
 #[derive(
     Clone,
     Debug,
@@ -64,6 +69,58 @@ pub struct Refund {
     pub issuer_error_message: Option<String>,
 }
 
+#[cfg(all(feature = "v2", feature = "refunds_v2"))]
+#[derive(
+    Clone,
+    Debug,
+    Eq,
+    Identifiable,
+    Queryable,
+    Selectable,
+    PartialEq,
+    serde::Serialize,
+    serde::Deserialize,
+)]
+#[diesel(table_name = refund, primary_key(id), check_for_backend(diesel::pg::Pg))]
+pub struct Refund {
+    pub payment_id: common_utils::id_type::GlobalPaymentId,
+    pub merchant_id: common_utils::id_type::MerchantId,
+    pub connector_transaction_id: ConnectorTransactionId,
+    pub connector: String,
+    pub connector_refund_id: Option<ConnectorTransactionId>,
+    pub external_reference_id: Option<String>,
+    pub refund_type: storage_enums::RefundType,
+    pub total_amount: MinorUnit,
+    pub currency: storage_enums::Currency,
+    pub refund_amount: MinorUnit,
+    pub refund_status: storage_enums::RefundStatus,
+    pub sent_to_gateway: bool,
+    pub refund_error_message: Option<String>,
+    pub metadata: Option<pii::SecretSerdeValue>,
+    pub refund_arn: Option<String>,
+    #[serde(with = "common_utils::custom_serde::iso8601")]
+    pub created_at: PrimitiveDateTime,
+    #[serde(with = "common_utils::custom_serde::iso8601")]
+    pub modified_at: PrimitiveDateTime,
+    pub description: Option<String>,
+    pub attempt_id: common_utils::id_type::GlobalAttemptId,
+    pub refund_reason: Option<String>,
+    pub refund_error_code: Option<String>,
+    pub profile_id: Option<common_utils::id_type::ProfileId>,
+    pub updated_by: String,
+    pub charges: Option<ChargeRefunds>,
+    pub organization_id: common_utils::id_type::OrganizationId,
+    pub split_refunds: Option<common_types::refunds::SplitRefund>,
+    pub unified_code: Option<String>,
+    pub unified_message: Option<String>,
+    pub processor_refund_data: Option<String>,
+    pub processor_transaction_data: Option<String>,
+    pub id: common_utils::id_type::GlobalRefundId,
+    pub merchant_reference_id: common_utils::id_type::RefundReferenceId,
+    pub connector_id: Option<common_utils::id_type::MerchantConnectorAccountId>,
+}
+
+#[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "refunds_v2")))]
 #[derive(
     Clone,
     Debug,
@@ -110,6 +167,54 @@ pub struct RefundNew {
     pub processor_transaction_data: Option<String>,
 }
 
+#[cfg(all(feature = "v2", feature = "refunds_v2"))]
+#[derive(
+    Clone,
+    Debug,
+    Eq,
+    PartialEq,
+    Insertable,
+    router_derive::DebugAsDisplay,
+    serde::Serialize,
+    serde::Deserialize,
+    router_derive::Setter,
+)]
+#[diesel(table_name = refund)]
+pub struct RefundNew {
+    pub merchant_reference_id: common_utils::id_type::RefundReferenceId,
+    pub payment_id: common_utils::id_type::GlobalPaymentId,
+    pub merchant_id: common_utils::id_type::MerchantId,
+    pub id: common_utils::id_type::GlobalRefundId,
+    pub external_reference_id: Option<String>,
+    pub connector_transaction_id: ConnectorTransactionId,
+    pub connector: String,
+    pub connector_refund_id: Option<ConnectorTransactionId>,
+    pub refund_type: storage_enums::RefundType,
+    pub total_amount: MinorUnit,
+    pub currency: storage_enums::Currency,
+    pub refund_amount: MinorUnit,
+    pub refund_status: storage_enums::RefundStatus,
+    pub sent_to_gateway: bool,
+    pub metadata: Option<pii::SecretSerdeValue>,
+    pub refund_arn: Option<String>,
+    #[serde(with = "common_utils::custom_serde::iso8601")]
+    pub created_at: PrimitiveDateTime,
+    #[serde(with = "common_utils::custom_serde::iso8601")]
+    pub modified_at: PrimitiveDateTime,
+    pub description: Option<String>,
+    pub attempt_id: common_utils::id_type::GlobalAttemptId,
+    pub refund_reason: Option<String>,
+    pub profile_id: Option<common_utils::id_type::ProfileId>,
+    pub updated_by: String,
+    pub connector_id: Option<common_utils::id_type::MerchantConnectorAccountId>,
+    pub charges: Option<ChargeRefunds>,
+    pub organization_id: common_utils::id_type::OrganizationId,
+    pub split_refunds: Option<common_types::refunds::SplitRefund>,
+    pub processor_refund_data: Option<String>,
+    pub processor_transaction_data: Option<String>,
+}
+
+#[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "refunds_v2")))]
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 pub enum RefundUpdate {
     Update {
@@ -153,6 +258,49 @@ pub enum RefundUpdate {
     },
 }
 
+#[cfg(all(feature = "v2", feature = "refunds_v2"))]
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+pub enum RefundUpdate {
+    Update {
+        connector_refund_id: ConnectorTransactionId,
+        refund_status: storage_enums::RefundStatus,
+        sent_to_gateway: bool,
+        refund_error_message: Option<String>,
+        refund_arn: String,
+        updated_by: String,
+        processor_refund_data: Option<String>,
+    },
+    MetadataAndReasonUpdate {
+        metadata: Option<pii::SecretSerdeValue>,
+        reason: Option<String>,
+        updated_by: String,
+    },
+    StatusUpdate {
+        connector_refund_id: Option<ConnectorTransactionId>,
+        sent_to_gateway: bool,
+        refund_status: storage_enums::RefundStatus,
+        updated_by: String,
+        processor_refund_data: Option<String>,
+    },
+    ErrorUpdate {
+        refund_status: Option<storage_enums::RefundStatus>,
+        refund_error_message: Option<String>,
+        refund_error_code: Option<String>,
+        updated_by: String,
+        connector_refund_id: Option<ConnectorTransactionId>,
+        processor_refund_data: Option<String>,
+        unified_code: Option<String>,
+        unified_message: Option<String>,
+    },
+    ManualUpdate {
+        refund_status: Option<storage_enums::RefundStatus>,
+        refund_error_message: Option<String>,
+        refund_error_code: Option<String>,
+        updated_by: String,
+    },
+}
+
+#[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "refunds_v2")))]
 #[derive(Clone, Debug, AsChangeset, router_derive::DebugAsDisplay)]
 #[diesel(table_name = refund)]
 pub struct RefundUpdateInternal {
@@ -173,6 +321,48 @@ pub struct RefundUpdateInternal {
     issuer_error_message: Option<String>,
 }
 
+#[cfg(all(feature = "v2", feature = "refunds_v2"))]
+#[derive(Clone, Debug, AsChangeset, router_derive::DebugAsDisplay)]
+#[diesel(table_name = refund)]
+pub struct RefundUpdateInternal {
+    connector_refund_id: Option<ConnectorTransactionId>,
+    refund_status: Option<storage_enums::RefundStatus>,
+    sent_to_gateway: Option<bool>,
+    refund_error_message: Option<String>,
+    refund_arn: Option<String>,
+    metadata: Option<pii::SecretSerdeValue>,
+    refund_reason: Option<String>,
+    refund_error_code: Option<String>,
+    updated_by: String,
+    modified_at: PrimitiveDateTime,
+    processor_refund_data: Option<String>,
+    unified_code: Option<String>,
+    unified_message: Option<String>,
+}
+
+#[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "refunds_v2")))]
+impl RefundUpdateInternal {
+    pub fn create_refund(self, source: Refund) -> Refund {
+        Refund {
+            connector_refund_id: self.connector_refund_id,
+            refund_status: self.refund_status.unwrap_or_default(),
+            sent_to_gateway: self.sent_to_gateway.unwrap_or_default(),
+            refund_error_message: self.refund_error_message,
+            refund_arn: self.refund_arn,
+            metadata: self.metadata,
+            refund_reason: self.refund_reason,
+            refund_error_code: self.refund_error_code,
+            updated_by: self.updated_by,
+            modified_at: self.modified_at,
+            processor_refund_data: self.processor_refund_data,
+            unified_code: self.unified_code,
+            unified_message: self.unified_message,
+            ..source
+        }
+    }
+}
+
+#[cfg(all(feature = "v2", feature = "refunds_v2"))]
 impl RefundUpdateInternal {
     pub fn create_refund(self, source: Refund) -> Refund {
         Refund {
@@ -194,6 +384,7 @@ impl RefundUpdateInternal {
     }
 }
 
+#[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "refunds_v2")))]
 impl From<RefundUpdate> for RefundUpdateInternal {
     fn from(refund_update: RefundUpdate) -> Self {
         match refund_update {
@@ -320,6 +511,122 @@ impl From<RefundUpdate> for RefundUpdateInternal {
     }
 }
 
+#[cfg(all(feature = "v2", feature = "refunds_v2"))]
+impl From<RefundUpdate> for RefundUpdateInternal {
+    fn from(refund_update: RefundUpdate) -> Self {
+        match refund_update {
+            RefundUpdate::Update {
+                connector_refund_id,
+                refund_status,
+                sent_to_gateway,
+                refund_error_message,
+                refund_arn,
+                updated_by,
+                processor_refund_data,
+            } => Self {
+                connector_refund_id: Some(connector_refund_id),
+                refund_status: Some(refund_status),
+                sent_to_gateway: Some(sent_to_gateway),
+                refund_error_message,
+                refund_arn: Some(refund_arn),
+                updated_by,
+                processor_refund_data,
+                metadata: None,
+                refund_reason: None,
+                refund_error_code: None,
+                modified_at: common_utils::date_time::now(),
+                unified_code: None,
+                unified_message: None,
+            },
+            RefundUpdate::MetadataAndReasonUpdate {
+                metadata,
+                reason,
+                updated_by,
+            } => Self {
+                metadata,
+                refund_reason: reason,
+                updated_by,
+                connector_refund_id: None,
+                refund_status: None,
+                sent_to_gateway: None,
+                refund_error_message: None,
+                refund_arn: None,
+                refund_error_code: None,
+                modified_at: common_utils::date_time::now(),
+                processor_refund_data: None,
+                unified_code: None,
+                unified_message: None,
+            },
+            RefundUpdate::StatusUpdate {
+                connector_refund_id,
+                sent_to_gateway,
+                refund_status,
+                updated_by,
+                processor_refund_data,
+            } => Self {
+                connector_refund_id,
+                sent_to_gateway: Some(sent_to_gateway),
+                refund_status: Some(refund_status),
+                updated_by,
+                processor_refund_data,
+                refund_error_message: None,
+                refund_arn: None,
+                metadata: None,
+                refund_reason: None,
+                refund_error_code: None,
+                modified_at: common_utils::date_time::now(),
+                unified_code: None,
+                unified_message: None,
+            },
+            RefundUpdate::ErrorUpdate {
+                refund_status,
+                refund_error_message,
+                refund_error_code,
+                unified_code,
+                unified_message,
+                updated_by,
+                connector_refund_id,
+                processor_refund_data,
+            } => Self {
+                refund_status,
+                refund_error_message,
+                refund_error_code,
+                updated_by,
+                connector_refund_id,
+                processor_refund_data,
+                sent_to_gateway: None,
+                refund_arn: None,
+                metadata: None,
+                refund_reason: None,
+                modified_at: common_utils::date_time::now(),
+                unified_code,
+                unified_message,
+            },
+            RefundUpdate::ManualUpdate {
+                refund_status,
+                refund_error_message,
+                refund_error_code,
+                updated_by,
+            } => Self {
+                refund_status,
+                refund_error_message,
+                refund_error_code,
+                updated_by,
+                connector_refund_id: None,
+                sent_to_gateway: None,
+                refund_arn: None,
+                metadata: None,
+                refund_reason: None,
+                modified_at: common_utils::date_time::now(),
+                processor_refund_data: None,
+                unified_code: None,
+                unified_message: None,
+            },
+        }
+    }
+}
+
+#[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "refunds_v2")))]
 impl RefundUpdate {
     pub fn apply_changeset(self, source: Refund) -> Refund {
         let RefundUpdateInternal {
@@ -360,6 +667,43 @@ impl RefundUpdate {
     }
 }
 
+#[cfg(all(feature = "v2", feature = "refunds_v2"))]
+impl RefundUpdate {
+    pub fn apply_changeset(self, source: Refund) -> Refund {
+        let RefundUpdateInternal {
+            connector_refund_id,
+            refund_status,
+            sent_to_gateway,
+            refund_error_message,
+            refund_arn,
+            metadata,
+            refund_reason,
+            refund_error_code,
+            updated_by,
+            modified_at: _,
+            processor_refund_data,
+            unified_code,
+            unified_message,
+        } = self.into();
+        Refund {
+            connector_refund_id: connector_refund_id.or(source.connector_refund_id),
+            refund_status: refund_status.unwrap_or(source.refund_status),
+            sent_to_gateway: sent_to_gateway.unwrap_or(source.sent_to_gateway),
+            refund_error_message: refund_error_message.or(source.refund_error_message),
+            refund_error_code: refund_error_code.or(source.refund_error_code),
+            refund_arn: refund_arn.or(source.refund_arn),
+            metadata: metadata.or(source.metadata),
+            refund_reason: refund_reason.or(source.refund_reason),
+            updated_by,
+            modified_at: common_utils::date_time::now(),
+            processor_refund_data: processor_refund_data.or(source.processor_refund_data),
+            unified_code: unified_code.or(source.unified_code),
+            unified_message: unified_message.or(source.unified_message),
+            ..source
+        }
+    }
+}
+
 #[derive(Debug, Eq, PartialEq, Deserialize, Serialize)]
 pub struct RefundCoreWorkflow {
     pub refund_internal_reference_id: String,
@@ -379,6 +723,16 @@ impl common_utils::events::ApiEventMetric for Refund {
     }
 }
 
+#[cfg(feature = "v2")]
+impl common_utils::events::ApiEventMetric for Refund {
+    fn get_api_event_type(&self) -> Option<common_utils::events::ApiEventsType> {
+        Some(common_utils::events::ApiEventsType::Refund {
+            payment_id: self.payment_id.clone(),
+            refund_id: self.id.clone(),
+        })
+    }
+}
+
 impl ConnectorTransactionIdTrait for Refund {
     fn get_optional_connector_refund_id(&self) -> Option<&String> {
         match self
diff --git a/crates/diesel_models/src/schema_v2.rs b/crates/diesel_models/src/schema_v2.rs
index 5d028ad5214..d97d4fec68a 100644
--- a/crates/diesel_models/src/schema_v2.rs
+++ b/crates/diesel_models/src/schema_v2.rs
@@ -1160,11 +1160,7 @@ diesel::table! {
     use diesel::sql_types::*;
     use crate::enums::diesel_exports::*;
 
-    refund (merchant_id, refund_id) {
-        #[max_length = 64]
-        internal_reference_id -> Varchar,
-        #[max_length = 64]
-        refund_id -> Varchar,
+    refund (id) {
         #[max_length = 64]
         payment_id -> Varchar,
         #[max_length = 64]
@@ -1200,8 +1196,6 @@ diesel::table! {
         profile_id -> Nullable<Varchar>,
         #[max_length = 32]
         updated_by -> Varchar,
-        #[max_length = 32]
-        merchant_connector_id -> Nullable<Varchar>,
         charges -> Nullable<Jsonb>,
         #[max_length = 32]
         organization_id -> Varchar,
@@ -1212,6 +1206,12 @@ diesel::table! {
         unified_message -> Nullable<Varchar>,
         processor_refund_data -> Nullable<Text>,
         processor_transaction_data -> Nullable<Text>,
+        #[max_length = 64]
+        id -> Varchar,
+        #[max_length = 64]
+        merchant_reference_id -> Varchar,
+        #[max_length = 64]
+        connector_id -> Nullable<Varchar>,
     }
 }
 
diff --git a/crates/router/Cargo.toml b/crates/router/Cargo.toml
index 2012028169c..780d4509db7 100644
--- a/crates/router/Cargo.toml
+++ b/crates/router/Cargo.toml
@@ -33,12 +33,13 @@ payouts = ["api_models/payouts", "common_enums/payouts", "hyperswitch_connectors
 payout_retry = ["payouts"]
 recon = ["email", "api_models/recon"]
 retry = []
-v2 = ["customer_v2", "payment_methods_v2", "common_default", "api_models/v2", "diesel_models/v2", "hyperswitch_domain_models/v2", "storage_impl/v2", "kgraph_utils/v2", "common_utils/v2", "hyperswitch_connectors/v2","hyperswitch_interfaces/v2", "common_types/v2","revenue_recovery","scheduler/v2"]
+v2 = ["customer_v2", "payment_methods_v2", "common_default", "api_models/v2", "diesel_models/v2", "hyperswitch_domain_models/v2", "storage_impl/v2", "kgraph_utils/v2", "common_utils/v2", "hyperswitch_connectors/v2","hyperswitch_interfaces/v2", "common_types/v2","revenue_recovery","refunds_v2","scheduler/v2"]
 v1 = ["common_default", "api_models/v1", "diesel_models/v1", "hyperswitch_domain_models/v1", "storage_impl/v1", "hyperswitch_interfaces/v1", "kgraph_utils/v1", "common_utils/v1", "hyperswitch_connectors/v1", "common_types/v1","scheduler/v1"]
 customer_v2 = ["api_models/customer_v2", "diesel_models/customer_v2", "hyperswitch_domain_models/customer_v2", "storage_impl/customer_v2"]
 payment_methods_v2 = ["api_models/payment_methods_v2", "diesel_models/payment_methods_v2", "hyperswitch_domain_models/payment_methods_v2", "storage_impl/payment_methods_v2", "common_utils/payment_methods_v2"]
 dynamic_routing = ["external_services/dynamic_routing", "storage_impl/dynamic_routing", "api_models/dynamic_routing"]
 revenue_recovery =["api_models/revenue_recovery","hyperswitch_interfaces/revenue_recovery","hyperswitch_domain_models/revenue_recovery","hyperswitch_connectors/revenue_recovery"]
+refunds_v2 = ["diesel_models/refunds_v2", "storage_impl/refunds_v2"]
 
 # Partial Auth
 # The feature reduces the overhead of the router authenticating the merchant for every request, and trusts on `x-merchant-id` header to be present in the request.
diff --git a/crates/router/src/db/kafka_store.rs b/crates/router/src/db/kafka_store.rs
index d06d13a3123..9b72ba26c26 100644
--- a/crates/router/src/db/kafka_store.rs
+++ b/crates/router/src/db/kafka_store.rs
@@ -2647,6 +2647,7 @@ impl CaptureInterface for KafkaStore {
 
 #[async_trait::async_trait]
 impl RefundInterface for KafkaStore {
+    #[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "refunds_v2")))]
     async fn find_refund_by_internal_reference_id_merchant_id(
         &self,
         internal_reference_id: &str,
@@ -2662,6 +2663,7 @@ impl RefundInterface for KafkaStore {
             .await
     }
 
+    #[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "refunds_v2")))]
     async fn find_refund_by_payment_id_merchant_id(
         &self,
         payment_id: &id_type::PaymentId,
@@ -2673,6 +2675,7 @@ impl RefundInterface for KafkaStore {
             .await
     }
 
+    #[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "refunds_v2")))]
     async fn find_refund_by_merchant_id_refund_id(
         &self,
         merchant_id: &id_type::MerchantId,
@@ -2684,6 +2687,7 @@ impl RefundInterface for KafkaStore {
             .await
     }
 
+    #[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "refunds_v2")))]
     async fn find_refund_by_merchant_id_connector_refund_id_connector(
         &self,
         merchant_id: &id_type::MerchantId,
@@ -2722,6 +2726,7 @@ impl RefundInterface for KafkaStore {
         Ok(refund)
     }
 
+    #[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "refunds_v2")))]
     async fn find_refund_by_merchant_id_connector_transaction_id(
         &self,
         merchant_id: &id_type::MerchantId,
@@ -2737,6 +2742,17 @@ impl RefundInterface for KafkaStore {
             .await
     }
 
+    #[cfg(all(feature = "v2", feature = "refunds_v2"))]
+    async fn find_refund_by_id(
+        &self,
+        id: &id_type::GlobalRefundId,
+        storage_scheme: MerchantStorageScheme,
+    ) -> CustomResult<storage::Refund, errors::StorageError> {
+        self.diesel_store
+            .find_refund_by_id(id, storage_scheme)
+            .await
+    }
+
     async fn insert_refund(
         &self,
         new: storage::RefundNew,
@@ -2754,7 +2770,11 @@ impl RefundInterface for KafkaStore {
         Ok(refund)
     }
 
-    #[cfg(feature = "olap")]
+    #[cfg(all(
+        any(feature = "v1", feature = "v2"),
+        not(feature = "refunds_v2"),
+        feature = "olap"
+    ))]
     async fn filter_refund_by_constraints(
         &self,
         merchant_id: &id_type::MerchantId,
@@ -2774,7 +2794,11 @@ impl RefundInterface for KafkaStore {
             .await
     }
 
-    #[cfg(feature = "olap")]
+    #[cfg(all(
+        any(feature = "v1", feature = "v2"),
+        not(feature = "refunds_v2"),
+        feature = "olap"
+    ))]
     async fn filter_refund_by_meta_constraints(
         &self,
         merchant_id: &id_type::MerchantId,
@@ -2786,7 +2810,11 @@ impl RefundInterface for KafkaStore {
             .await
     }
 
-    #[cfg(feature = "olap")]
+    #[cfg(all(
+        any(feature = "v1", feature = "v2"),
+        not(feature = "refunds_v2"),
+        feature = "olap"
+    ))]
     async fn get_refund_status_with_count(
         &self,
         merchant_id: &id_type::MerchantId,
@@ -2799,7 +2827,11 @@ impl RefundInterface for KafkaStore {
             .await
     }
 
-    #[cfg(feature = "olap")]
+    #[cfg(all(
+        any(feature = "v1", feature = "v2"),
+        not(feature = "refunds_v2"),
+        feature = "olap"
+    ))]
     async fn get_total_count_of_refunds(
         &self,
         merchant_id: &id_type::MerchantId,
diff --git a/crates/router/src/db/refund.rs b/crates/router/src/db/refund.rs
index 1f54eeb5fdc..211f8ddbc48 100644
--- a/crates/router/src/db/refund.rs
+++ b/crates/router/src/db/refund.rs
@@ -17,6 +17,7 @@ const MAX_LIMIT: usize = 100;
 
 #[async_trait::async_trait]
 pub trait RefundInterface {
+    #[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "refunds_v2")))]
     async fn find_refund_by_internal_reference_id_merchant_id(
         &self,
         internal_reference_id: &str,
@@ -24,6 +25,7 @@ pub trait RefundInterface {
         storage_scheme: enums::MerchantStorageScheme,
     ) -> CustomResult<storage_types::Refund, errors::StorageError>;
 
+    #[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "refunds_v2")))]
     async fn find_refund_by_payment_id_merchant_id(
         &self,
         payment_id: &common_utils::id_type::PaymentId,
@@ -31,6 +33,7 @@ pub trait RefundInterface {
         storage_scheme: enums::MerchantStorageScheme,
     ) -> CustomResult<Vec<storage_types::Refund>, errors::StorageError>;
 
+    #[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "refunds_v2")))]
     async fn find_refund_by_merchant_id_refund_id(
         &self,
         merchant_id: &common_utils::id_type::MerchantId,
@@ -38,6 +41,7 @@ pub trait RefundInterface {
         storage_scheme: enums::MerchantStorageScheme,
     ) -> CustomResult<storage_types::Refund, errors::StorageError>;
 
+    #[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "refunds_v2")))]
     async fn find_refund_by_merchant_id_connector_refund_id_connector(
         &self,
         merchant_id: &common_utils::id_type::MerchantId,
@@ -53,6 +57,7 @@ pub trait RefundInterface {
         storage_scheme: enums::MerchantStorageScheme,
     ) -> CustomResult<storage_types::Refund, errors::StorageError>;
 
+    #[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "refunds_v2")))]
     async fn find_refund_by_merchant_id_connector_transaction_id(
         &self,
         merchant_id: &common_utils::id_type::MerchantId,
@@ -60,13 +65,24 @@ pub trait RefundInterface {
         storage_scheme: enums::MerchantStorageScheme,
     ) -> CustomResult<Vec<storage_types::Refund>, errors::StorageError>;
 
+    #[cfg(all(feature = "v2", feature = "refunds_v2"))]
+    async fn find_refund_by_id(
+        &self,
+        id: &common_utils::id_type::GlobalRefundId,
+        storage_scheme: enums::MerchantStorageScheme,
+    ) -> CustomResult<storage_types::Refund, errors::StorageError>;
+
     async fn insert_refund(
         &self,
         new: storage_types::RefundNew,
         storage_scheme: enums::MerchantStorageScheme,
     ) -> CustomResult<storage_types::Refund, errors::StorageError>;
 
-    #[cfg(feature = "olap")]
+    #[cfg(all(
+        any(feature = "v1", feature = "v2"),
+        not(feature = "refunds_v2"),
+        feature = "olap"
+    ))]
     async fn filter_refund_by_constraints(
         &self,
         merchant_id: &common_utils::id_type::MerchantId,
@@ -76,7 +92,11 @@ pub trait RefundInterface {
         offset: i64,
     ) -> CustomResult<Vec<diesel_models::refund::Refund>, errors::StorageError>;
 
-    #[cfg(feature = "olap")]
+    #[cfg(all(
+        any(feature = "v1", feature = "v2"),
+        not(feature = "refunds_v2"),
+        feature = "olap"
+    ))]
     async fn filter_refund_by_meta_constraints(
         &self,
         merchant_id: &common_utils::id_type::MerchantId,
@@ -84,7 +104,11 @@ pub trait RefundInterface {
         storage_scheme: enums::MerchantStorageScheme,
     ) -> CustomResult<api_models::refunds::RefundListMetaData, errors::StorageError>;
 
-    #[cfg(feature = "olap")]
+    #[cfg(all(
+        any(feature = "v1", feature = "v2"),
+        not(feature = "refunds_v2"),
+        feature = "olap"
+    ))]
     async fn get_refund_status_with_count(
         &self,
         merchant_id: &common_utils::id_type::MerchantId,
@@ -93,7 +117,11 @@ pub trait RefundInterface {
         storage_scheme: enums::MerchantStorageScheme,
     ) -> CustomResult<Vec<(common_enums::RefundStatus, i64)>, errors::StorageError>;
 
-    #[cfg(feature = "olap")]
+    #[cfg(all(
+        any(feature = "v1", feature = "v2"),
+        not(feature = "refunds_v2"),
+        feature = "olap"
+    ))]
     async fn get_total_count_of_refunds(
         &self,
         merchant_id: &common_utils::id_type::MerchantId,
@@ -320,6 +348,7 @@ mod storage {
     };
     #[async_trait::async_trait]
     impl RefundInterface for Store {
+        #[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "refunds_v2")))]
         #[instrument(skip_all)]
         async fn find_refund_by_internal_reference_id_merchant_id(
             &self,
@@ -376,6 +405,7 @@ mod storage {
             }
         }
 
+        #[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "refunds_v2")))]
         #[instrument(skip_all)]
         async fn insert_refund(
             &self,
@@ -529,6 +559,20 @@ mod storage {
             }
         }
 
+        #[cfg(all(feature = "v2", feature = "refunds_v2"))]
+        #[instrument(skip_all)]
+        async fn insert_refund(
+            &self,
+            new: storage_types::RefundNew,
+            _storage_scheme: enums::MerchantStorageScheme,
+        ) -> CustomResult<storage_types::Refund, errors::StorageError> {
+            let conn = connection::pg_connection_write(self).await?;
+            new.insert(&conn)
+                .await
+                .map_err(|error| report!(errors::StorageError::from(error)))
+        }
+
+        #[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "refunds_v2")))]
         #[instrument(skip_all)]
         async fn find_refund_by_merchant_id_connector_transaction_id(
             &self,
@@ -588,6 +632,7 @@ mod storage {
             }
         }
 
+        #[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "refunds_v2")))]
         #[instrument(skip_all)]
         async fn update_refund(
             &self,
@@ -652,6 +697,21 @@ mod storage {
             }
         }
 
+        #[cfg(all(feature = "v2", feature = "refunds_v2"))]
+        #[instrument(skip_all)]
+        async fn update_refund(
+            &self,
+            this: storage_types::Refund,
+            refund: storage_types::RefundUpdate,
+            _storage_scheme: enums::MerchantStorageScheme,
+        ) -> CustomResult<storage_types::Refund, errors::StorageError> {
+            let conn = connection::pg_connection_write(self).await?;
+            this.update_with_id(&conn, refund)
+                .await
+                .map_err(|error| report!(errors::StorageError::from(error)))
+        }
+
+        #[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "refunds_v2")))]
         #[instrument(skip_all)]
         async fn find_refund_by_merchant_id_refund_id(
             &self,
@@ -702,6 +762,7 @@ mod storage {
             }
         }
 
+        #[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "refunds_v2")))]
         #[instrument(skip_all)]
         async fn find_refund_by_merchant_id_connector_refund_id_connector(
             &self,
@@ -760,6 +821,7 @@ mod storage {
             }
         }
 
+        #[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "refunds_v2")))]
         #[instrument(skip_all)]
         async fn find_refund_by_payment_id_merchant_id(
             &self,
@@ -807,7 +869,24 @@ mod storage {
             }
         }
 
-        #[cfg(feature = "olap")]
+        #[cfg(all(feature = "v2", feature = "refunds_v2"))]
+        #[instrument(skip_all)]
+        async fn find_refund_by_id(
+            &self,
+            id: &common_utils::id_type::GlobalRefundId,
+            _storage_scheme: enums::MerchantStorageScheme,
+        ) -> CustomResult<storage_types::Refund, errors::StorageError> {
+            let conn = connection::pg_connection_read(self).await?;
+            storage_types::Refund::find_by_global_id(&conn, id)
+                .await
+                .map_err(|error| report!(errors::StorageError::from(error)))
+        }
+
+        #[cfg(all(
+            any(feature = "v1", feature = "v2"),
+            not(feature = "refunds_v2"),
+            feature = "olap"
+        ))]
         #[instrument(skip_all)]
         async fn filter_refund_by_constraints(
             &self,
@@ -829,7 +908,11 @@ mod storage {
             .map_err(|error| report!(errors::StorageError::from(error)))
         }
 
-        #[cfg(feature = "olap")]
+        #[cfg(all(
+            any(feature = "v1", feature = "v2"),
+            not(feature = "refunds_v2"),
+            feature = "olap"
+        ))]
         #[instrument(skip_all)]
         async fn filter_refund_by_meta_constraints(
             &self,
@@ -843,7 +926,11 @@ mod storage {
                         .map_err(|error|report!(errors::StorageError::from(error)))
         }
 
-        #[cfg(feature = "olap")]
+        #[cfg(all(
+            any(feature = "v1", feature = "v2"),
+            not(feature = "refunds_v2"),
+            feature = "olap"
+        ))]
         #[instrument(skip_all)]
         async fn get_refund_status_with_count(
             &self,
@@ -858,7 +945,11 @@ mod storage {
             .map_err(|error| report!(errors::StorageError::from(error)))
         }
 
-        #[cfg(feature = "olap")]
+        #[cfg(all(
+            any(feature = "v1", feature = "v2"),
+            not(feature = "refunds_v2"),
+            feature = "olap"
+        ))]
         #[instrument(skip_all)]
         async fn get_total_count_of_refunds(
             &self,
@@ -880,6 +971,7 @@ mod storage {
 
 #[async_trait::async_trait]
 impl RefundInterface for MockDb {
+    #[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "refunds_v2")))]
     async fn find_refund_by_internal_reference_id_merchant_id(
         &self,
         internal_reference_id: &str,
@@ -899,6 +991,7 @@ impl RefundInterface for MockDb {
             })
     }
 
+    #[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "refunds_v2")))]
     async fn insert_refund(
         &self,
         new: storage_types::RefundNew,
@@ -950,6 +1043,56 @@ impl RefundInterface for MockDb {
         refunds.push(refund.clone());
         Ok(refund)
     }
+
+    #[cfg(all(feature = "v2", feature = "refunds_v2"))]
+    async fn insert_refund(
+        &self,
+        new: storage_types::RefundNew,
+        _storage_scheme: enums::MerchantStorageScheme,
+    ) -> CustomResult<storage_types::Refund, errors::StorageError> {
+        let mut refunds = self.refunds.lock().await;
+        let current_time = common_utils::date_time::now();
+
+        let refund = storage_types::Refund {
+            id: new.id,
+            merchant_reference_id: new.merchant_reference_id,
+            payment_id: new.payment_id,
+            merchant_id: new.merchant_id,
+            attempt_id: new.attempt_id,
+            connector_transaction_id: new.connector_transaction_id,
+            connector: new.connector,
+            connector_refund_id: new.connector_refund_id,
+            external_reference_id: new.external_reference_id,
+            refund_type: new.refund_type,
+            total_amount: new.total_amount,
+            currency: new.currency,
+            refund_amount: new.refund_amount,
+            refund_status: new.refund_status,
+            sent_to_gateway: new.sent_to_gateway,
+            refund_error_message: None,
+            refund_error_code: None,
+            metadata: new.metadata,
+            refund_arn: new.refund_arn.clone(),
+            created_at: new.created_at,
+            modified_at: current_time,
+            description: new.description,
+            refund_reason: new.refund_reason.clone(),
+            profile_id: new.profile_id,
+            updated_by: new.updated_by,
+            connector_id: new.connector_id,
+            charges: new.charges,
+            split_refunds: new.split_refunds,
+            organization_id: new.organization_id,
+            unified_code: None,
+            unified_message: None,
+            processor_refund_data: new.processor_refund_data.clone(),
+            processor_transaction_data: new.processor_transaction_data.clone(),
+        };
+        refunds.push(refund.clone());
+        Ok(refund)
+    }
+
+    #[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "refunds_v2")))]
     async fn find_refund_by_merchant_id_connector_transaction_id(
         &self,
         merchant_id: &common_utils::id_type::MerchantId,
@@ -968,6 +1111,7 @@ impl RefundInterface for MockDb {
             .collect::<Vec<_>>())
     }
 
+    #[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "refunds_v2")))]
     async fn update_refund(
         &self,
         this: storage_types::Refund,
@@ -990,6 +1134,30 @@ impl RefundInterface for MockDb {
             })
     }
 
+    #[cfg(all(feature = "v2", feature = "refunds_v2"))]
+    async fn update_refund(
+        &self,
+        this: storage_types::Refund,
+        refund: storage_types::RefundUpdate,
+        _storage_scheme: enums::MerchantStorageScheme,
+    ) -> CustomResult<storage_types::Refund, errors::StorageError> {
+        self.refunds
+            .lock()
+            .await
+            .iter_mut()
+            .find(|refund| this.merchant_reference_id == refund.merchant_reference_id)
+            .map(|r| {
+                let refund_updated = RefundUpdateInternal::from(refund).create_refund(r.clone());
+                *r = refund_updated.clone();
+                refund_updated
+            })
+            .ok_or_else(|| {
+                errors::StorageError::ValueNotFound("cannot find refund to update".to_string())
+                    .into()
+            })
+    }
+
+    #[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "refunds_v2")))]
     async fn find_refund_by_merchant_id_refund_id(
         &self,
         merchant_id: &common_utils::id_type::MerchantId,
@@ -1007,6 +1175,7 @@ impl RefundInterface for MockDb {
             })
     }
 
+    #[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "refunds_v2")))]
     async fn find_refund_by_merchant_id_connector_refund_id_connector(
         &self,
         merchant_id: &common_utils::id_type::MerchantId,
@@ -1032,6 +1201,7 @@ impl RefundInterface for MockDb {
             })
     }
 
+    #[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "refunds_v2")))]
     async fn find_refund_by_payment_id_merchant_id(
         &self,
         payment_id: &common_utils::id_type::PaymentId,
@@ -1047,7 +1217,28 @@ impl RefundInterface for MockDb {
             .collect::<Vec<_>>())
     }
 
-    #[cfg(feature = "olap")]
+    #[cfg(all(feature = "v2", feature = "refunds_v2"))]
+    async fn find_refund_by_id(
+        &self,
+        id: &common_utils::id_type::GlobalRefundId,
+        _storage_scheme: enums::MerchantStorageScheme,
+    ) -> CustomResult<storage_types::Refund, errors::StorageError> {
+        let refunds = self.refunds.lock().await;
+
+        refunds
+            .iter()
+            .find(|refund| refund.id == *id)
+            .cloned()
+            .ok_or_else(|| {
+                errors::StorageError::DatabaseError(DatabaseError::NotFound.into()).into()
+            })
+    }
+
+    #[cfg(all(
+        any(feature = "v1", feature = "v2"),
+        not(feature = "refunds_v2"),
+        feature = "olap"
+    ))]
     async fn filter_refund_by_constraints(
         &self,
         merchant_id: &common_utils::id_type::MerchantId,
@@ -1162,7 +1353,11 @@ impl RefundInterface for MockDb {
         Ok(filtered_refunds)
     }
 
-    #[cfg(feature = "olap")]
+    #[cfg(all(
+        any(feature = "v1", feature = "v2"),
+        not(feature = "refunds_v2"),
+        feature = "olap"
+    ))]
     async fn filter_refund_by_meta_constraints(
         &self,
         _merchant_id: &common_utils::id_type::MerchantId,
@@ -1209,7 +1404,11 @@ impl RefundInterface for MockDb {
         Ok(refund_meta_data)
     }
 
-    #[cfg(feature = "olap")]
+    #[cfg(all(
+        any(feature = "v1", feature = "v2"),
+        not(feature = "refunds_v2"),
+        feature = "olap"
+    ))]
     async fn get_refund_status_with_count(
         &self,
         _merchant_id: &common_utils::id_type::MerchantId,
@@ -1255,7 +1454,11 @@ impl RefundInterface for MockDb {
         Ok(result)
     }
 
-    #[cfg(feature = "olap")]
+    #[cfg(all(
+        any(feature = "v1", feature = "v2"),
+        not(feature = "refunds_v2"),
+        feature = "olap"
+    ))]
     async fn get_total_count_of_refunds(
         &self,
         merchant_id: &common_utils::id_type::MerchantId,
diff --git a/crates/router/src/services/kafka/refund.rs b/crates/router/src/services/kafka/refund.rs
index a7e37930cf2..19635b2d8c2 100644
--- a/crates/router/src/services/kafka/refund.rs
+++ b/crates/router/src/services/kafka/refund.rs
@@ -5,6 +5,9 @@ use common_utils::{
 use diesel_models::{enums as storage_enums, refund::Refund};
 use time::OffsetDateTime;
 
+use crate::events;
+
+#[cfg(feature = "v1")]
 #[derive(serde::Serialize, Debug)]
 pub struct KafkaRefund<'a> {
     pub internal_reference_id: &'a String,
@@ -35,6 +38,7 @@ pub struct KafkaRefund<'a> {
     pub organization_id: &'a id_type::OrganizationId,
 }
 
+#[cfg(feature = "v1")]
 impl<'a> KafkaRefund<'a> {
     pub fn from_storage(refund: &'a Refund) -> Self {
         Self {
@@ -66,6 +70,70 @@ impl<'a> KafkaRefund<'a> {
     }
 }
 
+#[cfg(feature = "v2")]
+#[derive(serde::Serialize, Debug)]
+pub struct KafkaRefund<'a> {
+    pub id: &'a id_type::GlobalRefundId,
+    pub merchant_reference_id: &'a id_type::RefundReferenceId,
+    pub payment_id: &'a id_type::GlobalPaymentId,
+    pub merchant_id: &'a id_type::MerchantId,
+    pub connector_transaction_id: &'a String,
+    pub connector: &'a String,
+    pub connector_refund_id: Option<&'a String>,
+    pub external_reference_id: Option<&'a String>,
+    pub refund_type: &'a storage_enums::RefundType,
+    pub total_amount: &'a MinorUnit,
+    pub currency: &'a storage_enums::Currency,
+    pub refund_amount: &'a MinorUnit,
+    pub refund_status: &'a storage_enums::RefundStatus,
+    pub sent_to_gateway: &'a bool,
+    pub refund_error_message: Option<&'a String>,
+    pub refund_arn: Option<&'a String>,
+    #[serde(default, with = "time::serde::timestamp")]
+    pub created_at: OffsetDateTime,
+    #[serde(default, with = "time::serde::timestamp")]
+    pub modified_at: OffsetDateTime,
+    pub description: Option<&'a String>,
+    pub attempt_id: &'a id_type::GlobalAttemptId,
+    pub refund_reason: Option<&'a String>,
+    pub refund_error_code: Option<&'a String>,
+    pub profile_id: Option<&'a id_type::ProfileId>,
+    pub organization_id: &'a id_type::OrganizationId,
+}
+
+#[cfg(feature = "v2")]
+impl<'a> KafkaRefund<'a> {
+    pub fn from_storage(refund: &'a Refund) -> Self {
+        Self {
+            id: &refund.id,
+            merchant_reference_id: &refund.merchant_reference_id,
+            payment_id: &refund.payment_id,
+            merchant_id: &refund.merchant_id,
+            connector_transaction_id: refund.get_connector_transaction_id(),
+            connector: &refund.connector,
+            connector_refund_id: refund.get_optional_connector_refund_id(),
+            external_reference_id: refund.external_reference_id.as_ref(),
+            refund_type: &refund.refund_type,
+            total_amount: &refund.total_amount,
+            currency: &refund.currency,
+            refund_amount: &refund.refund_amount,
+            refund_status: &refund.refund_status,
+            sent_to_gateway: &refund.sent_to_gateway,
+            refund_error_message: refund.refund_error_message.as_ref(),
+            refund_arn: refund.refund_arn.as_ref(),
+            created_at: refund.created_at.assume_utc(),
+            modified_at: refund.modified_at.assume_utc(),
+            description: refund.description.as_ref(),
+            attempt_id: &refund.attempt_id,
+            refund_reason: refund.refund_reason.as_ref(),
+            refund_error_code: refund.refund_error_code.as_ref(),
+            profile_id: refund.profile_id.as_ref(),
+            organization_id: &refund.organization_id,
+        }
+    }
+}
+
+#[cfg(feature = "v1")]
 impl super::KafkaMessage for KafkaRefund<'_> {
     fn key(&self) -> String {
         format!(
@@ -77,7 +145,24 @@ impl super::KafkaMessage for KafkaRefund<'_> {
         )
     }
 
-    fn event_type(&self) -> crate::events::EventType {
-        crate::events::EventType::Refund
+    fn event_type(&self) -> events::EventType {
+        events::EventType::Refund
+    }
+}
+
+#[cfg(feature = "v2")]
+impl super::KafkaMessage for KafkaRefund<'_> {
+    fn key(&self) -> String {
+        format!(
+            "{}_{}_{}_{}",
+            self.merchant_id.get_string_repr(),
+            self.payment_id.get_string_repr(),
+            self.attempt_id.get_string_repr(),
+            self.merchant_reference_id.get_string_repr()
+        )
+    }
+
+    fn event_type(&self) -> events::EventType {
+        events::EventType::Refund
     }
 }
diff --git a/crates/router/src/services/kafka/refund_event.rs b/crates/router/src/services/kafka/refund_event.rs
index b9b3db17b58..34ed0b85c55 100644
--- a/crates/router/src/services/kafka/refund_event.rs
+++ b/crates/router/src/services/kafka/refund_event.rs
@@ -5,6 +5,9 @@ use common_utils::{
 use diesel_models::{enums as storage_enums, refund::Refund};
 use time::OffsetDateTime;
 
+use crate::events;
+
+#[cfg(feature = "v1")]
 #[serde_with::skip_serializing_none]
 #[derive(serde::Serialize, Debug)]
 pub struct KafkaRefundEvent<'a> {
@@ -36,6 +39,7 @@ pub struct KafkaRefundEvent<'a> {
     pub organization_id: &'a id_type::OrganizationId,
 }
 
+#[cfg(feature = "v1")]
 impl<'a> KafkaRefundEvent<'a> {
     pub fn from_storage(refund: &'a Refund) -> Self {
         Self {
@@ -67,6 +71,71 @@ impl<'a> KafkaRefundEvent<'a> {
     }
 }
 
+#[cfg(feature = "v2")]
+#[serde_with::skip_serializing_none]
+#[derive(serde::Serialize, Debug)]
+pub struct KafkaRefundEvent<'a> {
+    pub id: &'a id_type::GlobalRefundId,
+    pub merchant_reference_id: &'a id_type::RefundReferenceId,
+    pub payment_id: &'a id_type::GlobalPaymentId,
+    pub merchant_id: &'a id_type::MerchantId,
+    pub connector_transaction_id: &'a String,
+    pub connector: &'a String,
+    pub connector_refund_id: Option<&'a String>,
+    pub external_reference_id: Option<&'a String>,
+    pub refund_type: &'a storage_enums::RefundType,
+    pub total_amount: &'a MinorUnit,
+    pub currency: &'a storage_enums::Currency,
+    pub refund_amount: &'a MinorUnit,
+    pub refund_status: &'a storage_enums::RefundStatus,
+    pub sent_to_gateway: &'a bool,
+    pub refund_error_message: Option<&'a String>,
+    pub refund_arn: Option<&'a String>,
+    #[serde(default, with = "time::serde::timestamp::nanoseconds")]
+    pub created_at: OffsetDateTime,
+    #[serde(default, with = "time::serde::timestamp::nanoseconds")]
+    pub modified_at: OffsetDateTime,
+    pub description: Option<&'a String>,
+    pub attempt_id: &'a id_type::GlobalAttemptId,
+    pub refund_reason: Option<&'a String>,
+    pub refund_error_code: Option<&'a String>,
+    pub profile_id: Option<&'a id_type::ProfileId>,
+    pub organization_id: &'a id_type::OrganizationId,
+}
+
+#[cfg(feature = "v2")]
+impl<'a> KafkaRefundEvent<'a> {
+    pub fn from_storage(refund: &'a Refund) -> Self {
+        Self {
+            id: &refund.id,
+            merchant_reference_id: &refund.merchant_reference_id,
+            payment_id: &refund.payment_id,
+            merchant_id: &refund.merchant_id,
+            connector_transaction_id: refund.get_connector_transaction_id(),
+            connector: &refund.connector,
+            connector_refund_id: refund.get_optional_connector_refund_id(),
+            external_reference_id: refund.external_reference_id.as_ref(),
+            refund_type: &refund.refund_type,
+            total_amount: &refund.total_amount,
+            currency: &refund.currency,
+            refund_amount: &refund.refund_amount,
+            refund_status: &refund.refund_status,
+            sent_to_gateway: &refund.sent_to_gateway,
+            refund_error_message: refund.refund_error_message.as_ref(),
+            refund_arn: refund.refund_arn.as_ref(),
+            created_at: refund.created_at.assume_utc(),
+            modified_at: refund.modified_at.assume_utc(),
+            description: refund.description.as_ref(),
+            attempt_id: &refund.attempt_id,
+            refund_reason: refund.refund_reason.as_ref(),
+            refund_error_code: refund.refund_error_code.as_ref(),
+            profile_id: refund.profile_id.as_ref(),
+            organization_id: &refund.organization_id,
+        }
+    }
+}
+
+#[cfg(feature = "v1")]
 impl super::KafkaMessage for KafkaRefundEvent<'_> {
     fn key(&self) -> String {
         format!(
@@ -77,8 +146,23 @@ impl super::KafkaMessage for KafkaRefundEvent<'_> {
             self.refund_id
         )
     }
+    fn event_type(&self) -> events::EventType {
+        events::EventType::Refund
+    }
+}
 
-    fn event_type(&self) -> crate::events::EventType {
-        crate::events::EventType::Refund
+#[cfg(feature = "v2")]
+impl super::KafkaMessage for KafkaRefundEvent<'_> {
+    fn key(&self) -> String {
+        format!(
+            "{}_{}_{}_{}",
+            self.merchant_id.get_string_repr(),
+            self.payment_id.get_string_repr(),
+            self.attempt_id.get_string_repr(),
+            self.merchant_reference_id.get_string_repr()
+        )
+    }
+    fn event_type(&self) -> events::EventType {
+        events::EventType::Refund
     }
 }
diff --git a/crates/router/src/types/storage/refund.rs b/crates/router/src/types/storage/refund.rs
index e46fc1a3f47..646b2742d5b 100644
--- a/crates/router/src/types/storage/refund.rs
+++ b/crates/router/src/types/storage/refund.rs
@@ -18,6 +18,7 @@ use crate::{connection::PgPooledConn, logger};
 
 #[async_trait::async_trait]
 pub trait RefundDbExt: Sized {
+    #[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "refunds_v2")))]
     async fn filter_by_constraints(
         conn: &PgPooledConn,
         merchant_id: &common_utils::id_type::MerchantId,
@@ -26,18 +27,21 @@ pub trait RefundDbExt: Sized {
         offset: i64,
     ) -> CustomResult<Vec<Self>, errors::DatabaseError>;
 
+    #[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "refunds_v2")))]
     async fn filter_by_meta_constraints(
         conn: &PgPooledConn,
         merchant_id: &common_utils::id_type::MerchantId,
         refund_list_details: &common_utils::types::TimeRange,
     ) -> CustomResult<api_models::refunds::RefundListMetaData, errors::DatabaseError>;
 
+    #[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "refunds_v2")))]
     async fn get_refunds_count(
         conn: &PgPooledConn,
         merchant_id: &common_utils::id_type::MerchantId,
         refund_list_details: &refunds::RefundListConstraints,
     ) -> CustomResult<i64, errors::DatabaseError>;
 
+    #[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "refunds_v2")))]
     async fn get_refund_status_with_count(
         conn: &PgPooledConn,
         merchant_id: &common_utils::id_type::MerchantId,
@@ -48,6 +52,7 @@ pub trait RefundDbExt: Sized {
 
 #[async_trait::async_trait]
 impl RefundDbExt for Refund {
+    #[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "refunds_v2")))]
     async fn filter_by_constraints(
         conn: &PgPooledConn,
         merchant_id: &common_utils::id_type::MerchantId,
@@ -159,6 +164,7 @@ impl RefundDbExt for Refund {
         .attach_printable_lazy(|| "Error filtering records by predicate")
     }
 
+    #[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "refunds_v2")))]
     async fn filter_by_meta_constraints(
         conn: &PgPooledConn,
         merchant_id: &common_utils::id_type::MerchantId,
@@ -214,6 +220,7 @@ impl RefundDbExt for Refund {
         Ok(meta)
     }
 
+    #[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "refunds_v2")))]
     async fn get_refunds_count(
         conn: &PgPooledConn,
         merchant_id: &common_utils::id_type::MerchantId,
@@ -302,6 +309,7 @@ impl RefundDbExt for Refund {
             .attach_printable_lazy(|| "Error filtering count of refunds")
     }
 
+    #[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "refunds_v2")))]
     async fn get_refund_status_with_count(
         conn: &PgPooledConn,
         merchant_id: &common_utils::id_type::MerchantId,
diff --git a/crates/storage_impl/Cargo.toml b/crates/storage_impl/Cargo.toml
index ac3598cea58..06b551d8312 100644
--- a/crates/storage_impl/Cargo.toml
+++ b/crates/storage_impl/Cargo.toml
@@ -17,6 +17,7 @@ v1 = ["api_models/v1", "diesel_models/v1", "hyperswitch_domain_models/v1", "comm
 v2 = ["api_models/v2", "diesel_models/v2", "hyperswitch_domain_models/v2", "common_utils/v2"]
 customer_v2 = ["api_models/customer_v2", "diesel_models/customer_v2", "hyperswitch_domain_models/customer_v2"]
 payment_methods_v2 = ["diesel_models/payment_methods_v2", "api_models/payment_methods_v2", "hyperswitch_domain_models/payment_methods_v2"]
+refunds_v2 = ["diesel_models/refunds_v2"]
 
 [dependencies]
 # First Party dependencies
diff --git a/crates/storage_impl/src/lib.rs b/crates/storage_impl/src/lib.rs
index 13b4671e863..657114d1067 100644
--- a/crates/storage_impl/src/lib.rs
+++ b/crates/storage_impl/src/lib.rs
@@ -346,6 +346,7 @@ impl UniqueConstraints for diesel_models::PaymentAttempt {
     }
 }
 
+#[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "refunds_v2")))]
 impl UniqueConstraints for diesel_models::Refund {
     fn unique_constraints(&self) -> Vec<String> {
         vec![format!(
@@ -359,6 +360,16 @@ impl UniqueConstraints for diesel_models::Refund {
     }
 }
 
+#[cfg(all(feature = "v2", feature = "refunds_v2"))]
+impl UniqueConstraints for diesel_models::Refund {
+    fn unique_constraints(&self) -> Vec<String> {
+        vec![self.id.get_string_repr().to_owned()]
+    }
+    fn table_name(&self) -> &str {
+        "Refund"
+    }
+}
+
 impl UniqueConstraints for diesel_models::ReverseLookup {
     fn unique_constraints(&self) -> Vec<String> {
         vec![format!("reverselookup_{}", self.lookup_id)]
diff --git a/v2_compatible_migrations/2024-08-28-081721_add_v2_columns/down.sql b/v2_compatible_migrations/2024-08-28-081721_add_v2_columns/down.sql
index 92f74b1d9ac..e3c87773716 100644
--- a/v2_compatible_migrations/2024-08-28-081721_add_v2_columns/down.sql
+++ b/v2_compatible_migrations/2024-08-28-081721_add_v2_columns/down.sql
@@ -52,4 +52,9 @@ ALTER TABLE merchant_connector_account
 ALTER TABLE payment_methods
     DROP COLUMN IF EXISTS locker_fingerprint_id,
     DROP COLUMN IF EXISTS payment_method_type_v2,
-    DROP COLUMN IF EXISTS payment_method_subtype;
\ No newline at end of file
+    DROP COLUMN IF EXISTS payment_method_subtype;
+
+ALTER TABLE refund
+    DROP COLUMN IF EXISTS id,
+    DROP COLUMN IF EXISTS merchant_reference_id,
+    DROP COLUMN IF EXISTS connector_id;
\ No newline at end of file
diff --git a/v2_compatible_migrations/2024-08-28-081721_add_v2_columns/up.sql b/v2_compatible_migrations/2024-08-28-081721_add_v2_columns/up.sql
index 3094cdb3320..1dcedbff563 100644
--- a/v2_compatible_migrations/2024-08-28-081721_add_v2_columns/up.sql
+++ b/v2_compatible_migrations/2024-08-28-081721_add_v2_columns/up.sql
@@ -62,4 +62,9 @@ ALTER TABLE merchant_connector_account
 ALTER TABLE payment_methods
     ADD COLUMN IF NOT EXISTS locker_fingerprint_id VARCHAR(64),
     ADD COLUMN IF NOT EXISTS payment_method_type_v2 VARCHAR(64),
-    ADD COLUMN IF NOT EXISTS payment_method_subtype VARCHAR(64);
\ No newline at end of file
+    ADD COLUMN IF NOT EXISTS payment_method_subtype VARCHAR(64);
+
+ALTER TABLE refund
+    ADD COLUMN IF NOT EXISTS id VARCHAR(64),
+    ADD COLUMN IF NOT EXISTS merchant_reference_id VARCHAR(64),
+    ADD COLUMN IF NOT EXISTS connector_id VARCHAR(64);
\ No newline at end of file
diff --git a/v2_compatible_migrations/2024-08-28-081757_drop_not_null_constraints_on_v1_columns/down.sql b/v2_compatible_migrations/2024-08-28-081757_drop_not_null_constraints_on_v1_columns/down.sql
index 10fd9adfa55..c0b0efbd385 100644
--- a/v2_compatible_migrations/2024-08-28-081757_drop_not_null_constraints_on_v1_columns/down.sql
+++ b/v2_compatible_migrations/2024-08-28-081757_drop_not_null_constraints_on_v1_columns/down.sql
@@ -37,3 +37,7 @@ ALTER TABLE payment_attempt
 ---------------------- Payment Methods ----------------------
 ALTER TABLE payment_methods 
     ADD PRIMARY KEY (payment_method_id);
+
+---------------------- Refunds ----------------------
+ALTER TABLE refund
+    ADD PRIMARY KEY (refund_id,merchant_id);
\ No newline at end of file
diff --git a/v2_compatible_migrations/2024-08-28-081757_drop_not_null_constraints_on_v1_columns/up.sql b/v2_compatible_migrations/2024-08-28-081757_drop_not_null_constraints_on_v1_columns/up.sql
index e348f2e1cbb..731d488297e 100644
--- a/v2_compatible_migrations/2024-08-28-081757_drop_not_null_constraints_on_v1_columns/up.sql
+++ b/v2_compatible_migrations/2024-08-28-081757_drop_not_null_constraints_on_v1_columns/up.sql
@@ -37,3 +37,7 @@ ALTER TABLE payment_attempt
 ALTER TABLE payment_methods 
     DROP CONSTRAINT payment_methods_pkey,
     ALTER COLUMN payment_method_id DROP NOT NULL;
+
+ALTER TABLE refund
+    DROP CONSTRAINT refund_pkey,
+    ALTER COLUMN refund_id DROP NOT NULL;
\ No newline at end of file
diff --git a/v2_migrations/2024-08-28-081839_add_not_null_constraints_to_v2_columns/down.sql b/v2_migrations/2024-08-28-081837_add_not_null_constraints_to_v2_columns/down.sql
similarity index 87%
rename from v2_migrations/2024-08-28-081839_add_not_null_constraints_to_v2_columns/down.sql
rename to v2_migrations/2024-08-28-081837_add_not_null_constraints_to_v2_columns/down.sql
index 1941d9c1b8e..06b78440d17 100644
--- a/v2_migrations/2024-08-28-081839_add_not_null_constraints_to_v2_columns/down.sql
+++ b/v2_migrations/2024-08-28-081837_add_not_null_constraints_to_v2_columns/down.sql
@@ -24,4 +24,7 @@ ALTER TABLE payment_attempt
     ALTER COLUMN net_amount DROP NOT NULL,
     ALTER COLUMN authentication_type DROP NOT NULL,
     ALTER COLUMN payment_method_type_v2 DROP NOT NULL,
-    ALTER COLUMN payment_method_subtype DROP NOT NULL;
\ No newline at end of file
+    ALTER COLUMN payment_method_subtype DROP NOT NULL;
+
+ALTER TABLE refund
+    ALTER COLUMN merchant_reference_id DROP NOT NULL;
\ No newline at end of file
diff --git a/v2_migrations/2024-08-28-081839_add_not_null_constraints_to_v2_columns/up.sql b/v2_migrations/2024-08-28-081837_add_not_null_constraints_to_v2_columns/up.sql
similarity index 83%
rename from v2_migrations/2024-08-28-081839_add_not_null_constraints_to_v2_columns/up.sql
rename to v2_migrations/2024-08-28-081837_add_not_null_constraints_to_v2_columns/up.sql
index 36fc7d2893d..3641d2bdab6 100644
--- a/v2_migrations/2024-08-28-081839_add_not_null_constraints_to_v2_columns/up.sql
+++ b/v2_migrations/2024-08-28-081837_add_not_null_constraints_to_v2_columns/up.sql
@@ -23,4 +23,8 @@ ALTER TABLE payment_attempt
     ALTER COLUMN net_amount SET NOT NULL,
     ALTER COLUMN authentication_type SET NOT NULL,
     ALTER COLUMN payment_method_type_v2 SET NOT NULL,
-    ALTER COLUMN payment_method_subtype SET NOT NULL;
\ No newline at end of file
+    ALTER COLUMN payment_method_subtype SET NOT NULL;
+
+-- This migration is to make fields mandatory in refund table
+ALTER TABLE refund
+    ALTER COLUMN merchant_reference_id SET NOT NULL;
\ No newline at end of file
diff --git a/v2_migrations/2024-08-28-081838_update_v2_primary_key_constraints/down.sql b/v2_migrations/2024-08-28-081838_update_v2_primary_key_constraints/down.sql
index 9b04dd6d3f5..a48c2b660f5 100644
--- a/v2_migrations/2024-08-28-081838_update_v2_primary_key_constraints/down.sql
+++ b/v2_migrations/2024-08-28-081838_update_v2_primary_key_constraints/down.sql
@@ -65,3 +65,7 @@ ALTER TABLE payment_attempt ALTER COLUMN id DROP NOT NULL;
 ------------------------ Payment Methods -----------------------
 ALTER TABLE payment_methods DROP CONSTRAINT payment_methods_pkey;
 ALTER TABLE payment_methods ALTER COLUMN id DROP NOT NULL;
+
+------------------------ Refunds -----------------------
+ALTER TABLE refund DROP CONSTRAINT refund_pkey;
+ALTER TABLE refund ALTER COLUMN id DROP NOT NULL;
\ No newline at end of file
diff --git a/v2_migrations/2024-08-28-081838_update_v2_primary_key_constraints/up.sql b/v2_migrations/2024-08-28-081838_update_v2_primary_key_constraints/up.sql
index 3b155115119..df0c434d68f 100644
--- a/v2_migrations/2024-08-28-081838_update_v2_primary_key_constraints/up.sql
+++ b/v2_migrations/2024-08-28-081838_update_v2_primary_key_constraints/up.sql
@@ -76,3 +76,8 @@ ADD PRIMARY KEY (id);
 -- Add primary key constraint
 ALTER TABLE payment_methods 
 ADD PRIMARY KEY (id);
+
+------------------------ Refunds -----------------------
+-- Add primary key constraint
+ALTER TABLE refund
+ADD PRIMARY KEY (id);
\ No newline at end of file
diff --git a/v2_migrations/2025-01-13-081847_drop_v1_columns/down.sql b/v2_migrations/2025-01-13-081847_drop_v1_columns/down.sql
index efcb56101f9..483ef12a276 100644
--- a/v2_migrations/2025-01-13-081847_drop_v1_columns/down.sql
+++ b/v2_migrations/2025-01-13-081847_drop_v1_columns/down.sql
@@ -130,4 +130,9 @@ ALTER TABLE payment_methods
 ALTER TABLE refund ADD COLUMN connector_refund_data VARCHAR(512),
     ADD COLUMN connector_transaction_data VARCHAR(512);
 
-ALTER TABLE captures ADD COLUMN connector_capture_data VARCHAR(512);
\ No newline at end of file
+ALTER TABLE captures ADD COLUMN connector_capture_data VARCHAR(512);
+
+ALTER TABLE refund 
+    ADD COLUMN IF NOT EXISTS internal_reference_id VARCHAR(64),
+    ADD COLUMN IF NOT EXISTS refund_id VARCHAR(64),
+    ADD COLUMN IF NOT EXISTS merchant_connector_id VARCHAR(64);
\ No newline at end of file
diff --git a/v2_migrations/2025-01-13-081847_drop_v1_columns/up.sql b/v2_migrations/2025-01-13-081847_drop_v1_columns/up.sql
index cbb2b264d45..69357bb04ac 100644
--- a/v2_migrations/2025-01-13-081847_drop_v1_columns/up.sql
+++ b/v2_migrations/2025-01-13-081847_drop_v1_columns/up.sql
@@ -121,3 +121,9 @@ ALTER TABLE refund DROP COLUMN connector_refund_data,
 
 -- Run below queries only when V1 is deprecated
 ALTER TABLE captures DROP COLUMN connector_capture_data;
+
+-- Run below queries only when V1 is deprecated
+ALTER TABLE refund 
+    DROP COLUMN IF EXISTS internal_reference_id,
+    DROP COLUMN IF EXISTS refund_id,
+    DROP COLUMN IF EXISTS merchant_connector_id;