ankurah_core/
changes.rs

1use crate::{entity::Entity, error::MutationError, model::View, reactor::ChangeNotification};
2use ankurah_proto::{Attested, Event};
3
4#[derive(Debug, Clone)]
5pub struct EntityChange {
6    entity: Entity,
7    events: Vec<Attested<Event>>,
8}
9
10// Implement the trait for EntityChange
11impl ChangeNotification for EntityChange {
12    type Entity = Entity;
13    type Event = ankurah_proto::Attested<Event>;
14
15    fn into_parts(self) -> (Self::Entity, Vec<Self::Event>) { (self.entity, self.events) }
16    fn entity(&self) -> &Self::Entity { &self.entity }
17    fn events(&self) -> &[Self::Event] { &self.events }
18}
19
20// TODO consider a flattened version of EntityChange that includes the entity and Vec<(operations, parent, attestations)> rather than a Vec<Attested<Event>>
21impl EntityChange {
22    pub fn new(entity: Entity, events: Vec<Attested<Event>>) -> Result<Self, MutationError> {
23        // Every event must belong to this entity and be part of its current
24        // history: either a head tip, or the parent of a later event in the
25        // same batch (an ancestor superseded within an ordered multi-event
26        // batch, e.g. a bridge or a multi-event subscription item). Requiring
27        // head membership alone rejects legitimate parent-then-child batches
28        // after both events applied.
29        let head = entity.head();
30        for (i, event) in events.iter().enumerate() {
31            if event.payload.entity_id != entity.id {
32                return Err(MutationError::InvalidEvent);
33            }
34            let id = event.payload.id();
35            let in_head = head.contains(&id);
36            let superseded_in_batch = events[i + 1..].iter().any(|later| later.payload.parent.contains(&id));
37            if !in_head && !superseded_in_batch {
38                return Err(MutationError::InvalidEvent);
39            }
40        }
41        Ok(Self { entity, events })
42    }
43    pub fn into_parts(self) -> (Entity, Vec<Attested<Event>>) { (self.entity, self.events) }
44}
45
46#[derive(Debug, Clone)]
47pub enum ItemChange<I> {
48    /// Initial retrieval of an item upon subscription
49    Initial { item: I },
50    /// A new item was added OR changed such that it now matches the subscription
51    Add { item: I, events: Vec<Attested<Event>> },
52    /// A item that previously matched the subscription has changed in a way that has not changed the matching condition
53    Update { item: I, events: Vec<Attested<Event>> },
54    /// A item that previously matched the subscription has changed in a way that no longer matches the subscription
55    Remove { item: I, events: Vec<Attested<Event>> },
56}
57
58impl<I> ItemChange<I> {
59    pub fn entity(&self) -> &I {
60        match self {
61            ItemChange::Initial { item }
62            | ItemChange::Add { item, .. }
63            | ItemChange::Update { item, .. }
64            | ItemChange::Remove { item, .. } => item,
65        }
66    }
67
68    pub fn events(&self) -> &[Attested<Event>] {
69        match self {
70            ItemChange::Add { events, .. } | ItemChange::Update { events, .. } | ItemChange::Remove { events, .. } => events,
71            _ => &[],
72        }
73    }
74    pub fn kind(&self) -> ChangeKind { ChangeKind::from(self) }
75}
76
77impl<I> std::fmt::Display for ItemChange<I>
78where I: View
79{
80    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
81        match self {
82            ItemChange::Initial { item } => {
83                write!(f, "Initial {}/{}", I::collection(), item.id())
84            }
85            ItemChange::Add { item, .. } => {
86                write!(f, "Add {}/{}", I::collection(), item.id())
87            }
88            ItemChange::Update { item, .. } => {
89                write!(f, "Update {}/{}", I::collection(), item.id())
90            }
91            ItemChange::Remove { item, .. } => {
92                write!(f, "Remove {}/{}", I::collection(), item.id())
93            }
94        }
95    }
96}
97
98impl std::fmt::Display for EntityChange {
99    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
100        write!(f, "EntityChange {}/{}", self.entity.collection(), self.entity.id())
101    }
102}
103
104use crate::resultset::ResultSet;
105
106#[derive(Debug, Clone)]
107pub struct ChangeSet<R: View> {
108    pub resultset: ResultSet<R>,
109    pub changes: Vec<ItemChange<R>>,
110}
111
112impl<R: View> ChangeSet<R>
113where R: Clone
114{
115    /// Returns items from the initial query load (before subscription was active)
116    pub fn initial(&self) -> Vec<R> {
117        self.changes
118            .iter()
119            .filter_map(|change| match change {
120                ItemChange::Initial { item } => Some(item.clone()),
121                _ => None,
122            })
123            .collect()
124    }
125
126    /// Returns genuinely new items (added after subscription, or now match the predicate)
127    pub fn added(&self) -> Vec<R> {
128        self.changes
129            .iter()
130            .filter_map(|change| match change {
131                ItemChange::Add { item, .. } => Some(item.clone()),
132                _ => None,
133            })
134            .collect()
135    }
136
137    /// Returns all items that appeared in the result set (initial load + newly added)
138    pub fn appeared(&self) -> Vec<R> {
139        self.changes
140            .iter()
141            .filter_map(|change| match change {
142                ItemChange::Add { item, .. } | ItemChange::Initial { item } => Some(item.clone()),
143                _ => None,
144            })
145            .collect()
146    }
147
148    #[deprecated(since = "0.7.10", note = "Use `appeared()`, `initial()`, or `added()` instead")]
149    /// Returns all items that were added or now match the query
150    pub fn adds(&self) -> Vec<R> { self.appeared() }
151
152    /// Returns all items that were removed or no longer match the query
153    pub fn removed(&self) -> Vec<R> {
154        self.changes
155            .iter()
156            .filter_map(|change| match change {
157                ItemChange::Remove { item, .. } => Some(item.clone()),
158                _ => None,
159            })
160            .collect()
161    }
162
163    #[deprecated(since = "0.7.10", note = "Use `removed()` instead")]
164    /// Returns all items that were removed or no longer match the query
165    pub fn removes(&self) -> Vec<R> { self.removed() }
166
167    /// Returns all items that were updated but still match the query
168    pub fn updated(&self) -> Vec<R> {
169        self.changes
170            .iter()
171            .filter_map(|change| match change {
172                ItemChange::Update { item, .. } => Some(item.clone()),
173                _ => None,
174            })
175            .collect()
176    }
177
178    #[deprecated(since = "0.7.10", note = "Use `updated()` instead")]
179    /// Returns all items that were updated but still match the query
180    pub fn updates(&self) -> Vec<R> { self.updated() }
181}
182
183impl<I> std::fmt::Display for ChangeSet<I>
184where I: View + Clone + 'static
185{
186    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
187        // print the number of results in the resultset, and then display each change
188        use ankurah_signals::Peek;
189        let results = self.resultset.peek().len();
190        write!(f, "ChangeSet({results} results): {}", self.changes.iter().map(|c| c.to_string()).collect::<Vec<_>>().join(", "))
191    }
192}
193
194// Note: ChangeSet<Entity> conversion removed since Entity doesn't implement View
195// and ChangeSet is no longer used by Reactor
196
197impl<I> From<ItemChange<Entity>> for ItemChange<I>
198where I: View
199{
200    fn from(change: ItemChange<Entity>) -> Self {
201        match change {
202            ItemChange::Initial { item } => ItemChange::Initial { item: I::from_entity(item) },
203            ItemChange::Add { item, events } => ItemChange::Add { item: I::from_entity(item), events },
204            ItemChange::Update { item, events } => ItemChange::Update { item: I::from_entity(item), events },
205            ItemChange::Remove { item, events } => ItemChange::Remove { item: I::from_entity(item), events },
206        }
207    }
208}
209
210#[derive(Debug, Clone, PartialEq)]
211pub enum ChangeKind {
212    Initial,
213    Add,
214    Remove,
215    Update,
216}
217
218impl<R> From<&ItemChange<R>> for ChangeKind {
219    fn from(change: &ItemChange<R>) -> Self {
220        match change {
221            ItemChange::Initial { .. } => ChangeKind::Initial,
222            ItemChange::Add { .. } => ChangeKind::Add,
223            ItemChange::Remove { .. } => ChangeKind::Remove,
224            ItemChange::Update { .. } => ChangeKind::Update,
225        }
226    }
227}
228
229// Moved all the ReactorUpdate stuff into reactor.rs because it's specific to the reactor
230// and we have several types of updates, so it's essential to keep them organized
ankurah_core/changes.rs

ankurah_core/
changes.rs
