infrastore_server/
service.rs

1//! `tonic` service implementation backed by a local `Store`.
2
3use std::path::Path;
4use std::sync::Arc;
5
6use chrono::{DateTime, Utc};
7use infrastore_core::{
8    KeyIdentity, ListFilter, OwnerCategory, Period, Store, TimeSeriesError, TimeSeriesType,
9};
10
11/// Parse an ISO-8601 period from a request, mapping failures to an
12/// `invalid_argument` status.
13fn parse_period(s: &str) -> Result<Period, Status> {
14    Period::from_iso8601(s).map_err(|e| Status::invalid_argument(e.to_string()))
15}
16
17/// Build a [`ListFilter`] from a `ListReq`, mapping bad enums / periods to
18/// `invalid_argument`. Shared by `ListTimeSeries` and `ListKeys`.
19fn filter_from_list_req(req: ListReq) -> Result<ListFilter, Status> {
20    let mut filter = ListFilter::new();
21    if let Some(id) = req.owner_id {
22        filter = filter.owner_id(id);
23    }
24    if let Some(c) = req.owner_category {
25        let pb_c = pb::OwnerCategory::try_from(c)
26            .map_err(|_| Status::invalid_argument(format!("unknown owner_category {c}")))?;
27        filter = filter.owner_category(OwnerCategory::from(pb_c));
28    }
29    if let Some(t) = req.owner_type {
30        filter = filter.owner_type(t);
31    }
32    if let Some(t) = req.time_series_type {
33        let pb_t = pb::TimeSeriesType::try_from(t)
34            .map_err(|_| Status::invalid_argument(format!("unknown time_series_type {t}")))?;
35        filter = filter.time_series_type(TimeSeriesType::from(pb_t));
36    }
37    if let Some(name) = req.name {
38        filter = filter.name(name);
39    }
40    if let Some(iso) = req.resolution {
41        filter = filter.resolution(parse_period(&iso)?);
42    }
43    if let Some(iso) = req.interval {
44        filter = filter.interval(parse_period(&iso)?);
45    }
46    if let Some(f) = req.features {
47        filter = filter.features(features_from_pb(f).map_err(map_convert_err)?);
48    }
49    Ok(filter)
50}
51
52/// An inclusive-start, exclusive-end UTC time range.
53type TimeRange = (DateTime<Utc>, DateTime<Utc>);
54
55/// Parse an optional `(start, end)` RFC3339 range; both must be supplied
56/// together or neither. Shared by `GetTimeSeries` and `BulkRead`.
57fn parse_time_range(
58    start: Option<String>,
59    end: Option<String>,
60) -> Result<Option<TimeRange>, Status> {
61    match (start, end) {
62        (Some(s), Some(e)) => {
63            let start = DateTime::parse_from_rfc3339(&s)
64                .map_err(|err| Status::invalid_argument(format!("start: {err}")))?
65                .with_timezone(&Utc);
66            let end = DateTime::parse_from_rfc3339(&e)
67                .map_err(|err| Status::invalid_argument(format!("end: {err}")))?
68                .with_timezone(&Utc);
69            Ok(Some((start, end)))
70        }
71        (None, None) => Ok(None),
72        _ => Err(Status::invalid_argument(
73            "start_rfc3339 and end_rfc3339 must be supplied together",
74        )),
75    }
76}
77use infrastore_proto::convert::{
78    features_from_pb, forecast_summary_row_to_pb, full_key_to_pb, key_from_pb, metadata_to_pb,
79    requested_type_from_pb, static_summary_row_to_pb, time_series_data_to_get_resp,
80};
81use infrastore_proto::pb::{
82    self, BulkReadReq, BulkReadResp, ConsistencyReq, ConsistencyResp, CountsByTypeResp, CountsReq,
83    CountsResp, DetailedCountsResp, EmptyReq, ForecastParamsReq, ForecastParamsResp,
84    ForecastSummaryResp, GetReq, GetResp, HasReq, HasResp, IntervalsReq, IntervalsResp, KeyReq,
85    KeysReq, KeysResp, ListKeysReq, ListKeysResp, ListOwnerIdsReq, ListOwnerIdsResp, ListReq,
86    ListResp, ResolutionsReq, ResolutionsResp, ResolveForecastKeyReq, ResolveForecastKeyResp,
87    StaticSummaryResp, TimeSeriesMetadata, VerifyReq, VerifyResp,
88    catalog_store_server::{CatalogStore as CatalogStoreSvc, CatalogStoreServer},
89};
90use tokio::sync::Mutex;
91use tonic::{Request, Response, Status};
92
93/// Trait service backed by a `Store`. Read-only RPCs only.
94pub struct CatalogStoreService {
95    store: Arc<Mutex<Store>>,
96}
97
98impl CatalogStoreService {
99    pub fn new(store: Store) -> Self {
100        Self {
101            store: Arc::new(Mutex::new(store)),
102        }
103    }
104
105    pub fn from_path(path: &Path) -> Result<Self, TimeSeriesError> {
106        let store = Store::open(path, true)?;
107        Ok(Self::new(store))
108    }
109
110    pub fn into_server(self) -> CatalogStoreServer<Self> {
111        CatalogStoreServer::new(self)
112    }
113}
114
115/// Map a core `TimeSeriesError` onto a tonic `Status`.
116fn map_err(e: TimeSeriesError) -> Status {
117    match e {
118        TimeSeriesError::NotFound => Status::not_found("time series not found"),
119        TimeSeriesError::DuplicateTimeSeries => Status::already_exists("duplicate"),
120        TimeSeriesError::InvalidParameter(m) => Status::invalid_argument(m),
121        TimeSeriesError::IntegrityError(m) => Status::data_loss(m),
122        TimeSeriesError::ReadOnlyStore => Status::failed_precondition("store is read-only"),
123        TimeSeriesError::ConnectionError(m) => Status::unavailable(m),
124        TimeSeriesError::IncompatibleForecast => {
125            Status::failed_precondition("incompatible forecast")
126        }
127        e @ TimeSeriesError::IncompatibleFormat { .. } => {
128            Status::failed_precondition(e.to_string())
129        }
130        TimeSeriesError::Io(e) => Status::internal(format!("io: {e}")),
131        TimeSeriesError::Sqlite(e) => Status::internal(format!("sqlite: {e}")),
132        TimeSeriesError::Serde(e) => Status::internal(format!("serde: {e}")),
133        // `TimeSeriesError` is non_exhaustive; surface future variants as
134        // internal rather than failing to compile against a newer core.
135        e => Status::internal(e.to_string()),
136    }
137}
138
139fn map_convert_err(e: infrastore_proto::convert::ConvertError) -> Status {
140    Status::invalid_argument(e.to_string())
141}
142
143#[tonic::async_trait]
144impl CatalogStoreSvc for CatalogStoreService {
145    async fn list_time_series(
146        &self,
147        request: Request<ListReq>,
148    ) -> Result<Response<ListResp>, Status> {
149        let filter = filter_from_list_req(request.into_inner())?;
150        let store = self.store.lock().await;
151        let metas = store.list_time_series(filter).map_err(map_err)?;
152        Ok(Response::new(ListResp {
153            metadata: metas.iter().map(metadata_to_pb).collect(),
154        }))
155    }
156
157    async fn get_time_series(&self, request: Request<GetReq>) -> Result<Response<GetResp>, Status> {
158        let req = request.into_inner();
159        let key = req
160            .key
161            .ok_or_else(|| Status::invalid_argument("missing key"))?;
162        let key = key_from_pb(key).map_err(map_convert_err)?;
163        let time_range = parse_time_range(req.start_rfc3339, req.end_rfc3339)?;
164        let store = self.store.lock().await;
165        let data = store.get_time_series(&key, time_range).map_err(map_err)?;
166        Ok(Response::new(time_series_data_to_get_resp(&data)))
167    }
168
169    async fn get_time_series_keys(
170        &self,
171        request: Request<KeysReq>,
172    ) -> Result<Response<KeysResp>, Status> {
173        let req = request.into_inner();
174        let owner_category = pb::OwnerCategory::try_from(req.owner_category)
175            .map_err(|_| {
176                Status::invalid_argument(format!("unknown owner_category {}", req.owner_category))
177            })
178            .map(OwnerCategory::from)?;
179        let store = self.store.lock().await;
180        let keys = store
181            .get_time_series_keys(req.owner_id, owner_category)
182            .map_err(map_err)?;
183        Ok(Response::new(KeysResp {
184            keys: keys.iter().map(full_key_to_pb).collect(),
185        }))
186    }
187
188    async fn get_resolutions(
189        &self,
190        request: Request<ResolutionsReq>,
191    ) -> Result<Response<ResolutionsResp>, Status> {
192        let req = request.into_inner();
193        let ts_type = match req.time_series_type {
194            Some(t) => Some(TimeSeriesType::from(
195                pb::TimeSeriesType::try_from(t).map_err(|_| {
196                    Status::invalid_argument(format!("unknown time_series_type {t}"))
197                })?,
198            )),
199            None => None,
200        };
201        let store = self.store.lock().await;
202        let durations = store.get_resolutions(ts_type).map_err(map_err)?;
203        Ok(Response::new(ResolutionsResp {
204            resolution: durations.iter().map(|p| p.to_iso8601()).collect(),
205        }))
206    }
207
208    async fn get_counts(
209        &self,
210        _request: Request<CountsReq>,
211    ) -> Result<Response<CountsResp>, Status> {
212        let store = self.store.lock().await;
213        let counts = store.get_time_series_counts().map_err(map_err)?;
214        Ok(Response::new(CountsResp {
215            components_with_time_series: counts.components_with_time_series,
216            static_time_series: counts.static_time_series,
217            forecasts: counts.forecasts,
218        }))
219    }
220
221    async fn get_forecast_parameters(
222        &self,
223        request: Request<ForecastParamsReq>,
224    ) -> Result<Response<ForecastParamsResp>, Status> {
225        let req = request.into_inner();
226        let resolution = req.resolution.as_deref().map(parse_period).transpose()?;
227        let interval = req.interval.as_deref().map(parse_period).transpose()?;
228        let store = self.store.lock().await;
229        let params = store
230            .get_forecast_parameters(resolution, interval)
231            .map_err(map_err)?;
232        Ok(Response::new(ForecastParamsResp {
233            horizon: params.horizon.map(|p| p.to_iso8601()),
234            interval: params.interval.map(|p| p.to_iso8601()),
235            count: params.count.map(|c| c as u64),
236            resolution: params.resolution.map(|p| p.to_iso8601()),
237            initial_timestamp_rfc3339: params.initial_timestamp.map(|t| t.to_rfc3339()),
238        }))
239    }
240
241    async fn has_time_series(&self, request: Request<HasReq>) -> Result<Response<HasResp>, Status> {
242        let req = request.into_inner();
243        let key = req
244            .key
245            .ok_or_else(|| Status::invalid_argument("missing key"))?;
246        let key = key_from_pb(key).map_err(map_convert_err)?;
247        let store = self.store.lock().await;
248        let present = store.has_time_series(&key).map_err(map_err)?;
249        Ok(Response::new(HasResp { present }))
250    }
251
252    async fn verify_integrity(
253        &self,
254        _request: Request<VerifyReq>,
255    ) -> Result<Response<VerifyResp>, Status> {
256        let store = self.store.lock().await;
257        let report = store.verify_integrity().map_err(map_err)?;
258        Ok(Response::new(VerifyResp {
259            errors: report.errors,
260        }))
261    }
262
263    // ---- Additive read RPCs (Phase 4.4) ----
264
265    async fn list_keys(
266        &self,
267        request: Request<ListKeysReq>,
268    ) -> Result<Response<ListKeysResp>, Status> {
269        let req = request.into_inner();
270        let filter = filter_from_list_req(req.filter.unwrap_or_default())?;
271        let store = self.store.lock().await;
272        let rows = if req.with_hash {
273            store
274                .list_keys_with_hash(filter)
275                .map_err(map_err)?
276                .into_iter()
277                .map(|(k, h)| pb::list_keys_resp::Row {
278                    key: Some(full_key_to_pb(&k)),
279                    data_hash: Some(h.to_vec()),
280                })
281                .collect()
282        } else {
283            store
284                .list_keys(filter)
285                .map_err(map_err)?
286                .into_iter()
287                .map(|k| pb::list_keys_resp::Row {
288                    key: Some(full_key_to_pb(&k)),
289                    data_hash: None,
290                })
291                .collect()
292        };
293        Ok(Response::new(ListKeysResp { rows }))
294    }
295
296    async fn get_metadata(
297        &self,
298        request: Request<KeyReq>,
299    ) -> Result<Response<TimeSeriesMetadata>, Status> {
300        let key = request
301            .into_inner()
302            .key
303            .ok_or_else(|| Status::invalid_argument("missing key"))?;
304        let key = key_from_pb(key).map_err(map_convert_err)?;
305        let store = self.store.lock().await;
306        let meta = store.get_metadata(&key).map_err(map_err)?;
307        Ok(Response::new(metadata_to_pb(&meta)))
308    }
309
310    async fn bulk_read(
311        &self,
312        request: Request<BulkReadReq>,
313    ) -> Result<Response<BulkReadResp>, Status> {
314        let req = request.into_inner();
315        let time_range = parse_time_range(req.start_rfc3339, req.end_rfc3339)?;
316        let keys = req
317            .keys
318            .into_iter()
319            .map(key_from_pb)
320            .collect::<Result<Vec<_>, _>>()
321            .map_err(map_convert_err)?;
322        let refs: Vec<&KeyIdentity> = keys.iter().collect();
323        let store = self.store.lock().await;
324        let datas = store.bulk_read_range(&refs, time_range).map_err(map_err)?;
325        Ok(Response::new(BulkReadResp {
326            items: datas.iter().map(time_series_data_to_get_resp).collect(),
327        }))
328    }
329
330    async fn get_detailed_counts(
331        &self,
332        _request: Request<EmptyReq>,
333    ) -> Result<Response<DetailedCountsResp>, Status> {
334        let store = self.store.lock().await;
335        let c = store.time_series_counts_detailed().map_err(map_err)?;
336        Ok(Response::new(DetailedCountsResp {
337            components_with_time_series: c.components_with_time_series,
338            supplemental_attributes_with_time_series: c.supplemental_attributes_with_time_series,
339            static_time_series_count: c.static_time_series_count,
340            forecast_count: c.forecast_count,
341        }))
342    }
343
344    async fn get_counts_by_type(
345        &self,
346        _request: Request<EmptyReq>,
347    ) -> Result<Response<CountsByTypeResp>, Status> {
348        let store = self.store.lock().await;
349        let entries = store
350            .counts_by_type()
351            .map_err(map_err)?
352            .into_iter()
353            .map(|(t, n)| pb::counts_by_type_resp::Entry {
354                time_series_type: pb::TimeSeriesType::from(t) as i32,
355                count: n,
356            })
357            .collect();
358        Ok(Response::new(CountsByTypeResp { entries }))
359    }
360
361    async fn list_owner_ids(
362        &self,
363        request: Request<ListOwnerIdsReq>,
364    ) -> Result<Response<ListOwnerIdsResp>, Status> {
365        let req = request.into_inner();
366        let category = pb::OwnerCategory::try_from(req.owner_category)
367            .map_err(|_| {
368                Status::invalid_argument(format!("unknown owner_category {}", req.owner_category))
369            })
370            .map(OwnerCategory::from)?;
371        let ts_type = match req.time_series_type {
372            Some(t) => Some(TimeSeriesType::from(
373                pb::TimeSeriesType::try_from(t).map_err(|_| {
374                    Status::invalid_argument(format!("unknown time_series_type {t}"))
375                })?,
376            )),
377            None => None,
378        };
379        let resolution = req.resolution.as_deref().map(parse_period).transpose()?;
380        let store = self.store.lock().await;
381        let ids = store
382            .list_owner_ids(category, ts_type, resolution)
383            .map_err(map_err)?;
384        Ok(Response::new(ListOwnerIdsResp { owner_id: ids }))
385    }
386
387    async fn get_intervals(
388        &self,
389        request: Request<IntervalsReq>,
390    ) -> Result<Response<IntervalsResp>, Status> {
391        let req = request.into_inner();
392        let ts_type = match req.time_series_type {
393            Some(t) => Some(TimeSeriesType::from(
394                pb::TimeSeriesType::try_from(t).map_err(|_| {
395                    Status::invalid_argument(format!("unknown time_series_type {t}"))
396                })?,
397            )),
398            None => None,
399        };
400        let store = self.store.lock().await;
401        let intervals = store.get_intervals(ts_type).map_err(map_err)?;
402        Ok(Response::new(IntervalsResp {
403            interval: intervals.iter().map(|p| p.to_iso8601()).collect(),
404        }))
405    }
406
407    async fn get_static_summary(
408        &self,
409        _request: Request<EmptyReq>,
410    ) -> Result<Response<StaticSummaryResp>, Status> {
411        let store = self.store.lock().await;
412        let rows = store.static_summary().map_err(map_err)?;
413        Ok(Response::new(StaticSummaryResp {
414            rows: rows.iter().map(static_summary_row_to_pb).collect(),
415        }))
416    }
417
418    async fn get_forecast_summary(
419        &self,
420        _request: Request<EmptyReq>,
421    ) -> Result<Response<ForecastSummaryResp>, Status> {
422        let store = self.store.lock().await;
423        let rows = store.forecast_summary().map_err(map_err)?;
424        Ok(Response::new(ForecastSummaryResp {
425            rows: rows.iter().map(forecast_summary_row_to_pb).collect(),
426        }))
427    }
428
429    async fn check_static_consistency(
430        &self,
431        request: Request<ConsistencyReq>,
432    ) -> Result<Response<ConsistencyResp>, Status> {
433        let resolution = request
434            .into_inner()
435            .resolution
436            .as_deref()
437            .map(parse_period)
438            .transpose()?;
439        let store = self.store.lock().await;
440        let rows = store
441            .check_static_consistency(resolution)
442            .map_err(map_err)?
443            .into_iter()
444            .map(|c| pb::consistency_resp::Row {
445                resolution: c.resolution.to_iso8601(),
446                initial_timestamp_rfc3339: c.initial_timestamp.to_rfc3339(),
447                length: c.length as u64,
448            })
449            .collect();
450        Ok(Response::new(ConsistencyResp { rows }))
451    }
452
453    async fn resolve_forecast_key(
454        &self,
455        request: Request<ResolveForecastKeyReq>,
456    ) -> Result<Response<ResolveForecastKeyResp>, Status> {
457        let req = request.into_inner();
458        let category = pb::OwnerCategory::try_from(req.owner_category)
459            .map_err(|_| {
460                Status::invalid_argument(format!("unknown owner_category {}", req.owner_category))
461            })
462            .map(OwnerCategory::from)?;
463        let resolution = req.resolution.as_deref().map(parse_period).transpose()?;
464        let interval = req.interval.as_deref().map(parse_period).transpose()?;
465        let features = match req.features {
466            Some(f) => features_from_pb(f).map_err(map_convert_err)?,
467            None => infrastore_core::Features::new(),
468        };
469        let requested = requested_type_from_pb(
470            req.requested
471                .ok_or_else(|| Status::invalid_argument("missing requested type"))?,
472        )
473        .map_err(map_convert_err)?;
474        let store = self.store.lock().await;
475        let key = store
476            .resolve_forecast_key(
477                req.owner_id,
478                category,
479                &req.name,
480                resolution,
481                interval,
482                features,
483                requested,
484            )
485            .map_err(map_err)?;
486        Ok(Response::new(ResolveForecastKeyResp {
487            key: Some(full_key_to_pb(&key)),
488        }))
489    }
490}
infrastore_server/service.rs

infrastore_server/
service.rs
