es_entity/snapshot.rs
1//! Core types for entity snapshotting.
2//!
3//! A repo with `#[es_repo(snapshot)]` persists, in the same statement as the
4//! events it appends, an author-defined state `S` (the fold of events
5//! `1..=k`) into `<tbl>_snapshots`. Hydration then loads `S` plus only the
6//! events after `k` in one round trip.
7
8use chrono::{DateTime, Utc};
9
10/// Matches no stored snapshot; binds on full-history loads and is
11/// `NoSnapshot`'s fingerprint.
12pub const NO_SNAPSHOT_FINGERPRINT: i64 = i64::MIN;
13
14/// Implemented by `#[derive(EsSnapshot)]` for user state types and by hand for
15/// [`NoSnapshot`].
16pub trait EsSnapshot:
17 serde::Serialize + serde::de::DeserializeOwned + std::fmt::Debug + Send + Sync + 'static
18{
19 /// `false` only for [`NoSnapshot`].
20 const IS_SNAPSHOT: bool;
21 const FINGERPRINT: i64;
22 #[doc(hidden)]
23 const HAS_FORGETTABLE_FIELDS: bool;
24 /// JSON keys (field idents) of the top-level `Forgettable<T>` fields.
25 #[doc(hidden)]
26 const FORGETTABLE_JSON_FIELDS: &'static [&'static str];
27 #[doc(hidden)]
28 fn extract_forgettable_payloads(&self) -> Option<serde_json::Value>;
29 #[doc(hidden)]
30 fn forget_forgettable_payloads(&mut self);
31}
32
33/// The default `S` of `EntityEvents<E, S>`: this entity has no snapshot.
34#[derive(Debug, Clone, Copy, PartialEq, Eq)]
35pub enum NoSnapshot {}
36
37impl serde::Serialize for NoSnapshot {
38 fn serialize<Ser: serde::Serializer>(&self, _serializer: Ser) -> Result<Ser::Ok, Ser::Error> {
39 match *self {}
40 }
41}
42
43impl<'de> serde::Deserialize<'de> for NoSnapshot {
44 fn deserialize<D: serde::Deserializer<'de>>(_deserializer: D) -> Result<Self, D::Error> {
45 Err(serde::de::Error::custom(
46 "NoSnapshot cannot be deserialized",
47 ))
48 }
49}
50
51impl EsSnapshot for NoSnapshot {
52 const IS_SNAPSHOT: bool = false;
53 const FINGERPRINT: i64 = NO_SNAPSHOT_FINGERPRINT;
54 const HAS_FORGETTABLE_FIELDS: bool = false;
55 const FORGETTABLE_JSON_FIELDS: &'static [&'static str] = &[];
56
57 fn extract_forgettable_payloads(&self) -> Option<serde_json::Value> {
58 match *self {}
59 }
60
61 fn forget_forgettable_payloads(&mut self) {
62 match *self {}
63 }
64}
65
66/// A persisted snapshot as loaded: the state plus the metadata of the events
67/// it summarises.
68pub struct SnapshotRecord<S> {
69 /// Events `1..=sequence` are folded into `state`.
70 pub sequence: usize,
71 pub state: S,
72 pub recorded_at: DateTime<Utc>,
73 /// `recorded_at` of event 1 — keeps `entity_first_persisted_at()` O(1).
74 pub first_recorded_at: DateTime<Utc>,
75}
76
77impl<S: std::fmt::Debug> std::fmt::Debug for SnapshotRecord<S> {
78 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
79 f.debug_struct("SnapshotRecord")
80 .field("sequence", &self.sequence)
81 .field("state", &self.state)
82 .field("recorded_at", &self.recorded_at)
83 .field("first_recorded_at", &self.first_recorded_at)
84 .finish()
85 }
86}
87
88impl<S: Clone> Clone for SnapshotRecord<S> {
89 fn clone(&self) -> Self {
90 Self {
91 sequence: self.sequence,
92 state: self.state.clone(),
93 recorded_at: self.recorded_at,
94 first_recorded_at: self.first_recorded_at,
95 }
96 }
97}
98
99impl<S: PartialEq> PartialEq for SnapshotRecord<S> {
100 fn eq(&self, other: &Self) -> bool {
101 self.sequence == other.sequence
102 && self.state == other.state
103 && self.recorded_at == other.recorded_at
104 && self.first_recorded_at == other.first_recorded_at
105 }
106}
107
108/// One item of an entity's replay. The snapshot, when present, is the FIRST
109/// item in forward order and the LAST item in reverse order. Deliberately
110/// exhaustive — do not add `#[non_exhaustive]`.
111#[derive(Debug, Clone, Copy)]
112pub enum Replay<'a, E, S> {
113 Snapshot(&'a S),
114 Event(&'a E),
115}
116
117/// Normalises `idempotency_guard!` input: a plain `&E` stream (today's call
118/// sites) and a `replay()` stream both become [`Replay`]. Do not implement
119/// for other types.
120pub trait IntoReplay<'a, E, S> {
121 fn into_replay(self) -> Replay<'a, E, S>;
122}
123
124impl<'a, E> IntoReplay<'a, E, NoSnapshot> for &'a E {
125 fn into_replay(self) -> Replay<'a, E, NoSnapshot> {
126 Replay::Event(self)
127 }
128}
129
130impl<'a, E, S> IntoReplay<'a, E, S> for Replay<'a, E, S> {
131 fn into_replay(self) -> Self {
132 self
133 }
134}
135
136/// Reached by `idempotency_guard!` when a stream yields `Replay::Snapshot`
137/// but the guard has no `snapshot:` clause. Only [`NoSnapshot`] (and
138/// `SnapshotRecord<NoSnapshot>`) implement it, so on a real snapshot this is
139/// a compile error with the message below.
140#[diagnostic::on_unimplemented(
141 message = "`idempotency_guard!` over a snapshotted event stream needs a `snapshot: <pattern> [if <guard>]` clause",
142 label = "this guard iterates `replay()` of an entity whose snapshot type is `{Self}`",
143 note = "add `snapshot: s if <what the snapshot says about this operation>`; use `snapshot: _ if false` if the snapshot can never imply the operation was applied"
144)]
145pub trait GuardWithoutSnapshotClause {
146 fn reached<R>(&self) -> R;
147}
148
149impl GuardWithoutSnapshotClause for NoSnapshot {
150 fn reached<R>(&self) -> R {
151 match *self {}
152 }
153}
154
155impl GuardWithoutSnapshotClause for SnapshotRecord<NoSnapshot> {
156 fn reached<R>(&self) -> R {
157 match self.state {}
158 }
159}
160
161/// Implemented by every entity whose repo enables `snapshot`.
162///
163/// The `#[derive(EsRepo)]` macro checks that the entity's own `Snapshot`
164/// associated type and the repo's `snapshot` flag agree, so widening the
165/// repo without also widening the entity's `EntityEvents<E, S>` (or the
166/// reverse) does not compile:
167///
168/// ```ignore
169/// impl HeadSnapshot for Meter {
170/// fn capture(&self) -> Option<MeterSnapshot> {
171/// (self.events.tail_len() >= 4).then(|| MeterSnapshot { .. })
172/// }
173/// }
174/// ```
175///
176/// ```compile_fail
177/// use es_entity::*;
178/// use serde::{Serialize, Deserialize};
179/// # fn main() {}
180/// # es_entity::entity_id! { SnapGuardMeterId }
181/// # #[derive(EsEvent, Debug, Clone, Serialize, Deserialize)]
182/// # #[serde(tag = "type", rename_all = "snake_case")]
183/// # #[es_event(id = "SnapGuardMeterId")]
184/// # pub enum SnapGuardMeterEvent {
185/// # Initialized { id: SnapGuardMeterId },
186/// # }
187/// # pub struct NewSnapGuardMeter { id: SnapGuardMeterId }
188/// # impl IntoEvents<SnapGuardMeterEvent> for NewSnapGuardMeter {
189/// # fn into_events(self) -> EntityEvents<SnapGuardMeterEvent> {
190/// # EntityEvents::init(self.id, [SnapGuardMeterEvent::Initialized { id: self.id }])
191/// # }
192/// # }
193/// // Missing: a real `Snapshot` — `events` stays
194/// // `EntityEvents<SnapGuardMeterEvent>` (implicit `NoSnapshot`).
195/// #[derive(EsEntity)]
196/// pub struct SnapGuardMeter {
197/// pub id: SnapGuardMeterId,
198/// events: EntityEvents<SnapGuardMeterEvent>,
199/// }
200/// # impl TryFromEvents<SnapGuardMeterEvent> for SnapGuardMeter {
201/// # fn try_from_events(events: EntityEvents<SnapGuardMeterEvent>) -> Result<Self, EntityHydrationError> {
202/// # Ok(SnapGuardMeter { id: *events.id(), events })
203/// # }
204/// # }
205/// # impl HeadSnapshot for SnapGuardMeter {
206/// # fn capture(&self) -> Option<NoSnapshot> { None }
207/// # }
208/// // error: entity snapshot type and `#[es_repo(snapshot)]` disagree.
209/// #[derive(EsRepo, Debug)]
210/// #[es_repo(
211/// entity = "SnapGuardMeter",
212/// tbl = "meters",
213/// events_tbl = "meter_events",
214/// snapshot,
215/// snapshot_tbl = "meter_snapshots"
216/// )]
217/// pub struct SnapGuardMeters {
218/// pool: es_entity::db::Pool,
219/// }
220/// ```
221///
222/// A snapshot type with a `Forgettable<T>` field also requires the repo to
223/// enable `forgettable` — otherwise the payload would never be scrubbed:
224///
225/// ```compile_fail
226/// use es_entity::*;
227/// use serde::{Serialize, Deserialize};
228/// # fn main() {}
229/// # es_entity::entity_id! { SnapGuardClientId }
230/// # #[derive(EsEvent, Debug, Clone, Serialize, Deserialize)]
231/// # #[serde(tag = "type", rename_all = "snake_case")]
232/// # #[es_event(id = "SnapGuardClientId")]
233/// # pub enum SnapGuardClientEvent {
234/// # Initialized { id: SnapGuardClientId, email: Forgettable<String> },
235/// # }
236/// #[derive(EsSnapshot, Debug, Clone, Serialize, Deserialize)]
237/// #[es_snapshot(version = 1)]
238/// pub struct SnapGuardClientSnapshot {
239/// pub id: SnapGuardClientId,
240/// pub email: Forgettable<String>,
241/// }
242/// # pub struct NewSnapGuardClient { id: SnapGuardClientId, email: String }
243/// # impl IntoEvents<SnapGuardClientEvent> for NewSnapGuardClient {
244/// # fn into_events(self) -> EntityEvents<SnapGuardClientEvent> {
245/// # EntityEvents::init(
246/// # self.id,
247/// # [SnapGuardClientEvent::Initialized { id: self.id, email: Forgettable::new(self.email) }],
248/// # )
249/// # }
250/// # }
251/// # #[derive(EsEntity)]
252/// # pub struct SnapGuardClient {
253/// # pub id: SnapGuardClientId,
254/// # events: EntityEvents<SnapGuardClientEvent, SnapGuardClientSnapshot>,
255/// # }
256/// # impl HeadSnapshot for SnapGuardClient {
257/// # fn capture(&self) -> Option<SnapGuardClientSnapshot> { None }
258/// # }
259/// # impl TryFromEvents<SnapGuardClientEvent, SnapGuardClientSnapshot> for SnapGuardClient {
260/// # fn try_from_events(events: EntityEvents<SnapGuardClientEvent, SnapGuardClientSnapshot>) -> Result<Self, EntityHydrationError> {
261/// # Ok(SnapGuardClient { id: *events.id(), events })
262/// # }
263/// # }
264/// // error: snapshot type has Forgettable fields but this repo does not
265/// // enable `forgettable`.
266/// #[derive(EsRepo, Debug)]
267/// #[es_repo(
268/// entity = "SnapGuardClient",
269/// tbl = "clients",
270/// events_tbl = "client_events",
271/// snapshot,
272/// snapshot_tbl = "client_snapshots"
273/// )]
274/// pub struct SnapGuardClients {
275/// pool: es_entity::db::Pool,
276/// }
277/// ```
278pub trait HeadSnapshot: crate::EsEntity {
279 /// Called on every write the entity passes through — with staged events,
280 /// or with none when the loaded state had no matching snapshot. `Some(s)`
281 /// = persist `s` as the snapshot at the current head; `None` = keep
282 /// whatever snapshot exists and let the tail grow.
283 fn capture(&self) -> Option<Self::Snapshot>;
284}