millipede_core/storage/
dataset.rs

1//! Dataset storage contracts.
2
3use super::{StorageError, StorageResult};
4use futures_util::{StreamExt, stream::BoxStream};
5use serde::{Serialize, de::DeserializeOwned};
6
7/// Options controlling dataset listing order and pagination.
8#[derive(Debug, Clone, Default)]
9#[non_exhaustive]
10#[must_use = "list options do nothing unless passed to Dataset::list"]
11pub struct ListOptions {
12    /// Number of items to skip.
13    pub offset: u64,
14    /// Maximum number of items to return.
15    pub limit: Option<u64>,
16    /// Whether to list newest items first.
17    pub desc: bool,
18}
19
20/// A page of dataset items and pagination metadata.
21#[derive(Debug, Clone)]
22pub struct Page<T> {
23    /// Items in this page.
24    pub items: Vec<T>,
25    /// Total number of items in the dataset.
26    pub total: u64,
27    /// Offset used for this page.
28    pub offset: u64,
29    /// Limit used for this page.
30    pub limit: Option<u64>,
31}
32
33/// Dataset identity and timestamps.
34#[derive(Debug, Clone)]
35#[non_exhaustive]
36pub struct DatasetInfo {
37    /// Dataset name.
38    pub name: String,
39    /// Number of stored items.
40    pub item_count: u64,
41    /// Creation timestamp.
42    pub created_at: time::OffsetDateTime,
43    /// Most recent modification timestamp.
44    pub modified_at: time::OffsetDateTime,
45}
46
47impl DatasetInfo {
48    /// Creates dataset metadata for a storage backend.
49    pub fn new(
50        name: String,
51        item_count: u64,
52        created_at: time::OffsetDateTime,
53        modified_at: time::OffsetDateTime,
54    ) -> Self {
55        Self {
56            name,
57            item_count,
58            created_at,
59            modified_at,
60        }
61    }
62}
63
64/// Object-safe storage for append-only JSON records.
65#[async_trait::async_trait]
66pub trait Dataset: Send + Sync {
67    /// Appends one raw JSON value.
68    async fn push_json(&self, item: serde_json::Value) -> StorageResult<()>;
69    /// Appends a batch of raw JSON values.
70    async fn push_json_batch(&self, items: Vec<serde_json::Value>) -> StorageResult<()>;
71    /// Lists raw JSON values according to the supplied options.
72    async fn list_raw(&self, opts: ListOptions) -> StorageResult<Page<serde_json::Value>>;
73    /// Streams raw JSON values according to the supplied options.
74    fn stream_raw(&self, opts: ListOptions) -> BoxStream<'_, StorageResult<serde_json::Value>>;
75    /// Exports the complete dataset as JSON.
76    async fn export_json(&self, path: &std::path::Path) -> StorageResult<()>;
77    /// Exports the complete dataset as CSV.
78    async fn export_csv(&self, path: &std::path::Path) -> StorageResult<()>;
79    /// Returns dataset metadata.
80    async fn info(&self) -> StorageResult<DatasetInfo>;
81}
82
83/// Typed convenience operations available on every [`Dataset`].
84#[async_trait::async_trait]
85pub trait DatasetExt: Dataset {
86    /// Serializes and appends one item.
87    async fn push<T: Serialize + Send + Sync>(&self, item: &T) -> StorageResult<()> {
88        self.push_json(serde_json::to_value(item)?).await
89    }
90
91    /// Serializes and appends a batch of items.
92    async fn push_batch<T: Serialize + Send + Sync>(&self, items: &[T]) -> StorageResult<()> {
93        let items = items
94            .iter()
95            .map(serde_json::to_value)
96            .collect::<Result<Vec<_>, _>>()?;
97        self.push_json_batch(items).await
98    }
99
100    /// Lists and deserializes typed items.
101    async fn list<T: DeserializeOwned>(&self, opts: ListOptions) -> StorageResult<Page<T>> {
102        let page = self.list_raw(opts).await?;
103        Ok(Page {
104            items: page
105                .items
106                .into_iter()
107                .map(serde_json::from_value)
108                .collect::<Result<Vec<_>, _>>()?,
109            total: page.total,
110            offset: page.offset,
111            limit: page.limit,
112        })
113    }
114
115    /// Streams and deserializes typed items.
116    fn stream<T: DeserializeOwned + Send + 'static>(
117        &self,
118        opts: ListOptions,
119    ) -> BoxStream<'_, StorageResult<T>> {
120        Box::pin(self.stream_raw(opts).map(|result| {
121            result.and_then(|value| serde_json::from_value(value).map_err(StorageError::from))
122        }))
123    }
124}
125
126impl<D: Dataset + ?Sized> DatasetExt for D {}
millipede_core/storage/dataset.rs

millipede_core/storage/
dataset.rs
