Bug: [FEATURE] Implement V2 UCS (Unified Connector Service) Integration for Authorization Flow 



### Feature Description

# [Feature Request] Implement V2 UCS (Unified Connector Service) Integration for Authorization Flow

## Issue Type
- [x] Feature Request
- [ ] Bug Report
- [ ] Enhancement
- [ ] Documentation

## Priority
- [x] High
- [ ] Medium 
- [ ] Low

## Labels
`A-core` `C-feature` `payment-flows` `ucs-integration` `v2-api`

## Summary

Implement V2 UCS (Unified Connector Service) integration for the authorization payment flow, upgrading from the existing V1 UCS integration to support the updated gRPC schema and enhanced V2 payment architecture.

## Background

Currently, Hyperswitch has V1 UCS integration that works with the legacy gRPC schema. However, the UCS service has evolved to V2 with improved:
- Enhanced gRPC schema with better type safety
- Improved field structures for card details, addresses, and identifiers
- Better error handling and response parsing
- Enhanced payment method data transformation

## Problem Statement

### Current State
- V1 UCS integration uses outdated `PaymentsAuthorizeRequest` schema
- Limited support for V2 payment flow architecture
- Outdated transformer implementations for V2 data structures
- Monolithic UCS function structure affecting maintainability

### Pain Points
1. **Schema Incompatibility**: V1 gRPC schema doesn't support V2 enhanced data models
2. **Limited Error Handling**: V1 response parsing lacks V2 identifier structure support
3. **Technical Debt**: Monolithic UCS functions make debugging and enhancement difficult
4. **Missing V2 Features**: Cannot leverage V2 payment flow improvements through UCS


### Possible Implementation

## Proposed Solution

### High-Level Approach
Implement V2 UCS integration that:
1. **Updates gRPC Schema**: Migrate to `PaymentServiceAuthorizeRequest` with V2 field structure
2. **Enhances Architecture**: Refactor payment flow with better separation of concerns
3. **Improves Transformers**: Update V2 transformers for new schema compatibility
4. **Maintains Backward Compatibility**: Preserve V1 integration with feature flag separation

### Technical Requirements

#### 1. gRPC Schema Migration
- [ ] Update `rust-grpc-client` dependency to latest version
- [ ] Migrate from `PaymentsAuthorizeRequest` to `PaymentServiceAuthorizeRequest`
- [ ] Update authentication headers and request metadata handling
- [ ] Implement new identifier structure for response parsing

#### 2. Payment Flow Architecture
- [ ] Refactor `call_connector_service()` into separate prerequisite and decision functions
- [ ] Implement `call_connector_service_prerequisites()` for V2 flow preparation
- [ ] Create `decide_unified_connector_service_call()` for UCS routing logic
- [ ] Maintain feature flag separation between V1 and V2 implementations

#### 3. V2 Transformer Enhancements
- [ ] Update card data transformation for new schema structure
- [ ] Enhance address handling with improved field mapping
- [ ] Implement customer ID support in V2 transformers

#### 4. Error Handling & Response Parsing
- [ ] Update response handlers for new identifier structure
- [ ] Enhance error context and fallback mechanisms
- [ ] Maintain automatic fallback to traditional connector flow

#### 5. Testing & Validation
- [ ] Create comprehensive test cases for V2 UCS flow
- [ ] Test schema compatibility and field mapping
- [ ] Validate fallback mechanisms and error handling



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

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

### Have you read the Contributing Guidelines?

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

### Are you willing to submit a PR?

Yes, I am willing to submit a PR!
diff --git a/Cargo.lock b/Cargo.lock
index 848ccc6648f..f3ab37bbb5b 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1651,6 +1651,20 @@ dependencies = [
  "thiserror 1.0.69",
 ]
 
+[[package]]
+name = "cargo_metadata"
+version = "0.19.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "dd5eb614ed4c27c5d706420e4320fbe3216ab31fa1c33cd8246ac36dae4479ba"
+dependencies = [
+ "camino",
+ "cargo-platform",
+ "semver 1.0.26",
+ "serde",
+ "serde_json",
+ "thiserror 2.0.12",
+]
+
 [[package]]
 name = "cast"
 version = "0.3.0"
@@ -3347,13 +3361,18 @@ dependencies = [
 
 [[package]]
 name = "g2h"
-version = "0.2.0"
+version = "0.4.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "312ad594dd0e3c26f860a52c8b703ab509c546931920f277f1afa9b7127fd755"
+checksum = "0aece561ff748cdd2a37c8ee938a47bbf9b2c03823b393a332110599b14ee827"
 dependencies = [
+ "cargo_metadata 0.19.2",
  "heck 0.5.0",
+ "proc-macro2",
+ "prost",
  "prost-build",
+ "prost-types",
  "quote",
+ "thiserror 2.0.12",
  "tonic-build",
 ]
 
@@ -3506,7 +3525,7 @@ dependencies = [
 [[package]]
 name = "grpc-api-types"
 version = "0.1.0"
-source = "git+https://github.com/juspay/connector-service?rev=afaef3427f815befe253da152e5f37d3858bbc9f#afaef3427f815befe253da152e5f37d3858bbc9f"
+source = "git+https://github.com/juspay/connector-service?rev=4918efedd5ea6c33e4a1600b988b2cf4948bed10#4918efedd5ea6c33e4a1600b988b2cf4948bed10"
 dependencies = [
  "axum 0.8.4",
  "error-stack 0.5.0",
@@ -6827,7 +6846,7 @@ dependencies = [
 name = "router_env"
 version = "0.1.0"
 dependencies = [
- "cargo_metadata",
+ "cargo_metadata 0.18.1",
  "config",
  "error-stack 0.4.1",
  "gethostname",
@@ -6880,7 +6899,7 @@ dependencies = [
 [[package]]
 name = "rust-grpc-client"
 version = "0.1.0"
-source = "git+https://github.com/juspay/connector-service?rev=afaef3427f815befe253da152e5f37d3858bbc9f#afaef3427f815befe253da152e5f37d3858bbc9f"
+source = "git+https://github.com/juspay/connector-service?rev=4918efedd5ea6c33e4a1600b988b2cf4948bed10#4918efedd5ea6c33e4a1600b988b2cf4948bed10"
 dependencies = [
  "grpc-api-types",
 ]
@@ -9501,7 +9520,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "2990d9ea5967266ea0ccf413a4aa5c42a93dbcfda9cb49a97de6931726b12566"
 dependencies = [
  "anyhow",
- "cargo_metadata",
+ "cargo_metadata 0.18.1",
  "cfg-if 1.0.0",
  "git2",
  "regex",
diff --git a/crates/external_services/Cargo.toml b/crates/external_services/Cargo.toml
index 906f8e9b596..30ee8ee5a2c 100644
--- a/crates/external_services/Cargo.toml
+++ b/crates/external_services/Cargo.toml
@@ -53,7 +53,7 @@ reqwest = { version = "0.11.27", features = ["rustls-tls"] }
 http = "0.2.12"
 url = { version = "2.5.4", features = ["serde"] }
 quick-xml = { version = "0.31.0", features = ["serialize"] }
-unified-connector-service-client = { git = "https://github.com/juspay/connector-service", rev = "afaef3427f815befe253da152e5f37d3858bbc9f", package = "rust-grpc-client" }
+unified-connector-service-client = { git = "https://github.com/juspay/connector-service", rev = "4918efedd5ea6c33e4a1600b988b2cf4948bed10", package = "rust-grpc-client" }
 
 
 # First party crates
diff --git a/crates/external_services/src/grpc_client/unified_connector_service.rs b/crates/external_services/src/grpc_client/unified_connector_service.rs
index 0ad9602985d..5a4a912b798 100644
--- a/crates/external_services/src/grpc_client/unified_connector_service.rs
+++ b/crates/external_services/src/grpc_client/unified_connector_service.rs
@@ -84,6 +84,10 @@ pub enum UnifiedConnectorServiceError {
     /// Failed to perform Payment Authorize from gRPC Server
     #[error("Failed to perform Payment Authorize from gRPC Server")]
     PaymentAuthorizeFailure,
+
+    /// Failed to perform Payment Get from gRPC Server
+    #[error("Failed to perform Payment Get from gRPC Server")]
+    PaymentGetFailure,
 }
 
 /// Result type for Dynamic Routing
@@ -191,6 +195,28 @@ impl UnifiedConnectorServiceClient {
             .change_context(UnifiedConnectorServiceError::PaymentAuthorizeFailure)
             .inspect_err(|error| logger::error!(?error))
     }
+
+    /// Performs Payment Sync/Get
+    pub async fn payment_get(
+        &self,
+        payment_get_request: payments_grpc::PaymentServiceGetRequest,
+        connector_auth_metadata: ConnectorAuthMetadata,
+        grpc_headers: GrpcHeaders,
+    ) -> UnifiedConnectorServiceResult<tonic::Response<payments_grpc::PaymentServiceGetResponse>>
+    {
+        let mut request = tonic::Request::new(payment_get_request);
+
+        let metadata =
+            build_unified_connector_service_grpc_headers(connector_auth_metadata, grpc_headers)?;
+        *request.metadata_mut() = metadata;
+
+        self.client
+            .clone()
+            .get(request)
+            .await
+            .change_context(UnifiedConnectorServiceError::PaymentGetFailure)
+            .inspect_err(|error| logger::error!(?error))
+    }
 }
 
 /// Build the gRPC Headers for Unified Connector Service Request
diff --git a/crates/hyperswitch_domain_models/src/merchant_connector_account.rs b/crates/hyperswitch_domain_models/src/merchant_connector_account.rs
index 92ec2f51de2..1e460ee7306 100644
--- a/crates/hyperswitch_domain_models/src/merchant_connector_account.rs
+++ b/crates/hyperswitch_domain_models/src/merchant_connector_account.rs
@@ -163,7 +163,9 @@ impl MerchantConnectorAccountTypeDetails {
             Self::MerchantConnectorAccount(merchant_connector_account) => {
                 Some(merchant_connector_account.connector_name)
             }
-            Self::MerchantConnectorDetails(_) => None,
+            Self::MerchantConnectorDetails(merchant_connector_details) => {
+                Some(merchant_connector_details.connector_name)
+            }
         }
     }
 
diff --git a/crates/router/Cargo.toml b/crates/router/Cargo.toml
index 585c49ffe18..8b80d22dc2a 100644
--- a/crates/router/Cargo.toml
+++ b/crates/router/Cargo.toml
@@ -88,7 +88,7 @@ reqwest = { version = "0.11.27", features = ["json", "rustls-tls", "gzip", "mult
 ring = "0.17.14"
 rust_decimal = { version = "1.37.1", features = ["serde-with-float", "serde-with-str"] }
 rust-i18n = { git = "https://github.com/kashif-m/rust-i18n", rev = "f2d8096aaaff7a87a847c35a5394c269f75e077a" }
-unified-connector-service-client = { git = "https://github.com/juspay/connector-service", rev = "afaef3427f815befe253da152e5f37d3858bbc9f", package = "rust-grpc-client" }
+unified-connector-service-client = { git = "https://github.com/juspay/connector-service", rev = "4918efedd5ea6c33e4a1600b988b2cf4948bed10", package = "rust-grpc-client" }
 rustc-hash = "1.1.0"
 rustls = "0.22"
 rustls-pemfile = "2"
diff --git a/crates/router/src/core/payments.rs b/crates/router/src/core/payments.rs
index 8a0058b79e4..5319e823614 100644
--- a/crates/router/src/core/payments.rs
+++ b/crates/router/src/core/payments.rs
@@ -216,7 +216,27 @@ where
 
     let (payment_data, connector_response_data) = match connector {
         ConnectorCallType::PreDetermined(connector_data) => {
-            let router_data = call_connector_service(
+            let (mca_type_details, updated_customer, router_data) =
+                call_connector_service_prerequisites(
+                    state,
+                    req_state.clone(),
+                    &merchant_context,
+                    connector_data.connector_data.clone(),
+                    &operation,
+                    &mut payment_data,
+                    &customer,
+                    call_connector_action.clone(),
+                    None,
+                    header_payload.clone(),
+                    None,
+                    profile,
+                    false,
+                    false, //should_retry_with_pan is set to false in case of PreDetermined ConnectorCallType
+                    req.should_return_raw_response(),
+                )
+                .await?;
+
+            let router_data = decide_unified_connector_service_call(
                 state,
                 req_state.clone(),
                 &merchant_context,
@@ -225,16 +245,17 @@ where
                 &mut payment_data,
                 &customer,
                 call_connector_action.clone(),
-                None,
+                None, // schedule_time is not used in PreDetermined ConnectorCallType
                 header_payload.clone(),
                 #[cfg(feature = "frm")]
                 None,
-                #[cfg(not(feature = "frm"))]
-                None,
                 profile,
                 false,
                 false, //should_retry_with_pan is set to false in case of PreDetermined ConnectorCallType
                 req.should_return_raw_response(),
+                mca_type_details,
+                router_data,
+                updated_customer,
             )
             .await?;
 
@@ -271,7 +292,28 @@ where
         ConnectorCallType::Retryable(connectors) => {
             let mut connectors = connectors.clone().into_iter();
             let connector_data = get_connector_data(&mut connectors)?;
-            let router_data = call_connector_service(
+
+            let (mca_type_details, updated_customer, router_data) =
+                call_connector_service_prerequisites(
+                    state,
+                    req_state.clone(),
+                    &merchant_context,
+                    connector_data.connector_data.clone(),
+                    &operation,
+                    &mut payment_data,
+                    &customer,
+                    call_connector_action.clone(),
+                    None,
+                    header_payload.clone(),
+                    None,
+                    profile,
+                    false,
+                    false, //should_retry_with_pan is set to false in case of Retryable ConnectorCallType
+                    req.should_return_raw_response(),
+                )
+                .await?;
+
+            let router_data = decide_unified_connector_service_call(
                 state,
                 req_state.clone(),
                 &merchant_context,
@@ -280,16 +322,17 @@ where
                 &mut payment_data,
                 &customer,
                 call_connector_action.clone(),
-                None,
+                None, // schedule_time is not used in Retryable ConnectorCallType
                 header_payload.clone(),
                 #[cfg(feature = "frm")]
                 None,
-                #[cfg(not(feature = "frm"))]
-                None,
                 profile,
-                false,
+                true,
                 false, //should_retry_with_pan is set to false in case of PreDetermined ConnectorCallType
                 req.should_return_raw_response(),
+                mca_type_details,
+                router_data,
+                updated_customer,
             )
             .await?;
 
@@ -409,17 +452,37 @@ where
         .get_connector_from_request(state, &req, &mut payment_data)
         .await?;
 
-    let router_data = internal_call_connector_service(
+    let merchant_connector_account = payment_data
+        .get_merchant_connector_details()
+        .map(domain::MerchantConnectorAccountTypeDetails::MerchantConnectorDetails)
+        .ok_or_else(|| {
+            error_stack::report!(errors::ApiErrorResponse::InternalServerError)
+                .attach_printable("Merchant connector details not found in payment data")
+        })?;
+
+    operation
+        .to_domain()?
+        .populate_payment_data(
+            state,
+            &mut payment_data,
+            &merchant_context,
+            profile,
+            &connector_data,
+        )
+        .await?;
+
+    let router_data = connector_service_decider(
         state,
         req_state.clone(),
         &merchant_context,
-        connector_data,
+        connector_data.clone(),
         &operation,
         &mut payment_data,
         call_connector_action.clone(),
         header_payload.clone(),
         profile,
         req.should_return_raw_response(),
+        merchant_connector_account,
     )
     .await?;
 
@@ -3508,27 +3571,29 @@ where
 
     router_data = router_data.add_session_token(state, &connector).await?;
 
-    let mut should_continue_further = access_token::update_router_data_with_access_token_result(
+    let should_continue_further = access_token::update_router_data_with_access_token_result(
         &add_access_token_result,
         &mut router_data,
         &call_connector_action,
     );
 
-    let add_create_order_result = router_data
+    let should_continue_further = match router_data
         .create_order_at_connector(state, &connector, should_continue_further)
-        .await?;
+        .await?
+    {
+        Some(create_order_response) => {
+            if let Ok(order_id) = create_order_response.clone().create_order_result {
+                payment_data.set_connector_response_reference_id(Some(order_id.clone()))
+            }
 
-    if add_create_order_result.is_create_order_performed {
-        if let Ok(order_id_opt) = &add_create_order_result.create_order_result {
-            payment_data.set_connector_response_reference_id(order_id_opt.clone());
+            // Set the response in routerdata response to carry forward
+            router_data
+                .update_router_data_with_create_order_response(create_order_response.clone());
+            create_order_response.create_order_result.ok().is_some()
         }
-        should_continue_further = router_data
-            .update_router_data_with_create_order_result(
-                add_create_order_result,
-                should_continue_further,
-            )
-            .await?;
-    }
+        // If create order is not required, then we can proceed with further processing
+        None => true,
+    };
 
     let updated_customer = call_create_connector_customer_if_required(
         state,
@@ -3948,6 +4013,9 @@ pub async fn call_connector_service<F, RouterDReq, ApiRequest, D>(
     is_retry_payment: bool,
     should_retry_with_pan: bool,
     return_raw_connector_response: Option<bool>,
+    merchant_connector_account_type_details: domain::MerchantConnectorAccountTypeDetails,
+    mut router_data: RouterData<F, RouterDReq, router_types::PaymentsResponseData>,
+    updated_customer: Option<storage::CustomerUpdate>,
 ) -> RouterResult<RouterData<F, RouterDReq, router_types::PaymentsResponseData>>
 where
     F: Send + Clone + Sync,
@@ -3961,50 +4029,6 @@ where
     dyn api::Connector:
         services::api::ConnectorIntegration<F, RouterDReq, router_types::PaymentsResponseData>,
 {
-    let stime_connector = Instant::now();
-
-    let merchant_connector_account =
-        domain::MerchantConnectorAccountTypeDetails::MerchantConnectorAccount(Box::new(
-            helpers::get_merchant_connector_account_v2(
-                state,
-                merchant_context.get_merchant_key_store(),
-                connector.merchant_connector_id.as_ref(),
-            )
-            .await?,
-        ));
-
-    operation
-        .to_domain()?
-        .populate_payment_data(
-            state,
-            payment_data,
-            merchant_context,
-            business_profile,
-            &connector,
-        )
-        .await?;
-
-    let updated_customer = call_create_connector_customer_if_required(
-        state,
-        customer,
-        merchant_context,
-        &merchant_connector_account,
-        payment_data,
-    )
-    .await?;
-
-    let mut router_data = payment_data
-        .construct_router_data(
-            state,
-            connector.connector.id(),
-            merchant_context,
-            customer,
-            &merchant_connector_account,
-            None,
-            None,
-        )
-        .await?;
-
     let add_access_token_result = router_data
         .add_access_token(
             state,
@@ -4016,37 +4040,43 @@ where
 
     router_data = router_data.add_session_token(state, &connector).await?;
 
-    let mut should_continue_further = access_token::update_router_data_with_access_token_result(
+    let should_continue_further = access_token::update_router_data_with_access_token_result(
         &add_access_token_result,
         &mut router_data,
         &call_connector_action,
     );
 
-    let add_create_order_result = router_data
+    let should_continue = match router_data
         .create_order_at_connector(state, &connector, should_continue_further)
-        .await?;
+        .await?
+    {
+        Some(create_order_response) => {
+            if let Ok(order_id) = create_order_response.clone().create_order_result {
+                payment_data.set_connector_response_reference_id(Some(order_id))
+            }
 
-    if add_create_order_result.is_create_order_performed {
-        if let Ok(order_id_opt) = &add_create_order_result.create_order_result {
-            payment_data.set_connector_response_reference_id(order_id_opt.clone());
+            // Set the response in routerdata response to carry forward
+            router_data
+                .update_router_data_with_create_order_response(create_order_response.clone());
+            create_order_response.create_order_result.ok().map(|_| ())
         }
-        should_continue_further = router_data
-            .update_router_data_with_create_order_result(
-                add_create_order_result,
-                should_continue_further,
-            )
-            .await?;
-    }
+        // If create order is not required, then we can proceed with further processing
+        None => Some(()),
+    };
 
     // In case of authorize flow, pre-task and post-tasks are being called in build request
     // if we do not want to proceed further, then the function will return Ok(None, false)
-    let (connector_request, should_continue_further) = if should_continue_further {
-        // Check if the actual flow specific request can be built with available data
-        router_data
-            .build_flow_specific_connector_request(state, &connector, call_connector_action.clone())
-            .await?
-    } else {
-        (None, false)
+    let (connector_request, should_continue_further) = match should_continue {
+        Some(_) => {
+            router_data
+                .build_flow_specific_connector_request(
+                    state,
+                    &connector,
+                    call_connector_action.clone(),
+                )
+                .await?
+        }
+        None => (None, false),
     };
 
     // Update the payment trackers just before calling the connector
@@ -4089,18 +4119,14 @@ where
         Ok(router_data)
     }?;
 
-    let etime_connector = Instant::now();
-    let duration_connector = etime_connector.saturating_duration_since(stime_connector);
-    tracing::info!(duration = format!("Duration taken: {}", duration_connector.as_millis()));
-
     Ok(router_data)
 }
 
-#[cfg(feature = "v1")]
-// This function does not perform the tokenization action, as the payment method is not saved in this flow.
+#[cfg(feature = "v2")]
 #[allow(clippy::too_many_arguments)]
+#[allow(clippy::type_complexity)]
 #[instrument(skip_all)]
-pub async fn proxy_for_call_connector_service<F, RouterDReq, ApiRequest, D>(
+pub async fn call_connector_service_prerequisites<F, RouterDReq, ApiRequest, D>(
     state: &SessionState,
     req_state: ReqState,
     merchant_context: &domain::MerchantContext,
@@ -4109,15 +4135,17 @@ pub async fn proxy_for_call_connector_service<F, RouterDReq, ApiRequest, D>(
     payment_data: &mut D,
     customer: &Option<domain::Customer>,
     call_connector_action: CallConnectorAction,
-    validate_result: &operations::ValidateResult,
     schedule_time: Option<time::PrimitiveDateTime>,
     header_payload: HeaderPayload,
-
+    frm_suggestion: Option<storage_enums::FrmSuggestion>,
     business_profile: &domain::Profile,
-    return_raw_connector_response: Option<bool>,
+    is_retry_payment: bool,
+    should_retry_with_pan: bool,
+    all_keys_required: Option<bool>,
 ) -> RouterResult<(
+    domain::MerchantConnectorAccountTypeDetails,
+    Option<storage::CustomerUpdate>,
     RouterData<F, RouterDReq, router_types::PaymentsResponseData>,
-    helpers::MerchantConnectorAccountType,
 )>
 where
     F: Send + Clone + Sync,
@@ -4131,145 +4159,354 @@ where
     dyn api::Connector:
         services::api::ConnectorIntegration<F, RouterDReq, router_types::PaymentsResponseData>,
 {
-    let stime_connector = Instant::now();
+    let merchant_connector_account_type_details =
+        domain::MerchantConnectorAccountTypeDetails::MerchantConnectorAccount(Box::new(
+            helpers::get_merchant_connector_account_v2(
+                state,
+                merchant_context.get_merchant_key_store(),
+                connector.merchant_connector_id.as_ref(),
+            )
+            .await?,
+        ));
 
-    let merchant_connector_account = construct_profile_id_and_get_mca(
+    operation
+        .to_domain()?
+        .populate_payment_data(
+            state,
+            payment_data,
+            merchant_context,
+            business_profile,
+            &connector,
+        )
+        .await?;
+
+    let updated_customer = call_create_connector_customer_if_required(
         state,
+        customer,
         merchant_context,
+        &merchant_connector_account_type_details,
         payment_data,
-        &connector.connector_name.to_string(),
-        connector.merchant_connector_id.as_ref(),
-        false,
     )
     .await?;
 
-    if payment_data
-        .get_payment_attempt()
-        .merchant_connector_id
-        .is_none()
-    {
-        payment_data.set_merchant_connector_id_in_attempt(merchant_connector_account.get_mca_id());
-    }
-
-    let merchant_recipient_data = None;
-
-    let mut router_data = payment_data
+    let router_data = payment_data
         .construct_router_data(
             state,
             connector.connector.id(),
             merchant_context,
             customer,
-            &merchant_connector_account,
-            merchant_recipient_data,
+            &merchant_connector_account_type_details,
+            None,
             None,
         )
         .await?;
 
-    let add_access_token_result = router_data
-        .add_access_token(
+    Ok((
+        merchant_connector_account_type_details,
+        updated_customer,
+        router_data,
+    ))
+}
+
+#[cfg(feature = "v2")]
+#[instrument(skip_all)]
+pub async fn internal_call_connector_service_prerequisites<F, RouterDReq, ApiRequest, D>(
+    state: &SessionState,
+    merchant_context: &domain::MerchantContext,
+    connector: api::ConnectorData,
+    operation: &BoxedOperation<'_, F, ApiRequest, D>,
+    payment_data: &mut D,
+    business_profile: &domain::Profile,
+) -> RouterResult<(
+    domain::MerchantConnectorAccountTypeDetails,
+    RouterData<F, RouterDReq, router_types::PaymentsResponseData>,
+)>
+where
+    F: Send + Clone + Sync,
+    RouterDReq: Send + Sync,
+
+    // To create connector flow specific interface data
+    D: OperationSessionGetters<F> + OperationSessionSetters<F> + Send + Sync + Clone,
+    D: ConstructFlowSpecificData<F, RouterDReq, router_types::PaymentsResponseData>,
+    RouterData<F, RouterDReq, router_types::PaymentsResponseData>: Feature<F, RouterDReq> + Send,
+    // To construct connector flow specific api
+    dyn api::Connector:
+        services::api::ConnectorIntegration<F, RouterDReq, router_types::PaymentsResponseData>,
+{
+    let merchant_connector_details =
+        payment_data
+            .get_merchant_connector_details()
+            .ok_or_else(|| {
+                error_stack::report!(errors::ApiErrorResponse::InternalServerError)
+                    .attach_printable("Merchant connector details not found in payment data")
+            })?;
+    let merchant_connector_account =
+        domain::MerchantConnectorAccountTypeDetails::MerchantConnectorDetails(
+            merchant_connector_details,
+        );
+
+    operation
+        .to_domain()?
+        .populate_payment_data(
             state,
-            &connector,
+            payment_data,
             merchant_context,
-            payment_data.get_creds_identifier(),
+            business_profile,
+            &connector,
         )
         .await?;
 
-    router_data = router_data.add_session_token(state, &connector).await?;
+    let router_data = payment_data
+        .construct_router_data(
+            state,
+            connector.connector.id(),
+            merchant_context,
+            &None,
+            &merchant_connector_account,
+            None,
+            None,
+        )
+        .await?;
 
-    let mut should_continue_further = access_token::update_router_data_with_access_token_result(
-        &add_access_token_result,
-        &mut router_data,
-        &call_connector_action,
-    );
+    Ok((merchant_connector_account, router_data))
+}
 
-    (router_data, should_continue_further) = complete_preprocessing_steps_if_required(
-        state,
-        &connector,
-        payment_data,
-        router_data,
-        operation,
-        should_continue_further,
-    )
-    .await?;
+#[cfg(feature = "v2")]
+#[allow(clippy::too_many_arguments)]
+#[instrument(skip_all)]
+pub async fn connector_service_decider<F, RouterDReq, ApiRequest, D>(
+    state: &SessionState,
+    req_state: ReqState,
+    merchant_context: &domain::MerchantContext,
+    connector: api::ConnectorData,
+    operation: &BoxedOperation<'_, F, ApiRequest, D>,
+    payment_data: &mut D,
+    call_connector_action: CallConnectorAction,
+    header_payload: HeaderPayload,
+    business_profile: &domain::Profile,
+    return_raw_connector_response: Option<bool>,
+    merchant_connector_account_type_details: domain::MerchantConnectorAccountTypeDetails,
+) -> RouterResult<RouterData<F, RouterDReq, router_types::PaymentsResponseData>>
+where
+    F: Send + Clone + Sync,
+    RouterDReq: Send + Sync,
 
-    if let Ok(router_types::PaymentsResponseData::PreProcessingResponse {
-        session_token: Some(session_token),
-        ..
-    }) = router_data.response.to_owned()
-    {
-        payment_data.push_sessions_token(session_token);
-    };
+    // To create connector flow specific interface data
+    D: OperationSessionGetters<F> + OperationSessionSetters<F> + Send + Sync + Clone,
+    D: ConstructFlowSpecificData<F, RouterDReq, router_types::PaymentsResponseData>,
+    RouterData<F, RouterDReq, router_types::PaymentsResponseData>: Feature<F, RouterDReq> + Send,
+    // To construct connector flow specific api
+    dyn api::Connector:
+        services::api::ConnectorIntegration<F, RouterDReq, router_types::PaymentsResponseData>,
+{
+    let mut router_data = payment_data
+        .construct_router_data(
+            state,
+            connector.connector.id(),
+            merchant_context,
+            &None,
+            &merchant_connector_account_type_details,
+            None,
+            None,
+        )
+        .await?;
 
-    let (connector_request, should_continue_further) = if should_continue_further {
-        // Check if the actual flow specific request can be built with available data
-        router_data
-            .build_flow_specific_connector_request(state, &connector, call_connector_action.clone())
+    // do order creation
+    let should_call_unified_connector_service =
+        should_call_unified_connector_service(state, merchant_context, &router_data).await?;
+
+    let (connector_request, should_continue_further) = if !should_call_unified_connector_service {
+        let mut should_continue_further = true;
+
+        let should_continue = match router_data
+            .create_order_at_connector(state, &connector, should_continue_further)
             .await?
+        {
+            Some(create_order_response) => {
+                if let Ok(order_id) = create_order_response.clone().create_order_result {
+                    payment_data.set_connector_response_reference_id(Some(order_id))
+                }
+
+                // Set the response in routerdata response to carry forward
+                router_data
+                    .update_router_data_with_create_order_response(create_order_response.clone());
+                create_order_response.create_order_result.ok().map(|_| ())
+            }
+            // If create order is not required, then we can proceed with further processing
+            None => Some(()),
+        };
+
+        let should_continue: (Option<common_utils::request::Request>, bool) = match should_continue
+        {
+            Some(_) => {
+                router_data
+                    .build_flow_specific_connector_request(
+                        state,
+                        &connector,
+                        call_connector_action.clone(),
+                    )
+                    .await?
+            }
+            None => (None, false),
+        };
+        should_continue
     } else {
+        // If unified connector service is called, these values are not used
+        // as the request is built in the unified connector service call
         (None, false)
     };
 
-    if should_add_task_to_process_tracker(payment_data) {
-        operation
-            .to_domain()?
-            .add_task_to_process_tracker(
-                state,
-                payment_data.get_payment_attempt(),
-                validate_result.requeue,
-                schedule_time,
-            )
-            .await
-            .map_err(|error| logger::error!(process_tracker_error=?error))
-            .ok();
-    }
-
-    let updated_customer = None;
-    let frm_suggestion = None;
-
     (_, *payment_data) = operation
         .to_update_tracker()?
         .update_trackers(
             state,
             req_state,
             payment_data.clone(),
-            customer.clone(),
+            None, // customer is not used in internal flows
             merchant_context.get_merchant_account().storage_scheme,
-            updated_customer,
+            None,
             merchant_context.get_merchant_key_store(),
-            frm_suggestion,
+            None, // frm_suggestion is not used in internal flows
             header_payload.clone(),
         )
         .await?;
 
-    let router_data = if should_continue_further {
-        // The status of payment_attempt and intent will be updated in the previous step
-        // update this in router_data.
-        // This is added because few connector integrations do not update the status,
-        // and rely on previous status set in router_data
-        router_data.status = payment_data.get_payment_attempt().status;
-        router_data
-            .decide_flows(
+    record_time_taken_with(|| async {
+        if should_call_unified_connector_service {
+            router_data
+                .call_unified_connector_service(
+                    state,
+                    merchant_connector_account_type_details.clone(),
+                    merchant_context,
+                )
+                .await?;
+
+            Ok(router_data)
+        } else {
+            let router_data = if should_continue_further {
+                router_data
+                    .decide_flows(
+                        state,
+                        &connector,
+                        call_connector_action,
+                        connector_request,
+                        business_profile,
+                        header_payload.clone(),
+                        return_raw_connector_response,
+                    )
+                    .await
+            } else {
+                Ok(router_data)
+            }?;
+            Ok(router_data)
+        }
+    })
+    .await
+}
+
+#[cfg(feature = "v2")]
+#[allow(clippy::too_many_arguments)]
+#[instrument(skip_all)]
+pub async fn decide_unified_connector_service_call<F, RouterDReq, ApiRequest, D>(
+    state: &SessionState,
+    req_state: ReqState,
+    merchant_context: &domain::MerchantContext,
+    connector: api::ConnectorData,
+    operation: &BoxedOperation<'_, F, ApiRequest, D>,
+    payment_data: &mut D,
+    customer: &Option<domain::Customer>,
+    call_connector_action: CallConnectorAction,
+    schedule_time: Option<time::PrimitiveDateTime>,
+    header_payload: HeaderPayload,
+    frm_suggestion: Option<storage_enums::FrmSuggestion>,
+    business_profile: &domain::Profile,
+    is_retry_payment: bool,
+    should_retry_with_pan: bool,
+    return_raw_connector_response: Option<bool>,
+    merchant_connector_account_type_details: domain::MerchantConnectorAccountTypeDetails,
+    mut router_data: RouterData<F, RouterDReq, router_types::PaymentsResponseData>,
+    updated_customer: Option<storage::CustomerUpdate>,
+) -> RouterResult<RouterData<F, RouterDReq, router_types::PaymentsResponseData>>
+where
+    F: Send + Clone + Sync,
+    RouterDReq: Send + Sync,
+
+    // To create connector flow specific interface data
+    D: OperationSessionGetters<F> + OperationSessionSetters<F> + Send + Sync + Clone,
+    D: ConstructFlowSpecificData<F, RouterDReq, router_types::PaymentsResponseData>,
+    RouterData<F, RouterDReq, router_types::PaymentsResponseData>: Feature<F, RouterDReq> + Send,
+    // To construct connector flow specific api
+    dyn api::Connector:
+        services::api::ConnectorIntegration<F, RouterDReq, router_types::PaymentsResponseData>,
+{
+    record_time_taken_with(|| async {
+        if should_call_unified_connector_service(state, merchant_context, &router_data).await? {
+            if should_add_task_to_process_tracker(payment_data) {
+                operation
+                    .to_domain()?
+                    .add_task_to_process_tracker(
+                        state,
+                        payment_data.get_payment_attempt(),
+                        false,
+                        schedule_time,
+                    )
+                    .await
+                    .map_err(|error| logger::error!(process_tracker_error=?error))
+                    .ok();
+            }
+
+            (_, *payment_data) = operation
+                .to_update_tracker()?
+                .update_trackers(
+                    state,
+                    req_state,
+                    payment_data.clone(),
+                    customer.clone(),
+                    merchant_context.get_merchant_account().storage_scheme,
+                    None,
+                    merchant_context.get_merchant_key_store(),
+                    frm_suggestion,
+                    header_payload.clone(),
+                )
+                .await?;
+
+            router_data
+                .call_unified_connector_service(
+                    state,
+                    merchant_connector_account_type_details.clone(),
+                    merchant_context,
+                )
+                .await?;
+
+            Ok(router_data)
+        } else {
+            call_connector_service(
                 state,
-                &connector,
+                req_state,
+                merchant_context,
+                connector,
+                operation,
+                payment_data,
+                customer,
                 call_connector_action,
-                connector_request,
+                schedule_time,
+                header_payload,
+                frm_suggestion,
                 business_profile,
-                header_payload.clone(),
+                is_retry_payment,
+                should_retry_with_pan,
                 return_raw_connector_response,
+                merchant_connector_account_type_details,
+                router_data,
+                updated_customer,
             )
             .await
-    } else {
-        Ok(router_data)
-    }?;
-
-    let etime_connector = Instant::now();
-    let duration_connector = etime_connector.saturating_duration_since(stime_connector);
-    tracing::info!(duration = format!("Duration taken: {}", duration_connector.as_millis()));
-
-    Ok((router_data, merchant_connector_account))
+        }
+    })
+    .await
 }
-
-#[cfg(feature = "v2")]
+#[cfg(feature = "v1")]
+// This function does not perform the tokenization action, as the payment method is not saved in this flow.
 #[allow(clippy::too_many_arguments)]
 #[instrument(skip_all)]
 pub async fn proxy_for_call_connector_service<F, RouterDReq, ApiRequest, D>(
@@ -4279,11 +4516,18 @@ pub async fn proxy_for_call_connector_service<F, RouterDReq, ApiRequest, D>(
     connector: api::ConnectorData,
     operation: &BoxedOperation<'_, F, ApiRequest, D>,
     payment_data: &mut D,
+    customer: &Option<domain::Customer>,
     call_connector_action: CallConnectorAction,
+    validate_result: &operations::ValidateResult,
+    schedule_time: Option<time::PrimitiveDateTime>,
     header_payload: HeaderPayload,
+
     business_profile: &domain::Profile,
     return_raw_connector_response: Option<bool>,
-) -> RouterResult<RouterData<F, RouterDReq, router_types::PaymentsResponseData>>
+) -> RouterResult<(
+    RouterData<F, RouterDReq, router_types::PaymentsResponseData>,
+    helpers::MerchantConnectorAccountType,
+)>
 where
     F: Send + Clone + Sync,
     RouterDReq: Send + Sync,
@@ -4298,34 +4542,34 @@ where
 {
     let stime_connector = Instant::now();
 
-    let merchant_connector_account =
-        domain::MerchantConnectorAccountTypeDetails::MerchantConnectorAccount(Box::new(
-            helpers::get_merchant_connector_account_v2(
-                state,
-                merchant_context.get_merchant_key_store(),
-                connector.merchant_connector_id.as_ref(),
-            )
-            .await?,
-        ));
-    operation
-        .to_domain()?
-        .populate_payment_data(
-            state,
-            payment_data,
-            merchant_context,
-            business_profile,
-            &connector,
-        )
-        .await?;
+    let merchant_connector_account = construct_profile_id_and_get_mca(
+        state,
+        merchant_context,
+        payment_data,
+        &connector.connector_name.to_string(),
+        connector.merchant_connector_id.as_ref(),
+        false,
+    )
+    .await?;
+
+    if payment_data
+        .get_payment_attempt()
+        .merchant_connector_id
+        .is_none()
+    {
+        payment_data.set_merchant_connector_id_in_attempt(merchant_connector_account.get_mca_id());
+    }
+
+    let merchant_recipient_data = None;
 
     let mut router_data = payment_data
         .construct_router_data(
             state,
             connector.connector.id(),
             merchant_context,
-            &None,
+            customer,
             &merchant_connector_account,
-            None,
+            merchant_recipient_data,
             None,
         )
         .await?;
@@ -4347,7 +4591,26 @@ where
         &call_connector_action,
     );
 
+    (router_data, should_continue_further) = complete_preprocessing_steps_if_required(
+        state,
+        &connector,
+        payment_data,
+        router_data,
+        operation,
+        should_continue_further,
+    )
+    .await?;
+
+    if let Ok(router_types::PaymentsResponseData::PreProcessingResponse {
+        session_token: Some(session_token),
+        ..
+    }) = router_data.response.to_owned()
+    {
+        payment_data.push_sessions_token(session_token);
+    };
+
     let (connector_request, should_continue_further) = if should_continue_further {
+        // Check if the actual flow specific request can be built with available data
         router_data
             .build_flow_specific_connector_request(state, &connector, call_connector_action.clone())
             .await?
@@ -4355,22 +4618,44 @@ where
         (None, false)
     };
 
+    if should_add_task_to_process_tracker(payment_data) {
+        operation
+            .to_domain()?
+            .add_task_to_process_tracker(
+                state,
+                payment_data.get_payment_attempt(),
+                validate_result.requeue,
+                schedule_time,
+            )
+            .await
+            .map_err(|error| logger::error!(process_tracker_error=?error))
+            .ok();
+    }
+
+    let updated_customer = None;
+    let frm_suggestion = None;
+
     (_, *payment_data) = operation
         .to_update_tracker()?
         .update_trackers(
             state,
             req_state,
             payment_data.clone(),
-            None,
+            customer.clone(),
             merchant_context.get_merchant_account().storage_scheme,
-            None,
+            updated_customer,
             merchant_context.get_merchant_key_store(),
-            None,
+            frm_suggestion,
             header_payload.clone(),
         )
         .await?;
 
     let router_data = if should_continue_further {
+        // The status of payment_attempt and intent will be updated in the previous step
+        // update this in router_data.
+        // This is added because few connector integrations do not update the status,
+        // and rely on previous status set in router_data
+        router_data.status = payment_data.get_payment_attempt().status;
         router_data
             .decide_flows(
                 state,
@@ -4390,13 +4675,13 @@ where
     let duration_connector = etime_connector.saturating_duration_since(stime_connector);
     tracing::info!(duration = format!("Duration taken: {}", duration_connector.as_millis()));
 
-    Ok(router_data)
+    Ok((router_data, merchant_connector_account))
 }
 
 #[cfg(feature = "v2")]
 #[allow(clippy::too_many_arguments)]
 #[instrument(skip_all)]
-pub async fn internal_call_connector_service<F, RouterDReq, ApiRequest, D>(
+pub async fn proxy_for_call_connector_service<F, RouterDReq, ApiRequest, D>(
     state: &SessionState,
     req_state: ReqState,
     merchant_context: &domain::MerchantContext,
@@ -4422,18 +4707,15 @@ where
 {
     let stime_connector = Instant::now();
 
-    let merchant_connector_details =
-        payment_data
-            .get_merchant_connector_details()
-            .ok_or_else(|| {
-                error_stack::report!(errors::ApiErrorResponse::InternalServerError)
-                    .attach_printable("Merchant connector details not found in payment data")
-            })?;
     let merchant_connector_account =
-        domain::MerchantConnectorAccountTypeDetails::MerchantConnectorDetails(
-            merchant_connector_details,
-        );
-
+        domain::MerchantConnectorAccountTypeDetails::MerchantConnectorAccount(Box::new(
+            helpers::get_merchant_connector_account_v2(
+                state,
+                merchant_context.get_merchant_key_store(),
+                connector.merchant_connector_id.as_ref(),
+            )
+            .await?,
+        ));
     operation
         .to_domain()?
         .populate_payment_data(
@@ -4474,22 +4756,6 @@ where
         &call_connector_action,
     );
 
-    let add_create_order_result = router_data
-        .create_order_at_connector(state, &connector, should_continue_further)
-        .await?;
-
-    if add_create_order_result.is_create_order_performed {
-        if let Ok(order_id_opt) = &add_create_order_result.create_order_result {
-            payment_data.set_connector_response_reference_id(order_id_opt.clone());
-        }
-        should_continue_further = router_data
-            .update_router_data_with_create_order_result(
-                add_create_order_result,
-                should_continue_further,
-            )
-            .await?;
-    }
-
     let (connector_request, should_continue_further) = if should_continue_further {
         router_data
             .build_flow_specific_connector_request(state, &connector, call_connector_action.clone())
diff --git a/crates/router/src/core/payments/flows.rs b/crates/router/src/core/payments/flows.rs
index a93e2f391fe..2cae2f0600d 100644
--- a/crates/router/src/core/payments/flows.rs
+++ b/crates/router/src/core/payments/flows.rs
@@ -185,35 +185,27 @@ pub trait Feature<F, T> {
         _state: &SessionState,
         _connector: &api::ConnectorData,
         _should_continue_payment: bool,
-    ) -> RouterResult<types::CreateOrderResult>
+    ) -> RouterResult<Option<types::CreateOrderResult>>
     where
         F: Clone,
         Self: Sized,
         dyn api::Connector: services::ConnectorIntegration<F, T, types::PaymentsResponseData>,
     {
-        Ok(types::CreateOrderResult {
-            create_order_result: Ok(None),
-            is_create_order_performed: false,
-        })
+        Ok(None)
     }
 
-    async fn update_router_data_with_create_order_result(
+    fn update_router_data_with_create_order_response(
         &mut self,
         _create_order_result: types::CreateOrderResult,
-        should_continue_payment: bool,
-    ) -> RouterResult<bool>
-    where
-        F: Clone,
-        Self: Sized,
-        dyn api::Connector: services::ConnectorIntegration<F, T, types::PaymentsResponseData>,
-    {
-        Ok(should_continue_payment)
+    ) {
     }
 
     async fn call_unified_connector_service<'a>(
         &mut self,
         _state: &SessionState,
-        _merchant_connector_account: helpers::MerchantConnectorAccountType,
+        #[cfg(feature = "v1")] _merchant_connector_account: helpers::MerchantConnectorAccountType,
+        #[cfg(feature = "v2")]
+        _merchant_connector_account: domain::MerchantConnectorAccountTypeDetails,
         _merchant_context: &domain::MerchantContext,
     ) -> RouterResult<()>
     where
diff --git a/crates/router/src/core/payments/flows/authorize_flow.rs b/crates/router/src/core/payments/flows/authorize_flow.rs
index 1532fffa3f9..f42efa0c1e4 100644
--- a/crates/router/src/core/payments/flows/authorize_flow.rs
+++ b/crates/router/src/core/payments/flows/authorize_flow.rs
@@ -431,7 +431,7 @@ impl Feature<api::Authorize, types::PaymentsAuthorizeData> for types::PaymentsAu
         state: &SessionState,
         connector: &api::ConnectorData,
         should_continue_payment: bool,
-    ) -> RouterResult<types::CreateOrderResult> {
+    ) -> RouterResult<Option<types::CreateOrderResult>> {
         if connector
             .connector_name
             .requires_order_creation_before_payment(self.payment_method)
@@ -471,7 +471,7 @@ impl Feature<api::Authorize, types::PaymentsAuthorizeData> for types::PaymentsAu
                     if let types::PaymentsResponseData::PaymentsCreateOrderResponse { order_id } =
                         res
                     {
-                        Ok(Some(order_id))
+                        Ok(order_id)
                     } else {
                         Err(error_stack::report!(ApiErrorResponse::InternalServerError)
                             .attach_printable(format!(
@@ -482,47 +482,37 @@ impl Feature<api::Authorize, types::PaymentsAuthorizeData> for types::PaymentsAu
                 Err(error) => Err(error),
             };
 
-            Ok(types::CreateOrderResult {
+            Ok(Some(types::CreateOrderResult {
                 create_order_result: create_order_resp,
-                is_create_order_performed: true,
-            })
+            }))
         } else {
-            Ok(types::CreateOrderResult {
-                create_order_result: Ok(None),
-                is_create_order_performed: false,
-            })
+            // If the connector does not require order creation, return None
+            Ok(None)
         }
     }
 
-    async fn update_router_data_with_create_order_result(
+    fn update_router_data_with_create_order_response(
         &mut self,
         create_order_result: types::CreateOrderResult,
-        should_continue_further: bool,
-    ) -> RouterResult<bool> {
-        if create_order_result.is_create_order_performed {
-            match create_order_result.create_order_result {
-                Ok(Some(order_id)) => {
-                    self.request.order_id = Some(order_id.clone());
-                    self.response =
-                        Ok(types::PaymentsResponseData::PaymentsCreateOrderResponse { order_id });
-                    Ok(true)
-                }
-                Ok(None) => Err(error_stack::report!(ApiErrorResponse::InternalServerError)
-                    .attach_printable("Order Id not found."))?,
-                Err(err) => {
-                    self.response = Err(err.clone());
-                    Ok(false)
-                }
+    ) {
+        match create_order_result.create_order_result {
+            Ok(order_id) => {
+                self.request.order_id = Some(order_id.clone()); // ? why this is assigned here and ucs also wants this to populate data
+                self.response =
+                    Ok(types::PaymentsResponseData::PaymentsCreateOrderResponse { order_id });
+            }
+            Err(err) => {
+                self.response = Err(err.clone());
             }
-        } else {
-            Ok(should_continue_further)
         }
     }
 
     async fn call_unified_connector_service<'a>(
         &mut self,
         state: &SessionState,
-        merchant_connector_account: helpers::MerchantConnectorAccountType,
+        #[cfg(feature = "v1")] merchant_connector_account: helpers::MerchantConnectorAccountType,
+        #[cfg(feature = "v2")]
+        merchant_connector_account: domain::MerchantConnectorAccountTypeDetails,
         merchant_context: &domain::MerchantContext,
     ) -> RouterResult<()> {
         let client = state
@@ -558,13 +548,14 @@ impl Feature<api::Authorize, types::PaymentsAuthorizeData> for types::PaymentsAu
 
         let (status, router_data_response) =
             handle_unified_connector_service_response_for_payment_authorize(
-                payment_authorize_response,
+                payment_authorize_response.clone(),
             )
             .change_context(ApiErrorResponse::InternalServerError)
             .attach_printable("Failed to deserialize UCS response")?;
 
         self.status = status;
         self.response = router_data_response;
+        self.raw_connector_response = payment_authorize_response.raw_connector_response;
 
         Ok(())
     }
diff --git a/crates/router/src/core/payments/flows/psync_flow.rs b/crates/router/src/core/payments/flows/psync_flow.rs
index f2ffbe24d11..e27c9974406 100644
--- a/crates/router/src/core/payments/flows/psync_flow.rs
+++ b/crates/router/src/core/payments/flows/psync_flow.rs
@@ -1,6 +1,8 @@
 use std::collections::HashMap;
 
 use async_trait::async_trait;
+use error_stack::ResultExt;
+use unified_connector_service_client::payments as payments_grpc;
 
 use super::{ConstructFlowSpecificData, Feature};
 use crate::{
@@ -8,10 +10,14 @@ use crate::{
     core::{
         errors::{ApiErrorResponse, ConnectorErrorExt, RouterResult},
         payments::{self, access_token, helpers, transformers, PaymentData},
+        unified_connector_service::{
+            build_unified_connector_service_auth_metadata,
+            handle_unified_connector_service_response_for_payment_get,
+        },
     },
     routes::SessionState,
     services::{self, api::ConnectorValidation, logger},
-    types::{self, api, domain},
+    types::{self, api, domain, transformers::ForeignTryFrom},
 };
 
 #[cfg(feature = "v1")]
@@ -204,6 +210,56 @@ impl Feature<api::PSync, types::PaymentsSyncData>
 
         Ok((request, true))
     }
+
+    async fn call_unified_connector_service<'a>(
+        &mut self,
+        state: &SessionState,
+        #[cfg(feature = "v1")] merchant_connector_account: helpers::MerchantConnectorAccountType,
+        #[cfg(feature = "v2")]
+        merchant_connector_account: domain::MerchantConnectorAccountTypeDetails,
+        merchant_context: &domain::MerchantContext,
+    ) -> RouterResult<()> {
+        let client = state
+            .grpc_client
+            .unified_connector_service_client
+            .clone()
+            .ok_or(ApiErrorResponse::InternalServerError)
+            .attach_printable("Failed to fetch Unified Connector Service client")?;
+
+        let payment_get_request = payments_grpc::PaymentServiceGetRequest::foreign_try_from(self)
+            .change_context(ApiErrorResponse::InternalServerError)
+            .attach_printable("Failed to construct Payment Get Request")?;
+
+        let connector_auth_metadata = build_unified_connector_service_auth_metadata(
+            merchant_connector_account,
+            merchant_context,
+        )
+        .change_context(ApiErrorResponse::InternalServerError)
+        .attach_printable("Failed to construct request metadata")?;
+
+        let response = client
+            .payment_get(
+                payment_get_request,
+                connector_auth_metadata,
+                state.get_grpc_headers(),
+            )
+            .await
+            .change_context(ApiErrorResponse::InternalServerError)
+            .attach_printable("Failed to get payment")?;
+
+        let payment_get_response = response.into_inner();
+
+        let (status, router_data_response) =
+            handle_unified_connector_service_response_for_payment_get(payment_get_response.clone())
+                .change_context(ApiErrorResponse::InternalServerError)
+                .attach_printable("Failed to deserialize UCS response")?;
+
+        self.status = status;
+        self.response = router_data_response;
+        self.raw_connector_response = payment_get_response.raw_connector_response;
+
+        Ok(())
+    }
 }
 
 #[async_trait]
diff --git a/crates/router/src/core/payments/transformers.rs b/crates/router/src/core/payments/transformers.rs
index 2379dddb8fd..59359429080 100644
--- a/crates/router/src/core/payments/transformers.rs
+++ b/crates/router/src/core/payments/transformers.rs
@@ -304,7 +304,7 @@ pub async fn construct_payment_router_data_for_authorize<'a>(
         router_return_url: Some(router_return_url),
         webhook_url,
         complete_authorize_url,
-        customer_id: None,
+        customer_id: customer_id.clone(),
         surcharge_details: None,
         request_extended_authorization: None,
         request_incremental_authorization: matches!(
@@ -3466,8 +3466,126 @@ where
 impl<F: Clone> TryFrom<PaymentAdditionalData<'_, F>> for types::PaymentsAuthorizeData {
     type Error = error_stack::Report<errors::ApiErrorResponse>;
 
-    fn try_from(_additional_data: PaymentAdditionalData<'_, F>) -> Result<Self, Self::Error> {
-        todo!()
+    fn try_from(additional_data: PaymentAdditionalData<'_, F>) -> Result<Self, Self::Error> {
+        let payment_data = additional_data.payment_data.clone();
+        let router_base_url = &additional_data.router_base_url;
+        let connector_name = &additional_data.connector_name;
+        let attempt = &payment_data.payment_attempt;
+        let browser_info: Option<types::BrowserInformation> = attempt
+            .browser_info
+            .clone()
+            .map(types::BrowserInformation::from);
+
+        let complete_authorize_url = Some(helpers::create_complete_authorize_url(
+            router_base_url,
+            attempt,
+            connector_name,
+            payment_data.creds_identifier.as_deref(),
+        ));
+
+        let merchant_connector_account_id_or_connector_name = payment_data
+            .payment_attempt
+            .merchant_connector_id
+            .as_ref()
+            .map(|mca_id| mca_id.get_string_repr())
+            .unwrap_or(connector_name);
+
+        let webhook_url = Some(helpers::create_webhook_url(
+            router_base_url,
+            &attempt.merchant_id,
+            merchant_connector_account_id_or_connector_name,
+        ));
+        let router_return_url = Some(helpers::create_redirect_url(
+            router_base_url,
+            attempt,
+            connector_name,
+            payment_data.creds_identifier.as_deref(),
+        ));
+
+        let payment_method_data = payment_data.payment_method_data.or_else(|| {
+            if payment_data.mandate_id.is_some() {
+                Some(domain::PaymentMethodData::MandatePayment)
+            } else {
+                None
+            }
+        });
+
+        let amount = payment_data
+            .payment_attempt
+            .get_total_amount()
+            .get_amount_as_i64();
+
+        let customer_name = additional_data
+            .customer_data
+            .as_ref()
+            .and_then(|customer_data| {
+                customer_data
+                    .name
+                    .as_ref()
+                    .map(|customer| customer.clone().into_inner())
+            });
+
+        let customer_id = additional_data
+            .customer_data
+            .as_ref()
+            .and_then(|data| data.get_id().clone().try_into().ok());
+
+        let merchant_order_reference_id = payment_data
+            .payment_intent
+            .merchant_reference_id
+            .map(|s| s.get_string_repr().to_string());
+
+        let shipping_cost = payment_data.payment_intent.amount_details.shipping_cost;
+
+        Ok(Self {
+            payment_method_data: payment_method_data
+                .unwrap_or(domain::PaymentMethodData::Card(domain::Card::default())),
+            amount,
+            order_tax_amount: None, // V2 doesn't currently support order tax amount
+            email: None,            // V2 doesn't store email directly in payment_intent
+            customer_name,
+            currency: payment_data.currency,
+            confirm: true,
+            statement_descriptor_suffix: None,
+            statement_descriptor: None,
+            capture_method: Some(payment_data.payment_intent.capture_method),
+            router_return_url,
+            webhook_url,
+            complete_authorize_url,
+            setup_future_usage: Some(payment_data.payment_intent.setup_future_usage),
+            mandate_id: payment_data.mandate_id.clone(),
+            off_session: get_off_session(payment_data.mandate_id.as_ref(), None),
+            customer_acceptance: None,
+            setup_mandate_details: None,
+            browser_info,
+            order_details: None,
+            order_category: None,
+            session_token: None,
+            enrolled_for_3ds: false,
+            related_transaction_id: None,
+            payment_experience: None,
+            payment_method_type: Some(payment_data.payment_attempt.payment_method_subtype),
+            surcharge_details: None,
+            customer_id,
+            request_incremental_authorization: false,
+            metadata: payment_data
+                .payment_intent
+                .metadata
+                .clone()
+                .map(|m| m.expose()),
+            authentication_data: None,
+            request_extended_authorization: None,
+            split_payments: None,
+            minor_amount: payment_data.payment_attempt.get_total_amount(),
+            merchant_order_reference_id,
+            integrity_object: None,
+            shipping_cost,
+            additional_payment_method_data: None,
+            merchant_account_id: None,
+            merchant_config_currency: None,
+            connector_testing_data: None,
+            order_id: None,
+        })
     }
 }
 
diff --git a/crates/router/src/core/unified_connector_service.rs b/crates/router/src/core/unified_connector_service.rs
index 3cdcb28d4d8..14836ff20c6 100644
--- a/crates/router/src/core/unified_connector_service.rs
+++ b/crates/router/src/core/unified_connector_service.rs
@@ -1,4 +1,3 @@
-use api_models::admin::ConnectorAuthType;
 use common_enums::{AttemptStatus, PaymentMethodType};
 use common_utils::{errors::CustomResult, ext_traits::ValueExt};
 use error_stack::ResultExt;
@@ -6,9 +5,11 @@ use external_services::grpc_client::unified_connector_service::{
     ConnectorAuthMetadata, UnifiedConnectorServiceError,
 };
 use hyperswitch_connectors::utils::CardData;
+#[cfg(feature = "v2")]
+use hyperswitch_domain_models::merchant_connector_account::MerchantConnectorAccountTypeDetails;
 use hyperswitch_domain_models::{
     merchant_context::MerchantContext,
-    router_data::{ErrorResponse, RouterData},
+    router_data::{ConnectorAuthType, ErrorResponse, RouterData},
     router_response_types::{PaymentsResponseData, RedirectForm},
 };
 use masking::{ExposeInterface, PeekInterface, Secret};
@@ -101,7 +102,7 @@ pub fn build_unified_connector_service_payment_method(
                 }
                 _ => {
                     return Err(UnifiedConnectorServiceError::NotImplemented(format!(
-                        "Unimplemented card payment method type: {payment_method_type:?}"
+                        "Unimplemented payment method subtype: {payment_method_type:?}"
                     ))
                     .into());
                 }
@@ -113,24 +114,52 @@ pub fn build_unified_connector_service_payment_method(
                 })),
             })
         }
+        hyperswitch_domain_models::payment_method_data::PaymentMethodData::Upi(upi_data) => {
+            let upi_type = match upi_data {
+                hyperswitch_domain_models::payment_method_data::UpiData::UpiCollect(
+                    upi_collect_data,
+                ) => {
+                    let vpa_id = upi_collect_data.vpa_id.map(|vpa| vpa.expose());
+                    let upi_details = payments_grpc::UpiCollect { vpa_id };
+                    PaymentMethod::UpiCollect(upi_details)
+                }
+                _ => {
+                    return Err(UnifiedConnectorServiceError::NotImplemented(format!(
+                        "Unimplemented payment method subtype: {payment_method_type:?}"
+                    ))
+                    .into());
+                }
+            };
 
-        _ => Err(UnifiedConnectorServiceError::NotImplemented(
-            "Unimplemented Payment Method".to_string(),
-        )
+            Ok(payments_grpc::PaymentMethod {
+                payment_method: Some(upi_type),
+            })
+        }
+        _ => Err(UnifiedConnectorServiceError::NotImplemented(format!(
+            "Unimplemented payment method: {payment_method_data:?}"
+        ))
         .into()),
     }
 }
 
 pub fn build_unified_connector_service_auth_metadata(
-    merchant_connector_account: MerchantConnectorAccountType,
+    #[cfg(feature = "v1")] merchant_connector_account: MerchantConnectorAccountType,
+    #[cfg(feature = "v2")] merchant_connector_account: MerchantConnectorAccountTypeDetails,
     merchant_context: &MerchantContext,
 ) -> CustomResult<ConnectorAuthMetadata, UnifiedConnectorServiceError> {
+    #[cfg(feature = "v1")]
     let auth_type: ConnectorAuthType = merchant_connector_account
         .get_connector_account_details()
         .parse_value("ConnectorAuthType")
         .change_context(UnifiedConnectorServiceError::FailedToObtainAuthType)
         .attach_printable("Failed while parsing value for ConnectorAuthType")?;
 
+    #[cfg(feature = "v2")]
+    let auth_type: ConnectorAuthType = merchant_connector_account
+        .get_connector_account_details()
+        .change_context(UnifiedConnectorServiceError::FailedToObtainAuthType)
+        .attach_printable("Failed to obtain ConnectorAuthType")?;
+
     let connector_name = {
         #[cfg(feature = "v1")]
         {
@@ -211,30 +240,126 @@ pub fn handle_unified_connector_service_response_for_payment_authorize(
                 })
         });
 
+    let transaction_id = response.transaction_id.as_ref().and_then(|id| {
+        id.id_type.clone().and_then(|id_type| match id_type {
+            payments_grpc::identifier::IdType::Id(id) => Some(id),
+            payments_grpc::identifier::IdType::EncodedData(encoded_data) => Some(encoded_data),
+            payments_grpc::identifier::IdType::NoResponseIdMarker(_) => None,
+        })
+    });
+
     let router_data_response = match status {
         AttemptStatus::Charged |
-        AttemptStatus::Authorized |
-        AttemptStatus::AuthenticationPending |
-        AttemptStatus::DeviceDataCollectionPending => Ok(PaymentsResponseData::TransactionResponse {
-            resource_id: match connector_response_reference_id.as_ref() {
+                AttemptStatus::Authorized |
+                AttemptStatus::AuthenticationPending |
+                AttemptStatus::DeviceDataCollectionPending |
+                AttemptStatus::Started |
+                AttemptStatus::AuthenticationSuccessful |
+                AttemptStatus::Authorizing |
+                AttemptStatus::ConfirmationAwaited |
+                AttemptStatus::Pending => Ok(PaymentsResponseData::TransactionResponse {
+                    resource_id: match transaction_id.as_ref() {
+                        Some(transaction_id) => hyperswitch_domain_models::router_request_types::ResponseId::ConnectorTransactionId(transaction_id.clone()),
+                        None => hyperswitch_domain_models::router_request_types::ResponseId::NoResponseId,
+                    },
+                    redirection_data: Box::new(
+                        response
+                            .redirection_data
+                            .clone()
+                            .map(RedirectForm::foreign_try_from)
+                            .transpose()?
+                    ),
+                    mandate_reference: Box::new(None),
+                    connector_metadata: None,
+                    network_txn_id: response.network_txn_id.clone(),
+                    connector_response_reference_id,
+                    incremental_authorization_allowed: response.incremental_authorization_allowed,
+                    charges: None,
+                }),
+        AttemptStatus::AuthenticationFailed
+                | AttemptStatus::AuthorizationFailed
+                | AttemptStatus::Unresolved
+                | AttemptStatus::Failure => Err(ErrorResponse {
+                    code: response.error_code().to_owned(),
+                    message: response.error_message().to_owned(),
+                    reason: Some(response.error_message().to_owned()),
+                    status_code: 500,
+                    attempt_status: Some(status),
+                    connector_transaction_id: connector_response_reference_id,
+                    network_decline_code: None,
+                    network_advice_code: None,
+                    network_error_message: None,
+                }),
+        AttemptStatus::RouterDeclined |
+                    AttemptStatus::CodInitiated |
+                    AttemptStatus::Voided |
+                    AttemptStatus::VoidInitiated |
+                    AttemptStatus::CaptureInitiated |
+                    AttemptStatus::VoidFailed |
+                    AttemptStatus::AutoRefunded |
+                    AttemptStatus::PartialCharged |
+                    AttemptStatus::PartialChargedAndChargeable |
+                    AttemptStatus::PaymentMethodAwaited |
+                    AttemptStatus::CaptureFailed |
+                    AttemptStatus::IntegrityFailure => return Err(UnifiedConnectorServiceError::NotImplemented(format!(
+                        "AttemptStatus {status:?} is not implemented for Unified Connector Service"
+                    )).into()),
+                };
+
+    Ok((status, router_data_response))
+}
+
+pub fn handle_unified_connector_service_response_for_payment_get(
+    response: payments_grpc::PaymentServiceGetResponse,
+) -> CustomResult<
+    (AttemptStatus, Result<PaymentsResponseData, ErrorResponse>),
+    UnifiedConnectorServiceError,
+> {
+    let status = AttemptStatus::foreign_try_from(response.status())?;
+
+    let connector_response_reference_id =
+        response.response_ref_id.as_ref().and_then(|identifier| {
+            identifier
+                .id_type
+                .clone()
+                .and_then(|id_type| match id_type {
+                    payments_grpc::identifier::IdType::Id(id) => Some(id),
+                    payments_grpc::identifier::IdType::EncodedData(encoded_data) => {
+                        Some(encoded_data)
+                    }
+                    payments_grpc::identifier::IdType::NoResponseIdMarker(_) => None,
+                })
+        });
+
+    let router_data_response = match status {
+        AttemptStatus::Charged |
+                AttemptStatus::Authorized |
+                AttemptStatus::AuthenticationPending |
+                AttemptStatus::DeviceDataCollectionPending |
+                AttemptStatus::Started |
+                AttemptStatus::AuthenticationSuccessful |
+                AttemptStatus::Authorizing |
+                AttemptStatus::ConfirmationAwaited |
+                AttemptStatus::Pending => Ok(
+            PaymentsResponseData::TransactionResponse {
+                resource_id: match connector_response_reference_id.as_ref() {
                 Some(connector_response_reference_id) => hyperswitch_domain_models::router_request_types::ResponseId::ConnectorTransactionId(connector_response_reference_id.clone()),
                 None => hyperswitch_domain_models::router_request_types::ResponseId::NoResponseId,
             },
-            redirection_data: Box::new(
-                response
-                    .redirection_data
-                    .clone()
-                    .map(RedirectForm::foreign_try_from)
-                    .transpose()?
-            ),
-            mandate_reference: Box::new(None),
-            connector_metadata: None,
-            network_txn_id: response.network_txn_id.clone(),
-            connector_response_reference_id,
-            incremental_authorization_allowed: response.incremental_authorization_allowed,
-            charges: None,
-        }),
-        _ => Err(ErrorResponse {
+                redirection_data: Box::new(
+                    None
+                ),
+                mandate_reference: Box::new(None),
+                connector_metadata: None,
+                network_txn_id: response.network_txn_id.clone(),
+                connector_response_reference_id,
+                incremental_authorization_allowed: None,
+                charges: None,
+            }
+        ),
+        AttemptStatus::AuthenticationFailed
+                | AttemptStatus::AuthorizationFailed
+                | AttemptStatus::Failure => Err(ErrorResponse {
             code: response.error_code().to_owned(),
             message: response.error_message().to_owned(),
             reason: Some(response.error_message().to_owned()),
@@ -244,7 +369,22 @@ pub fn handle_unified_connector_service_response_for_payment_authorize(
             network_decline_code: None,
             network_advice_code: None,
             network_error_message: None,
-        })
+        }),
+        AttemptStatus::RouterDeclined |
+                    AttemptStatus::CodInitiated |
+                    AttemptStatus::Voided |
+                    AttemptStatus::VoidInitiated |
+                    AttemptStatus::CaptureInitiated |
+                    AttemptStatus::VoidFailed |
+                    AttemptStatus::AutoRefunded |
+                    AttemptStatus::PartialCharged |
+                    AttemptStatus::PartialChargedAndChargeable |
+                    AttemptStatus::Unresolved |
+                    AttemptStatus::PaymentMethodAwaited |
+                    AttemptStatus::CaptureFailed |
+                    AttemptStatus::IntegrityFailure => return Err(UnifiedConnectorServiceError::NotImplemented(format!(
+                        "AttemptStatus {status:?} is not implemented for Unified Connector Service"
+                    )).into()),
     };
 
     Ok((status, router_data_response))
diff --git a/crates/router/src/core/unified_connector_service/transformers.rs b/crates/router/src/core/unified_connector_service/transformers.rs
index 0e2ae6fd343..e937a7e7d2d 100644
--- a/crates/router/src/core/unified_connector_service/transformers.rs
+++ b/crates/router/src/core/unified_connector_service/transformers.rs
@@ -7,8 +7,8 @@ use error_stack::ResultExt;
 use external_services::grpc_client::unified_connector_service::UnifiedConnectorServiceError;
 use hyperswitch_domain_models::{
     router_data::RouterData,
-    router_flow_types::payments::Authorize,
-    router_request_types::{AuthenticationData, PaymentsAuthorizeData},
+    router_flow_types::payments::{Authorize, PSync},
+    router_request_types::{AuthenticationData, PaymentsAuthorizeData, PaymentsSyncData},
     router_response_types::{PaymentsResponseData, RedirectForm},
 };
 use masking::{ExposeInterface, PeekInterface};
@@ -18,6 +18,37 @@ use crate::{
     core::unified_connector_service::build_unified_connector_service_payment_method,
     types::transformers::ForeignTryFrom,
 };
+impl ForeignTryFrom<&RouterData<PSync, PaymentsSyncData, PaymentsResponseData>>
+    for payments_grpc::PaymentServiceGetRequest
+{
+    type Error = error_stack::Report<UnifiedConnectorServiceError>;
+
+    fn foreign_try_from(
+        router_data: &RouterData<PSync, PaymentsSyncData, PaymentsResponseData>,
+    ) -> Result<Self, Self::Error> {
+        let connector_transaction_id = router_data
+            .request
+            .connector_transaction_id
+            .get_connector_transaction_id()
+            .map(|id| Identifier {
+                id_type: Some(payments_grpc::identifier::IdType::Id(id)),
+            })
+            .ok();
+
+        let connector_ref_id = router_data
+            .request
+            .connector_reference_id
+            .clone()
+            .map(|id| Identifier {
+                id_type: Some(payments_grpc::identifier::IdType::Id(id)),
+            });
+
+        Ok(Self {
+            transaction_id: connector_transaction_id,
+            request_ref_id: connector_ref_id,
+        })
+    }
+}
 
 impl ForeignTryFrom<&RouterData<Authorize, PaymentsAuthorizeData, PaymentsResponseData>>
     for payments_grpc::PaymentServiceAuthorizeRequest
@@ -113,7 +144,7 @@ impl ForeignTryFrom<&RouterData<Authorize, PaymentsAuthorizeData, PaymentsRespon
                 .map(|shipping_cost| shipping_cost.get_amount_as_i64()),
             request_ref_id: Some(Identifier {
                 id_type: Some(payments_grpc::identifier::IdType::Id(
-                    router_data.payment_id.clone(),
+                    router_data.connector_request_reference_id.clone(),
                 )),
             }),
             connector_customer_id: router_data
@@ -181,89 +212,115 @@ impl ForeignTryFrom<hyperswitch_domain_models::payment_address::PaymentAddress>
     fn foreign_try_from(
         payment_address: hyperswitch_domain_models::payment_address::PaymentAddress,
     ) -> Result<Self, Self::Error> {
-        let shipping = payment_address.get_shipping().and_then(|address| {
-            let details = address.address.as_ref()?;
+        let shipping = payment_address.get_shipping().map(|address| {
+            let details = address.address.as_ref();
 
             let get_str =
                 |opt: &Option<masking::Secret<String>>| opt.as_ref().map(|s| s.peek().to_owned());
 
             let get_plain = |opt: &Option<String>| opt.clone();
 
-            let country = details
-                .country
-                .as_ref()
-                .and_then(|c| payments_grpc::CountryAlpha2::from_str_name(&c.to_string()))
-                .ok_or_else(|| {
-                    UnifiedConnectorServiceError::RequestEncodingFailedWithReason(
-                        "Invalid country code".to_string(),
-                    )
-                })
-                .attach_printable("Invalid country code")
-                .ok()? // Return None if invalid
-                .into();
-
-            Some(payments_grpc::Address {
-                first_name: get_str(&details.first_name),
-                last_name: get_str(&details.last_name),
-                line1: get_str(&details.line1),
-                line2: get_str(&details.line2),
-                line3: get_str(&details.line3),
-                city: get_plain(&details.city),
-                state: get_str(&details.state),
-                zip_code: get_str(&details.zip),
-                country_alpha2_code: Some(country),
+            let country = details.and_then(|details| {
+                details
+                    .country
+                    .as_ref()
+                    .and_then(|c| payments_grpc::CountryAlpha2::from_str_name(&c.to_string()))
+                    .map(|country| country.into())
+            });
+
+            payments_grpc::Address {
+                first_name: get_str(&details.and_then(|d| d.first_name.clone())),
+                last_name: get_str(&details.and_then(|d| d.last_name.clone())),
+                line1: get_str(&details.and_then(|d| d.line1.clone())),
+                line2: get_str(&details.and_then(|d| d.line2.clone())),
+                line3: get_str(&details.and_then(|d| d.line3.clone())),
+                city: get_plain(&details.and_then(|d| d.city.clone())),
+                state: get_str(&details.and_then(|d| d.state.clone())),
+                zip_code: get_str(&details.and_then(|d| d.zip.clone())),
+                country_alpha2_code: country,
                 email: address.email.as_ref().map(|e| e.peek().to_string()),
                 phone_number: address
                     .phone
                     .as_ref()
                     .and_then(|phone| phone.number.as_ref().map(|n| n.peek().to_string())),
                 phone_country_code: address.phone.as_ref().and_then(|p| p.country_code.clone()),
-            })
+            }
         });
 
-        let billing = payment_address.get_payment_billing().and_then(|address| {
-            let details = address.address.as_ref()?;
+        let billing = payment_address.get_payment_billing().map(|address| {
+            let details = address.address.as_ref();
 
             let get_str =
                 |opt: &Option<masking::Secret<String>>| opt.as_ref().map(|s| s.peek().to_owned());
 
             let get_plain = |opt: &Option<String>| opt.clone();
 
-            let country = details
-                .country
-                .as_ref()
-                .and_then(|c| payments_grpc::CountryAlpha2::from_str_name(&c.to_string()))
-                .ok_or_else(|| {
-                    UnifiedConnectorServiceError::RequestEncodingFailedWithReason(
-                        "Invalid country code".to_string(),
-                    )
-                })
-                .attach_printable("Invalid country code")
-                .ok()? // Return None if invalid
-                .into();
-
-            Some(payments_grpc::Address {
-                first_name: get_str(&details.first_name),
-                last_name: get_str(&details.last_name),
-                line1: get_str(&details.line1),
-                line2: get_str(&details.line2),
-                line3: get_str(&details.line3),
-                city: get_plain(&details.city),
-                state: get_str(&details.state),
-                zip_code: get_str(&details.zip),
-                country_alpha2_code: Some(country),
+            let country = details.and_then(|details| {
+                details
+                    .country
+                    .as_ref()
+                    .and_then(|c| payments_grpc::CountryAlpha2::from_str_name(&c.to_string()))
+                    .map(|country| country.into())
+            });
+
+            payments_grpc::Address {
+                first_name: get_str(&details.and_then(|d| d.first_name.clone())),
+                last_name: get_str(&details.and_then(|d| d.last_name.clone())),
+                line1: get_str(&details.and_then(|d| d.line1.clone())),
+                line2: get_str(&details.and_then(|d| d.line2.clone())),
+                line3: get_str(&details.and_then(|d| d.line3.clone())),
+                city: get_plain(&details.and_then(|d| d.city.clone())),
+                state: get_str(&details.and_then(|d| d.state.clone())),
+                zip_code: get_str(&details.and_then(|d| d.zip.clone())),
+                country_alpha2_code: country,
                 email: address.email.as_ref().map(|e| e.peek().to_string()),
                 phone_number: address
                     .phone
                     .as_ref()
                     .and_then(|phone| phone.number.as_ref().map(|n| n.peek().to_string())),
                 phone_country_code: address.phone.as_ref().and_then(|p| p.country_code.clone()),
-            })
+            }
         });
 
+        let unified_payment_method_billing =
+            payment_address.get_payment_method_billing().map(|address| {
+                let details = address.address.as_ref();
+
+                let get_str = |opt: &Option<masking::Secret<String>>| {
+                    opt.as_ref().map(|s| s.peek().to_owned())
+                };
+
+                let get_plain = |opt: &Option<String>| opt.clone();
+
+                let country = details.and_then(|details| {
+                    details
+                        .country
+                        .as_ref()
+                        .and_then(|c| payments_grpc::CountryAlpha2::from_str_name(&c.to_string()))
+                        .map(|country| country.into())
+                });
+
+                payments_grpc::Address {
+                    first_name: get_str(&details.and_then(|d| d.first_name.clone())),
+                    last_name: get_str(&details.and_then(|d| d.last_name.clone())),
+                    line1: get_str(&details.and_then(|d| d.line1.clone())),
+                    line2: get_str(&details.and_then(|d| d.line2.clone())),
+                    line3: get_str(&details.and_then(|d| d.line3.clone())),
+                    city: get_plain(&details.and_then(|d| d.city.clone())),
+                    state: get_str(&details.and_then(|d| d.state.clone())),
+                    zip_code: get_str(&details.and_then(|d| d.zip.clone())),
+                    country_alpha2_code: country,
+                    email: address.email.as_ref().map(|e| e.peek().to_string()),
+                    phone_number: address
+                        .phone
+                        .as_ref()
+                        .and_then(|phone| phone.number.as_ref().map(|n| n.peek().to_string())),
+                    phone_country_code: address.phone.as_ref().and_then(|p| p.country_code.clone()),
+                }
+            });
         Ok(Self {
-            shipping_address: shipping,
-            billing_address: billing,
+            shipping_address: shipping.or(unified_payment_method_billing.clone()),
+            billing_address: billing.or(unified_payment_method_billing),
         })
     }
 }
@@ -391,6 +448,12 @@ impl ForeignTryFrom<payments_grpc::RedirectForm> for RedirectForm {
             Some(payments_grpc::redirect_form::FormType::Html(html)) => Ok(Self::Html {
                 html_data: html.html_data,
             }),
+            Some(payments_grpc::redirect_form::FormType::Uri(_)) => Err(
+                UnifiedConnectorServiceError::RequestEncodingFailedWithReason(
+                    "URI form type is not implemented".to_string(),
+                )
+                .into(),
+            ),
             None => Err(
                 UnifiedConnectorServiceError::RequestEncodingFailedWithReason(
                     "Missing form type".to_string(),
diff --git a/crates/router/src/types.rs b/crates/router/src/types.rs
index a5740561c7f..57043bbcc05 100644
--- a/crates/router/src/types.rs
+++ b/crates/router/src/types.rs
@@ -514,9 +514,9 @@ pub struct PaymentMethodTokenResult {
     pub connector_response: Option<ConnectorResponseData>,
 }
 
+#[derive(Clone)]
 pub struct CreateOrderResult {
-    pub create_order_result: Result<Option<String>, ErrorResponse>,
-    pub is_create_order_performed: bool,
+    pub create_order_result: Result<String, ErrorResponse>,
 }
 
 pub struct PspTokenResult {

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

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

  ### Fixes #8440

  ## Description
  <!-- Describe your changes in detail -->

  This PR implements **V2 UCS (Unified Connector Service) Integration** for authorization and synchronization flows, building upon the existing V1 UCS integration (PR#[8280](https://github.com/juspay/hyperswitch/pull/8280)), with enhanced
  V2 schema support according to updated gRPC contract.

  **Key Features Implemented:**

  1. **Authorization Flow (Payments)**: Full V2 schema support for payment authorization
  2. **PSync Flow**: Payment status synchronization through UCS V2
  3. **UPI Payment Method**: Support for UPI Collect payment method
  4. **Enhanced Response Handling**: Improved status mapping and error handling
  5. **Address Transformation Fix**: Better handling of optional address fields

  **Key Changes from V1 to V2:**

  1. **Updated gRPC Schema**:
     - Migrated from `PaymentsAuthorizeRequest` to `PaymentServiceAuthorizeRequest`
     - Added `PaymentServiceGetRequest` for PSync flow
     - Enhanced field structure for better type safety

  2. **New Payment Methods**:
     - Added UPI Collect payment method support
     - Proper VPA ID extraction and handling

  3. **Enhanced Transformers**:
     - Updated transformers for authorization and sync flows
     - Improved card details, address structure, and payment method handling
     - Fixed address transformation to handle None values gracefully

  4. **V2 Flow Architecture**:
     - Refactored payment flow to separate UCS prerequisites and decision logic
     - Added PSync flow integration through UCS
     - Feature flag protection for clean V1/V2 separation

  5. **Improved Response Handling**:
     - Enhanced response parsing with new identifier structure
     - Added support for more payment states (Started, AuthenticationSuccessful, Authorizing, etc.)
     - Better error handling and status mapping

  **V2 Architecture Flow:**
  V2 Payment Request → call_connector_service_prerequisites() → decide_unified_connector_service_call() →
  ├── UCS Available → V2 Transformers → gRPC Call → V2 Response Handling
  └── UCS Unavailable → Traditional Connector Flow (Fallback)

  V2 PSync Request → call_unified_connector_service() →
  ├── UCS Available → V2 PSync Transformer → gRPC Get Call → Status Response
  └── UCS Unavailable → Traditional Connector Sync Flow

  **Key Technical Improvements:**
  - **Refactored Payment Functions**: Split `call_connector_service()` into `call_connector_service_prerequisites()` and `decide_unified_connector_service_call()` for better separation of concerns
  - **Enhanced V2 Transformers**: Updated card data transformation, address structure, and authentication data handling
  - **PSync Flow Integration**: Added complete payment synchronization support through UCS
  - **UPI Payment Support**: Added UPI Collect payment method with VPA ID handling
  - **Improved Error Handling**: Better response parsing with new identifier structure
  - **Customer ID Support**: Added customer ID mapping in V2 transformer implementation
  - **Enhanced Authentication**: Added tenant_id and merchant_id to ConnectorAuthMetadata

  **Files Modified:**
  - `crates/router/Cargo.toml` - Updated rust-grpc-client dependency to main branch
  - `crates/router/src/core/payments.rs` - Refactored UCS integration with V2 prerequisite and decision functions
  - `crates/router/src/core/payments/flows/authorize_flow.rs` - Updated to use `PaymentServiceAuthorizeRequest`
  - `crates/router/src/core/payments/flows/psync_flow.rs` - Added UCS V2 support for payment synchronization
  - `crates/router/src/core/payments/transformers.rs` - Added V2 PaymentsAuthorizeData transformer implementation and customer_id support
  - `crates/router/src/core/unified_connector_service/` - Updated transformers and response handlers for V2 schema
  - `crates/router/src/core/unified_connector_service.rs` - Added UPI payment method support and enhanced response handling
  - `crates/router/src/core/unified_connector_service/transformers.rs` - Major updates for V2 gRPC schema compatibility
  - `crates/external_services/src/grpc_client/unified_connector_service.rs` - Added payment get operations support

  ### Additional Changes

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

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

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

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

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

  **V2 UCS Integration Evolution:**

  This change builds upon the existing V1 UCS integration to support the updated V2 gRPC schema and enhanced payment flow architecture. Key motivations:

  **Schema Evolution:**
  - **Updated gRPC Contracts**: V2 introduces improved field structures for better type safety and extensibility
  - **Enhanced Data Models**: New card details structure, improved address handling, and better identifier management
  - **Backward Compatibility**: V2 implementation maintains compatibility while leveraging new schema features

  **Feature Additions:**
  - **PSync Flow Support**: Enables payment status synchronization through UCS, reducing direct connector calls
  - **UPI Payment Method**: Adds support for popular Indian payment method through unified interface
  - **Enhanced Status Handling**: More granular payment states for better payment lifecycle tracking

  **Architectural Improvements:**
  - **Separation of Concerns**: Split monolithic UCS functions into prerequisite preparation and decision logic phases
  - **Better Feature Flag Management**: Clean V1/V2 separation using `#[cfg(feature = "v2")]` prevents conflicts
  - **Enhanced Error Handling**: Improved response parsing and identifier handling for better reliability
  - **Maintainability**: Cleaner code structure for easier debugging and future enhancements

  **Business Value:**
  - **Future-Proofing**: V2 schema supports upcoming UCS enhancements and new payment flows
  - **Improved Reliability**: Better error handling and response parsing reduces failure rates
  - **Enhanced Developer Experience**: Cleaner API interfaces and better type safety
  - **Operational Excellence**: Maintains existing rollout controls while improving monitoring capabilities
  - **Broader Payment Support**: UPI integration opens up access to Indian market payment preferences

  ## How did you test it?
  <!--
  Did you write an integration/unit/API test to verify the code changes?
  Or did you test this change manually (provide relevant screenshots)?
-->

### Manual Testing (HAPPY CASE)

#### **V2 Payment Intent Creation (Adyen via UCS)**

*Request:*
```bash
curl --location 'http://localhost:8080/v2/payments/create-intent' \
--header 'api-key: dev_ymXSN8By3WESknjGnFyLVwDE4tK9v8ZjnkMZsKb2cM7KgjlzLfU1SOnJ63LsIrfR' \
--header 'Content-Type: application/json' \
--header 'x-profile-id: pro_2MBjeGslBsr2Lr7AoLvr' \
--header 'Authorization: api-key=dev_ymXSN8By3WESknjGnFyLVwDE4tK9v8ZjnkMZsKb2cM7KgjlzLfU1SOnJ63LsIrfR' \
--data '{
    "amount_details": {
        "order_amount": 100,
        "currency": "USD"
    },
    "capture_method": "automatic",
    "authentication_type": "no_three_ds",
    "customer_id": "12345_cus_0197abe38df57a408b928b4111ff5038",
    "billing": {
        "address": {
            "line1": "1467",
            "line2": "Harrison Street",
            "line3": "Harrison Street",
            "city": "San Fransico",
            "state": "California",
            "zip": "94122",
            "country": "US",
            "first_name": "joseph",
            "last_name": "Doe"
        }
    },
    "shipping": {
        "address": {
            "line1": "1467",
            "line2": "Harrison Street",
            "line3": "Harrison Street",
            "city": "San Fransico",
            "state": "California",
            "zip": "94122",
            "country": "US",
            "first_name": "joseph",
            "last_name": "Doe"
        }
    }
}'
```

*Response:*
```json
{
    "id": "12345_pay_0197ac4f440c73c0b34c7050d248e483",
    "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": "cs_0197ac4f4421744094d3abffdeae634c",
    "profile_id": "pro_2MBjeGslBsr2Lr7AoLvr",
    "merchant_reference_id": null,
    "routing_algorithm_id": null,
    "capture_method": "automatic",
    "authentication_type": "no_three_ds",
    "billing": {
        "address": {
            "city": "San Fransico",
            "country": "US",
            "line1": "1467",
            "line2": "Harrison Street",
            "line3": "Harrison Street",
            "zip": "94122",
            "state": "California",
            "first_name": "joseph",
            "last_name": "Doe"
        },
        "phone": null,
        "email": null
    },
    "shipping": {
        "address": {
            "city": "San Fransico",
            "country": "US",
            "line1": "1467",
            "line2": "Harrison Street",
            "line3": "Harrison Street",
            "zip": "94122",
            "state": "California",
            "first_name": "joseph",
            "last_name": "Doe"
        },
        "phone": null,
        "email": null
    },
    "customer_id": "12345_cus_0197abe38df57a408b928b4111ff5038",
    "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-06-26T13:11:05.403Z",
    "frm_metadata": null,
    "request_external_three_ds_authentication": "Skip"
}
```

#### **V2 Payment Confirmation (Adyen via UCS)**

*Request:*
```bash
curl --location 'http://localhost:8080/v2/payments/12345_pay_0197ac4f440c73c0b34c7050d248e483/confirm-intent' \
--header 'x-client-secret: cs_0197ac4f4421744094d3abffdeae634c' \
--header 'x-profile-id: pro_2MBjeGslBsr2Lr7AoLvr' \
--header 'x-connected-merchant-id: cloth_seller_2_X5u3WjKKaUMMtPMYCNQM' \
--header 'Authorization: publishable-key=pk_dev_b78d41b686d54407842e55c1be492fce,client-secret=cs_0197ac4f4421744094d3abffdeae634c' \
--header 'Content-Type: application/json' \
--header 'api-key: pk_dev_b78d41b686d54407842e55c1be492fce' \
--data '{
    "payment_method_type": "card",
    "payment_method_subtype": "credit",
    "payment_method_data": {
        "card": {
            "card_number": "4111111111111111",
            "card_exp_month": "03",
            "card_exp_year": "30",
            "card_holder_name": "Joseph Doe",
            "card_cvc": "737"
        }
    },
    "browser_info": {
        "user_agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/70.0.3538.110 Safari/537.36",
        "accept_header": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8",
        "language": "nl-NL",
        "color_depth": 24,
        "screen_height": 723,
        "screen_width": 1536,
        "time_zone": 0,
        "java_enabled": true,
        "java_script_enabled": true
    }
}'
```

*Response:*
```json
{
    "id": "12345_pay_0197ac4f440c73c0b34c7050d248e483",
    "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": "12345_cus_0197abe38df57a408b928b4111ff5038",
    "connector": "adyen",
    "created": "2025-06-26T12:56:05.403Z",
    "payment_method_data": {
        "billing": null
    },
    "payment_method_type": "card",
    "payment_method_subtype": "credit",
    "connector_transaction_id": "12345_pay_0197ac4f440c73c0b34c7050d248e483",
    "connector_reference_id": null,
    "merchant_connector_id": "mca_YUhrsoeHfh0EQJE6Zqut",
    "browser_info": null,
    "error": 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",
    "is_iframe_redirection_enabled": null,
    "merchant_reference_id": null
}
```

#### **V2 Payment Intent Creation (Adyen via Traditional Connector Flow)**

*Request:*
```bash
curl --location 'http://localhost:8080/v2/payments/create-intent' \
--header 'api-key: dev_ymXSN8By3WESknjGnFyLVwDE4tK9v8ZjnkMZsKb2cM7KgjlzLfU1SOnJ63LsIrfR' \
--header 'Content-Type: application/json' \
--header 'x-profile-id: pro_2MBjeGslBsr2Lr7AoLvr' \
--header 'Authorization: api-key=dev_ymXSN8By3WESknjGnFyLVwDE4tK9v8ZjnkMZsKb2cM7KgjlzLfU1SOnJ63LsIrfR' \
--data '{
    "amount_details": {
        "order_amount": 100,
        "currency": "USD"
    },
    "capture_method": "automatic",
    "authentication_type": "no_three_ds",
    "customer_id": "12345_cus_0197abe38df57a408b928b4111ff5038",
    "billing": {
        "address": {
            "line1": "1467",
            "line2": "Harrison Street",
            "line3": "Harrison Street",
            "city": "San Fransico",
            "state": "California",
            "zip": "94122",
            "country": "US",
            "first_name": "joseph",
            "last_name": "Doe"
        }
    },
    "shipping": {
        "address": {
            "line1": "1467",
            "line2": "Harrison Street",
            "line3": "Harrison Street",
            "city": "San Fransico",
            "state": "California",
            "zip": "94122",
            "country": "US",
            "first_name": "joseph",
            "last_name": "Doe"
        }
    }
}'
```

*Response:*
```json
{
    "id": "12345_pay_0197ac4c67657d81a796bc358b364d04",
    "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": "cs_0197ac4c67ac7731907d52f9dac88996",
    "profile_id": "pro_2MBjeGslBsr2Lr7AoLvr",
    "merchant_reference_id": null,
    "routing_algorithm_id": null,
    "capture_method": "automatic",
    "authentication_type": "no_three_ds",
    "billing": {
        "address": {
            "city": "San Fransico",
            "country": "US",
            "line1": "1467",
            "line2": "Harrison Street",
            "line3": "Harrison Street",
            "zip": "94122",
            "state": "California",
            "first_name": "joseph",
            "last_name": "Doe"
        },
        "phone": null,
        "email": null
    },
    "shipping": {
        "address": {
            "city": "San Fransico",
            "country": "US",
            "line1": "1467",
            "line2": "Harrison Street",
            "line3": "Harrison Street",
            "zip": "94122",
            "state": "California",
            "first_name": "joseph",
            "last_name": "Doe"
        },
        "phone": null,
        "email": null
    },
    "customer_id": "12345_cus_0197abe38df57a408b928b4111ff5038",
    "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-06-26T13:07:57.887Z",
    "frm_metadata": null,
    "request_external_three_ds_authentication": "Skip"
}
```

#### **V2 Payment Confirmation (Adyen via Traditional Connector Flow)**

*Request:*
```bash
curl --location 'http://localhost:8080/v2/payments/12345_pay_0197ac4c67657d81a796bc358b364d04/confirm-intent' \
--header 'x-client-secret: cs_0197ac4c67ac7731907d52f9dac88996' \
--header 'x-profile-id: pro_2MBjeGslBsr2Lr7AoLvr' \
--header 'x-connected-merchant-id: cloth_seller_2_X5u3WjKKaUMMtPMYCNQM' \
--header 'Authorization: publishable-key=pk_dev_b78d41b686d54407842e55c1be492fce,client-secret=cs_0197ac4c67ac7731907d52f9dac88996' \
--header 'Content-Type: application/json' \
--header 'api-key: pk_dev_b78d41b686d54407842e55c1be492fce' \
--data '{
    "payment_method_type": "card",
    "payment_method_subtype": "credit",
    "payment_method_data": {
        "card": {
            "card_number": "4111111111111111",
            "card_exp_month": "03",
            "card_exp_year": "30",
            "card_holder_name": "Joseph Doe",
            "card_cvc": "737"
        }
    },
    "browser_info": {
        "user_agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/70.0.3538.110 Safari/537.36",
        "accept_header": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8",
        "language": "nl-NL",
        "color_depth": 24,
        "screen_height": 723,
        "screen_width": 1536,
        "time_zone": 0,
        "java_enabled": true,
        "java_script_enabled": true
    }
}'
```

*Response:*
```json
{
    "id": "12345_pay_0197ac4c67657d81a796bc358b364d04",
    "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": "12345_cus_0197abe38df57a408b928b4111ff5038",
    "connector": "adyen",
    "created": "2025-06-26T12:52:57.887Z",
    "payment_method_data": {
        "billing": null
    },
    "payment_method_type": "card",
    "payment_method_subtype": "credit",
    "connector_transaction_id": "K2LGMPJFQZ9TPM75",
    "connector_reference_id": null,
    "merchant_connector_id": "mca_YUhrsoeHfh0EQJE6Zqut",
    "browser_info": null,
    "error": 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",
    "is_iframe_redirection_enabled": null,
    "merchant_reference_id": null
}
```

  ### Manual Testing (FAILURE CASE)

#### **V2 Payment Intent Creation (Adyen via UCS)**

*Request:*
```bash
curl --location 'http://localhost:8080/v2/payments/create-intent' \
--header 'api-key: dev_ymXSN8By3WESknjGnFyLVwDE4tK9v8ZjnkMZsKb2cM7KgjlzLfU1SOnJ63LsIrfR' \
--header 'Content-Type: application/json' \
--header 'x-profile-id: pro_2MBjeGslBsr2Lr7AoLvr' \
--header 'Authorization: api-key=dev_ymXSN8By3WESknjGnFyLVwDE4tK9v8ZjnkMZsKb2cM7KgjlzLfU1SOnJ63LsIrfR' \
--data '{
    "amount_details": {
        "order_amount": 100,
        "currency": "USD"
    },
    "capture_method": "automatic",
    "authentication_type": "no_three_ds",
    "customer_id": "12345_cus_0197abe38df57a408b928b4111ff5038",
    "billing": {
        "address": {
            "line1": "1467",
            "line2": "Harrison Street",
            "line3": "Harrison Street",
            "city": "San Fransico",
            "state": "California",
            "zip": "94122",
            "country": "US",
            "first_name": "joseph",
            "last_name": "Doe"
        }
    },
    "shipping": {
        "address": {
            "line1": "1467",
            "line2": "Harrison Street",
            "line3": "Harrison Street",
            "city": "San Fransico",
            "state": "California",
            "zip": "94122",
            "country": "US",
            "first_name": "joseph",
            "last_name": "Doe"
        }
    }
}'
```

*Response:*
```json
{
    "id": "12345_pay_0197ac52cee5796099a0ae352092a707",
    "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": "cs_0197ac52cefb77e2be6097f5d54b1152",
    "profile_id": "pro_2MBjeGslBsr2Lr7AoLvr",
    "merchant_reference_id": null,
    "routing_algorithm_id": null,
    "capture_method": "automatic",
    "authentication_type": "no_three_ds",
    "billing": {
        "address": {
            "city": "San Fransico",
            "country": "US",
            "line1": "1467",
            "line2": "Harrison Street",
            "line3": "Harrison Street",
            "zip": "94122",
            "state": "California",
            "first_name": "joseph",
            "last_name": "Doe"
        },
        "phone": null,
        "email": null
    },
    "shipping": {
        "address": {
            "city": "San Fransico",
            "country": "US",
            "line1": "1467",
            "line2": "Harrison Street",
            "line3": "Harrison Street",
            "zip": "94122",
            "state": "California",
            "first_name": "joseph",
            "last_name": "Doe"
        },
        "phone": null,
        "email": null
    },
    "customer_id": "12345_cus_0197abe38df57a408b928b4111ff5038",
    "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-06-26T13:14:57.559Z",
    "frm_metadata": null,
    "request_external_three_ds_authentication": "Skip"
}
```

#### **V2 Payment Confirmation (Adyen via UCS)**

*Request:*
```bash
curl --location 'http://localhost:8080/v2/payments/12345_pay_0197ac52cee5796099a0ae352092a707/confirm-intent' \
--header 'x-client-secret: cs_0197ac52cefb77e2be6097f5d54b1152' \
--header 'x-profile-id: pro_2MBjeGslBsr2Lr7AoLvr' \
--header 'x-connected-merchant-id: cloth_seller_2_X5u3WjKKaUMMtPMYCNQM' \
--header 'Authorization: publishable-key=pk_dev_b78d41b686d54407842e55c1be492fce,client-secret=cs_0197ac52cefb77e2be6097f5d54b1152' \
--header 'Content-Type: application/json' \
--header 'api-key: pk_dev_b78d41b686d54407842e55c1be492fce' \
--data '{
    "payment_method_type": "card",
    "payment_method_subtype": "credit",
    "payment_method_data": {
        "card": {
            "card_number": "4111111111111111",
            "card_exp_month": "03",
            "card_exp_year": "30",
            "card_holder_name": "Joseph Doe",
            "card_cvc": "730"
        }
    },
    "browser_info": {
        "user_agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/70.0.3538.110 Safari/537.36",
        "accept_header": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8",
        "language": "nl-NL",
        "color_depth": 24,
        "screen_height": 723,
        "screen_width": 1536,
        "time_zone": 0,
        "java_enabled": true,
        "java_script_enabled": true
    }
}'
```

*Response:*
```json
{
    "id": "12345_pay_0197ac52cee5796099a0ae352092a707",
    "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": 0
    },
    "customer_id": "12345_cus_0197abe38df57a408b928b4111ff5038",
    "connector": "adyen",
    "created": "2025-06-26T12:59:57.559Z",
    "payment_method_data": {
        "billing": null
    },
    "payment_method_type": "card",
    "payment_method_subtype": "credit",
    "connector_transaction_id": "G4QT66CH8WCFHJ75",
    "connector_reference_id": null,
    "merchant_connector_id": "mca_YUhrsoeHfh0EQJE6Zqut",
    "browser_info": null,
    "error": {
        "code": "24",
        "message": "CVC Declined",
        "unified_code": null,
        "unified_message": null,
        "network_advice_code": null,
        "network_decline_code": null,
        "network_error_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",
    "is_iframe_redirection_enabled": null,
    "merchant_reference_id": null
}
```

#### **V2 Payment Intent Creation (Adyen via Traditional Connector Flow)**

*Request:*
```bash
curl --location 'http://localhost:8080/v2/payments/create-intent' \
--header 'api-key: dev_ymXSN8By3WESknjGnFyLVwDE4tK9v8ZjnkMZsKb2cM7KgjlzLfU1SOnJ63LsIrfR' \
--header 'Content-Type: application/json' \
--header 'x-profile-id: pro_2MBjeGslBsr2Lr7AoLvr' \
--header 'Authorization: api-key=dev_ymXSN8By3WESknjGnFyLVwDE4tK9v8ZjnkMZsKb2cM7KgjlzLfU1SOnJ63LsIrfR' \
--data '{
    "amount_details": {
        "order_amount": 100,
        "currency": "USD"
    },
    "capture_method": "automatic",
    "authentication_type": "no_three_ds",
    "customer_id": "12345_cus_0197abe38df57a408b928b4111ff5038",
    "billing": {
        "address": {
            "line1": "1467",
            "line2": "Harrison Street",
            "line3": "Harrison Street",
            "city": "San Fransico",
            "state": "California",
            "zip": "94122",
            "country": "US",
            "first_name": "joseph",
            "last_name": "Doe"
        }
    },
    "shipping": {
        "address": {
            "line1": "1467",
            "line2": "Harrison Street",
            "line3": "Harrison Street",
            "city": "San Fransico",
            "state": "California",
            "zip": "94122",
            "country": "US",
            "first_name": "joseph",
            "last_name": "Doe"
        }
    }
}'
```

*Response:*
```json
{
    "id": "12345_pay_0197ac5574b37982b150606298590cfb",
    "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": "cs_0197ac5574c57c50b773a7230ffffe69",
    "profile_id": "pro_2MBjeGslBsr2Lr7AoLvr",
    "merchant_reference_id": null,
    "routing_algorithm_id": null,
    "capture_method": "automatic",
    "authentication_type": "no_three_ds",
    "billing": {
        "address": {
            "city": "San Fransico",
            "country": "US",
            "line1": "1467",
            "line2": "Harrison Street",
            "line3": "Harrison Street",
            "zip": "94122",
            "state": "California",
            "first_name": "joseph",
            "last_name": "Doe"
        },
        "phone": null,
        "email": null
    },
    "shipping": {
        "address": {
            "city": "San Fransico",
            "country": "US",
            "line1": "1467",
            "line2": "Harrison Street",
            "line3": "Harrison Street",
            "zip": "94122",
            "state": "California",
            "first_name": "joseph",
            "last_name": "Doe"
        },
        "phone": null,
        "email": null
    },
    "customer_id": "12345_cus_0197abe38df57a408b928b4111ff5038",
    "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-06-26T13:17:51.072Z",
    "frm_metadata": null,
    "request_external_three_ds_authentication": "Skip"
}
```

#### **V2 Payment Confirmation (Adyen via Traditional Connector Flow)**

*Request:*
```bash
curl --location 'http://localhost:8080/v2/payments/12345_pay_0197ac5574b37982b150606298590cfb/confirm-intent' \
--header 'x-client-secret: cs_0197ac5574c57c50b773a7230ffffe69' \
--header 'x-profile-id: pro_2MBjeGslBsr2Lr7AoLvr' \
--header 'x-connected-merchant-id: cloth_seller_2_X5u3WjKKaUMMtPMYCNQM' \
--header 'Authorization: publishable-key=pk_dev_b78d41b686d54407842e55c1be492fce,client-secret=cs_0197ac5574c57c50b773a7230ffffe69' \
--header 'Content-Type: application/json' \
--header 'api-key: pk_dev_b78d41b686d54407842e55c1be492fce' \
--data '{
    "payment_method_type": "card",
    "payment_method_subtype": "credit",
    "payment_method_data": {
        "card": {
            "card_number": "4111111111111111",
            "card_exp_month": "03",
            "card_exp_year": "30",
            "card_holder_name": "Joseph Doe",
            "card_cvc": "730"
        }
    },
    "browser_info": {
        "user_agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/70.0.3538.110 Safari/537.36",
        "accept_header": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8",
        "language": "nl-NL",
        "color_depth": 24,
        "screen_height": 723,
        "screen_width": 1536,
        "time_zone": 0,
        "java_enabled": true,
        "java_script_enabled": true
    }
}'
```

*Response:*
```json
{
    "id": "12345_pay_0197ac5574b37982b150606298590cfb",
    "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": 0
    },
    "customer_id": "12345_cus_0197abe38df57a408b928b4111ff5038",
    "connector": "adyen",
    "created": "2025-06-26T13:02:51.072Z",
    "payment_method_data": {
        "billing": null
    },
    "payment_method_type": "card",
    "payment_method_subtype": "credit",
    "connector_transaction_id": "DXZQPDL34QGTPM75",
    "connector_reference_id": null,
    "merchant_connector_id": "mca_YUhrsoeHfh0EQJE6Zqut",
    "browser_info": null,
    "error": {
        "code": "24",
        "message": "CVC Declined",
        "unified_code": null,
        "unified_message": null,
        "network_advice_code": null,
        "network_decline_code": null,
        "network_error_message": "DECLINED CVC Incorrect"
    },
    "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",
    "is_iframe_redirection_enabled": null,
    "merchant_reference_id": null
}
```

###UPI Collect payment through UCS 
*Request*

```bash
curl --location 'http://localhost:8080/v2/payments' \
--header 'x-profile-id: pro_2MBjeGslBsr2Lr7AoLvr' \
--header 'x-merchant-id: cloth_seller_2_H3Oh1h4iu0wg8BWGvMzu' \
--header 'x-tenant-id: public' \
--header 'Content-Type: application/json' \
--header 'api-key: dev_ymXSN8By3WESknjGnFyLVwDE4tK9v8ZjnkMZsKb2cM7KgjlzLfU1SOnJ63LsIrfR' \
--data-raw '{
    "amount_details": {
        "currency": "INR",
        "order_amount": 1000
    },    
    "merchant_connector_details": {
        "connector_name": "razorpay",
        "merchant_connector_creds": {
            "auth_type": "SignatureKey",
            "key1": "*************************",
            "api_key": "********************",
            "api_secret": "*******************"
        }
    },
    "merchant_reference_id": "12345_pay_01970c787911751013554",
    "payment_method_data": {
        "upi": {
            "upi_collect": {
                "vpa_id": "failure@razorpay"
            }
        },
        "billing": {
            "phone": {
                "number": "8800574493",
                "country_code": "+91"
            },
            "email": "uzair.khan@juspay.in"
        }
    },
    "payment_method_subtype": "upi_collect",
    "payment_method_type": "upi"
}'
```
*Response*
```json
{
    "id": "12345_pay_0197ceee564473609731b2ab35827cc7",
    "status": "processing",
    "amount": {
        "order_amount": 1000,
        "currency": "INR",
        "shipping_cost": null,
        "order_tax_amount": null,
        "external_tax_calculation": "skip",
        "surcharge_calculation": "skip",
        "surcharge_amount": null,
        "tax_on_surcharge": null,
        "net_amount": 1000,
        "amount_to_capture": null,
        "amount_capturable": 0,
        "amount_captured": null
    },
    "customer_id": null,
    "connector": "razorpay",
    "created": "2025-07-03T06:16:55.621Z",
    "payment_method_data": {
        "billing": {
            "address": null,
            "phone": {
                "number": "8800574493",
                "country_code": "+91"
            },
            "email": "uzair.khan@juspay.in"
        }
    },
    "payment_method_type": "upi",
    "payment_method_subtype": "upi_collect",
    "connector_transaction_id": "pay_QoUbCBtcPCfaWY",
    "connector_reference_id": "order_QoUbBDL6PKnuZY",
    "merchant_connector_id": null,
    "browser_info": null,
    "error": 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": null,
    "authentication_type_applied": "no_three_ds",
    "is_iframe_redirection_enabled": null,
    "merchant_reference_id": "12345_pay_01970c787911751013554",
    "whole_connector_response": null
}
```
### PSync through UCS 

*Request*

```bash
curl --location 'http://localhost:8080/v2/payments/12345_pay_0197ceee564473609731b2ab35827cc7' \
--header 'x-profile-id: pro_2MBjeGslBsr2Lr7AoLvr' \
--header 'x-merchant-id: cloth_seller_2_H3Oh1h4iu0wg8BWGvMzu' \
--header 'x-tenant-id: public' \
--header 'Content-Type: application/json' \
--header 'api-key: dev_ymXSN8By3WESknjGnFyLVwDE4tK9v8ZjnkMZsKb2cM7KgjlzLfU1SOnJ63LsIrfR' \
--data '{
    "merchant_connector_details": {
        "connector_name": "razorpay",
        "merchant_connector_creds": {
            "auth_type": "SignatureKey",
            "key1": "**************",
            "api_key": "********************",
            "api_secret": "******************"
        }
    },
    "force_sync": true
}'```
*Response*
```json
{
    "id": "12345_pay_0197ceee564473609731b2ab35827cc7",
    "status": "failed",
    "amount": {
        "order_amount": 1000,
        "currency": "INR",
        "shipping_cost": null,
        "order_tax_amount": null,
        "external_tax_calculation": "skip",
        "surcharge_calculation": "skip",
        "surcharge_amount": null,
        "tax_on_surcharge": null,
        "net_amount": 1000,
        "amount_to_capture": null,
        "amount_capturable": 0,
        "amount_captured": 0
    },
    "customer_id": null,
    "connector": "razorpay",
    "created": "2025-07-03T06:16:55.621Z",
    "payment_method_data": {
        "billing": {
            "address": null,
            "phone": {
                "number": "8800574493",
                "country_code": "+91"
            },
            "email": "uzair.khan@juspay.in"
        }
    },
    "payment_method_type": "upi",
    "payment_method_subtype": "upi_collect",
    "connector_transaction_id": "pay_QoUbCBtcPCfaWY",
    "connector_reference_id": "order_QoUbBDL6PKnuZY",
    "merchant_connector_id": null,
    "browser_info": null,
    "error": {
        "code": "",
        "message": "",
        "unified_code": null,
        "unified_message": null,
        "network_advice_code": null,
        "network_decline_code": null,
        "network_error_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": null,
    "authentication_type_applied": null,
    "is_iframe_redirection_enabled": null,
    "merchant_reference_id": "12345_pay_01970c787911751013554",
    "whole_connector_response": null
}
```
**V2 Schema Testing:**
- V2 gRPC schema compatibility verification
- `PaymentServiceAuthorizeRequest` field mapping validation
- Enhanced card details and address structure testing
- Response identifier parsing with new structure

**Fallback Testing:**
- UCS service unavailable → Automatic fallback to traditional connector flow
- Invalid authentication → Fallback with error logging
- V2 transformer errors → Graceful degradation to V1 flow
- Rollout percentage = 0% → Traditional flow used
- Rollout percentage = 100% → UCS V2 flow used

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