Store invalid messages in dead letter queue

Cargo.toml

@@ -3,5 +3,6 @@ members = [
   "packages-rs/drainpipe",
   "packages-rs/drainpipe-cli",
   "packages-rs/drainpipe-store",
+  "packages-rs/jetstream",
 ]
 resolver = "2"

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

@@ -5,6 +5,7 @@ use serde::{Deserialize, Serialize};
 use sled::Tree;
 
 pub struct Store {
+    db: sled::Db,
     cursor_tree: Tree,
     dead_letter_tree: Tree,
 }
@@ -50,18 +51,13 @@ impl DeadLetter {
             error_message,
         })
     }
-
-    pub fn key(&self) -> &String {
-        match &self.0 {
-            DeadLetterInner::V1 { key, .. } => key,
-        }
-    }
 }
 
 impl Store {
     pub fn open(path: &PathBuf) -> anyhow::Result<Store> {
         let db = sled::open(path)?;
         Ok(Self {
+            db: db.clone(),
             cursor_tree: db.open_tree("cursor")?,
             dead_letter_tree: db.open_tree("dead_letter")?,
         })
@@ -92,9 +88,27 @@ impl Store {
             .transpose()
     }
 
-    pub fn record_dead_letter(&self, dead_letter: &DeadLetter) -> anyhow::Result<()> {
-        self.dead_letter_tree
-            .insert(dead_letter.key(), bincode::serialize(&dead_letter)?)?;
+    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/Cargo.toml

@@ -6,21 +6,16 @@ edition = "2021"
 [dependencies]
 anyhow = "1.0.93"
 atrium-api = "0.24.7"
-bincode = "1.3.3"
-chrono = { version = "0.4.38", features = ["serde"] }
+chrono = "0.4.38"
 dotenv-flow = "0.16.2"
 drainpipe-store = { path = "../drainpipe-store" }
 env_logger = "0.11.5"
-flume = "0.11.1"
 futures-util = "0.3"
+jetstream = { path = "../jetstream" }
 log = "0.4.22"
 reqwest = { version = "0.12.4", features = ["json"] }
 serde = { version = "1.0.215", features = ["derive"] }
 serde_json = "1.0.132"
 sled = "0.34.7"
-thiserror = "2.0.3"
 tokio = { version = "1.38.0", features = ["full"] }
 tokio-metrics = "0.3.1"
-tokio-tungstenite = { version = "0.24.0", features = ["native-tls"] }
-url = "2.5.3"
-zstd = "0.13.2"

packages-rs/drainpipe/src/main.rs

@@ -1,5 +1,4 @@
 mod config;
-mod jetstream;
 
 use chrono::{TimeZone, Utc};
 use config::Config;
@@ -76,11 +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(&drainpipe_store::DeadLetter::new(
-                                        commit.info().time_us.to_string(),
-                                        serde_json::to_string(commit)?,
-                                        e.to_string(),
-                                    ))
+                                    store.record_dead_letter_commit(&commit, e.to_string())
                                 })?
                             }
 
@@ -92,7 +87,7 @@ async fn main() -> anyhow::Result<()> {
                 }
 
                 Ok(Err(e)) => {
-                    // TODO: This should add a dead letter
+                    store.record_dead_letter_jetstream_error(&e)?;
                     log::error!(
                         "Error receiving event (possible junk event structure?): {:?}",
                         e

packages-rs/jetstream/Cargo.toml

@@ -0,0 +1,18 @@
+[package]
+name = "jetstream"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+atrium-api = "0.24.8"
+chrono = { version = "0.4.38", features = ["serde"] }
+flume = "0.11.1"
+futures-util = "0.3.31"
+log = "0.4.22"
+serde = { version = "1.0.215", features = ["derive"] }
+serde_json = "1.0.132"
+thiserror = "2.0.3"
+tokio = { version = "1.38.0", features = ["full"] }
+tokio-tungstenite = { version = "0.24.0", features = ["native-tls"] }
+url = "2.5.4"
+zstd = "0.13.2"

packages-rs/jetstream/README.md

@@ -0,0 +1,5 @@
+# Jetstream Client
+
+A client for [jetstream](https://github.com/bluesky-social/jetstream).
+
+This started as a fork of [jetstream-oxide](https://github.com/videah/jetstream-oxide) but has since diverged with a few breaking changes and critical fixes.