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

minco_plugin_events/
lib.rs
