awaken_server_contract/contract/
event_store.rs

1//! Server/store-owned canonical event store traits and store-output types.
2//!
3//! The input/identity vocabulary (drafts, options, ids, cursors, scopes, kinds,
4//! visibility) stays in `awaken-runtime-contract` and is re-exported below by
5//! name. The store-output types (the persisted `CanonicalEvent`, `EventPage`,
6//! `AppendResult`, subscription handles and `FidelityClass`) and the capability
7//! traits are server/store concerns and live here.
8
9use std::collections::{BTreeMap, BTreeSet};
10
11use async_trait::async_trait;
12use futures::stream::BoxStream;
13use serde::{Deserialize, Serialize};
14use serde_json::Value;
15
16// Canonical event input & identity vocabulary shared with the runtime write
17// boundary; defined in runtime-contract.
18pub use awaken_runtime_contract::contract::event_store::{
19    AppendOptions, CanonicalEventDraft, CanonicalEventId, CanonicalEventKind, EventCursor,
20    EventScope, EventScopeFamily, EventScopeIds, EventStoreError, EventVisibility,
21};
22
23/// Durability class used by compacted and full-fidelity event capture.
24#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
25#[serde(rename_all = "snake_case")]
26pub enum FidelityClass {
27    ObservedRuntimeEvent,
28    CommittedRuntimeEvent,
29    DomainEvent,
30    ControlEvent,
31}
32
33/// EventStore append output.
34#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
35pub struct CanonicalEvent {
36    pub event_id: CanonicalEventId,
37    pub scopes: Vec<EventScope>,
38    pub cursors_by_scope: BTreeMap<EventScope, EventCursor>,
39    pub event_kind: CanonicalEventKind,
40    #[serde(default)]
41    pub payload: Value,
42    #[serde(default, skip_serializing_if = "Option::is_none")]
43    pub thread_id: Option<String>,
44    #[serde(default, skip_serializing_if = "Option::is_none")]
45    pub run_id: Option<String>,
46    #[serde(default, skip_serializing_if = "Option::is_none")]
47    pub causation_id: Option<String>,
48    #[serde(default, skip_serializing_if = "Option::is_none")]
49    pub correlation_id: Option<String>,
50    pub origin: String,
51    #[serde(default)]
52    pub visibility: EventVisibility,
53    pub schema_version: u32,
54    pub created_at: u64,
55}
56
57impl CanonicalEvent {
58    /// Build a persisted canonical event from an accepted draft and store output.
59    pub fn from_append(
60        event_id: CanonicalEventId,
61        cursors_by_scope: BTreeMap<EventScope, EventCursor>,
62        created_at: u64,
63        draft: CanonicalEventDraft,
64    ) -> Result<Self, EventStoreError> {
65        draft.validate()?;
66        validate_cursor_coverage(&draft.scopes, &cursors_by_scope)?;
67        let ids = draft.scope_ids()?;
68        Ok(Self {
69            event_id,
70            scopes: draft.scopes,
71            cursors_by_scope,
72            event_kind: draft.event_kind,
73            payload: draft.payload,
74            thread_id: ids.thread_id,
75            run_id: ids.run_id,
76            causation_id: draft.causation_id,
77            correlation_id: draft.correlation_id,
78            origin: draft.origin,
79            visibility: draft.visibility,
80            schema_version: draft.schema_version,
81            created_at,
82        })
83    }
84}
85
86/// Result returned by an append call.
87#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
88pub struct AppendResult {
89    pub event: CanonicalEvent,
90}
91
92impl AppendResult {
93    #[must_use]
94    pub fn event_id(&self) -> &CanonicalEventId {
95        &self.event.event_id
96    }
97
98    #[must_use]
99    pub fn cursors_by_scope(&self) -> &BTreeMap<EventScope, EventCursor> {
100        &self.event.cursors_by_scope
101    }
102}
103
104/// Paged canonical event list response.
105#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
106pub struct EventPage {
107    pub events: Vec<CanonicalEvent>,
108    #[serde(default, skip_serializing_if = "Option::is_none")]
109    pub next_cursor: Option<EventCursor>,
110    pub has_more: bool,
111}
112
113/// Start position for event subscription.
114#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
115#[serde(rename_all = "snake_case")]
116pub enum SubscribeStart {
117    FromStart,
118    FromCursor(EventCursor),
119    FromNow,
120}
121
122/// Live canonical event stream.
123pub type CanonicalEventStream = BoxStream<'static, Result<CanonicalEvent, EventStoreError>>;
124
125/// Result returned when a live subscription is opened.
126pub struct SubscribeHandle {
127    pub start_cursor: Option<EventCursor>,
128    pub stream: CanonicalEventStream,
129}
130
131fn validate_cursor_coverage(
132    scopes: &[EventScope],
133    cursors: &BTreeMap<EventScope, EventCursor>,
134) -> Result<(), EventStoreError> {
135    let scope_set = scopes.iter().collect::<BTreeSet<_>>();
136    let cursor_scope_set = cursors.keys().collect::<BTreeSet<_>>();
137    if scope_set != cursor_scope_set {
138        return Err(EventStoreError::Validation(
139            "cursors_by_scope must exactly match event scopes".to_string(),
140        ));
141    }
142    Ok(())
143}
144
145/// Append canonical events.
146#[async_trait]
147pub trait EventWriter: Send + Sync {
148    async fn append(
149        &self,
150        draft: CanonicalEventDraft,
151        options: AppendOptions,
152    ) -> Result<AppendResult, EventStoreError>;
153}
154
155/// Read canonical event history.
156#[async_trait]
157pub trait EventReader: Send + Sync {
158    async fn list(
159        &self,
160        scope: EventScope,
161        from: Option<EventCursor>,
162        limit: usize,
163    ) -> Result<EventPage, EventStoreError>;
164
165    async fn count(&self, scope: EventScope) -> Result<u64, EventStoreError>;
166}
167
168/// Lookup canonical events by stable event id.
169#[async_trait]
170pub trait EventLookup: Send + Sync {
171    async fn load_event(
172        &self,
173        event_id: &CanonicalEventId,
174    ) -> Result<CanonicalEvent, EventStoreError>;
175}
176
177/// Subscribe to canonical event history and live tail.
178#[async_trait]
179pub trait EventSubscriber: Send + Sync {
180    async fn subscribe(
181        &self,
182        scope: EventScope,
183        start: SubscribeStart,
184    ) -> Result<SubscribeHandle, EventStoreError>;
185}
186
187/// Full canonical event store capability.
188pub trait EventStore: EventWriter + EventReader + EventLookup + EventSubscriber {}
189
190impl<T> EventStore for T where T: EventWriter + EventReader + EventLookup + EventSubscriber {}
191
192#[cfg(test)]
193mod tests {
194    use super::*;
195    use std::collections::BTreeMap;
196
197    fn kind() -> CanonicalEventKind {
198        CanonicalEventKind::new("RunStarted").unwrap()
199    }
200
201    fn cursor(value: &str) -> EventCursor {
202        EventCursor::new(value).unwrap()
203    }
204
205    fn event_id(value: &str) -> CanonicalEventId {
206        CanonicalEventId::new(value).unwrap()
207    }
208
209    #[test]
210    fn persisted_event_requires_cursor_for_each_scope() {
211        let draft = CanonicalEventDraft::new(
212            vec![EventScope::thread("t1"), EventScope::run("r1")],
213            kind(),
214            serde_json::Value::Null,
215            "server",
216        )
217        .unwrap();
218        let mut cursors = BTreeMap::new();
219        cursors.insert(EventScope::thread("t1"), cursor("cur_t_1"));
220
221        let err = CanonicalEvent::from_append(event_id("evt_1"), cursors, 1, draft).unwrap_err();
222        assert!(
223            matches!(err, EventStoreError::Validation(message) if message.contains("cursors_by_scope"))
224        );
225    }
226
227    #[test]
228    fn persisted_event_carries_store_assigned_fields_and_denormalized_ids() {
229        let draft = CanonicalEventDraft::new(
230            vec![EventScope::thread("t1"), EventScope::run("r1")],
231            kind(),
232            serde_json::json!({"ok": true}),
233            "server",
234        )
235        .unwrap();
236        let mut cursors = BTreeMap::new();
237        cursors.insert(EventScope::thread("t1"), cursor("cur_t_1"));
238        cursors.insert(EventScope::run("r1"), cursor("cur_r_1"));
239
240        let event = CanonicalEvent::from_append(event_id("evt_1"), cursors, 42, draft).unwrap();
241        assert_eq!(event.event_id.as_str(), "evt_1");
242        assert_eq!(event.thread_id.as_deref(), Some("t1"));
243        assert_eq!(event.run_id.as_deref(), Some("r1"));
244        assert_eq!(event.created_at, 42);
245        assert_eq!(event.payload, serde_json::json!({"ok": true}));
246    }
247}
awaken_server_contract/contract/event_store.rs

awaken_server_contract/contract/
event_store.rs
