s2_api/v1/
config.rs

1use std::{str::FromStr, time::Duration};
2
3use http::{HeaderName, HeaderValue};
4use s2_common::{http::ParseableHeader, maybe::Maybe};
5use serde::{Deserialize, Serialize};
6
7#[rustfmt::skip]
8#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
9#[cfg_attr(feature = "utoipa", derive(utoipa::ToSchema))]
10#[serde(rename_all = "kebab-case")]
11pub enum StorageClass {
12    /// Append tail latency under 400 milliseconds with s2.dev.
13    Standard,
14    /// Append tail latency under 40 milliseconds with s2.dev.
15    Express,
16}
17
18impl From<StorageClass> for s2_common::config::StorageClass {
19    fn from(value: StorageClass) -> Self {
20        match value {
21            StorageClass::Express => Self::Express,
22            StorageClass::Standard => Self::Standard,
23        }
24    }
25}
26
27impl From<s2_common::config::StorageClass> for StorageClass {
28    fn from(value: s2_common::config::StorageClass) -> Self {
29        match value {
30            s2_common::config::StorageClass::Express => Self::Express,
31            s2_common::config::StorageClass::Standard => Self::Standard,
32        }
33    }
34}
35
36#[rustfmt::skip]
37#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
38#[cfg_attr(feature = "utoipa", derive(utoipa::ToSchema))]
39#[serde(rename_all = "kebab-case")]
40pub enum RetentionPolicy {
41    /// Age in seconds for automatic trimming of records older than this threshold.
42    /// This must be set to a value greater than 0 seconds.
43    Age(u64),
44    /// Retain records unless explicitly trimmed.
45    Infinite(InfiniteRetention)
46}
47
48#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
49#[cfg_attr(feature = "utoipa", derive(utoipa::ToSchema))]
50#[serde(rename_all = "kebab-case")]
51pub struct InfiniteRetention {}
52
53impl TryFrom<RetentionPolicy> for s2_common::config::RetentionPolicy {
54    type Error = s2_common::ValidationError;
55
56    fn try_from(value: RetentionPolicy) -> Result<Self, Self::Error> {
57        let policy = match value {
58            RetentionPolicy::Age(age) => Self::Age(Duration::from_secs(age)),
59            RetentionPolicy::Infinite(_) => Self::Infinite(),
60        };
61        policy.validate()
62    }
63}
64
65impl From<s2_common::config::RetentionPolicy> for RetentionPolicy {
66    fn from(value: s2_common::config::RetentionPolicy) -> Self {
67        match value {
68            s2_common::config::RetentionPolicy::Age(age) => Self::Age(age.as_secs()),
69            s2_common::config::RetentionPolicy::Infinite() => Self::Infinite(InfiniteRetention {}),
70        }
71    }
72}
73
74#[rustfmt::skip]
75#[derive(Debug, Default, PartialEq, Eq, Clone, Serialize, Deserialize)]
76#[cfg_attr(feature = "utoipa", derive(utoipa::ToSchema))]
77#[serde(rename_all = "kebab-case")]
78pub enum TimestampingMode {
79    /// Prefer client-specified timestamp if present otherwise use arrival time.
80    #[default]
81    ClientPrefer,
82    /// Require a client-specified timestamp and reject the append if it is missing.
83    ClientRequire,
84    /// Use the arrival time and ignore any client-specified timestamp.
85    Arrival,
86}
87
88impl From<TimestampingMode> for s2_common::config::TimestampingMode {
89    fn from(value: TimestampingMode) -> Self {
90        match value {
91            TimestampingMode::ClientPrefer => Self::ClientPrefer,
92            TimestampingMode::ClientRequire => Self::ClientRequire,
93            TimestampingMode::Arrival => Self::Arrival,
94        }
95    }
96}
97
98impl From<s2_common::config::TimestampingMode> for TimestampingMode {
99    fn from(value: s2_common::config::TimestampingMode) -> Self {
100        match value {
101            s2_common::config::TimestampingMode::ClientPrefer => Self::ClientPrefer,
102            s2_common::config::TimestampingMode::ClientRequire => Self::ClientRequire,
103            s2_common::config::TimestampingMode::Arrival => Self::Arrival,
104        }
105    }
106}
107
108#[rustfmt::skip]
109#[derive(Debug, Default, PartialEq, Eq, Clone, Serialize, Deserialize)]
110#[cfg_attr(feature = "utoipa", derive(utoipa::ToSchema))]
111pub struct TimestampingConfig {
112    /// Timestamping mode for appends that influences how timestamps are handled.
113    pub mode: Option<TimestampingMode>,
114    /// Allow client-specified timestamps to exceed the arrival time.
115    /// If this is `false` or not set, client timestamps will be capped at the arrival time.
116    pub uncapped: Option<bool>,
117}
118
119impl TimestampingConfig {
120    pub fn to_opt(config: s2_common::config::OptionalTimestampingConfig) -> Option<Self> {
121        let config = TimestampingConfig {
122            mode: config.mode.map(Into::into),
123            uncapped: config.uncapped,
124        };
125        if config == Self::default() {
126            None
127        } else {
128            Some(config)
129        }
130    }
131}
132
133impl From<s2_common::config::TimestampingConfig> for TimestampingConfig {
134    fn from(value: s2_common::config::TimestampingConfig) -> Self {
135        Self {
136            mode: Some(value.mode.into()),
137            uncapped: Some(value.uncapped),
138        }
139    }
140}
141
142impl From<s2_common::config::OptionalTimestampingConfig> for TimestampingConfig {
143    fn from(value: s2_common::config::OptionalTimestampingConfig) -> Self {
144        Self {
145            mode: value.mode.map(Into::into),
146            uncapped: value.uncapped,
147        }
148    }
149}
150
151impl From<TimestampingConfig> for s2_common::config::OptionalTimestampingConfig {
152    fn from(value: TimestampingConfig) -> Self {
153        Self {
154            mode: value.mode.map(Into::into),
155            uncapped: value.uncapped,
156        }
157    }
158}
159
160#[rustfmt::skip]
161#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
162#[cfg_attr(feature = "utoipa", derive(utoipa::ToSchema))]
163pub struct TimestampingReconfiguration {
164    /// Timestamping mode for appends that influences how timestamps are handled.
165    #[serde(default, skip_serializing_if = "Maybe::is_unspecified")]
166    #[cfg_attr(feature = "utoipa", schema(value_type = Option<TimestampingMode>))]
167    pub mode: Maybe<Option<TimestampingMode>>,
168    /// Allow client-specified timestamps to exceed the arrival time.
169    #[serde(default, skip_serializing_if = "Maybe::is_unspecified")]
170    #[cfg_attr(feature = "utoipa", schema(value_type = Option<bool>))]
171    pub uncapped: Maybe<Option<bool>>,
172}
173
174impl From<TimestampingReconfiguration> for s2_common::config::TimestampingReconfiguration {
175    fn from(value: TimestampingReconfiguration) -> Self {
176        Self {
177            mode: value.mode.map_opt(Into::into),
178            uncapped: value.uncapped,
179        }
180    }
181}
182
183impl From<s2_common::config::TimestampingReconfiguration> for TimestampingReconfiguration {
184    fn from(value: s2_common::config::TimestampingReconfiguration) -> Self {
185        Self {
186            mode: value.mode.map_opt(Into::into),
187            uncapped: value.uncapped,
188        }
189    }
190}
191
192#[rustfmt::skip]
193#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
194#[cfg_attr(feature = "utoipa", derive(utoipa::ToSchema))]
195pub struct DeleteOnEmptyConfig {
196    /// Minimum age in seconds before an empty stream can be deleted.
197    /// Set to 0 (default) to disable delete-on-empty (don't delete automatically).
198    #[serde(default)]
199    pub min_age_secs: u64,
200}
201
202impl DeleteOnEmptyConfig {
203    pub fn to_opt(config: s2_common::config::OptionalDeleteOnEmptyConfig) -> Option<Self> {
204        config.min_age.map(|min_age| DeleteOnEmptyConfig {
205            min_age_secs: min_age.as_secs(),
206        })
207    }
208}
209
210impl From<s2_common::config::DeleteOnEmptyConfig> for DeleteOnEmptyConfig {
211    fn from(value: s2_common::config::DeleteOnEmptyConfig) -> Self {
212        Self {
213            min_age_secs: value.min_age.as_secs(),
214        }
215    }
216}
217
218impl From<s2_common::config::OptionalDeleteOnEmptyConfig> for DeleteOnEmptyConfig {
219    fn from(value: s2_common::config::OptionalDeleteOnEmptyConfig) -> Self {
220        Self {
221            min_age_secs: value.min_age.unwrap_or_default().as_secs(),
222        }
223    }
224}
225
226impl From<DeleteOnEmptyConfig> for s2_common::config::DeleteOnEmptyConfig {
227    fn from(value: DeleteOnEmptyConfig) -> Self {
228        Self {
229            min_age: Duration::from_secs(value.min_age_secs),
230        }
231    }
232}
233
234impl From<DeleteOnEmptyConfig> for s2_common::config::OptionalDeleteOnEmptyConfig {
235    fn from(value: DeleteOnEmptyConfig) -> Self {
236        Self {
237            min_age: Some(Duration::from_secs(value.min_age_secs)),
238        }
239    }
240}
241
242#[rustfmt::skip]
243#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
244#[cfg_attr(feature = "utoipa", derive(utoipa::ToSchema))]
245pub struct DeleteOnEmptyReconfiguration {
246    /// Minimum age in seconds before an empty stream can be deleted.
247    /// Set to 0 to disable delete-on-empty (don't delete automatically).
248    #[serde(default, skip_serializing_if = "Maybe::is_unspecified")]
249    #[cfg_attr(feature = "utoipa", schema(value_type = Option<u64>))]
250    pub min_age_secs: Maybe<Option<u64>>,
251}
252
253impl From<DeleteOnEmptyReconfiguration> for s2_common::config::DeleteOnEmptyReconfiguration {
254    fn from(value: DeleteOnEmptyReconfiguration) -> Self {
255        Self {
256            min_age: value.min_age_secs.map_opt(Duration::from_secs),
257        }
258    }
259}
260
261impl From<s2_common::config::DeleteOnEmptyReconfiguration> for DeleteOnEmptyReconfiguration {
262    fn from(value: s2_common::config::DeleteOnEmptyReconfiguration) -> Self {
263        Self {
264            min_age_secs: value.min_age.map_opt(|d| d.as_secs()),
265        }
266    }
267}
268
269#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize, Deserialize)]
270#[cfg_attr(feature = "utoipa", derive(utoipa::ToSchema))]
271pub enum EncryptionAlgorithm {
272    /// AEGIS-256 authenticated encryption.
273    #[serde(rename = "aegis-256")]
274    Aegis256,
275    /// AES-256-GCM authenticated encryption.
276    #[serde(rename = "aes-256-gcm")]
277    Aes256Gcm,
278}
279
280impl From<EncryptionAlgorithm> for s2_common::encryption::EncryptionAlgorithm {
281    fn from(value: EncryptionAlgorithm) -> Self {
282        match value {
283            EncryptionAlgorithm::Aegis256 => Self::Aegis256,
284            EncryptionAlgorithm::Aes256Gcm => Self::Aes256Gcm,
285        }
286    }
287}
288
289impl From<s2_common::encryption::EncryptionAlgorithm> for EncryptionAlgorithm {
290    fn from(value: s2_common::encryption::EncryptionAlgorithm) -> Self {
291        match value {
292            s2_common::encryption::EncryptionAlgorithm::Aegis256 => Self::Aegis256,
293            s2_common::encryption::EncryptionAlgorithm::Aes256Gcm => Self::Aes256Gcm,
294        }
295    }
296}
297
298#[rustfmt::skip]
299#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
300#[cfg_attr(feature = "utoipa", derive(utoipa::ToSchema))]
301pub struct StreamConfig {
302    /// Storage class for recent writes.
303    pub storage_class: Option<StorageClass>,
304    /// Retention policy for the stream.
305    /// If unspecified, the default is to retain records for 7 days.
306    pub retention_policy: Option<RetentionPolicy>,
307    /// Timestamping behavior.
308    pub timestamping: Option<TimestampingConfig>,
309    /// Delete-on-empty configuration.
310    #[serde(default)]
311    pub delete_on_empty: Option<DeleteOnEmptyConfig>,
312}
313
314impl StreamConfig {
315    pub fn to_opt(config: s2_common::config::OptionalStreamConfig) -> Option<Self> {
316        let s2_common::config::OptionalStreamConfig {
317            storage_class,
318            retention_policy,
319            timestamping,
320            delete_on_empty,
321        } = config;
322
323        let config = StreamConfig {
324            storage_class: storage_class.map(Into::into),
325            retention_policy: retention_policy.map(Into::into),
326            timestamping: TimestampingConfig::to_opt(timestamping),
327            delete_on_empty: DeleteOnEmptyConfig::to_opt(delete_on_empty),
328        };
329        if config == Self::default() {
330            None
331        } else {
332            Some(config)
333        }
334    }
335
336    /// Encode as compact JSON for the `s2-stream-config` header.
337    pub fn to_header_value(&self) -> HeaderValue {
338        let json = serde_json::to_string(self).expect("StreamConfig serializes to JSON");
339        HeaderValue::from_str(&json).expect("compact JSON of StreamConfig is a valid header value")
340    }
341}
342
343impl From<s2_common::config::StreamConfig> for StreamConfig {
344    fn from(value: s2_common::config::StreamConfig) -> Self {
345        let s2_common::config::StreamConfig {
346            storage_class,
347            retention_policy,
348            timestamping,
349            delete_on_empty,
350        } = value;
351
352        Self {
353            storage_class: Some(storage_class.into()),
354            retention_policy: Some(retention_policy.into()),
355            timestamping: Some(timestamping.into()),
356            delete_on_empty: Some(delete_on_empty.into()),
357        }
358    }
359}
360
361pub static STREAM_CONFIG_HEADER: HeaderName = HeaderName::from_static("s2-stream-config");
362
363/// Value of the `s2-stream-config` header: a JSON-encoded [`StreamConfig`] to apply over the
364/// basin's default stream config if the stream is created on append or read. Ignored if the
365/// stream already exists.
366#[derive(Debug, Clone, PartialEq, Eq)]
367pub struct StreamConfigHeader(pub s2_common::config::OptionalStreamConfig);
368
369impl FromStr for StreamConfigHeader {
370    type Err = s2_common::ValidationError;
371
372    fn from_str(s: &str) -> Result<Self, Self::Err> {
373        let config: StreamConfig =
374            serde_json::from_str(s).map_err(|e| format!("invalid JSON: {e}"))?;
375        Ok(Self(config.try_into()?))
376    }
377}
378
379impl ParseableHeader for StreamConfigHeader {
380    fn name() -> &'static HeaderName {
381        &STREAM_CONFIG_HEADER
382    }
383}
384
385impl TryFrom<StreamConfig> for s2_common::config::OptionalStreamConfig {
386    type Error = s2_common::ValidationError;
387
388    fn try_from(value: StreamConfig) -> Result<Self, Self::Error> {
389        let StreamConfig {
390            storage_class,
391            retention_policy,
392            timestamping,
393            delete_on_empty,
394        } = value;
395
396        let retention_policy = match retention_policy {
397            None => None,
398            Some(policy) => Some(policy.try_into()?),
399        };
400
401        let config = Self {
402            storage_class: storage_class.map(Into::into),
403            retention_policy,
404            timestamping: timestamping.map(Into::into).unwrap_or_default(),
405            delete_on_empty: delete_on_empty.map(Into::into).unwrap_or_default(),
406        };
407        config.validate()?;
408        Ok(config)
409    }
410}
411
412#[rustfmt::skip]
413#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
414#[cfg_attr(feature = "utoipa", derive(utoipa::ToSchema))]
415pub struct StreamReconfiguration {
416    /// Storage class for recent writes.
417    #[serde(default, skip_serializing_if = "Maybe::is_unspecified")]
418    #[cfg_attr(feature = "utoipa", schema(value_type = Option<StorageClass>))]
419    pub storage_class: Maybe<Option<StorageClass>>,
420    /// Retention policy for the stream.
421    /// If unspecified, the default is to retain records for 7 days.
422    #[serde(default, skip_serializing_if = "Maybe::is_unspecified")]
423    #[cfg_attr(feature = "utoipa", schema(value_type = Option<RetentionPolicy>))]
424    pub retention_policy: Maybe<Option<RetentionPolicy>>,
425    /// Timestamping behavior.
426    #[serde(default, skip_serializing_if = "Maybe::is_unspecified")]
427    #[cfg_attr(feature = "utoipa", schema(value_type = Option<TimestampingReconfiguration>))]
428    pub timestamping: Maybe<Option<TimestampingReconfiguration>>,
429    /// Delete-on-empty configuration.
430    #[serde(default, skip_serializing_if = "Maybe::is_unspecified")]
431    #[cfg_attr(feature = "utoipa", schema(value_type = Option<DeleteOnEmptyReconfiguration>))]
432    pub delete_on_empty: Maybe<Option<DeleteOnEmptyReconfiguration>>,
433}
434
435impl TryFrom<StreamReconfiguration> for s2_common::config::StreamReconfiguration {
436    type Error = s2_common::ValidationError;
437
438    fn try_from(value: StreamReconfiguration) -> Result<Self, Self::Error> {
439        let StreamReconfiguration {
440            storage_class,
441            retention_policy,
442            timestamping,
443            delete_on_empty,
444        } = value;
445
446        Ok(Self {
447            storage_class: storage_class.map_opt(Into::into),
448            retention_policy: retention_policy.try_map_opt(TryInto::try_into)?,
449            timestamping: timestamping.map_opt(Into::into),
450            delete_on_empty: delete_on_empty.map_opt(Into::into),
451        })
452    }
453}
454
455impl From<s2_common::config::StreamReconfiguration> for StreamReconfiguration {
456    fn from(value: s2_common::config::StreamReconfiguration) -> Self {
457        let s2_common::config::StreamReconfiguration {
458            storage_class,
459            retention_policy,
460            timestamping,
461            delete_on_empty,
462        } = value;
463
464        Self {
465            storage_class: storage_class.map_opt(Into::into),
466            retention_policy: retention_policy.map_opt(Into::into),
467            timestamping: timestamping.map_opt(Into::into),
468            delete_on_empty: delete_on_empty.map_opt(Into::into),
469        }
470    }
471}
472
473#[rustfmt::skip]
474#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
475#[cfg_attr(feature = "utoipa", derive(utoipa::ToSchema))]
476pub struct BasinConfig {
477    /// Default stream configuration.
478    pub default_stream_config: Option<StreamConfig>,
479    /// Encryption algorithm to apply to newly created streams in the basin.
480    pub stream_cipher: Option<EncryptionAlgorithm>,
481    /// Create stream on append if it doesn't exist, using the default stream configuration.
482    #[serde(default)]
483    #[cfg_attr(feature = "utoipa", schema(default = false))]
484    pub create_stream_on_append: bool,
485    /// Create stream on read if it doesn't exist, using the default stream configuration.
486    #[serde(default)]
487    #[cfg_attr(feature = "utoipa", schema(default = false))]
488    pub create_stream_on_read: bool,
489}
490
491impl TryFrom<BasinConfig> for s2_common::config::BasinConfig {
492    type Error = s2_common::ValidationError;
493
494    fn try_from(value: BasinConfig) -> Result<Self, Self::Error> {
495        let BasinConfig {
496            default_stream_config,
497            stream_cipher,
498            create_stream_on_append,
499            create_stream_on_read,
500        } = value;
501
502        let config = Self {
503            default_stream_config: match default_stream_config {
504                Some(config) => config.try_into()?,
505                None => Default::default(),
506            },
507            stream_cipher: stream_cipher.map(Into::into),
508            create_stream_on_append,
509            create_stream_on_read,
510        };
511        config.validate()?;
512        Ok(config)
513    }
514}
515
516impl From<s2_common::config::BasinConfig> for BasinConfig {
517    fn from(value: s2_common::config::BasinConfig) -> Self {
518        let s2_common::config::BasinConfig {
519            default_stream_config,
520            stream_cipher,
521            create_stream_on_append,
522            create_stream_on_read,
523        } = value;
524
525        Self {
526            default_stream_config: StreamConfig::to_opt(default_stream_config),
527            stream_cipher: stream_cipher.map(Into::into),
528            create_stream_on_append,
529            create_stream_on_read,
530        }
531    }
532}
533
534#[rustfmt::skip]
535#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
536#[cfg_attr(feature = "utoipa", derive(utoipa::ToSchema))]
537pub struct BasinReconfiguration {
538    /// Basin configuration.
539    #[serde(default, skip_serializing_if = "Maybe::is_unspecified")]
540    #[cfg_attr(feature = "utoipa", schema(value_type = Option<StreamReconfiguration>))]
541    pub default_stream_config: Maybe<Option<StreamReconfiguration>>,
542    /// Encryption algorithm to apply to newly created streams in the basin.
543    #[serde(default, skip_serializing_if = "Maybe::is_unspecified")]
544    #[cfg_attr(feature = "utoipa", schema(value_type = Option<EncryptionAlgorithm>))]
545    pub stream_cipher: Maybe<Option<EncryptionAlgorithm>>,
546    /// Create a stream on append.
547    #[serde(default, skip_serializing_if = "Maybe::is_unspecified")]
548    #[cfg_attr(feature = "utoipa", schema(value_type = Option<bool>))]
549    pub create_stream_on_append: Maybe<bool>,
550    /// Create a stream on read.
551    #[serde(default, skip_serializing_if = "Maybe::is_unspecified")]
552    #[cfg_attr(feature = "utoipa", schema(value_type = Option<bool>))]
553    pub create_stream_on_read: Maybe<bool>,
554}
555
556impl TryFrom<BasinReconfiguration> for s2_common::config::BasinReconfiguration {
557    type Error = s2_common::ValidationError;
558
559    fn try_from(value: BasinReconfiguration) -> Result<Self, Self::Error> {
560        let BasinReconfiguration {
561            default_stream_config,
562            stream_cipher,
563            create_stream_on_append,
564            create_stream_on_read,
565        } = value;
566
567        Ok(Self {
568            default_stream_config: default_stream_config.try_map_opt(TryInto::try_into)?,
569            stream_cipher: stream_cipher.map_opt(Into::into),
570            create_stream_on_append: create_stream_on_append.map(Into::into),
571            create_stream_on_read: create_stream_on_read.map(Into::into),
572        })
573    }
574}
575
576impl From<s2_common::config::BasinReconfiguration> for BasinReconfiguration {
577    fn from(value: s2_common::config::BasinReconfiguration) -> Self {
578        let s2_common::config::BasinReconfiguration {
579            default_stream_config,
580            stream_cipher,
581            create_stream_on_append,
582            create_stream_on_read,
583        } = value;
584
585        Self {
586            default_stream_config: default_stream_config.map_opt(Into::into),
587            stream_cipher: stream_cipher.map_opt(Into::into),
588            create_stream_on_append: create_stream_on_append.map(Into::into),
589            create_stream_on_read: create_stream_on_read.map(Into::into),
590        }
591    }
592}
593
594#[cfg(test)]
595mod tests {
596    use proptest::prelude::*;
597
598    use super::*;
599
600    fn gen_storage_class() -> impl Strategy<Value = StorageClass> {
601        prop_oneof![Just(StorageClass::Standard), Just(StorageClass::Express)]
602    }
603
604    fn gen_timestamping_mode() -> impl Strategy<Value = TimestampingMode> {
605        prop_oneof![
606            Just(TimestampingMode::ClientPrefer),
607            Just(TimestampingMode::ClientRequire),
608            Just(TimestampingMode::Arrival),
609        ]
610    }
611
612    fn gen_retention_policy() -> impl Strategy<Value = RetentionPolicy> {
613        prop_oneof![
614            any::<u64>().prop_map(RetentionPolicy::Age),
615            Just(RetentionPolicy::Infinite(InfiniteRetention {})),
616        ]
617    }
618
619    fn gen_timestamping_config() -> impl Strategy<Value = TimestampingConfig> {
620        (
621            proptest::option::of(gen_timestamping_mode()),
622            proptest::option::of(any::<bool>()),
623        )
624            .prop_map(|(mode, uncapped)| TimestampingConfig { mode, uncapped })
625    }
626
627    fn gen_delete_on_empty_config() -> impl Strategy<Value = DeleteOnEmptyConfig> {
628        any::<u64>().prop_map(|min_age_secs| DeleteOnEmptyConfig { min_age_secs })
629    }
630
631    fn gen_encryption_algorithm() -> impl Strategy<Value = EncryptionAlgorithm> {
632        prop_oneof![
633            Just(EncryptionAlgorithm::Aegis256),
634            Just(EncryptionAlgorithm::Aes256Gcm),
635        ]
636    }
637
638    fn gen_stream_config() -> impl Strategy<Value = StreamConfig> {
639        (
640            proptest::option::of(gen_storage_class()),
641            proptest::option::of(gen_retention_policy()),
642            proptest::option::of(gen_timestamping_config()),
643            proptest::option::of(gen_delete_on_empty_config()),
644        )
645            .prop_map(
646                |(storage_class, retention_policy, timestamping, delete_on_empty)| StreamConfig {
647                    storage_class,
648                    retention_policy,
649                    timestamping,
650                    delete_on_empty,
651                },
652            )
653    }
654
655    fn gen_basin_config() -> impl Strategy<Value = BasinConfig> {
656        (
657            proptest::option::of(gen_stream_config()),
658            proptest::option::of(gen_encryption_algorithm()),
659            any::<bool>(),
660            any::<bool>(),
661        )
662            .prop_map(
663                |(
664                    default_stream_config,
665                    stream_cipher,
666                    create_stream_on_append,
667                    create_stream_on_read,
668                )| {
669                    BasinConfig {
670                        default_stream_config,
671                        stream_cipher,
672                        create_stream_on_append,
673                        create_stream_on_read,
674                    }
675                },
676            )
677    }
678
679    fn gen_maybe<T: std::fmt::Debug + Clone + 'static>(
680        inner: impl Strategy<Value = T>,
681    ) -> impl Strategy<Value = Maybe<Option<T>>> {
682        prop_oneof![
683            Just(Maybe::Unspecified),
684            Just(Maybe::Specified(None)),
685            inner.prop_map(|v| Maybe::Specified(Some(v))),
686        ]
687    }
688
689    fn gen_stream_reconfiguration() -> impl Strategy<Value = StreamReconfiguration> {
690        (
691            gen_maybe(gen_storage_class()),
692            gen_maybe(gen_retention_policy()),
693            gen_maybe(gen_timestamping_reconfiguration()),
694            gen_maybe(gen_delete_on_empty_reconfiguration()),
695        )
696            .prop_map(
697                |(storage_class, retention_policy, timestamping, delete_on_empty)| {
698                    StreamReconfiguration {
699                        storage_class,
700                        retention_policy,
701                        timestamping,
702                        delete_on_empty,
703                    }
704                },
705            )
706    }
707
708    fn gen_timestamping_reconfiguration() -> impl Strategy<Value = TimestampingReconfiguration> {
709        (gen_maybe(gen_timestamping_mode()), gen_maybe(any::<bool>()))
710            .prop_map(|(mode, uncapped)| TimestampingReconfiguration { mode, uncapped })
711    }
712
713    fn gen_delete_on_empty_reconfiguration() -> impl Strategy<Value = DeleteOnEmptyReconfiguration>
714    {
715        gen_maybe(any::<u64>())
716            .prop_map(|min_age_secs| DeleteOnEmptyReconfiguration { min_age_secs })
717    }
718
719    fn gen_basin_reconfiguration() -> impl Strategy<Value = BasinReconfiguration> {
720        (
721            gen_maybe(gen_stream_reconfiguration()),
722            gen_maybe(gen_encryption_algorithm()),
723            prop_oneof![
724                Just(Maybe::Unspecified),
725                any::<bool>().prop_map(Maybe::Specified),
726            ],
727            prop_oneof![
728                Just(Maybe::Unspecified),
729                any::<bool>().prop_map(Maybe::Specified),
730            ],
731        )
732            .prop_map(
733                |(
734                    default_stream_config,
735                    stream_cipher,
736                    create_stream_on_append,
737                    create_stream_on_read,
738                )| BasinReconfiguration {
739                    default_stream_config,
740                    stream_cipher,
741                    create_stream_on_append,
742                    create_stream_on_read,
743                },
744            )
745    }
746
747    fn gen_internal_optional_stream_config()
748    -> impl Strategy<Value = s2_common::config::OptionalStreamConfig> {
749        (
750            proptest::option::of(gen_storage_class()),
751            proptest::option::of(gen_retention_policy()),
752            proptest::option::of(gen_timestamping_mode()),
753            proptest::option::of(any::<bool>()),
754            proptest::option::of(any::<u64>()),
755        )
756            .prop_map(|(sc, rp, ts_mode, ts_uncapped, doe)| {
757                s2_common::config::OptionalStreamConfig {
758                    storage_class: sc.map(Into::into),
759                    retention_policy: rp.map(|rp| match rp {
760                        RetentionPolicy::Age(secs) => {
761                            s2_common::config::RetentionPolicy::Age(Duration::from_secs(secs))
762                        }
763                        RetentionPolicy::Infinite(_) => {
764                            s2_common::config::RetentionPolicy::Infinite()
765                        }
766                    }),
767                    timestamping: s2_common::config::OptionalTimestampingConfig {
768                        mode: ts_mode.map(Into::into),
769                        uncapped: ts_uncapped,
770                    },
771                    delete_on_empty: s2_common::config::OptionalDeleteOnEmptyConfig {
772                        min_age: doe.map(Duration::from_secs),
773                    },
774                }
775            })
776    }
777
778    proptest! {
779        #[test]
780        fn stream_config_conversion_validates(config in gen_stream_config()) {
781            let has_zero_age = matches!(config.retention_policy, Some(RetentionPolicy::Age(0)));
782            let result: Result<s2_common::config::OptionalStreamConfig, _> = config.try_into();
783
784            if has_zero_age {
785                prop_assert!(result.is_err());
786            } else {
787                prop_assert!(result.is_ok());
788            }
789        }
790
791        #[test]
792        fn basin_config_conversion_validates(config in gen_basin_config()) {
793            let has_invalid_config = config.default_stream_config.as_ref().is_some_and(|sc| {
794                matches!(sc.retention_policy, Some(RetentionPolicy::Age(0)))
795            });
796
797            let result: Result<s2_common::config::BasinConfig, _> = config.try_into();
798
799            if has_invalid_config {
800                prop_assert!(result.is_err());
801            } else {
802                prop_assert!(result.is_ok());
803            }
804        }
805
806        #[test]
807        fn stream_reconfiguration_conversion_validates(reconfig in gen_stream_reconfiguration()) {
808            let has_zero_age = matches!(
809                reconfig.retention_policy,
810                Maybe::Specified(Some(RetentionPolicy::Age(0)))
811            );
812            let result: Result<s2_common::config::StreamReconfiguration, _> = reconfig.try_into();
813
814            if has_zero_age {
815                prop_assert!(result.is_err());
816            } else {
817                prop_assert!(result.is_ok());
818            }
819        }
820
821        #[test]
822        fn merge_stream_or_basin_or_default(
823            stream in gen_internal_optional_stream_config(),
824            basin in gen_internal_optional_stream_config(),
825        ) {
826            let merged = stream.clone().merge(basin.clone());
827
828            prop_assert_eq!(
829                merged.storage_class,
830                stream.storage_class.or(basin.storage_class).unwrap_or_default()
831            );
832            prop_assert_eq!(
833                merged.retention_policy,
834                stream.retention_policy.or(basin.retention_policy).unwrap_or_default()
835            );
836            prop_assert_eq!(
837                merged.timestamping.mode,
838                stream.timestamping.mode.or(basin.timestamping.mode).unwrap_or_default()
839            );
840            prop_assert_eq!(
841                merged.timestamping.uncapped,
842                stream.timestamping.uncapped.or(basin.timestamping.uncapped).unwrap_or_default()
843            );
844            prop_assert_eq!(
845                merged.delete_on_empty.min_age,
846                stream.delete_on_empty.min_age.or(basin.delete_on_empty.min_age).unwrap_or_default()
847            );
848        }
849
850        #[test]
851        fn reconfigure_unspecified_preserves_base(base in gen_internal_optional_stream_config()) {
852            let reconfig = s2_common::config::StreamReconfiguration::default();
853            let result = base.clone().reconfigure(reconfig);
854
855            prop_assert_eq!(result.storage_class, base.storage_class);
856            prop_assert_eq!(result.retention_policy, base.retention_policy);
857            prop_assert_eq!(result.timestamping.mode, base.timestamping.mode);
858            prop_assert_eq!(result.timestamping.uncapped, base.timestamping.uncapped);
859            prop_assert_eq!(result.delete_on_empty.min_age, base.delete_on_empty.min_age);
860        }
861
862        #[test]
863        fn reconfigure_specified_none_clears(base in gen_internal_optional_stream_config()) {
864            let reconfig = s2_common::config::StreamReconfiguration {
865                storage_class: Maybe::Specified(None),
866                retention_policy: Maybe::Specified(None),
867                timestamping: Maybe::Specified(None),
868                delete_on_empty: Maybe::Specified(None),
869            };
870            let result = base.reconfigure(reconfig);
871
872            prop_assert!(result.storage_class.is_none());
873            prop_assert!(result.retention_policy.is_none());
874            prop_assert!(result.timestamping.mode.is_none());
875            prop_assert!(result.timestamping.uncapped.is_none());
876            prop_assert!(result.delete_on_empty.min_age.is_none());
877        }
878
879        #[test]
880        fn reconfigure_specified_some_sets_value(
881            base in gen_internal_optional_stream_config(),
882            new_sc in gen_storage_class(),
883            new_rp_secs in 1u64..u64::MAX,
884        ) {
885            let reconfig = s2_common::config::StreamReconfiguration {
886                storage_class: Maybe::Specified(Some(new_sc.into())),
887                retention_policy: Maybe::Specified(Some(
888                    s2_common::config::RetentionPolicy::Age(Duration::from_secs(new_rp_secs))
889                )),
890                ..Default::default()
891            };
892            let result = base.reconfigure(reconfig);
893
894            prop_assert_eq!(result.storage_class, Some(new_sc.into()));
895            prop_assert_eq!(
896                result.retention_policy,
897                Some(s2_common::config::RetentionPolicy::Age(Duration::from_secs(new_rp_secs)))
898            );
899        }
900
901        #[test]
902        fn to_opt_returns_some_for_non_defaults(
903            sc in gen_storage_class(),
904            doe_secs in 1u64..u64::MAX,
905            ts_mode in gen_timestamping_mode(),
906        ) {
907            // non-default storage class -> Some
908            let internal = s2_common::config::OptionalStreamConfig {
909                storage_class: Some(sc.into()),
910                ..Default::default()
911            };
912            prop_assert!(StreamConfig::to_opt(internal).is_some());
913
914            // non-zero delete_on_empty -> Some
915            let internal = s2_common::config::OptionalDeleteOnEmptyConfig {
916                min_age: Some(Duration::from_secs(doe_secs)),
917            };
918            let api = DeleteOnEmptyConfig::to_opt(internal);
919            prop_assert!(api.is_some());
920            prop_assert_eq!(api.unwrap().min_age_secs, doe_secs);
921
922            // non-default timestamping -> Some
923            let internal = s2_common::config::OptionalTimestampingConfig {
924                mode: Some(ts_mode.into()),
925                uncapped: None,
926            };
927            prop_assert!(TimestampingConfig::to_opt(internal).is_some());
928        }
929
930        #[test]
931        fn basin_reconfiguration_conversion_validates(reconfig in gen_basin_reconfiguration()) {
932            let has_zero_age = matches!(
933                &reconfig.default_stream_config,
934                Maybe::Specified(Some(sr)) if matches!(
935                    sr.retention_policy,
936                    Maybe::Specified(Some(RetentionPolicy::Age(0)))
937                )
938            );
939            let result: Result<s2_common::config::BasinReconfiguration, _> = reconfig.try_into();
940
941            if has_zero_age {
942                prop_assert!(result.is_err());
943            } else {
944                prop_assert!(result.is_ok());
945            }
946        }
947
948        #[test]
949        fn reconfigure_basin_unspecified_preserves(
950            base_sc in proptest::option::of(gen_storage_class()),
951            base_algorithm in proptest::option::of(gen_encryption_algorithm()),
952            base_on_append in any::<bool>(),
953            base_on_read in any::<bool>(),
954        ) {
955            let base = s2_common::config::BasinConfig {
956                default_stream_config: s2_common::config::OptionalStreamConfig {
957                    storage_class: base_sc.map(Into::into),
958                    ..Default::default()
959                },
960                stream_cipher: base_algorithm.map(Into::into),
961                create_stream_on_append: base_on_append,
962                create_stream_on_read: base_on_read,
963            };
964
965            let reconfig = s2_common::config::BasinReconfiguration::default();
966            let result = base.clone().reconfigure(reconfig);
967
968            prop_assert_eq!(result.default_stream_config.storage_class, base.default_stream_config.storage_class);
969            prop_assert_eq!(result.stream_cipher, base.stream_cipher);
970            prop_assert_eq!(result.create_stream_on_append, base.create_stream_on_append);
971            prop_assert_eq!(result.create_stream_on_read, base.create_stream_on_read);
972        }
973
974        #[test]
975        fn reconfigure_basin_specified_updates(
976            base_on_append in any::<bool>(),
977            new_on_append in any::<bool>(),
978            new_sc in gen_storage_class(),
979            new_algorithm in gen_encryption_algorithm(),
980        ) {
981            let base = s2_common::config::BasinConfig {
982                create_stream_on_append: base_on_append,
983                ..Default::default()
984            };
985
986            let reconfig = s2_common::config::BasinReconfiguration {
987                default_stream_config: Maybe::Specified(Some(s2_common::config::StreamReconfiguration {
988                    storage_class: Maybe::Specified(Some(new_sc.into())),
989                    ..Default::default()
990                })),
991                stream_cipher: Maybe::Specified(Some(new_algorithm.into())),
992                create_stream_on_append: Maybe::Specified(new_on_append),
993                ..Default::default()
994            };
995            let result = base.reconfigure(reconfig);
996
997            prop_assert_eq!(result.default_stream_config.storage_class, Some(new_sc.into()));
998            prop_assert_eq!(result.stream_cipher, Some(new_algorithm.into()));
999            prop_assert_eq!(result.create_stream_on_append, new_on_append);
1000        }
1001
1002        #[test]
1003        fn reconfigure_nested_partial_update(
1004            base_mode in gen_timestamping_mode(),
1005            base_uncapped in any::<bool>(),
1006            new_mode in gen_timestamping_mode(),
1007        ) {
1008            let base = s2_common::config::OptionalStreamConfig {
1009                timestamping: s2_common::config::OptionalTimestampingConfig {
1010                    mode: Some(base_mode.into()),
1011                    uncapped: Some(base_uncapped),
1012                },
1013                ..Default::default()
1014            };
1015
1016            let expected_mode: s2_common::config::TimestampingMode = new_mode.into();
1017
1018            let reconfig = s2_common::config::StreamReconfiguration {
1019                timestamping: Maybe::Specified(Some(s2_common::config::TimestampingReconfiguration {
1020                    mode: Maybe::Specified(Some(expected_mode)),
1021                    uncapped: Maybe::Unspecified,
1022                })),
1023                ..Default::default()
1024            };
1025            let result = base.reconfigure(reconfig);
1026
1027            prop_assert_eq!(result.timestamping.mode, Some(expected_mode));
1028            prop_assert_eq!(result.timestamping.uncapped, Some(base_uncapped));
1029        }
1030    }
1031
1032    #[test]
1033    fn to_opt_returns_none_for_defaults() {
1034        // default stream config -> None
1035        assert!(StreamConfig::to_opt(s2_common::config::OptionalStreamConfig::default()).is_none());
1036
1037        // delete_on_empty: None -> None
1038        let doe_none = s2_common::config::OptionalDeleteOnEmptyConfig { min_age: None };
1039        assert!(DeleteOnEmptyConfig::to_opt(doe_none).is_none());
1040
1041        // default timestamping -> None
1042        assert!(
1043            TimestampingConfig::to_opt(s2_common::config::OptionalTimestampingConfig::default())
1044                .is_none()
1045        );
1046    }
1047
1048    #[test]
1049    fn optional_stream_config_to_opt_preserves_explicit_zero_delete_on_empty() {
1050        let api = StreamConfig::to_opt(s2_common::config::OptionalStreamConfig {
1051            delete_on_empty: s2_common::config::OptionalDeleteOnEmptyConfig {
1052                min_age: Some(Duration::ZERO),
1053            },
1054            ..Default::default()
1055        })
1056        .unwrap();
1057
1058        assert_eq!(
1059            api.delete_on_empty,
1060            Some(DeleteOnEmptyConfig { min_age_secs: 0 })
1061        );
1062    }
1063
1064    #[test]
1065    fn empty_json_converts_to_all_none() {
1066        let json = serde_json::json!({});
1067        let parsed: StreamConfig = serde_json::from_value(json).unwrap();
1068        let internal: s2_common::config::OptionalStreamConfig = parsed.try_into().unwrap();
1069
1070        assert!(
1071            internal.storage_class.is_none(),
1072            "storage_class should be None"
1073        );
1074        assert!(
1075            internal.retention_policy.is_none(),
1076            "retention_policy should be None"
1077        );
1078        assert!(
1079            internal.timestamping.mode.is_none(),
1080            "timestamping.mode should be None"
1081        );
1082        assert!(
1083            internal.timestamping.uncapped.is_none(),
1084            "timestamping.uncapped should be None"
1085        );
1086        assert!(
1087            internal.delete_on_empty.min_age.is_none(),
1088            "delete_on_empty.min_age should be None"
1089        );
1090    }
1091
1092    #[test]
1093    fn stream_config_header_parses_and_validates() {
1094        let header: StreamConfigHeader =
1095            r#"{"retention_policy":{"age":3600},"delete_on_empty":{"min_age_secs":300}}"#
1096                .parse()
1097                .unwrap();
1098        assert_eq!(
1099            header.0,
1100            s2_common::config::OptionalStreamConfig {
1101                retention_policy: Some(s2_common::config::RetentionPolicy::Age(
1102                    Duration::from_secs(3600)
1103                )),
1104                delete_on_empty: s2_common::config::OptionalDeleteOnEmptyConfig {
1105                    min_age: Some(Duration::from_secs(300)),
1106                },
1107                ..Default::default()
1108            }
1109        );
1110
1111        for spaced in [
1112            r#"{ "retention_policy": { "age": 3600 }, "delete_on_empty": { "min_age_secs": 300 } }"#,
1113            "{\t\"delete_on_empty\":\t{\"min_age_secs\":\t300},\t\"retention_policy\":\t{\"age\":\t3600}\t}",
1114            "  {\"retention_policy\":{\"age\":3600},\"delete_on_empty\":{\"min_age_secs\":300}}  ",
1115        ] {
1116            let parsed: StreamConfigHeader = spaced.parse().unwrap();
1117            assert_eq!(parsed, header, "{spaced:?}");
1118        }
1119
1120        let empty: StreamConfigHeader = "{}".parse().unwrap();
1121        assert_eq!(empty.0, Default::default());
1122
1123        let invalid_json = "not json".parse::<StreamConfigHeader>().unwrap_err();
1124        assert!(invalid_json.to_string().contains("invalid JSON"));
1125
1126        let invalid_age =
1127            r#"{"retention_policy":{"age":0}}"#.parse::<StreamConfigHeader>().unwrap_err();
1128        assert!(
1129            invalid_age
1130                .to_string()
1131                .contains("age must be greater than 0 seconds"),
1132            "{invalid_age}"
1133        );
1134    }
1135
1136    #[test]
1137    fn stream_config_header_value_roundtrips() {
1138        let config = StreamConfig {
1139            storage_class: Some(StorageClass::Express),
1140            retention_policy: Some(RetentionPolicy::Infinite(InfiniteRetention {})),
1141            timestamping: Some(TimestampingConfig {
1142                mode: Some(TimestampingMode::ClientRequire),
1143                uncapped: Some(true),
1144            }),
1145            delete_on_empty: Some(DeleteOnEmptyConfig { min_age_secs: 60 }),
1146        };
1147        let value = config.to_header_value();
1148        let parsed: StreamConfigHeader = value.to_str().unwrap().parse().unwrap();
1149        assert_eq!(
1150            parsed.0,
1151            s2_common::config::OptionalStreamConfig::try_from(config).unwrap()
1152        );
1153    }
1154}
s2_api/v1/config.rs

s2_api/v1/
config.rs
