refactor(storage): update paymentintent object to provide a relation with attempts (#2502)

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

Author: lsampras

Cargo.lock

@@ -23,3 +23,28 @@ pub enum MerchantStorageScheme {
     PostgresOnly,
     RedisKv,
 }
+
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub enum RemoteStorageObject<T: ForeignIDRef> {
+    ForeignID(String),
+    Object(T),
+}
+
+impl<T: ForeignIDRef> From<T> for RemoteStorageObject<T> {
+    fn from(value: T) -> Self {
+        Self::Object(value)
+    }
+}
+
+pub trait ForeignIDRef {
+    fn foreign_id(&self) -> String;
+}
+
+impl<T: ForeignIDRef> RemoteStorageObject<T> {
+    pub fn get_id(&self) -> String {
+        match self {
+            Self::ForeignID(id) => id.clone(),
+            Self::Object(i) => i.foreign_id(),
+        }
+    }
+}

crates/data_models/src/lib.rs

@@ -1,2 +1,50 @@
+use common_utils::pii;
+use time::PrimitiveDateTime;
+
 pub mod payment_attempt;
 pub mod payment_intent;
+
+use common_enums as storage_enums;
+
+use self::payment_attempt::PaymentAttempt;
+use crate::RemoteStorageObject;
+
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct PaymentIntent {
+    pub id: i32,
+    pub payment_id: String,
+    pub merchant_id: String,
+    pub status: storage_enums::IntentStatus,
+    pub amount: i64,
+    pub currency: Option<storage_enums::Currency>,
+    pub amount_captured: Option<i64>,
+    pub customer_id: Option<String>,
+    pub description: Option<String>,
+    pub return_url: Option<String>,
+    pub metadata: Option<pii::SecretSerdeValue>,
+    pub connector_id: Option<String>,
+    pub shipping_address_id: Option<String>,
+    pub billing_address_id: Option<String>,
+    pub statement_descriptor_name: Option<String>,
+    pub statement_descriptor_suffix: Option<String>,
+    pub created_at: PrimitiveDateTime,
+    pub modified_at: PrimitiveDateTime,
+    pub last_synced: Option<PrimitiveDateTime>,
+    pub setup_future_usage: Option<storage_enums::FutureUsage>,
+    pub off_session: Option<bool>,
+    pub client_secret: Option<String>,
+    pub active_attempt: RemoteStorageObject<PaymentAttempt>,
+    pub business_country: Option<storage_enums::CountryAlpha2>,
+    pub business_label: Option<String>,
+    pub order_details: Option<Vec<pii::SecretSerdeValue>>,
+    pub allowed_payment_method_types: Option<serde_json::Value>,
+    pub connector_metadata: Option<serde_json::Value>,
+    pub feature_metadata: Option<serde_json::Value>,
+    pub attempt_count: i16,
+    pub profile_id: Option<String>,
+    pub payment_link_id: Option<String>,
+    // Denotes the action(approve or reject) taken by merchant in case of manual review.
+    // Manual review can occur when the transaction is marked as risky by the frm_processor, payment processor or when there is underpayment/over payment incase of crypto payment
+    pub merchant_decision: Option<String>,
+    pub payment_confirm_source: Option<storage_enums::PaymentSource>,
+}

crates/data_models/src/payments.rs

@@ -3,8 +3,8 @@ use common_enums as storage_enums;
 use serde::{Deserialize, Serialize};
 use time::PrimitiveDateTime;
 
-use super::payment_intent::PaymentIntent;
-use crate::{errors, mandates::MandateDataType, MerchantStorageScheme};
+use super::PaymentIntent;
+use crate::{errors, mandates::MandateDataType, ForeignIDRef, MerchantStorageScheme};
 
 #[async_trait::async_trait]
 pub trait PaymentAttemptInterface {
@@ -314,3 +314,9 @@ pub enum PaymentAttemptUpdate {
         surcharge_metadata: Option<serde_json::Value>,
     },
 }
+
+impl ForeignIDRef for PaymentAttempt {
+    fn foreign_id(&self) -> String {
+        self.attempt_id.clone()
+    }
+}

crates/data_models/src/payments/payment_attempt.rs

@@ -6,7 +6,8 @@ use common_utils::{
 use serde::{Deserialize, Serialize};
 use time::PrimitiveDateTime;
 
-use crate::{errors, MerchantStorageScheme};
+use super::{payment_attempt::PaymentAttempt, PaymentIntent};
+use crate::{errors, MerchantStorageScheme, RemoteStorageObject};
 #[async_trait::async_trait]
 pub trait PaymentIntentInterface {
     async fn update_payment_intent(
@@ -29,6 +30,12 @@ pub trait PaymentIntentInterface {
         storage_scheme: MerchantStorageScheme,
     ) -> error_stack::Result<PaymentIntent, errors::StorageError>;
 
+    async fn get_active_payment_attempt(
+        &self,
+        payment: &mut PaymentIntent,
+        storage_scheme: MerchantStorageScheme,
+    ) -> error_stack::Result<PaymentAttempt, errors::StorageError>;
+
     #[cfg(feature = "olap")]
     async fn filter_payment_intent_by_constraints(
         &self,
@@ -51,10 +58,7 @@ pub trait PaymentIntentInterface {
         merchant_id: &str,
         constraints: &PaymentIntentFetchConstraints,
         storage_scheme: MerchantStorageScheme,
-    ) -> error_stack::Result<
-        Vec<(PaymentIntent, super::payment_attempt::PaymentAttempt)>,
-        errors::StorageError,
-    >;
+    ) -> error_stack::Result<Vec<(PaymentIntent, PaymentAttempt)>, errors::StorageError>;
 
     #[cfg(feature = "olap")]
     async fn get_filtered_active_attempt_ids_for_total_count(
@@ -65,50 +69,7 @@ pub trait PaymentIntentInterface {
     ) -> error_stack::Result<Vec<String>, errors::StorageError>;
 }
 
-#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
-pub struct PaymentIntent {
-    pub id: i32,
-    pub payment_id: String,
-    pub merchant_id: String,
-    pub status: storage_enums::IntentStatus,
-    pub amount: i64,
-    pub currency: Option<storage_enums::Currency>,
-    pub amount_captured: Option<i64>,
-    pub customer_id: Option<String>,
-    pub description: Option<String>,
-    pub return_url: Option<String>,
-    pub metadata: Option<pii::SecretSerdeValue>,
-    pub connector_id: Option<String>,
-    pub shipping_address_id: Option<String>,
-    pub billing_address_id: Option<String>,
-    pub statement_descriptor_name: Option<String>,
-    pub statement_descriptor_suffix: Option<String>,
-    #[serde(with = "common_utils::custom_serde::iso8601")]
-    pub created_at: PrimitiveDateTime,
-    #[serde(with = "common_utils::custom_serde::iso8601")]
-    pub modified_at: PrimitiveDateTime,
-    #[serde(default, with = "common_utils::custom_serde::iso8601::option")]
-    pub last_synced: Option<PrimitiveDateTime>,
-    pub setup_future_usage: Option<storage_enums::FutureUsage>,
-    pub off_session: Option<bool>,
-    pub client_secret: Option<String>,
-    pub active_attempt_id: String,
-    pub business_country: Option<storage_enums::CountryAlpha2>,
-    pub business_label: Option<String>,
-    pub order_details: Option<Vec<pii::SecretSerdeValue>>,
-    pub allowed_payment_method_types: Option<serde_json::Value>,
-    pub connector_metadata: Option<serde_json::Value>,
-    pub feature_metadata: Option<serde_json::Value>,
-    pub attempt_count: i16,
-    pub profile_id: Option<String>,
-    // Denotes the action(approve or reject) taken by merchant in case of manual review.
-    // Manual review can occur when the transaction is marked as risky by the frm_processor, payment processor or when there is underpayment/over payment incase of crypto payment
-    pub merchant_decision: Option<String>,
-    pub payment_link_id: Option<String>,
-    pub payment_confirm_source: Option<storage_enums::PaymentSource>,
-}
-
-#[derive(Clone, Debug, Default, Eq, PartialEq, Serialize, Deserialize)]
+#[derive(Clone, Debug, Eq, PartialEq)]
 pub struct PaymentIntentNew {
     pub payment_id: String,
     pub merchant_id: String,
@@ -125,16 +86,13 @@ pub struct PaymentIntentNew {
     pub billing_address_id: Option<String>,
     pub statement_descriptor_name: Option<String>,
     pub statement_descriptor_suffix: Option<String>,
-    #[serde(default, with = "common_utils::custom_serde::iso8601::option")]
     pub created_at: Option<PrimitiveDateTime>,
-    #[serde(default, with = "common_utils::custom_serde::iso8601::option")]
     pub modified_at: Option<PrimitiveDateTime>,
-    #[serde(default, with = "common_utils::custom_serde::iso8601::option")]
     pub last_synced: Option<PrimitiveDateTime>,
     pub setup_future_usage: Option<storage_enums::FutureUsage>,
     pub off_session: Option<bool>,
     pub client_secret: Option<String>,
-    pub active_attempt_id: String,
+    pub active_attempt: RemoteStorageObject<PaymentAttempt>,
     pub business_country: Option<storage_enums::CountryAlpha2>,
     pub business_label: Option<String>,
     pub order_details: Option<Vec<pii::SecretSerdeValue>>,

crates/data_models/src/payments/payment_intent.rs

@@ -6,9 +6,10 @@ use crate::{
     connector_response::{ConnectorResponse, ConnectorResponseNew, ConnectorResponseUpdate},
     errors,
     payment_attempt::{PaymentAttempt, PaymentAttemptNew, PaymentAttemptUpdate},
-    payment_intent::{PaymentIntent, PaymentIntentNew, PaymentIntentUpdate},
+    payment_intent::{PaymentIntentNew, PaymentIntentUpdate},
     refund::{Refund, RefundNew, RefundUpdate},
     reverse_lookup::ReverseLookupNew,
+    PaymentIntent,
 };
 
 #[derive(Debug, Serialize, Deserialize)]

crates/diesel_models/src/kv.rs

@@ -165,7 +165,6 @@ pub enum PaymentIntentUpdate {
 
 #[derive(Clone, Debug, Default, AsChangeset, router_derive::DebugAsDisplay)]
 #[diesel(table_name = payment_intent)]
-
 pub struct PaymentIntentUpdateInternal {
     pub amount: Option<i64>,
     pub currency: Option<storage_enums::Currency>,

crates/diesel_models/src/payment_intent.rs

@@ -15,10 +15,9 @@ use crate::{
     payment_attempt::{
         PaymentAttempt, PaymentAttemptNew, PaymentAttemptUpdate, PaymentAttemptUpdateInternal,
     },
-    payment_intent::PaymentIntent,
     query::generics::db_metrics,
     schema::payment_attempt::dsl,
-    PgPooledConn, StorageResult,
+    PaymentIntent, PgPooledConn, StorageResult,
 };
 
 impl PaymentAttemptNew {

crates/diesel_models/src/query/payment_attempt.rs

@@ -122,7 +122,8 @@ async fn drainer_handler(
     active_tasks.fetch_add(1, atomic::Ordering::Release);
 
     let stream_name = utils::get_drainer_stream_name(store.clone(), stream_index);
-    let drainer_result = drainer(store.clone(), max_read_count, stream_name.as_str()).await;
+    let drainer_result =
+        Box::pin(drainer(store.clone(), max_read_count, stream_name.as_str())).await;
 
     if let Err(error) = drainer_result {
         logger::error!(?error)

crates/drainer/src/lib.rs

@@ -115,7 +115,7 @@ impl ConnectorCommon for Trustpay {
 
         match response {
             Ok(response_data) => {
-                let error_list = response_data.errors.clone().unwrap_or(vec![]);
+                let error_list = response_data.errors.clone().unwrap_or_default();
                 let option_error_code_message = get_error_code_error_message_based_on_priority(
                     self.clone(),
                     error_list.into_iter().map(|errors| errors.into()).collect(),