s2_api/v1/
config.rs

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

s2_api/v1/
config.rs
