1use super::behavior::{Behavior, BehaviorContext};
2use super::graph::{record_hash, replay_strict, StructuralHash};
3use super::{
4 GraphDiff, GraphEvent, GraphEventRecord, GraphObject, GraphPatch, GraphRelation,
5 PatchOperation, ReplayError, StateGraph, GRAPH_EVENT_SCHEMA_VERSION,
6};
7use std::collections::{BTreeMap, VecDeque};
8use std::sync::Arc;
9use thiserror::Error;
10
11#[derive(Debug, Clone, Copy)]
12pub struct RuntimeLimits {
13 pub max_events: usize,
14 pub max_behavior_depth: usize,
15}
16
17impl Default for RuntimeLimits {
18 fn default() -> Self {
19 Self {
20 max_events: 10_000,
21 max_behavior_depth: 64,
22 }
23 }
24}
25
26pub struct GraphRuntime {
27 branch_id: String,
28 correlation_id: Option<String>,
29 graph: StateGraph,
30 structural_hash: StructuralHash,
31 events: Vec<GraphEventRecord>,
32 behaviors: Vec<Arc<dyn Behavior>>,
33 pending: VecDeque<(usize, GraphEventRecord)>,
34 limits: RuntimeLimits,
35 external_cursors: BTreeMap<(String, String), ExternalCursor>,
36}
37
38#[derive(Debug, Clone, PartialEq, Eq)]
39struct ExternalCursor {
40 sequence: u64,
41 event_id: String,
42 observed: BTreeMap<u64, String>,
43}
44
45#[derive(Debug, Clone, Copy, PartialEq, Eq)]
46pub enum ExternalProjectionOutcome {
47 Applied,
48 Duplicate,
49}
50
51#[derive(Debug, Clone, PartialEq)]
52pub struct ExternalEvent {
53 pub source: String,
54 pub stream_id: String,
55 pub sequence: u64,
56 pub event_id: String,
57 pub name: String,
58 pub payload: serde_json::Value,
59}
60
61impl GraphRuntime {
62 pub fn new() -> Self {
63 Self::with_limits(RuntimeLimits::default())
64 }
65
66 pub fn with_limits(limits: RuntimeLimits) -> Self {
67 Self {
68 branch_id: new_id("branch"),
69 correlation_id: None,
70 graph: StateGraph::default(),
71 structural_hash: StructuralHash::default(),
72 events: Vec::new(),
73 behaviors: Vec::new(),
74 pending: VecDeque::new(),
75 limits,
76 external_cursors: BTreeMap::new(),
77 }
78 }
79
80 pub fn with_correlation_id(mut self, correlation_id: impl Into<String>) -> Self {
81 self.correlation_id = Some(correlation_id.into());
82 self
83 }
84
85 pub fn restore(events: Vec<GraphEventRecord>) -> Result<Self, ReplayError> {
86 let graph = replay_strict(&events)?;
87 let structural_hash = StructuralHash::from_graph(&graph)?;
88 let external_cursors = external_cursors(&events)?;
89 let branch_id = events
90 .last()
91 .map(|record| record.branch_id.clone())
92 .unwrap_or_else(|| new_id("branch"));
93 let correlation_id = events
94 .iter()
95 .rev()
96 .find_map(|record| record.correlation_id.clone());
97 Ok(Self {
98 branch_id,
99 correlation_id,
100 graph,
101 structural_hash,
102 events,
103 behaviors: Vec::new(),
104 pending: VecDeque::new(),
105 limits: RuntimeLimits::default(),
106 external_cursors,
107 })
108 }
109
110 pub fn branch_id(&self) -> &str {
111 &self.branch_id
112 }
113
114 pub fn graph(&self) -> &StateGraph {
115 &self.graph
116 }
117
118 pub fn events(&self) -> &[GraphEventRecord] {
119 &self.events
120 }
121
122 pub fn register(&mut self, behavior: Arc<dyn Behavior>) -> Result<(), RuntimeError> {
123 if self
124 .behaviors
125 .iter()
126 .any(|registered| registered.name() == behavior.name())
127 {
128 return Err(RuntimeError::DuplicateBehavior(behavior.name().to_string()));
129 }
130 self.behaviors.push(behavior);
131 Ok(())
132 }
133
134 pub fn emit(&mut self, event: GraphEvent) -> Result<GraphEventRecord, RuntimeError> {
135 let record = self.append(event, None)?;
136 self.pending.push_back((0, record.clone()));
137 self.drain_behaviors()?;
138 Ok(record)
139 }
140
141 pub fn run_goal(&mut self, goal: impl Into<String>) -> Result<GraphEventRecord, RuntimeError> {
142 self.emit(GraphEvent::GoalCreated { goal: goal.into() })
143 }
144
145 pub fn propose_patch(
146 &mut self,
147 patch: GraphPatch,
148 causation_id: Option<String>,
149 ) -> Result<bool, RuntimeError> {
150 let records = self.apply_patch(patch, causation_id, 0)?;
151 self.drain_behaviors()?;
152 Ok(records)
153 }
154
155 pub fn fork_at(&self, sequence_exclusive: u64) -> Result<Self, RuntimeError> {
156 let end = usize::try_from(sequence_exclusive)
157 .map_err(|_| RuntimeError::InvalidFork(sequence_exclusive))?;
158 if end > self.events.len() {
159 return Err(RuntimeError::InvalidFork(sequence_exclusive));
160 }
161 let events = self.events[..end].to_vec();
162 let graph = replay_strict(&events)?;
163 let structural_hash = StructuralHash::from_graph(&graph)?;
164 let external_cursors = external_cursors(&events)?;
165 let parent_branch_id = self.branch_id.clone();
166 let mut fork = Self {
167 branch_id: new_id("branch"),
168 correlation_id: self.correlation_id.clone(),
169 graph,
170 structural_hash,
171 events,
172 behaviors: self.behaviors.clone(),
173 pending: VecDeque::new(),
174 limits: self.limits,
175 external_cursors,
176 };
177 let cause = fork.events.last().map(|record| record.id.clone());
178 fork.append(
179 GraphEvent::BranchForked {
180 parent_branch_id,
181 fork_sequence: sequence_exclusive,
182 },
183 cause,
184 )?;
185 Ok(fork)
186 }
187
188 pub fn diff(&self, other: &Self) -> GraphDiff {
189 self.graph.diff(&other.graph)
190 }
191
192 pub fn strict_replay(records: &[GraphEventRecord]) -> Result<StateGraph, ReplayError> {
193 external_cursors(records)?;
194 replay_strict(records)
195 }
196
197 pub fn check_external(
199 &self,
200 event: &ExternalEvent,
201 ) -> Result<Option<ExternalProjectionOutcome>, RuntimeError> {
202 let key = (event.source.clone(), event.stream_id.clone());
203 let expected = self
204 .external_cursors
205 .get(&key)
206 .map_or(1, |cursor| cursor.sequence.saturating_add(1));
207 if let Some(cursor) = self.external_cursors.get(&key) {
208 if event.sequence <= cursor.sequence {
209 if cursor.observed.get(&event.sequence) == Some(&event.event_id) {
210 return Ok(Some(ExternalProjectionOutcome::Duplicate));
211 }
212 return Err(RuntimeError::ExternalEventConflict {
213 source_name: event.source.clone(),
214 stream_id: event.stream_id.clone(),
215 sequence: event.sequence,
216 });
217 }
218 }
219 if event.sequence != expected {
220 return Err(RuntimeError::ExternalSequenceDiverged {
221 source_name: event.source.clone(),
222 stream_id: event.stream_id.clone(),
223 expected,
224 actual: event.sequence,
225 });
226 }
227 Ok(None)
228 }
229
230 pub fn project_external(
232 &mut self,
233 event: ExternalEvent,
234 patch: GraphPatch,
235 ) -> Result<ExternalProjectionOutcome, RuntimeError> {
236 let key = (event.source.clone(), event.stream_id.clone());
237 if let Some(outcome) = self.check_external(&event)? {
238 return Ok(outcome);
239 }
240 let mutation_events = self
241 .validate_patch(&patch)
242 .map_err(RuntimeError::InvalidExternalProjection)?;
243 let required = mutation_events.len().saturating_add(3);
244 if self.events.len().saturating_add(required) > self.limits.max_events {
245 return Err(RuntimeError::EventLimitExceeded(self.limits.max_events));
246 }
247
248 let patch_id = new_id("patch");
249 let proposed = self.append(
250 GraphEvent::PatchProposed {
251 patch_id: patch_id.clone(),
252 patch,
253 },
254 None,
255 )?;
256 let mut last_cause = proposed.id;
257 for mutation in mutation_events {
258 let record = self.append(mutation, Some(last_cause))?;
259 last_cause = record.id;
260 }
261 let applied = self.append(GraphEvent::PatchApplied { patch_id }, Some(last_cause))?;
262 let observed = self.append(
263 GraphEvent::ExternalEventObserved {
264 source: event.source.clone(),
265 stream_id: event.stream_id.clone(),
266 sequence: event.sequence,
267 event_id: event.event_id.clone(),
268 name: event.name,
269 payload: event.payload,
270 },
271 Some(applied.id),
272 )?;
273 let cursor = self
274 .external_cursors
275 .entry(key)
276 .or_insert_with(|| ExternalCursor {
277 sequence: 0,
278 event_id: String::new(),
279 observed: BTreeMap::new(),
280 });
281 cursor.sequence = event.sequence;
282 cursor.event_id = event.event_id.clone();
283 cursor.observed.insert(event.sequence, event.event_id);
284 self.pending.push_back((0, observed));
285 self.drain_behaviors()?;
286 Ok(ExternalProjectionOutcome::Applied)
287 }
288
289 fn drain_behaviors(&mut self) -> Result<(), RuntimeError> {
290 while let Some((depth, triggering_event)) = self.pending.pop_front() {
291 if depth >= self.limits.max_behavior_depth {
292 return Err(RuntimeError::BehaviorDepthExceeded(
293 self.limits.max_behavior_depth,
294 ));
295 }
296 let matching = self
297 .behaviors
298 .iter()
299 .filter(|behavior| behavior.filter().matches(&triggering_event, &self.graph))
300 .cloned()
301 .collect::<Vec<_>>();
302 for behavior in matching {
303 let cause = Some(triggering_event.id.clone());
304 self.append(
305 GraphEvent::BehaviorStarted {
306 name: behavior.name().to_string(),
307 },
308 cause.clone(),
309 )?;
310 let result = behavior.evaluate(BehaviorContext {
311 graph: &self.graph,
312 event: &triggering_event,
313 });
314 match result {
315 Ok(patches) => {
316 for patch in patches {
317 self.apply_patch(patch, cause.clone(), depth + 1)?;
318 }
319 self.append(
320 GraphEvent::BehaviorCompleted {
321 name: behavior.name().to_string(),
322 },
323 cause,
324 )?;
325 }
326 Err(error) => {
327 self.append(
328 GraphEvent::BehaviorFailed {
329 name: behavior.name().to_string(),
330 error: error.to_string(),
331 },
332 cause,
333 )?;
334 }
335 }
336 }
337 }
338 Ok(())
339 }
340
341 fn apply_patch(
342 &mut self,
343 patch: GraphPatch,
344 causation_id: Option<String>,
345 depth: usize,
346 ) -> Result<bool, RuntimeError> {
347 let patch_id = new_id("patch");
348 let validation = self.validate_patch(&patch);
349 let required_events = validation
350 .as_ref()
351 .map_or(2, |mutation_events| mutation_events.len().saturating_add(2));
352 if self.events.len().saturating_add(required_events) > self.limits.max_events {
353 return Err(RuntimeError::EventLimitExceeded(self.limits.max_events));
354 }
355 let proposed = self.append(
356 GraphEvent::PatchProposed {
357 patch_id: patch_id.clone(),
358 patch: patch.clone(),
359 },
360 causation_id,
361 )?;
362 self.pending.push_back((depth, proposed.clone()));
363
364 let mutation_events = match validation {
365 Ok(events) => events,
366 Err(reason) => {
367 let rejected = self.append(
368 GraphEvent::PatchRejected { patch_id, reason },
369 Some(proposed.id),
370 )?;
371 self.pending.push_back((depth, rejected));
372 return Ok(false);
373 }
374 };
375
376 let mut last_cause = proposed.id;
377 for event in mutation_events {
378 let record = self.append(event, Some(last_cause))?;
379 last_cause = record.id.clone();
380 self.pending.push_back((depth, record));
381 }
382 let applied = self.append(GraphEvent::PatchApplied { patch_id }, Some(last_cause))?;
383 self.pending.push_back((depth, applied));
384 Ok(true)
385 }
386
387 fn validate_patch(&self, patch: &GraphPatch) -> Result<Vec<GraphEvent>, String> {
388 if patch.expected_graph_version != self.graph.version() {
389 return Err(format!(
390 "graph version conflict: expected {}, current {}",
391 patch.expected_graph_version,
392 self.graph.version()
393 ));
394 }
395 let mut candidate = PatchValidationView::new(&self.graph);
396 let mut events = Vec::with_capacity(patch.operations.len());
397 for operation in &patch.operations {
398 events.push(candidate.apply(operation)?);
399 }
400 Ok(events)
401 }
402
403 fn append(
404 &mut self,
405 event: GraphEvent,
406 causation_id: Option<String>,
407 ) -> Result<GraphEventRecord, RuntimeError> {
408 if self.events.len() >= self.limits.max_events {
409 return Err(RuntimeError::EventLimitExceeded(self.limits.max_events));
410 }
411 let state_version_before = self.graph.version();
412 let mut structural_hash = self.structural_hash;
413 structural_hash.apply(&event, &self.graph)?;
414 self.graph.apply(&event)?;
415 self.structural_hash = structural_hash;
416 let state_hash_after = if self.graph.version() == state_version_before {
417 match self.events.last() {
418 Some(record) if record.schema_version == GRAPH_EVENT_SCHEMA_VERSION => {
419 record.state_hash_after.clone()
420 }
421 _ => self.structural_hash.digest(self.graph.version())?,
422 }
423 } else {
424 self.structural_hash.digest(self.graph.version())?
425 };
426 let mut record = GraphEventRecord {
427 schema_version: GRAPH_EVENT_SCHEMA_VERSION,
428 id: new_id("event"),
429 sequence: self.events.len() as u64,
430 timestamp_ms: now_ms(),
431 branch_id: self.branch_id.clone(),
432 causation_id,
433 correlation_id: self.correlation_id.clone(),
434 state_version_before,
435 state_version_after: self.graph.version(),
436 state_hash_after,
437 previous_record_hash: self.events.last().map(|record| record.record_hash.clone()),
438 record_hash: String::new(),
439 event,
440 };
441 record.record_hash = record_hash(&record)?;
442 self.events.push(record.clone());
443 Ok(record)
444 }
445}
446
447impl Default for GraphRuntime {
448 fn default() -> Self {
449 Self::new()
450 }
451}
452
453struct PatchValidationView<'a> {
454 graph: &'a StateGraph,
455 objects: BTreeMap<String, Option<GraphObject>>,
456 relations: BTreeMap<String, Option<GraphRelation>>,
457}
458
459impl<'a> PatchValidationView<'a> {
460 fn new(graph: &'a StateGraph) -> Self {
461 Self {
462 graph,
463 objects: BTreeMap::new(),
464 relations: BTreeMap::new(),
465 }
466 }
467
468 fn object(&self, id: &str) -> Option<&GraphObject> {
469 self.objects
470 .get(id)
471 .map_or_else(|| self.graph.object(id), Option::as_ref)
472 }
473
474 fn relation(&self, id: &str) -> Option<&GraphRelation> {
475 self.relations
476 .get(id)
477 .map_or_else(|| self.graph.relation(id), Option::as_ref)
478 }
479
480 fn object_has_relations(&self, id: &str) -> bool {
481 self.graph.relations().any(|base| {
482 self.relation(&base.id)
483 .is_some_and(|relation| relation.source == id || relation.target == id)
484 }) || self.relations.values().flatten().any(|relation| {
485 self.graph.relation(&relation.id).is_none()
486 && (relation.source == id || relation.target == id)
487 })
488 }
489
490 fn apply(&mut self, operation: &PatchOperation) -> Result<GraphEvent, String> {
491 Ok(match operation {
492 PatchOperation::AddObject {
493 id,
494 object_type,
495 data,
496 } => {
497 if self.object(id).is_some() {
498 return Err(format!("object `{id}` already exists"));
499 }
500 self.objects.insert(
501 id.clone(),
502 Some(GraphObject {
503 id: id.clone(),
504 object_type: object_type.clone(),
505 data: data.clone(),
506 version: 1,
507 }),
508 );
509 GraphEvent::ObjectCreated {
510 id: id.clone(),
511 object_type: object_type.clone(),
512 data: data.clone(),
513 }
514 }
515 PatchOperation::UpdateObject {
516 id,
517 expected_version,
518 data,
519 } => {
520 let current = self
521 .object(id)
522 .cloned()
523 .ok_or_else(|| format!("object `{id}` does not exist"))?;
524 if current.version != *expected_version {
525 return Err(format!(
526 "object `{id}` version conflict: expected {expected_version}, current {}",
527 current.version
528 ));
529 }
530 let version = current.version + 1;
531 self.objects.insert(
532 id.clone(),
533 Some(GraphObject {
534 data: data.clone(),
535 version,
536 ..current
537 }),
538 );
539 GraphEvent::ObjectUpdated {
540 id: id.clone(),
541 version,
542 data: data.clone(),
543 }
544 }
545 PatchOperation::RemoveObject {
546 id,
547 expected_version,
548 } => {
549 let current = self
550 .object(id)
551 .cloned()
552 .ok_or_else(|| format!("object `{id}` does not exist"))?;
553 if current.version != *expected_version {
554 return Err(format!("object `{id}` version conflict"));
555 }
556 if self.object_has_relations(id) {
557 return Err(format!("object `{id}` still has relations"));
558 }
559 self.objects.insert(id.clone(), None);
560 GraphEvent::ObjectRemoved {
561 id: id.clone(),
562 version: current.version + 1,
563 }
564 }
565 PatchOperation::AddRelation {
566 id,
567 relation_type,
568 source,
569 target,
570 data,
571 } => {
572 if self.relation(id).is_some() {
573 return Err(format!("relation `{id}` already exists"));
574 }
575 if self.object(source).is_none() || self.object(target).is_none() {
576 return Err(format!("relation `{id}` has a missing endpoint"));
577 }
578 self.relations.insert(
579 id.clone(),
580 Some(GraphRelation {
581 id: id.clone(),
582 relation_type: relation_type.clone(),
583 source: source.clone(),
584 target: target.clone(),
585 data: data.clone(),
586 version: 1,
587 }),
588 );
589 GraphEvent::RelationCreated {
590 id: id.clone(),
591 relation_type: relation_type.clone(),
592 source: source.clone(),
593 target: target.clone(),
594 data: data.clone(),
595 }
596 }
597 PatchOperation::UpdateRelation {
598 id,
599 expected_version,
600 data,
601 } => {
602 let current = self
603 .relation(id)
604 .cloned()
605 .ok_or_else(|| format!("relation `{id}` does not exist"))?;
606 if current.version != *expected_version {
607 return Err(format!("relation `{id}` version conflict"));
608 }
609 let version = current.version + 1;
610 self.relations.insert(
611 id.clone(),
612 Some(GraphRelation {
613 data: data.clone(),
614 version,
615 ..current
616 }),
617 );
618 GraphEvent::RelationUpdated {
619 id: id.clone(),
620 version,
621 data: data.clone(),
622 }
623 }
624 PatchOperation::RemoveRelation {
625 id,
626 expected_version,
627 } => {
628 let current = self
629 .relation(id)
630 .cloned()
631 .ok_or_else(|| format!("relation `{id}` does not exist"))?;
632 if current.version != *expected_version {
633 return Err(format!("relation `{id}` version conflict"));
634 }
635 self.relations.insert(id.clone(), None);
636 GraphEvent::RelationRemoved {
637 id: id.clone(),
638 version: current.version + 1,
639 }
640 }
641 })
642 }
643}
644
645fn new_id(prefix: &str) -> String {
646 format!("{prefix}-{}", uuid::Uuid::new_v4())
647}
648
649fn now_ms() -> u64 {
650 use std::time::{SystemTime, UNIX_EPOCH};
651 SystemTime::now()
652 .duration_since(UNIX_EPOCH)
653 .unwrap_or_default()
654 .as_millis()
655 .min(u128::from(u64::MAX)) as u64
656}
657
658fn external_cursors(
659 events: &[GraphEventRecord],
660) -> Result<BTreeMap<(String, String), ExternalCursor>, ReplayError> {
661 let mut cursors: BTreeMap<(String, String), ExternalCursor> = BTreeMap::new();
662 for record in events {
663 let GraphEvent::ExternalEventObserved {
664 source,
665 stream_id,
666 sequence,
667 event_id,
668 ..
669 } = &record.event
670 else {
671 continue;
672 };
673 let key = (source.clone(), stream_id.clone());
674 let expected = cursors
675 .get(&key)
676 .map_or(1, |cursor| cursor.sequence.saturating_add(1));
677 if *sequence != expected {
678 return Err(ReplayError::InvalidMutation(format!(
679 "external stream `{source}/{stream_id}` sequence diverged: expected {expected}, got {sequence}"
680 )));
681 }
682 let cursor = cursors.entry(key).or_insert_with(|| ExternalCursor {
683 sequence: 0,
684 event_id: String::new(),
685 observed: BTreeMap::new(),
686 });
687 cursor.sequence = *sequence;
688 cursor.event_id = event_id.clone();
689 cursor.observed.insert(*sequence, event_id.clone());
690 }
691 Ok(cursors)
692}
693
694#[derive(Debug, Error)]
695pub enum RuntimeError {
696 #[error(transparent)]
697 Replay(#[from] ReplayError),
698 #[error("behavior `{0}` is already registered")]
699 DuplicateBehavior(String),
700 #[error("event limit of {0} exceeded")]
701 EventLimitExceeded(usize),
702 #[error("behavior recursion depth of {0} exceeded")]
703 BehaviorDepthExceeded(usize),
704 #[error("cannot fork at exclusive sequence {0}")]
705 InvalidFork(u64),
706 #[error("external stream `{source_name}/{stream_id}` sequence diverged: expected {expected}, got {actual}")]
707 ExternalSequenceDiverged {
708 source_name: String,
709 stream_id: String,
710 expected: u64,
711 actual: u64,
712 },
713 #[error("external stream `{source_name}/{stream_id}` event at sequence {sequence} conflicts with the observed event id")]
714 ExternalEventConflict {
715 source_name: String,
716 stream_id: String,
717 sequence: u64,
718 },
719 #[error("invalid external projection: {0}")]
720 InvalidExternalProjection(String),
721}