spectra_core/
storage.rs

1//! Storage backend traits and query filter types for metrics and events.
2//!
3//! # Design notes
4//!
5//! Adapters implement these traits in `spectra-backend-*` crates and inject them through
6//! [`spectra_runtime::SpectraBuilder`]. The default trait methods provide safe stubs so
7//! partial implementations can ship incrementally.
8//!
9//! # Lifecycle example
10//!
11//! ```no_run
12//! # use std::sync::Arc;
13//! # use chrono::Utc;
14//! # use serde_json::json;
15//! # use spectra_core::{
16//! #     EventStorageBackend, MetricsStorageBackend, NoOpEventBackend, NoOpMetricsBackend,
17//! #     MetricsQueryRange, EventsQueryFilter,
18//! # };
19//! # async fn demo() -> spectra_core::Result<()> {
20//! let metrics = Arc::new(NoOpMetricsBackend);
21//! let events = Arc::new(NoOpEventBackend);
22//!
23//! metrics.record_counter("cache_hits", &json!({}), 1, Utc::now()).await?;
24//! events.append_row("request_log", &json!({"ok": true}), Utc::now(), None).await?;
25//!
26//! let points = metrics.query_range(MetricsQueryRange {
27//!     metric_name: "cache_hits".into(),
28//!     start: Utc::now() - chrono::Duration::hours(1),
29//!     end: Utc::now(),
30//!     label_matchers: vec![],
31//! }).await?;
32//! let rows = events.query_rows(EventsQueryFilter {
33//!     table: "request_log".into(),
34//!     ..Default::default()
35//! }).await?;
36//! # let _ = (points, rows);
37//! # Ok(())
38//! # }
39//! ```
40//!
41use std::sync::Arc;
42
43use async_trait::async_trait;
44use chrono::{DateTime, Utc};
45use serde_json::Value;
46
47use crate::error::Result;
48
49/// Storage engine identifier for query routing and UI.
50#[derive(Debug, Clone, Copy, PartialEq, Eq)]
51pub enum StorageEngineType {
52    /// No-op backend (discards writes, returns empty reads).
53    NoOp,
54    /// In-memory backend.
55    Mem,
56    /// SQLite embedded backend.
57    Sqlite,
58    /// TensorBase backend.
59    TensorBase,
60    /// ClickHouse backend.
61    ClickHouse,
62}
63
64/// Time-series query parameters.
65#[derive(Debug, Clone)]
66pub struct MetricsQueryRange {
67    /// Metric family name.
68    pub metric_name: String,
69    /// Inclusive range start timestamp.
70    pub start: DateTime<Utc>,
71    /// Inclusive range end timestamp.
72    pub end: DateTime<Utc>,
73    /// Label equality matchers.
74    pub label_matchers: Vec<crate::query::LabelMatcher>,
75}
76
77/// Row filter for event log queries (adapter input).
78#[derive(Debug, Clone, Default)]
79pub struct EventsQueryFilter {
80    /// Event table name.
81    pub table: String,
82    /// Optional inclusive range start.
83    pub start: Option<DateTime<Utc>>,
84    /// Optional inclusive range end.
85    pub end: Option<DateTime<Utc>>,
86    /// Optional partition granularity (`"hourly"` or `"daily"`).
87    pub partition: Option<String>,
88    /// Maximum rows to return.
89    pub limit: Option<u32>,
90    /// Row offset for pagination.
91    pub offset: Option<u32>,
92    /// Column to sort by.
93    pub sort_field: Option<String>,
94    /// Whether sort is descending.
95    pub sort_desc: bool,
96    /// Structured row filter model.
97    pub filter: crate::query::GridFilterModel,
98}
99
100/// Aggregate query for chart views (adapter input).
101#[derive(Debug, Clone)]
102pub struct EventsAggregateFilter {
103    /// Event table name.
104    pub table: String,
105    /// Inclusive range start timestamp.
106    pub start: DateTime<Utc>,
107    /// Inclusive range end timestamp.
108    pub end: DateTime<Utc>,
109    /// Optional partition granularity (`"hourly"` or `"daily"`).
110    pub partition: Option<String>,
111    /// Structured row filter model.
112    pub filter: crate::query::GridFilterModel,
113    /// Aggregation measure.
114    pub measure: crate::query::EventMeasure,
115    /// Field to sum when measure is sum.
116    pub measure_field: Option<String>,
117    /// Time bucket width in seconds.
118    pub time_bucket_secs: Option<u64>,
119    /// Field to group by for slice views.
120    pub group_by_field: Option<String>,
121}
122
123/// Metric point for query results.
124#[derive(Debug, Clone)]
125pub struct MetricPoint {
126    /// Sample timestamp.
127    pub ts: DateTime<Utc>,
128    /// Sample value.
129    pub value: f64,
130    /// Label set as JSON.
131    pub labels: Value,
132}
133
134/// Event row for query results.
135#[derive(Debug, Clone)]
136pub struct EventRow {
137    /// Event timestamp.
138    pub ts: DateTime<Utc>,
139    /// Event field payload.
140    pub fields: Value,
141}
142
143/// One metrics write row for batch insert (counter or gauge).
144#[derive(Debug, Clone)]
145pub struct MetricWriteRow {
146    /// Metric family name.
147    pub name: String,
148    /// `"counter"` or `"gauge"`.
149    pub kind: &'static str,
150    /// Counter delta or gauge value as JSON number.
151    pub value: Value,
152    /// Label set as JSON.
153    pub labels: Value,
154    /// Sample timestamp.
155    pub ts: DateTime<Utc>,
156    /// Optional correlation identifier.
157    pub correlation_id: Option<String>,
158}
159
160/// One event write row for batch insert.
161#[derive(Debug, Clone)]
162pub struct EventWriteRow {
163    /// Event table name.
164    pub table: String,
165    /// Event field payload.
166    pub fields: Value,
167    /// Event timestamp.
168    pub ts: DateTime<Utc>,
169    /// Optional correlation identifier.
170    pub correlation_id: Option<String>,
171}
172
173/// Metrics storage: subscribers call `record_*`; explore UI calls `query_range`.
174#[async_trait]
175pub trait MetricsStorageBackend: Send + Sync {
176    /// Returns the engine type for this backend.
177    fn engine_type(&self) -> StorageEngineType;
178
179    /// Records a counter increment.
180    async fn record_counter(
181        &self,
182        name: &str,
183        labels: &Value,
184        delta: i64,
185        ts: DateTime<Utc>,
186    ) -> Result<()>;
187
188    /// Records a gauge sample.
189    async fn record_gauge(
190        &self,
191        name: &str,
192        labels: &Value,
193        value: f64,
194        ts: DateTime<Utc>,
195    ) -> Result<()>;
196
197    /// Queries a time range of metric points.
198    async fn query_range(&self, _query: MetricsQueryRange) -> Result<Vec<MetricPoint>> {
199        Ok(Vec::new())
200    }
201
202    /// Batch write metrics (default: per-row dispatch).
203    async fn record_metrics_batch(&self, rows: &[MetricWriteRow]) -> Result<()> {
204        for row in rows {
205            if row.kind == "counter" {
206                let delta = row.value.as_i64().unwrap_or(0);
207                self.record_counter(&row.name, &row.labels, delta, row.ts)
208                    .await?;
209            } else {
210                let value = row.value.as_f64().unwrap_or(0.0);
211                self.record_gauge(&row.name, &row.labels, value, row.ts)
212                    .await?;
213            }
214        }
215        Ok(())
216    }
217}
218
219/// Event storage: subscribers call `append_row`; explore UI calls `query_rows`.
220#[async_trait]
221pub trait EventStorageBackend: Send + Sync {
222    /// Returns the engine type for this backend.
223    fn engine_type(&self) -> StorageEngineType;
224
225    /// Appends one event row.
226    async fn append_row(
227        &self,
228        table: &str,
229        fields: &Value,
230        ts: DateTime<Utc>,
231        correlation_id: Option<&str>,
232    ) -> Result<()>;
233
234    /// Queries event rows matching the filter.
235    async fn query_rows(&self, _filter: EventsQueryFilter) -> Result<Vec<EventRow>> {
236        Ok(Vec::new())
237    }
238
239    /// Queries aggregated chart data.
240    async fn query_aggregate(
241        &self,
242        _filter: EventsAggregateFilter,
243    ) -> Result<crate::query::EventAggregateResult> {
244        Ok(crate::query::EventAggregateResult::TimeSeries {
245            series: Vec::new(),
246            headline: Vec::new(),
247        })
248    }
249
250    /// Batch append event rows (default: per-row dispatch).
251    async fn append_rows_batch(&self, rows: &[EventWriteRow]) -> Result<()> {
252        for row in rows {
253            self.append_row(
254                &row.table,
255                &row.fields,
256                row.ts,
257                row.correlation_id.as_deref(),
258            )
259            .await?;
260        }
261        Ok(())
262    }
263}
264
265/// No-op metrics backend (default before host wiring).
266#[derive(Debug, Default, Clone, Copy)]
267pub struct NoOpMetricsBackend;
268
269#[async_trait]
270impl MetricsStorageBackend for NoOpMetricsBackend {
271    fn engine_type(&self) -> StorageEngineType {
272        StorageEngineType::NoOp
273    }
274
275    async fn record_counter(&self, _: &str, _: &Value, _: i64, _: DateTime<Utc>) -> Result<()> {
276        Ok(())
277    }
278
279    async fn record_gauge(&self, _: &str, _: &Value, _: f64, _: DateTime<Utc>) -> Result<()> {
280        Ok(())
281    }
282}
283
284/// No-op event backend (default before host wiring).
285#[derive(Debug, Default, Clone, Copy)]
286pub struct NoOpEventBackend;
287
288#[async_trait]
289impl EventStorageBackend for NoOpEventBackend {
290    fn engine_type(&self) -> StorageEngineType {
291        StorageEngineType::NoOp
292    }
293
294    async fn append_row(
295        &self,
296        _: &str,
297        _: &Value,
298        _: DateTime<Utc>,
299        _: Option<&str>,
300    ) -> Result<()> {
301        Ok(())
302    }
303}
304
305/// Shared handle to a metrics storage backend.
306pub type SharedMetricsBackend = Arc<dyn MetricsStorageBackend>;
307/// Shared handle to an event storage backend.
308pub type SharedEventBackend = Arc<dyn EventStorageBackend>;
spectra_core/storage.rs

spectra_core/
storage.rs
