minco_plugin_events/
lib.rs

1//! Domain event publishing and explicit transactional-outbox primitives.
2#![forbid(unsafe_code)]
3
4use async_trait::async_trait;
5use chrono::{DateTime, TimeDelta, Utc};
6use minco_core::{
7    CapabilityProvision, DataClass, Plugin, PluginContext, PluginDescriptor, PluginError, PluginId,
8    PluginStability,
9};
10use semver::{Version, VersionReq};
11use serde::{Deserialize, Serialize};
12use std::{collections::BTreeMap, sync::Arc};
13use tokio::sync::RwLock;
14use uuid::Uuid;
15
16#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
17pub struct DomainEvent {
18    pub id: Uuid,
19    pub event_type: String,
20    pub aggregate_type: String,
21    pub aggregate_id: String,
22    pub correlation_id: Uuid,
23    pub occurred_at: DateTime<Utc>,
24    pub payload: serde_json::Value,
25    #[serde(default)]
26    pub metadata: BTreeMap<String, serde_json::Value>,
27}
28
29impl DomainEvent {
30    pub fn new(
31        event_type: impl Into<String>,
32        aggregate_type: impl Into<String>,
33        aggregate_id: impl Into<String>,
34        correlation_id: Uuid,
35        payload: serde_json::Value,
36    ) -> Self {
37        Self {
38            id: Uuid::now_v7(),
39            event_type: event_type.into(),
40            aggregate_type: aggregate_type.into(),
41            aggregate_id: aggregate_id.into(),
42            correlation_id,
43            occurred_at: Utc::now(),
44            payload,
45            metadata: BTreeMap::new(),
46        }
47    }
48}
49
50#[async_trait]
51pub trait EventPublisher: Send + Sync + std::fmt::Debug {
52    async fn publish(&self, event: &DomainEvent) -> Result<(), EventError>;
53}
54
55#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
56#[serde(rename_all = "snake_case")]
57pub enum OutboxStatus {
58    Pending,
59    Claimed,
60    Published,
61    Failed,
62}
63
64#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
65pub struct OutboxRecord {
66    pub event: DomainEvent,
67    pub status: OutboxStatus,
68    pub attempt_count: u32,
69    pub available_at: DateTime<Utc>,
70    #[serde(default, skip_serializing_if = "Option::is_none")]
71    pub claimed_by: Option<String>,
72    #[serde(default, skip_serializing_if = "Option::is_none")]
73    pub claim_expires_at: Option<DateTime<Utc>>,
74    #[serde(default, skip_serializing_if = "Option::is_none")]
75    pub last_error: Option<String>,
76}
77
78impl OutboxRecord {
79    pub fn pending(event: DomainEvent) -> Self {
80        Self {
81            event,
82            status: OutboxStatus::Pending,
83            attempt_count: 0,
84            available_at: Utc::now(),
85            claimed_by: None,
86            claim_expires_at: None,
87            last_error: None,
88        }
89    }
90}
91
92/// Persistence boundary for a transactional outbox.
93///
94/// Implementations must claim records atomically. Reading pending rows and updating them in a
95/// second statement is not a conforming implementation because multiple workers could publish the
96/// same event concurrently.
97#[async_trait]
98pub trait OutboxStore: Send + Sync + std::fmt::Debug {
99    async fn enqueue(&self, record: OutboxRecord) -> Result<(), EventError>;
100
101    async fn claim_pending(
102        &self,
103        worker_id: &str,
104        limit: usize,
105        claim_expires_at: DateTime<Utc>,
106    ) -> Result<Vec<OutboxRecord>, EventError>;
107
108    /// Atomically claims one known event for request-assisted publication.
109    async fn claim_event(
110        &self,
111        event_id: Uuid,
112        worker_id: &str,
113        claim_expires_at: DateTime<Utc>,
114    ) -> Result<Option<OutboxRecord>, EventError>;
115
116    async fn mark_published(&self, event_id: Uuid, worker_id: &str) -> Result<(), EventError>;
117
118    async fn mark_failed(
119        &self,
120        event_id: Uuid,
121        worker_id: &str,
122        error: String,
123        retry_at: DateTime<Utc>,
124    ) -> Result<(), EventError>;
125
126    async fn recover_expired_claims(&self, now: DateTime<Utc>) -> Result<usize, EventError>;
127}
128
129#[derive(Clone)]
130pub struct EventServices {
131    pub publisher: Arc<dyn EventPublisher>,
132    pub outbox: Arc<dyn OutboxStore>,
133}
134
135impl std::fmt::Debug for EventServices {
136    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
137        formatter
138            .debug_struct("EventServices")
139            .finish_non_exhaustive()
140    }
141}
142
143impl EventServices {
144    /// Executes one bounded, explicit outbox-dispatch pass.
145    ///
146    /// Minco never schedules this automatically. Applications choose request-assisted delivery,
147    /// an SQS-triggered worker, an operator command, or an explicitly costed recovery schedule.
148    pub async fn dispatch_once(
149        &self,
150        worker_id: &str,
151        limit: usize,
152        lease: TimeDelta,
153    ) -> Result<DispatchReport, EventError> {
154        validate_worker(worker_id, limit, lease)?;
155        let now = Utc::now();
156        self.outbox.recover_expired_claims(now).await?;
157        let claimed = self
158            .outbox
159            .claim_pending(worker_id, limit, now + lease)
160            .await?;
161        let mut report = DispatchReport {
162            claimed: claimed.len(),
163            ..DispatchReport::default()
164        };
165        for record in claimed {
166            match self.publisher.publish(&record.event).await {
167                Ok(()) => {
168                    self.outbox
169                        .mark_published(record.event.id, worker_id)
170                        .await?;
171                    report.published += 1;
172                }
173                Err(error) => {
174                    self.outbox
175                        .mark_failed(
176                            record.event.id,
177                            worker_id,
178                            error.to_string(),
179                            Utc::now() + TimeDelta::seconds(30),
180                        )
181                        .await?;
182                    report.failed += 1;
183                }
184            }
185        }
186        Ok(report)
187    }
188}
189
190#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
191pub struct DispatchReport {
192    pub claimed: usize,
193    pub published: usize,
194    pub failed: usize,
195}
196
197#[derive(Debug, Default)]
198pub struct MemoryEventBus {
199    published: RwLock<Vec<DomainEvent>>,
200    outbox: RwLock<BTreeMap<Uuid, OutboxRecord>>,
201}
202
203impl MemoryEventBus {
204    pub async fn published(&self) -> Vec<DomainEvent> {
205        self.published.read().await.clone()
206    }
207
208    pub async fn outbox_records(&self) -> Vec<OutboxRecord> {
209        self.outbox.read().await.values().cloned().collect()
210    }
211}
212
213#[async_trait]
214impl EventPublisher for MemoryEventBus {
215    async fn publish(&self, event: &DomainEvent) -> Result<(), EventError> {
216        validate_event(event)?;
217        self.published.write().await.push(event.clone());
218        Ok(())
219    }
220}
221
222#[async_trait]
223impl OutboxStore for MemoryEventBus {
224    async fn enqueue(&self, record: OutboxRecord) -> Result<(), EventError> {
225        validate_event(&record.event)?;
226        if record.status != OutboxStatus::Pending {
227            return Err(EventError::InvalidOutboxState);
228        }
229        let mut outbox = self.outbox.write().await;
230        if outbox.contains_key(&record.event.id) {
231            return Err(EventError::DuplicateEvent(record.event.id));
232        }
233        outbox.insert(record.event.id, record);
234        drop(outbox);
235        Ok(())
236    }
237
238    async fn claim_pending(
239        &self,
240        worker_id: &str,
241        limit: usize,
242        claim_expires_at: DateTime<Utc>,
243    ) -> Result<Vec<OutboxRecord>, EventError> {
244        if worker_id.trim().is_empty() || limit == 0 || claim_expires_at <= Utc::now() {
245            return Err(EventError::InvalidClaim);
246        }
247        let now = Utc::now();
248        let mut outbox = self.outbox.write().await;
249        let ids = outbox
250            .values()
251            .filter(|record| {
252                matches!(record.status, OutboxStatus::Pending | OutboxStatus::Failed)
253                    && record.available_at <= now
254            })
255            .take(limit)
256            .map(|record| record.event.id)
257            .collect::<Vec<_>>();
258        let mut claimed = Vec::with_capacity(ids.len());
259        for id in ids {
260            let record = outbox.get_mut(&id).ok_or(EventError::MissingEvent(id))?;
261            record.status = OutboxStatus::Claimed;
262            record.claimed_by = Some(worker_id.to_owned());
263            record.claim_expires_at = Some(claim_expires_at);
264            record.attempt_count = record.attempt_count.saturating_add(1);
265            claimed.push(record.clone());
266        }
267        drop(outbox);
268        Ok(claimed)
269    }
270
271    async fn claim_event(
272        &self,
273        event_id: Uuid,
274        worker_id: &str,
275        claim_expires_at: DateTime<Utc>,
276    ) -> Result<Option<OutboxRecord>, EventError> {
277        if worker_id.trim().is_empty() || claim_expires_at <= Utc::now() {
278            return Err(EventError::InvalidClaim);
279        }
280        let now = Utc::now();
281        let mut outbox = self.outbox.write().await;
282        let claimed = outbox.get_mut(&event_id).and_then(|record| {
283            if !matches!(record.status, OutboxStatus::Pending | OutboxStatus::Failed)
284                || record.available_at > now
285            {
286                return None;
287            }
288            record.status = OutboxStatus::Claimed;
289            record.claimed_by = Some(worker_id.to_owned());
290            record.claim_expires_at = Some(claim_expires_at);
291            record.attempt_count = record.attempt_count.saturating_add(1);
292            Some(record.clone())
293        });
294        drop(outbox);
295        Ok(claimed)
296    }
297
298    async fn mark_published(&self, event_id: Uuid, worker_id: &str) -> Result<(), EventError> {
299        let mut outbox = self.outbox.write().await;
300        {
301            let record = claimed_by(&mut outbox, event_id, worker_id)?;
302            record.status = OutboxStatus::Published;
303            record.claimed_by = None;
304            record.claim_expires_at = None;
305            record.last_error = None;
306        }
307        drop(outbox);
308        Ok(())
309    }
310
311    async fn mark_failed(
312        &self,
313        event_id: Uuid,
314        worker_id: &str,
315        error: String,
316        retry_at: DateTime<Utc>,
317    ) -> Result<(), EventError> {
318        let mut outbox = self.outbox.write().await;
319        {
320            let record = claimed_by(&mut outbox, event_id, worker_id)?;
321            record.status = OutboxStatus::Failed;
322            record.claimed_by = None;
323            record.claim_expires_at = None;
324            record.available_at = retry_at;
325            record.last_error = Some(error);
326        }
327        drop(outbox);
328        Ok(())
329    }
330
331    async fn recover_expired_claims(&self, now: DateTime<Utc>) -> Result<usize, EventError> {
332        let mut recovered = 0;
333        let mut outbox = self.outbox.write().await;
334        for record in outbox.values_mut() {
335            if record.status == OutboxStatus::Claimed
336                && record
337                    .claim_expires_at
338                    .is_some_and(|expires| expires <= now)
339            {
340                record.status = OutboxStatus::Pending;
341                record.claimed_by = None;
342                record.claim_expires_at = None;
343                recovered += 1;
344            }
345        }
346        drop(outbox);
347        Ok(recovered)
348    }
349}
350
351fn claimed_by<'a>(
352    outbox: &'a mut BTreeMap<Uuid, OutboxRecord>,
353    event_id: Uuid,
354    worker_id: &str,
355) -> Result<&'a mut OutboxRecord, EventError> {
356    let record = outbox
357        .get_mut(&event_id)
358        .ok_or(EventError::MissingEvent(event_id))?;
359    if record.status != OutboxStatus::Claimed || record.claimed_by.as_deref() != Some(worker_id) {
360        return Err(EventError::ClaimOwnership {
361            event_id,
362            worker_id: worker_id.to_owned(),
363        });
364    }
365    Ok(record)
366}
367
368#[derive(Debug, Clone)]
369pub struct EventsPlugin {
370    services: EventServices,
371}
372
373impl EventsPlugin {
374    pub fn new(publisher: Arc<dyn EventPublisher>, outbox: Arc<dyn OutboxStore>) -> Self {
375        Self {
376            services: EventServices { publisher, outbox },
377        }
378    }
379
380    pub fn memory() -> (Self, Arc<MemoryEventBus>) {
381        let bus = Arc::new(MemoryEventBus::default());
382        (Self::new(bus.clone(), bus.clone()), bus)
383    }
384}
385
386impl Plugin for EventsPlugin {
387    fn descriptor(&self) -> PluginDescriptor {
388        let mut descriptor = PluginDescriptor::new(
389            PluginId::new("events").expect("static plugin ID"),
390            Version::new(1, 0, 0),
391            "Domain event publisher and transactional outbox ports without hidden schedules",
392        );
393        descriptor.documentation = Some("https://docs.rs/minco-plugin-events".into());
394        descriptor.core_compatibility =
395            VersionReq::parse(concat!("^", env!("CARGO_PKG_VERSION"))).expect("package version");
396        descriptor.stability = PluginStability::Beta;
397        descriptor
398            .data_classes
399            .extend([DataClass::Internal, DataClass::CustomerProvided]);
400        descriptor.provides.extend([
401            CapabilityProvision {
402                name: "events.publish".into(),
403                version: Version::new(1, 0, 0),
404            },
405            CapabilityProvision {
406                name: "events.outbox".into(),
407                version: Version::new(1, 0, 0),
408            },
409        ]);
410        descriptor
411    }
412
413    fn install(&self, context: &mut PluginContext<'_>) -> Result<(), PluginError> {
414        context.services().insert(Arc::new(self.services.clone()))?;
415        Ok(())
416    }
417}
418
419fn validate_event(event: &DomainEvent) -> Result<(), EventError> {
420    if event.event_type.trim().is_empty()
421        || event.aggregate_type.trim().is_empty()
422        || event.aggregate_id.trim().is_empty()
423    {
424        return Err(EventError::InvalidEvent);
425    }
426    Ok(())
427}
428
429fn validate_worker(worker_id: &str, limit: usize, lease: TimeDelta) -> Result<(), EventError> {
430    if worker_id.trim().is_empty() || limit == 0 || lease <= TimeDelta::zero() {
431        Err(EventError::InvalidClaim)
432    } else {
433        Ok(())
434    }
435}
436
437#[derive(Debug, thiserror::Error)]
438pub enum EventError {
439    #[error("event type, aggregate type, and aggregate ID are required")]
440    InvalidEvent,
441    #[error("outbox records must be enqueued in pending state")]
442    InvalidOutboxState,
443    #[error("outbox claim requires a worker ID, positive limit, and future lease")]
444    InvalidClaim,
445    #[error("event already exists in the outbox: {0}")]
446    DuplicateEvent(Uuid),
447    #[error("event does not exist in the outbox: {0}")]
448    MissingEvent(Uuid),
449    #[error("worker {worker_id} does not own the claim for event {event_id}")]
450    ClaimOwnership { event_id: Uuid, worker_id: String },
451    #[error("event infrastructure failed: {0}")]
452    Infrastructure(String),
453}
454
455#[cfg(test)]
456mod tests {
457    use super::*;
458
459    fn event() -> DomainEvent {
460        DomainEvent::new(
461            "order.placed",
462            "order",
463            "order-1",
464            Uuid::now_v7(),
465            serde_json::json!({"total": 10}),
466        )
467    }
468
469    #[tokio::test]
470    async fn memory_outbox_claims_atomically_and_requires_claim_ownership() {
471        let bus = MemoryEventBus::default();
472        let event = event();
473        bus.enqueue(OutboxRecord::pending(event.clone()))
474            .await
475            .unwrap();
476
477        let first = bus
478            .claim_pending("worker-a", 10, Utc::now() + TimeDelta::minutes(1))
479            .await
480            .unwrap();
481        let second = bus
482            .claim_pending("worker-b", 10, Utc::now() + TimeDelta::minutes(1))
483            .await
484            .unwrap();
485        assert_eq!(first.len(), 1);
486        assert!(second.is_empty());
487        assert!(matches!(
488            bus.mark_published(event.id, "worker-b").await,
489            Err(EventError::ClaimOwnership { .. })
490        ));
491        bus.mark_published(event.id, "worker-a").await.unwrap();
492        assert!(
493            bus.claim_pending("worker-b", 10, Utc::now() + TimeDelta::minutes(1))
494                .await
495                .unwrap()
496                .is_empty()
497        );
498    }
499
500    #[tokio::test]
501    async fn dispatch_is_explicit_and_bounded() {
502        let bus = Arc::new(MemoryEventBus::default());
503        bus.enqueue(OutboxRecord::pending(event())).await.unwrap();
504        let services = EventServices {
505            publisher: bus.clone(),
506            outbox: bus.clone(),
507        };
508        let report = services
509            .dispatch_once("worker-a", 10, TimeDelta::minutes(1))
510            .await
511            .unwrap();
512        assert_eq!(
513            report,
514            DispatchReport {
515                claimed: 1,
516                published: 1,
517                failed: 0,
518            }
519        );
520        assert_eq!(bus.published().await.len(), 1);
521    }
522}
minco_plugin_events/lib.rs

minco_plugin_events/
lib.rs
