vertep_runtime/
journal.rs

1//! Journal entries as RDF: what one turn records and how it is read back.
2//!
3//! An entry is the named graph named `<address>/entry/<position>`. An input entry records the
4//! accepted input (event name, type, payload, send id, origin, origin type, invoke id, message
5//! id), the chart version, the timestamp stamped at dequeue, the configuration digest after the
6//! macrostep, and its effects: outgoing sends, timers scheduled and cancelled, invocations
7//! started, requests issued and answered, transitions taken (annotated with the entry as their
8//! cause) and whether the session terminated. Migration and fork entries carry snapshots
9//! instead. The store adds `rdf:type vt:JournalEntry`, `vt:actor` and `vt:position`.
10
11use core::fmt;
12use std::collections::BTreeSet;
13
14use oxrdf::vocab::xsd;
15use oxrdf::{Graph, Literal, NamedNode, NamedNodeRef, Term, TermRef, Triple};
16use serde::{Deserialize, Serialize};
17use sha2::{Digest, Sha256};
18use vertep_core::datamodel::{Event, EventType, Value};
19use vertep_store::reification::Annotation;
20use vertep_store::{Entry, JournalRecord};
21
22use crate::address::{self, Address, hex};
23use crate::chart::UnhandledPolicy;
24use crate::json;
25use crate::vocab::{self as vt, node};
26
27/// A malformed journal entry or snapshot.
28#[derive(Clone, Debug, PartialEq, Eq)]
29pub struct DecodeError(pub String);
30
31impl fmt::Display for DecodeError {
32    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
33        write!(f, "malformed journal record: {}", self.0)
34    }
35}
36
37impl std::error::Error for DecodeError {}
38
39/// A reminder registration carried by a timer.
40// @tg: implements:: [[openspec:durable-timers#Recurring Reminders]]
41#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
42pub struct ReminderSpec {
43    /// The reminder name.
44    pub name: String,
45    /// The period in milliseconds.
46    pub period_ms: i64,
47}
48
49/// A durable timer: a delayed send, a request timeout or a reminder.
50// @tg: implements:: [[openspec:durable-timers#Delayed Send Is Journaled]]
51#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
52pub struct TimerRecord {
53    /// The timer IRI, `<entry>/timer/<send id>`; also the message id of its event (for a
54    /// reminder, of its occurrences plus `/<n>`).
55    pub iri: String,
56    /// The actor that scheduled it.
57    pub actor: String,
58    /// The send id (for a reminder, its name).
59    pub send_id: String,
60    /// Due time in milliseconds: the scheduling entry's timestamp plus the delay (for a
61    /// reminder, of the first occurrence).
62    pub due_ms: i64,
63    /// The event delivered when it fires.
64    pub event: Event,
65    /// The address the event is delivered to.
66    pub recipient: String,
67    /// Set for a reminder.
68    pub reminder: Option<ReminderSpec>,
69    /// Set for a request timeout: the request's correlation id.
70    pub correlation: Option<String>,
71}
72
73impl TimerRecord {
74    /// The message id of occurrence `n` (0 for a one-shot timer).
75    pub fn message_id(&self, n: u64) -> String {
76        if self.reminder.is_some() {
77            format!("{}/{n}", self.iri)
78        } else {
79            self.iri.clone()
80        }
81    }
82
83    /// The due time of occurrence `n` of a reminder (`n` from 0), or of the timer.
84    // @tg: implements:: [[openspec:durable-timers#Timer Time Source]]
85    pub fn due_of(&self, n: u64) -> i64 {
86        match &self.reminder {
87            Some(r) => self.due_ms.saturating_add(
88                r.period_ms
89                    .saturating_mul(i64::try_from(n).unwrap_or(i64::MAX)),
90            ),
91            None => self.due_ms,
92        }
93    }
94}
95
96/// An outgoing send recorded in an entry; dispatched by the outbox after the append.
97#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
98pub struct SendRecord {
99    /// The send IRI `<entry>/send/<k>`, which is also the message id it is delivered with.
100    pub iri: String,
101    /// The send id.
102    pub send_id: String,
103    /// The event.
104    pub event: Event,
105    /// The target as the chart gave it.
106    pub target: Option<String>,
107    /// The resolved recipient address, when the target names one.
108    pub recipient: Option<String>,
109    /// The Event I/O Processor type.
110    pub type_: String,
111    /// A request's correlation id, or the correlation id a reply answers.
112    pub correlation: Option<String>,
113}
114
115/// The accepted input of an entry.
116#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
117pub struct InputRecord {
118    /// The event as delivered (origin, origin type, send id, invoke id, payload).
119    pub event: Event,
120    /// The deduplication key.
121    pub message_id: String,
122    /// The correlation id the message carried (a request, or a reply).
123    pub correlation: Option<String>,
124    /// The timer whose firing this input is.
125    pub timer: Option<String>,
126}
127
128/// A snapshot of an actor at a journal position.
129// @tg: implements:: [[openspec:event-journal#Snapshots]]
130#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
131pub struct SnapshotData {
132    /// The snapshot IRI.
133    pub iri: String,
134    /// The actor.
135    pub actor: String,
136    /// The position after which it was taken.
137    pub position: u64,
138    /// The chart version it is valid under.
139    pub version: String,
140    /// The configuration digest at that position.
141    pub digest: String,
142    /// The interpreter session as JSON.
143    pub session: String,
144    /// The datamodel snapshot as JSON (Vertep value encoding).
145    pub datamodel: String,
146    /// The runtime state (timers, reminders, requests) as JSON.
147    pub runtime: String,
148}
149
150/// What an input entry records besides its input.
151#[derive(Clone, Debug, Default, PartialEq)]
152pub struct Effects {
153    /// Outgoing sends, in order.
154    pub sends: Vec<SendRecord>,
155    /// Timers scheduled.
156    pub timers: Vec<TimerRecord>,
157    /// IRIs of timers cancelled.
158    pub cancels: Vec<String>,
159    /// Invocations started, as JSON.
160    pub invokes: Vec<String>,
161    /// Correlation ids of requests issued.
162    pub requests: Vec<String>,
163    /// The correlation id the input answers.
164    pub answers: Option<String>,
165    /// Node IRIs of the transitions taken.
166    pub transitions: Vec<String>,
167    /// Whether the session terminated.
168    pub terminated: bool,
169    /// The unhandled-event policy applied, when the input enabled no transition.
170    pub unhandled: Option<UnhandledPolicy>,
171}
172
173/// The kind-specific content of an entry.
174#[derive(Clone, Debug, PartialEq)]
175pub enum EntryBody {
176    /// One accepted input and its macrostep.
177    Input {
178        /// The input.
179        input: InputRecord,
180        /// Its effects.
181        effects: Effects,
182    },
183    /// A migration.
184    Migration {
185        /// The migration IRI.
186        migration: String,
187        /// The version migrated from.
188        from: String,
189        /// The state before, under `from`.
190        before: SnapshotData,
191        /// The state after, under the entry's version.
192        after: SnapshotData,
193    },
194    /// The first entry of a fork.
195    Fork {
196        /// The source actor.
197        source: String,
198        /// The source position.
199        position: u64,
200        /// The source's state there.
201        snapshot: SnapshotData,
202    },
203}
204
205/// One journal entry, decoded.
206// @tg: implements:: [[openspec:event-journal#Journal Of Accepted Inputs]]
207// @tg: implements:: [[openspec:event-journal#Journal Entry Content]]
208// @tg: implements:: [[openspec:chart-versioning#Version Recorded Per Journal Segment]]
209#[derive(Clone, Debug, PartialEq)]
210pub struct EntryData {
211    /// The actor.
212    pub actor: Address,
213    /// The position.
214    pub position: u64,
215    /// The chart version IRI the entry ran under.
216    pub version: String,
217    /// The message id the entry accepted.
218    pub message_id: String,
219    /// Milliseconds since the epoch, stamped at dequeue.
220    pub timestamp_ms: i64,
221    /// The configuration digest after the entry.
222    pub digest: String,
223    /// The kind-specific content.
224    pub body: EntryBody,
225}
226
227/// The digest of a configuration and datamodel snapshot: SHA-256 (hex) of the active state ids,
228/// the datamodel JSON and the termination flag.
229pub fn digest(active: &[String], datamodel: &Value, terminated: bool) -> String {
230    let mut h = Sha256::new();
231    for s in active {
232        h.update(s.as_bytes());
233        h.update([0u8]);
234    }
235    h.update([1u8]);
236    h.update(json::to_text(datamodel).as_bytes());
237    h.update([u8::from(terminated)]);
238    hex(&h.finalize())
239}
240
241fn int(n: i64) -> Literal {
242    Literal::new_typed_literal(n.to_string(), xsd::INTEGER)
243}
244
245fn uint(n: u64) -> Literal {
246    Literal::new_typed_literal(n.to_string(), xsd::INTEGER)
247}
248
249fn boolean(b: bool) -> Literal {
250    Literal::new_typed_literal(if b { "true" } else { "false" }, xsd::BOOLEAN)
251}
252
253fn str_lit(s: &str) -> Literal {
254    Literal::new_simple_literal(s)
255}
256
257/// Builds triples about one subject.
258struct Out<'a> {
259    g: &'a mut Vec<Triple>,
260}
261
262impl Out<'_> {
263    fn add(&mut self, s: &NamedNode, p: &str, o: impl Into<Term>) {
264        self.g.push(Triple::new(s.clone(), node(p), o.into()));
265    }
266
267    fn event(&mut self, s: &NamedNode, e: &Event) {
268        self.add(s, vt::EVENT_NAME, str_lit(&e.name));
269        self.add(s, vt::EVENT_TYPE, str_lit(event_type_name(e.type_)));
270        if let Some(x) = &e.sendid {
271            self.add(s, vt::SEND_ID, str_lit(x));
272        }
273        if let Some(x) = &e.origin {
274            self.add(s, vt::ORIGIN, str_lit(x));
275        }
276        if let Some(x) = &e.origintype {
277            self.add(s, vt::ORIGIN_TYPE, str_lit(x));
278        }
279        // Written only when set, so journals of events without a verified sender are unchanged.
280        if e.originverified {
281            self.add(s, vt::ORIGIN_VERIFIED, boolean(true));
282        }
283        if let Some(x) = &e.invokeid {
284            self.add(s, vt::INVOKE_ID, str_lit(x));
285        }
286        self.add(s, vt::PAYLOAD, str_lit(&json::to_text(&e.data)));
287    }
288
289    fn timer(&mut self, t: &TimerRecord) {
290        let s = NamedNode::new_unchecked(&t.iri);
291        self.add(
292            &s,
293            vt::RDF_TYPE,
294            node(if t.reminder.is_some() {
295                vt::REMINDER
296            } else {
297                vt::TIMER
298            }),
299        );
300        self.add(&s, vt::ACTOR, NamedNode::new_unchecked(&t.actor));
301        self.add(&s, vt::SEND_ID, str_lit(&t.send_id));
302        self.add(&s, vt::DUE_AT, int(t.due_ms));
303        self.add(&s, vt::RECIPIENT, NamedNode::new_unchecked(&t.recipient));
304        self.add(&s, vt::MESSAGE_ID, str_lit(&t.message_id(0)));
305        self.add(
306            &s,
307            vt::ENVELOPE,
308            str_lit(&serde_json::to_string(t).unwrap_or_default()),
309        );
310        if let Some(r) = &t.reminder {
311            self.add(&s, vt::REMINDER_NAME, str_lit(&r.name));
312            self.add(&s, vt::PERIOD, int(r.period_ms));
313        }
314        if let Some(c) = &t.correlation {
315            self.add(&s, vt::CORRELATION_ID, str_lit(c));
316        }
317        self.add(&s, vt::EVENT_NAME, str_lit(&t.event.name));
318    }
319
320    fn snapshot(&mut self, s: &SnapshotData) {
321        let n = NamedNode::new_unchecked(&s.iri);
322        self.add(&n, vt::RDF_TYPE, node(vt::SNAPSHOT));
323        self.add(&n, vt::SNAPSHOT_OF, NamedNode::new_unchecked(&s.actor));
324        self.add(&n, vt::POSITION, uint(s.position));
325        self.add(&n, vt::CHART_VERSION, NamedNode::new_unchecked(&s.version));
326        self.add(&n, vt::CONFIGURATION_DIGEST, str_lit(&s.digest));
327        self.add(&n, vt::SESSION_STATE, str_lit(&s.session));
328        self.add(&n, vt::DATAMODEL_STATE, str_lit(&s.datamodel));
329        self.add(&n, vt::RUNTIME_STATE, str_lit(&s.runtime));
330    }
331}
332
333/// The triples of a snapshot record.
334pub fn snapshot_triples(s: &SnapshotData) -> Vec<Triple> {
335    let mut g = Vec::new();
336    Out { g: &mut g }.snapshot(s);
337    g
338}
339
340/// The triples describing a pending timer in its actor's current-state graph.
341pub fn timer_triples(t: &TimerRecord) -> Vec<Triple> {
342    let mut g = Vec::new();
343    Out { g: &mut g }.timer(t);
344    g
345}
346
347fn event_type_name(t: EventType) -> &'static str {
348    match t {
349        EventType::External => "external",
350        EventType::Internal => "internal",
351        EventType::Platform => "platform",
352    }
353}
354
355fn event_type_of(s: &str) -> EventType {
356    match s {
357        "internal" => EventType::Internal,
358        "platform" => EventType::Platform,
359        _ => EventType::External,
360    }
361}
362
363impl EntryData {
364    /// The entry IRI.
365    pub fn iri(&self) -> NamedNode {
366        self.actor.entry_iri(self.position)
367    }
368
369    /// The effects of an input entry; empty for other kinds.
370    pub fn effects(&self) -> Option<&Effects> {
371        match &self.body {
372            EntryBody::Input { effects, .. } => Some(effects),
373            _ => None,
374        }
375    }
376
377    /// The entry as an RDF graph for [`vertep_store::JournalStore::append`].
378    pub fn to_entry(&self) -> Entry {
379        let e = self.iri();
380        let mut g = Vec::new();
381        let mut o = Out { g: &mut g };
382        o.add(
383            &e,
384            vt::CHART_VERSION,
385            NamedNode::new_unchecked(&self.version),
386        );
387        o.add(&e, vt::MESSAGE_ID, str_lit(&self.message_id));
388        o.add(&e, vt::TIMESTAMP, int(self.timestamp_ms));
389        o.add(&e, vt::CONFIGURATION_DIGEST, str_lit(&self.digest));
390        match &self.body {
391            EntryBody::Input { input, effects } => {
392                o.add(&e, vt::RDF_TYPE, node(vt::INPUT_ENTRY));
393                o.add(&e, vt::RDF_TYPE, node(vt::MACROSTEP));
394                let i = address::input_iri(&e);
395                o.add(&e, vt::INPUT, i.clone());
396                o.add(&i, vt::RDF_TYPE, node(vt::INPUT_EVENT));
397                o.add(&i, vt::MESSAGE_ID, str_lit(&input.message_id));
398                o.event(&i, &input.event);
399                if let Some(c) = &input.correlation {
400                    o.add(&i, vt::CORRELATION_ID, str_lit(c));
401                }
402                if let Some(t) = &input.timer {
403                    o.add(&e, vt::FIRED_TIMER, NamedNode::new_unchecked(t));
404                }
405                for (k, s) in effects.sends.iter().enumerate() {
406                    let sn = NamedNode::new_unchecked(&s.iri);
407                    o.add(&e, vt::SEND, sn.clone());
408                    o.add(&sn, vt::RDF_TYPE, node(vt::OUTGOING_SEND));
409                    o.add(&sn, vt::ORDER, uint(k as u64));
410                    o.add(&sn, vt::MESSAGE_ID, str_lit(&s.iri));
411                    o.add(&sn, vt::SEND_TYPE, str_lit(&s.type_));
412                    if let Some(t) = &s.target {
413                        o.add(&sn, vt::SEND_TARGET, str_lit(t));
414                    }
415                    if let Some(r) = &s.recipient {
416                        o.add(&sn, vt::RECIPIENT, NamedNode::new_unchecked(r));
417                    }
418                    if let Some(c) = &s.correlation {
419                        o.add(&sn, vt::CORRELATION_ID, str_lit(c));
420                    }
421                    o.add(&sn, vt::EVENT_NAME, str_lit(&s.event.name));
422                    o.add(&sn, vt::PAYLOAD, str_lit(&json::to_text(&s.event.data)));
423                    o.add(
424                        &sn,
425                        vt::ENVELOPE,
426                        str_lit(&serde_json::to_string(s).unwrap_or_default()),
427                    );
428                }
429                for t in &effects.timers {
430                    o.add(&e, vt::SCHEDULES, NamedNode::new_unchecked(&t.iri));
431                    o.timer(t);
432                    o.add(
433                        &NamedNode::new_unchecked(&t.iri),
434                        vt::SCHEDULED_BY,
435                        e.clone(),
436                    );
437                }
438                for c in &effects.cancels {
439                    o.add(&e, vt::CANCELS, NamedNode::new_unchecked(c));
440                }
441                if !effects.invokes.is_empty() {
442                    let arr = format!("[{}]", effects.invokes.join(","));
443                    o.add(&e, vt::INVOKES, str_lit(&arr));
444                }
445                for r in &effects.requests {
446                    o.add(&e, vt::REQUESTS, str_lit(r));
447                }
448                if let Some(a) = &effects.answers {
449                    o.add(&e, vt::ANSWERS, str_lit(a));
450                }
451                if effects.terminated {
452                    o.add(&e, vt::TERMINATED_HERE, boolean(true));
453                }
454                if let Some(p) = effects.unhandled {
455                    o.add(&e, vt::UNHANDLED, str_lit(p.as_str()));
456                }
457                for t in &effects.transitions {
458                    o.add(&e, vt::TOOK_TRANSITION, NamedNode::new_unchecked(t));
459                }
460            }
461            EntryBody::Migration {
462                migration,
463                from,
464                before,
465                after,
466            } => {
467                o.add(&e, vt::RDF_TYPE, node(vt::MIGRATION_ENTRY));
468                o.add(&e, vt::MIGRATION_P, NamedNode::new_unchecked(migration));
469                o.add(
470                    &NamedNode::new_unchecked(migration),
471                    vt::RDF_TYPE,
472                    node(vt::MIGRATION),
473                );
474                o.add(&e, vt::FROM_VERSION, NamedNode::new_unchecked(from));
475                o.add(&e, vt::TO_VERSION, NamedNode::new_unchecked(&self.version));
476                o.add(
477                    &e,
478                    vt::SNAPSHOT_BEFORE,
479                    NamedNode::new_unchecked(&before.iri),
480                );
481                o.add(&e, vt::SNAPSHOT_AFTER, NamedNode::new_unchecked(&after.iri));
482                o.snapshot(before);
483                o.snapshot(after);
484            }
485            EntryBody::Fork {
486                source,
487                position,
488                snapshot,
489            } => {
490                o.add(&e, vt::RDF_TYPE, node(vt::FORK_ENTRY));
491                o.add(&e, vt::FORK_OF, NamedNode::new_unchecked(source));
492                o.add(&e, vt::FORK_POSITION, uint(*position));
493                o.add(&e, vt::SNAPSHOT_P, NamedNode::new_unchecked(&snapshot.iri));
494                o.snapshot(snapshot);
495            }
496        }
497        Entry::new(e, g)
498    }
499
500    /// The graph in which the annotations of this entry's transitions are written.
501    pub fn annotation_graph(&self) -> NamedNode {
502        NamedNode::new_unchecked(format!("{}/annotations", self.iri().as_str()))
503    }
504
505    /// The statements `<actor> vt:tookTransition <transition>`, each annotated through a
506    /// blank-node reifier (tiramemsu refuses reifier IRIs) with `vt:causedBy <entry>`. They are written to the actor's graph store, in
507    /// [`EntryData::annotation_graph`], as RDF 1.2 reifiers when the store supports them and as
508    /// `rdf:Statement` reification otherwise (journal entries themselves hold no triple terms).
509    // @tg: implements:: [[openspec:runtime-ontology#Triple Term Annotations]]
510    pub fn annotations(&self) -> Vec<Annotation> {
511        let e = self.iri();
512        let g = self.annotation_graph();
513        self.effects()
514            .map(|eff| {
515                eff.transitions
516                    .iter()
517                    .map(|t| {
518                        Annotation::new(
519                            g.clone(),
520                            Triple::new(
521                                self.actor.node().clone(),
522                                node(vt::TOOK_TRANSITION),
523                                NamedNode::new_unchecked(t),
524                            ),
525                        )
526                        .with(node(vt::CAUSED_BY), e.clone())
527                    })
528                    .collect()
529            })
530            .unwrap_or_default()
531    }
532
533    /// Decodes a journal record.
534    ///
535    /// # Errors
536    ///
537    /// [`DecodeError`] when a required statement is missing or malformed.
538    pub fn from_record(actor: &Address, rec: &JournalRecord) -> Result<Self, DecodeError> {
539        let g = rec.entry.triples();
540        let e = rec.entry.name().clone();
541        let r = Reader { g };
542        let version = r.iri(&e, vt::CHART_VERSION)?;
543        let message_id = r.string(&e, vt::MESSAGE_ID)?;
544        let timestamp_ms = r.int(&e, vt::TIMESTAMP)?;
545        let digest = r.string(&e, vt::CONFIGURATION_DIGEST)?;
546        let body = if r.has_type(&e, vt::MIGRATION_ENTRY) {
547            let before = r.iri(&e, vt::SNAPSHOT_BEFORE)?;
548            let after = r.iri(&e, vt::SNAPSHOT_AFTER)?;
549            EntryBody::Migration {
550                migration: r.iri(&e, vt::MIGRATION_P)?,
551                from: r.iri(&e, vt::FROM_VERSION)?,
552                before: r.snapshot(&NamedNode::new_unchecked(before))?,
553                after: r.snapshot(&NamedNode::new_unchecked(after))?,
554            }
555        } else if r.has_type(&e, vt::FORK_ENTRY) {
556            let s = r.iri(&e, vt::SNAPSHOT_P)?;
557            EntryBody::Fork {
558                source: r.iri(&e, vt::FORK_OF)?,
559                position: u64::try_from(r.int(&e, vt::FORK_POSITION)?)
560                    .map_err(|_| DecodeError("negative fork position".into()))?,
561                snapshot: r.snapshot(&NamedNode::new_unchecked(s))?,
562            }
563        } else {
564            let i = NamedNode::new_unchecked(r.iri(&e, vt::INPUT)?);
565            let input = InputRecord {
566                event: r.event(&i)?,
567                message_id: r.string(&i, vt::MESSAGE_ID)?,
568                correlation: r.opt_string(&i, vt::CORRELATION_ID),
569                timer: r.opt_iri(&e, vt::FIRED_TIMER),
570            };
571            let mut sends = Vec::new();
572            for s in r.iris(&e, vt::SEND) {
573                let sn = NamedNode::new_unchecked(&s);
574                let order = r.int(&sn, vt::ORDER)?;
575                let j = r.string(&sn, vt::ENVELOPE)?;
576                let rec = serde_json::from_str::<SendRecord>(&j)
577                    .map_err(|x| DecodeError(format!("send {s}: {x}")))?;
578                sends.push((order, rec));
579            }
580            sends.sort_by_key(|(o, _)| *o);
581            let mut timers = Vec::new();
582            for t in r.iris(&e, vt::SCHEDULES) {
583                let tn = NamedNode::new_unchecked(&t);
584                let j = r.string(&tn, vt::ENVELOPE)?;
585                timers.push(
586                    serde_json::from_str::<TimerRecord>(&j)
587                        .map_err(|x| DecodeError(format!("timer {t}: {x}")))?,
588                );
589            }
590            timers.sort_by(|a, b| a.iri.cmp(&b.iri));
591            let mut cancels = r.iris(&e, vt::CANCELS);
592            cancels.sort();
593            let invokes = match r.opt_string(&e, vt::INVOKES) {
594                Some(arr) => serde_json::from_str::<Vec<serde_json::Value>>(&arr)
595                    .map_err(|x| DecodeError(format!("invokes: {x}")))?
596                    .into_iter()
597                    .map(|v| v.to_string())
598                    .collect(),
599                None => Vec::new(),
600            };
601            let mut requests = r.strings(&e, vt::REQUESTS);
602            requests.sort();
603            let mut transitions = r.objects_iri(&e, vt::TOOK_TRANSITION);
604            transitions.sort();
605            EntryBody::Input {
606                input,
607                effects: Effects {
608                    sends: sends.into_iter().map(|(_, s)| s).collect(),
609                    timers,
610                    cancels,
611                    invokes,
612                    requests,
613                    answers: r.opt_string(&e, vt::ANSWERS),
614                    transitions,
615                    terminated: r.opt_bool(&e, vt::TERMINATED_HERE),
616                    unhandled: r
617                        .opt_string(&e, vt::UNHANDLED)
618                        .and_then(|p| UnhandledPolicy::parse(&p)),
619                },
620            }
621        };
622        Ok(EntryData {
623            actor: actor.clone(),
624            position: rec.position.get(),
625            version,
626            message_id,
627            timestamp_ms,
628            digest,
629            body,
630        })
631    }
632}
633
634/// Reads statements of one graph.
635pub(crate) struct Reader<'a> {
636    pub(crate) g: &'a Graph,
637}
638
639impl Reader<'_> {
640    fn objects<'b>(
641        &'b self,
642        s: &'b NamedNode,
643        p: &'static str,
644    ) -> impl Iterator<Item = TermRef<'b>> + 'b {
645        let p = NamedNodeRef::new_unchecked(p);
646        self.g.objects_for_subject_predicate(s, p)
647    }
648
649    pub(crate) fn has_type(&self, s: &NamedNode, class: &str) -> bool {
650        self.objects(s, vt::RDF_TYPE)
651            .any(|o| matches!(o, TermRef::NamedNode(n) if n.as_str() == class))
652    }
653
654    pub(crate) fn opt_string(&self, s: &NamedNode, p: &'static str) -> Option<String> {
655        self.objects(s, p).find_map(|o| match o {
656            TermRef::Literal(l) => Some(l.value().to_owned()),
657            _ => None,
658        })
659    }
660
661    pub(crate) fn string(&self, s: &NamedNode, p: &'static str) -> Result<String, DecodeError> {
662        self.opt_string(s, p)
663            .ok_or_else(|| DecodeError(format!("{s} has no {p}")))
664    }
665
666    fn strings(&self, s: &NamedNode, p: &'static str) -> Vec<String> {
667        self.objects(s, p)
668            .filter_map(|o| match o {
669                TermRef::Literal(l) => Some(l.value().to_owned()),
670                _ => None,
671            })
672            .collect()
673    }
674
675    pub(crate) fn int(&self, s: &NamedNode, p: &'static str) -> Result<i64, DecodeError> {
676        self.string(s, p)?
677            .parse()
678            .map_err(|_| DecodeError(format!("{s} {p} is not an integer")))
679    }
680
681    fn opt_bool(&self, s: &NamedNode, p: &'static str) -> bool {
682        self.opt_string(s, p)
683            .is_some_and(|v| v == "true" || v == "1")
684    }
685
686    pub(crate) fn opt_iri(&self, s: &NamedNode, p: &'static str) -> Option<String> {
687        self.objects(s, p).find_map(|o| match o {
688            TermRef::NamedNode(n) => Some(n.as_str().to_owned()),
689            _ => None,
690        })
691    }
692
693    pub(crate) fn iri(&self, s: &NamedNode, p: &'static str) -> Result<String, DecodeError> {
694        self.opt_iri(s, p)
695            .ok_or_else(|| DecodeError(format!("{s} has no {p}")))
696    }
697
698    fn iris(&self, s: &NamedNode, p: &'static str) -> Vec<String> {
699        self.objects(s, p)
700            .filter_map(|o| match o {
701                TermRef::NamedNode(n) => Some(n.as_str().to_owned()),
702                _ => None,
703            })
704            .collect()
705    }
706
707    fn objects_iri(&self, s: &NamedNode, p: &'static str) -> Vec<String> {
708        let set: BTreeSet<String> = self.iris(s, p).into_iter().collect();
709        set.into_iter().collect()
710    }
711
712    fn event(&self, s: &NamedNode) -> Result<Event, DecodeError> {
713        let data = json::from_text(&self.string(s, vt::PAYLOAD)?)
714            .map_err(|e| DecodeError(format!("{s} payload: {e}")))?;
715        Ok(Event {
716            name: self.string(s, vt::EVENT_NAME)?,
717            type_: event_type_of(&self.opt_string(s, vt::EVENT_TYPE).unwrap_or_default()),
718            sendid: self.opt_string(s, vt::SEND_ID),
719            origin: self.opt_string(s, vt::ORIGIN),
720            origintype: self.opt_string(s, vt::ORIGIN_TYPE),
721            originverified: self.opt_bool(s, vt::ORIGIN_VERIFIED),
722            invokeid: self.opt_string(s, vt::INVOKE_ID),
723            data,
724        })
725    }
726
727    pub(crate) fn snapshot(&self, n: &NamedNode) -> Result<SnapshotData, DecodeError> {
728        Ok(SnapshotData {
729            iri: n.as_str().to_owned(),
730            actor: self.iri(n, vt::SNAPSHOT_OF)?,
731            position: u64::try_from(self.int(n, vt::POSITION)?)
732                .map_err(|_| DecodeError("negative snapshot position".into()))?,
733            version: self.iri(n, vt::CHART_VERSION)?,
734            digest: self.string(n, vt::CONFIGURATION_DIGEST)?,
735            session: self.string(n, vt::SESSION_STATE)?,
736            datamodel: self.string(n, vt::DATAMODEL_STATE)?,
737            runtime: self.string(n, vt::RUNTIME_STATE)?,
738        })
739    }
740}
741
742#[cfg(test)]
743mod tests {
744    use super::*;
745    use vertep_store::Position;
746
747    fn sample() -> EntryData {
748        let a = Address::parse("urn:vertep:test:actor").unwrap();
749        let e = a.entry_iri(1);
750        EntryData {
751            actor: a.clone(),
752            position: 1,
753            version: "urn:vertep:chart:sha256:00".into(),
754            message_id: "m1".into(),
755            timestamp_ms: 1_000,
756            digest: "d".into(),
757            body: EntryBody::Input {
758                input: InputRecord {
759                    event: Event {
760                        name: "go".into(),
761                        data: Value::Object(vec![("x".into(), Value::Number(1.0))]),
762                        origin: Some("urn:vertep:test:sender".into()),
763                        ..Event::default()
764                    },
765                    message_id: "m1".into(),
766                    correlation: None,
767                    timer: None,
768                },
769                effects: Effects {
770                    sends: vec![SendRecord {
771                        iri: address::send_iri(&e, 0).as_str().into(),
772                        send_id: "s1".into(),
773                        event: Event::named("out", EventType::External),
774                        target: Some("urn:vertep:test:b".into()),
775                        recipient: Some("urn:vertep:test:b".into()),
776                        type_: "scxml".into(),
777                        correlation: None,
778                    }],
779                    timers: vec![TimerRecord {
780                        iri: address::timer_iri(&e, "r1").as_str().into(),
781                        actor: a.as_str().into(),
782                        send_id: "r1".into(),
783                        due_ms: 6_000,
784                        event: Event::named("remind", EventType::External),
785                        recipient: a.as_str().into(),
786                        reminder: None,
787                        correlation: None,
788                    }],
789                    transitions: vec!["urn:vertep:chart:sha256:00#s.a:t0".into()],
790                    ..Effects::default()
791                },
792            },
793        }
794    }
795
796    #[test]
797    fn entry_round_trips_through_rdf() {
798        let d = sample();
799        let entry = d
800            .to_entry()
801            .stamp(d.actor.node(), Position::new(1))
802            .unwrap();
803        let rec = JournalRecord {
804            position: Position::new(1),
805            entry,
806        };
807        assert_eq!(EntryData::from_record(&d.actor, &rec).unwrap(), d);
808        assert_eq!(d.annotations().len(), 1);
809    }
810
811    // @tg: verifies:: [[openspec:did-identity#Journaled Verification Outcome#Replay after key rotation]] {level: unit}
812    #[test]
813    fn verified_origin_round_trips_through_the_journal() {
814        let mut d = sample();
815        let EntryBody::Input { input, .. } = &mut d.body else {
816            unreachable!()
817        };
818        input.event = Event::named("go", EventType::External)
819            .with_verified_origin("did:key:z6MkhaXgBZDvotDkL5257faiztiGiC2QtKLGpbnnEGta2doK");
820        let entry = d
821            .to_entry()
822            .stamp(d.actor.node(), Position::new(1))
823            .unwrap();
824        let rec = JournalRecord {
825            position: Position::new(1),
826            entry,
827        };
828        let back = EntryData::from_record(&d.actor, &rec).unwrap();
829        assert_eq!(back, d);
830        let EntryBody::Input { input, .. } = &back.body else {
831            unreachable!()
832        };
833        assert!(input.event.originverified);
834    }
835
836    #[test]
837    fn digest_depends_on_configuration_and_data() {
838        let a = digest(&["s1".into()], &Value::Undefined, false);
839        assert_ne!(a, digest(&["s2".into()], &Value::Undefined, false));
840        assert_ne!(a, digest(&["s1".into()], &Value::Null, false));
841        assert_eq!(a, digest(&["s1".into()], &Value::Undefined, false));
842    }
843}
vertep_runtime/journal.rs

vertep_runtime/
journal.rs
