use std::{ops::Range, str::FromStr};
use bytes::{Buf, BufMut, Bytes, BytesMut};
use enum_ordinalize::Ordinalize;
use s2_common::{
bash::Bash,
caps::MIN_BASIN_NAME_LEN,
types::{
basin::{BasinName, BasinNamePrefix, BasinNameStartAfter},
config::BasinConfig,
},
};
use serde::{Deserialize, Serialize};
use time::OffsetDateTime;
use super::{
DeserializationError, KeyType, check_min_size, deser_json_value, increment_bytes,
invalid_value_err, ser_json_value,
};
#[derive(Debug, Clone)]
pub struct BasinMeta {
pub config: BasinConfig,
pub created_at: OffsetDateTime,
pub deleted_at: Option<OffsetDateTime>,
pub creation_idempotency_key: Option<Bash>,
}
#[derive(Debug, Serialize, Deserialize)]
struct BasinMetaSerde {
config: Option<s2_api::v1::config::BasinConfig>,
#[serde(with = "time::serde::rfc3339")]
created_at: OffsetDateTime,
#[serde(with = "time::serde::rfc3339::option")]
deleted_at: Option<OffsetDateTime>,
creation_idempotency_key: Option<Bash>,
}
impl From<BasinMeta> for BasinMetaSerde {
fn from(meta: BasinMeta) -> Self {
Self {
config: Some(meta.config.into()),
created_at: meta.created_at,
deleted_at: meta.deleted_at,
creation_idempotency_key: meta.creation_idempotency_key,
}
}
}
impl TryFrom<BasinMetaSerde> for BasinMeta {
type Error = s2_common::types::ValidationError;
fn try_from(serde: BasinMetaSerde) -> Result<Self, Self::Error> {
let config = match serde.config {
Some(api_config) => api_config.try_into()?,
None => BasinConfig::default(),
};
Ok(Self {
config,
created_at: serde.created_at,
deleted_at: serde.deleted_at,
creation_idempotency_key: serde.creation_idempotency_key,
})
}
}
pub fn ser_key_prefix(prefix: &BasinNamePrefix) -> Bytes {
ser_key_internal(prefix.as_bytes()).freeze()
}
pub fn ser_key_prefix_end(prefix: &BasinNamePrefix) -> Bytes {
increment_bytes(ser_key_internal(prefix.as_bytes())).expect("non-empty")
}
pub fn ser_key_start_after(start_after: &BasinNameStartAfter) -> Bytes {
let start_after_bytes = start_after.as_bytes();
let mut bytes = Vec::with_capacity(start_after_bytes.len() + 1);
bytes.extend_from_slice(start_after_bytes);
bytes.push(b'\0');
ser_key_internal(&bytes).freeze()
}
pub fn ser_key_range(prefix: &BasinNamePrefix, start_after: &BasinNameStartAfter) -> Range<Bytes> {
let prefix_start = ser_key_prefix(prefix);
let start = if !start_after.is_empty() {
let start_after_key = ser_key_start_after(start_after);
std::cmp::max(prefix_start, start_after_key)
} else {
prefix_start
};
let end = ser_key_prefix_end(prefix);
start..end
}
pub fn ser_key(basin: &BasinName) -> Bytes {
ser_key_internal(basin.as_bytes()).freeze()
}
fn ser_key_internal(basin: &[u8]) -> BytesMut {
let capacity = 1 + basin.len();
let mut buf = BytesMut::with_capacity(capacity);
buf.put_u8(KeyType::BasinMeta.ordinal());
buf.put_slice(basin);
debug_assert_eq!(buf.len(), capacity, "serialized length mismatch");
buf
}
pub fn deser_key(mut bytes: Bytes) -> Result<BasinName, DeserializationError> {
check_min_size(&bytes, 1 + MIN_BASIN_NAME_LEN)?;
let ordinal = bytes.get_u8();
if ordinal != KeyType::BasinMeta.ordinal() {
return Err(DeserializationError::InvalidOrdinal(ordinal));
}
let basin_str = std::str::from_utf8(&bytes).map_err(|e| invalid_value_err("basin", e))?;
BasinName::from_str(basin_str).map_err(|e| invalid_value_err("basin", e))
}
pub fn ser_value(basin_meta: &BasinMeta) -> Bytes {
ser_json_value::<BasinMeta, BasinMetaSerde>(basin_meta, "BasinMeta")
}
pub fn deser_value(bytes: Bytes) -> Result<BasinMeta, DeserializationError> {
deser_json_value::<BasinMeta, BasinMetaSerde>(bytes, "basin_meta")
}
#[cfg(test)]
mod tests {
use std::str::FromStr;
use bytes::Bytes;
use enum_ordinalize::Ordinalize;
use proptest::prelude::*;
use s2_common::{
bash::Bash,
types::{
basin::{BasinName, BasinNamePrefix, BasinNameStartAfter},
config::BasinConfig,
},
};
use time::OffsetDateTime;
use crate::backend::kv::{
DeserializationError, KeyType, proptest_strategies::basin_name_strategy,
};
fn basin(name: &str) -> BasinName {
BasinName::from_str(name).unwrap()
}
fn basin_prefix(prefix: &str) -> BasinNamePrefix {
BasinNamePrefix::from_str(prefix).unwrap()
}
fn basin_start_after(name: &str) -> BasinNameStartAfter {
BasinNameStartAfter::from_str(name).unwrap()
}
#[test]
fn basin_meta_ser_key_prefix() {
let prefix = basin_prefix("test-prefix");
let key = super::ser_key_prefix(&prefix);
assert_eq!(key[0], KeyType::BasinMeta.ordinal());
assert_eq!(&key[1..], b"test-prefix");
}
#[test]
fn basin_meta_ser_key_prefix_empty() {
let prefix = BasinNamePrefix::default();
let key = super::ser_key_prefix(&prefix);
assert_eq!(key.len(), 1);
assert_eq!(key[0], KeyType::BasinMeta.ordinal());
}
#[test]
fn basin_meta_ser_key_prefix_end_empty() {
let prefix = BasinNamePrefix::default();
let end_key = super::ser_key_prefix_end(&prefix);
assert_eq!(end_key.len(), 1);
assert_eq!(end_key[0], KeyType::BasinMeta.ordinal() + 1);
}
#[test]
fn basin_meta_ser_key_prefix_end_advances() {
for (input, expected_suffix) in [("test-a", &b"test-b"[..]), ("test-abc", &b"test-abd"[..])]
{
let end_key = super::ser_key_prefix_end(&basin_prefix(input));
assert_eq!(end_key[0], KeyType::BasinMeta.ordinal());
assert_eq!(&end_key[1..], expected_suffix);
}
}
#[test]
fn basin_meta_ser_key_start_after() {
let key = super::ser_key_start_after(&basin_start_after("my-basin"));
assert_eq!(key[0], KeyType::BasinMeta.ordinal());
assert_eq!(&key[1..key.len() - 1], b"my-basin");
assert_eq!(
key[key.len() - 1],
b'\0',
"should end with null byte for exclusion"
);
}
#[test]
fn basin_meta_key_range_handles_pagination() {
let prefix = basin_prefix("test-");
let basin1 = basin("test-aaa");
let basin2 = basin("test-bbb");
let basin3 = basin("test-ccc");
let outside = basin("staging-service1");
let page1 = super::ser_key_range(&prefix, &BasinNameStartAfter::default());
assert_eq!(page1.start, super::ser_key_prefix(&prefix));
assert_eq!(page1.end, super::ser_key_prefix_end(&prefix));
let key1 = super::ser_key(&basin1);
let key2 = super::ser_key(&basin2);
let key3 = super::ser_key(&basin3);
let key_outside = super::ser_key(&outside);
assert!(key1 >= page1.start && key1 < page1.end);
assert!(key2 >= page1.start && key2 < page1.end);
assert!(key3 >= page1.start && key3 < page1.end);
assert!(key_outside < page1.start || key_outside >= page1.end);
let start_after = BasinNameStartAfter::from(basin1.clone());
let cursor_key = super::ser_key_start_after(&start_after);
assert!(cursor_key > key1);
assert!(cursor_key < key2);
let page2 = super::ser_key_range(&prefix, &start_after);
assert_eq!(page2.start, super::ser_key_start_after(&start_after));
assert_eq!(page2.end, super::ser_key_prefix_end(&prefix));
assert!(key1 < page2.start);
assert!(key2 >= page2.start && key2 < page2.end);
assert!(key3 >= page2.start && key3 < page2.end);
}
#[test]
fn value_roundtrip_basin_meta() {
let config = BasinConfig {
create_stream_on_append: true,
..Default::default()
};
let created_at = OffsetDateTime::from_unix_timestamp(1234567890)
.unwrap()
.replace_nanosecond(123456789)
.unwrap();
let deleted_at = Some(
OffsetDateTime::from_unix_timestamp(1234567890)
.unwrap()
.replace_nanosecond(123456789)
.unwrap(),
);
let basin_meta = super::BasinMeta {
config: config.clone(),
created_at,
deleted_at,
creation_idempotency_key: Some(Bash::length_prefixed(&[
b"test-basin",
b"request-token-123",
])),
};
let bytes = super::ser_value(&basin_meta);
let decoded = super::deser_value(bytes).unwrap();
assert_eq!(
basin_meta.config.create_stream_on_append,
decoded.config.create_stream_on_append
);
assert_eq!(
basin_meta.config.create_stream_on_read,
decoded.config.create_stream_on_read
);
assert_eq!(basin_meta.created_at, decoded.created_at);
assert_eq!(basin_meta.deleted_at, decoded.deleted_at);
}
#[test]
fn basin_meta_deser_defaults_config_missing() {
let serde_value = super::BasinMetaSerde {
config: None,
created_at: OffsetDateTime::from_unix_timestamp(1_234_567).unwrap(),
deleted_at: None,
creation_idempotency_key: Some(Bash::length_prefixed(&[b"my-basin", b"req-789"])),
};
let bytes = Bytes::from(serde_json::to_vec(&serde_value).unwrap());
let decoded = super::deser_value(bytes).unwrap();
let default_config = BasinConfig::default();
assert_eq!(
decoded.config.create_stream_on_append,
default_config.create_stream_on_append
);
assert_eq!(
decoded.config.create_stream_on_read,
default_config.create_stream_on_read
);
assert_eq!(
decoded.config.default_stream_config.storage_class,
default_config.default_stream_config.storage_class
);
assert_eq!(
decoded.config.default_stream_config.retention_policy,
default_config.default_stream_config.retention_policy
);
assert_eq!(
decoded.config.default_stream_config.timestamping.mode,
default_config.default_stream_config.timestamping.mode
);
assert_eq!(
decoded.config.default_stream_config.timestamping.uncapped,
default_config.default_stream_config.timestamping.uncapped
);
assert_eq!(
decoded.config.default_stream_config.delete_on_empty.min_age,
default_config.default_stream_config.delete_on_empty.min_age
);
assert_eq!(decoded.created_at, serde_value.created_at);
assert_eq!(decoded.deleted_at, serde_value.deleted_at);
}
#[test]
fn basin_meta_deser_invalid_json() {
let err = super::deser_value(Bytes::from_static(b"{")).unwrap_err();
assert!(matches!(err, DeserializationError::JsonDeserialization(_)));
}
fn basin_name_prefix_strategy() -> impl Strategy<Value = BasinNamePrefix> {
prop_oneof![
Just(BasinNamePrefix::default()),
"[a-z][a-z0-9-]{0,46}".prop_map(|s| BasinNamePrefix::from_str(&s).unwrap()),
]
}
#[test]
fn basin_meta_range_start_after_before_prefix() {
let prefix = BasinNamePrefix::from_str("staging-").unwrap();
let start_after = BasinNameStartAfter::from_str("prod-api").unwrap();
let range = super::ser_key_range(&prefix, &start_after);
assert!(
range.start < range.end,
"range should be valid when start_after is before prefix range"
);
let staging_basin = BasinName::from_str("staging-api").unwrap();
let staging_key = super::ser_key(&staging_basin);
assert!(
staging_key >= range.start && staging_key < range.end,
"basins matching prefix should be in range"
);
let prod_basin = BasinName::from_str("prod-service").unwrap();
let prod_key = super::ser_key(&prod_basin);
assert!(
prod_key < range.start,
"basins before prefix should NOT be in range"
);
}
proptest! {
#[test]
fn roundtrip_basin_meta_key(basin in basin_name_strategy()) {
let bytes = super::ser_key(&basin);
let decoded = super::deser_key(bytes).unwrap();
prop_assert_eq!(basin.as_ref(), decoded.as_ref());
}
#[test]
fn basin_meta_range_contains_prefixed_keys(
prefix in basin_name_prefix_strategy(),
basin in basin_name_strategy(),
) {
let prefix_str = prefix.as_ref();
let basin_str = basin.as_ref();
let matches_prefix = prefix_str.is_empty() || basin_str.starts_with(prefix_str);
let range = super::ser_key_range(&prefix, &BasinNameStartAfter::default());
let key = super::ser_key(&basin);
if matches_prefix {
prop_assert!(key >= range.start, "key {:?} should be >= range.start {:?}", key, range.start);
prop_assert!(key < range.end, "key {:?} should be < range.end {:?}", key, range.end);
} else {
prop_assert!(key < range.start || key >= range.end);
}
}
#[test]
fn basin_meta_keys_preserve_ordering(
basin1 in basin_name_strategy(),
basin2 in basin_name_strategy(),
) {
let key1 = super::ser_key(&basin1);
let key2 = super::ser_key(&basin2);
let basin_cmp = basin1.as_ref().cmp(basin2.as_ref());
let key_cmp = key1.cmp(&key2);
prop_assert_eq!(basin_cmp, key_cmp, "ordering should be preserved");
}
#[test]
fn basin_meta_start_after_excludes_cursor(
prefix in basin_name_prefix_strategy(),
basin1 in basin_name_strategy(),
basin2 in basin_name_strategy(),
) {
if basin1.as_ref() >= basin2.as_ref() {
return Ok(());
}
let start_after = BasinNameStartAfter::from(basin1.clone());
let range = super::ser_key_range(&prefix, &start_after);
let key1 = super::ser_key(&basin1);
let key2 = super::ser_key(&basin2);
let prefix_str = prefix.as_ref();
let basin1_matches = prefix_str.is_empty() || basin1.as_ref().starts_with(prefix_str);
let basin2_matches = prefix_str.is_empty() || basin2.as_ref().starts_with(prefix_str);
prop_assert!(key1 < range.start, "cursor basin should be excluded (before range.start)");
if basin2_matches && (basin1_matches || basin2.as_ref() > prefix_str) {
prop_assert!(key2 >= range.start, "later basin matching prefix should be included (at or after range.start)");
}
}
}
}