Bug: [FEATURE]: Add refunds create core flow for v2 apis.



Implement refunds create core flow for v2 apis.
diff --git a/api-reference-v2/openapi_spec.json b/api-reference-v2/openapi_spec.json
index 3125f3ac29c..10682866ab7 100644
--- a/api-reference-v2/openapi_spec.json
+++ b/api-reference-v2/openapi_spec.json
@@ -20357,82 +20357,6 @@
           }
         }
       },
-      "RefundRequest": {
-        "type": "object",
-        "required": [
-          "payment_id"
-        ],
-        "properties": {
-          "payment_id": {
-            "type": "string",
-            "description": "The payment id against which refund is to be initiated",
-            "example": "pay_mbabizu24mvu3mela5njyhpit4",
-            "maxLength": 30,
-            "minLength": 30
-          },
-          "refund_id": {
-            "type": "string",
-            "description": "Unique Identifier for the Refund. This is to ensure idempotency for multiple partial refunds initiated against the same payment. If this is not passed by the merchant, this field shall be auto generated and provided in the API response. It is recommended to generate uuid(v4) as the refund_id.",
-            "example": "ref_mbabizu24mvu3mela5njyhpit4",
-            "nullable": true,
-            "maxLength": 30,
-            "minLength": 30
-          },
-          "merchant_id": {
-            "type": "string",
-            "description": "The identifier for the Merchant Account",
-            "example": "y3oqhf46pyzuxjbcn2giaqnb44",
-            "nullable": true,
-            "maxLength": 255
-          },
-          "amount": {
-            "type": "integer",
-            "format": "int64",
-            "description": "Total amount for which the refund is to be initiated. Amount for the payment in lowest denomination of the currency. (i.e) in cents for USD denomination, in paisa for INR denomination etc., If not provided, this will default to the full payment amount",
-            "example": 6540,
-            "nullable": true,
-            "minimum": 100
-          },
-          "reason": {
-            "type": "string",
-            "description": "Reason for the refund. Often useful for displaying to users and your customer support executive. In case the payment went through Stripe, this field needs to be passed with one of these enums: `duplicate`, `fraudulent`, or `requested_by_customer`",
-            "example": "Customer returned the product",
-            "nullable": true,
-            "maxLength": 255
-          },
-          "refund_type": {
-            "allOf": [
-              {
-                "$ref": "#/components/schemas/RefundType"
-              }
-            ],
-            "default": "Instant",
-            "nullable": true
-          },
-          "metadata": {
-            "type": "object",
-            "description": "You can specify up to 50 keys, with key names up to 40 characters long and values up to 500 characters long. Metadata is useful for storing additional, structured information on an object.",
-            "nullable": true
-          },
-          "merchant_connector_details": {
-            "allOf": [
-              {
-                "$ref": "#/components/schemas/MerchantConnectorDetailsWrap"
-              }
-            ],
-            "nullable": true
-          },
-          "split_refunds": {
-            "allOf": [
-              {
-                "$ref": "#/components/schemas/SplitRefund"
-              }
-            ],
-            "nullable": true
-          }
-        },
-        "additionalProperties": false
-      },
       "RefundResponse": {
         "type": "object",
         "required": [
@@ -20535,7 +20459,7 @@
       },
       "RefundType": {
         "type": "string",
-        "description": "To indicate whether to refund needs to be instant or scheduled",
+        "description": "To indicate whether the refund needs to be instant or scheduled",
         "enum": [
           "scheduled",
           "instant"
@@ -20562,7 +20486,8 @@
       "RefundsCreateRequest": {
         "type": "object",
         "required": [
-          "payment_id"
+          "payment_id",
+          "merchant_reference_id"
         ],
         "properties": {
           "payment_id": {
@@ -20574,11 +20499,17 @@
           },
           "merchant_reference_id": {
             "type": "string",
-            "description": "Unique Identifier for the Refund. This is to ensure idempotency for multiple partial refunds initiated against the same payment.",
+            "description": "Unique Identifier for the Refund given by the Merchant.",
             "example": "ref_mbabizu24mvu3mela5njyhpit4",
+            "maxLength": 64,
+            "minLength": 1
+          },
+          "merchant_id": {
+            "type": "string",
+            "description": "The identifier for the Merchant Account",
+            "example": "y3oqhf46pyzuxjbcn2giaqnb44",
             "nullable": true,
-            "maxLength": 30,
-            "minLength": 30
+            "maxLength": 255
           },
           "amount": {
             "type": "integer",
diff --git a/crates/api_models/Cargo.toml b/crates/api_models/Cargo.toml
index f89558e6267..327406c198c 100644
--- a/crates/api_models/Cargo.toml
+++ b/crates/api_models/Cargo.toml
@@ -17,9 +17,10 @@ olap = []
 openapi = ["common_enums/openapi", "olap", "recon", "dummy_connector", "olap"]
 recon = []
 v1 = ["common_utils/v1"]
-v2 = ["common_types/v2", "common_utils/v2", "customer_v2"]
+v2 = ["common_types/v2", "common_utils/v2", "customer_v2", "refunds_v2"]
 customer_v2 = ["common_utils/customer_v2"]
 payment_methods_v2 = ["common_utils/payment_methods_v2"]
+refunds_v2 = []
 dynamic_routing = []
 control_center_theme = ["dep:actix-web", "dep:actix-multipart"]
 revenue_recovery = []
diff --git a/crates/api_models/src/events/refund.rs b/crates/api_models/src/events/refund.rs
index e87ae154471..888918d836e 100644
--- a/crates/api_models/src/events/refund.rs
+++ b/crates/api_models/src/events/refund.rs
@@ -1,9 +1,13 @@
 use common_utils::events::{ApiEventMetric, ApiEventsType};
 
+#[cfg(feature = "v1")]
+use crate::refunds::RefundRequest;
+#[cfg(feature = "v2")]
+use crate::refunds::RefundsCreateRequest;
 use crate::refunds::{
     RefundAggregateResponse, RefundListFilters, RefundListMetaData, RefundListRequest,
-    RefundListResponse, RefundManualUpdateRequest, RefundRequest, RefundResponse,
-    RefundUpdateRequest, RefundsRetrieveRequest,
+    RefundListResponse, RefundManualUpdateRequest, RefundResponse, RefundUpdateRequest,
+    RefundsRetrieveRequest,
 };
 
 #[cfg(feature = "v1")]
@@ -19,6 +23,13 @@ impl ApiEventMetric for RefundRequest {
     }
 }
 
+#[cfg(feature = "v2")]
+impl ApiEventMetric for RefundsCreateRequest {
+    fn get_api_event_type(&self) -> Option<ApiEventsType> {
+        None
+    }
+}
+
 #[cfg(feature = "v1")]
 impl ApiEventMetric for RefundResponse {
     fn get_api_event_type(&self) -> Option<ApiEventsType> {
diff --git a/crates/api_models/src/refunds.rs b/crates/api_models/src/refunds.rs
index ece1edb72bd..e9a7a77592d 100644
--- a/crates/api_models/src/refunds.rs
+++ b/crates/api_models/src/refunds.rs
@@ -12,6 +12,7 @@ use crate::{
     enums,
 };
 
+#[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "refunds_v2")))]
 #[derive(Default, Debug, ToSchema, Clone, Deserialize, Serialize)]
 #[serde(deny_unknown_fields)]
 pub struct RefundRequest {
@@ -61,7 +62,7 @@ pub struct RefundRequest {
     pub split_refunds: Option<common_types::refunds::SplitRefund>,
 }
 
-#[cfg(feature = "v2")]
+#[cfg(all(feature = "v2", feature = "refunds_v2"))]
 #[derive(Debug, ToSchema, Clone, Deserialize, Serialize)]
 #[serde(deny_unknown_fields)]
 pub struct RefundsCreateRequest {
@@ -74,14 +75,18 @@ pub struct RefundsCreateRequest {
     )]
     pub payment_id: common_utils::id_type::GlobalPaymentId,
 
-    /// Unique Identifier for the Refund. This is to ensure idempotency for multiple partial refunds initiated against the same payment.
+    /// Unique Identifier for the Refund given by the Merchant.
     #[schema(
-        max_length = 30,
-        min_length = 30,
+        max_length = 64,
+        min_length = 1,
         example = "ref_mbabizu24mvu3mela5njyhpit4",
-        value_type = Option<String>,
+        value_type = String,
     )]
-    pub merchant_reference_id: Option<common_utils::id_type::RefundReferenceId>,
+    pub merchant_reference_id: common_utils::id_type::RefundReferenceId,
+
+    /// The identifier for the Merchant Account
+    #[schema(max_length = 255, example = "y3oqhf46pyzuxjbcn2giaqnb44", value_type = Option<String>)]
+    pub merchant_id: Option<common_utils::id_type::MerchantId>,
 
     /// Total amount for which the refund is to be initiated. Amount for the payment in lowest denomination of the currency. (i.e) in cents for USD denomination, in paisa for INR denomination etc., If not provided, this will default to the amount_captured of the payment
     #[schema(value_type = Option<i64> , minimum = 100, example = 6540)]
@@ -153,6 +158,7 @@ pub struct RefundManualUpdateRequest {
     pub error_message: Option<String>,
 }
 
+#[cfg(feature = "v1")]
 /// To indicate whether to refund needs to be instant or scheduled
 #[derive(
     Default, Debug, Clone, Copy, ToSchema, Deserialize, Serialize, Eq, PartialEq, strum::Display,
@@ -164,6 +170,18 @@ pub enum RefundType {
     Instant,
 }
 
+#[cfg(feature = "v2")]
+/// To indicate whether the refund needs to be instant or scheduled
+#[derive(
+    Default, Debug, Clone, Copy, ToSchema, Deserialize, Serialize, Eq, PartialEq, strum::Display,
+)]
+#[serde(rename_all = "snake_case")]
+pub enum RefundType {
+    Scheduled,
+    #[default]
+    Instant,
+}
+
 #[cfg(feature = "v1")]
 #[derive(Debug, Clone, Eq, PartialEq, Deserialize, Serialize, ToSchema)]
 pub struct RefundResponse {
diff --git a/crates/common_utils/src/types.rs b/crates/common_utils/src/types.rs
index 4b24cf281cb..0dbf026263f 100644
--- a/crates/common_utils/src/types.rs
+++ b/crates/common_utils/src/types.rs
@@ -1204,6 +1204,14 @@ impl ConnectorTransactionId {
         }
     }
 
+    /// Implementation for extracting hashed data
+    pub fn extract_hashed_data(&self) -> Option<String> {
+        match self {
+            Self::TxnId(_) => None,
+            Self::HashedData(src) => Some(src.clone()),
+        }
+    }
+
     /// Implementation for retrieving
     pub fn get_txn_id<'a>(
         &'a self,
diff --git a/crates/diesel_models/src/query/payment_attempt.rs b/crates/diesel_models/src/query/payment_attempt.rs
index cfcbc2bd199..2543ab41461 100644
--- a/crates/diesel_models/src/query/payment_attempt.rs
+++ b/crates/diesel_models/src/query/payment_attempt.rs
@@ -161,6 +161,29 @@ impl PaymentAttempt {
         .ok_or(report!(DatabaseError::NotFound))
     }
 
+    #[cfg(feature = "v2")]
+    pub async fn find_last_successful_or_partially_captured_attempt_by_payment_id(
+        conn: &PgPooledConn,
+        payment_id: &common_utils::id_type::GlobalPaymentId,
+    ) -> StorageResult<Self> {
+        // perform ordering on the application level instead of database level
+        generics::generic_filter::<<Self as HasTable>::Table, _, _, Self>(
+            conn,
+            dsl::payment_id.eq(payment_id.to_owned()).and(
+                dsl::status
+                    .eq(enums::AttemptStatus::Charged)
+                    .or(dsl::status.eq(enums::AttemptStatus::PartialCharged)),
+            ),
+            Some(1),
+            None,
+            Some(dsl::modified_at.desc()),
+        )
+        .await?
+        .into_iter()
+        .nth(0)
+        .ok_or(report!(DatabaseError::NotFound))
+    }
+
     #[cfg(feature = "v1")]
     pub async fn find_by_merchant_id_connector_txn_id(
         conn: &PgPooledConn,
diff --git a/crates/diesel_models/src/query/refund.rs b/crates/diesel_models/src/query/refund.rs
index adae21c413f..cf42134cab0 100644
--- a/crates/diesel_models/src/query/refund.rs
+++ b/crates/diesel_models/src/query/refund.rs
@@ -170,4 +170,26 @@ impl Refund {
         )
         .await
     }
+
+    pub async fn find_by_merchant_id_connector_transaction_id(
+        conn: &PgPooledConn,
+        merchant_id: &common_utils::id_type::MerchantId,
+        connector_transaction_id: &str,
+    ) -> StorageResult<Vec<Self>> {
+        generics::generic_filter::<
+            <Self as HasTable>::Table,
+            _,
+            <<Self as HasTable>::Table as Table>::PrimaryKey,
+            _,
+        >(
+            conn,
+            dsl::merchant_id
+                .eq(merchant_id.to_owned())
+                .and(dsl::connector_transaction_id.eq(connector_transaction_id.to_owned())),
+            None,
+            None,
+            None,
+        )
+        .await
+    }
 }
diff --git a/crates/diesel_models/src/refund.rs b/crates/diesel_models/src/refund.rs
index 6a40baba9dd..e197bc3c211 100644
--- a/crates/diesel_models/src/refund.rs
+++ b/crates/diesel_models/src/refund.rs
@@ -11,7 +11,6 @@ use crate::enums as storage_enums;
 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,
@@ -702,8 +701,65 @@ impl RefundUpdate {
             ..source
         }
     }
+
+    pub fn build_error_update_for_unified_error_and_message(
+        unified_error_object: (String, String),
+        refund_error_message: Option<String>,
+        refund_error_code: Option<String>,
+        storage_scheme: &storage_enums::MerchantStorageScheme,
+    ) -> Self {
+        let (unified_code, unified_message) = unified_error_object;
+
+        Self::ErrorUpdate {
+            refund_status: Some(storage_enums::RefundStatus::Failure),
+            refund_error_message,
+            refund_error_code,
+            updated_by: storage_scheme.to_string(),
+            connector_refund_id: None,
+            processor_refund_data: None,
+            unified_code: Some(unified_code),
+            unified_message: Some(unified_message),
+        }
+    }
+
+    pub fn build_error_update_for_integrity_check_failure(
+        integrity_check_failed_fields: String,
+        connector_refund_id: Option<ConnectorTransactionId>,
+        storage_scheme: &storage_enums::MerchantStorageScheme,
+    ) -> Self {
+        Self::ErrorUpdate {
+            refund_status: Some(storage_enums::RefundStatus::ManualReview),
+            refund_error_message: Some(format!(
+                "Integrity Check Failed! as data mismatched for fields {}",
+                integrity_check_failed_fields
+            )),
+            refund_error_code: Some("IE".to_string()),
+            updated_by: storage_scheme.to_string(),
+            connector_refund_id: connector_refund_id.clone(),
+            processor_refund_data: connector_refund_id.and_then(|x| x.extract_hashed_data()),
+            unified_code: None,
+            unified_message: None,
+        }
+    }
+
+    pub fn build_refund_update(
+        connector_refund_id: ConnectorTransactionId,
+        refund_status: storage_enums::RefundStatus,
+        storage_scheme: &storage_enums::MerchantStorageScheme,
+    ) -> Self {
+        Self::Update {
+            connector_refund_id: connector_refund_id.clone(),
+            refund_status,
+            sent_to_gateway: true,
+            refund_error_message: None,
+            refund_arn: "".to_string(),
+            updated_by: storage_scheme.to_string(),
+            processor_refund_data: connector_refund_id.extract_hashed_data(),
+        }
+    }
 }
 
+#[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "refunds_v2")))]
 #[derive(Debug, Eq, PartialEq, Deserialize, Serialize)]
 pub struct RefundCoreWorkflow {
     pub refund_internal_reference_id: String,
@@ -713,6 +769,16 @@ pub struct RefundCoreWorkflow {
     pub processor_transaction_data: Option<String>,
 }
 
+#[cfg(all(feature = "v2", feature = "refunds_v2"))]
+#[derive(Debug, Eq, PartialEq, Deserialize, Serialize)]
+pub struct RefundCoreWorkflow {
+    pub refund_id: common_utils::id_type::GlobalRefundId,
+    pub connector_transaction_id: ConnectorTransactionId,
+    pub merchant_id: common_utils::id_type::MerchantId,
+    pub payment_id: common_utils::id_type::GlobalPaymentId,
+    pub processor_transaction_data: Option<String>,
+}
+
 #[cfg(feature = "v1")]
 impl common_utils::events::ApiEventMetric for Refund {
     fn get_api_event_type(&self) -> Option<common_utils::events::ApiEventsType> {
diff --git a/crates/hyperswitch_domain_models/src/payments.rs b/crates/hyperswitch_domain_models/src/payments.rs
index 90ba708c109..8179fd66581 100644
--- a/crates/hyperswitch_domain_models/src/payments.rs
+++ b/crates/hyperswitch_domain_models/src/payments.rs
@@ -674,6 +674,29 @@ impl PaymentIntent {
     pub fn get_feature_metadata(&self) -> Option<FeatureMetadata> {
         self.feature_metadata.clone()
     }
+
+    pub fn get_optional_customer_id(
+        &self,
+    ) -> CustomResult<Option<id_type::CustomerId>, common_utils::errors::ValidationError> {
+        self.customer_id
+            .as_ref()
+            .map(|customer_id| id_type::CustomerId::try_from(customer_id.clone()))
+            .transpose()
+    }
+
+    pub fn get_optional_connector_metadata(
+        &self,
+    ) -> CustomResult<
+        Option<api_models::payments::ConnectorMetadata>,
+        common_utils::errors::ParsingError,
+    > {
+        self.connector_metadata
+            .clone()
+            .map(|cm| {
+                cm.parse_value::<api_models::payments::ConnectorMetadata>("ConnectorMetadata")
+            })
+            .transpose()
+    }
 }
 
 #[cfg(feature = "v1")]
diff --git a/crates/hyperswitch_domain_models/src/payments/payment_attempt.rs b/crates/hyperswitch_domain_models/src/payments/payment_attempt.rs
index 494950d42c4..69e5064e6e1 100644
--- a/crates/hyperswitch_domain_models/src/payments/payment_attempt.rs
+++ b/crates/hyperswitch_domain_models/src/payments/payment_attempt.rs
@@ -117,6 +117,15 @@ pub trait PaymentAttemptInterface {
         storage_scheme: storage_enums::MerchantStorageScheme,
     ) -> error_stack::Result<PaymentAttempt, Self::Error>;
 
+    #[cfg(feature = "v2")]
+    async fn find_payment_attempt_last_successful_or_partially_captured_attempt_by_payment_id(
+        &self,
+        key_manager_state: &KeyManagerState,
+        merchant_key_store: &MerchantKeyStore,
+        payment_id: &id_type::GlobalPaymentId,
+        storage_scheme: storage_enums::MerchantStorageScheme,
+    ) -> error_stack::Result<PaymentAttempt, Self::Error>;
+
     #[cfg(feature = "v1")]
     async fn find_payment_attempt_by_merchant_id_connector_txn_id(
         &self,
@@ -1014,7 +1023,7 @@ impl PaymentAttempt {
     }
 
     pub fn get_total_surcharge_amount(&self) -> Option<MinorUnit> {
-        todo!();
+        self.amount_details.surcharge_amount
     }
 }
 
diff --git a/crates/openapi/Cargo.toml b/crates/openapi/Cargo.toml
index 7e7e5a6fb3c..33c48e2f0ae 100644
--- a/crates/openapi/Cargo.toml
+++ b/crates/openapi/Cargo.toml
@@ -17,7 +17,7 @@ common_types = { version = "0.1.0", path = "../common_types" }
 router_env = { version = "0.1.0", path = "../router_env" }
 
 [features]
-v2 = ["api_models/v2", "api_models/customer_v2", "common_utils/v2", "api_models/payment_methods_v2", "common_utils/payment_methods_v2"]
+v2 = ["api_models/v2", "api_models/customer_v2", "common_utils/v2", "api_models/payment_methods_v2", "common_utils/payment_methods_v2", "api_models/refunds_v2"]
 v1 = ["api_models/v1", "common_utils/v1"]
 
 [lints]
diff --git a/crates/openapi/src/openapi_v2.rs b/crates/openapi/src/openapi_v2.rs
index d2404e1069d..2629b1b8b7d 100644
--- a/crates/openapi/src/openapi_v2.rs
+++ b/crates/openapi/src/openapi_v2.rs
@@ -186,7 +186,6 @@ Never share your secret api keys. Keep them guarded and secure.
         common_types::refunds::SplitRefund,
         common_types::payments::ConnectorChargeResponseData,
         common_types::payments::StripeChargeResponseData,
-        api_models::refunds::RefundRequest,
         api_models::refunds::RefundsCreateRequest,
         api_models::refunds::RefundErrorDetails,
         api_models::refunds::RefundType,
diff --git a/crates/router/Cargo.toml b/crates/router/Cargo.toml
index 0eae31ace87..7dfb87c3669 100644
--- a/crates/router/Cargo.toml
+++ b/crates/router/Cargo.toml
@@ -39,7 +39,7 @@ customer_v2 = ["api_models/customer_v2", "diesel_models/customer_v2", "hyperswit
 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"]
+refunds_v2 = ["api_models/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/core.rs b/crates/router/src/core.rs
index 83c1058b426..e96aa339e9e 100644
--- a/crates/router/src/core.rs
+++ b/crates/router/src/core.rs
@@ -42,6 +42,9 @@ pub mod poll;
 pub mod recon;
 #[cfg(feature = "v1")]
 pub mod refunds;
+#[cfg(feature = "v2")]
+pub mod refunds_v2;
+
 pub mod routing;
 pub mod surcharge_decision_config;
 #[cfg(feature = "olap")]
diff --git a/crates/router/src/core/refunds.rs b/crates/router/src/core/refunds.rs
index 6e576509a18..649e9ea57a0 100644
--- a/crates/router/src/core/refunds.rs
+++ b/crates/router/src/core/refunds.rs
@@ -1,6 +1,3 @@
-pub mod transformers;
-pub mod validator;
-
 #[cfg(feature = "olap")]
 use std::collections::HashMap;
 
@@ -27,7 +24,10 @@ use crate::{
         errors::{self, ConnectorErrorExt, RouterResponse, RouterResult, StorageErrorExt},
         payments::{self, access_token, helpers},
         refunds::transformers::SplitRefundInput,
-        utils as core_utils,
+        utils::{
+            self as core_utils, refunds_transformers as transformers,
+            refunds_validator as validator,
+        },
     },
     db, logger,
     routes::{metrics, SessionState},
diff --git a/crates/router/src/core/refunds_v2.rs b/crates/router/src/core/refunds_v2.rs
new file mode 100644
index 00000000000..7186bb76bff
--- /dev/null
+++ b/crates/router/src/core/refunds_v2.rs
@@ -0,0 +1,840 @@
+use std::str::FromStr;
+
+use api_models::{enums::Connector, refunds::RefundErrorDetails};
+use common_utils::{id_type, types as common_utils_types};
+use error_stack::{report, ResultExt};
+use hyperswitch_domain_models::router_data::{ErrorResponse, RouterData};
+use hyperswitch_interfaces::integrity::{CheckIntegrity, FlowIntegrity, GetIntegrityObject};
+use router_env::{instrument, tracing};
+
+use crate::{
+    consts,
+    core::{
+        errors::{self, ConnectorErrorExt, StorageErrorExt},
+        payments::{self, access_token, helpers},
+        utils::{self as core_utils, refunds_validator},
+    },
+    db, logger,
+    routes::{metrics, SessionState},
+    services,
+    types::{
+        self,
+        api::{self, refunds},
+        domain,
+        storage::{self, enums},
+        transformers::{ForeignFrom, ForeignTryFrom},
+    },
+    utils,
+};
+
+#[instrument(skip_all)]
+pub async fn refund_create_core(
+    state: SessionState,
+    merchant_account: domain::MerchantAccount,
+    key_store: domain::MerchantKeyStore,
+    req: refunds::RefundsCreateRequest,
+) -> errors::RouterResponse<refunds::RefundResponse> {
+    let db = &*state.store;
+    let (payment_intent, payment_attempt, amount);
+
+    payment_intent = db
+        .find_payment_intent_by_id(
+            &(&state).into(),
+            &req.payment_id,
+            &key_store,
+            merchant_account.storage_scheme,
+        )
+        .await
+        .to_not_found_response(errors::ApiErrorResponse::PaymentNotFound)?;
+
+    utils::when(
+        !(payment_intent.status == enums::IntentStatus::Succeeded
+            || payment_intent.status == enums::IntentStatus::PartiallyCaptured),
+        || {
+            Err(report!(errors::ApiErrorResponse::PaymentUnexpectedState {
+                current_flow: "refund".into(),
+                field_name: "status".into(),
+                current_value: payment_intent.status.to_string(),
+                states: "succeeded, partially_captured".to_string()
+            })
+            .attach_printable("unable to refund for a unsuccessful payment intent"))
+        },
+    )?;
+
+    let captured_amount = payment_intent
+        .amount_captured
+        .ok_or(errors::ApiErrorResponse::InternalServerError)
+        .attach_printable("amount captured is none in a successful payment")?;
+
+    // Amount is not passed in request refer from payment intent.
+    amount = req.amount.unwrap_or(captured_amount);
+
+    utils::when(amount <= common_utils_types::MinorUnit::new(0), || {
+        Err(report!(errors::ApiErrorResponse::InvalidDataFormat {
+            field_name: "amount".to_string(),
+            expected_format: "positive integer".to_string()
+        })
+        .attach_printable("amount less than or equal to zero"))
+    })?;
+
+    payment_attempt = db
+        .find_payment_attempt_last_successful_or_partially_captured_attempt_by_payment_id(
+            &(&state).into(),
+            &key_store,
+            &req.payment_id,
+            merchant_account.storage_scheme,
+        )
+        .await
+        .to_not_found_response(errors::ApiErrorResponse::SuccessfulPaymentNotFound)?;
+
+    let global_refund_id = id_type::GlobalRefundId::generate(&state.conf.cell_information.id);
+
+    tracing::Span::current().record("global_refund_id", global_refund_id.get_string_repr());
+
+    Box::pin(validate_and_create_refund(
+        &state,
+        &merchant_account,
+        &key_store,
+        &payment_attempt,
+        &payment_intent,
+        amount,
+        req,
+        global_refund_id,
+    ))
+    .await
+    .map(services::ApplicationResponse::Json)
+}
+
+#[allow(clippy::too_many_arguments)]
+#[instrument(skip_all)]
+pub async fn trigger_refund_to_gateway(
+    state: &SessionState,
+    refund: &storage::Refund,
+    merchant_account: &domain::MerchantAccount,
+    key_store: &domain::MerchantKeyStore,
+    payment_attempt: &storage::PaymentAttempt,
+    payment_intent: &storage::PaymentIntent,
+) -> errors::RouterResult<storage::Refund> {
+    let db = &*state.store;
+
+    let mca_id = payment_attempt.get_attempt_merchant_connector_account_id()?;
+
+    let storage_scheme = merchant_account.storage_scheme;
+
+    let mca = db
+        .find_merchant_connector_account_by_id(&state.into(), &mca_id, key_store)
+        .await
+        .change_context(errors::ApiErrorResponse::InternalServerError)
+        .attach_printable("Failed to fetch merchant connector account")?;
+
+    metrics::REFUND_COUNT.add(
+        1,
+        router_env::metric_attributes!(("connector", mca_id.get_string_repr().to_string())),
+    );
+
+    let connector_enum = mca.connector_name;
+
+    let connector: api::ConnectorData = api::ConnectorData::get_connector_by_name(
+        &state.conf.connectors,
+        &connector_enum.to_string(),
+        api::GetToken::Connector,
+        Some(mca_id.clone()),
+    )?;
+
+    refunds_validator::validate_for_valid_refunds(payment_attempt, connector.connector_name)?;
+
+    let mut router_data = core_utils::construct_refund_router_data(
+        state,
+        connector_enum,
+        merchant_account,
+        key_store,
+        payment_intent,
+        payment_attempt,
+        refund,
+        &mca,
+    )
+    .await?;
+
+    let add_access_token_result =
+        access_token::add_access_token(state, &connector, merchant_account, &router_data, None)
+            .await?;
+
+    logger::debug!(refund_router_data=?router_data);
+
+    access_token::update_router_data_with_access_token_result(
+        &add_access_token_result,
+        &mut router_data,
+        &payments::CallConnectorAction::Trigger,
+    );
+
+    let connector_response =
+        call_connector_service(state, &connector, add_access_token_result, router_data).await;
+
+    let refund_update = get_refund_update_object(
+        state,
+        &connector,
+        &storage_scheme,
+        merchant_account,
+        &connector_response,
+    )
+    .await;
+
+    let response = match refund_update {
+        Some(refund_update) => state
+            .store
+            .update_refund(
+                refund.to_owned(),
+                refund_update,
+                merchant_account.storage_scheme,
+            )
+            .await
+            .to_not_found_response(errors::ApiErrorResponse::InternalServerError)
+            .attach_printable_lazy(|| {
+                format!(
+                    "Failed while updating refund: refund_id: {}",
+                    refund.id.get_string_repr()
+                )
+            })?,
+        None => refund.to_owned(),
+    };
+    // Implement outgoing webhooks here
+    connector_response.to_refund_failed_response()?;
+    Ok(response)
+}
+
+async fn call_connector_service(
+    state: &SessionState,
+    connector: &api::ConnectorData,
+    add_access_token_result: types::AddAccessTokenResult,
+    router_data: RouterData<api::Execute, types::RefundsData, types::RefundsResponseData>,
+) -> Result<
+    RouterData<api::Execute, types::RefundsData, types::RefundsResponseData>,
+    error_stack::Report<errors::ConnectorError>,
+> {
+    if !(add_access_token_result.connector_supports_access_token
+        && router_data.access_token.is_none())
+    {
+        let connector_integration: services::BoxedRefundConnectorIntegrationInterface<
+            api::Execute,
+            types::RefundsData,
+            types::RefundsResponseData,
+        > = connector.connector.get_connector_integration();
+        services::execute_connector_processing_step(
+            state,
+            connector_integration,
+            &router_data,
+            payments::CallConnectorAction::Trigger,
+            None,
+        )
+        .await
+    } else {
+        Ok(router_data)
+    }
+}
+
+async fn get_refund_update_object(
+    state: &SessionState,
+    connector: &api::ConnectorData,
+    storage_scheme: &enums::MerchantStorageScheme,
+    merchant_account: &domain::MerchantAccount,
+    router_data_response: &Result<
+        RouterData<api::Execute, types::RefundsData, types::RefundsResponseData>,
+        error_stack::Report<errors::ConnectorError>,
+    >,
+) -> Option<storage::RefundUpdate> {
+    match router_data_response {
+        // This error is related to connector implementation i.e if no implementation for refunds for that specific connector in HS or the connector does not support refund itself.
+        Err(err) => get_connector_implementation_error_refund_update(err, *storage_scheme),
+        Ok(response) => {
+            let response = perform_integrity_check(response.clone());
+            match response.response.clone() {
+                Err(err) => Some(
+                    get_connector_error_refund_update(state, err, connector, storage_scheme).await,
+                ),
+                Ok(refund_response_data) => Some(get_refund_update_for_refund_response_data(
+                    response,
+                    connector,
+                    refund_response_data,
+                    storage_scheme,
+                    merchant_account,
+                )),
+            }
+        }
+    }
+}
+
+fn get_connector_implementation_error_refund_update(
+    error: &error_stack::Report<errors::ConnectorError>,
+    storage_scheme: enums::MerchantStorageScheme,
+) -> Option<storage::RefundUpdate> {
+    Option::<storage::RefundUpdate>::foreign_from((error.current_context(), storage_scheme))
+}
+
+async fn get_connector_error_refund_update(
+    state: &SessionState,
+    err: ErrorResponse,
+    connector: &api::ConnectorData,
+    storage_scheme: &enums::MerchantStorageScheme,
+) -> storage::RefundUpdate {
+    let unified_error_object = get_unified_error_and_message(state, &err, connector).await;
+
+    storage::RefundUpdate::build_error_update_for_unified_error_and_message(
+        unified_error_object,
+        err.reason.or(Some(err.message)),
+        Some(err.code),
+        storage_scheme,
+    )
+}
+
+async fn get_unified_error_and_message(
+    state: &SessionState,
+    err: &ErrorResponse,
+    connector: &api::ConnectorData,
+) -> (String, String) {
+    let option_gsm = helpers::get_gsm_record(
+        state,
+        Some(err.code.clone()),
+        Some(err.message.clone()),
+        connector.connector_name.to_string(),
+        consts::REFUND_FLOW_STR.to_string(),
+    )
+    .await;
+    // Note: Some connectors do not have a separate list of refund errors
+    // In such cases, the error codes and messages are stored under "Authorize" flow in GSM table
+    // So we will have to fetch the GSM using Authorize flow in case GSM is not found using "refund_flow"
+    let option_gsm = if option_gsm.is_none() {
+        helpers::get_gsm_record(
+            state,
+            Some(err.code.clone()),
+            Some(err.message.clone()),
+            connector.connector_name.to_string(),
+            consts::AUTHORIZE_FLOW_STR.to_string(),
+        )
+        .await
+    } else {
+        option_gsm
+    };
+
+    let gsm_unified_code = option_gsm.as_ref().and_then(|gsm| gsm.unified_code.clone());
+    let gsm_unified_message = option_gsm.and_then(|gsm| gsm.unified_message);
+
+    match gsm_unified_code.as_ref().zip(gsm_unified_message.as_ref()) {
+        Some((code, message)) => (code.to_owned(), message.to_owned()),
+        None => (
+            consts::DEFAULT_UNIFIED_ERROR_CODE.to_owned(),
+            consts::DEFAULT_UNIFIED_ERROR_MESSAGE.to_owned(),
+        ),
+    }
+}
+
+pub fn get_refund_update_for_refund_response_data(
+    router_data: RouterData<api::Execute, types::RefundsData, types::RefundsResponseData>,
+    connector: &api::ConnectorData,
+    refund_response_data: types::RefundsResponseData,
+    storage_scheme: &enums::MerchantStorageScheme,
+    merchant_account: &domain::MerchantAccount,
+) -> storage::RefundUpdate {
+    // match on connector integrity checks
+    match router_data.integrity_check.clone() {
+        Err(err) => {
+            let connector_refund_id = err
+                .connector_transaction_id
+                .map(common_utils_types::ConnectorTransactionId::from);
+
+            metrics::INTEGRITY_CHECK_FAILED.add(
+                1,
+                router_env::metric_attributes!(
+                    ("connector", connector.connector_name.to_string()),
+                    ("merchant_id", merchant_account.get_id().clone()),
+                ),
+            );
+
+            storage::RefundUpdate::build_error_update_for_integrity_check_failure(
+                err.field_names,
+                connector_refund_id,
+                storage_scheme,
+            )
+        }
+        Ok(()) => {
+            if refund_response_data.refund_status == diesel_models::enums::RefundStatus::Success {
+                metrics::SUCCESSFUL_REFUND.add(
+                    1,
+                    router_env::metric_attributes!((
+                        "connector",
+                        connector.connector_name.to_string(),
+                    )),
+                )
+            }
+
+            let connector_refund_id = common_utils_types::ConnectorTransactionId::from(
+                refund_response_data.connector_refund_id,
+            );
+
+            storage::RefundUpdate::build_refund_update(
+                connector_refund_id,
+                refund_response_data.refund_status,
+                storage_scheme,
+            )
+        }
+    }
+}
+
+pub fn perform_integrity_check(
+    mut router_data: RouterData<api::Execute, types::RefundsData, types::RefundsResponseData>,
+) -> RouterData<api::Execute, types::RefundsData, types::RefundsResponseData> {
+    // Initiating Integrity check
+    let integrity_result = check_refund_integrity(&router_data.request, &router_data.response);
+    router_data.integrity_check = integrity_result;
+    router_data
+}
+
+impl ForeignFrom<(&errors::ConnectorError, enums::MerchantStorageScheme)>
+    for Option<storage::RefundUpdate>
+{
+    fn foreign_from(
+        (from, storage_scheme): (&errors::ConnectorError, enums::MerchantStorageScheme),
+    ) -> Self {
+        match from {
+            errors::ConnectorError::NotImplemented(message) => {
+                Some(storage::RefundUpdate::ErrorUpdate {
+                    refund_status: Some(enums::RefundStatus::Failure),
+                    refund_error_message: Some(
+                        errors::ConnectorError::NotImplemented(message.to_owned()).to_string(),
+                    ),
+                    refund_error_code: Some("NOT_IMPLEMENTED".to_string()),
+                    updated_by: storage_scheme.to_string(),
+                    connector_refund_id: None,
+                    processor_refund_data: None,
+                    unified_code: None,
+                    unified_message: None,
+                })
+            }
+            errors::ConnectorError::NotSupported { message, connector } => {
+                Some(storage::RefundUpdate::ErrorUpdate {
+                    refund_status: Some(enums::RefundStatus::Failure),
+                    refund_error_message: Some(format!(
+                        "{message} is not supported by {connector}"
+                    )),
+                    refund_error_code: Some("NOT_SUPPORTED".to_string()),
+                    updated_by: storage_scheme.to_string(),
+                    connector_refund_id: None,
+                    processor_refund_data: None,
+                    unified_code: None,
+                    unified_message: None,
+                })
+            }
+            _ => None,
+        }
+    }
+}
+
+pub fn check_refund_integrity<T, Request>(
+    request: &Request,
+    refund_response_data: &Result<types::RefundsResponseData, ErrorResponse>,
+) -> Result<(), common_utils::errors::IntegrityCheckError>
+where
+    T: FlowIntegrity,
+    Request: GetIntegrityObject<T> + CheckIntegrity<Request, T>,
+{
+    let connector_refund_id = refund_response_data
+        .as_ref()
+        .map(|resp_data| resp_data.connector_refund_id.clone())
+        .ok();
+
+    request.check_integrity(request, connector_refund_id.to_owned())
+}
+
+// ********************************************** VALIDATIONS **********************************************
+
+#[instrument(skip_all)]
+#[allow(clippy::too_many_arguments)]
+pub async fn validate_and_create_refund(
+    state: &SessionState,
+    merchant_account: &domain::MerchantAccount,
+    key_store: &domain::MerchantKeyStore,
+    payment_attempt: &storage::PaymentAttempt,
+    payment_intent: &storage::PaymentIntent,
+    refund_amount: common_utils_types::MinorUnit,
+    req: refunds::RefundsCreateRequest,
+    global_refund_id: id_type::GlobalRefundId,
+) -> errors::RouterResult<refunds::RefundResponse> {
+    let db = &*state.store;
+
+    let refund_type = req.refund_type.unwrap_or_default();
+
+    let merchant_reference_id = req.merchant_reference_id;
+
+    let predicate = req
+        .merchant_id
+        .as_ref()
+        .map(|merchant_id| merchant_id != merchant_account.get_id());
+
+    utils::when(predicate.unwrap_or(false), || {
+        Err(report!(errors::ApiErrorResponse::InvalidDataFormat {
+            field_name: "merchant_id".to_string(),
+            expected_format: "merchant_id from merchant account".to_string()
+        })
+        .attach_printable("invalid merchant_id in request"))
+    })?;
+
+    let connector_payment_id = payment_attempt.clone().connector_payment_id.ok_or_else(|| {
+        report!(errors::ApiErrorResponse::InternalServerError)
+        .attach_printable("Transaction in invalid. Missing field \"connector_transaction_id\" in payment_attempt.")
+    })?;
+
+    let all_refunds = db
+        .find_refund_by_merchant_id_connector_transaction_id(
+            merchant_account.get_id(),
+            &connector_payment_id,
+            merchant_account.storage_scheme,
+        )
+        .await
+        .to_not_found_response(errors::ApiErrorResponse::RefundNotFound)?;
+
+    let currency = payment_intent.amount_details.currency;
+
+    refunds_validator::validate_payment_order_age(
+        &payment_intent.created_at,
+        state.conf.refund.max_age,
+    )
+    .change_context(errors::ApiErrorResponse::InvalidDataFormat {
+        field_name: "created_at".to_string(),
+        expected_format: format!(
+            "created_at not older than {} days",
+            state.conf.refund.max_age,
+        ),
+    })?;
+
+    let total_amount_captured = payment_intent
+        .amount_captured
+        .unwrap_or(payment_attempt.get_total_amount());
+
+    refunds_validator::validate_refund_amount(
+        total_amount_captured.get_amount_as_i64(),
+        &all_refunds,
+        refund_amount.get_amount_as_i64(),
+    )
+    .change_context(errors::ApiErrorResponse::RefundAmountExceedsPaymentAmount)?;
+
+    refunds_validator::validate_maximum_refund_against_payment_attempt(
+        &all_refunds,
+        state.conf.refund.max_attempts,
+    )
+    .change_context(errors::ApiErrorResponse::MaximumRefundCount)?;
+
+    let connector = payment_attempt
+        .connector
+        .clone()
+        .ok_or(errors::ApiErrorResponse::InternalServerError)
+        .attach_printable("No connector populated in payment attempt")?;
+    let (connector_transaction_id, processor_transaction_data) =
+        common_utils_types::ConnectorTransactionId::form_id_and_data(connector_payment_id);
+    let refund_create_req = storage::RefundNew {
+        id: global_refund_id,
+        merchant_reference_id: merchant_reference_id.clone(),
+        external_reference_id: Some(merchant_reference_id.get_string_repr().to_string()),
+        payment_id: req.payment_id,
+        merchant_id: merchant_account.get_id().clone(),
+        connector_transaction_id,
+        connector,
+        refund_type: enums::RefundType::foreign_from(req.refund_type.unwrap_or_default()),
+        total_amount: payment_attempt.get_total_amount(),
+        refund_amount,
+        currency,
+        created_at: common_utils::date_time::now(),
+        modified_at: common_utils::date_time::now(),
+        refund_status: enums::RefundStatus::Pending,
+        metadata: req.metadata,
+        description: req.reason.clone(),
+        attempt_id: payment_attempt.id.clone(),
+        refund_reason: req.reason,
+        profile_id: Some(payment_intent.profile_id.clone()),
+        connector_id: payment_attempt.merchant_connector_id.clone(),
+        charges: None,
+        split_refunds: None,
+        connector_refund_id: None,
+        sent_to_gateway: Default::default(),
+        refund_arn: None,
+        updated_by: Default::default(),
+        organization_id: merchant_account.organization_id.clone(),
+        processor_transaction_data,
+        processor_refund_data: None,
+    };
+
+    let refund = match db
+        .insert_refund(refund_create_req, merchant_account.storage_scheme)
+        .await
+    {
+        Ok(refund) => {
+            Box::pin(schedule_refund_execution(
+                state,
+                refund.clone(),
+                refund_type,
+                merchant_account,
+                key_store,
+                payment_attempt,
+                payment_intent,
+            ))
+            .await?
+        }
+        Err(err) => {
+            if err.current_context().is_db_unique_violation() {
+                Err(errors::ApiErrorResponse::DuplicateRefundRequest)?
+            } else {
+                Err(err)
+                    .change_context(errors::ApiErrorResponse::RefundFailed { data: None })
+                    .attach_printable("Failed to insert refund")?
+            }
+        }
+    };
+
+    let unified_translated_message =
+        match (refund.unified_code.clone(), refund.unified_message.clone()) {
+            (Some(unified_code), Some(unified_message)) => helpers::get_unified_translation(
+                state,
+                unified_code,
+                unified_message.clone(),
+                state.locale.to_string(),
+            )
+            .await
+            .or(Some(unified_message)),
+            _ => refund.unified_message,
+        };
+
+    let refund = storage::Refund {
+        unified_message: unified_translated_message,
+        ..refund
+    };
+
+    api::RefundResponse::foreign_try_from(refund)
+}
+
+impl ForeignTryFrom<storage::Refund> for api::RefundResponse {
+    type Error = error_stack::Report<errors::ApiErrorResponse>;
+    fn foreign_try_from(refund: storage::Refund) -> Result<Self, Self::Error> {
+        let refund = refund;
+
+        let profile_id = refund
+            .profile_id
+            .clone()
+            .ok_or(errors::ApiErrorResponse::InternalServerError)
+            .attach_printable("Profile id not found")?;
+
+        let merchant_connector_id = refund
+            .connector_id
+            .clone()
+            .ok_or(errors::ApiErrorResponse::InternalServerError)
+            .attach_printable("Connector id not found")?;
+
+        let connector_name = refund.connector;
+        let connector = Connector::from_str(&connector_name)
+            .change_context(errors::ConnectorError::InvalidConnectorName)
+            .change_context(errors::ApiErrorResponse::InvalidDataValue {
+                field_name: "connector",
+            })
+            .attach_printable_lazy(|| {
+                format!("unable to parse connector name {connector_name:?}")
+            })?;
+
+        Ok(Self {
+            payment_id: refund.payment_id,
+            id: refund.id.clone(),
+            amount: refund.refund_amount,
+            currency: refund.currency,
+            reason: refund.refund_reason,
+            status: refunds::RefundStatus::foreign_from(refund.refund_status),
+            profile_id,
+            metadata: refund.metadata,
+            created_at: refund.created_at,
+            updated_at: refund.modified_at,
+            connector,
+            merchant_connector_id,
+            merchant_reference_id: Some(refund.merchant_reference_id),
+            error_details: Some(RefundErrorDetails {
+                code: refund.refund_error_code.unwrap_or_default(),
+                message: refund.refund_error_message.unwrap_or_default(),
+            }),
+            connector_refund_reference_id: None,
+        })
+    }
+}
+
+// ********************************************** PROCESS TRACKER **********************************************
+
+#[instrument(skip_all)]
+#[allow(clippy::too_many_arguments)]
+pub async fn schedule_refund_execution(
+    state: &SessionState,
+    refund: storage::Refund,
+    refund_type: api_models::refunds::RefundType,
+    merchant_account: &domain::MerchantAccount,
+    key_store: &domain::MerchantKeyStore,
+    payment_attempt: &storage::PaymentAttempt,
+    payment_intent: &storage::PaymentIntent,
+) -> errors::RouterResult<storage::Refund> {
+    let db = &*state.store;
+    let runner = storage::ProcessTrackerRunner::RefundWorkflowRouter;
+    let task = "EXECUTE_REFUND";
+    let task_id = format!("{runner}_{task}_{}", refund.id.get_string_repr());
+
+    let refund_process = db
+        .find_process_by_id(&task_id)
+        .await
+        .change_context(errors::ApiErrorResponse::InternalServerError)
+        .attach_printable("Failed to find the process id")?;
+
+    let result = match refund.refund_status {
+        enums::RefundStatus::Pending | enums::RefundStatus::ManualReview => {
+            match (refund.sent_to_gateway, refund_process) {
+                (false, None) => {
+                    // Execute the refund task based on refund_type
+                    match refund_type {
+                        api_models::refunds::RefundType::Scheduled => {
+                            add_refund_execute_task(db, &refund, runner)
+                                .await
+                                .change_context(errors::ApiErrorResponse::InternalServerError)
+                                .attach_printable_lazy(|| format!("Failed while pushing refund execute task to scheduler, refund_id: {}", refund.id.get_string_repr()))?;
+
+                            Ok(refund)
+                        }
+                        api_models::refunds::RefundType::Instant => {
+                            let update_refund = Box::pin(trigger_refund_to_gateway(
+                                state,
+                                &refund,
+                                merchant_account,
+                                key_store,
+                                payment_attempt,
+                                payment_intent,
+                            ))
+                            .await;
+
+                            match update_refund {
+                                Ok(updated_refund_data) => {
+                                    add_refund_sync_task(db, &updated_refund_data, runner)
+                                        .await
+                                        .change_context(errors::ApiErrorResponse::InternalServerError)
+                                        .attach_printable_lazy(|| format!(
+                                            "Failed while pushing refund sync task in scheduler: refund_id: {}",
+                                            refund.id.get_string_repr()
+                                        ))?;
+                                    Ok(updated_refund_data)
+                                }
+                                Err(err) => Err(err),
+                            }
+                        }
+                    }
+                }
+                _ => {
+                    // Sync the refund for status check
+                    //[#300]: return refund status response
+                    match refund_type {
+                        api_models::refunds::RefundType::Scheduled => {
+                            add_refund_sync_task(db, &refund, runner)
+                                .await
+                                .change_context(errors::ApiErrorResponse::InternalServerError)
+                                .attach_printable_lazy(|| format!("Failed while pushing refund sync task in scheduler: refund_id: {}", refund.id.get_string_repr()))?;
+                            Ok(refund)
+                        }
+                        api_models::refunds::RefundType::Instant => {
+                            // [#255]: This is not possible in schedule_refund_execution as it will always be scheduled
+                            // sync_refund_with_gateway(data, &refund).await
+                            Ok(refund)
+                        }
+                    }
+                }
+            }
+        }
+        //  [#255]: This is not allowed to be otherwise or all
+        _ => Ok(refund),
+    }?;
+    Ok(result)
+}
+
+#[instrument]
+pub fn refund_to_refund_core_workflow_model(
+    refund: &storage::Refund,
+) -> storage::RefundCoreWorkflow {
+    storage::RefundCoreWorkflow {
+        refund_id: refund.id.clone(),
+        connector_transaction_id: refund.connector_transaction_id.clone(),
+        merchant_id: refund.merchant_id.clone(),
+        payment_id: refund.payment_id.clone(),
+        processor_transaction_data: refund.processor_transaction_data.clone(),
+    }
+}
+
+#[instrument(skip_all)]
+pub async fn add_refund_execute_task(
+    db: &dyn db::StorageInterface,
+    refund: &storage::Refund,
+    runner: storage::ProcessTrackerRunner,
+) -> errors::RouterResult<storage::ProcessTracker> {
+    let task = "EXECUTE_REFUND";
+    let process_tracker_id = format!("{runner}_{task}_{}", refund.id.get_string_repr());
+    let tag = ["REFUND"];
+    let schedule_time = common_utils::date_time::now();
+    let refund_workflow_tracking_data = refund_to_refund_core_workflow_model(refund);
+    let process_tracker_entry = storage::ProcessTrackerNew::new(
+        process_tracker_id,
+        task,
+        runner,
+        tag,
+        refund_workflow_tracking_data,
+        None,
+        schedule_time,
+        common_types::consts::API_VERSION,
+    )
+    .change_context(errors::ApiErrorResponse::InternalServerError)
+    .attach_printable("Failed to construct refund execute process tracker task")?;
+
+    let response = db
+        .insert_process(process_tracker_entry)
+        .await
+        .to_duplicate_response(errors::ApiErrorResponse::DuplicateRefundRequest)
+        .attach_printable_lazy(|| {
+            format!(
+                "Failed while inserting task in process_tracker: refund_id: {}",
+                refund.id.get_string_repr()
+            )
+        })?;
+    Ok(response)
+}
+
+#[instrument(skip_all)]
+pub async fn add_refund_sync_task(
+    db: &dyn db::StorageInterface,
+    refund: &storage::Refund,
+    runner: storage::ProcessTrackerRunner,
+) -> errors::RouterResult<storage::ProcessTracker> {
+    let task = "SYNC_REFUND";
+    let process_tracker_id = format!("{runner}_{task}_{}", refund.id.get_string_repr());
+    let schedule_time = common_utils::date_time::now();
+    let refund_workflow_tracking_data = refund_to_refund_core_workflow_model(refund);
+    let tag = ["REFUND"];
+    let process_tracker_entry = storage::ProcessTrackerNew::new(
+        process_tracker_id,
+        task,
+        runner,
+        tag,
+        refund_workflow_tracking_data,
+        None,
+        schedule_time,
+        common_types::consts::API_VERSION,
+    )
+    .change_context(errors::ApiErrorResponse::InternalServerError)
+    .attach_printable("Failed to construct refund sync process tracker task")?;
+
+    let response = db
+        .insert_process(process_tracker_entry)
+        .await
+        .to_duplicate_response(errors::ApiErrorResponse::DuplicateRefundRequest)
+        .attach_printable_lazy(|| {
+            format!(
+                "Failed while inserting task in process_tracker: refund_id: {}",
+                refund.id.get_string_repr()
+            )
+        })?;
+    metrics::TASKS_ADDED_COUNT.add(1, router_env::metric_attributes!(("flow", "Refund")));
+
+    Ok(response)
+}
diff --git a/crates/router/src/core/utils.rs b/crates/router/src/core/utils.rs
index 87dce4ade0b..35901ed8ac3 100644
--- a/crates/router/src/core/utils.rs
+++ b/crates/router/src/core/utils.rs
@@ -1,6 +1,9 @@
+pub mod refunds_transformers;
+pub mod refunds_validator;
+
 use std::{collections::HashSet, marker::PhantomData, str::FromStr};
 
-use api_models::enums::{DisputeStage, DisputeStatus};
+use api_models::enums::{Connector, DisputeStage, DisputeStatus};
 #[cfg(feature = "payouts")]
 use api_models::payouts::PayoutVendorAccountDetails;
 use common_enums::{IntentStatus, RequestIncrementalAuthorization};
@@ -16,6 +19,8 @@ use hyperswitch_domain_models::{
     merchant_connector_account::MerchantConnectorAccount, payment_address::PaymentAddress,
     router_data::ErrorResponse, router_request_types, types::OrderDetailsWithAmount,
 };
+#[cfg(all(feature = "v2", feature = "refunds_v2"))]
+use masking::ExposeOptionInterface;
 #[cfg(feature = "payouts")]
 use masking::{ExposeInterface, PeekInterface};
 use maud::{html, PreEscaped};
@@ -232,18 +237,155 @@ pub async fn construct_payout_router_data<'a, F>(
 #[instrument(skip_all)]
 #[allow(clippy::too_many_arguments)]
 pub async fn construct_refund_router_data<'a, F>(
-    _state: &'a SessionState,
-    _connector_id: &str,
-    _merchant_account: &domain::MerchantAccount,
-    _key_store: &domain::MerchantKeyStore,
-    _money: (MinorUnit, enums::Currency),
-    _payment_intent: &'a storage::PaymentIntent,
-    _payment_attempt: &storage::PaymentAttempt,
-    _refund: &'a storage::Refund,
-    _creds_identifier: Option<String>,
-    _split_refunds: Option<router_request_types::SplitRefundsRequest>,
+    state: &'a SessionState,
+    connector_enum: Connector,
+    merchant_account: &domain::MerchantAccount,
+    key_store: &domain::MerchantKeyStore,
+    payment_intent: &'a storage::PaymentIntent,
+    payment_attempt: &storage::PaymentAttempt,
+    refund: &'a storage::Refund,
+    merchant_connector_account: &MerchantConnectorAccount,
 ) -> RouterResult<types::RefundsRouterData<F>> {
-    todo!()
+    let auth_type = merchant_connector_account
+        .get_connector_account_details()
+        .change_context(errors::ApiErrorResponse::InternalServerError)?;
+
+    let status = payment_attempt.status;
+
+    let payment_amount = payment_attempt.get_total_amount();
+    let currency = payment_intent.amount_details.currency;
+
+    let payment_method_type = payment_attempt.payment_method_type;
+
+    let merchant_connector_account_id = &merchant_connector_account.id;
+
+    let webhook_url = Some(helpers::create_webhook_url(
+        &state.base_url.clone(),
+        merchant_account.get_id(),
+        merchant_connector_account_id.get_string_repr(),
+    ));
+
+    let supported_connector = &state
+        .conf
+        .multiple_api_version_supported_connectors
+        .supported_connectors;
+
+    let connector_api_version = if supported_connector.contains(&connector_enum) {
+        state
+            .store
+            .find_config_by_key(&format!("connector_api_version_{connector_enum}"))
+            .await
+            .map(|value| value.config)
+            .ok()
+    } else {
+        None
+    };
+
+    let browser_info = payment_attempt
+        .browser_info
+        .clone()
+        .map(types::BrowserInformation::from);
+
+    let connector_refund_id = refund.get_optional_connector_refund_id().cloned();
+    let capture_method = payment_intent.capture_method;
+
+    let customer_id = payment_intent
+        .get_optional_customer_id()
+        .change_context(errors::ApiErrorResponse::InternalServerError)
+        .attach_printable("Failed to get optional customer id")?;
+
+    let braintree_metadata = payment_intent
+        .get_optional_connector_metadata()
+        .change_context(errors::ApiErrorResponse::InternalServerError)
+        .attach_printable("Failed parsing ConnectorMetadata")?
+        .and_then(|cm| cm.braintree);
+
+    let merchant_account_id = braintree_metadata
+        .as_ref()
+        .and_then(|braintree| braintree.merchant_account_id.clone());
+
+    let merchant_config_currency =
+        braintree_metadata.and_then(|braintree| braintree.merchant_config_currency);
+
+    let router_data = types::RouterData {
+        flow: PhantomData,
+        merchant_id: merchant_account.get_id().clone(),
+        customer_id,
+        tenant_id: state.tenant.tenant_id.clone(),
+        connector: connector_enum.to_string(),
+        payment_id: payment_attempt.payment_id.get_string_repr().to_owned(),
+        attempt_id: payment_attempt.id.get_string_repr().to_string().clone(),
+        status,
+        payment_method: payment_method_type,
+        connector_auth_type: auth_type,
+        description: None,
+        // Does refund need shipping/billing address ?
+        address: PaymentAddress::default(),
+        auth_type: payment_attempt.authentication_type,
+        connector_meta_data: merchant_connector_account.get_metadata(),
+        connector_wallets_details: merchant_connector_account.get_connector_wallets_details(),
+        amount_captured: payment_intent
+            .amount_captured
+            .map(|amt| amt.get_amount_as_i64()),
+        payment_method_status: None,
+        minor_amount_captured: payment_intent.amount_captured,
+        request: types::RefundsData {
+            refund_id: refund.id.get_string_repr().to_string(),
+            connector_transaction_id: refund.get_connector_transaction_id().clone(),
+            refund_amount: refund.refund_amount.get_amount_as_i64(),
+            minor_refund_amount: refund.refund_amount,
+            currency,
+            payment_amount: payment_amount.get_amount_as_i64(),
+            minor_payment_amount: payment_amount,
+            webhook_url,
+            connector_metadata: payment_attempt.connector_metadata.clone().expose_option(),
+            reason: refund.refund_reason.clone(),
+            connector_refund_id: connector_refund_id.clone(),
+            browser_info,
+            split_refunds: None,
+            integrity_object: None,
+            refund_status: refund.refund_status,
+            merchant_account_id,
+            merchant_config_currency,
+            refund_connector_metadata: refund.metadata.clone(),
+            capture_method: Some(capture_method),
+        },
+
+        response: Ok(types::RefundsResponseData {
+            connector_refund_id: connector_refund_id.unwrap_or_default(),
+            refund_status: refund.refund_status,
+        }),
+        access_token: None,
+        session_token: None,
+        reference_id: None,
+        payment_method_token: None,
+        connector_customer: None,
+        recurring_mandate_payment_data: None,
+        preprocessing_id: None,
+        connector_request_reference_id: refund.id.get_string_repr().to_string().clone(),
+        #[cfg(feature = "payouts")]
+        payout_method_data: None,
+        #[cfg(feature = "payouts")]
+        quote_id: None,
+        test_mode: None,
+        payment_method_balance: None,
+        connector_api_version,
+        connector_http_status_code: None,
+        external_latency: None,
+        apple_pay_flow: None,
+        frm_metadata: None,
+        refund_id: Some(refund.id.get_string_repr().to_string().clone()),
+        dispute_id: None,
+        connector_response: None,
+        integrity_check: Ok(()),
+        additional_merchant_data: None,
+        header_payload: None,
+        connector_mandate_request_reference_id: None,
+        authentication_id: None,
+        psd2_sca_exemption_type: None,
+    };
+
+    Ok(router_data)
 }
 
 #[cfg(feature = "v1")]
@@ -309,7 +451,7 @@ pub async fn construct_refund_router_data<'a, F>(
         .conf
         .multiple_api_version_supported_connectors
         .supported_connectors;
-    let connector_enum = api_models::enums::Connector::from_str(connector_id)
+    let connector_enum = Connector::from_str(connector_id)
         .change_context(errors::ConnectorError::InvalidConnectorName)
         .change_context(errors::ApiErrorResponse::InvalidDataValue {
             field_name: "connector",
@@ -485,7 +627,7 @@ pub fn validate_uuid(uuid: String, key: &str) -> Result<String, errors::ApiError
 
 #[cfg(feature = "v1")]
 pub fn get_split_refunds(
-    split_refund_input: super::refunds::transformers::SplitRefundInput,
+    split_refund_input: refunds_transformers::SplitRefundInput,
 ) -> RouterResult<Option<router_request_types::SplitRefundsRequest>> {
     match split_refund_input.split_payment_request.as_ref() {
         Some(common_types::payments::SplitPaymentsRequest::StripeSplitPayment(stripe_payment)) => {
@@ -515,7 +657,7 @@ pub fn get_split_refunds(
             };
 
             if let Some(charge_id) = charge_id_option {
-                let options = super::refunds::validator::validate_stripe_charge_refund(
+                let options = refunds_validator::validate_stripe_charge_refund(
                     charge_type_option,
                     &split_refund_input.refund_request,
                 )?;
@@ -543,7 +685,7 @@ pub fn get_split_refunds(
                         split_refund_request,
                     )) = split_refund_input.refund_request.clone()
                     {
-                        super::refunds::validator::validate_adyen_charge_refund(
+                        refunds_validator::validate_adyen_charge_refund(
                             adyen_split_payment_response,
                             &split_refund_request,
                         )?;
@@ -573,7 +715,7 @@ pub fn get_split_refunds(
                         split_refund_request,
                     )),
                 ) => {
-                    let user_id = super::refunds::validator::validate_xendit_charge_refund(
+                    let user_id = refunds_validator::validate_xendit_charge_refund(
                         xendit_split_payment_response,
                         split_refund_request,
                     )?;
@@ -1472,7 +1614,7 @@ pub async fn validate_and_get_business_profile(
 }
 
 fn connector_needs_business_sub_label(connector_name: &str) -> bool {
-    let connectors_list = [api_models::enums::Connector::Cybersource];
+    let connectors_list = [Connector::Cybersource];
     connectors_list
         .map(|connector| connector.to_string())
         .contains(&connector_name.to_string())
diff --git a/crates/router/src/core/refunds/transformers.rs b/crates/router/src/core/utils/refunds_transformers.rs
similarity index 100%
rename from crates/router/src/core/refunds/transformers.rs
rename to crates/router/src/core/utils/refunds_transformers.rs
diff --git a/crates/router/src/core/refunds/validator.rs b/crates/router/src/core/utils/refunds_validator.rs
similarity index 90%
rename from crates/router/src/core/refunds/validator.rs
rename to crates/router/src/core/utils/refunds_validator.rs
index 573543dfbf7..5f982388264 100644
--- a/crates/router/src/core/refunds/validator.rs
+++ b/crates/router/src/core/utils/refunds_validator.rs
@@ -110,6 +110,7 @@ pub fn validate_refund_list(limit: Option<i64>) -> CustomResult<i64, errors::Api
     }
 }
 
+#[cfg(feature = "v1")]
 pub fn validate_for_valid_refunds(
     payment_attempt: &hyperswitch_domain_models::payments::payment_attempt::PaymentAttempt,
     connector: api_models::enums::Connector,
@@ -146,6 +147,38 @@ pub fn validate_for_valid_refunds(
     }
 }
 
+#[cfg(feature = "v2")]
+pub fn validate_for_valid_refunds(
+    payment_attempt: &hyperswitch_domain_models::payments::payment_attempt::PaymentAttempt,
+    connector: api_models::enums::Connector,
+) -> RouterResult<()> {
+    let payment_method_type = payment_attempt.payment_method_type;
+
+    match payment_method_type {
+        diesel_models::enums::PaymentMethod::PayLater
+        | diesel_models::enums::PaymentMethod::Wallet => {
+            let payment_method_subtype = payment_attempt.payment_method_subtype;
+
+            utils::when(
+                matches!(
+                    (connector, payment_method_subtype),
+                    (
+                        api_models::enums::Connector::Braintree,
+                        diesel_models::enums::PaymentMethodType::Paypal,
+                    )
+                ),
+                || {
+                    Err(errors::ApiErrorResponse::RefundNotPossible {
+                        connector: connector.to_string(),
+                    }
+                    .into())
+                },
+            )
+        }
+        _ => Ok(()),
+    }
+}
+
 pub fn validate_stripe_charge_refund(
     charge_type_option: Option<api_enums::PaymentChargeType>,
     split_refund_request: &Option<common_types::refunds::SplitRefund>,
diff --git a/crates/router/src/db/kafka_store.rs b/crates/router/src/db/kafka_store.rs
index 866df70e69b..304428ca141 100644
--- a/crates/router/src/db/kafka_store.rs
+++ b/crates/router/src/db/kafka_store.rs
@@ -1721,6 +1721,24 @@ impl PaymentAttemptInterface for KafkaStore {
             .await
     }
 
+    #[cfg(feature = "v2")]
+    async fn find_payment_attempt_last_successful_or_partially_captured_attempt_by_payment_id(
+        &self,
+        key_manager_state: &KeyManagerState,
+        merchant_key_store: &domain::MerchantKeyStore,
+        payment_id: &id_type::GlobalPaymentId,
+        storage_scheme: MerchantStorageScheme,
+    ) -> CustomResult<storage::PaymentAttempt, errors::StorageError> {
+        self.diesel_store
+            .find_payment_attempt_last_successful_or_partially_captured_attempt_by_payment_id(
+                key_manager_state,
+                merchant_key_store,
+                payment_id,
+                storage_scheme,
+            )
+            .await
+    }
+
     #[cfg(feature = "v1")]
     async fn find_payment_attempt_by_preprocessing_id_merchant_id(
         &self,
@@ -2759,7 +2777,6 @@ 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,
diff --git a/crates/router/src/db/refund.rs b/crates/router/src/db/refund.rs
index 211f8ddbc48..fbbe0a6d159 100644
--- a/crates/router/src/db/refund.rs
+++ b/crates/router/src/db/refund.rs
@@ -57,7 +57,6 @@ 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,
@@ -632,6 +631,24 @@ mod storage {
             }
         }
 
+        #[cfg(all(feature = "v2", feature = "refunds_v2"))]
+        #[instrument(skip_all)]
+        async fn find_refund_by_merchant_id_connector_transaction_id(
+            &self,
+            merchant_id: &common_utils::id_type::MerchantId,
+            connector_transaction_id: &str,
+            _storage_scheme: enums::MerchantStorageScheme,
+        ) -> CustomResult<Vec<storage_types::Refund>, errors::StorageError> {
+            let conn = connection::pg_connection_read(self).await?;
+            storage_types::Refund::find_by_merchant_id_connector_transaction_id(
+                &conn,
+                merchant_id,
+                connector_transaction_id,
+            )
+            .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 update_refund(
@@ -1092,7 +1109,6 @@ impl RefundInterface for MockDb {
         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,
diff --git a/crates/router/src/lib.rs b/crates/router/src/lib.rs
index dc7cb1233fa..2b499c882c8 100644
--- a/crates/router/src/lib.rs
+++ b/crates/router/src/lib.rs
@@ -154,7 +154,9 @@ pub fn mk_app(
 
         #[cfg(all(feature = "v2", feature = "oltp"))]
         {
-            server_app = server_app.service(routes::PaymentMethodSession::server(state.clone()));
+            server_app = server_app
+                .service(routes::PaymentMethodSession::server(state.clone()))
+                .service(routes::Refunds::server(state.clone()));
         }
 
         #[cfg(feature = "v1")]
diff --git a/crates/router/src/routes.rs b/crates/router/src/routes.rs
index 34feb63ea78..03a5589f770 100644
--- a/crates/router/src/routes.rs
+++ b/crates/router/src/routes.rs
@@ -44,7 +44,6 @@ pub mod poll;
 pub mod profiles;
 #[cfg(feature = "recon")]
 pub mod recon;
-#[cfg(feature = "v1")]
 pub mod refunds;
 #[cfg(feature = "olap")]
 pub mod routing;
diff --git a/crates/router/src/routes/app.rs b/crates/router/src/routes/app.rs
index d2d31790914..9dc67be1d2d 100644
--- a/crates/router/src/routes/app.rs
+++ b/crates/router/src/routes/app.rs
@@ -48,6 +48,8 @@ use super::pm_auth;
 use super::poll;
 #[cfg(all(feature = "v2", feature = "revenue_recovery", feature = "oltp"))]
 use super::recovery_webhooks::*;
+#[cfg(all(feature = "oltp", feature = "v2"))]
+use super::refunds;
 #[cfg(feature = "olap")]
 use super::routing;
 #[cfg(all(feature = "olap", feature = "v1"))]
@@ -1163,6 +1165,17 @@ impl Refunds {
     }
 }
 
+#[cfg(all(feature = "v2", feature = "refunds_v2", feature = "oltp"))]
+impl Refunds {
+    pub fn server(state: AppState) -> Scope {
+        let mut route = web::scope("/v2/refunds").app_data(web::Data::new(state));
+
+        route = route.service(web::resource("").route(web::post().to(refunds::refunds_create)));
+
+        route
+    }
+}
+
 #[cfg(feature = "payouts")]
 pub struct Payouts;
 
diff --git a/crates/router/src/routes/refunds.rs b/crates/router/src/routes/refunds.rs
index aa2afa1c329..c4d157d7de4 100644
--- a/crates/router/src/routes/refunds.rs
+++ b/crates/router/src/routes/refunds.rs
@@ -1,16 +1,21 @@
 use actix_web::{web, HttpRequest, HttpResponse};
+use common_utils;
 use router_env::{instrument, tracing, Flow};
 
 use super::app::AppState;
+#[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "refunds_v2")))]
+use crate::core::refunds::*;
+#[cfg(all(feature = "v2", feature = "refunds_v2"))]
+use crate::core::refunds_v2::*;
 use crate::{
-    core::{api_locking, refunds::*},
+    core::api_locking,
     services::{api, authentication as auth, authorization::permissions::Permission},
     types::api::refunds,
 };
-
 /// Refunds - Create
 ///
 /// To create a refund against an already processed payment
+#[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "refunds_v2")))]
 #[utoipa::path(
     post,
     path = "/refunds",
@@ -59,6 +64,41 @@ pub async fn refunds_create(
     ))
     .await
 }
+
+#[cfg(all(feature = "v2", feature = "refunds_v2"))]
+#[instrument(skip_all, fields(flow = ?Flow::RefundsCreate))]
+// #[post("")]
+pub async fn refunds_create(
+    state: web::Data<AppState>,
+    req: HttpRequest,
+    json_payload: web::Json<refunds::RefundsCreateRequest>,
+) -> HttpResponse {
+    let flow = Flow::RefundsCreate;
+
+    Box::pin(api::server_wrap(
+        flow,
+        state,
+        &req,
+        json_payload.into_inner(),
+        |state, auth: auth::AuthenticationData, req, _| {
+            refund_create_core(state, auth.merchant_account, auth.key_store, req)
+        },
+        auth::auth_type(
+            &auth::V2ApiKeyAuth {
+                is_connected_allowed: false,
+                is_platform_allowed: false,
+            },
+            &auth::JWTAuth {
+                permission: Permission::ProfileRefundWrite,
+            },
+            req.headers(),
+        ),
+        api_locking::LockAction::NotApplicable,
+    ))
+    .await
+}
+
+#[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "refunds_v2")))]
 /// Refunds - Retrieve (GET)
 ///
 /// To retrieve the properties of a Refund. This may be used to get the status of a previously initiated payment or next action for an ongoing payment
@@ -125,6 +165,8 @@ pub async fn refunds_retrieve(
     ))
     .await
 }
+
+#[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "refunds_v2")))]
 /// Refunds - Retrieve (POST)
 ///
 /// To retrieve the properties of a Refund. This may be used to get the status of a previously initiated payment or next action for an ongoing payment
@@ -176,6 +218,8 @@ pub async fn refunds_retrieve_with_body(
     ))
     .await
 }
+
+#[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "refunds_v2")))]
 /// Refunds - Update
 ///
 /// To update the properties of a Refund object. This may include attaching a reason for the refund or metadata fields
@@ -221,6 +265,12 @@ pub async fn refunds_update(
     ))
     .await
 }
+
+#[cfg(all(
+    any(feature = "v1", feature = "v2"),
+    not(feature = "refunds_v2"),
+    feature = "olap"
+))]
 /// Refunds - List
 ///
 /// To list the refunds associated with a payment_id or with the merchant, if payment_id is not provided
@@ -236,7 +286,6 @@ pub async fn refunds_update(
     security(("api_key" = []))
 )]
 #[instrument(skip_all, fields(flow = ?Flow::RefundsList))]
-#[cfg(feature = "olap")]
 pub async fn refunds_list(
     state: web::Data<AppState>,
     req: HttpRequest,
@@ -266,6 +315,11 @@ pub async fn refunds_list(
     .await
 }
 
+#[cfg(all(
+    any(feature = "v1", feature = "v2"),
+    not(feature = "refunds_v2"),
+    feature = "olap"
+))]
 /// Refunds - List at profile level
 ///
 /// To list the refunds associated with a payment_id or with the merchant, if payment_id is not provided
@@ -281,7 +335,6 @@ pub async fn refunds_list(
     security(("api_key" = []))
 )]
 #[instrument(skip_all, fields(flow = ?Flow::RefundsList))]
-#[cfg(feature = "olap")]
 pub async fn refunds_list_profile(
     state: web::Data<AppState>,
     req: HttpRequest,
@@ -316,6 +369,11 @@ pub async fn refunds_list_profile(
     .await
 }
 
+#[cfg(all(
+    any(feature = "v1", feature = "v2"),
+    not(feature = "refunds_v2"),
+    feature = "olap"
+))]
 /// Refunds - Filter
 ///
 /// To list the refunds filters associated with list of connectors, currencies and payment statuses
@@ -331,7 +389,6 @@ pub async fn refunds_list_profile(
     security(("api_key" = []))
 )]
 #[instrument(skip_all, fields(flow = ?Flow::RefundsList))]
-#[cfg(feature = "olap")]
 pub async fn refunds_filter_list(
     state: web::Data<AppState>,
     req: HttpRequest,
@@ -361,6 +418,11 @@ pub async fn refunds_filter_list(
     .await
 }
 
+#[cfg(all(
+    any(feature = "v1", feature = "v2"),
+    not(feature = "refunds_v2"),
+    feature = "olap"
+))]
 /// Refunds - Filter V2
 ///
 /// To list the refunds filters associated with list of connectors, currencies and payment statuses
@@ -375,7 +437,6 @@ pub async fn refunds_filter_list(
     security(("api_key" = []))
 )]
 #[instrument(skip_all, fields(flow = ?Flow::RefundsFilters))]
-#[cfg(feature = "olap")]
 pub async fn get_refunds_filters(state: web::Data<AppState>, req: HttpRequest) -> HttpResponse {
     let flow = Flow::RefundsFilters;
     Box::pin(api::server_wrap(
@@ -401,6 +462,11 @@ pub async fn get_refunds_filters(state: web::Data<AppState>, req: HttpRequest) -
     .await
 }
 
+#[cfg(all(
+    any(feature = "v1", feature = "v2"),
+    not(feature = "refunds_v2"),
+    feature = "olap"
+))]
 /// Refunds - Filter V2 at profile level
 ///
 /// To list the refunds filters associated with list of connectors, currencies and payment statuses
@@ -415,7 +481,6 @@ pub async fn get_refunds_filters(state: web::Data<AppState>, req: HttpRequest) -
     security(("api_key" = []))
 )]
 #[instrument(skip_all, fields(flow = ?Flow::RefundsFilters))]
-#[cfg(feature = "olap")]
 pub async fn get_refunds_filters_profile(
     state: web::Data<AppState>,
     req: HttpRequest,
@@ -448,8 +513,12 @@ pub async fn get_refunds_filters_profile(
     .await
 }
 
+#[cfg(all(
+    any(feature = "v1", feature = "v2"),
+    not(feature = "refunds_v2"),
+    feature = "olap"
+))]
 #[instrument(skip_all, fields(flow = ?Flow::RefundsAggregate))]
-#[cfg(feature = "olap")]
 pub async fn get_refunds_aggregates(
     state: web::Data<AppState>,
     req: HttpRequest,
@@ -480,8 +549,12 @@ pub async fn get_refunds_aggregates(
     .await
 }
 
+#[cfg(all(
+    any(feature = "v1", feature = "v2"),
+    not(feature = "refunds_v2"),
+    feature = "olap"
+))]
 #[instrument(skip_all, fields(flow = ?Flow::RefundsManualUpdate))]
-#[cfg(feature = "olap")]
 pub async fn refunds_manual_update(
     state: web::Data<AppState>,
     req: HttpRequest,
@@ -503,8 +576,12 @@ pub async fn refunds_manual_update(
     .await
 }
 
+#[cfg(all(
+    any(feature = "v1", feature = "v2"),
+    not(feature = "refunds_v2"),
+    feature = "olap"
+))]
 #[instrument(skip_all, fields(flow = ?Flow::RefundsAggregate))]
-#[cfg(feature = "olap")]
 pub async fn get_refunds_aggregate_profile(
     state: web::Data<AppState>,
     req: HttpRequest,
diff --git a/crates/router/src/types/api/refunds.rs b/crates/router/src/types/api/refunds.rs
index 7aafe4de152..ff1f192fd45 100644
--- a/crates/router/src/types/api/refunds.rs
+++ b/crates/router/src/types/api/refunds.rs
@@ -1,6 +1,9 @@
+#[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "refunds_v2")))]
+pub use api_models::refunds::RefundRequest;
+#[cfg(all(feature = "v2", feature = "refunds_v2"))]
+pub use api_models::refunds::RefundsCreateRequest;
 pub use api_models::refunds::{
-    RefundRequest, RefundResponse, RefundStatus, RefundType, RefundUpdateRequest,
-    RefundsRetrieveRequest,
+    RefundResponse, RefundStatus, RefundType, RefundUpdateRequest, RefundsRetrieveRequest,
 };
 pub use hyperswitch_domain_models::router_flow_types::refunds::{Execute, RSync};
 pub use hyperswitch_interfaces::api::refunds::{Refund, RefundExecute, RefundSync};
diff --git a/crates/storage_impl/src/mock_db/payment_attempt.rs b/crates/storage_impl/src/mock_db/payment_attempt.rs
index 06f98bb26b0..baafc60f316 100644
--- a/crates/storage_impl/src/mock_db/payment_attempt.rs
+++ b/crates/storage_impl/src/mock_db/payment_attempt.rs
@@ -342,4 +342,26 @@ impl PaymentAttemptInterface for MockDb {
             .cloned()
             .unwrap())
     }
+
+    #[cfg(feature = "v2")]
+    #[allow(clippy::unwrap_used)]
+    async fn find_payment_attempt_last_successful_or_partially_captured_attempt_by_payment_id(
+        &self,
+        _key_manager_state: &KeyManagerState,
+        _merchant_key_store: &MerchantKeyStore,
+        payment_id: &id_type::GlobalPaymentId,
+        _storage_scheme: storage_enums::MerchantStorageScheme,
+    ) -> CustomResult<PaymentAttempt, StorageError> {
+        let payment_attempts = self.payment_attempts.lock().await;
+
+        Ok(payment_attempts
+            .iter()
+            .find(|payment_attempt| {
+                payment_attempt.payment_id == *payment_id
+                    && (payment_attempt.status == storage_enums::AttemptStatus::PartialCharged
+                        || payment_attempt.status == storage_enums::AttemptStatus::Charged)
+            })
+            .cloned()
+            .unwrap())
+    }
 }
diff --git a/crates/storage_impl/src/payments/payment_attempt.rs b/crates/storage_impl/src/payments/payment_attempt.rs
index 0156270484c..9720545821b 100644
--- a/crates/storage_impl/src/payments/payment_attempt.rs
+++ b/crates/storage_impl/src/payments/payment_attempt.rs
@@ -216,6 +216,33 @@ impl<T: DatabaseStore> PaymentAttemptInterface for RouterStore<T> {
         .map(PaymentAttempt::from_storage_model)
     }
 
+    #[cfg(feature = "v2")]
+    #[instrument(skip_all)]
+    async fn find_payment_attempt_last_successful_or_partially_captured_attempt_by_payment_id(
+        &self,
+        key_manager_state: &KeyManagerState,
+        merchant_key_store: &MerchantKeyStore,
+        payment_id: &common_utils::id_type::GlobalPaymentId,
+        _storage_scheme: MerchantStorageScheme,
+    ) -> CustomResult<PaymentAttempt, errors::StorageError> {
+        let conn = pg_connection_read(self).await?;
+        DieselPaymentAttempt::find_last_successful_or_partially_captured_attempt_by_payment_id(
+            &conn, payment_id,
+        )
+        .await
+        .map_err(|er| {
+            let new_err = diesel_error_to_data_error(*er.current_context());
+            er.change_context(new_err)
+        })?
+        .convert(
+            key_manager_state,
+            merchant_key_store.key.get_inner(),
+            merchant_key_store.merchant_id.clone().into(),
+        )
+        .await
+        .change_context(errors::StorageError::DecryptionError)
+    }
+
     #[instrument(skip_all)]
     #[cfg(feature = "v1")]
     async fn find_payment_attempt_by_merchant_id_connector_txn_id(
@@ -1050,6 +1077,26 @@ impl<T: DatabaseStore> PaymentAttemptInterface for KVRouterStore<T> {
         }
     }
 
+    #[cfg(feature = "v2")]
+    #[instrument(skip_all)]
+    async fn find_payment_attempt_last_successful_or_partially_captured_attempt_by_payment_id(
+        &self,
+        key_manager_state: &KeyManagerState,
+        merchant_key_store: &MerchantKeyStore,
+        payment_id: &common_utils::id_type::GlobalPaymentId,
+        storage_scheme: MerchantStorageScheme,
+    ) -> error_stack::Result<PaymentAttempt, errors::StorageError> {
+        // Ignoring storage scheme for v2 implementation
+        self.router_store
+            .find_payment_attempt_last_successful_or_partially_captured_attempt_by_payment_id(
+                key_manager_state,
+                merchant_key_store,
+                payment_id,
+                storage_scheme,
+            )
+            .await
+    }
+
     #[cfg(feature = "v2")]
     async fn find_payment_attempt_by_profile_id_connector_transaction_id(
         &self,

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

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

## Description
<!-- Describe your changes in detail -->
This PR is a part of v2 implementation of refunds. This PR contains refunds create core flow, i.e we can be able to create refunds for payments in v2.

**Important**
We are moving `refunds/transformers.rs` and `refunds/validator.rs` to a separate location because of accessibility
to both v1 and v2 refund code.  

### Additional Changes

- [x] 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 support for creating refunds for payments 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)?
-->
- Create Payment Intent api call
```
curl --location 'http://localhost:8080/v2/payments/create-intent' \
--header 'api-key: xyz' \
--header 'Content-Type: application/json' \
--header 'x-profile-id: pro_' \
--header 'Authorization: api-key=xyz' \
--data-raw '{
    "amount_details": {
        "order_amount": 100,
        "currency": "USD"
    },
    "capture_method":"automatic",
    "authentication_type": "no_three_ds",
    "billing": {
        "address": {
            "first_name": "John",
            "last_name": "Dough"
        },
        "email": "example@example.com"
    },
    "shipping": {
        "address": {
            "first_name": "John",
            "last_name": "Dough",
            "city": "Karwar",
            "zip": "581301",
            "state": "Karnataka"
        },
        "email": "example@example.com"
    }
}'
```
- Response from the above call
```json
{
  "id": "12345_pay_0195d13dd6497c12a3b187039fc4b71e",
  "status": "requires_payment_method",
  "amount_details": {
    "order_amount": 100,
    "currency": "USD",
    "shipping_cost": null,
    "order_tax_amount": null,
    "external_tax_calculation": "skip",
    "surcharge_calculation": "skip",
    "surcharge_amount": null,
    "tax_on_surcharge": null
  },
  "client_secret": "12345_pay_0195d13dd6497c12a3b187039fc4b71e_secret_0195d13dd6547a938bfe7b97b36203a5",
  "profile_id": "pro_TujgCYCsvWvO6YNsRAmC",
  "merchant_reference_id": null,
  "routing_algorithm_id": null,
  "capture_method": "automatic",
  "authentication_type": "no_three_ds",
  "billing": {
    "address": {
      "city": null,
      "country": null,
      "line1": null,
      "line2": null,
      "line3": null,
      "zip": null,
      "state": null,
      "first_name": "John",
      "last_name": "Dough"
    },
    "phone": null,
    "email": "example@example.com"
  },
  "shipping": {
    "address": {
      "city": "Karwar",
      "country": null,
      "line1": null,
      "line2": null,
      "line3": null,
      "zip": "581301",
      "state": "Karnataka",
      "first_name": "John",
      "last_name": "Dough"
    },
    "phone": null,
    "email": "example@example.com"
  },
  "customer_id": null,
  "customer_present": "present",
  "description": null,
  "return_url": null,
  "setup_future_usage": "on_session",
  "apply_mit_exemption": "Skip",
  "statement_descriptor": null,
  "order_details": null,
  "allowed_payment_method_types": null,
  "metadata": null,
  "connector_metadata": null,
  "feature_metadata": null,
  "payment_link_enabled": "Skip",
  "payment_link_config": null,
  "request_incremental_authorization": "default",
  "expires_on": "2025-03-26T07:12:25.588Z",
  "frm_metadata": null,
  "request_external_three_ds_authentication": "Skip"
}
```
- Confirm Intent API call
```
curl --location 'http://localhost:8080/v2/payments/12345_pay_0195d13dd6497c12a3b187039fc4b71e/confirm-intent' \
--header 'x-client-secret: 12345_pay_0195d13dd6497c12a3b187039fc4b71e_secret_0195d13dd6547a938bfe7b97b36203a5' \
--header 'x-profile-id: pro_TujgCYCsvWvO6YNsRAmC' \
--header 'Content-Type: application/json' \
--header 'Authorization: publishable-key=pk_dev_58559cdf90434adcb0b92eeb6a8b633f,client-secret=12345_pay_0195d13dd6497c12a3b187039fc4b71e_secret_0195d13dd6547a938bfe7b97b36203a5' \
--data '{
    "payment_method_data": {
        "card": {
            "card_number": "4242424242424242",
            "card_exp_month": "09",
            "card_exp_year": "25",
            "card_holder_name": "John Doe",
            "card_cvc": "100"
        }
    },
    "payment_method_type": "card",
    "payment_method_subtype": "credit"
}'
```
- Response from the above call
```json
{
  "id": "12345_pay_0195d13dd6497c12a3b187039fc4b71e",
  "status": "succeeded",
  "amount": {
    "order_amount": 100,
    "currency": "USD",
    "shipping_cost": null,
    "order_tax_amount": null,
    "external_tax_calculation": "skip",
    "surcharge_calculation": "skip",
    "surcharge_amount": null,
    "tax_on_surcharge": null,
    "net_amount": 100,
    "amount_to_capture": null,
    "amount_capturable": 0,
    "amount_captured": 100
  },
  "customer_id": null,
  "connector": "stripe",
  "client_secret": "12345_pay_0195d13dd6497c12a3b187039fc4b71e_secret_0195d13dd6547a938bfe7b97b36203a5",
  "created": "2025-03-26T06:57:25.588Z",
  "payment_method_data": {
    "billing": null
  },
  "payment_method_type": "card",
  "payment_method_subtype": "credit",
  "connector_transaction_id": "pi_3R6ntzD5R7gDAGff0MYUxZFn",
  "connector_reference_id": null,
  "merchant_connector_id": "mca_FhEKUEAUdgv7mVCs3A9U",
  "browser_info": null,
  "error": null,
  "shipping": null,
  "billing": null,
  "attempts": null,
  "connector_token_details": {
    "token": "pm_1R6ntzD5R7gDAGff07DogD0c",
    "connector_token_request_reference_id": "oMf8ppmFul0oTqI6Zl"
  },
  "payment_method_id": null,
  "next_action": null,
  "return_url": "https://google.com/success",
  "authentication_type": "no_three_ds",
  "authentication_type_applied": "no_three_ds"
}
```
- Create refund for a successful payment
```
curl --location 'http://localhost:8080/v2/refunds' \
--header 'X-Profile-Id: pro_TujgCYCsvWvO6YNsRAmC' \
--header 'Content-Type: application/json' \
--header 'api-key: dev_kksGlKY14xNypCzsrJIXFCzL3t2j5kUs7ChmloPryJsdRy2RWYljonzPCxJSDmuZ' \
--data '{
    "payment_id":"12345_pay_0195d13dd6497c12a3b187039fc4b71e",
    "merchant_reference_id":"1742972385",
    "reason":"Paid by mistake",
    "metadata":{
        "foo":"bar"
    }
}'
```
- Response from the above call
```json
{
  "id": "12345_ref_0195d13dfedb7790bc1a84be825e64d5",
  "payment_id": "12345_pay_0195d13dd6497c12a3b187039fc4b71e",
  "merchant_reference_id": "1742972256",
  "amount": 100,
  "currency": "USD",
  "status": "succeeded",
  "reason": "Paid by mistake",
  "metadata": {
    "foo": "bar"
  },
  "error_details": {
    "code": "",
    "message": ""
  },
  "created_at": "2025-03-26T06:57:35.993Z",
  "updated_at": "2025-03-26T06:57:36.981Z",
  "connector": "stripe",
  "profile_id": "pro_TujgCYCsvWvO6YNsRAmC",
  "merchant_connector_id": "mca_FhEKUEAUdgv7mVCs3A9U",
  "connector_refund_reference_id": "12345_ref_0195d13dfedb7790bc1a84be825e64d5"
}
```
- Trying the same refund again will give us this response
```json
{"error":{"type":"invalid_request","message":"The refund amount exceeds the amount captured","code":"IR_13"}}
```
- What if the payment was failed and we try to refund it. (Invalid card number in confirm intent call)
```
curl --location 'http://localhost:8080/v2/payments/12345_pay_0195d142100474c3bcb32e40935e290d/confirm-intent' \
--header 'x-client-secret: 12345_pay_0195d142100474c3bcb32e40935e290d_secret_0195d14210127d30b524ec0e063b3c0a' \
--header 'x-profile-id: pro_TujgCYCsvWvO6YNsRAmC' \
--header 'Content-Type: application/json' \
--header 'Authorization: publishable-key=pk_dev_58559cdf90434adcb0b92eeb6a8b633f,client-secret=12345_pay_0195d142100474c3bcb32e40935e290d_secret_0195d14210127d30b524ec0e063b3c0a' \
--data '{
    "payment_method_data": {
        "card": {
            "card_number": "4242424242424242",
            "card_exp_month": "01",
            "card_exp_year": "25",
            "card_holder_name": "John Doe",
            "card_cvc": "100"
        }
    },
    "payment_method_type": "card",
    "payment_method_subtype": "credit"
}'
```
- Response from the above call
```json
{
  "id": "12345_pay_0195d142100474c3bcb32e40935e290d",
  "status": "failed",
  "amount": {
    "order_amount": 100,
    "currency": "USD",
    "shipping_cost": null,
    "order_tax_amount": null,
    "external_tax_calculation": "skip",
    "surcharge_calculation": "skip",
    "surcharge_amount": null,
    "tax_on_surcharge": null,
    "net_amount": 100,
    "amount_to_capture": null,
    "amount_capturable": 0,
    "amount_captured": null
  },
  "customer_id": null,
  "connector": "stripe",
  "client_secret": "12345_pay_0195d142100474c3bcb32e40935e290d_secret_0195d14210127d30b524ec0e063b3c0a",
  "created": "2025-03-26T07:02:02.514Z",
  "payment_method_data": {
    "billing": null
  },
  "payment_method_type": "card",
  "payment_method_subtype": "credit",
  "connector_transaction_id": null,
  "connector_reference_id": null,
  "merchant_connector_id": "mca_FhEKUEAUdgv7mVCs3A9U",
  "browser_info": null,
  "error": {
    "code": "invalid_expiry_month",
    "message": "invalid_expiry_month",
    "unified_code": null,
    "unified_message": null
  },
  "shipping": null,
  "billing": null,
  "attempts": null,
  "connector_token_details": null,
  "payment_method_id": null,
  "next_action": null,
  "return_url": "https://google.com/success",
  "authentication_type": "no_three_ds",
  "authentication_type_applied": "no_three_ds"
}
```

- Trying to refund failed payment
```
curl --location 'http://localhost:8080/v2/refunds' \
--header 'X-Profile-Id: pro_TujgCYCsvWvO6YNsRAmC' \
--header 'Content-Type: application/json' \
--header 'api-key: dev_kksGlKY14xNypCzsrJIXFCzL3t2j5kUs7ChmloPryJsdRy2RWYljonzPCxJSDmuZ' \
--data '{
    "payment_id":"12345_pay_0195d142100474c3bcb32e40935e290d",
    "merchant_reference_id":"1742972638",
    "reason":"Paid by mistake",
    "metadata":{
        "foo":"bar"
    }
}'
```
- Response from the above call
```json
{
  "error": {
    "type": "invalid_request",
    "message": "This Payment could not be refund because it has a status of failed. The expected state is succeeded, partially_captured",
    "code": "IR_14"
  }
}
```

Error case responses
- Connector NotImplemented Error (501)
```
{
    "error": {
        "type": "invalid_request",
        "message": "get_url method is not implemented",
        "code": "IR_00"
    }
}
``` 
- Connector NotSupported Error (400)
```
{
    "error": {
        "type": "invalid_request",
        "message": "Payment method type not supported",
        "code": "IR_19",
        "reason": "Refund is not supported by Stripe"
    }
}
```

## 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
