aedb/api/
event_query.rs

1//! Ergonomic, first-class querying over the canonical event log.
2//!
3//! Events are appended to the `event_outbox` system table (primary key
4//! `(commit_seq, topic, event_key)`) by [`crate::AedbInstance::emit_event`] and
5//! by `commit_effect_batch`. This module turns that raw, append-only log into a
6//! query surface suitable for building UIs and monitors: filter by event type,
7//! project/scope, commit-sequence range, time range, and arbitrary payload
8//! fields, in either direction, with commit-atomic cursor pagination.
9//!
10//! The query reads through a snapshot ([`ConsistencyMode`]) and is bounded by
11//! `max_scan_rows` so a selective filter can never trigger an unbounded scan;
12//! when the budget is hit mid-page the returned cursor lets the caller resume.
13
14use std::ops::Bound;
15
16use crate::catalog::SYSTEM_PROJECT_ID;
17use crate::catalog::types::Value;
18use crate::error::AedbError;
19use crate::permission::{CallerContext, Permission};
20use crate::query::plan::ConsistencyMode;
21use crate::query_authorization::ensure_external_caller_allowed;
22use crate::storage::encoded_key::EncodedKey;
23use crate::{
24    AedbInstance, EVENT_OUTBOX_TABLE, EventFieldFilter, EventOrder, EventOutboxRecord, EventQuery,
25    EventStreamPage, SYSTEM_SCOPE_ID,
26};
27
28impl EventQuery {
29    /// Start an empty query (ascending, no filters). Set `limit` before use;
30    /// a `limit` of 0 yields an empty page.
31    pub fn new() -> Self {
32        Self::default()
33    }
34
35    /// Filter by event type (the emit `topic`).
36    pub fn topic(mut self, topic: impl Into<String>) -> Self {
37        self.topic = Some(topic.into());
38        self
39    }
40
41    /// Restrict to a single project / scope namespace.
42    pub fn namespace(mut self, project_id: impl Into<String>, scope_id: impl Into<String>) -> Self {
43        self.project_id = Some(project_id.into());
44        self.scope_id = Some(scope_id.into());
45        self
46    }
47
48    /// Resume an ascending scan after this commit sequence (exclusive).
49    pub fn after_commit_seq(mut self, seq: u64) -> Self {
50        self.from_commit_seq_exclusive = Some(seq);
51        self
52    }
53
54    /// Resume a descending scan before this commit sequence (exclusive).
55    pub fn before_commit_seq(mut self, seq: u64) -> Self {
56        self.to_commit_seq_exclusive = Some(seq);
57        self
58    }
59
60    /// Inclusive timestamp window (micros since epoch).
61    pub fn time_range(mut self, from_micros: Option<u64>, to_micros: Option<u64>) -> Self {
62        self.from_ts_micros = from_micros;
63        self.to_ts_micros = to_micros;
64        self
65    }
66
67    /// Add a payload-field equality filter (e.g. recipient, instance_id, block).
68    pub fn where_field(mut self, field: impl Into<String>, equals: impl Into<String>) -> Self {
69        self.field_filters.push(EventFieldFilter {
70            field: field.into(),
71            equals: equals.into(),
72        });
73        self
74    }
75
76    /// Newest-first ordering (for "latest N by type").
77    pub fn descending(mut self) -> Self {
78        self.order = EventOrder::Descending;
79        self
80    }
81
82    /// Page size.
83    pub fn limit(mut self, limit: usize) -> Self {
84        self.limit = limit;
85        self
86    }
87}
88
89impl AedbInstance {
90    /// Query the canonical event log. See [`EventQuery`].
91    ///
92    /// In secure mode (authenticated calls required) this rejects anonymous
93    /// access; use [`AedbInstance::query_events_as`] with a caller instead.
94    pub async fn query_events(
95        &self,
96        query: &EventQuery,
97        consistency: ConsistencyMode,
98    ) -> Result<EventStreamPage, AedbError> {
99        if self.require_authenticated_calls {
100            return Err(AedbError::PermissionDenied(
101                "authenticated caller required in secure mode".into(),
102            ));
103        }
104        self.query_events_internal(query, consistency, None).await
105    }
106
107    /// Query the canonical event log on behalf of `caller`, enforcing
108    /// `TableRead` permission on the event log.
109    pub async fn query_events_as(
110        &self,
111        caller: &CallerContext,
112        query: &EventQuery,
113        consistency: ConsistencyMode,
114    ) -> Result<EventStreamPage, AedbError> {
115        ensure_external_caller_allowed(caller)?;
116        self.query_events_internal(query, consistency, Some(caller))
117            .await
118    }
119
120    /// Convenience: the most recent `n` events of a given type, newest first.
121    pub async fn latest_events_by_topic(
122        &self,
123        topic: &str,
124        n: usize,
125        consistency: ConsistencyMode,
126    ) -> Result<Vec<EventOutboxRecord>, AedbError> {
127        let query = EventQuery::new().topic(topic).descending().limit(n);
128        Ok(self.query_events(&query, consistency).await?.events)
129    }
130
131    async fn query_events_internal(
132        &self,
133        query: &EventQuery,
134        consistency: ConsistencyMode,
135        caller: Option<&CallerContext>,
136    ) -> Result<EventStreamPage, AedbError> {
137        let lease = self.acquire_snapshot(consistency).await?;
138        let snapshot_seq = lease.view.seq;
139
140        if query.limit == 0 {
141            return Ok(EventStreamPage {
142                events: Vec::new(),
143                next_commit_seq: None,
144                snapshot_seq,
145            });
146        }
147
148        if let Some(caller) = caller {
149            let required = Permission::TableRead {
150                project_id: SYSTEM_PROJECT_ID.to_string(),
151                scope_id: SYSTEM_SCOPE_ID.to_string(),
152                table_name: EVENT_OUTBOX_TABLE.to_string(),
153            };
154            if !lease
155                .view
156                .catalog
157                .has_permission(&caller.caller_id, &required)
158            {
159                return Err(AedbError::PermissionDenied("permission denied".into()));
160            }
161        }
162
163        let Some(table) =
164            lease
165                .view
166                .keyspace
167                .table(SYSTEM_PROJECT_ID, SYSTEM_SCOPE_ID, EVENT_OUTBOX_TABLE)
168        else {
169            return Ok(EventStreamPage {
170                events: Vec::new(),
171                next_commit_seq: None,
172                snapshot_seq,
173            });
174        };
175
176        // Translate the commit-sequence cursor bounds into key bounds. The PK
177        // is (commit_seq, topic, event_key); a key with an empty topic sorts
178        // before every real row at the same commit_seq, so excluding it cleanly
179        // drops an entire commit_seq from the range.
180        let start_bound = match query.from_commit_seq_exclusive {
181            Some(from) => match from.checked_add(1) {
182                Some(start_seq) => match i64::try_from(start_seq) {
183                    Ok(start_seq_i64) => Bound::Included(seq_key(start_seq_i64)),
184                    // start beyond i64::MAX: nothing can match.
185                    Err(_) => {
186                        return Ok(EventStreamPage {
187                            events: Vec::new(),
188                            next_commit_seq: None,
189                            snapshot_seq,
190                        });
191                    }
192                },
193                None => {
194                    return Ok(EventStreamPage {
195                        events: Vec::new(),
196                        next_commit_seq: None,
197                        snapshot_seq,
198                    });
199                }
200            },
201            None => Bound::Unbounded,
202        };
203        let end_bound = match query.to_commit_seq_exclusive {
204            Some(to) => match i64::try_from(to) {
205                Ok(to_i64) => Bound::Excluded(seq_key(to_i64)),
206                // upper bound beyond i64::MAX: no effective cap.
207                Err(_) => Bound::Unbounded,
208            },
209            None => Bound::Unbounded,
210        };
211
212        let max_scan_rows = self._config.max_scan_rows;
213        let limit = query.limit;
214
215        let mut events: Vec<EventOutboxRecord> = Vec::new();
216        let mut examined: usize = 0;
217        // Once set, we are finishing the events of the boundary commit and will
218        // break as soon as we cross to a different commit_seq. This keeps a page
219        // commit-atomic so cursor resumption never skips or duplicates events.
220        let mut boundary_seq: Option<u64> = None;
221        let mut stopped_early = false;
222
223        let range = table.rows.range((start_bound, end_bound));
224        match query.order {
225            EventOrder::Ascending => {
226                for (_, stored) in range {
227                    let row = lease.view.keyspace.materialize_row(stored)?;
228                    if !Self::consume_event_row(
229                        &row,
230                        query,
231                        limit,
232                        max_scan_rows,
233                        &mut events,
234                        &mut examined,
235                        &mut boundary_seq,
236                        &mut stopped_early,
237                    ) {
238                        break;
239                    }
240                }
241            }
242            EventOrder::Descending => {
243                for (_, stored) in range.rev() {
244                    let row = lease.view.keyspace.materialize_row(stored)?;
245                    if !Self::consume_event_row(
246                        &row,
247                        query,
248                        limit,
249                        max_scan_rows,
250                        &mut events,
251                        &mut examined,
252                        &mut boundary_seq,
253                        &mut stopped_early,
254                    ) {
255                        break;
256                    }
257                }
258            }
259        }
260
261        // The cursor is the boundary commit_seq only when we stopped early
262        // (there may be more); a naturally exhausted range reports no cursor.
263        let next_commit_seq = if stopped_early { boundary_seq } else { None };
264
265        Ok(EventStreamPage {
266            events,
267            next_commit_seq,
268            snapshot_seq,
269        })
270    }
271
272    /// Process one event-log row. Returns `false` when the scan should stop.
273    /// The boundary logic is direction-agnostic: iteration order is already
274    /// fixed by the caller, and a page boundary always ends at the trailing
275    /// edge of the current commit_seq.
276    #[allow(clippy::too_many_arguments)]
277    fn consume_event_row(
278        row: &crate::catalog::types::Row,
279        query: &EventQuery,
280        limit: usize,
281        max_scan_rows: usize,
282        events: &mut Vec<EventOutboxRecord>,
283        examined: &mut usize,
284        boundary_seq: &mut Option<u64>,
285        stopped_early: &mut bool,
286    ) -> bool {
287        let Some(record) = decode_event_row(row) else {
288            return true;
289        };
290
291        // If we are finishing a boundary commit, stop the moment we leave it.
292        if let Some(b) = *boundary_seq
293            && record.commit_seq != b
294        {
295            *stopped_early = true;
296            return false;
297        }
298
299        *examined += 1;
300
301        if event_matches(&record, query) {
302            events.push(record.clone());
303        }
304
305        // After fully processing a row, decide whether this commit_seq becomes
306        // the page boundary (limit reached, or scan budget exhausted). We finish
307        // the rest of this commit's rows before breaking.
308        if boundary_seq.is_none() && (events.len() >= limit || *examined >= max_scan_rows) {
309            *boundary_seq = Some(record.commit_seq);
310        }
311        true
312    }
313}
314
315/// Build a range key for the start of a commit_seq (empty topic / event_key
316/// sort before every real row at that sequence).
317fn seq_key(commit_seq_i64: i64) -> EncodedKey {
318    EncodedKey::from_values(&[
319        Value::Integer(commit_seq_i64),
320        Value::Text("".into()),
321        Value::Text("".into()),
322    ])
323}
324
325/// Decode an `event_outbox` row into a public record, or `None` if the row
326/// shape is unexpected (defensive; rows are written by a single code path).
327fn decode_event_row(row: &crate::catalog::types::Row) -> Option<EventOutboxRecord> {
328    let (
329        Some(Value::Integer(commit_seq_i64)),
330        Some(Value::Timestamp(ts_i64)),
331        Some(Value::Text(project_id)),
332        Some(Value::Text(scope_id)),
333        Some(Value::Text(topic)),
334        Some(Value::Text(event_key)),
335        Some(Value::Json(payload)),
336    ) = (
337        row.values.first(),
338        row.values.get(1),
339        row.values.get(2),
340        row.values.get(3),
341        row.values.get(4),
342        row.values.get(5),
343        row.values.get(6),
344    )
345    else {
346        return None;
347    };
348    let commit_seq = u64::try_from(*commit_seq_i64).ok()?;
349    let ts_micros = u64::try_from(*ts_i64).ok()?;
350    Some(EventOutboxRecord {
351        commit_seq,
352        ts_micros,
353        project_id: project_id.to_string(),
354        scope_id: scope_id.to_string(),
355        topic: topic.to_string(),
356        event_key: event_key.to_string(),
357        payload_json: payload.to_string(),
358    })
359}
360
361/// Apply the non-range filters (topic, namespace, time window, payload fields).
362fn event_matches(record: &EventOutboxRecord, query: &EventQuery) -> bool {
363    if let Some(topic) = &query.topic
364        && &record.topic != topic
365    {
366        return false;
367    }
368    if let Some(project_id) = &query.project_id
369        && &record.project_id != project_id
370    {
371        return false;
372    }
373    if let Some(scope_id) = &query.scope_id
374        && &record.scope_id != scope_id
375    {
376        return false;
377    }
378    if let Some(from) = query.from_ts_micros
379        && record.ts_micros < from
380    {
381        return false;
382    }
383    if let Some(to) = query.to_ts_micros
384        && record.ts_micros > to
385    {
386        return false;
387    }
388    payload_fields_match(&record.payload_json, &query.field_filters)
389}
390
391/// Match top-level JSON payload fields by their string rendering. A payload
392/// that is not a JSON object, or that lacks a filtered field, does not match.
393fn payload_fields_match(payload_json: &str, filters: &[EventFieldFilter]) -> bool {
394    if filters.is_empty() {
395        return true;
396    }
397    let Ok(parsed) = serde_json::from_str::<serde_json::Value>(payload_json) else {
398        return false;
399    };
400    let Some(obj) = parsed.as_object() else {
401        return false;
402    };
403    for filter in filters {
404        let Some(value) = obj.get(&filter.field) else {
405            return false;
406        };
407        let matches = match value {
408            serde_json::Value::String(s) => s == &filter.equals,
409            serde_json::Value::Number(n) => n.to_string() == filter.equals,
410            serde_json::Value::Bool(b) => b.to_string() == filter.equals,
411            _ => false,
412        };
413        if !matches {
414            return false;
415        }
416    }
417    true
418}
419
420#[cfg(test)]
421mod tests;
aedb/api/event_query.rs

aedb/api/
event_query.rs
