spectra_backend_tensorbase/
events.rs

1//! TensorBase events storage adapter.
2
3use async_trait::async_trait;
4use chrono::{DateTime, Utc};
5use serde_json::Value;
6use spectra_backend_remote_common::RemoteEventsBackend;
7use spectra_core::{
8    EventAggregateResult, EventRow, EventStorageBackend, EventWriteRow, EventsAggregateFilter,
9    EventsQueryFilter, Result, StorageEngineType,
10};
11
12/// Scale-out TensorBase structured-event storage over its ClickHouse-compatible protocol.
13///
14/// Use [`connect`](Self::connect) with a full protocol URL or
15/// [`connect_host`](Self::connect_host) for the default native port (`9528`). Pair this type
16/// with `TensorBaseMetricsBackend` when building the runtime.
17///
18/// # Examples
19///
20/// Facade wiring through `Spectra::builder()` (requires the `spectra` crate with the
21/// `tensorbase` feature):
22///
23/// ```ignore
24/// use std::sync::Arc;
25/// use spectra::{Spectra, TensorBaseEventsBackend, TensorBaseMetricsBackend};
26///
27/// # async fn start() -> spectra::Result<()> {
28/// let url = "tcp://127.0.0.1:9528";
29/// let spectra = Spectra::builder()
30///     .metrics_backend(Arc::new(TensorBaseMetricsBackend::connect(url).await?))
31///     .events_backend(Arc::new(TensorBaseEventsBackend::connect(url).await?))
32///     .build()?;
33/// # let _ = spectra;
34/// # Ok(())
35/// # }
36/// ```
37pub struct TensorBaseEventsBackend(RemoteEventsBackend);
38
39impl TensorBaseEventsBackend {
40    /// Connect with a full TensorBase protocol URL and ensure event tables exist.
41    ///
42    /// Accepts `tcp://host:9528` (native) or an HTTP ClickHouse-compatible endpoint. The call
43    /// is async and executes DDL.
44    ///
45    /// # Examples
46    ///
47    /// ```ignore
48    /// # async fn example() -> spectra_core::Result<()> {
49    /// use spectra_backend_tensorbase::TensorBaseEventsBackend;
50    ///
51    /// let backend = TensorBaseEventsBackend::connect("tcp://127.0.0.1:9528").await?;
52    /// # let _ = backend;
53    /// # Ok(())
54    /// # }
55    /// ```
56    pub async fn connect(url: &str) -> Result<Self> {
57        Ok(Self(
58            RemoteEventsBackend::connect(
59                url,
60                StorageEngineType::TensorBase,
61                &crate::ddl::events_ddl(),
62            )
63            .await?,
64        ))
65    }
66
67    /// Connect using a host name and the default native port (`9528`).
68    ///
69    /// Equivalent to [`connect`](Self::connect) with `tcp://{host}:9528`.
70    ///
71    /// # Examples
72    ///
73    /// ```ignore
74    /// # async fn example() -> spectra_core::Result<()> {
75    /// use spectra_backend_tensorbase::TensorBaseEventsBackend;
76    ///
77    /// let backend = TensorBaseEventsBackend::connect_host("127.0.0.1").await?;
78    /// # let _ = backend;
79    /// # Ok(())
80    /// # }
81    /// ```
82    pub async fn connect_host(host: &str) -> Result<Self> {
83        Self::connect(&crate::ddl::default_url(host)).await
84    }
85
86    /// In-memory stub that preserves the TensorBase engine type for storage-contract tests.
87    ///
88    /// # Examples
89    ///
90    /// ```
91    /// use spectra_backend_tensorbase::TensorBaseEventsBackend;
92    /// use spectra_core::{EventStorageBackend, StorageEngineType};
93    ///
94    /// let backend = TensorBaseEventsBackend::in_memory_stub();
95    /// assert_eq!(backend.engine_type(), StorageEngineType::TensorBase);
96    /// ```
97    pub fn in_memory_stub() -> Self {
98        Self(RemoteEventsBackend::in_memory_for_test(
99            StorageEngineType::TensorBase,
100        ))
101    }
102
103    /// In-memory stub for unit tests in this crate.
104    #[cfg(test)]
105    pub fn in_memory_for_test() -> Self {
106        Self::in_memory_stub()
107    }
108}
109
110#[async_trait]
111impl EventStorageBackend for TensorBaseEventsBackend {
112    fn engine_type(&self) -> StorageEngineType {
113        self.0.engine_type()
114    }
115
116    async fn append_row(
117        &self,
118        table: &str,
119        fields: &Value,
120        ts: DateTime<Utc>,
121        correlation_id: Option<&str>,
122    ) -> Result<()> {
123        self.0.append_row(table, fields, ts, correlation_id).await
124    }
125
126    async fn append_rows_batch(&self, rows: &[EventWriteRow]) -> Result<()> {
127        self.0.append_rows_batch(rows).await
128    }
129
130    async fn query_rows(&self, filter: EventsQueryFilter) -> Result<Vec<EventRow>> {
131        self.0.query_rows(filter).await
132    }
133
134    async fn query_aggregate(&self, filter: EventsAggregateFilter) -> Result<EventAggregateResult> {
135        self.0.query_aggregate(filter).await
136    }
137}
138
139#[cfg(test)]
140mod tests {
141    use super::*;
142    use serde_json::json;
143
144    #[tokio::test]
145    async fn tensorbase_events_roundtrip_in_memory() {
146        let backend = TensorBaseEventsBackend::in_memory_for_test();
147        let ts = Utc::now();
148        backend
149            .append_row("req", &json!({"x": 1}), ts, None)
150            .await
151            .expect("write");
152        let rows = backend
153            .query_rows(EventsQueryFilter {
154                table: "req".into(),
155                ..Default::default()
156            })
157            .await
158            .expect("query");
159        assert_eq!(rows.len(), 1);
160    }
161
162    #[tokio::test]
163    #[ignore = "requires SPECTRA_TENSORBASE_URL"]
164    async fn tensorbase_events_integration() {
165        let url = std::env::var("SPECTRA_TENSORBASE_URL").expect("SPECTRA_TENSORBASE_URL");
166        let backend = TensorBaseEventsBackend::connect(&url)
167            .await
168            .expect("connect");
169        let ts = Utc::now();
170        backend
171            .append_row("integration_req", &json!({"ok": true}), ts, None)
172            .await
173            .expect("write");
174        let rows = backend
175            .query_rows(EventsQueryFilter {
176                table: "integration_req".into(),
177                ..Default::default()
178            })
179            .await
180            .expect("query");
181        assert!(!rows.is_empty());
182    }
183}
spectra_backend_tensorbase/events.rs

spectra_backend_tensorbase/
events.rs
