refactor(drainer): change logic for trimming the stream and refactor for modularity

crates/diesel_models/src/kv.rs

@@ -7,16 +7,50 @@ use crate::{
     payment_attempt::{PaymentAttempt, PaymentAttemptNew, PaymentAttemptUpdate},
     payment_intent::{PaymentIntentNew, PaymentIntentUpdate},
     refund::{Refund, RefundNew, RefundUpdate},
-    reverse_lookup::ReverseLookupNew,
-    PaymentIntent,
+    reverse_lookup::{ReverseLookup, ReverseLookupNew},
+    PaymentIntent, PgPooledConn,
 };
 
 #[derive(Debug, Serialize, Deserialize)]
 #[serde(rename_all = "snake_case", tag = "db_op", content = "data")]
 pub enum DBOperation {
     Insert { insertable: Insertable },
     Update { updatable: Updateable },
-    Delete,
+}
+
+impl DBOperation {
+    pub fn operation<'a>(&self) -> &'a str {
+        match self {
+            Self::Insert { .. } => "insert",
+            Self::Update { .. } => "update",
+        }
+    }
+    pub fn table<'a>(&self) -> &'a str {
+        match self {
+            Self::Insert { insertable } => match insertable {
+                Insertable::PaymentIntent(_) => "payment_intent",
+                Insertable::PaymentAttempt(_) => "payment_attempt",
+                Insertable::Refund(_) => "refund",
+                Insertable::Address(_) => "address",
+                Insertable::ReverseLookUp(_) => "reverse_lookup",
+            },
+            Self::Update { updatable } => match updatable {
+                Updateable::PaymentIntentUpdate(_) => "payment_intent",
+                Updateable::PaymentAttemptUpdate(_) => "payment_attempt",
+                Updateable::RefundUpdate(_) => "refund",
+                Updateable::AddressUpdate(_) => "address",
+            },
+        }
+    }
+}
+
+#[derive(Debug)]
+pub enum DBResult {
+    PaymentIntent(Box<PaymentIntent>),
+    PaymentAttempt(Box<PaymentAttempt>),
+    Refund(Box<Refund>),
+    Address(Box<Address>),
+    ReverseLookUp(Box<ReverseLookup>),
 }
 
 #[derive(Debug, Serialize, Deserialize)]
@@ -25,6 +59,40 @@ pub struct TypedSql {
     pub op: DBOperation,
 }
 
+impl DBOperation {
+    pub async fn execute(self, conn: &PgPooledConn) -> crate::StorageResult<DBResult> {
+        Ok(match self {
+            Self::Insert { insertable } => match insertable {
+                Insertable::PaymentIntent(a) => {
+                    DBResult::PaymentIntent(Box::new(a.insert(conn).await?))
+                }
+                Insertable::PaymentAttempt(a) => {
+                    DBResult::PaymentAttempt(Box::new(a.insert(conn).await?))
+                }
+                Insertable::Refund(a) => DBResult::Refund(Box::new(a.insert(conn).await?)),
+                Insertable::Address(addr) => DBResult::Address(Box::new(addr.insert(conn).await?)),
+                Insertable::ReverseLookUp(rev) => {
+                    DBResult::ReverseLookUp(Box::new(rev.insert(conn).await?))
+                }
+            },
+            Self::Update { updatable } => match updatable {
+                Updateable::PaymentIntentUpdate(a) => {
+                    DBResult::PaymentIntent(Box::new(a.orig.update(conn, a.update_data).await?))
+                }
+                Updateable::PaymentAttemptUpdate(a) => DBResult::PaymentAttempt(Box::new(
+                    a.orig.update_with_attempt_id(conn, a.update_data).await?,
+                )),
+                Updateable::RefundUpdate(a) => {
+                    DBResult::Refund(Box::new(a.orig.update(conn, a.update_data).await?))
+                }
+                Updateable::AddressUpdate(a) => {
+                    DBResult::Address(Box::new(a.orig.update(conn, a.update_data).await?))
+                }
+            },
+        })
+    }
+}
+
 impl TypedSql {
     pub fn to_field_value_pairs(
         &self,

crates/drainer/Cargo.toml

@@ -25,6 +25,7 @@ serde_json = "1.0.108"
 serde_path_to_error = "0.1.14"
 thiserror = "1.0.40"
 tokio = { version = "1.28.2", features = ["macros", "rt-multi-thread"] }
+async-trait = "0.1.74"
 
 # First Party Crates
 common_utils = { version = "0.1.0", path = "../common_utils", features = ["signals"] }

crates/drainer/src/errors.rs

@@ -11,6 +11,8 @@ pub enum DrainerError {
     ConfigurationError(config::ConfigError),
     #[error("Error while configuring signals: {0}")]
     SignalError(String),
+    #[error("Error while parsing data from the stream: {0:?}")]
+    ParsingError(error_stack::Report<common_utils::errors::ParsingError>),
     #[error("Unexpected error occurred: {0}")]
     UnexpectedError(String),
 }