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

spectra_backend_tensorbase/
events.rs
