made_core/ports/
append_outcome.rs

1use crate::entities::AuditRecord;
2use crate::ports::PositionedRecord;
3use crate::value_objects::{GlobalPosition, StreamVersion};
4
5/// What an event store did with a batch of facts.
6///
7/// A conflict is an outcome rather than an error: another caller got
8/// there first, nothing landed, and the right response is to reload
9/// and decide again — not to give up.
10#[derive(Debug, Clone, PartialEq, Eq)]
11pub enum AppendOutcome {
12    /// Every fact was sealed and landed. `version` is the stream's new
13    /// head, `records` the sealed facts in order, and `first_position`
14    /// where the first of them sits in the global order; the rest
15    /// follow it contiguously.
16    Appended {
17        version: StreamVersion,
18        records: Vec<AuditRecord>,
19        first_position: GlobalPosition,
20    },
21    /// The stream was not at the version the caller decided against.
22    /// Nothing was written.
23    Conflict {
24        expected: StreamVersion,
25        actual: StreamVersion,
26    },
27}
28
29impl AppendOutcome {
30    #[must_use]
31    pub fn is_conflict(&self) -> bool {
32        matches!(self, Self::Conflict { .. })
33    }
34
35    /// The records that landed — none for a conflict.
36    #[must_use]
37    pub fn records(&self) -> &[AuditRecord] {
38        match self {
39            Self::Appended { records, .. } => records,
40            Self::Conflict { .. } => &[],
41        }
42    }
43
44    /// The stream's head after the append, if it landed.
45    #[must_use]
46    pub fn appended_version(&self) -> Option<StreamVersion> {
47        match self {
48            Self::Appended { version, .. } => Some(*version),
49            Self::Conflict { .. } => None,
50        }
51    }
52
53    /// The records that landed, each with its place in the global
54    /// order — none for a conflict.
55    ///
56    /// Derived rather than stored: the positions of one append are
57    /// contiguous from `first_position` by contract, so keeping one
58    /// per record would be the same fact written `n` times and `n`
59    /// chances for an adapter to write it differently. Whoever needs
60    /// the pair — a projection being told what was just sealed, a
61    /// consumer placing a cursor — gets it the same way here.
62    #[must_use]
63    pub fn positioned(&self) -> Vec<PositionedRecord> {
64        let Self::Appended {
65            records,
66            first_position,
67            ..
68        } = self
69        else {
70            return Vec::new();
71        };
72        let mut position = *first_position;
73        records
74            .iter()
75            .map(|record| {
76                let at = position;
77                position = position.next();
78                PositionedRecord {
79                    position: at,
80                    record: record.clone(),
81                }
82            })
83            .collect()
84    }
85}
86
87#[cfg(test)]
88mod tests {
89    use time::OffsetDateTime;
90
91    use super::*;
92    use crate::entities::ceremony_events::CeremonyCompleted;
93    use crate::entities::{AuditFact, CeremonyEvent};
94    use crate::value_objects::{
95        AuditActor, AuditActorKind, CeremonyId, CeremonyName, CeremonyVersion, EventId, StateId,
96    };
97
98    fn record(event_id: &str) -> AuditRecord {
99        AuditRecord::first(AuditFact {
100            event_id: EventId::new(event_id).unwrap(),
101            event: CeremonyEvent::CeremonyCompleted(CeremonyCompleted {
102                final_state: StateId::new("DONE").unwrap(),
103                completed_at: OffsetDateTime::UNIX_EPOCH,
104            }),
105            ceremony_id: CeremonyId::new("c1").unwrap(),
106            definition_name: CeremonyName::new("positions").unwrap(),
107            definition_version: CeremonyVersion::v1(),
108            occurred_at: OffsetDateTime::UNIX_EPOCH,
109            actor: AuditActor::new("test", AuditActorKind::Engine, None).unwrap(),
110            correlation_id: None,
111            causation_id: None,
112            trace: None,
113        })
114        .unwrap()
115    }
116
117    #[test]
118    fn positions_of_an_append_run_on_from_the_first() {
119        let outcome = AppendOutcome::Appended {
120            version: StreamVersion::new(3),
121            records: vec![record("e1"), record("e2"), record("e3")],
122            first_position: GlobalPosition::new(7).unwrap(),
123        };
124
125        let positioned = outcome.positioned();
126
127        assert_eq!(
128            positioned
129                .iter()
130                .map(|entry| entry.position.value())
131                .collect::<Vec<_>>(),
132            [7, 8, 9]
133        );
134        assert_eq!(
135            positioned
136                .iter()
137                .map(|entry| entry.record.clone())
138                .collect::<Vec<_>>(),
139            outcome.records()
140        );
141    }
142
143    #[test]
144    fn a_conflict_positions_nothing() {
145        let outcome = AppendOutcome::Conflict {
146            expected: StreamVersion::EMPTY,
147            actual: StreamVersion::new(2),
148        };
149
150        assert!(outcome.positioned().is_empty());
151        assert!(outcome.records().is_empty());
152        assert!(outcome.appended_version().is_none());
153        assert!(outcome.is_conflict());
154    }
155}
made_core/ports/append_outcome.rs

made_core/ports/
append_outcome.rs
