feat: event infrastructure — payload, transport, NATS adapter

- EventPublisher now takes &DomainEvent (11 call sites + 3 impls updated)
- EventEnvelope + EventConsumer port in domain
- event-payload: serializable DomainEvent mirror with subject routing
- event-transport: generic Transport/MessageSource traits, publisher/consumer adapters
- adapters-nats: JetStream publish + durable pull consumer

crates/adapters/event-payload/Cargo.toml

@@ -0,0 +1,11 @@
+[package]
+name = "event-payload"
+version = "0.1.0"
+edition = "2024"
+
+[dependencies]
+domain     = { workspace = true }
+serde      = { workspace = true }
+serde_json = { workspace = true }
+uuid       = { workspace = true }
+chrono     = { workspace = true }

crates/adapters/event-payload/src/lib.rs

@@ -0,0 +1,252 @@
+use domain::{errors::DomainError, events::DomainEvent, value_objects::SystemId};
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(tag = "type", content = "data")]
+pub enum EventPayload {
+    AssetIngested {
+        asset_id: String,
+        owner_user_id: String,
+        timestamp: String,
+    },
+    MetadataUpdated {
+        asset_id: String,
+        updated_by: String,
+        timestamp: String,
+    },
+    AssetDeleted {
+        asset_id: String,
+        deleted_by: String,
+        timestamp: String,
+    },
+    ShareCreated {
+        scope_id: String,
+        shareable_id: String,
+        created_by: String,
+        timestamp: String,
+    },
+    ShareRevoked {
+        scope_id: String,
+        revoked_by: String,
+        timestamp: String,
+    },
+    SidecarSyncRequested {
+        asset_id: String,
+        timestamp: String,
+    },
+    JobEnqueued {
+        job_id: String,
+        job_type: String,
+        timestamp: String,
+    },
+    JobCompleted {
+        job_id: String,
+        timestamp: String,
+    },
+    JobFailed {
+        job_id: String,
+        error: String,
+        timestamp: String,
+    },
+}
+
+impl EventPayload {
+    pub fn subject(&self) -> &'static str {
+        match self {
+            Self::AssetIngested { .. } => "assets.ingested",
+            Self::MetadataUpdated { .. } => "metadata.updated",
+            Self::AssetDeleted { .. } => "assets.deleted",
+            Self::ShareCreated { .. } => "shares.created",
+            Self::ShareRevoked { .. } => "shares.revoked",
+            Self::SidecarSyncRequested { .. } => "sidecars.sync_requested",
+            Self::JobEnqueued { .. } => "jobs.enqueued",
+            Self::JobCompleted { .. } => "jobs.completed",
+            Self::JobFailed { .. } => "jobs.failed",
+        }
+    }
+}
+
+impl From<&DomainEvent> for EventPayload {
+    fn from(e: &DomainEvent) -> Self {
+        match e {
+            DomainEvent::AssetIngested {
+                asset_id,
+                owner_user_id,
+                timestamp,
+            } => Self::AssetIngested {
+                asset_id: asset_id.to_string(),
+                owner_user_id: owner_user_id.to_string(),
+                timestamp: timestamp.to_string(),
+            },
+            DomainEvent::MetadataUpdated {
+                asset_id,
+                updated_by,
+                timestamp,
+            } => Self::MetadataUpdated {
+                asset_id: asset_id.to_string(),
+                updated_by: updated_by.to_string(),
+                timestamp: timestamp.to_string(),
+            },
+            DomainEvent::AssetDeleted {
+                asset_id,
+                deleted_by,
+                timestamp,
+            } => Self::AssetDeleted {
+                asset_id: asset_id.to_string(),
+                deleted_by: deleted_by.to_string(),
+                timestamp: timestamp.to_string(),
+            },
+            DomainEvent::ShareCreated {
+                scope_id,
+                shareable_id,
+                created_by,
+                timestamp,
+            } => Self::ShareCreated {
+                scope_id: scope_id.to_string(),
+                shareable_id: shareable_id.to_string(),
+                created_by: created_by.to_string(),
+                timestamp: timestamp.to_string(),
+            },
+            DomainEvent::ShareRevoked {
+                scope_id,
+                revoked_by,
+                timestamp,
+            } => Self::ShareRevoked {
+                scope_id: scope_id.to_string(),
+                revoked_by: revoked_by.to_string(),
+                timestamp: timestamp.to_string(),
+            },
+            DomainEvent::SidecarSyncRequested {
+                asset_id,
+                timestamp,
+            } => Self::SidecarSyncRequested {
+                asset_id: asset_id.to_string(),
+                timestamp: timestamp.to_string(),
+            },
+            DomainEvent::JobEnqueued {
+                job_id,
+                job_type,
+                timestamp,
+            } => Self::JobEnqueued {
+                job_id: job_id.to_string(),
+                job_type: job_type.clone(),
+                timestamp: timestamp.to_string(),
+            },
+            DomainEvent::JobCompleted { job_id, timestamp } => Self::JobCompleted {
+                job_id: job_id.to_string(),
+                timestamp: timestamp.to_string(),
+            },
+            DomainEvent::JobFailed {
+                job_id,
+                error,
+                timestamp,
+            } => Self::JobFailed {
+                job_id: job_id.to_string(),
+                error: error.clone(),
+                timestamp: timestamp.to_string(),
+            },
+        }
+    }
+}
+
+fn parse_uuid(s: &str, field: &str) -> Result<uuid::Uuid, DomainError> {
+    uuid::Uuid::parse_str(s)
+        .map_err(|_| DomainError::Internal(format!("invalid uuid for {field}: {s}")))
+}
+
+fn parse_timestamp(s: &str) -> Result<domain::value_objects::DateTimeStamp, DomainError> {
+    use chrono::DateTime;
+    let dt = DateTime::parse_from_rfc3339(s)
+        .map_err(|_| DomainError::Internal(format!("invalid timestamp: {s}")))?;
+    Ok(domain::value_objects::DateTimeStamp::from_datetime(
+        dt.with_timezone(&chrono::Utc),
+    ))
+}
+
+impl TryFrom<EventPayload> for DomainEvent {
+    type Error = DomainError;
+
+    fn try_from(p: EventPayload) -> Result<Self, DomainError> {
+        Ok(match p {
+            EventPayload::AssetIngested {
+                asset_id,
+                owner_user_id,
+                timestamp,
+            } => DomainEvent::AssetIngested {
+                asset_id: SystemId::from_uuid(parse_uuid(&asset_id, "asset_id")?),
+                owner_user_id: SystemId::from_uuid(parse_uuid(&owner_user_id, "owner_user_id")?),
+                timestamp: parse_timestamp(&timestamp)?,
+            },
+            EventPayload::MetadataUpdated {
+                asset_id,
+                updated_by,
+                timestamp,
+            } => DomainEvent::MetadataUpdated {
+                asset_id: SystemId::from_uuid(parse_uuid(&asset_id, "asset_id")?),
+                updated_by: SystemId::from_uuid(parse_uuid(&updated_by, "updated_by")?),
+                timestamp: parse_timestamp(&timestamp)?,
+            },
+            EventPayload::AssetDeleted {
+                asset_id,
+                deleted_by,
+                timestamp,
+            } => DomainEvent::AssetDeleted {
+                asset_id: SystemId::from_uuid(parse_uuid(&asset_id, "asset_id")?),
+                deleted_by: SystemId::from_uuid(parse_uuid(&deleted_by, "deleted_by")?),
+                timestamp: parse_timestamp(&timestamp)?,
+            },
+            EventPayload::ShareCreated {
+                scope_id,
+                shareable_id,
+                created_by,
+                timestamp,
+            } => DomainEvent::ShareCreated {
+                scope_id: SystemId::from_uuid(parse_uuid(&scope_id, "scope_id")?),
+                shareable_id: SystemId::from_uuid(parse_uuid(&shareable_id, "shareable_id")?),
+                created_by: SystemId::from_uuid(parse_uuid(&created_by, "created_by")?),
+                timestamp: parse_timestamp(&timestamp)?,
+            },
+            EventPayload::ShareRevoked {
+                scope_id,
+                revoked_by,
+                timestamp,
+            } => DomainEvent::ShareRevoked {
+                scope_id: SystemId::from_uuid(parse_uuid(&scope_id, "scope_id")?),
+                revoked_by: SystemId::from_uuid(parse_uuid(&revoked_by, "revoked_by")?),
+                timestamp: parse_timestamp(&timestamp)?,
+            },
+            EventPayload::SidecarSyncRequested {
+                asset_id,
+                timestamp,
+            } => DomainEvent::SidecarSyncRequested {
+                asset_id: SystemId::from_uuid(parse_uuid(&asset_id, "asset_id")?),
+                timestamp: parse_timestamp(&timestamp)?,
+            },
+            EventPayload::JobEnqueued {
+                job_id,
+                job_type,
+                timestamp,
+            } => DomainEvent::JobEnqueued {
+                job_id: SystemId::from_uuid(parse_uuid(&job_id, "job_id")?),
+                job_type,
+                timestamp: parse_timestamp(&timestamp)?,
+            },
+            EventPayload::JobCompleted { job_id, timestamp } => DomainEvent::JobCompleted {
+                job_id: SystemId::from_uuid(parse_uuid(&job_id, "job_id")?),
+                timestamp: parse_timestamp(&timestamp)?,
+            },
+            EventPayload::JobFailed {
+                job_id,
+                error,
+                timestamp,
+            } => DomainEvent::JobFailed {
+                job_id: SystemId::from_uuid(parse_uuid(&job_id, "job_id")?),
+                error,
+                timestamp: parse_timestamp(&timestamp)?,
+            },
+        })
+    }
+}
+
+#[cfg(test)]
+mod tests;

crates/adapters/event-payload/src/tests.rs

@@ -0,0 +1,89 @@
+use crate::EventPayload;
+use domain::{events::DomainEvent, value_objects::SystemId};
+
+fn make_timestamp() -> domain::value_objects::DateTimeStamp {
+    domain::value_objects::DateTimeStamp::now()
+}
+
+#[test]
+fn subject_mapping() {
+    let cases = vec![
+        (
+            DomainEvent::AssetIngested {
+                asset_id: SystemId::new(),
+                owner_user_id: SystemId::new(),
+                timestamp: make_timestamp(),
+            },
+            "assets.ingested",
+        ),
+        (
+            DomainEvent::JobEnqueued {
+                job_id: SystemId::new(),
+                job_type: "extract_metadata".into(),
+                timestamp: make_timestamp(),
+            },
+            "jobs.enqueued",
+        ),
+        (
+            DomainEvent::JobFailed {
+                job_id: SystemId::new(),
+                error: "boom".into(),
+                timestamp: make_timestamp(),
+            },
+            "jobs.failed",
+        ),
+    ];
+
+    for (event, expected_subject) in cases {
+        let payload = EventPayload::from(&event);
+        assert_eq!(payload.subject(), expected_subject);
+    }
+}
+
+#[test]
+fn roundtrip_asset_ingested() {
+    let id = SystemId::new();
+    let owner = SystemId::new();
+    let event = DomainEvent::AssetIngested {
+        asset_id: id,
+        owner_user_id: owner,
+        timestamp: make_timestamp(),
+    };
+
+    let payload = EventPayload::from(&event);
+    let json = serde_json::to_vec(&payload).unwrap();
+    let back: EventPayload = serde_json::from_slice(&json).unwrap();
+    let restored = DomainEvent::try_from(back).unwrap();
+
+    if let DomainEvent::AssetIngested {
+        asset_id,
+        owner_user_id,
+        ..
+    } = restored
+    {
+        assert_eq!(asset_id, id);
+        assert_eq!(owner_user_id, owner);
+    } else {
+        panic!("wrong variant");
+    }
+}
+
+#[test]
+fn roundtrip_job_failed() {
+    let jid = SystemId::new();
+    let event = DomainEvent::JobFailed {
+        job_id: jid,
+        error: "plugin crashed".into(),
+        timestamp: make_timestamp(),
+    };
+
+    let payload = EventPayload::from(&event);
+    let back = DomainEvent::try_from(payload).unwrap();
+
+    if let DomainEvent::JobFailed { job_id, error, .. } = back {
+        assert_eq!(job_id, jid);
+        assert_eq!(error, "plugin crashed");
+    } else {
+        panic!("wrong variant");
+    }
+}

crates/adapters/event-transport/Cargo.toml

@@ -0,0 +1,15 @@
+[package]
+name = "event-transport"
+version = "0.1.0"
+edition = "2024"
+
+[dependencies]
+domain        = { workspace = true }
+event-payload = { workspace = true }
+serde_json    = { workspace = true }
+async-trait   = { workspace = true }
+tracing       = { workspace = true }
+futures       = { workspace = true }
+
+[dev-dependencies]
+tokio = { workspace = true, features = ["macros", "rt-multi-thread"] }

crates/adapters/event-transport/src/lib.rs

@@ -0,0 +1,99 @@
+use async_trait::async_trait;
+use domain::{
+    errors::DomainError,
+    events::{DomainEvent, EventEnvelope},
+    ports::{EventConsumer, EventPublisher},
+};
+use event_payload::EventPayload;
+use futures::stream::BoxStream;
+
+#[async_trait]
+pub trait Transport: Send + Sync {
+    async fn publish_bytes(&self, subject: &str, bytes: &[u8]) -> Result<(), DomainError>;
+}
+
+pub struct EventPublisherAdapter<T: Transport> {
+    transport: T,
+}
+
+impl<T: Transport> EventPublisherAdapter<T> {
+    pub fn new(transport: T) -> Self {
+        Self { transport }
+    }
+}
+
+#[async_trait]
+impl<T: Transport> EventPublisher for EventPublisherAdapter<T> {
+    async fn publish(&self, event: &DomainEvent) -> Result<(), DomainError> {
+        let payload = EventPayload::from(event);
+        let subject = payload.subject();
+        let bytes =
+            serde_json::to_vec(&payload).map_err(|e| DomainError::Internal(e.to_string()))?;
+        tracing::debug!(subject, "publishing event");
+        self.transport.publish_bytes(subject, &bytes).await
+    }
+}
+
+pub struct RawMessage {
+    pub subject: String,
+    pub payload: Vec<u8>,
+    pub delivery_count: u64,
+    pub ack: Box<dyn Fn() + Send + Sync>,
+    pub nack: Box<dyn Fn() + Send + Sync>,
+}
+
+pub trait MessageSource: Send + Sync {
+    fn messages(&self) -> BoxStream<'_, Result<RawMessage, DomainError>>;
+}
+
+pub struct EventConsumerAdapter<S: MessageSource> {
+    source: S,
+}
+
+impl<S: MessageSource> EventConsumerAdapter<S> {
+    pub fn new(source: S) -> Self {
+        Self { source }
+    }
+}
+
+impl<S: MessageSource> EventConsumer for EventConsumerAdapter<S> {
+    fn consume(&self) -> BoxStream<'_, Result<EventEnvelope, DomainError>> {
+        use futures::StreamExt;
+        let stream = self.source.messages();
+        Box::pin(stream.filter_map(|result| async move {
+            match result {
+                Err(e) => {
+                    tracing::warn!("transport error: {e}");
+                    None
+                }
+                Ok(msg) => {
+                    let payload = match serde_json::from_slice::<EventPayload>(&msg.payload) {
+                        Ok(p) => p,
+                        Err(e) => {
+                            tracing::warn!("failed to deserialize event payload, acking: {e}");
+                            (msg.ack)();
+                            return None;
+                        }
+                    };
+                    let event = match DomainEvent::try_from(payload) {
+                        Ok(e) => e,
+                        Err(e) => {
+                            tracing::warn!("unknown event type, acking: {e}");
+                            (msg.ack)();
+                            return None;
+                        }
+                    };
+                    Some(Ok(EventEnvelope {
+                        event,
+                        delivery_count: msg.delivery_count,
+                        ack: msg.ack,
+                        nack: msg.nack,
+                    }))
+                }
+            }
+        }))
+    }
+}
+
+#[cfg(test)]
+mod tests;

crates/adapters/event-transport/src/tests.rs

@@ -0,0 +1,61 @@
+use crate::{EventPublisherAdapter, Transport};
+use async_trait::async_trait;
+use domain::{
+    errors::DomainError, events::DomainEvent, ports::EventPublisher, value_objects::SystemId,
+};
+use std::sync::{Arc, Mutex};
+
+struct RecordingTransport {
+    messages: Arc<Mutex<Vec<(String, Vec<u8>)>>>,
+}
+
+#[async_trait]
+impl Transport for RecordingTransport {
+    async fn publish_bytes(&self, subject: &str, bytes: &[u8]) -> Result<(), DomainError> {
+        self.messages
+            .lock()
+            .unwrap()
+            .push((subject.to_string(), bytes.to_vec()));
+        Ok(())
+    }
+}
+
+#[tokio::test]
+async fn adapter_publishes_with_correct_subject() {
+    let messages = Arc::new(Mutex::new(Vec::new()));
+    let adapter = EventPublisherAdapter::new(RecordingTransport {
+        messages: messages.clone(),
+    });
+
+    let event = DomainEvent::JobCompleted {
+        job_id: SystemId::new(),
+        timestamp: domain::value_objects::DateTimeStamp::now(),
+    };
+
+    adapter.publish(&event).await.unwrap();
+
+    let recorded = messages.lock().unwrap();
+    assert_eq!(recorded.len(), 1);
+    assert_eq!(recorded[0].0, "jobs.completed");
+}
+
+#[tokio::test]
+async fn published_bytes_are_valid_json() {
+    let messages = Arc::new(Mutex::new(Vec::new()));
+    let adapter = EventPublisherAdapter::new(RecordingTransport {
+        messages: messages.clone(),
+    });
+
+    let event = DomainEvent::AssetIngested {
+        asset_id: SystemId::new(),
+        owner_user_id: SystemId::new(),
+        timestamp: domain::value_objects::DateTimeStamp::now(),
+    };
+
+    adapter.publish(&event).await.unwrap();
+
+    let recorded = messages.lock().unwrap();
+    let payload: event_payload::EventPayload =
+        serde_json::from_slice(&recorded[0].1).expect("should be valid JSON");
+    assert_eq!(payload.subject(), "assets.ingested");
+}

crates/adapters/nats/Cargo.toml

@@ -0,0 +1,13 @@
+[package]
+name = "adapters-nats"
+version = "0.1.0"
+edition = "2024"
+
+[dependencies]
+domain          = { workspace = true }
+event-transport = { workspace = true }
+async-nats      = { workspace = true }
+async-trait     = { workspace = true }
+futures         = { workspace = true }
+tokio           = { workspace = true }
+tracing         = { workspace = true }

crates/adapters/nats/src/lib.rs

@@ -0,0 +1,214 @@
+use async_nats::jetstream::{self, AckKind, stream::Config as StreamConfig};
+use async_trait::async_trait;
+use domain::errors::DomainError;
+use event_transport::{MessageSource, RawMessage, Transport};
+use futures::stream::BoxStream;
+use std::sync::Arc;
+
+const STREAM_NAME: &str = "KPHOTOS_EVENTS";
+const STREAM_SUBJECT: &str = "kphotos-events.>";
+const CONSUMER_NAME: &str = "worker";
+const MAX_MESSAGES: i64 = 100_000;
+
+pub const CONSUMER_MAX_DELIVER: i64 = 5;
+const CONSUMER_ACK_WAIT_SECS: u64 = 30;
+const ACK_TASK_TIMEOUT_SECS: u64 = 5;
+
+fn stream_config() -> StreamConfig {
+    StreamConfig {
+        name: STREAM_NAME.to_string(),
+        subjects: vec![STREAM_SUBJECT.to_string()],
+        max_messages: MAX_MESSAGES,
+        ..Default::default()
+    }
+}
+
+pub async fn ensure_stream(client: &async_nats::Client) -> Result<(), DomainError> {
+    let js = jetstream::new(client.clone());
+
+    if js.update_stream(stream_config()).await.is_ok() {
+        tracing::info!(subject = STREAM_SUBJECT, "JetStream stream updated");
+        return Ok(());
+    }
+
+    tracing::warn!(
+        "JetStream stream update failed (incompatible config), deleting '{STREAM_NAME}' and recreating"
+    );
+    let _ = js.delete_stream(STREAM_NAME).await;
+
+    js.create_stream(stream_config())
+        .await
+        .map(|_| ())
+        .map_err(|e| DomainError::Internal(format!("JetStream stream create failed: {e}")))
+}
+
+pub struct NatsTransport {
+    jetstream: jetstream::Context,
+}
+
+impl NatsTransport {
+    pub fn new(client: async_nats::Client) -> Self {
+        Self {
+            jetstream: jetstream::new(client),
+        }
+    }
+}
+
+#[async_trait]
+impl Transport for NatsTransport {
+    async fn publish_bytes(&self, subject: &str, bytes: &[u8]) -> Result<(), DomainError> {
+        let full_subject = format!("kphotos-events.{subject}");
+        self.jetstream
+            .publish(full_subject, bytes.to_vec().into())
+            .await
+            .map_err(|e| DomainError::Internal(e.to_string()))?
+            .await
+            .map_err(|e| DomainError::Internal(e.to_string()))?;
+        Ok(())
+    }
+}
+
+pub struct NatsMessageSource {
+    jetstream: jetstream::Context,
+}
+
+impl NatsMessageSource {
+    pub fn new(client: async_nats::Client) -> Self {
+        Self {
+            jetstream: jetstream::new(client),
+        }
+    }
+}
+
+impl MessageSource for NatsMessageSource {
+    fn messages(&self) -> BoxStream<'_, Result<RawMessage, DomainError>> {
+        use futures::stream;
+        use tokio::sync::Mutex as TokioMutex;
+
+        let js = self.jetstream.clone();
+        let (tx, rx) = tokio::sync::mpsc::channel::<Result<RawMessage, DomainError>>(128);
+
+        tokio::spawn(async move {
+            let stream = match js.get_stream(STREAM_NAME).await {
+                Ok(s) => s,
+                Err(e) => {
+                    let _ = tx.send(Err(DomainError::Internal(e.to_string()))).await;
+                    return;
+                }
+            };
+
+            if let Ok(info) = stream.consumer_info(CONSUMER_NAME).await
+                && info.config.deliver_subject.is_some()
+            {
+                tracing::info!(
+                    "deleting old push consumer '{CONSUMER_NAME}', replacing with pull"
+                );
+                let _ = stream.delete_consumer(CONSUMER_NAME).await;
+            }
+
+            let consumer = match stream
+                .get_or_create_consumer(
+                    CONSUMER_NAME,
+                    jetstream::consumer::pull::Config {
+                        durable_name: Some(CONSUMER_NAME.to_string()),
+                        deliver_policy: jetstream::consumer::DeliverPolicy::New,
+                        ack_policy: jetstream::consumer::AckPolicy::Explicit,
+                        ack_wait: std::time::Duration::from_secs(CONSUMER_ACK_WAIT_SECS),
+                        max_deliver: CONSUMER_MAX_DELIVER,
+                        ..Default::default()
+                    },
+                )
+                .await
+            {
+                Ok(c) => c,
+                Err(e) => {
+                    let _ = tx.send(Err(DomainError::Internal(e.to_string()))).await;
+                    return;
+                }
+            };
+
+            tracing::info!("NATS pull consumer ready");
+
+            loop {
+                let mut messages = match consumer.messages().await {
+                    Ok(m) => m,
+                    Err(e) => {
+                        tracing::error!("NATS messages() failed: {e}");
+                        let _ = tx.send(Err(DomainError::Internal(e.to_string()))).await;
+                        return;
+                    }
+                };
+
+                use futures::StreamExt;
+                while let Some(result) = messages.next().await {
+                    let msg = match result {
+                        Ok(m) => m,
+                        Err(e) => {
+                            tracing::warn!("NATS message error: {e}");
+                            continue;
+                        }
+                    };
+
+                    let subject = msg.subject.to_string();
+                    let payload = msg.payload.to_vec();
+                    let delivery_count = msg
+                        .info()
+                        .map(|info| info.delivered.max(0) as u64)
+                        .unwrap_or(1);
+                    let msg = Arc::new(msg);
+                    let msg_nack = Arc::clone(&msg);
+
+                    let raw = RawMessage {
+                        subject,
+                        payload,
+                        delivery_count,
+                        ack: Box::new(move || {
+                            let m = Arc::clone(&msg);
+                            tokio::spawn(async move {
+                                let result = tokio::time::timeout(
+                                    std::time::Duration::from_secs(ACK_TASK_TIMEOUT_SECS),
+                                    m.ack(),
+                                )
+                                .await;
+                                match result {
+                                    Ok(Ok(())) => {}
+                                    Ok(Err(e)) => tracing::warn!("NATS ack failed: {e}"),
+                                    Err(_) => tracing::warn!(
+                                        "NATS ack timed out after {ACK_TASK_TIMEOUT_SECS}s"
+                                    ),
+                                }
+                            });
+                        }),
+                        nack: Box::new(move || {
+                            let m = Arc::clone(&msg_nack);
+                            tokio::spawn(async move {
+                                let result = tokio::time::timeout(
+                                    std::time::Duration::from_secs(ACK_TASK_TIMEOUT_SECS),
+                                    m.ack_with(AckKind::Nak(None)),
+                                )
+                                .await;
+                                match result {
+                                    Ok(Ok(())) => {}
+                                    Ok(Err(e)) => tracing::warn!("NATS nack failed: {e}"),
+                                    Err(_) => tracing::warn!(
+                                        "NATS nack timed out after {ACK_TASK_TIMEOUT_SECS}s"
+                                    ),
+                                }
+                            });
+                        }),
+                    };
+
+                    if tx.send(Ok(raw)).await.is_err() {
+                        return;
+                    }
+                }
+            }
+        });
+
+        let rx = Arc::new(TokioMutex::new(rx));
+        Box::pin(stream::unfold(rx, |rx| async move {
+            let item = rx.lock().await.recv().await?;
+            Some((item, rx))
+        }))
+    }
+}

crates/application/src/catalog/commands/register_asset.rs

@@ -69,7 +69,7 @@ impl RegisterAssetHandler {
         };
 
         self.event_pub
-            .publish(DomainEvent::AssetIngested {
+            .publish(&DomainEvent::AssetIngested {
                 asset_id: asset.asset_id,
                 owner_user_id: asset.owner_user_id,
                 timestamp: DateTimeStamp::now(),

crates/application/src/catalog/commands/update_metadata.rs

@@ -43,7 +43,7 @@ impl UpdateMetadataHandler {
         self.metadata_repo.save(&metadata).await?;
 
         self.event_pub
-            .publish(DomainEvent::MetadataUpdated {
+            .publish(&DomainEvent::MetadataUpdated {
                 asset_id: cmd.asset_id,
                 updated_by: cmd.user_id,
                 timestamp: DateTimeStamp::now(),

crates/application/src/processing/commands/complete_job.rs

@@ -49,7 +49,7 @@ impl CompleteJobHandler {
             self.batch_repo.save(&batch).await?;
         }
         self.event_pub
-            .publish(DomainEvent::JobCompleted {
+            .publish(&DomainEvent::JobCompleted {
                 job_id: job.job_id,
                 timestamp: DateTimeStamp::now(),
             })

crates/application/src/processing/commands/enqueue_job.rs

@@ -39,7 +39,7 @@ impl EnqueueJobHandler {
         }
         self.job_repo.save(&job).await?;
         self.event_pub
-            .publish(DomainEvent::JobEnqueued {
+            .publish(&DomainEvent::JobEnqueued {
                 job_id: job.job_id,
                 job_type: format!("{:?}", cmd.job_type),
                 timestamp: DateTimeStamp::now(),

crates/application/src/processing/commands/execute_pipeline.rs

@@ -77,7 +77,7 @@ impl ExecutePipelineHandler {
                 self.job_repo.save(&job).await?;
                 self.update_batch_on_complete(&job).await?;
                 self.event_pub
-                    .publish(DomainEvent::JobCompleted {
+                    .publish(&DomainEvent::JobCompleted {
                         job_id: job.job_id,
                         timestamp: DateTimeStamp::now(),
                     })
@@ -89,7 +89,7 @@ impl ExecutePipelineHandler {
                 self.job_repo.save(&job).await?;
                 self.update_batch_on_fail(&job).await?;
                 self.event_pub
-                    .publish(DomainEvent::JobFailed {
+                    .publish(&DomainEvent::JobFailed {
                         job_id: job.job_id,
                         error: error_msg,
                         timestamp: DateTimeStamp::now(),

crates/application/src/processing/commands/fail_job.rs

@@ -49,7 +49,7 @@ impl FailJobHandler {
                 self.batch_repo.save(&batch).await?;
             }
             self.event_pub
-                .publish(DomainEvent::JobFailed {
+                .publish(&DomainEvent::JobFailed {
                     job_id: job.job_id,
                     error: cmd.error,
                     timestamp: DateTimeStamp::now(),
@@ -57,7 +57,7 @@ impl FailJobHandler {
                 .await?;
         } else if job.status == JobStatus::Queued {
             self.event_pub
-                .publish(DomainEvent::JobEnqueued {
+                .publish(&DomainEvent::JobEnqueued {
                     job_id: job.job_id,
                     job_type: format!("{:?}", job.job_type),
                     timestamp: DateTimeStamp::now(),

crates/application/src/sharing/commands/revoke_share.rs

@@ -36,7 +36,7 @@ impl RevokeShareHandler {
         self.share_repo.delete_scope(&cmd.scope_id).await?;
 
         self.event_pub
-            .publish(DomainEvent::ShareRevoked {
+            .publish(&DomainEvent::ShareRevoked {
                 scope_id: cmd.scope_id,
                 revoked_by: cmd.revoked_by,
                 timestamp: DateTimeStamp::now(),

crates/application/src/sharing/commands/share_resource.rs

@@ -51,7 +51,7 @@ impl ShareResourceHandler {
         self.share_repo.save_target(&target).await?;
 
         self.event_pub
-            .publish(DomainEvent::ShareCreated {
+            .publish(&DomainEvent::ShareCreated {
                 scope_id: scope.scope_id,
                 shareable_id: cmd.shareable_id,
                 created_by: cmd.created_by,

crates/application/src/storage/commands/ingest_asset.rs

@@ -145,7 +145,7 @@ impl IngestAssetHandler {
         self.ledger_repo.record(&entry).await?;
 
         self.event_pub
-            .publish(DomainEvent::AssetIngested {
+            .publish(&DomainEvent::AssetIngested {
                 asset_id: asset.asset_id,
                 owner_user_id: cmd.uploader_id,
                 timestamp: DateTimeStamp::now(),

crates/application/src/testing/fakes.rs

@@ -37,8 +37,8 @@ impl Default for StubEventPublisher {
 
 #[async_trait]
 impl EventPublisher for StubEventPublisher {
-    async fn publish(&self, event: DomainEvent) -> Result<(), DomainError> {
-        self.events.lock().await.push(event);
+    async fn publish(&self, event: &DomainEvent) -> Result<(), DomainError> {
+        self.events.lock().await.push(event.clone());
         Ok(())
     }
 }

crates/bootstrap/src/log_event_publisher.rs

@@ -5,7 +5,7 @@ pub struct LogEventPublisher;
 
 #[async_trait]
 impl EventPublisher for LogEventPublisher {
-    async fn publish(&self, event: DomainEvent) -> Result<(), DomainError> {
+    async fn publish(&self, event: &DomainEvent) -> Result<(), DomainError> {
         tracing::info!(?event, "domain event published");
         Ok(())
     }

crates/domain/src/common/events.rs

@@ -1,5 +1,12 @@
 use crate::common::value_objects::{DateTimeStamp, SystemId};
 
+pub struct EventEnvelope {
+    pub event: DomainEvent,
+    pub delivery_count: u64,
+    pub ack: Box<dyn Fn() + Send + Sync>,
+    pub nack: Box<dyn Fn() + Send + Sync>,
+}
+
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 pub enum DomainEvent {
     AssetIngested {

crates/domain/src/common/ports.rs

@@ -1,8 +1,13 @@
 use crate::common::errors::DomainError;
-use crate::common::events::DomainEvent;
+use crate::common::events::{DomainEvent, EventEnvelope};
 use async_trait::async_trait;
+use futures::stream::BoxStream;
 
 #[async_trait]
 pub trait EventPublisher: Send + Sync {
-    async fn publish(&self, event: DomainEvent) -> Result<(), DomainError>;
+    async fn publish(&self, event: &DomainEvent) -> Result<(), DomainError>;
+}
+
+pub trait EventConsumer: Send + Sync {
+    fn consume(&self) -> BoxStream<'_, Result<EventEnvelope, DomainError>>;
 }

crates/worker/src/main.rs

@@ -55,7 +55,7 @@ struct LogEventPublisher;
 impl domain::ports::EventPublisher for LogEventPublisher {
     async fn publish(
         &self,
-        event: domain::events::DomainEvent,
+        event: &domain::events::DomainEvent,
     ) -> Result<(), domain::errors::DomainError> {
         info!(event = ?event, "domain event");
         Ok(())