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