laminar_connectors/mongodb/timeseries/
mod.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;
laminar_connectors/mongodb/timeseries/mod.rs

laminar_connectors/mongodb/timeseries/
mod.rs
