refactor(events): centralize lifecycle-event payload on the domain entity

Add email_id()/sender_name()/error_class()/payload() accessors to LifecycleEvent
(mirroring event_type()) and make all four event publishers match-free, so the
payload JSON has one source of truth instead of being hand-duplicated across
webhook/NATS/sqlite/postgres (where it had already drifted — sender_name was
missing from the SQL payloads). Full-body contract tests lock the webhook and
NATS shapes and prove them identical.

The SQL publishers now always store the payload as an object, so a queued event
without a correlation id stores {"correlation_id": null} instead of NULL,
matching the pushed payload. The read path stays nullable and old rows keep
their NULL payload (no data migration).

Author: jdrouet

adapter/outbound-nats/src/event_publisher.rs

@@ -1,6 +1,5 @@
 use anyhow::Context;
 use catapulte_domain::entity::lifecycle_event::LifecycleEvent;
-use catapulte_domain::entity::sender::SenderName;
 use catapulte_domain::port::event_publisher::{EventPublisher, EventPublisherError};
 
 #[derive(Clone)]
@@ -17,59 +16,10 @@ impl NatsEventPublisher {
 }
 
 fn event_to_json(event: &LifecycleEvent) -> serde_json::Value {
-    let (email_id, extra) = match event {
-        LifecycleEvent::Queued { id, correlation_id } => {
-            (id, serde_json::json!({ "correlation_id": correlation_id }))
-        }
-        LifecycleEvent::Sending {
-            id,
-            attempt,
-            correlation_id,
-        } => (
-            id,
-            serde_json::json!({ "attempt": attempt, "correlation_id": correlation_id }),
-        ),
-        LifecycleEvent::Sent {
-            id,
-            sender_name,
-            correlation_id,
-        } => (
-            id,
-            serde_json::json!({
-                "sender_name": sender_name.as_str(),
-                "correlation_id": correlation_id,
-            }),
-        ),
-        LifecycleEvent::Retrying {
-            id,
-            attempt,
-            reason,
-            error_class,
-            sender_name,
-            correlation_id,
-        }
-        | LifecycleEvent::Failed {
-            id,
-            attempt,
-            reason,
-            error_class,
-            sender_name,
-            correlation_id,
-        } => (
-            id,
-            serde_json::json!({
-                "attempt": attempt,
-                "reason": reason,
-                "error_class": error_class.as_str(),
-                "sender_name": sender_name.as_ref().map(SenderName::as_str),
-                "correlation_id": correlation_id,
-            }),
-        ),
-    };
     serde_json::json!({
         "event_type": event.event_type(),
-        "email_id": email_id.as_uuid().to_string(),
-        "payload": extra,
+        "email_id": event.email_id().as_uuid().to_string(),
+        "payload": event.payload(),
     })
 }
 
@@ -116,3 +66,72 @@ impl NatsEventConfig {
         Ok(Some(NatsEventPublisher::new(client, self.subject)))
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use catapulte_domain::entity::email::EmailId;
+    use catapulte_domain::entity::error_class::ErrorClass;
+    use catapulte_domain::entity::lifecycle_event::LifecycleEvent;
+    use catapulte_domain::entity::sender::SenderName;
+
+    use super::event_to_json;
+
+    /// Build the canonical expected body for a given event — shared between the
+    /// webhook and NATS contract-lock tests so they are provably identical.
+    fn expected_body(event: &LifecycleEvent) -> serde_json::Value {
+        serde_json::json!({
+            "event_type": event.event_type(),
+            "email_id": event.email_id().as_uuid().to_string(),
+            "payload": event.payload(),
+        })
+    }
+
+    /// Contract-lock: Sent event — full JSON body equals canonical shape.
+    #[test]
+    fn contract_sent_full_body() {
+        let id = EmailId::default();
+        let event = LifecycleEvent::Sent {
+            id,
+            sender_name: SenderName::new("primary"),
+            correlation_id: Some("corr-sent".to_owned()),
+        };
+        let expected = serde_json::json!({
+            "event_type": "delivery.succeeded",
+            "email_id": id.as_uuid().to_string(),
+            "payload": {
+                "sender_name": "primary",
+                "correlation_id": "corr-sent",
+            },
+        });
+        assert_eq!(event_to_json(&event), expected);
+        // Confirm parity with the helper used by the webhook contract-lock test.
+        assert_eq!(event_to_json(&event), expected_body(&event));
+    }
+
+    /// Contract-lock: Failed event — full JSON body equals canonical shape.
+    #[test]
+    fn contract_failed_full_body() {
+        let id = EmailId::default();
+        let event = LifecycleEvent::Failed {
+            id,
+            attempt: 3,
+            reason: "smtp error".to_owned(),
+            error_class: ErrorClass::Delivery,
+            sender_name: Some(SenderName::new("primary")),
+            correlation_id: Some("corr-fail".to_owned()),
+        };
+        let expected = serde_json::json!({
+            "event_type": "delivery.failed",
+            "email_id": id.as_uuid().to_string(),
+            "payload": {
+                "attempt": 3,
+                "reason": "smtp error",
+                "error_class": "delivery",
+                "sender_name": "primary",
+                "correlation_id": "corr-fail",
+            },
+        });
+        assert_eq!(event_to_json(&event), expected);
+        assert_eq!(event_to_json(&event), expected_body(&event));
+    }
+}

adapter/outbound-postgres/src/event_publisher.rs

@@ -1,5 +1,6 @@
 use anyhow::Context;
 use catapulte_domain::entity::lifecycle_event::LifecycleEvent;
+use catapulte_domain::entity::sender::SenderName;
 use catapulte_domain::port::event_publisher::{EventPublisher, EventPublisherError};
 
 use crate::PostgresAdapter;
@@ -9,76 +10,25 @@ impl EventPublisher for PostgresAdapter {
     ///
     /// Returns `EventPublisherError::Publish` when the database insert fails.
     async fn publish(&self, event: &LifecycleEvent) -> Result<(), EventPublisherError> {
-        let (email_id_uuid, payload, sender_name, error_class) = match event {
-            LifecycleEvent::Queued { id, correlation_id } => (
-                id.as_uuid(),
-                correlation_id
-                    .as_ref()
-                    .map(|cid| serde_json::json!({ "correlation_id": cid })),
-                None,
-                None,
-            ),
-            LifecycleEvent::Sending {
-                id,
-                attempt,
-                correlation_id,
-            } => (
-                id.as_uuid(),
-                Some(serde_json::json!({ "attempt": attempt, "correlation_id": correlation_id })),
-                None,
-                None,
-            ),
-            LifecycleEvent::Sent {
-                id,
-                sender_name,
-                correlation_id,
-            } => (
-                id.as_uuid(),
-                Some(serde_json::json!({
-                    "sender_name": sender_name.as_str(),
-                    "correlation_id": correlation_id,
-                })),
-                Some(sender_name.as_str().to_owned()),
-                None,
-            ),
-            LifecycleEvent::Retrying {
-                id,
-                attempt,
-                reason,
-                error_class,
-                sender_name,
-                correlation_id,
-            }
-            | LifecycleEvent::Failed {
-                id,
-                attempt,
-                reason,
-                error_class,
-                sender_name,
-                correlation_id,
-            } => (
-                id.as_uuid(),
-                Some(serde_json::json!({
-                    "attempt": attempt,
-                    "reason": reason,
-                    "error_class": error_class.as_str(),
-                    "sender_name": sender_name.as_ref().map(catapulte_domain::entity::sender::SenderName::as_str),
-                    "correlation_id": correlation_id,
-                })),
-                sender_name.as_ref().map(|s| s.as_str().to_owned()),
-                Some(error_class.as_str().to_owned()),
-            ),
-        };
-
         let event_id = uuid::Uuid::now_v7();
+        let email_id_uuid = event.email_id().as_uuid();
+        // payload is always written as a JSON object so new rows are consistent
+        // with the pushed (webhook/NATS) payload. The column remains nullable so
+        // rows written before this change keep their NULL value; do not add NOT
+        // NULL to the column without a data migration.
+        let payload = event.payload();
+        let sender_name = event.sender_name().map(SenderName::as_str);
+        let error_class = event
+            .error_class()
+            .map(catapulte_domain::entity::error_class::ErrorClass::as_str);
 
         sqlx::query(
             "INSERT INTO lifecycle_events (id, email_id, event_type, payload, sender_name, error_class) VALUES ($1, $2, $3, $4, $5, $6)",
         )
         .bind(event_id)
         .bind(email_id_uuid)
         .bind(event.event_type())
-        .bind(payload)
+        .bind(Some(payload))
         .bind(sender_name)
         .bind(error_class)
         .execute(self.pool())
@@ -224,8 +174,10 @@ mod tests {
         assert_eq!(p["error_class"].as_str(), Some("delivery"));
     }
 
+    /// Queued without `correlation_id` now stores `{"correlation_id":null}` — an
+    /// object — matching the webhook/NATS wire shape (parity fix).
     #[tokio::test]
-    async fn publish_queued_inserts_row_with_event_type_queued_and_null_payload() {
+    async fn publish_queued_inserts_row_with_correlation_id_payload() {
         let id = EmailId::default();
         let adapter = adapter_with_email(id).await;
         adapter
@@ -247,7 +199,8 @@ mod tests {
                 .fetch_one(adapter.pool())
                 .await
                 .unwrap();
-        assert!(payload.is_none());
+        let p = payload.unwrap();
+        assert_eq!(p, serde_json::json!({ "correlation_id": null }));
     }
 
     #[tokio::test]

adapter/outbound-sqlite/src/event_publisher.rs

@@ -1,5 +1,6 @@
 use anyhow::Context;
 use catapulte_domain::entity::lifecycle_event::LifecycleEvent;
+use catapulte_domain::entity::sender::SenderName;
 use catapulte_domain::port::event_publisher::{EventPublisher, EventPublisherError};
 
 use crate::SqliteAdapter;
@@ -9,76 +10,25 @@ impl EventPublisher for SqliteAdapter {
     ///
     /// Returns `EventPublisherError::Publish` when the database insert fails.
     async fn publish(&self, event: &LifecycleEvent) -> Result<(), EventPublisherError> {
-        let (email_id_uuid, payload, sender_name, error_class) = match event {
-            LifecycleEvent::Queued { id, correlation_id } => (
-                id.as_uuid(),
-                correlation_id
-                    .as_ref()
-                    .map(|cid| serde_json::json!({ "correlation_id": cid })),
-                None,
-                None,
-            ),
-            LifecycleEvent::Sending {
-                id,
-                attempt,
-                correlation_id,
-            } => (
-                id.as_uuid(),
-                Some(serde_json::json!({ "attempt": attempt, "correlation_id": correlation_id })),
-                None,
-                None,
-            ),
-            LifecycleEvent::Sent {
-                id,
-                sender_name,
-                correlation_id,
-            } => (
-                id.as_uuid(),
-                Some(serde_json::json!({
-                    "sender_name": sender_name.as_str(),
-                    "correlation_id": correlation_id,
-                })),
-                Some(sender_name.as_str().to_owned()),
-                None,
-            ),
-            LifecycleEvent::Retrying {
-                id,
-                attempt,
-                reason,
-                error_class,
-                sender_name,
-                correlation_id,
-            }
-            | LifecycleEvent::Failed {
-                id,
-                attempt,
-                reason,
-                error_class,
-                sender_name,
-                correlation_id,
-            } => (
-                id.as_uuid(),
-                Some(serde_json::json!({
-                    "attempt": attempt,
-                    "reason": reason,
-                    "error_class": error_class.as_str(),
-                    "sender_name": sender_name.as_ref().map(catapulte_domain::entity::sender::SenderName::as_str),
-                    "correlation_id": correlation_id,
-                })),
-                sender_name.as_ref().map(|s| s.as_str().to_owned()),
-                Some(error_class.as_str().to_owned()),
-            ),
-        };
-        let email_id_bytes = email_id_uuid.as_bytes().to_vec();
+        let email_id_bytes = event.email_id().as_uuid().as_bytes().to_vec();
         let event_id_bytes = uuid::Uuid::now_v7().as_bytes().to_vec();
+        // payload is always written as a JSON object so new rows are consistent
+        // with the pushed (webhook/NATS) payload. The column remains nullable so
+        // rows written before this change keep their NULL value; do not add NOT
+        // NULL to the column without a data migration.
+        let payload = sqlx::types::Json(event.payload());
+        let sender_name = event.sender_name().map(SenderName::as_str);
+        let error_class = event
+            .error_class()
+            .map(catapulte_domain::entity::error_class::ErrorClass::as_str);
 
         sqlx::query(
             "INSERT INTO lifecycle_events (id, email_id, event_type, payload, sender_name, error_class) VALUES (?, ?, ?, ?, ?, ?)",
         )
         .bind(event_id_bytes)
         .bind(email_id_bytes)
         .bind(event.event_type())
-        .bind(payload.as_ref().map(sqlx::types::Json))
+        .bind(Some(payload))
         .bind(sender_name)
         .bind(error_class)
         .execute(self.pool())
@@ -165,8 +115,10 @@ mod tests {
         assert_eq!(p["error_class"], "delivery");
     }
 
+    /// Queued without `correlation_id` now stores `{"correlation_id":null}` — an
+    /// object — matching the webhook/NATS wire shape (parity fix).
     #[tokio::test]
-    async fn publish_queued_inserts_row_with_event_type_queued_and_null_payload() {
+    async fn publish_queued_inserts_row_with_correlation_id_payload() {
         let id = EmailId::default();
         let adapter = adapter_with_email(id).await;
         adapter
@@ -188,7 +140,8 @@ mod tests {
                 .fetch_one(adapter.pool())
                 .await
                 .unwrap();
-        assert!(payload.is_none());
+        let p = payload.unwrap().0;
+        assert_eq!(p, serde_json::json!({ "correlation_id": null }));
     }
 
     #[tokio::test]