sourcerer 0.1.1

Event-sourcing framework providing aggregates, repositories, stores, snapshots and upcasters for Rust applications.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92

//! Defines the upcasting mechanism for handling event schema versioning.
use serde_json::Value;

use crate::{Event, Result};

/// A raw, stored event, used for upcasting before deserialization.
#[derive(Debug)]
pub struct RawStoredEvent {
    /// The ID of the aggregate this event belongs to.
    pub aggregate_id: String,
    /// The version of the aggregate after this event was applied.
    pub version: i64,
    /// The version of the event's schema.
    pub event_version: u16,
    /// The type of the event.
    pub event_type: String,
    /// The event payload itself.
    pub payload: Value,
}

/// Defines the interface for an upcaster.
///
/// An upcaster is responsible for transforming an event from an older version
/// to a newer one. This is a key part of handling schema evolution.
pub trait Upcaster<E: Event>: Send + Sync {
    /// The type of event this upcaster can handle.
    fn event_type(&self) -> &'static str;

    /// The version of the event this upcaster can transform from.
    fn source_version(&self) -> u16;

    /// The version of the event this upcaster transforms to.
    fn target_version(&self) -> u16 {
        self.source_version() + 1
    }

    /// Transforms a JSON payload of an event into its next version.
    fn upcast(&self, payload: Value) -> Result<Value>;
}

/// A chain of upcasters that can be applied sequentially to an event.
pub struct UpcasterChain<E: Event> {
    upcasters: Vec<Box<dyn Upcaster<E>>>,
}

impl<E: Event> Default for UpcasterChain<E> {
    fn default() -> Self {
        Self {
            upcasters: Vec::new(),
        }
    }
}

impl<E: Event> UpcasterChain<E> {
    /// Creates a new, empty upcaster chain.
    pub fn new() -> Self {
        Self::default()
    }

    /// Adds an upcaster to the chain.
    pub fn with<U: Upcaster<E> + 'static>(mut self, upcaster: U) -> Self {
        self.upcasters.push(Box::new(upcaster));
        self
    }

    /// Applies the upcasting chain to a raw stored event.
    ///
    /// It will continue to apply upcasters until the event's version matches
    /// the latest version known to the application.
    pub(crate) fn upcast(&self, event: RawStoredEvent) -> Result<RawStoredEvent> {
        let mut current_version = event.event_version;
        let mut payload = event.payload;
        let event_type = event.event_type.clone();

        while let Some(upcaster) = self
            .upcasters
            .iter()
            .find(|u| u.event_type() == event_type && u.source_version() == current_version)
        {
            payload = upcaster.upcast(payload)?;
            current_version = upcaster.target_version();
        }

        Ok(RawStoredEvent {
            aggregate_id: event.aggregate_id,
            version: event.version,
            event_version: current_version,
            event_type,
            payload,
        })
    }
}