khive_db/backend/core_stores/
event.rs

1use super::super::StorageBackend;
2use super::map_open_error;
3use crate::stores::event::event_insert_statements;
4use async_trait::async_trait;
5use khive_storage::event::{
6    Event, EventFilter, EventGroupBy, EventPageQuery, EventPageWindow, EventStore,
7    IdempotentEventBatchResult,
8};
9use khive_storage::{
10    BatchWriteSummary, Page, PageRequest, StorageCapability, StorageError, StorageResult,
11};
12use std::collections::BTreeMap;
13use uuid::Uuid;
14
15#[async_trait]
16impl EventStore for StorageBackend {
17    async fn append_event(&self, event: Event) -> StorageResult<()> {
18        self.events()
19            .map_err(|error| map_open_error(error, StorageCapability::Events, "events"))?
20            .append_event(event)
21            .await
22    }
23
24    async fn append_events(&self, events: Vec<Event>) -> StorageResult<BatchWriteSummary> {
25        self.events()
26            .map_err(|error| map_open_error(error, StorageCapability::Events, "events"))?
27            .append_events(events)
28            .await
29    }
30
31    async fn get_event(&self, id: Uuid) -> StorageResult<Option<Event>> {
32        self.events()
33            .map_err(|error| map_open_error(error, StorageCapability::Events, "events"))?
34            .get_event(id)
35            .await
36    }
37
38    async fn query_events(
39        &self,
40        filter: EventFilter,
41        page: PageRequest,
42    ) -> StorageResult<Page<Event>> {
43        self.events()
44            .map_err(|error| map_open_error(error, StorageCapability::Events, "events"))?
45            .query_events(filter, page)
46            .await
47    }
48
49    async fn query_event_page(&self, query: EventPageQuery) -> StorageResult<EventPageWindow> {
50        self.events()
51            .map_err(|error| map_open_error(error, StorageCapability::Events, "events"))?
52            .query_event_page(query)
53            .await
54    }
55
56    async fn count_events(&self, filter: EventFilter) -> StorageResult<u64> {
57        self.events()
58            .map_err(|error| map_open_error(error, StorageCapability::Events, "events"))?
59            .count_events(filter)
60            .await
61    }
62
63    async fn count_events_grouped(
64        &self,
65        filter: EventFilter,
66        group_by: EventGroupBy,
67    ) -> StorageResult<BTreeMap<String, u64>> {
68        self.events()
69            .map_err(|error| map_open_error(error, StorageCapability::Events, "events"))?
70            .count_events_grouped(filter, group_by)
71            .await
72    }
73
74    fn preflight_event(&self, event: &Event) -> StorageResult<()> {
75        event_insert_statements(event).map(|_| ()).map_err(|error| {
76            StorageError::driver(StorageCapability::Events, "preflight_event", error)
77        })
78    }
79
80    async fn append_events_idempotent(
81        &self,
82        events: Vec<Event>,
83    ) -> StorageResult<IdempotentEventBatchResult> {
84        self.events()
85            .map_err(|error| map_open_error(error, StorageCapability::Events, "events"))?
86            .append_events_idempotent(events)
87            .await
88    }
89
90    fn supports_idempotent_audit_batch(&self) -> bool {
91        true
92    }
93}
khive_db/backend/core_stores/event.rs

khive_db/backend/core_stores/
event.rs
