khive_storage/
event.rs

1//! Event storage capability — append-only operation log.
2
3use std::collections::BTreeMap;
4
5use async_trait::async_trait;
6use serde::{Deserialize, Serialize};
7use serde_json::Value;
8use uuid::Uuid;
9
10use khive_types::{EventKind, EventOutcome, RefResolution, SubstrateKind};
11
12use crate::capability::StorageCapability;
13use crate::error::StorageError;
14use crate::types::{BatchWriteSummary, Page, PageRequest, SqlValue, StorageResult};
15
16/// Storage-level event record. Every verb execution produces one.
17/// Immutable once appended; projection rows are written beside it at append time.
18#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
19pub struct Event {
20    pub id: Uuid,
21    pub namespace: String,
22    pub verb: String,
23    pub substrate: SubstrateKind,
24    pub actor: String,
25    pub kind: EventKind,
26    pub outcome: EventOutcome,
27    pub payload: Value,
28    pub payload_schema_version: u32,
29    pub profile_state_version: Option<u64>,
30    pub duration_us: i64,
31    pub target_id: Option<Uuid>,
32    pub session_id: Option<Uuid>,
33    pub aggregate_kind: Option<String>,
34    pub aggregate_id: Option<Uuid>,
35    pub created_at: i64,
36    /// Original request parser position; absent for legacy/non-request events.
37    #[serde(default)]
38    pub op_index: Option<u32>,
39    /// Reference provenance, present exactly when `op_index` is present.
40    #[serde(default)]
41    pub ref_resolution: Option<RefResolution>,
42}
43
44impl Event {
45    /// Create a new event with a generated UUID and current timestamp.
46    pub fn new(
47        namespace: impl Into<String>,
48        verb: impl Into<String>,
49        kind: EventKind,
50        substrate: SubstrateKind,
51        actor: impl Into<String>,
52    ) -> Self {
53        let operation = crate::operation_context::current_operation_attribution();
54        Self {
55            id: Uuid::new_v4(),
56            namespace: namespace.into(),
57            verb: verb.into(),
58            substrate,
59            actor: actor.into(),
60            kind,
61            outcome: EventOutcome::Success,
62            payload: Value::Object(Default::default()),
63            payload_schema_version: 1,
64            profile_state_version: None,
65            duration_us: 0,
66            target_id: None,
67            session_id: None,
68            aggregate_kind: None,
69            aggregate_id: None,
70            created_at: chrono::Utc::now().timestamp_micros(),
71            op_index: operation.map(|operation| operation.op_index),
72            ref_resolution: operation.map(|operation| operation.ref_resolution),
73        }
74    }
75
76    /// Set the event outcome (success/failure).
77    pub fn with_outcome(mut self, o: EventOutcome) -> Self {
78        self.outcome = o;
79        self
80    }
81
82    /// Set the event payload JSON.
83    pub fn with_payload(mut self, payload: Value) -> Self {
84        self.payload = payload;
85        self
86    }
87
88    /// Set the payload schema version for forward compatibility.
89    pub fn with_payload_schema_version(mut self, version: u32) -> Self {
90        self.payload_schema_version = version;
91        self
92    }
93
94    /// Set the brain profile state version at event time.
95    pub fn with_profile_state_version(mut self, version: u64) -> Self {
96        self.profile_state_version = Some(version);
97        self
98    }
99
100    /// Set the operation duration in microseconds.
101    pub fn with_duration_us(mut self, us: i64) -> Self {
102        self.duration_us = us;
103        self
104    }
105
106    /// Set the exact subject UUID for this event; it need not be a graph record.
107    pub fn with_target(mut self, id: Uuid) -> Self {
108        self.target_id = Some(id);
109        self
110    }
111
112    /// Set the session ID for correlating related events.
113    pub fn with_session_id(mut self, id: Uuid) -> Self {
114        self.session_id = Some(id);
115        self
116    }
117
118    /// Set the aggregate kind and ID for event-sourced projections.
119    pub fn with_aggregate(mut self, kind: impl Into<String>, id: Uuid) -> Self {
120        self.aggregate_kind = Some(kind.into());
121        self.aggregate_id = Some(id);
122        self
123    }
124}
125
126/// Which durable record family an event observation refers to. Edge remains
127/// relational storage rather than a fourth [`SubstrateKind`]; this narrower
128/// discriminant exists so edge lifecycle events can still be queried through
129/// `observed=[edge_id]`.
130#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
131#[serde(rename_all = "snake_case")]
132pub enum ReferentKind {
133    Entity,
134    Note,
135    Edge,
136}
137
138impl ReferentKind {
139    /// Return the lowercase string name for this referent kind.
140    pub const fn name(self) -> &'static str {
141        match self {
142            Self::Entity => "entity",
143            Self::Note => "note",
144            Self::Edge => "edge",
145        }
146    }
147}
148
149/// Role of a referent in a brain observation (candidate, selected, target, signal).
150#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
151#[serde(rename_all = "snake_case")]
152pub enum ObservationRole {
153    Candidate,
154    Selected,
155    Target,
156    Signal,
157}
158
159impl ObservationRole {
160    /// Return the lowercase string name for this observation role.
161    pub const fn name(self) -> &'static str {
162        match self {
163            Self::Candidate => "candidate",
164            Self::Selected => "selected",
165            Self::Target => "target",
166            Self::Signal => "signal",
167        }
168    }
169}
170
171/// A single entity observation recorded alongside an event.
172#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
173pub struct EventObservation {
174    pub event_id: Uuid,
175    pub entity_id: Uuid,
176    pub referent_kind: ReferentKind,
177    pub role: ObservationRole,
178    pub position: u32,
179}
180
181/// An event together with its associated observations.
182#[derive(Debug, Clone, Serialize, Deserialize)]
183pub struct EventView {
184    pub event: Event,
185    pub observations: Vec<EventObservation>,
186}
187
188/// Closed grouping keys for storage-level event counts.
189#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
190#[serde(rename_all = "snake_case")]
191pub enum EventGroupBy {
192    /// Exact stored verb name.
193    Verb,
194    /// Canonical stored event kind.
195    Kind,
196    /// Exact stored actor attribution.
197    Actor,
198}
199
200impl EventGroupBy {
201    /// Return the fixed stored column name; never caller-supplied SQL.
202    pub const fn column(self) -> &'static str {
203        match self {
204            Self::Verb => "verb",
205            Self::Kind => "kind",
206            Self::Actor => "actor",
207        }
208    }
209}
210
211/// Filter for querying events. Namespace is implicit in the scoped EventStore.
212///
213/// New fields deserialize to defaults for older serialized filters. Complete
214/// external Rust struct literals must specify them or use `..Default::default()`.
215#[derive(Clone, Debug, Default, Serialize, Deserialize)]
216pub struct EventFilter {
217    pub ids: Vec<Uuid>,
218    /// Exact event subject, independent of the graph observation projection.
219    #[serde(default)]
220    pub target_id: Option<Uuid>,
221    pub kinds: Vec<EventKind>,
222    pub verbs: Vec<String>,
223    pub substrates: Vec<SubstrateKind>,
224    pub actors: Vec<String>,
225    pub after: Option<i64>,
226    pub before: Option<i64>,
227    pub session_id: Option<Uuid>,
228    pub observed: Vec<Uuid>,
229    pub selected: Vec<Uuid>,
230    pub payload_proposal_id: Option<Uuid>,
231    /// Exact outcome; no restriction when absent.
232    #[serde(default)]
233    pub outcome: Option<EventOutcome>,
234    /// ANDed SQL JSON equalities; see [`EventFilter::payload_eq`].
235    #[serde(default)]
236    pub payload_equalities: Vec<(String, SqlValue)>,
237}
238
239impl EventFilter {
240    /// Match one outcome, replacing any previous outcome selection.
241    pub fn outcome(mut self, outcome: EventOutcome) -> Self {
242        self.outcome = Some(outcome);
243        self
244    }
245
246    /// Require SQL `json_extract(payload, path) = value` equality.
247    ///
248    /// Paths must be `$.field[.subfield]` with nonempty ASCII alphanumeric or
249    /// underscore segments; queries reject other paths. Calls are ANDed.
250    /// SQL NULL matches neither a missing field nor explicit JSON null.
251    /// JSON booleans compare as 0/1, including numeric coercion. JSON objects
252    /// and arrays compare their serialized SQL text, not structural JSON equality.
253    pub fn payload_eq(mut self, path: impl Into<String>, value: SqlValue) -> Self {
254        self.payload_equalities.push((path.into(), value));
255        self
256    }
257}
258
259/// Persisted ordering key; the physical UUID spelling is not normalized.
260#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
261pub struct EventOrderKey {
262    pub created_at_us: i64,
263    pub physical_id: String,
264}
265
266/// Count-free ascending window within one scoped event store.
267#[derive(Clone, Debug, Serialize, Deserialize)]
268pub struct EventPageQuery {
269    pub since_us: i64,
270    pub until_us: i64,
271    pub kinds: Vec<EventKind>,
272    pub actors: Vec<String>,
273    pub exclude_namespaces: Vec<String>,
274    pub after: Option<EventOrderKey>,
275    pub max_rows: u32,
276}
277
278/// A decoded event with its exact persisted seek key.
279#[derive(Clone, Debug, Serialize, Deserialize)]
280pub struct EventPageRow {
281    pub event: Event,
282    pub order_key: EventOrderKey,
283}
284
285/// A bounded prefix, without a count of the complete matching window.
286///
287/// `budget_stop` names the row the leaf did not return because reading it would
288/// pass the leaf's raw-text budget; `rows` then holds every row ordered before it,
289/// whole, and is shorter than the requested row bound.
290#[derive(Clone, Debug, Serialize, Deserialize)]
291pub struct EventPageWindow {
292    pub rows: Vec<EventPageRow>,
293    #[serde(default)]
294    pub budget_stop: Option<EventOrderKey>,
295}
296
297/// Per-row outcome of an [`EventStore::append_events_idempotent`] call, in
298/// input order. Distinguishes a fresh insert from a retry that reproduced an
299/// identical row from a retry whose identity now disagrees with what is
300/// already stored.
301#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
302#[serde(rename_all = "snake_case")]
303pub enum EventAppendDisposition {
304    /// No prior row with this id existed; it was inserted.
305    Inserted,
306    /// A prior row with this id existed and every persisted column plus the
307    /// ordered observation projection matched exactly. Not re-inserted.
308    AlreadyPresentIdentical,
309    /// A prior row with this id existed but disagreed with the submitted
310    /// row. Not inserted; unrelated rows in the same batch are unaffected.
311    IdentityConflict,
312}
313
314/// Result of [`EventStore::append_events_idempotent`]. `rows` preserves the
315/// input order and length of the submitted batch.
316#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
317pub struct IdempotentEventBatchResult {
318    pub rows: Vec<EventAppendDisposition>,
319}
320
321/// Append-only operation log for verb executions.
322#[async_trait]
323pub trait EventStore: Send + Sync + 'static {
324    /// Append a single event to the log.
325    async fn append_event(&self, event: Event) -> StorageResult<()>;
326    /// Append a batch of events to the log.
327    async fn append_events(&self, events: Vec<Event>) -> StorageResult<BatchWriteSummary>;
328    /// Fetch an event by UUID, returning `None` if absent.
329    async fn get_event(&self, id: Uuid) -> StorageResult<Option<Event>>;
330    /// Query events matching a filter with pagination.
331    async fn query_events(
332        &self,
333        filter: EventFilter,
334        page: PageRequest,
335    ) -> StorageResult<Page<Event>>;
336    /// Read at most 4096 rows in ascending physical time/ID order, without counting.
337    /// Filters and exclusions apply before the row bound. A backend must refuse
338    /// unsupported paging rather than substitute an offset query.
339    async fn query_event_page(&self, query: EventPageQuery) -> StorageResult<EventPageWindow> {
340        let _ = query;
341        Err(StorageError::Unsupported {
342            capability: StorageCapability::Events,
343            operation: "query_event_page".into(),
344            message: "this EventStore backend does not implement query_event_page".into(),
345        })
346    }
347
348    /// Count events matching a filter.
349    async fn count_events(&self, filter: EventFilter) -> StorageResult<u64>;
350
351    /// Count matching rows per stored verb, kind, or actor in this store's namespace.
352    /// Empty groups are omitted. Backends must aggregate in storage rather than
353    /// substitute a bounded row query that could silently undercount.
354    async fn count_events_grouped(
355        &self,
356        filter: EventFilter,
357        group_by: EventGroupBy,
358    ) -> StorageResult<BTreeMap<String, u64>> {
359        let _ = (filter, group_by);
360        Err(StorageError::Unsupported {
361            capability: StorageCapability::Events,
362            operation: "count_events_grouped".into(),
363            message: "this EventStore backend does not implement grouped counts".into(),
364        })
365    }
366
367    /// Validate `event` against the exact insert/observation shape the
368    /// backend would build at append time, performing no I/O. Rejects a
369    /// malformed row before it is ever enqueued for a write, so one bad
370    /// producer input cannot poison a batch shared with other callers.
371    ///
372    /// Backends that do not implement pre-enqueue validation return
373    /// [`StorageError::Unsupported`].
374    fn preflight_event(&self, event: &Event) -> StorageResult<()> {
375        let _ = event;
376        Err(StorageError::Unsupported {
377            capability: StorageCapability::Events,
378            operation: "preflight_event".into(),
379            message: "this EventStore backend does not implement preflight_event".into(),
380        })
381    }
382
383    /// Append a batch of events with idempotent retry semantics: a row
384    /// carrying an id that already exists is compared against every
385    /// persisted column and its ordered observation projection rather than
386    /// treated as a write conflict. Exact equality reports
387    /// [`EventAppendDisposition::AlreadyPresentIdentical`] instead of
388    /// re-inserting; any mismatch reports
389    /// [`EventAppendDisposition::IdentityConflict`] for that row alone,
390    /// while unrelated rows in the same batch still commit.
391    ///
392    /// Backends that do not implement idempotent batching return
393    /// [`StorageError::Unsupported`].
394    async fn append_events_idempotent(
395        &self,
396        events: Vec<Event>,
397    ) -> StorageResult<IdempotentEventBatchResult> {
398        let _ = events;
399        Err(StorageError::Unsupported {
400            capability: StorageCapability::Events,
401            operation: "append_events_idempotent".into(),
402            message: "this EventStore backend does not implement append_events_idempotent".into(),
403        })
404    }
405
406    /// Whether this backend implements `preflight_event` and
407    /// `append_events_idempotent` for real, rather than inheriting their
408    /// `Unsupported`-returning defaults above.
409    ///
410    /// A caller that builds an ADR-133 audit-batch seam over a backend that
411    /// answers `false` here would have every audited row rejected at
412    /// preflight while the dispatch it audits still reports success — the
413    /// exact silent-loss failure mode the batch exists to prevent. Defaults
414    /// to `false` so an unmodified legacy backend is caught at registry
415    /// build time instead of appearing healthy.
416    fn supports_idempotent_audit_batch(&self) -> bool {
417        false
418    }
419}
420
421#[cfg(test)]
422#[path = "event/tests.rs"]
423mod tests;
khive_storage/event.rs

khive_storage/
event.rs
