laminar_connectors/mongodb/
timeseries.rs

1//! `MongoDB` time series collection configuration and validation.
2//!
3//! Time series collections in `MongoDB` use automatic bucketing for efficient
4//! storage and querying of time-stamped measurement data. This module
5//! provides typed configuration for creating and validating time series
6//! collections.
7//!
8//! # Important Constraints
9//!
10//! - Time series collections only accept `insert` operations; other write
11//!   modes are rejected at the sink level.
12//! - `MongoDB` does not support `watch()` (change streams) on time series
13//!   collections — a source targeting one named collection rejects these.
14
15use crate::error::ConnectorError;
16
17const MAX_CUSTOM_BUCKET_SPAN_SECONDS: u32 = 31_536_000;
18
19/// Whether the target collection is a standard or time series collection.
20#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
21#[serde(tag = "kind", rename_all = "snake_case")]
22pub enum CollectionKind {
23    /// Standard `MongoDB` collection.
24    #[default]
25    Standard,
26    /// Time series collection with bucketing configuration.
27    TimeSeries(TimeSeriesConfig),
28}
29
30/// Configuration for a `MongoDB` time series collection.
31///
32/// Maps to `MongoDB`'s `timeseries` collection option. The `time_field`
33/// is required; `meta_field` and TTL are optional.
34#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
35pub struct TimeSeriesConfig {
36    /// The field in each document that contains the date.
37    pub time_field: String,
38
39    /// An optional field that labels the data source (e.g., sensor ID).
40    /// Documents with the same meta value are bucketed together.
41    pub meta_field: Option<String>,
42
43    /// Bucketing granularity.
44    pub granularity: TimeSeriesGranularity,
45
46    /// Optional TTL: automatically delete documents after this many seconds.
47    pub expire_after_seconds: Option<u64>,
48}
49
50impl TimeSeriesConfig {
51    /// Validate server-enforced time-series collection invariants before I/O.
52    pub(crate) fn validate(&self) -> Result<(), ConnectorError> {
53        validate_field_name(&self.time_field, "time_field")?;
54        if let Some(meta_field) = self.meta_field.as_deref() {
55            validate_field_name(meta_field, "meta_field")?;
56            if meta_field == self.time_field {
57                return Err(ConnectorError::ConfigurationError(
58                    "time series meta_field must differ from time_field".into(),
59                ));
60            }
61            if meta_field == "_id" {
62                return Err(ConnectorError::ConfigurationError(
63                    "time series meta_field must not be '_id'".into(),
64                ));
65            }
66        }
67        if self
68            .expire_after_seconds
69            .is_some_and(|ttl| ttl > u64::try_from(i64::MAX).expect("i64::MAX fits u64"))
70        {
71            return Err(ConnectorError::ConfigurationError(
72                "time series expire_after_seconds exceeds MongoDB's signed 64-bit range".into(),
73            ));
74        }
75        if let TimeSeriesGranularity::Custom {
76            bucket_max_span_seconds,
77            bucket_rounding_seconds,
78        } = self.granularity
79        {
80            validate_custom_bucket(bucket_max_span_seconds, bucket_rounding_seconds)?;
81        }
82        Ok(())
83    }
84}
85
86fn validate_field_name(field: &str, label: &str) -> Result<(), ConnectorError> {
87    if field.trim().is_empty() {
88        return Err(ConnectorError::ConfigurationError(format!(
89            "time series {label} must not be empty"
90        )));
91    }
92    if field.contains('\0') {
93        return Err(ConnectorError::ConfigurationError(format!(
94            "time series {label} must not contain NUL"
95        )));
96    }
97    Ok(())
98}
99
100fn validate_custom_bucket(
101    bucket_max_span_seconds: u32,
102    bucket_rounding_seconds: u32,
103) -> Result<(), ConnectorError> {
104    if bucket_max_span_seconds != bucket_rounding_seconds {
105        return Err(ConnectorError::ConfigurationError(format!(
106            "time series custom granularity requires bucket_max_span_seconds ({bucket_max_span_seconds}) \
107             == bucket_rounding_seconds ({bucket_rounding_seconds})"
108        )));
109    }
110    if !(1..=MAX_CUSTOM_BUCKET_SPAN_SECONDS).contains(&bucket_max_span_seconds) {
111        return Err(ConnectorError::ConfigurationError(format!(
112            "time series custom granularity bucket_max_span_seconds must be between 1 and \
113             {MAX_CUSTOM_BUCKET_SPAN_SECONDS}"
114        )));
115    }
116    Ok(())
117}
118
119/// Time series bucketing granularity.
120///
121/// Controls the bucket span for time series collections:
122///
123/// | Granularity | Bucket Span |
124/// |-------------|-------------|
125/// | Seconds     | 1 hour      |
126/// | Minutes     | 24 hours    |
127/// | Hours       | 30 days     |
128/// | Custom      | User-defined (`MongoDB` ≥ 6.3) |
129///
130#[derive(
131    Debug, Clone, Copy, PartialEq, Eq, Hash, Default, serde::Serialize, serde::Deserialize,
132)]
133#[serde(tag = "type", rename_all = "snake_case")]
134pub enum TimeSeriesGranularity {
135    /// Bucket span = 1 hour (default).
136    #[default]
137    Seconds,
138    /// Bucket span = 24 hours.
139    Minutes,
140    /// Bucket span = 30 days.
141    Hours,
142    /// Custom bucketing (`MongoDB` ≥ 6.3).
143    ///
144    /// **Invariant**: `bucket_max_span_seconds` must equal
145    /// `bucket_rounding_seconds`. This is enforced at construction via
146    /// [`TimeSeriesGranularity::custom`].
147    Custom {
148        /// Maximum span of a single bucket in seconds.
149        bucket_max_span_seconds: u32,
150        /// Rounding boundary in seconds (must equal `bucket_max_span_seconds`).
151        bucket_rounding_seconds: u32,
152    },
153}
154
155impl TimeSeriesGranularity {
156    /// Creates a `Custom` granularity, enforcing the invariant that
157    /// `bucket_max_span_seconds == bucket_rounding_seconds`.
158    ///
159    /// # Errors
160    ///
161    /// Returns `ConnectorError::ConfigurationError` if the values differ.
162    pub fn custom(
163        bucket_max_span_seconds: u32,
164        bucket_rounding_seconds: u32,
165    ) -> Result<Self, ConnectorError> {
166        validate_custom_bucket(bucket_max_span_seconds, bucket_rounding_seconds)?;
167        Ok(Self::Custom {
168            bucket_max_span_seconds,
169            bucket_rounding_seconds,
170        })
171    }
172}
173
174impl std::fmt::Display for TimeSeriesGranularity {
175    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
176        match self {
177            Self::Seconds => f.write_str("seconds"),
178            Self::Minutes => f.write_str("minutes"),
179            Self::Hours => f.write_str("hours"),
180            Self::Custom {
181                bucket_max_span_seconds,
182                ..
183            } => write!(f, "custom({bucket_max_span_seconds}s)"),
184        }
185    }
186}
187
188#[cfg(test)]
189mod tests {
190    use super::*;
191
192    #[test]
193    fn test_granularity_default() {
194        assert_eq!(
195            TimeSeriesGranularity::default(),
196            TimeSeriesGranularity::Seconds
197        );
198    }
199
200    #[test]
201    fn test_custom_granularity_valid() {
202        let g = TimeSeriesGranularity::custom(3600, 3600).unwrap();
203        assert!(matches!(g, TimeSeriesGranularity::Custom { .. }));
204    }
205
206    #[test]
207    fn test_custom_granularity_mismatch() {
208        let err = TimeSeriesGranularity::custom(3600, 1800).unwrap_err();
209        assert!(err.to_string().contains("bucket_max_span_seconds"));
210    }
211
212    #[test]
213    fn test_custom_granularity_zero() {
214        let err = TimeSeriesGranularity::custom(0, 0).unwrap_err();
215        assert!(err.to_string().contains("between 1"));
216    }
217
218    #[test]
219    fn custom_granularity_rejects_server_limit_overflow() {
220        let error = TimeSeriesGranularity::custom(31_536_001, 31_536_001).unwrap_err();
221        assert!(error.to_string().contains("31536000"));
222    }
223
224    #[test]
225    fn test_collection_kind_default() {
226        assert!(matches!(
227            CollectionKind::default(),
228            CollectionKind::Standard
229        ));
230    }
231
232    #[test]
233    fn test_granularity_display() {
234        assert_eq!(TimeSeriesGranularity::Seconds.to_string(), "seconds");
235        assert_eq!(TimeSeriesGranularity::Minutes.to_string(), "minutes");
236        assert_eq!(TimeSeriesGranularity::Hours.to_string(), "hours");
237        let custom = TimeSeriesGranularity::custom(7200, 7200).unwrap();
238        assert_eq!(custom.to_string(), "custom(7200s)");
239    }
240
241    #[test]
242    fn time_series_config_rejects_conflicting_metadata_and_ttl_overflow() {
243        let mut config = TimeSeriesConfig {
244            time_field: "ts".into(),
245            meta_field: Some("ts".into()),
246            granularity: TimeSeriesGranularity::Seconds,
247            expire_after_seconds: None,
248        };
249        assert!(config
250            .validate()
251            .unwrap_err()
252            .to_string()
253            .contains("differ"));
254
255        config.meta_field = Some("_id".into());
256        assert!(config.validate().unwrap_err().to_string().contains("_id"));
257
258        config.meta_field = None;
259        config.expire_after_seconds = Some(u64::try_from(i64::MAX).unwrap() + 1);
260        assert!(config
261            .validate()
262            .unwrap_err()
263            .to_string()
264            .contains("signed 64-bit"));
265    }
266
267    #[test]
268    fn config_validation_cannot_bypass_field_and_custom_bucket_invariants() {
269        let mut config = TimeSeriesConfig {
270            time_field: "bad\0field".into(),
271            meta_field: None,
272            granularity: TimeSeriesGranularity::Seconds,
273            expire_after_seconds: None,
274        };
275        assert!(config.validate().unwrap_err().to_string().contains("NUL"));
276
277        config.time_field = "ts".into();
278        config.granularity = TimeSeriesGranularity::Custom {
279            bucket_max_span_seconds: 60,
280            bucket_rounding_seconds: 30,
281        };
282        assert!(config
283            .validate()
284            .unwrap_err()
285            .to_string()
286            .contains("bucket_rounding_seconds"));
287    }
288}
laminar_connectors/mongodb/timeseries.rs

laminar_connectors/mongodb/
timeseries.rs
