surreal_sync_runtime/pipeline/apply/
event.rs

1//! Unified apply events: row changes and relation (graph edge) changes.
2
3use surreal_sync_core::{Change, RelationChange};
4
5/// One item in the apply buffer / transform window / ordered sink queue.
6///
7/// Sources may interleave row CDC and relation CDC; the apply engine preserves
8/// source order across both kinds through the same max_in_flight window.
9///
10/// [`RelationChange`](Self::RelationChange) is boxed to keep the enum compact
11/// (`RelationChange` is substantially larger than [`Change`]).
12#[derive(Debug, Clone)]
13pub enum ApplyEvent {
14    /// Row create / update / delete.
15    Change(Change),
16    /// Graph-edge create / update / delete ([`RelationChange`]).
17    RelationChange(Box<RelationChange>),
18}
19
20impl ApplyEvent {
21    /// Wrap a row change.
22    pub fn change(change: Change) -> Self {
23        Self::Change(change)
24    }
25
26    /// Wrap a relation change.
27    pub fn relation_change(change: RelationChange) -> Self {
28        Self::RelationChange(Box::new(change))
29    }
30
31    /// Whether this is a row change.
32    pub fn is_change(&self) -> bool {
33        matches!(self, Self::Change(_))
34    }
35
36    /// Whether this is a relation change.
37    pub fn is_relation_change(&self) -> bool {
38        matches!(self, Self::RelationChange(_))
39    }
40}
41
42/// A source event plus the position to advance after sink success.
43#[derive(Debug, Clone)]
44pub struct PositionedEvent<P> {
45    /// Event to transform and apply.
46    pub event: ApplyEvent,
47    /// Source position associated with this event (checkpoint candidate).
48    pub position: P,
49}
50
51impl<P> PositionedEvent<P> {
52    /// Construct a positioned event.
53    pub fn new(event: ApplyEvent, position: P) -> Self {
54        Self { event, position }
55    }
56
57    /// Positioned row change.
58    pub fn change(change: Change, position: P) -> Self {
59        Self::new(ApplyEvent::Change(change), position)
60    }
61
62    /// Positioned relation change.
63    pub fn relation_change(change: RelationChange, position: P) -> Self {
64        Self::new(ApplyEvent::relation_change(change), position)
65    }
66}
surreal_sync_runtime/pipeline/apply/event.rs

surreal_sync_runtime/pipeline/apply/
event.rs
