made_core/entities/ceremony_instance/decisions/
start.rs

1use time::{Duration, OffsetDateTime};
2
3use crate::entities::ceremony_events::{
4    CeremonyInstanceStarted, MemoryRecalled, SuccessionCarried,
5};
6use crate::entities::{
7    CeremonyDefinition, CeremonyEvent, CeremonyInstance, PublishedCeremonyDefinition,
8};
9use crate::error::DomainError;
10use crate::value_objects::{
11    BudgetAccountId, CeremonyContext, CeremonyDeadline, CeremonyDefinitionDigest, CeremonyId,
12    CeremonyLineage, CeremonySuccession, SessionRecollection, StateDeadline, StateVisit,
13    SuccessionPlan,
14};
15
16impl CeremonyInstance {
17    /// The opening of a ceremony run from a definition supplied for it.
18    ///
19    /// A constructor rather than a command: there is no instance yet
20    /// to decide against. Required inputs are checked against this run's
21    /// context before any opening event is built. The event carries everything
22    /// [`Self::from_started`] needs to open the same instance without
23    /// the definition in hand.
24    ///
25    /// A batch rather than one event, because an opening is sometimes
26    /// two facts: what was started, and what it was told. See
27    /// [`Self::opening_batch`].
28    pub fn decide_start(
29        id: CeremonyId,
30        definition: &CeremonyDefinition,
31        context: CeremonyContext,
32        recollection: Option<SessionRecollection>,
33        now: OffsetDateTime,
34    ) -> Result<Vec<CeremonyEvent>, DomainError> {
35        Ok(Self::opening_batch(
36            Self::opening(id, definition, context, now, None, None, None)?,
37            recollection,
38            now,
39        ))
40    }
41
42    /// The opening of a ceremony bound to a published definition, its
43    /// digest recorded so a later reader can check which one ran.
44    pub fn decide_start_bound(
45        id: CeremonyId,
46        published: &PublishedCeremonyDefinition,
47        context: CeremonyContext,
48        recollection: Option<SessionRecollection>,
49        now: OffsetDateTime,
50    ) -> Result<Vec<CeremonyEvent>, DomainError> {
51        Ok(Self::opening_batch(
52            Self::opening(
53                id,
54                published.definition(),
55                context,
56                now,
57                Some(published.digest()),
58                None,
59                None,
60            )?,
61            recollection,
62            now,
63        ))
64    }
65
66    /// Open a published child using the exact recollection and lineage sealed by its parent plan.
67    pub fn decide_start_bound_child(
68        id: CeremonyId,
69        published: &PublishedCeremonyDefinition,
70        context: CeremonyContext,
71        lineage: CeremonyLineage,
72        recollection: Option<SessionRecollection>,
73        now: OffsetDateTime,
74    ) -> Result<Vec<CeremonyEvent>, DomainError> {
75        Ok(Self::opening_batch(
76            Self::opening(
77                id,
78                published.definition(),
79                context,
80                now,
81                Some(published.digest()),
82                Some(lineage),
83                None,
84            )?,
85            recollection,
86            now,
87        ))
88    }
89
90    /// Open a root ceremony and seal the shared ledger account for its whole tree.
91    pub fn decide_start_bound_budgeted(
92        id: CeremonyId,
93        published: &PublishedCeremonyDefinition,
94        context: CeremonyContext,
95        budget_account_id: BudgetAccountId,
96        recollection: Option<SessionRecollection>,
97        now: OffsetDateTime,
98    ) -> Result<Vec<CeremonyEvent>, DomainError> {
99        Ok(Self::opening_batch(
100            Self::opening(
101                id,
102                published.definition(),
103                context,
104                now,
105                Some(published.digest()),
106                None,
107                Some(budget_account_id),
108            )?,
109            recollection,
110            now,
111        ))
112    }
113
114    /// Open a child with the exact shared ledger account sealed by its parent plan.
115    pub fn decide_start_bound_budgeted_child(
116        id: CeremonyId,
117        published: &PublishedCeremonyDefinition,
118        context: CeremonyContext,
119        lineage: CeremonyLineage,
120        budget_account_id: BudgetAccountId,
121        recollection: Option<SessionRecollection>,
122        now: OffsetDateTime,
123    ) -> Result<Vec<CeremonyEvent>, DomainError> {
124        Ok(Self::opening_batch(
125            Self::opening(
126                id,
127                published.definition(),
128                context,
129                now,
130                Some(published.digest()),
131                Some(lineage),
132                Some(budget_account_id),
133            )?,
134            recollection,
135            now,
136        ))
137    }
138
139    /// Open a successor: the opening the plan settled, and what it was
140    /// given to start from.
141    ///
142    /// The second append of a succession, and the one that may be made
143    /// twice. Everything here is derived from facts already sealed in
144    /// the predecessor, so the batch a retry builds is byte for byte
145    /// the batch the first attempt built — which is what lets a caller
146    /// that crashed between the two appends verify the existing stream
147    /// instead of opening a second one.
148    ///
149    /// The successor seals its own deadlines from its own definition.
150    /// Inheriting the predecessor's would carry a clock the new
151    /// definition never agreed to.
152    pub fn decide_start_successor(
153        id: CeremonyId,
154        published: &PublishedCeremonyDefinition,
155        context: CeremonyContext,
156        succession: CeremonySuccession,
157        plan: &SuccessionPlan,
158        now: OffsetDateTime,
159    ) -> Result<Vec<CeremonyEvent>, DomainError> {
160        let mut opening = Self::opening(
161            id,
162            published.definition(),
163            context,
164            now,
165            Some(published.digest()),
166            None,
167            None,
168        )?;
169        opening.succession = Some(Box::new(succession));
170        let mut events = vec![CeremonyEvent::CeremonyInstanceStarted(opening)];
171        if !plan.carried().is_empty() {
172            events.push(CeremonyEvent::SuccessionCarried(SuccessionCarried {
173                carried: plan.carried().to_vec(),
174                carried_at: now,
175            }));
176        }
177        Ok(events)
178    }
179
180    /// The opening, and the recollection when there was one.
181    ///
182    /// The recollection is decided outside — reading memory is IO, and
183    /// `decide` stays pure — and handed in. What is decided here is
184    /// whether it becomes a fact: a recollection that came back empty
185    /// appends nothing, so a session with nothing to recall, which is
186    /// every session that declares no scope, has exactly the stream it
187    /// had before memory could be read at all.
188    fn opening_batch(
189        opening: CeremonyInstanceStarted,
190        recollection: Option<SessionRecollection>,
191        now: OffsetDateTime,
192    ) -> Vec<CeremonyEvent> {
193        let mut events = vec![CeremonyEvent::CeremonyInstanceStarted(opening)];
194        if let Some(recollection) = recollection.filter(|recalled| !recalled.is_empty()) {
195            events.push(CeremonyEvent::MemoryRecalled(MemoryRecalled {
196                recollection,
197                recalled_at: now,
198            }));
199        }
200        events
201    }
202
203    /// What starting derives from the definition: the initial state
204    /// and the steps that get a pending record.
205    pub(in crate::entities::ceremony_instance) fn opening(
206        id: CeremonyId,
207        definition: &CeremonyDefinition,
208        context: CeremonyContext,
209        now: OffsetDateTime,
210        bound_definition: Option<CeremonyDefinitionDigest>,
211        lineage: Option<CeremonyLineage>,
212        budget_account_id: Option<BudgetAccountId>,
213    ) -> Result<CeremonyInstanceStarted, DomainError> {
214        let missing = definition
215            .inputs()
216            .values()
217            .filter(|input| input.requirement().is_required())
218            .filter(|input| context.attributes().get(input.name().as_str()).is_none())
219            .map(|input| input.name().as_str())
220            .collect::<Vec<_>>();
221        if !missing.is_empty() {
222            return Err(DomainError::InvalidDocument {
223                reason: format!("missing required ceremony inputs: {}", missing.join(", ")),
224            });
225        }
226        let ceremony_deadline = definition
227            .ceremony_timeout()
228            .map(|timeout| checked_deadline(now, timeout.duration(), "ceremony_deadline"))
229            .transpose()?
230            .map(CeremonyDeadline::new);
231        let state_deadline = definition
232            .state_timeout()
233            .map(|timeout| checked_deadline(now, timeout.duration(), "state_deadline"))
234            .transpose()?
235            .map(|at| {
236                StateDeadline::new(definition.initial_state_id().clone(), StateVisit::FIRST, at)
237            });
238        Ok(CeremonyInstanceStarted {
239            ceremony_id: id,
240            definition_name: definition.name().clone(),
241            definition_version: definition.version().clone(),
242            initial_state: definition.initial_state_id().clone(),
243            step_ids: definition.steps().keys().cloned().collect(),
244            context,
245            bound_definition,
246            lineage,
247            succession: None,
248            budget_account_id,
249            ceremony_deadline,
250            state_deadline,
251            created_at: now,
252        })
253    }
254}
255
256pub(crate) fn checked_deadline(
257    now: OffsetDateTime,
258    duration: crate::value_objects::DurationMs,
259    field: &'static str,
260) -> Result<OffsetDateTime, DomainError> {
261    let millis = i64::try_from(duration.get()).map_err(|_| DomainError::OutOfRange {
262        field,
263        value: duration.get() as f64,
264        min: 0.0,
265        max: i64::MAX as f64,
266    })?;
267    now.checked_add(Duration::milliseconds(millis))
268        .ok_or(DomainError::OutOfRange {
269            field,
270            value: duration.get() as f64,
271            min: 0.0,
272            max: i64::MAX as f64,
273        })
274}
made_core/entities/ceremony_instance/decisions/start.rs

made_core/entities/ceremony_instance/decisions/
start.rs
