puerto_sqlx/events/
dead_letters.rs

1use async_trait::async_trait;
2use puerto_core::events::transport::{DeadLetterRecord, DeadLetterSink, TransportError};
3use sqlx::PgPool;
4use uuid::Uuid;
5
6use super::unavailable;
7
8/// `dead_letters`, whatever the transport. Recording the same event for the same consumer twice
9/// keeps the first record (`dead_letters_event_consumer`).
10pub struct PgDeadLetters {
11    pool: PgPool,
12}
13
14impl PgDeadLetters {
15    pub fn new(pool: PgPool) -> Self {
16        Self { pool }
17    }
18}
19
20#[async_trait]
21impl DeadLetterSink for PgDeadLetters {
22    async fn record(&self, record: &DeadLetterRecord) -> Result<(), TransportError> {
23        let attempts = i32::try_from(record.attempts).unwrap_or(i32::MAX);
24        sqlx::query(
25            "INSERT INTO dead_letters \
26             (id, event_id, consumer, event_type, attempts, reason, detail, payload) \
27             VALUES ($1, $2, $3, $4, $5, $6, $7, $8) \
28             ON CONFLICT (event_id, consumer) DO NOTHING",
29        )
30        .bind(Uuid::now_v7())
31        .bind(record.event_id)
32        .bind(&record.consumer)
33        .bind(&record.event_type)
34        .bind(attempts)
35        .bind(&record.reason)
36        .bind(&record.detail)
37        .bind(&record.payload)
38        .execute(&self.pool)
39        .await
40        .map_err(unavailable("dead_letters: record"))?;
41        tracing::warn!(
42            consumer = %record.consumer,
43            event_id = ?record.event_id,
44            reason = %record.reason,
45            "dead_letters: recorded"
46        );
47        Ok(())
48    }
49}
puerto_sqlx/events/dead_letters.rs

puerto_sqlx/events/
dead_letters.rs
