feat(core): Create Payout Webhook Flow

crates/api_models/src/webhooks.rs

@@ -3,6 +3,8 @@ use serde::{Deserialize, Serialize};
 use time::PrimitiveDateTime;
 use utoipa::ToSchema;
 
+#[cfg(feature = "payouts")]
+use crate::payouts;
 use crate::{disputes, enums as api_enums, mandates, payments, refunds};
 
 #[derive(Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize, Copy)]
@@ -39,10 +41,15 @@ pub enum IncomingWebhookEvent {
     MandateRevoked,
     EndpointVerification,
     ExternalAuthenticationARes,
+    PayoutSuccess,
+    PayoutFailure,
+    PayoutProcessing,
+    PayoutCancelled,
 }
 
 pub enum WebhookFlow {
     Payment,
+    Payout,
     Refund,
     Dispute,
     Subscription,
@@ -59,6 +66,10 @@ pub enum WebhookResponseTracker {
         payment_id: String,
         status: common_enums::IntentStatus,
     },
+    Payout {
+        payout_id: String,
+        status: common_enums::PayoutStatus,
+    },
     Refund {
         payment_id: String,
         refund_id: String,
@@ -82,7 +93,7 @@ impl WebhookResponseTracker {
             Self::Payment { payment_id, .. }
             | Self::Refund { payment_id, .. }
             | Self::Dispute { payment_id, .. } => Some(payment_id.to_string()),
-            Self::NoEffect | Self::Mandate { .. } => None,
+            Self::NoEffect | Self::Payout { .. } | Self::Mandate { .. } => None,
         }
     }
 }
@@ -119,6 +130,10 @@ impl From<IncomingWebhookEvent> for WebhookFlow {
             IncomingWebhookEvent::SourceChargeable
             | IncomingWebhookEvent::SourceTransactionCreated => Self::BankTransfer,
             IncomingWebhookEvent::ExternalAuthenticationARes => Self::ExternalAuthentication,
+            IncomingWebhookEvent::PayoutSuccess
+            | IncomingWebhookEvent::PayoutFailure
+            | IncomingWebhookEvent::PayoutProcessing
+            | IncomingWebhookEvent::PayoutCancelled => Self::Payout,
         }
     }
 }
@@ -143,12 +158,19 @@ pub enum AuthenticationIdType {
     ConnectorAuthenticationId(String),
 }
 
+#[derive(Clone)]
+pub enum PayoutIdType {
+    PayoutAttemptId(String),
+    ConnectorPayoutId(String),
+}
+
 #[derive(Clone)]
 pub enum ObjectReferenceId {
     PaymentId(payments::PaymentIdType),
     RefundId(RefundIdType),
     MandateId(MandateIdType),
     ExternalAuthenticationID(AuthenticationIdType),
+    PayoutId(PayoutIdType),
 }
 
 pub struct IncomingWebhookDetails {
@@ -187,6 +209,9 @@ pub enum OutgoingWebhookContent {
     DisputeDetails(Box<disputes::DisputeResponse>),
     #[schema(value_type = MandateResponse, title = "MandateResponse")]
     MandateDetails(Box<mandates::MandateResponse>),
+    #[cfg(feature = "payouts")]
+    #[schema(value_type = PayoutCreateResponse, title = "PayoutCreateResponse")]
+    PayoutDetails(payouts::PayoutCreateResponse),
 }
 
 #[derive(Debug, Clone, Serialize)]

crates/common_enums/src/enums.rs

@@ -1069,6 +1069,7 @@ pub enum EventClass {
     Refunds,
     Disputes,
     Mandates,
+    Payouts,
 }
 
 #[derive(
@@ -1107,6 +1108,10 @@ pub enum EventType {
     DisputeLost,
     MandateActive,
     MandateRevoked,
+    PayoutSuccess,
+    PayoutFailed,
+    PayoutProcessing,
+    PayoutCancelled,
 }
 
 #[derive(

crates/diesel_models/src/enums.rs

@@ -68,6 +68,7 @@ pub enum EventObjectType {
     RefundDetails,
     DisputeDetails,
     MandateDetails,
+    PayoutDetails,
 }
 
 #[derive(

crates/diesel_models/src/payout_attempt.rs

@@ -14,7 +14,7 @@ pub struct PayoutAttempt {
     pub merchant_id: String,
     pub address_id: String,
     pub connector: Option<String>,
-    pub connector_payout_id: String,
+    pub connector_payout_id: Option<String>,
     pub payout_token: Option<String>,
     pub status: storage_enums::PayoutStatus,
     pub is_eligible: Option<bool>,
@@ -51,7 +51,7 @@ pub struct PayoutAttemptNew {
     pub merchant_id: String,
     pub address_id: String,
     pub connector: Option<String>,
-    pub connector_payout_id: String,
+    pub connector_payout_id: Option<String>,
     pub payout_token: Option<String>,
     pub status: storage_enums::PayoutStatus,
     pub is_eligible: Option<bool>,
@@ -71,7 +71,7 @@ pub struct PayoutAttemptNew {
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub enum PayoutAttemptUpdate {
     StatusUpdate {
-        connector_payout_id: String,
+        connector_payout_id: Option<String>,
         status: storage_enums::PayoutStatus,
         error_message: Option<String>,
         error_code: Option<String>,
@@ -138,7 +138,7 @@ impl From<PayoutAttemptUpdate> for PayoutAttemptUpdateInternal {
                 error_code,
                 is_eligible,
             } => Self {
-                connector_payout_id: Some(connector_payout_id),
+                connector_payout_id,
                 status: Some(status),
                 error_message,
                 error_code,
@@ -182,7 +182,7 @@ impl PayoutAttemptUpdate {
         } = self.into();
         PayoutAttempt {
             payout_token: payout_token.or(source.payout_token),
-            connector_payout_id: connector_payout_id.unwrap_or(source.connector_payout_id),
+            connector_payout_id: connector_payout_id.or(source.connector_payout_id),
             status: status.unwrap_or(source.status),
             error_message: error_message.or(source.error_message),
             error_code: error_code.or(source.error_code),

crates/diesel_models/src/query/payout_attempt.rs

@@ -81,6 +81,20 @@ impl PayoutAttempt {
         .await
     }
 
+    pub async fn find_by_merchant_id_connector_payout_id(
+        conn: &PgPooledConn,
+        merchant_id: &str,
+        connector_payout_id: &str,
+    ) -> StorageResult<Self> {
+        generics::generic_find_one::<<Self as HasTable>::Table, _, _>(
+            conn,
+            dsl::merchant_id
+                .eq(merchant_id.to_owned())
+                .and(dsl::connector_payout_id.eq(connector_payout_id.to_owned())),
+        )
+        .await
+    }
+
     pub async fn update_by_merchant_id_payout_id(
         conn: &PgPooledConn,
         merchant_id: &str,

crates/diesel_models/src/schema.rs

@@ -949,7 +949,7 @@ diesel::table! {
         #[max_length = 64]
         connector -> Nullable<Varchar>,
         #[max_length = 128]
-        connector_payout_id -> Varchar,
+        connector_payout_id -> Nullable<Varchar>,
         #[max_length = 64]
         payout_token -> Nullable<Varchar>,
         status -> PayoutStatus,

crates/hyperswitch_domain_models/src/payouts/payout_attempt.rs

@@ -31,6 +31,13 @@ pub trait PayoutAttemptInterface {
         _storage_scheme: MerchantStorageScheme,
     ) -> error_stack::Result<PayoutAttempt, errors::StorageError>;
 
+    async fn find_payout_attempt_by_merchant_id_connector_payout_id(
+        &self,
+        _merchant_id: &str,
+        _connector_payout_id: &str,
+        _storage_scheme: MerchantStorageScheme,
+    ) -> error_stack::Result<PayoutAttempt, errors::StorageError>;
+
     async fn get_filters_for_payouts(
         &self,
         payout: &[Payouts],
@@ -55,7 +62,7 @@ pub struct PayoutAttempt {
     pub merchant_id: String,
     pub address_id: String,
     pub connector: Option<String>,
-    pub connector_payout_id: String,
+    pub connector_payout_id: Option<String>,
     pub payout_token: Option<String>,
     pub status: storage_enums::PayoutStatus,
     pub is_eligible: Option<bool>,
@@ -80,7 +87,7 @@ pub struct PayoutAttemptNew {
     pub merchant_id: String,
     pub address_id: String,
     pub connector: Option<String>,
-    pub connector_payout_id: String,
+    pub connector_payout_id: Option<String>,
     pub payout_token: Option<String>,
     pub status: storage_enums::PayoutStatus,
     pub is_eligible: Option<bool>,
@@ -106,7 +113,7 @@ impl Default for PayoutAttemptNew {
             merchant_id: String::default(),
             address_id: String::default(),
             connector: None,
-            connector_payout_id: String::default(),
+            connector_payout_id: Some(String::default()),
             payout_token: None,
             status: storage_enums::PayoutStatus::default(),
             is_eligible: None,
@@ -123,10 +130,10 @@ impl Default for PayoutAttemptNew {
     }
 }
 
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub enum PayoutAttemptUpdate {
     StatusUpdate {
-        connector_payout_id: String,
+        connector_payout_id: Option<String>,
         status: storage_enums::PayoutStatus,
         error_message: Option<String>,
         error_code: Option<String>,
@@ -173,7 +180,7 @@ impl From<PayoutAttemptUpdate> for PayoutAttemptUpdateInternal {
                 error_code,
                 is_eligible,
             } => Self {
-                connector_payout_id: Some(connector_payout_id),
+                connector_payout_id,
                 status: Some(status),
                 error_message,
                 error_code,

crates/router/src/compatibility/stripe/webhooks.rs

@@ -2,7 +2,12 @@ use api_models::{
     enums::{DisputeStatus, MandateStatus},
     webhooks::{self as api},
 };
-use common_utils::{crypto::SignMessage, date_time, ext_traits::Encode};
+use common_utils::{
+    crypto::{Encryptable, SignMessage},
+    date_time,
+    ext_traits::Encode,
+    pii::{self, Email},
+};
 use error_stack::ResultExt;
 use router_env::logger;
 use serde::Serialize;
@@ -81,6 +86,7 @@ pub enum StripeWebhookObject {
     Refund(StripeRefundResponse),
     Dispute(StripeDisputeResponse),
     Mandate(StripeMandateResponse),
+    Payout(StripePayoutResponse),
 }
 
 #[derive(Serialize, Debug)]
@@ -101,6 +107,72 @@ pub struct StripeMandateResponse {
     pub payment_method: String,
 }
 
+#[derive(Clone, Serialize, Debug)]
+pub struct StripePayoutResponse {
+    pub id: String,
+    pub amount: i64,
+    pub currency: String,
+    pub payout_type: common_enums::PayoutType,
+    pub status: StripePayoutStatus,
+    pub name: Option<masking::Secret<String>>,
+    pub email: Option<Email>,
+    pub phone: Option<masking::Secret<String>>,
+    pub phone_country_code: Option<String>,
+    pub created: Option<i64>,
+    pub metadata: Option<pii::SecretSerdeValue>,
+    pub entity_type: common_enums::PayoutEntityType,
+    pub recurring: bool,
+    pub error_message: Option<String>,
+    pub error_code: Option<String>,
+}
+
+#[derive(Clone, Serialize, Debug)]
+#[serde(rename_all = "snake_case")]
+pub enum StripePayoutStatus {
+    PayoutSuccess,
+    PayoutFailure,
+    PayoutProcessing,
+    PayoutCancelled,
+}
+
+impl From<common_enums::PayoutStatus> for StripePayoutStatus {
+    fn from(status: common_enums::PayoutStatus) -> Self {
+        match status {
+            common_enums::PayoutStatus::Success => Self::PayoutSuccess,
+            common_enums::PayoutStatus::Failed => Self::PayoutFailure,
+            common_enums::PayoutStatus::Cancelled => Self::PayoutCancelled,
+            common_enums::PayoutStatus::Pending
+            | common_enums::PayoutStatus::Ineligible
+            | common_enums::PayoutStatus::RequiresCreation
+            | common_enums::PayoutStatus::RequiresFulfillment
+            | common_enums::PayoutStatus::RequiresPayoutMethodData
+            | common_enums::PayoutStatus::RequiresVendorAccountCreation => Self::PayoutProcessing,
+        }
+    }
+}
+
+impl From<api_models::payouts::PayoutCreateResponse> for StripePayoutResponse {
+    fn from(res: api_models::payouts::PayoutCreateResponse) -> Self {
+        Self {
+            id: res.payout_id,
+            amount: res.amount,
+            currency: res.currency.to_string(),
+            payout_type: res.payout_type,
+            status: StripePayoutStatus::from(res.status),
+            name: res.name.map(Encryptable::into_inner),
+            email: res.email.map(|inner| inner.into()),
+            phone: res.phone.map(Encryptable::into_inner),
+            phone_country_code: res.phone_country_code,
+            created: res.created.map(|t| t.assume_utc().unix_timestamp()),
+            metadata: res.metadata,
+            entity_type: res.entity_type,
+            recurring: res.recurring,
+            error_message: res.error_message,
+            error_code: res.error_code,
+        }
+    }
+}
+
 #[derive(Serialize, Debug)]
 #[serde(rename_all = "snake_case")]
 pub enum StripeMandateStatus {
@@ -197,6 +269,10 @@ fn get_stripe_event_type(event_type: api_models::enums::EventType) -> &'static s
         }
         // stripe treats partially captured payments as succeeded.
         api_models::enums::EventType::PaymentCaptured => "payment_intent.succeeded",
+        api_models::enums::EventType::PayoutSuccess => "payout.paid",
+        api_models::enums::EventType::PayoutFailed => "payout.failed",
+        api_models::enums::EventType::PayoutCancelled => "payout.canceled",
+        api_models::enums::EventType::PayoutProcessing => "payout.created",
     }
 }
 
@@ -237,6 +313,7 @@ impl From<api::OutgoingWebhookContent> for StripeWebhookObject {
             api::OutgoingWebhookContent::MandateDetails(mandate) => {
                 Self::Mandate((*mandate).into())
             }
+            api::OutgoingWebhookContent::PayoutDetails(payout) => Self::Payout(payout.into()),
         }
     }
 }

crates/router/src/connector/adyen.rs

@@ -1805,6 +1805,11 @@ impl api::IncomingWebhook for Adyen {
                 ),
             ));
         }
+        if adyen::is_payout_event(&notif.event_code) {
+            return Ok(api_models::webhooks::ObjectReferenceId::PayoutId(
+                api_models::webhooks::PayoutIdType::PayoutAttemptId(notif.merchant_reference),
+            ));
+        }
         Err(report!(errors::ConnectorError::WebhookReferenceIdNotFound))
     }