feat(core): [Payouts] Add access_token flow for Payout Create and Fulfill flow (#4375)

Co-authored-by: hyperswitch-bot[bot] <148525504+hyperswitch-bot[bot]@users.noreply.github.com>

crates/api_models/src/enums.rs

@@ -135,6 +135,25 @@ pub enum Connector {
 }
 
 impl Connector {
+    #[cfg(feature = "payouts")]
+    pub fn supports_instant_payout(&self, payout_method: PayoutType) -> bool {
+        matches!(
+            (self, payout_method),
+            (Self::Paypal, PayoutType::Wallet) | (_, PayoutType::Card)
+        )
+    }
+    #[cfg(feature = "payouts")]
+    pub fn supports_create_recipient(&self, payout_method: PayoutType) -> bool {
+        matches!((self, payout_method), (_, PayoutType::Bank))
+    }
+    #[cfg(feature = "payouts")]
+    pub fn supports_payout_eligibility(&self, payout_method: PayoutType) -> bool {
+        matches!((self, payout_method), (_, PayoutType::Card))
+    }
+    #[cfg(feature = "payouts")]
+    pub fn supports_access_token_for_payout(&self, payout_method: PayoutType) -> bool {
+        matches!((self, payout_method), (Self::Paypal, _))
+    }
     pub fn supports_access_token(&self, payment_method: PaymentMethod) -> bool {
         matches!(
             (self, payment_method),
@@ -320,6 +339,7 @@ pub enum AuthenticationConnectors {
 pub enum PayoutConnectors {
     Adyen,
     Wise,
+    Paypal,
 }
 
 #[cfg(feature = "payouts")]
@@ -328,6 +348,7 @@ impl From<PayoutConnectors> for RoutableConnectors {
         match value {
             PayoutConnectors::Adyen => Self::Adyen,
             PayoutConnectors::Wise => Self::Wise,
+            PayoutConnectors::Paypal => Self::Paypal,
         }
     }
 }
@@ -338,6 +359,7 @@ impl From<PayoutConnectors> for Connector {
         match value {
             PayoutConnectors::Adyen => Self::Adyen,
             PayoutConnectors::Wise => Self::Wise,
+            PayoutConnectors::Paypal => Self::Paypal,
         }
     }
 }
@@ -349,6 +371,7 @@ impl TryFrom<Connector> for PayoutConnectors {
         match value {
             Connector::Adyen => Ok(Self::Adyen),
             Connector::Wise => Ok(Self::Wise),
+            Connector::Paypal => Ok(Self::Paypal),
             _ => Err(format!("Invalid payout connector {}", value)),
         }
     }

crates/connector_configs/src/connector.rs

@@ -215,6 +215,7 @@ impl ConnectorConfig {
         match connector {
             PayoutConnectors::Adyen => Ok(connector_data.adyen_payout),
             PayoutConnectors::Wise => Ok(connector_data.wise_payout),
+            PayoutConnectors::Paypal => Ok(connector_data.paypal),
         }
     }
 

crates/router/src/core/payouts.rs

@@ -1,3 +1,4 @@
+pub mod access_token;
 pub mod helpers;
 #[cfg(feature = "payout_retry")]
 pub mod retry;
@@ -160,7 +161,7 @@ pub async fn make_connector_decision(
                 key_store,
                 req,
                 &connector_data,
-                &mut payout_data,
+                payout_data,
             )
             .await?;
 
@@ -200,7 +201,7 @@ pub async fn make_connector_decision(
                 key_store,
                 req,
                 &connector_data,
-                &mut payout_data,
+                payout_data,
             )
             .await?;
 
@@ -858,7 +859,7 @@ pub async fn call_connector_payout(
     key_store: &domain::MerchantKeyStore,
     req: &payouts::PayoutCreateRequest,
     connector_data: &api::ConnectorData,
-    payout_data: &mut PayoutData,
+    mut payout_data: PayoutData,
 ) -> RouterResult<PayoutData> {
     let payout_attempt = &payout_data.payout_attempt.to_owned();
     let payouts = &payout_data.payouts.to_owned();
@@ -896,7 +897,7 @@ pub async fn call_connector_payout(
                 &payout_attempt.merchant_id,
                 Some(&payouts.payout_type),
                 key_store,
-                Some(payout_data),
+                Some(&mut payout_data),
                 merchant_account.storage_scheme,
             )
             .await?
@@ -906,10 +907,7 @@ pub async fn call_connector_payout(
 
     if let Some(true) = req.confirm {
         // Eligibility flow
-        if payouts.payout_type == storage_enums::PayoutType::Card
+        payout_data = complete_payout_eligibility(
-            && payout_attempt.is_eligible.is_none()
-        {
-            *payout_data = check_payout_eligibility(
             state,
             merchant_account,
             key_store,
@@ -917,29 +915,21 @@ pub async fn call_connector_payout(
             connector_data,
             payout_data,
         )
-            .await
+        .await?;
-            .attach_printable("Eligibility failed for given Payout request")?;
+
-        }
+        // Create customer flow
+        payout_data = complete_create_recipient(
+            state,
+            merchant_account,
+            key_store,
+            req,
+            connector_data,
+            payout_data,
+        )
+        .await?;
 
         // Payout creation flow
-        utils::when(
+        payout_data = complete_create_payout(
-            !payout_attempt
-                .is_eligible
-                .unwrap_or(state.conf.payouts.payout_eligibility),
-            || {
-                Err(report!(errors::ApiErrorResponse::PayoutFailed {
-                    data: Some(serde_json::json!({
-                        "message": "Payout method data is invalid"
-                    }))
-                })
-                .attach_printable("Payout data provided is invalid"))
-            },
-        )?;
-        if payout_data.payouts.payout_type == storage_enums::PayoutType::Bank
-            && payout_data.payout_attempt.status == storage_enums::PayoutStatus::RequiresCreation
-        {
-            // Create customer flow
-            *payout_data = create_recipient(
             state,
             merchant_account,
             key_store,
@@ -947,55 +937,53 @@ pub async fn call_connector_payout(
             connector_data,
             payout_data,
         )
-            .await
+        .await?;
-            .attach_printable("Creation of customer failed")?;
-
-            // Create payout flow
-            *payout_data = create_payout(
-                state,
-                merchant_account,
-                key_store,
-                req,
-                connector_data,
-                payout_data,
-            )
-            .await
-            .attach_printable("Payout creation failed for given Payout request")?;
-        }
-
-        if payout_data.payouts.payout_type == storage_enums::PayoutType::Wallet
-            && payout_data.payout_attempt.status == storage_enums::PayoutStatus::RequiresCreation
-        {
-            // Create payout flow
-            *payout_data = create_payout(
-                state,
-                merchant_account,
-                key_store,
-                req,
-                connector_data,
-                payout_data,
-            )
-            .await
-            .attach_printable("Payout creation failed for given Payout request")?;
-        }
     };
 
     // Auto fulfillment flow
     let status = payout_data.payout_attempt.status;
     if payouts.auto_fulfill && status == storage_enums::PayoutStatus::RequiresFulfillment {
-        *payout_data = fulfill_payout(
+        payout_data = fulfill_payout(
             state,
             merchant_account,
             key_store,
             &payouts::PayoutRequest::PayoutCreateRequest(req.to_owned()),
             connector_data,
-            payout_data,
+            &mut payout_data,
         )
         .await
         .attach_printable("Payout fulfillment failed for given Payout request")?;
     }
 
-    Ok(payout_data.to_owned())
+    Ok(payout_data)
+}
+
+pub async fn complete_create_recipient(
+    state: &AppState,
+    merchant_account: &domain::MerchantAccount,
+    key_store: &domain::MerchantKeyStore,
+    req: &payouts::PayoutCreateRequest,
+    connector_data: &api::ConnectorData,
+    mut payout_data: PayoutData,
+) -> RouterResult<PayoutData> {
+    if payout_data.payout_attempt.status == storage_enums::PayoutStatus::RequiresCreation
+        && connector_data
+            .connector_name
+            .supports_create_recipient(payout_data.payouts.payout_type)
+    {
+        payout_data = create_recipient(
+            state,
+            merchant_account,
+            key_store,
+            req,
+            connector_data,
+            &mut payout_data,
+        )
+        .await
+        .attach_printable("Creation of customer failed")?;
+    }
+
+    Ok(payout_data)
 }
 
 pub async fn create_recipient(
@@ -1084,6 +1072,50 @@ pub async fn create_recipient(
     Ok(payout_data.clone())
 }
 
+pub async fn complete_payout_eligibility(
+    state: &AppState,
+    merchant_account: &domain::MerchantAccount,
+    key_store: &domain::MerchantKeyStore,
+    req: &payouts::PayoutCreateRequest,
+    connector_data: &api::ConnectorData,
+    mut payout_data: PayoutData,
+) -> RouterResult<PayoutData> {
+    let payout_attempt = &payout_data.payout_attempt.to_owned();
+
+    if payout_attempt.is_eligible.is_none()
+        && connector_data
+            .connector_name
+            .supports_payout_eligibility(payout_data.payouts.payout_type)
+    {
+        payout_data = check_payout_eligibility(
+            state,
+            merchant_account,
+            key_store,
+            req,
+            connector_data,
+            &mut payout_data,
+        )
+        .await
+        .attach_printable("Eligibility failed for given Payout request")?;
+    }
+
+    utils::when(
+        !payout_attempt
+            .is_eligible
+            .unwrap_or(state.conf.payouts.payout_eligibility),
+        || {
+            Err(report!(errors::ApiErrorResponse::PayoutFailed {
+                data: Some(serde_json::json!({
+                    "message": "Payout method data is invalid"
+                }))
+            })
+            .attach_printable("Payout data provided is invalid"))
+        },
+    )?;
+
+    Ok(payout_data)
+}
+
 pub async fn check_payout_eligibility(
     state: &AppState,
     merchant_account: &domain::MerchantAccount,
@@ -1200,6 +1232,68 @@ pub async fn check_payout_eligibility(
     Ok(payout_data.clone())
 }
 
+pub async fn complete_create_payout(
+    state: &AppState,
+    merchant_account: &domain::MerchantAccount,
+    key_store: &domain::MerchantKeyStore,
+    req: &payouts::PayoutCreateRequest,
+    connector_data: &api::ConnectorData,
+    mut payout_data: PayoutData,
+) -> RouterResult<PayoutData> {
+    if payout_data.payout_attempt.status == storage_enums::PayoutStatus::RequiresCreation {
+        if connector_data
+            .connector_name
+            .supports_instant_payout(payout_data.payouts.payout_type)
+        {
+            // create payout_object only in router
+            let db = &*state.store;
+            let payout_attempt = &payout_data.payout_attempt;
+            let updated_payout_attempt = storage::PayoutAttemptUpdate::StatusUpdate {
+                connector_payout_id: "".to_string(),
+                status: storage::enums::PayoutStatus::RequiresFulfillment,
+                error_code: None,
+                error_message: None,
+                is_eligible: None,
+            };
+            payout_data.payout_attempt = db
+                .update_payout_attempt(
+                    payout_attempt,
+                    updated_payout_attempt,
+                    &payout_data.payouts,
+                    merchant_account.storage_scheme,
+                )
+                .await
+                .change_context(errors::ApiErrorResponse::InternalServerError)
+                .attach_printable("Error updating payout_attempt in db")?;
+            payout_data.payouts = db
+                .update_payout(
+                    &payout_data.payouts,
+                    storage::PayoutsUpdate::StatusUpdate {
+                        status: storage::enums::PayoutStatus::RequiresFulfillment,
+                    },
+                    &payout_data.payout_attempt,
+                    merchant_account.storage_scheme,
+                )
+                .await
+                .change_context(errors::ApiErrorResponse::InternalServerError)
+                .attach_printable("Error updating payouts in db")?;
+        } else {
+            // create payout_object in connector as well as router
+            payout_data = create_payout(
+                state,
+                merchant_account,
+                key_store,
+                req,
+                connector_data,
+                &mut payout_data,
+            )
+            .await
+            .attach_printable("Payout creation failed for given Payout request")?;
+        }
+    }
+    Ok(payout_data)
+}
+
 pub async fn create_payout(
     state: &AppState,
     merchant_account: &domain::MerchantAccount,
@@ -1219,7 +1313,17 @@ pub async fn create_payout(
     )
     .await?;
 
-    // 2. Fetch connector integration details
+    // 2. Get/Create access token
+    access_token::create_access_token(
+        state,
+        connector_data,
+        merchant_account,
+        &mut router_data,
+        payout_data.payouts.payout_type.to_owned(),
+    )
+    .await?;
+
+    // 3. Fetch connector integration details
     let connector_integration: services::BoxedConnectorIntegration<
         '_,
         api::PoCreate,
@@ -1227,13 +1331,13 @@ pub async fn create_payout(
         types::PayoutsResponseData,
     > = connector_data.connector.get_connector_integration();
 
-    // 3. Execute pretasks
+    // 4. Execute pretasks
     connector_integration
         .execute_pretasks(&mut router_data, state)
         .await
         .to_payout_failed_response()?;
 
-    // 4. Call connector service
+    // 5. Call connector service
     let router_data_resp = services::execute_connector_processing_step(
         state,
         connector_integration,
@@ -1244,7 +1348,7 @@ pub async fn create_payout(
     .await
     .to_payout_failed_response()?;
 
-    // 5. Process data returned by the connector
+    // 6. Process data returned by the connector
     let db = &*state.store;
     match router_data_resp.response {
         Ok(payout_response_data) => {
@@ -1439,7 +1543,7 @@ pub async fn fulfill_payout(
     payout_data: &mut PayoutData,
 ) -> RouterResult<PayoutData> {
     // 1. Form Router data
-    let router_data = core_utils::construct_payout_router_data(
+    let mut router_data = core_utils::construct_payout_router_data(
         state,
         &connector_data.connector_name.to_string(),
         merchant_account,
@@ -1449,7 +1553,17 @@ pub async fn fulfill_payout(
     )
     .await?;
 
-    // 2. Fetch connector integration details
+    // 2. Get/Create access token
+    access_token::create_access_token(
+        state,
+        connector_data,
+        merchant_account,
+        &mut router_data,
+        payout_data.payouts.payout_type.to_owned(),
+    )
+    .await?;
+
+    // 3. Fetch connector integration details
     let connector_integration: services::BoxedConnectorIntegration<
         '_,
         api::PoFulfill,
@@ -1457,7 +1571,7 @@ pub async fn fulfill_payout(
         types::PayoutsResponseData,
     > = connector_data.connector.get_connector_integration();
 
-    // 3. Call connector service
+    // 4. Call connector service
     let router_data_resp = services::execute_connector_processing_step(
         state,
         connector_integration,
@@ -1468,7 +1582,7 @@ pub async fn fulfill_payout(
     .await
     .to_payout_failed_response()?;
 
-    // 4. Process data returned by the connector
+    // 5. Process data returned by the connector
     let db = &*state.store;
     match router_data_resp.response {
         Ok(payout_response_data) => {

crates/router/src/core/payouts/access_token.rs

@@ -0,0 +1,194 @@
+use common_utils::ext_traits::AsyncExt;
+use error_stack::ResultExt;
+
+use crate::{
+    consts,
+    core::{
+        errors::{self, RouterResult},
+        payments,
+    },
+    routes::{metrics, AppState},
+    services,
+    types::{self, api as api_types, domain, storage::enums},
+};
+
+/// After we get the access token, check if there was an error and if the flow should proceed further
+/// Everything is well, continue with the flow
+/// There was an error, cannot proceed further
+#[cfg(feature = "payouts")]
+pub async fn create_access_token<F: Clone + 'static>(
+    state: &AppState,
+    connector_data: &api_types::ConnectorData,
+    merchant_account: &domain::MerchantAccount,
+    router_data: &mut types::PayoutsRouterData<F>,
+    payout_type: enums::PayoutType,
+) -> RouterResult<()> {
+    let connector_access_token = add_access_token_for_payout(
+        state,
+        connector_data,
+        merchant_account,
+        router_data,
+        payout_type,
+    )
+    .await?;
+
+    if connector_access_token.connector_supports_access_token {
+        match connector_access_token.access_token_result {
+            Ok(access_token) => {
+                router_data.access_token = access_token;
+            }
+            Err(connector_error) => {
+                router_data.response = Err(connector_error);
+            }
+        }
+    }
+
+    Ok(())
+}
+
+#[cfg(feature = "payouts")]
+pub async fn add_access_token_for_payout<F: Clone + 'static>(
+    state: &AppState,
+    connector: &api_types::ConnectorData,
+    merchant_account: &domain::MerchantAccount,
+    router_data: &types::PayoutsRouterData<F>,
+    payout_type: enums::PayoutType,
+) -> RouterResult<types::AddAccessTokenResult> {
+    if connector
+        .connector_name
+        .supports_access_token_for_payout(payout_type)
+    {
+        let merchant_id = &merchant_account.merchant_id;
+        let store = &*state.store;
+        let old_access_token = store
+            .get_access_token(merchant_id, connector.connector.id())
+            .await
+            .change_context(errors::ApiErrorResponse::InternalServerError)
+            .attach_printable("DB error when accessing the access token")?;
+
+        let res = match old_access_token {
+            Some(access_token) => Ok(Some(access_token)),
+            None => {
+                let cloned_router_data = router_data.clone();
+                let refresh_token_request_data = types::AccessTokenRequestData::try_from(
+                    router_data.connector_auth_type.clone(),
+                )
+                .attach_printable(
+                    "Could not create access token request, invalid connector account credentials",
+                )?;
+
+                let refresh_token_response_data: Result<types::AccessToken, types::ErrorResponse> =
+                    Err(types::ErrorResponse::default());
+                let refresh_token_router_data = payments::helpers::router_data_type_conversion::<
+                    _,
+                    api_types::AccessTokenAuth,
+                    _,
+                    _,
+                    _,
+                    _,
+                >(
+                    cloned_router_data,
+                    refresh_token_request_data,
+                    refresh_token_response_data,
+                );
+                refresh_connector_auth(
+                    state,
+                    connector,
+                    merchant_account,
+                    &refresh_token_router_data,
+                )
+                .await?
+                .async_map(|access_token| async {
+                    //Store the access token in db
+                    let store = &*state.store;
+                    // This error should not be propagated, we don't want payments to fail once we have
+                    // the access token, the next request will create new access token
+                    let _ = store
+                        .set_access_token(
+                            merchant_id,
+                            connector.connector.id(),
+                            access_token.clone(),
+                        )
+                        .await
+                        .change_context(errors::ApiErrorResponse::InternalServerError)
+                        .attach_printable("DB error when setting the access token");
+                    Some(access_token)
+                })
+                .await
+            }
+        };
+
+        Ok(types::AddAccessTokenResult {
+            access_token_result: res,
+            connector_supports_access_token: true,
+        })
+    } else {
+        Ok(types::AddAccessTokenResult {
+            access_token_result: Err(types::ErrorResponse::default()),
+            connector_supports_access_token: false,
+        })
+    }
+}
+
+#[cfg(feature = "payouts")]
+pub async fn refresh_connector_auth(
+    state: &AppState,
+    connector: &api_types::ConnectorData,
+    _merchant_account: &domain::MerchantAccount,
+    router_data: &types::RouterData<
+        api_types::AccessTokenAuth,
+        types::AccessTokenRequestData,
+        types::AccessToken,
+    >,
+) -> RouterResult<Result<types::AccessToken, types::ErrorResponse>> {
+    let connector_integration: services::BoxedConnectorIntegration<
+        '_,
+        api_types::AccessTokenAuth,
+        types::AccessTokenRequestData,
+        types::AccessToken,
+    > = connector.connector.get_connector_integration();
+
+    let access_token_router_data_result = services::execute_connector_processing_step(
+        state,
+        connector_integration,
+        router_data,
+        payments::CallConnectorAction::Trigger,
+        None,
+    )
+    .await;
+
+    let access_token_router_data = match access_token_router_data_result {
+        Ok(router_data) => Ok(router_data.response),
+        Err(connector_error) => {
+            // If we receive a timeout error from the connector, then
+            // the error has to be handled gracefully by updating the payment status to failed.
+            // further payment flow will not be continued
+            if connector_error.current_context().is_connector_timeout() {
+                let error_response = types::ErrorResponse {
+                    code: consts::REQUEST_TIMEOUT_ERROR_CODE.to_string(),
+                    message: consts::REQUEST_TIMEOUT_ERROR_MESSAGE.to_string(),
+                    reason: Some(consts::REQUEST_TIMEOUT_ERROR_MESSAGE.to_string()),
+                    status_code: 504,
+                    attempt_status: None,
+                    connector_transaction_id: None,
+                };
+
+                Ok(Err(error_response))
+            } else {
+                Err(connector_error
+                    .change_context(errors::ApiErrorResponse::InternalServerError)
+                    .attach_printable("Could not refresh access token"))
+            }
+        }
+    }?;
+
+    metrics::ACCESS_TOKEN_CREATION.add(
+        &metrics::CONTEXT,
+        1,
+        &[metrics::request::add_attributes(
+            "connector",
+            connector.connector_name.to_string(),
+        )],
+    );
+    Ok(access_token_router_data)
+}

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

@@ -258,7 +258,7 @@ pub async fn do_retry(
         key_store,
         req,
         &connector,
-        &mut payout_data,
+        payout_data,
     )
     .await
 }

openapi/openapi_spec.json

@@ -15607,7 +15607,8 @@
         "type": "string",
         "enum": [
           "adyen",
-          "wise"
+          "wise",
+          "paypal"
         ]
       },
       "PayoutCreateRequest": {