feat(kafka): Implement basic async kafka consumer (#17)

Co-authored-by: Evan Hicks <evanh@users.noreply.github.com>

Author: john-z-yang

.gitignore

@@ -2,3 +2,8 @@
 
 # Editors
 .DS_Store
+
+# Sqlite artifacts
+*.sqlite
+*.sqlite-shm
+*.sqlite-wal

Cargo.lock

@@ -6,21 +6,25 @@ edition = "2021"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-sentry_protos = "0.1.33"
+sentry_protos = "0.1.34"
+prost = "0.13"
+prost-types = "0.13.3"
 anyhow = "1.0.92"
 chrono = { version = "0.4.26" }
-sqlx = { version = "0.8.2", features = ["sqlite", "runtime-tokio", "chrono"] }
-prost = "0.13"
 tokio = { version = "1.41.0", features = ["full"] }
-prost-types = "0.13.3"
+tokio-util = "0.7.12"
+tokio-stream = { version = "0.1.16", features = ["full"] }
+async-stream = "0.3.5"
+futures = "0.3.31"
 rdkafka = { version = "0.36.2", features = ["cmake-build"] }
 serde = "1.0.214"
 serde_yaml = "0.9.34"
 figment = { version = "0.10.19", features = ["env", "yaml", "test"] }
+sqlx = { version = "0.8.2", features = ["sqlite", "runtime-tokio", "chrono"] }
 clap = { version = "4.5.20", features = ["derive"] }
 sentry = { version = "0.34.0", features = ["tracing"] }
-tracing-subscriber = { version = "0.3.18", features = ["json"] }
 tracing = "0.1.40"
+tracing-subscriber = { version = "0.3.18", features = ["json"] }
 metrics-exporter-statsd = "0.9.0"
 metrics = "0.24.0"
 

Cargo.toml

@@ -1,6 +1,7 @@
 CREATE TABLE IF NOT EXISTS inflight_taskactivations (
     id UUID NOT NULL PRIMARY KEY,
     activation BLOB NOT NULL,
+    partition INTEGER NOT NULL,
     offset BIGINTEGER NOT NULL,
     added_at DATETIME NOT NULL,
     deadletter_at DATETIME,

migrations/0001_create_inflight_taskactivations.sql

@@ -0,0 +1,33 @@
+use std::{sync::Arc, time::Duration};
+
+use anyhow::{anyhow, Error};
+use chrono::Utc;
+use prost::Message as _;
+use rdkafka::{message::OwnedMessage, Message};
+use sentry_protos::sentry::v1::TaskActivation;
+
+use crate::inflight_activation_store::{InflightActivation, TaskActivationStatus};
+
+pub struct Config {
+    pub deadletter_duration: Option<Duration>,
+}
+
+pub fn new(config: Config) -> impl Fn(Arc<OwnedMessage>) -> Result<InflightActivation, Error> {
+    move |msg: Arc<OwnedMessage>| {
+        let Some(payload) = msg.payload() else {
+            return Err(anyhow!("Message has no payload"));
+        };
+        let activation = TaskActivation::decode(payload)?;
+        Ok(InflightActivation {
+            activation,
+            status: TaskActivationStatus::Pending,
+            partition: msg.partition(),
+            offset: msg.offset(),
+            added_at: Utc::now(),
+            deadletter_at: config
+                .deadletter_duration
+                .map(|duration| Utc::now() + duration),
+            processing_deadline: None,
+        })
+    }
+}