Abstract dlq inputs

packages-rs/drainpipe-store/Cargo.toml

@@ -6,6 +6,8 @@ edition = "2021"
 [dependencies]
 anyhow = "1.0.93"
 bincode = "1.3.3"
+jetstream = { path = "../jetstream" }
 log = "0.4.22"
 serde = { version = "1.0.215", features = ["derive"] }
+serde_json = "1.0.133"
 sled = "0.34.7"

packages-rs/drainpipe-store/src/lib.rs

@@ -94,16 +94,27 @@ impl Store {
             .transpose()
     }
 
-    pub fn record_dead_letter(
-        &self,
-        commit_json: String,
-        error_message: String,
-    ) -> anyhow::Result<()> {
+    fn record_dead_letter(&self, commit_json: String, error_message: String) -> anyhow::Result<()> {
         let key = self.db.generate_id()?.to_string();
         self.dead_letter_tree.insert(
             key.clone(),
             bincode::serialize(&DeadLetter::new(key, commit_json, error_message))?,
         )?;
         Ok(())
     }
+
+    pub fn record_dead_letter_commit(
+        &self,
+        commit: &jetstream::event::CommitEvent,
+        error_message: String,
+    ) -> anyhow::Result<()> {
+        self.record_dead_letter(serde_json::to_string(commit)?, error_message)
+    }
+
+    pub fn record_dead_letter_jetstream_error(
+        &self,
+        error: &jetstream::error::JetstreamEventError,
+    ) -> anyhow::Result<()> {
+        self.record_dead_letter("null".into(), error.to_string())
+    }
 }

packages-rs/drainpipe/src/main.rs

@@ -75,10 +75,7 @@ async fn main() -> anyhow::Result<()> {
 
                                 send_frontpage_commit(&config, commit).await.or_else(|e| {
                                     log::error!("Error processing commit: {:?}", e);
-                                    store.record_dead_letter(
-                                        serde_json::to_string(commit)?,
-                                        e.to_string(),
-                                    )
+                                    store.record_dead_letter_commit(&commit, e.to_string())
                                 })?
                             }
 
@@ -90,7 +87,7 @@ async fn main() -> anyhow::Result<()> {
                 }
 
                 Ok(Err(e)) => {
-                    store.record_dead_letter("null".into(), e.to_string())?;
+                    store.record_dead_letter_jetstream_error(&e)?;
                     log::error!(
                         "Error receiving event (possible junk event structure?): {:?}",
                         e