puerto_sqlx/
events.rs

1//! The Postgres side of domain events (ADR-0018 § 4): the outbox the relay reads, the `db`
2//! transport (`event_deliveries`) and the dead-letter sink every transport writes to. The tables
3//! are `migrations/outbox.sql`, which `puerto new --events` writes into the project.
4
5mod dead_letters;
6mod deliveries;
7mod outbox;
8
9pub use dead_letters::PgDeadLetters;
10pub use deliveries::{DEFAULT_CLAIM_LEASE, PgDeliveries, PgSubscription};
11pub use outbox::{PgOutbox, RELAY_LOCK_KEY};
12
13use std::collections::BTreeMap;
14
15use chrono::{DateTime, Utc};
16use puerto_core::events::transport::TransportError;
17use puerto_core::events::{EnvelopeParts, EventEnvelope};
18use serde_json::Value;
19use sqlx::Row;
20use sqlx::postgres::PgRow;
21use uuid::Uuid;
22
23/// The outbox columns every query that hands out an envelope selects, prefixed by `o.`.
24const ENVELOPE_COLUMNS: &str = "o.id, o.seq, o.aggregate_type, o.aggregate_id, o.event_type, \
25                                o.schema_version, o.payload, o.headers, o.occurred_at";
26
27/// An outbox row as Postgres stores it, before it is checked to be an envelope.
28#[derive(Debug, Clone, PartialEq)]
29struct OutboxRow {
30    id: Uuid,
31    seq: i64,
32    aggregate_type: String,
33    aggregate_id: Uuid,
34    event_type: String,
35    schema_version: i32,
36    payload: Value,
37    headers: Value,
38    occurred_at: DateTime<Utc>,
39}
40
41impl OutboxRow {
42    fn from_pg(row: &PgRow) -> Result<Self, sqlx::Error> {
43        Ok(Self {
44            id: row.try_get("id")?,
45            seq: row.try_get("seq")?,
46            aggregate_type: row.try_get("aggregate_type")?,
47            aggregate_id: row.try_get("aggregate_id")?,
48            event_type: row.try_get("event_type")?,
49            schema_version: row.try_get("schema_version")?,
50            payload: row.try_get("payload")?,
51            headers: row.try_get("headers")?,
52            occurred_at: row.try_get("occurred_at")?,
53        })
54    }
55}
56
57impl TryFrom<OutboxRow> for EventEnvelope {
58    type Error = TransportError;
59
60    /// A schema version outside `u16`, or headers that are not a string map, are rows nobody
61    /// wrote through a unit of work.
62    fn try_from(row: OutboxRow) -> Result<Self, Self::Error> {
63        let schema_version =
64            u16::try_from(row.schema_version).map_err(|_| TransportError::InvalidData)?;
65        let headers: BTreeMap<String, String> =
66            serde_json::from_value(row.headers).map_err(|_| TransportError::InvalidData)?;
67        Ok(EventEnvelope::from_parts(EnvelopeParts {
68            id: row.id,
69            event_type: row.event_type,
70            aggregate_type: row.aggregate_type,
71            aggregate_id: row.aggregate_id,
72            schema_version,
73            occurred_at: row.occurred_at,
74            payload: row.payload,
75            headers,
76        }))
77    }
78}
79
80/// Reads one row as an envelope, logging why it could not.
81fn envelope_from_pg(row: &PgRow, context: &str) -> Result<EventEnvelope, TransportError> {
82    let row = OutboxRow::from_pg(row).map_err(|err| {
83        tracing::error!(error = %err, "{context}: unreadable outbox row");
84        TransportError::InvalidData
85    })?;
86    let id = row.id;
87    EventEnvelope::try_from(row).inspect_err(|_| {
88        tracing::error!(%id, "{context}: outbox row is not an envelope");
89    })
90}
91
92fn unavailable(context: &str) -> impl FnOnce(sqlx::Error) -> TransportError + '_ {
93    move |err| {
94        tracing::error!(error = %err, "{context}: database failed");
95        TransportError::Unavailable
96    }
97}
98
99#[cfg(test)]
100mod tests {
101    use super::*;
102    use serde_json::json;
103
104    fn row() -> OutboxRow {
105        OutboxRow {
106            id: Uuid::now_v7(),
107            seq: 7,
108            aggregate_type: "product".into(),
109            aggregate_id: Uuid::now_v7(),
110            event_type: "product_created".into(),
111            schema_version: 1,
112            payload: json!({ "name": "Desk lamp" }),
113            headers: json!({ "correlation_id": "req-1" }),
114            occurred_at: Utc::now(),
115        }
116    }
117
118    #[test]
119    fn should_read_every_column_when_the_row_is_an_envelope() {
120        let row = row();
121
122        let envelope = EventEnvelope::try_from(row.clone());
123
124        let Ok(envelope) = envelope else {
125            panic!("rejected a valid row: {envelope:?}");
126        };
127        assert_eq!(envelope.id, row.id);
128        assert_eq!(envelope.aggregate_type, "product");
129        assert_eq!(envelope.aggregate_id, row.aggregate_id);
130        assert_eq!(envelope.event_type, "product_created");
131        assert_eq!(envelope.schema_version, 1);
132        assert_eq!(envelope.payload, row.payload);
133        assert_eq!(envelope.occurred_at, row.occurred_at);
134        assert_eq!(
135            envelope.headers,
136            BTreeMap::from([("correlation_id".to_string(), "req-1".to_string())])
137        );
138    }
139
140    #[test]
141    fn should_reject_the_row_when_its_schema_version_is_out_of_range() {
142        for version in [-1, i32::from(u16::MAX) + 1] {
143            let row = OutboxRow {
144                schema_version: version,
145                ..row()
146            };
147            assert_eq!(
148                EventEnvelope::try_from(row),
149                Err(TransportError::InvalidData)
150            );
151        }
152    }
153
154    #[test]
155    fn should_accept_the_largest_schema_version_when_it_fits() {
156        let row = OutboxRow {
157            schema_version: i32::from(u16::MAX),
158            ..row()
159        };
160        assert_eq!(
161            EventEnvelope::try_from(row).map(|e| e.schema_version),
162            Ok(u16::MAX)
163        );
164    }
165
166    #[test]
167    fn should_reject_the_row_when_its_headers_are_not_a_string_map() {
168        let row = OutboxRow {
169            headers: json!({ "retries": 3 }),
170            ..row()
171        };
172        assert_eq!(
173            EventEnvelope::try_from(row),
174            Err(TransportError::InvalidData)
175        );
176    }
177}
puerto_sqlx/events.rs

puerto_sqlx/
events.rs
