feat: implement SQLite and Postgres event queue adapters
- Added SQLite and Postgres event queue implementations with migrations and payload structures. - Created migration scripts for both SQLite and Postgres event queues. - Implemented event publishing and consumption logic for both adapters. - Added serialization and deserialization for domain events to database payloads. - Updated presentation and worker crates to support new event queue features. - Refactored event handling to utilize the new database-backed event queues.
This commit is contained in:
@@ -1,14 +1,236 @@
|
||||
pub fn add(left: u64, right: u64) -> u64 {
|
||||
left + right
|
||||
mod migrations;
|
||||
mod payload;
|
||||
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
use async_trait::async_trait;
|
||||
use domain::{
|
||||
errors::DomainError,
|
||||
events::{AckHandle, DomainEvent, EventEnvelope},
|
||||
ports::{EventConsumer, EventPublisher},
|
||||
};
|
||||
use futures::stream::{self, BoxStream};
|
||||
use sqlx::SqlitePool;
|
||||
use tokio::sync::{Mutex, mpsc};
|
||||
|
||||
use payload::DbEventPayload;
|
||||
|
||||
pub struct DbEventQueueConfig {
|
||||
pub poll_interval_ms: u64,
|
||||
pub batch_size: i64,
|
||||
pub max_attempts: i32,
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn it_works() {
|
||||
let result = add(2, 2);
|
||||
assert_eq!(result, 4);
|
||||
impl DbEventQueueConfig {
|
||||
pub fn from_env() -> Self {
|
||||
Self {
|
||||
poll_interval_ms: std::env::var("EVENT_QUEUE_POLL_INTERVAL_MS")
|
||||
.ok().and_then(|v| v.parse().ok()).unwrap_or(500),
|
||||
batch_size: std::env::var("EVENT_QUEUE_BATCH_SIZE")
|
||||
.ok().and_then(|v| v.parse().ok()).unwrap_or(10),
|
||||
max_attempts: std::env::var("EVENT_QUEUE_MAX_ATTEMPTS")
|
||||
.ok().and_then(|v| v.parse().ok()).unwrap_or(5),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct SqliteEventQueue {
|
||||
pool: SqlitePool,
|
||||
config: Arc<DbEventQueueConfig>,
|
||||
}
|
||||
|
||||
impl SqliteEventQueue {
|
||||
pub async fn create(pool: SqlitePool, config: DbEventQueueConfig) -> anyhow::Result<Self> {
|
||||
migrations::run(&pool).await?;
|
||||
Ok(Self { pool, config: Arc::new(config) })
|
||||
}
|
||||
|
||||
pub async fn create_publisher(pool: SqlitePool) -> anyhow::Result<Arc<dyn EventPublisher>> {
|
||||
let q = Self::create(pool, DbEventQueueConfig::from_env()).await?;
|
||||
Ok(Arc::new(q))
|
||||
}
|
||||
|
||||
pub async fn create_channel(
|
||||
pool: SqlitePool,
|
||||
) -> anyhow::Result<(Arc<dyn EventPublisher>, Arc<dyn EventConsumer>)> {
|
||||
let q = Self::create(pool, DbEventQueueConfig::from_env()).await?;
|
||||
Ok((Arc::new(q.clone()), Arc::new(q)))
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl EventPublisher for SqliteEventQueue {
|
||||
async fn publish(&self, event: &DomainEvent) -> Result<(), DomainError> {
|
||||
let db_payload = DbEventPayload::from(event);
|
||||
let event_type = db_payload.event_type();
|
||||
let payload_json = serde_json::to_string(&db_payload)
|
||||
.map_err(|e| DomainError::InfrastructureError(format!("serialize: {e}")))?;
|
||||
|
||||
sqlx::query(
|
||||
"INSERT INTO event_queue (event_type, payload) VALUES (?, ?)"
|
||||
)
|
||||
.bind(event_type)
|
||||
.bind(payload_json)
|
||||
.execute(&self.pool)
|
||||
.await
|
||||
.map_err(|e| DomainError::InfrastructureError(format!("insert event: {e}")))?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
impl EventConsumer for SqliteEventQueue {
|
||||
fn consume(&self) -> BoxStream<'_, Result<EventEnvelope, DomainError>> {
|
||||
let pool = self.pool.clone();
|
||||
let config = Arc::clone(&self.config);
|
||||
let (tx, rx) = mpsc::channel(128);
|
||||
let rx = Arc::new(Mutex::new(rx));
|
||||
|
||||
tokio::spawn(async move {
|
||||
let poll_interval = Duration::from_millis(config.poll_interval_ms);
|
||||
loop {
|
||||
match claim_batch(&pool, &config).await {
|
||||
Err(e) => {
|
||||
tracing::error!("sqlite event queue claim error: {e}");
|
||||
tokio::time::sleep(poll_interval).await;
|
||||
}
|
||||
Ok(rows) if rows.is_empty() => {
|
||||
tokio::time::sleep(poll_interval).await;
|
||||
}
|
||||
Ok(rows) => {
|
||||
for row in rows {
|
||||
let envelope = decode_row(&pool, row, config.max_attempts);
|
||||
if tx.send(envelope).await.is_err() {
|
||||
tracing::info!("sqlite event queue consumer closed");
|
||||
return;
|
||||
}
|
||||
}
|
||||
// no sleep — re-poll immediately when batch was non-empty
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
Box::pin(stream::unfold(rx, |rx| async move {
|
||||
let item = rx.lock().await.recv().await?;
|
||||
Some((item, rx))
|
||||
}))
|
||||
}
|
||||
}
|
||||
|
||||
// ── Internal types ────────────────────────────────────────────────────────────
|
||||
|
||||
#[derive(sqlx::FromRow)]
|
||||
struct QueueRow {
|
||||
id: i64,
|
||||
payload: String,
|
||||
attempts: i32,
|
||||
}
|
||||
|
||||
async fn claim_batch(
|
||||
pool: &SqlitePool,
|
||||
config: &DbEventQueueConfig,
|
||||
) -> Result<Vec<QueueRow>, DomainError> {
|
||||
let mut tx = pool.begin().await
|
||||
.map_err(|e| DomainError::InfrastructureError(format!("begin tx: {e}")))?;
|
||||
|
||||
let rows = sqlx::query_as::<_, QueueRow>(
|
||||
"SELECT id, payload, attempts FROM event_queue
|
||||
WHERE status = 'pending'
|
||||
AND next_attempt_at <= strftime('%Y-%m-%dT%H:%M:%SZ', 'now')
|
||||
ORDER BY next_attempt_at ASC
|
||||
LIMIT ?"
|
||||
)
|
||||
.bind(config.batch_size)
|
||||
.fetch_all(&mut *tx)
|
||||
.await
|
||||
.map_err(|e| DomainError::InfrastructureError(format!("select pending: {e}")))?;
|
||||
|
||||
if rows.is_empty() {
|
||||
tx.rollback().await.ok();
|
||||
return Ok(vec![]);
|
||||
}
|
||||
|
||||
let placeholders = rows.iter().map(|_| "?").collect::<Vec<_>>().join(", ");
|
||||
let sql = format!(
|
||||
"UPDATE event_queue SET status = 'processing' WHERE id IN ({})",
|
||||
placeholders
|
||||
);
|
||||
let mut q = sqlx::query(&sql);
|
||||
for r in &rows { q = q.bind(r.id); }
|
||||
q.execute(&mut *tx).await
|
||||
.map_err(|e| DomainError::InfrastructureError(format!("mark processing: {e}")))?;
|
||||
|
||||
tx.commit().await
|
||||
.map_err(|e| DomainError::InfrastructureError(format!("commit claim: {e}")))?;
|
||||
|
||||
Ok(rows)
|
||||
}
|
||||
|
||||
fn decode_row(
|
||||
pool: &SqlitePool,
|
||||
row: QueueRow,
|
||||
max_attempts: i32,
|
||||
) -> Result<EventEnvelope, DomainError> {
|
||||
let db_payload: DbEventPayload = serde_json::from_str(&row.payload)
|
||||
.map_err(|e| DomainError::InfrastructureError(format!("deserialize: {e}")))?;
|
||||
let event = DomainEvent::try_from(db_payload)?;
|
||||
Ok(EventEnvelope::new(event, Box::new(DbAckHandle {
|
||||
pool: pool.clone(),
|
||||
row_id: row.id,
|
||||
attempts: row.attempts,
|
||||
max_attempts,
|
||||
})))
|
||||
}
|
||||
|
||||
struct DbAckHandle {
|
||||
pool: SqlitePool,
|
||||
row_id: i64,
|
||||
attempts: i32,
|
||||
max_attempts: i32,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl AckHandle for DbAckHandle {
|
||||
async fn ack(&self) -> Result<(), DomainError> {
|
||||
sqlx::query("UPDATE event_queue SET status = 'done' WHERE id = ?")
|
||||
.bind(self.row_id)
|
||||
.execute(&self.pool)
|
||||
.await
|
||||
.map_err(|e| DomainError::InfrastructureError(format!("ack: {e}")))?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn nack(&self) -> Result<(), DomainError> {
|
||||
let new_attempts = self.attempts + 1;
|
||||
if new_attempts >= self.max_attempts {
|
||||
sqlx::query(
|
||||
"UPDATE event_queue SET status = 'dead_lettered', attempts = ?, last_error = 'max attempts reached' WHERE id = ?"
|
||||
)
|
||||
.bind(new_attempts)
|
||||
.bind(self.row_id)
|
||||
.execute(&self.pool)
|
||||
.await
|
||||
.map_err(|e| DomainError::InfrastructureError(format!("nack dead-letter: {e}")))?;
|
||||
} else {
|
||||
let backoff = backoff_seconds(new_attempts);
|
||||
let sql = format!(
|
||||
"UPDATE event_queue SET status = 'pending', attempts = ?, next_attempt_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now', '+{backoff} seconds'), last_error = 'nack' WHERE id = ?"
|
||||
);
|
||||
sqlx::query(&sql)
|
||||
.bind(new_attempts)
|
||||
.bind(self.row_id)
|
||||
.execute(&self.pool)
|
||||
.await
|
||||
.map_err(|e| DomainError::InfrastructureError(format!("nack retry: {e}")))?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
fn backoff_seconds(attempts: i32) -> i64 {
|
||||
let base: i64 = 5 * (1i64 << attempts.min(6));
|
||||
base.min(300)
|
||||
}
|
||||
|
||||
6
crates/adapters/sqlite-event-queue/src/migrations.rs
Normal file
6
crates/adapters/sqlite-event-queue/src/migrations.rs
Normal file
@@ -0,0 +1,6 @@
|
||||
pub(crate) async fn run(pool: &sqlx::SqlitePool) -> anyhow::Result<()> {
|
||||
sqlx::migrate!("./migrations")
|
||||
.run(pool)
|
||||
.await
|
||||
.map_err(|e| anyhow::anyhow!("sqlite-event-queue migration failed: {e}"))
|
||||
}
|
||||
189
crates/adapters/sqlite-event-queue/src/payload.rs
Normal file
189
crates/adapters/sqlite-event-queue/src/payload.rs
Normal file
@@ -0,0 +1,189 @@
|
||||
use chrono::NaiveDateTime;
|
||||
use domain::{
|
||||
errors::DomainError,
|
||||
events::DomainEvent,
|
||||
value_objects::{ExternalMetadataId, MovieId, Rating, ReviewId, UserId},
|
||||
};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use uuid::Uuid;
|
||||
|
||||
#[derive(Debug, Serialize, Deserialize, PartialEq)]
|
||||
#[serde(tag = "type", content = "data")]
|
||||
pub enum DbEventPayload {
|
||||
ReviewLogged {
|
||||
review_id: String,
|
||||
movie_id: String,
|
||||
user_id: String,
|
||||
rating: u8,
|
||||
watched_at: i64,
|
||||
},
|
||||
ReviewUpdated {
|
||||
review_id: String,
|
||||
movie_id: String,
|
||||
user_id: String,
|
||||
rating: u8,
|
||||
watched_at: i64,
|
||||
},
|
||||
MovieDiscovered {
|
||||
movie_id: String,
|
||||
external_metadata_id: String,
|
||||
},
|
||||
}
|
||||
|
||||
impl DbEventPayload {
|
||||
pub fn event_type(&self) -> &'static str {
|
||||
match self {
|
||||
DbEventPayload::ReviewLogged { .. } => "ReviewLogged",
|
||||
DbEventPayload::ReviewUpdated { .. } => "ReviewUpdated",
|
||||
DbEventPayload::MovieDiscovered { .. } => "MovieDiscovered",
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn parse_uuid(s: &str, field: &str) -> Result<Uuid, DomainError> {
|
||||
Uuid::parse_str(s)
|
||||
.map_err(|e| DomainError::InfrastructureError(format!("{field}: {e}")))
|
||||
}
|
||||
|
||||
fn parse_ts(ts: i64) -> Result<NaiveDateTime, DomainError> {
|
||||
chrono::DateTime::from_timestamp(ts, 0)
|
||||
.map(|dt| dt.naive_utc())
|
||||
.ok_or_else(|| DomainError::InfrastructureError(format!("invalid timestamp: {ts}")))
|
||||
}
|
||||
|
||||
impl From<&DomainEvent> for DbEventPayload {
|
||||
fn from(event: &DomainEvent) -> Self {
|
||||
match event {
|
||||
DomainEvent::ReviewLogged { review_id, movie_id, user_id, rating, watched_at } => {
|
||||
DbEventPayload::ReviewLogged {
|
||||
review_id: review_id.value().to_string(),
|
||||
movie_id: movie_id.value().to_string(),
|
||||
user_id: user_id.value().to_string(),
|
||||
rating: rating.value(),
|
||||
watched_at: watched_at.and_utc().timestamp(),
|
||||
}
|
||||
}
|
||||
DomainEvent::ReviewUpdated { review_id, movie_id, user_id, rating, watched_at } => {
|
||||
DbEventPayload::ReviewUpdated {
|
||||
review_id: review_id.value().to_string(),
|
||||
movie_id: movie_id.value().to_string(),
|
||||
user_id: user_id.value().to_string(),
|
||||
rating: rating.value(),
|
||||
watched_at: watched_at.and_utc().timestamp(),
|
||||
}
|
||||
}
|
||||
DomainEvent::MovieDiscovered { movie_id, external_metadata_id } => {
|
||||
DbEventPayload::MovieDiscovered {
|
||||
movie_id: movie_id.value().to_string(),
|
||||
external_metadata_id: external_metadata_id.value().to_owned(),
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl TryFrom<DbEventPayload> for DomainEvent {
|
||||
type Error = DomainError;
|
||||
fn try_from(payload: DbEventPayload) -> Result<Self, DomainError> {
|
||||
match payload {
|
||||
DbEventPayload::ReviewLogged { review_id, movie_id, user_id, rating, watched_at } => {
|
||||
Ok(DomainEvent::ReviewLogged {
|
||||
review_id: ReviewId::from_uuid(parse_uuid(&review_id, "review_id")?),
|
||||
movie_id: MovieId::from_uuid(parse_uuid(&movie_id, "movie_id")?),
|
||||
user_id: UserId::from_uuid(parse_uuid(&user_id, "user_id")?),
|
||||
rating: Rating::new(rating)?,
|
||||
watched_at: parse_ts(watched_at)?,
|
||||
})
|
||||
}
|
||||
DbEventPayload::ReviewUpdated { review_id, movie_id, user_id, rating, watched_at } => {
|
||||
Ok(DomainEvent::ReviewUpdated {
|
||||
review_id: ReviewId::from_uuid(parse_uuid(&review_id, "review_id")?),
|
||||
movie_id: MovieId::from_uuid(parse_uuid(&movie_id, "movie_id")?),
|
||||
user_id: UserId::from_uuid(parse_uuid(&user_id, "user_id")?),
|
||||
rating: Rating::new(rating)?,
|
||||
watched_at: parse_ts(watched_at)?,
|
||||
})
|
||||
}
|
||||
DbEventPayload::MovieDiscovered { movie_id, external_metadata_id } => {
|
||||
Ok(DomainEvent::MovieDiscovered {
|
||||
movie_id: MovieId::from_uuid(parse_uuid(&movie_id, "movie_id")?),
|
||||
external_metadata_id: ExternalMetadataId::new(external_metadata_id)?,
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
fn fixed_dt() -> NaiveDateTime {
|
||||
chrono::DateTime::from_timestamp(1_700_000_000, 0).unwrap().naive_utc()
|
||||
}
|
||||
|
||||
fn review_logged() -> DomainEvent {
|
||||
DomainEvent::ReviewLogged {
|
||||
review_id: ReviewId::from_uuid(Uuid::new_v4()),
|
||||
movie_id: MovieId::from_uuid(Uuid::new_v4()),
|
||||
user_id: UserId::from_uuid(Uuid::new_v4()),
|
||||
rating: Rating::new(4).unwrap(),
|
||||
watched_at: fixed_dt(),
|
||||
}
|
||||
}
|
||||
|
||||
fn review_updated() -> DomainEvent {
|
||||
DomainEvent::ReviewUpdated {
|
||||
review_id: ReviewId::from_uuid(Uuid::new_v4()),
|
||||
movie_id: MovieId::from_uuid(Uuid::new_v4()),
|
||||
user_id: UserId::from_uuid(Uuid::new_v4()),
|
||||
rating: Rating::new(3).unwrap(),
|
||||
watched_at: fixed_dt(),
|
||||
}
|
||||
}
|
||||
|
||||
fn movie_discovered() -> DomainEvent {
|
||||
DomainEvent::MovieDiscovered {
|
||||
movie_id: MovieId::from_uuid(Uuid::new_v4()),
|
||||
external_metadata_id: ExternalMetadataId::new("tt1234567".into()).unwrap(),
|
||||
}
|
||||
}
|
||||
|
||||
fn round_trip(event: DomainEvent) {
|
||||
let payload = DbEventPayload::from(&event);
|
||||
let json = serde_json::to_string(&payload).expect("serialize");
|
||||
let back: DbEventPayload = serde_json::from_str(&json).expect("deserialize");
|
||||
let recovered = DomainEvent::try_from(back).expect("try_from");
|
||||
assert_eq!(DbEventPayload::from(&event), DbEventPayload::from(&recovered));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn round_trip_review_logged() {
|
||||
round_trip(review_logged());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn round_trip_review_updated() {
|
||||
round_trip(review_updated());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn round_trip_movie_discovered() {
|
||||
round_trip(movie_discovered());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn serialized_format_is_tagged() {
|
||||
let payload = DbEventPayload::from(&movie_discovered());
|
||||
let json = serde_json::to_string(&payload).unwrap();
|
||||
assert!(json.contains(r#""type":"MovieDiscovered""#));
|
||||
assert!(json.contains(r#""data":"#));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn event_type_strings() {
|
||||
assert_eq!(DbEventPayload::from(&review_logged()).event_type(), "ReviewLogged");
|
||||
assert_eq!(DbEventPayload::from(&review_updated()).event_type(), "ReviewUpdated");
|
||||
assert_eq!(DbEventPayload::from(&movie_discovered()).event_type(), "MovieDiscovered");
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user