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

s2_api/v1/
config.rs
