chore: added fork and headers

crates/router/src/core/payments.rs

@@ -4075,12 +4075,13 @@ pub async fn call_connector_service_prerequisites<F, RouterDReq, ApiRequest, D>(
 )>
 where
     F: Send + Clone + Sync,
-    RouterDReq: Send + Sync,
+    RouterDReq: Send + Clone + 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,
+    RouterData<F, RouterDReq, router_types::PaymentsResponseData>:
+        Feature<F, RouterDReq> + Send + Clone,
     // To construct connector flow specific api
     dyn api::Connector:
         services::api::ConnectorIntegration<F, RouterDReq, router_types::PaymentsResponseData>,
@@ -4237,16 +4238,18 @@ pub async fn decide_unified_connector_service_call<F, RouterDReq, ApiRequest, D>
 )>
 where
     F: Send + Clone + Sync,
-    RouterDReq: Send + Sync,
+    RouterDReq: Send + Sync + Clone,
 
     // 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,
+    RouterData<F, RouterDReq, router_types::PaymentsResponseData>:
+        Feature<F, RouterDReq> + Send + Clone,
     // To construct connector flow specific api
     dyn api::Connector:
         services::api::ConnectorIntegration<F, RouterDReq, router_types::PaymentsResponseData>,
 {
+    let mut rd = router_data.clone();
     record_time_taken_with(|| async {
         if !matches!(
             call_connector_action,
@@ -4326,6 +4329,13 @@ where
 
             // Update feature metadata to track Direct routing usage for stickiness
             update_gateway_system_in_feature_metadata(payment_data, GatewaySystem::Direct)?;
+            let _result = rd
+                .call_unified_connector_service(
+                    state,
+                    merchant_connector_account.clone(),
+                    merchant_context,
+                )
+                .await?;
 
             call_connector_service(
                 state,

crates/router/src/core/payments/retry.rs

@@ -49,7 +49,7 @@ pub async fn do_gsm_actions<F, ApiRequest, FData, D>(
 ) -> RouterResult<types::RouterData<F, FData, types::PaymentsResponseData>>
 where
     F: Clone + Send + Sync,
-    FData: Send + Sync + types::Capturable,
+    FData: Send + Sync + types::Capturable + Clone,
     payments::PaymentResponse: operations::Operation<F, FData>,
     D: payments::OperationSessionGetters<F>
         + payments::OperationSessionSetters<F>
@@ -357,7 +357,7 @@ pub async fn do_retry<F, ApiRequest, FData, D>(
 ) -> RouterResult<types::RouterData<F, FData, types::PaymentsResponseData>>
 where
     F: Clone + Send + Sync,
-    FData: Send + Sync + types::Capturable,
+    FData: Send + Sync + types::Capturable + Clone,
     payments::PaymentResponse: operations::Operation<F, FData>,
     D: payments::OperationSessionGetters<F>
         + payments::OperationSessionSetters<F>

crates/router/src/routes/app.rs

@@ -157,7 +157,10 @@ impl SessionState {
     }
     pub fn get_grpc_headers_ucs(&self) -> GrpcHeadersUcsBuilderInitial {
         let tenant_id = self.tenant.tenant_id.get_string_repr().to_string();
-        GrpcHeadersUcs::builder().tenant_id(tenant_id)
+        let request_id = self.request_id.map(|req_id| (*req_id).to_string());
+        GrpcHeadersUcs::builder()
+            .tenant_id(tenant_id)
+            .request_id(request_id)
     }
     #[cfg(all(feature = "revenue_recovery", feature = "v2"))]
     pub fn get_recovery_grpc_headers(&self) -> GrpcRecoveryHeaders {

crates/router/src/services/api.rs

@@ -20,7 +20,7 @@ pub use client::{ApiClient, MockApiClient, ProxyClient};
 pub use common_enums::enums::PaymentAction;
 pub use common_utils::request::{ContentType, Method, Request, RequestBuilder};
 use common_utils::{
-    consts::{DEFAULT_TENANT, TENANT_HEADER, X_HS_LATENCY},
+    consts::{DEFAULT_TENANT, TENANT_HEADER, X_CONNECTOR, X_FLOW, X_HS_LATENCY, X_REQUEST_ID},
     errors::{ErrorSwitch, ReportSwitchExt},
     request::RequestContent,
 };
@@ -326,7 +326,7 @@ where
             };
 
             match connector_request {
-                Some(request) => {
+                Some(mut request) => {
                     let masked_request_body = match &request.body {
                         Some(request) => match request {
                             RequestContent::Json(i)
@@ -339,6 +339,29 @@ where
                         },
                         None => serde_json::Value::Null,
                     };
+                    let flow_name = std::any::type_name::<T>()
+                        .split("::")
+                        .last()
+                        .unwrap_or_default();
+
+                    request.headers.insert((
+                        X_FLOW.to_string(),
+                        Maskable::Masked(Secret::new(flow_name.clone().to_string())),
+                    ));
+
+                    let connector_name = req.connector.clone();
+                    request.headers.insert((
+                        X_CONNECTOR.to_string(),
+                        Maskable::Masked(Secret::new(connector_name.clone().to_string())),
+                    ));
+                    state.request_id.as_ref().map(|id| {
+                        let request_id = id.to_string();
+                        request.headers.insert((
+                            X_REQUEST_ID.to_string(),
+                            Maskable::Masked(Secret::new(request_id.clone())),
+                        ));
+                        request_id
+                    });
                     let request_url = request.url.clone();
                     let request_method = request.method;
                     let current_time = Instant::now();