ankurah_core/
storage.rs

1use std::sync::Arc;
2
3use async_trait::async_trait;
4use futures::Stream;
5use tracing::warn;
6
7use crate::error::{MutationError, RetrievalError};
8use ankurah_proto::{Attested, CollectionId, EntityId, EntityState, Event, EventId};
9
10/// One raw logical record emitted by a storage dump.
11#[derive(Debug)]
12pub enum StorageDumpItem {
13    Event(Attested<Event>),
14    State(Attested<EntityState>),
15}
16
17/// Raw logical export implemented by storage engines that support portable
18/// dumps. Physical tables, cursor pages, and materialized values remain an
19/// implementation detail of the engine.
20#[async_trait]
21pub trait StorageDump: StorageEngine {
22    type DumpStream: Stream<Item = Result<StorageDumpItem, RetrievalError>> + Send + 'static;
23
24    /// Stream events followed by states without exposing the engine's
25    /// physical partitioning.
26    async fn dump(&self) -> Result<Self::DumpStream, RetrievalError>;
27}
28
29pub fn state_name(name: &str) -> String { format!("{}_state", name) }
30
31pub fn event_name(name: &str) -> String { format!("{}_event", name) }
32
33#[async_trait]
34pub trait StorageEngine: Send + Sync {
35    type Value;
36    // Opens and/or creates a storage collection.
37    async fn collection(&self, id: &CollectionId) -> Result<Arc<dyn StorageCollection>, RetrievalError>;
38    // Delete all collections and their data from the storage engine
39    async fn delete_all_collections(&self) -> Result<bool, MutationError>;
40}
41
42#[async_trait]
43pub trait StorageCollection: Send + Sync {
44    async fn set_state(&self, state: Attested<EntityState>) -> Result<bool, MutationError>;
45    async fn get_state(&self, id: EntityId) -> Result<Attested<EntityState>, RetrievalError>;
46
47    // Fetch raw entity states matching a selection (predicate + order by + limit)
48    async fn fetch_states(&self, selection: &ankql::ast::Selection) -> Result<Vec<Attested<EntityState>>, RetrievalError>;
49
50    async fn set_states(&self, states: Vec<Attested<EntityState>>) -> Result<(), MutationError> {
51        for state in states {
52            self.set_state(state).await?;
53        }
54        Ok(())
55    }
56
57    async fn get_states(&self, ids: Vec<EntityId>) -> Result<Vec<Attested<EntityState>>, RetrievalError> {
58        let mut states = Vec::new();
59        for id in ids {
60            match self.get_state(id).await {
61                Ok(state) => states.push(state),
62                Err(RetrievalError::EntityNotFound(_)) => {
63                    warn!("Entity not found: {:?}", id);
64                }
65                Err(e) => return Err(e),
66            }
67        }
68        Ok(states)
69    }
70
71    async fn add_event(&self, entity_event: &Attested<Event>) -> Result<bool, MutationError>;
72
73    /// Retrieve a list of events
74    async fn get_events(&self, event_ids: Vec<EventId>) -> Result<Vec<Attested<Event>>, RetrievalError>;
75
76    /// Retrieve all events from the collection
77    async fn dump_entity_events(&self, id: EntityId) -> Result<Vec<Attested<Event>>, RetrievalError>;
78}
79
80/// Manages the storage and state of the collection without any knowledge of the model type
81#[derive(Clone)]
82pub struct StorageCollectionWrapper(pub(crate) Arc<dyn StorageCollection>);
83
84/// Storage interface for a collection
85impl StorageCollectionWrapper {
86    pub fn new(bucket: Arc<dyn StorageCollection>) -> Self { Self(bucket) }
87}
88
89impl std::ops::Deref for StorageCollectionWrapper {
90    type Target = Arc<dyn StorageCollection>;
91    fn deref(&self) -> &Self::Target { &self.0 }
92}
ankurah_core/storage.rs

ankurah_core/
storage.rs
