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 TryFrom<StreamConfig> for s2_common::config::OptionalStreamConfig {
355    type Error = s2_common::ValidationError;
356
357    fn try_from(value: StreamConfig) -> Result<Self, Self::Error> {
358        let StreamConfig {
359            storage_class,
360            retention_policy,
361            timestamping,
362            delete_on_empty,
363        } = value;
364
365        let retention_policy = match retention_policy {
366            None => None,
367            Some(policy) => Some(policy.try_into()?),
368        };
369
370        let config = Self {
371            storage_class: storage_class.map(Into::into),
372            retention_policy,
373            timestamping: timestamping.map(Into::into).unwrap_or_default(),
374            delete_on_empty: delete_on_empty.map(Into::into).unwrap_or_default(),
375        };
376        config.validate()?;
377        Ok(config)
378    }
379}
380
381#[rustfmt::skip]
382#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
383#[cfg_attr(feature = "utoipa", derive(utoipa::ToSchema))]
384pub struct StreamReconfiguration {
385    /// Storage class for recent writes.
386    #[serde(default, skip_serializing_if = "Maybe::is_unspecified")]
387    #[cfg_attr(feature = "utoipa", schema(value_type = Option<StorageClass>))]
388    pub storage_class: Maybe<Option<StorageClass>>,
389    /// Retention policy for the stream.
390    /// If unspecified, the default is to retain records for 7 days.
391    #[serde(default, skip_serializing_if = "Maybe::is_unspecified")]
392    #[cfg_attr(feature = "utoipa", schema(value_type = Option<RetentionPolicy>))]
393    pub retention_policy: Maybe<Option<RetentionPolicy>>,
394    /// Timestamping behavior.
395    #[serde(default, skip_serializing_if = "Maybe::is_unspecified")]
396    #[cfg_attr(feature = "utoipa", schema(value_type = Option<TimestampingReconfiguration>))]
397    pub timestamping: Maybe<Option<TimestampingReconfiguration>>,
398    /// Delete-on-empty configuration.
399    #[serde(default, skip_serializing_if = "Maybe::is_unspecified")]
400    #[cfg_attr(feature = "utoipa", schema(value_type = Option<DeleteOnEmptyReconfiguration>))]
401    pub delete_on_empty: Maybe<Option<DeleteOnEmptyReconfiguration>>,
402}
403
404impl TryFrom<StreamReconfiguration> for s2_common::config::StreamReconfiguration {
405    type Error = s2_common::ValidationError;
406
407    fn try_from(value: StreamReconfiguration) -> Result<Self, Self::Error> {
408        let StreamReconfiguration {
409            storage_class,
410            retention_policy,
411            timestamping,
412            delete_on_empty,
413        } = value;
414
415        Ok(Self {
416            storage_class: storage_class.map_opt(Into::into),
417            retention_policy: retention_policy.try_map_opt(TryInto::try_into)?,
418            timestamping: timestamping.map_opt(Into::into),
419            delete_on_empty: delete_on_empty.map_opt(Into::into),
420        })
421    }
422}
423
424impl From<s2_common::config::StreamReconfiguration> for StreamReconfiguration {
425    fn from(value: s2_common::config::StreamReconfiguration) -> Self {
426        let s2_common::config::StreamReconfiguration {
427            storage_class,
428            retention_policy,
429            timestamping,
430            delete_on_empty,
431        } = value;
432
433        Self {
434            storage_class: storage_class.map_opt(Into::into),
435            retention_policy: retention_policy.map_opt(Into::into),
436            timestamping: timestamping.map_opt(Into::into),
437            delete_on_empty: delete_on_empty.map_opt(Into::into),
438        }
439    }
440}
441
442#[rustfmt::skip]
443#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
444#[cfg_attr(feature = "utoipa", derive(utoipa::ToSchema))]
445pub struct BasinConfig {
446    /// Default stream configuration.
447    pub default_stream_config: Option<StreamConfig>,
448    /// Encryption algorithm to apply to newly created streams in the basin.
449    pub stream_cipher: Option<EncryptionAlgorithm>,
450    /// Create stream on append if it doesn't exist, using the default stream configuration.
451    #[serde(default)]
452    #[cfg_attr(feature = "utoipa", schema(default = false))]
453    pub create_stream_on_append: bool,
454    /// Create stream on read if it doesn't exist, using the default stream configuration.
455    #[serde(default)]
456    #[cfg_attr(feature = "utoipa", schema(default = false))]
457    pub create_stream_on_read: bool,
458}
459
460impl TryFrom<BasinConfig> for s2_common::config::BasinConfig {
461    type Error = s2_common::ValidationError;
462
463    fn try_from(value: BasinConfig) -> Result<Self, Self::Error> {
464        let BasinConfig {
465            default_stream_config,
466            stream_cipher,
467            create_stream_on_append,
468            create_stream_on_read,
469        } = value;
470
471        let config = Self {
472            default_stream_config: match default_stream_config {
473                Some(config) => config.try_into()?,
474                None => Default::default(),
475            },
476            stream_cipher: stream_cipher.map(Into::into),
477            create_stream_on_append,
478            create_stream_on_read,
479        };
480        config.validate()?;
481        Ok(config)
482    }
483}
484
485impl From<s2_common::config::BasinConfig> for BasinConfig {
486    fn from(value: s2_common::config::BasinConfig) -> Self {
487        let s2_common::config::BasinConfig {
488            default_stream_config,
489            stream_cipher,
490            create_stream_on_append,
491            create_stream_on_read,
492        } = value;
493
494        Self {
495            default_stream_config: StreamConfig::to_opt(default_stream_config),
496            stream_cipher: stream_cipher.map(Into::into),
497            create_stream_on_append,
498            create_stream_on_read,
499        }
500    }
501}
502
503#[rustfmt::skip]
504#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
505#[cfg_attr(feature = "utoipa", derive(utoipa::ToSchema))]
506pub struct BasinReconfiguration {
507    /// Basin configuration.
508    #[serde(default, skip_serializing_if = "Maybe::is_unspecified")]
509    #[cfg_attr(feature = "utoipa", schema(value_type = Option<StreamReconfiguration>))]
510    pub default_stream_config: Maybe<Option<StreamReconfiguration>>,
511    /// Encryption algorithm to apply to newly created streams in the basin.
512    #[serde(default, skip_serializing_if = "Maybe::is_unspecified")]
513    #[cfg_attr(feature = "utoipa", schema(value_type = Option<EncryptionAlgorithm>))]
514    pub stream_cipher: Maybe<Option<EncryptionAlgorithm>>,
515    /// Create a stream on append.
516    #[serde(default, skip_serializing_if = "Maybe::is_unspecified")]
517    #[cfg_attr(feature = "utoipa", schema(value_type = Option<bool>))]
518    pub create_stream_on_append: Maybe<bool>,
519    /// Create a stream on read.
520    #[serde(default, skip_serializing_if = "Maybe::is_unspecified")]
521    #[cfg_attr(feature = "utoipa", schema(value_type = Option<bool>))]
522    pub create_stream_on_read: Maybe<bool>,
523}
524
525impl TryFrom<BasinReconfiguration> for s2_common::config::BasinReconfiguration {
526    type Error = s2_common::ValidationError;
527
528    fn try_from(value: BasinReconfiguration) -> Result<Self, Self::Error> {
529        let BasinReconfiguration {
530            default_stream_config,
531            stream_cipher,
532            create_stream_on_append,
533            create_stream_on_read,
534        } = value;
535
536        Ok(Self {
537            default_stream_config: default_stream_config.try_map_opt(TryInto::try_into)?,
538            stream_cipher: stream_cipher.map_opt(Into::into),
539            create_stream_on_append: create_stream_on_append.map(Into::into),
540            create_stream_on_read: create_stream_on_read.map(Into::into),
541        })
542    }
543}
544
545impl From<s2_common::config::BasinReconfiguration> for BasinReconfiguration {
546    fn from(value: s2_common::config::BasinReconfiguration) -> Self {
547        let s2_common::config::BasinReconfiguration {
548            default_stream_config,
549            stream_cipher,
550            create_stream_on_append,
551            create_stream_on_read,
552        } = value;
553
554        Self {
555            default_stream_config: default_stream_config.map_opt(Into::into),
556            stream_cipher: stream_cipher.map_opt(Into::into),
557            create_stream_on_append: create_stream_on_append.map(Into::into),
558            create_stream_on_read: create_stream_on_read.map(Into::into),
559        }
560    }
561}
562
563#[cfg(test)]
564mod tests {
565    use proptest::prelude::*;
566
567    use super::*;
568
569    fn gen_storage_class() -> impl Strategy<Value = StorageClass> {
570        prop_oneof![Just(StorageClass::Standard), Just(StorageClass::Express)]
571    }
572
573    fn gen_timestamping_mode() -> impl Strategy<Value = TimestampingMode> {
574        prop_oneof![
575            Just(TimestampingMode::ClientPrefer),
576            Just(TimestampingMode::ClientRequire),
577            Just(TimestampingMode::Arrival),
578        ]
579    }
580
581    fn gen_retention_policy() -> impl Strategy<Value = RetentionPolicy> {
582        prop_oneof![
583            any::<u64>().prop_map(RetentionPolicy::Age),
584            Just(RetentionPolicy::Infinite(InfiniteRetention {})),
585        ]
586    }
587
588    fn gen_timestamping_config() -> impl Strategy<Value = TimestampingConfig> {
589        (
590            proptest::option::of(gen_timestamping_mode()),
591            proptest::option::of(any::<bool>()),
592        )
593            .prop_map(|(mode, uncapped)| TimestampingConfig { mode, uncapped })
594    }
595
596    fn gen_delete_on_empty_config() -> impl Strategy<Value = DeleteOnEmptyConfig> {
597        any::<u64>().prop_map(|min_age_secs| DeleteOnEmptyConfig { min_age_secs })
598    }
599
600    fn gen_encryption_algorithm() -> impl Strategy<Value = EncryptionAlgorithm> {
601        prop_oneof![
602            Just(EncryptionAlgorithm::Aegis256),
603            Just(EncryptionAlgorithm::Aes256Gcm),
604        ]
605    }
606
607    fn gen_stream_config() -> impl Strategy<Value = StreamConfig> {
608        (
609            proptest::option::of(gen_storage_class()),
610            proptest::option::of(gen_retention_policy()),
611            proptest::option::of(gen_timestamping_config()),
612            proptest::option::of(gen_delete_on_empty_config()),
613        )
614            .prop_map(
615                |(storage_class, retention_policy, timestamping, delete_on_empty)| StreamConfig {
616                    storage_class,
617                    retention_policy,
618                    timestamping,
619                    delete_on_empty,
620                },
621            )
622    }
623
624    fn gen_basin_config() -> impl Strategy<Value = BasinConfig> {
625        (
626            proptest::option::of(gen_stream_config()),
627            proptest::option::of(gen_encryption_algorithm()),
628            any::<bool>(),
629            any::<bool>(),
630        )
631            .prop_map(
632                |(
633                    default_stream_config,
634                    stream_cipher,
635                    create_stream_on_append,
636                    create_stream_on_read,
637                )| {
638                    BasinConfig {
639                        default_stream_config,
640                        stream_cipher,
641                        create_stream_on_append,
642                        create_stream_on_read,
643                    }
644                },
645            )
646    }
647
648    fn gen_maybe<T: std::fmt::Debug + Clone + 'static>(
649        inner: impl Strategy<Value = T>,
650    ) -> impl Strategy<Value = Maybe<Option<T>>> {
651        prop_oneof![
652            Just(Maybe::Unspecified),
653            Just(Maybe::Specified(None)),
654            inner.prop_map(|v| Maybe::Specified(Some(v))),
655        ]
656    }
657
658    fn gen_stream_reconfiguration() -> impl Strategy<Value = StreamReconfiguration> {
659        (
660            gen_maybe(gen_storage_class()),
661            gen_maybe(gen_retention_policy()),
662            gen_maybe(gen_timestamping_reconfiguration()),
663            gen_maybe(gen_delete_on_empty_reconfiguration()),
664        )
665            .prop_map(
666                |(storage_class, retention_policy, timestamping, delete_on_empty)| {
667                    StreamReconfiguration {
668                        storage_class,
669                        retention_policy,
670                        timestamping,
671                        delete_on_empty,
672                    }
673                },
674            )
675    }
676
677    fn gen_timestamping_reconfiguration() -> impl Strategy<Value = TimestampingReconfiguration> {
678        (gen_maybe(gen_timestamping_mode()), gen_maybe(any::<bool>()))
679            .prop_map(|(mode, uncapped)| TimestampingReconfiguration { mode, uncapped })
680    }
681
682    fn gen_delete_on_empty_reconfiguration() -> impl Strategy<Value = DeleteOnEmptyReconfiguration>
683    {
684        gen_maybe(any::<u64>())
685            .prop_map(|min_age_secs| DeleteOnEmptyReconfiguration { min_age_secs })
686    }
687
688    fn gen_basin_reconfiguration() -> impl Strategy<Value = BasinReconfiguration> {
689        (
690            gen_maybe(gen_stream_reconfiguration()),
691            gen_maybe(gen_encryption_algorithm()),
692            prop_oneof![
693                Just(Maybe::Unspecified),
694                any::<bool>().prop_map(Maybe::Specified),
695            ],
696            prop_oneof![
697                Just(Maybe::Unspecified),
698                any::<bool>().prop_map(Maybe::Specified),
699            ],
700        )
701            .prop_map(
702                |(
703                    default_stream_config,
704                    stream_cipher,
705                    create_stream_on_append,
706                    create_stream_on_read,
707                )| BasinReconfiguration {
708                    default_stream_config,
709                    stream_cipher,
710                    create_stream_on_append,
711                    create_stream_on_read,
712                },
713            )
714    }
715
716    fn gen_internal_optional_stream_config()
717    -> impl Strategy<Value = s2_common::config::OptionalStreamConfig> {
718        (
719            proptest::option::of(gen_storage_class()),
720            proptest::option::of(gen_retention_policy()),
721            proptest::option::of(gen_timestamping_mode()),
722            proptest::option::of(any::<bool>()),
723            proptest::option::of(any::<u64>()),
724        )
725            .prop_map(|(sc, rp, ts_mode, ts_uncapped, doe)| {
726                s2_common::config::OptionalStreamConfig {
727                    storage_class: sc.map(Into::into),
728                    retention_policy: rp.map(|rp| match rp {
729                        RetentionPolicy::Age(secs) => {
730                            s2_common::config::RetentionPolicy::Age(Duration::from_secs(secs))
731                        }
732                        RetentionPolicy::Infinite(_) => {
733                            s2_common::config::RetentionPolicy::Infinite()
734                        }
735                    }),
736                    timestamping: s2_common::config::OptionalTimestampingConfig {
737                        mode: ts_mode.map(Into::into),
738                        uncapped: ts_uncapped,
739                    },
740                    delete_on_empty: s2_common::config::OptionalDeleteOnEmptyConfig {
741                        min_age: doe.map(Duration::from_secs),
742                    },
743                }
744            })
745    }
746
747    proptest! {
748        #[test]
749        fn stream_config_conversion_validates(config in gen_stream_config()) {
750            let has_zero_age = matches!(config.retention_policy, Some(RetentionPolicy::Age(0)));
751            let result: Result<s2_common::config::OptionalStreamConfig, _> = config.try_into();
752
753            if has_zero_age {
754                prop_assert!(result.is_err());
755            } else {
756                prop_assert!(result.is_ok());
757            }
758        }
759
760        #[test]
761        fn basin_config_conversion_validates(config in gen_basin_config()) {
762            let has_invalid_config = config.default_stream_config.as_ref().is_some_and(|sc| {
763                matches!(sc.retention_policy, Some(RetentionPolicy::Age(0)))
764            });
765
766            let result: Result<s2_common::config::BasinConfig, _> = config.try_into();
767
768            if has_invalid_config {
769                prop_assert!(result.is_err());
770            } else {
771                prop_assert!(result.is_ok());
772            }
773        }
774
775        #[test]
776        fn stream_reconfiguration_conversion_validates(reconfig in gen_stream_reconfiguration()) {
777            let has_zero_age = matches!(
778                reconfig.retention_policy,
779                Maybe::Specified(Some(RetentionPolicy::Age(0)))
780            );
781            let result: Result<s2_common::config::StreamReconfiguration, _> = reconfig.try_into();
782
783            if has_zero_age {
784                prop_assert!(result.is_err());
785            } else {
786                prop_assert!(result.is_ok());
787            }
788        }
789
790        #[test]
791        fn merge_stream_or_basin_or_default(
792            stream in gen_internal_optional_stream_config(),
793            basin in gen_internal_optional_stream_config(),
794        ) {
795            let merged = stream.clone().merge(basin.clone());
796
797            prop_assert_eq!(
798                merged.storage_class,
799                stream.storage_class.or(basin.storage_class).unwrap_or_default()
800            );
801            prop_assert_eq!(
802                merged.retention_policy,
803                stream.retention_policy.or(basin.retention_policy).unwrap_or_default()
804            );
805            prop_assert_eq!(
806                merged.timestamping.mode,
807                stream.timestamping.mode.or(basin.timestamping.mode).unwrap_or_default()
808            );
809            prop_assert_eq!(
810                merged.timestamping.uncapped,
811                stream.timestamping.uncapped.or(basin.timestamping.uncapped).unwrap_or_default()
812            );
813            prop_assert_eq!(
814                merged.delete_on_empty.min_age,
815                stream.delete_on_empty.min_age.or(basin.delete_on_empty.min_age).unwrap_or_default()
816            );
817        }
818
819        #[test]
820        fn reconfigure_unspecified_preserves_base(base in gen_internal_optional_stream_config()) {
821            let reconfig = s2_common::config::StreamReconfiguration::default();
822            let result = base.clone().reconfigure(reconfig);
823
824            prop_assert_eq!(result.storage_class, base.storage_class);
825            prop_assert_eq!(result.retention_policy, base.retention_policy);
826            prop_assert_eq!(result.timestamping.mode, base.timestamping.mode);
827            prop_assert_eq!(result.timestamping.uncapped, base.timestamping.uncapped);
828            prop_assert_eq!(result.delete_on_empty.min_age, base.delete_on_empty.min_age);
829        }
830
831        #[test]
832        fn reconfigure_specified_none_clears(base in gen_internal_optional_stream_config()) {
833            let reconfig = s2_common::config::StreamReconfiguration {
834                storage_class: Maybe::Specified(None),
835                retention_policy: Maybe::Specified(None),
836                timestamping: Maybe::Specified(None),
837                delete_on_empty: Maybe::Specified(None),
838            };
839            let result = base.reconfigure(reconfig);
840
841            prop_assert!(result.storage_class.is_none());
842            prop_assert!(result.retention_policy.is_none());
843            prop_assert!(result.timestamping.mode.is_none());
844            prop_assert!(result.timestamping.uncapped.is_none());
845            prop_assert!(result.delete_on_empty.min_age.is_none());
846        }
847
848        #[test]
849        fn reconfigure_specified_some_sets_value(
850            base in gen_internal_optional_stream_config(),
851            new_sc in gen_storage_class(),
852            new_rp_secs in 1u64..u64::MAX,
853        ) {
854            let reconfig = s2_common::config::StreamReconfiguration {
855                storage_class: Maybe::Specified(Some(new_sc.into())),
856                retention_policy: Maybe::Specified(Some(
857                    s2_common::config::RetentionPolicy::Age(Duration::from_secs(new_rp_secs))
858                )),
859                ..Default::default()
860            };
861            let result = base.reconfigure(reconfig);
862
863            prop_assert_eq!(result.storage_class, Some(new_sc.into()));
864            prop_assert_eq!(
865                result.retention_policy,
866                Some(s2_common::config::RetentionPolicy::Age(Duration::from_secs(new_rp_secs)))
867            );
868        }
869
870        #[test]
871        fn to_opt_returns_some_for_non_defaults(
872            sc in gen_storage_class(),
873            doe_secs in 1u64..u64::MAX,
874            ts_mode in gen_timestamping_mode(),
875        ) {
876            // non-default storage class -> Some
877            let internal = s2_common::config::OptionalStreamConfig {
878                storage_class: Some(sc.into()),
879                ..Default::default()
880            };
881            prop_assert!(StreamConfig::to_opt(internal).is_some());
882
883            // non-zero delete_on_empty -> Some
884            let internal = s2_common::config::OptionalDeleteOnEmptyConfig {
885                min_age: Some(Duration::from_secs(doe_secs)),
886            };
887            let api = DeleteOnEmptyConfig::to_opt(internal);
888            prop_assert!(api.is_some());
889            prop_assert_eq!(api.unwrap().min_age_secs, doe_secs);
890
891            // non-default timestamping -> Some
892            let internal = s2_common::config::OptionalTimestampingConfig {
893                mode: Some(ts_mode.into()),
894                uncapped: None,
895            };
896            prop_assert!(TimestampingConfig::to_opt(internal).is_some());
897        }
898
899        #[test]
900        fn basin_reconfiguration_conversion_validates(reconfig in gen_basin_reconfiguration()) {
901            let has_zero_age = matches!(
902                &reconfig.default_stream_config,
903                Maybe::Specified(Some(sr)) if matches!(
904                    sr.retention_policy,
905                    Maybe::Specified(Some(RetentionPolicy::Age(0)))
906                )
907            );
908            let result: Result<s2_common::config::BasinReconfiguration, _> = reconfig.try_into();
909
910            if has_zero_age {
911                prop_assert!(result.is_err());
912            } else {
913                prop_assert!(result.is_ok());
914            }
915        }
916
917        #[test]
918        fn reconfigure_basin_unspecified_preserves(
919            base_sc in proptest::option::of(gen_storage_class()),
920            base_algorithm in proptest::option::of(gen_encryption_algorithm()),
921            base_on_append in any::<bool>(),
922            base_on_read in any::<bool>(),
923        ) {
924            let base = s2_common::config::BasinConfig {
925                default_stream_config: s2_common::config::OptionalStreamConfig {
926                    storage_class: base_sc.map(Into::into),
927                    ..Default::default()
928                },
929                stream_cipher: base_algorithm.map(Into::into),
930                create_stream_on_append: base_on_append,
931                create_stream_on_read: base_on_read,
932            };
933
934            let reconfig = s2_common::config::BasinReconfiguration::default();
935            let result = base.clone().reconfigure(reconfig);
936
937            prop_assert_eq!(result.default_stream_config.storage_class, base.default_stream_config.storage_class);
938            prop_assert_eq!(result.stream_cipher, base.stream_cipher);
939            prop_assert_eq!(result.create_stream_on_append, base.create_stream_on_append);
940            prop_assert_eq!(result.create_stream_on_read, base.create_stream_on_read);
941        }
942
943        #[test]
944        fn reconfigure_basin_specified_updates(
945            base_on_append in any::<bool>(),
946            new_on_append in any::<bool>(),
947            new_sc in gen_storage_class(),
948            new_algorithm in gen_encryption_algorithm(),
949        ) {
950            let base = s2_common::config::BasinConfig {
951                create_stream_on_append: base_on_append,
952                ..Default::default()
953            };
954
955            let reconfig = s2_common::config::BasinReconfiguration {
956                default_stream_config: Maybe::Specified(Some(s2_common::config::StreamReconfiguration {
957                    storage_class: Maybe::Specified(Some(new_sc.into())),
958                    ..Default::default()
959                })),
960                stream_cipher: Maybe::Specified(Some(new_algorithm.into())),
961                create_stream_on_append: Maybe::Specified(new_on_append),
962                ..Default::default()
963            };
964            let result = base.reconfigure(reconfig);
965
966            prop_assert_eq!(result.default_stream_config.storage_class, Some(new_sc.into()));
967            prop_assert_eq!(result.stream_cipher, Some(new_algorithm.into()));
968            prop_assert_eq!(result.create_stream_on_append, new_on_append);
969        }
970
971        #[test]
972        fn reconfigure_nested_partial_update(
973            base_mode in gen_timestamping_mode(),
974            base_uncapped in any::<bool>(),
975            new_mode in gen_timestamping_mode(),
976        ) {
977            let base = s2_common::config::OptionalStreamConfig {
978                timestamping: s2_common::config::OptionalTimestampingConfig {
979                    mode: Some(base_mode.into()),
980                    uncapped: Some(base_uncapped),
981                },
982                ..Default::default()
983            };
984
985            let expected_mode: s2_common::config::TimestampingMode = new_mode.into();
986
987            let reconfig = s2_common::config::StreamReconfiguration {
988                timestamping: Maybe::Specified(Some(s2_common::config::TimestampingReconfiguration {
989                    mode: Maybe::Specified(Some(expected_mode)),
990                    uncapped: Maybe::Unspecified,
991                })),
992                ..Default::default()
993            };
994            let result = base.reconfigure(reconfig);
995
996            prop_assert_eq!(result.timestamping.mode, Some(expected_mode));
997            prop_assert_eq!(result.timestamping.uncapped, Some(base_uncapped));
998        }
999    }
1000
1001    #[test]
1002    fn to_opt_returns_none_for_defaults() {
1003        // default stream config -> None
1004        assert!(StreamConfig::to_opt(s2_common::config::OptionalStreamConfig::default()).is_none());
1005
1006        // delete_on_empty: None -> None
1007        let doe_none = s2_common::config::OptionalDeleteOnEmptyConfig { min_age: None };
1008        assert!(DeleteOnEmptyConfig::to_opt(doe_none).is_none());
1009
1010        // default timestamping -> None
1011        assert!(
1012            TimestampingConfig::to_opt(s2_common::config::OptionalTimestampingConfig::default())
1013                .is_none()
1014        );
1015    }
1016
1017    #[test]
1018    fn optional_stream_config_to_opt_preserves_explicit_zero_delete_on_empty() {
1019        let api = StreamConfig::to_opt(s2_common::config::OptionalStreamConfig {
1020            delete_on_empty: s2_common::config::OptionalDeleteOnEmptyConfig {
1021                min_age: Some(Duration::ZERO),
1022            },
1023            ..Default::default()
1024        })
1025        .unwrap();
1026
1027        assert_eq!(
1028            api.delete_on_empty,
1029            Some(DeleteOnEmptyConfig { min_age_secs: 0 })
1030        );
1031    }
1032
1033    #[test]
1034    fn empty_json_converts_to_all_none() {
1035        let json = serde_json::json!({});
1036        let parsed: StreamConfig = serde_json::from_value(json).unwrap();
1037        let internal: s2_common::config::OptionalStreamConfig = parsed.try_into().unwrap();
1038
1039        assert!(
1040            internal.storage_class.is_none(),
1041            "storage_class should be None"
1042        );
1043        assert!(
1044            internal.retention_policy.is_none(),
1045            "retention_policy should be None"
1046        );
1047        assert!(
1048            internal.timestamping.mode.is_none(),
1049            "timestamping.mode should be None"
1050        );
1051        assert!(
1052            internal.timestamping.uncapped.is_none(),
1053            "timestamping.uncapped should be None"
1054        );
1055        assert!(
1056            internal.delete_on_empty.min_age.is_none(),
1057            "delete_on_empty.min_age should be None"
1058        );
1059    }
1060}
s2_api/v1/config.rs

s2_api/v1/
config.rs
