use std::collections::BTreeMap;
use blake2::{digest::consts::U32, Blake2b, Digest};
use serde::de::DeserializeOwned;
use crate::error::CachekitError;
type Blake2b256 = Blake2b<U32>;
const INT_MIN: i128 = i64::MIN as i128;
const INT_MAX: i128 = u64::MAX as i128;
const F64_COLLAPSE_MIN: f64 = -9_223_372_036_854_775_808.0;
const F64_COLLAPSE_MAX: f64 = 18_446_744_073_709_551_616.0;
#[derive(Debug, Clone, PartialEq)]
pub enum InteropValue {
Null,
Bool(bool),
Int(i128),
Float(f64),
Str(String),
Bytes(Vec<u8>),
Array(Vec<InteropValue>),
Map(BTreeMap<String, InteropValue>),
Set(Vec<InteropValue>),
DateTime {
unix_micros: i64,
},
Uuid(uuid::Uuid),
}
impl InteropValue {
#[must_use]
pub fn bytes(bytes: impl Into<Vec<u8>>) -> Self {
Self::Bytes(bytes.into())
}
#[must_use]
pub fn datetime_from_unix_micros(unix_micros: i64) -> Self {
Self::DateTime { unix_micros }
}
}
impl From<bool> for InteropValue {
fn from(v: bool) -> Self {
Self::Bool(v)
}
}
impl From<i32> for InteropValue {
fn from(v: i32) -> Self {
Self::Int(i128::from(v))
}
}
impl From<i64> for InteropValue {
fn from(v: i64) -> Self {
Self::Int(i128::from(v))
}
}
impl From<u32> for InteropValue {
fn from(v: u32) -> Self {
Self::Int(i128::from(v))
}
}
impl From<u64> for InteropValue {
fn from(v: u64) -> Self {
Self::Int(i128::from(v))
}
}
impl From<i128> for InteropValue {
fn from(v: i128) -> Self {
Self::Int(v)
}
}
impl From<f64> for InteropValue {
fn from(v: f64) -> Self {
Self::Float(v)
}
}
impl From<&str> for InteropValue {
fn from(v: &str) -> Self {
Self::Str(v.to_owned())
}
}
impl From<String> for InteropValue {
fn from(v: String) -> Self {
Self::Str(v)
}
}
impl From<Vec<InteropValue>> for InteropValue {
fn from(v: Vec<InteropValue>) -> Self {
Self::Array(v)
}
}
impl From<BTreeMap<String, InteropValue>> for InteropValue {
fn from(v: BTreeMap<String, InteropValue>) -> Self {
Self::Map(v)
}
}
impl From<uuid::Uuid> for InteropValue {
fn from(v: uuid::Uuid) -> Self {
Self::Uuid(v)
}
}
impl TryFrom<std::time::SystemTime> for InteropValue {
type Error = CachekitError;
fn try_from(t: std::time::SystemTime) -> Result<Self, CachekitError> {
let out_of_range =
|| CachekitError::Serialization("datetime out of interop range".to_owned());
match t.duration_since(std::time::UNIX_EPOCH) {
Ok(after) => {
let micros = i64::try_from(after.as_micros()).map_err(|_| out_of_range())?;
Ok(Self::DateTime {
unix_micros: micros,
})
}
Err(before) => {
let micros_up = before.duration().as_nanos().div_ceil(1000);
let micros = i64::try_from(micros_up).map_err(|_| out_of_range())?;
Ok(Self::DateTime {
unix_micros: -micros,
})
}
}
}
}
fn validate_segment(kind: &str, segment: &str) -> Result<(), CachekitError> {
let bytes = segment.as_bytes();
let valid = matches!(bytes.first(), Some(b) if b.is_ascii_lowercase() || b.is_ascii_digit())
&& bytes.len() <= 64
&& bytes[1..].iter().all(|b| {
b.is_ascii_lowercase() || b.is_ascii_digit() || matches!(b, b'.' | b'_' | b'-')
});
if valid {
Ok(())
} else {
Err(CachekitError::InvalidKey(format!(
"interop {kind} {segment:?} must match ^[a-z0-9][a-z0-9._-]{{0,63}}$ \
(lowercase ASCII letters, digits, '.', '_', '-'; 1-64 chars)"
)))
}
}
pub fn interop_key(
namespace: &str,
operation: &str,
args: &[InteropValue],
) -> Result<String, CachekitError> {
validate_segment("namespace", namespace)?;
validate_segment("operation", operation)?;
let encoded = canonical_args(args)?;
let mut hasher = Blake2b256::new();
hasher.update(&encoded);
let args_hash = hex::encode(hasher.finalize());
Ok(format!("{namespace}:{operation}:{args_hash}"))
}
pub fn canonical_args(args: &[InteropValue]) -> Result<Vec<u8>, CachekitError> {
let mut buf = Vec::new();
encode_array_header(&mut buf, args.len())?;
for arg in args {
encode(&mut buf, arg, Profile::Args)?;
}
Ok(buf)
}
pub fn serialize_value(value: &InteropValue) -> Result<Vec<u8>, CachekitError> {
let mut buf = Vec::new();
encode(&mut buf, value, Profile::Value)?;
Ok(buf)
}
pub fn deserialize<T: DeserializeOwned>(bytes: &[u8]) -> Result<T, CachekitError> {
if bytes.starts_with(b"CK") {
return Err(CachekitError::Serialization(
"payload starts with 0x43 0x4B (\"CK\"): this is a Python-SDK-internal auto-mode \
entry (CK frame), not an interop-mode value — it cannot be read cross-SDK"
.to_owned(),
));
}
let mut remaining: &[u8] = bytes;
let mut de = rmp_serde::Deserializer::new(&mut remaining);
let value = T::deserialize(&mut de)
.map_err(|e| CachekitError::Serialization(format!("interop decode: {e}")))?;
if !remaining.is_empty() {
return Err(CachekitError::Serialization(format!(
"interop payload has {} trailing byte(s) after the MessagePack document — \
interop readers must consume exactly one document",
remaining.len()
)));
}
Ok(value)
}
#[derive(Clone, Copy)]
enum Profile {
Args,
Value,
}
fn encode(buf: &mut Vec<u8>, value: &InteropValue, profile: Profile) -> Result<(), CachekitError> {
match value {
InteropValue::Null => buf.push(0xc0),
InteropValue::Bool(false) => buf.push(0xc2),
InteropValue::Bool(true) => buf.push(0xc3),
InteropValue::Int(i) => encode_int(buf, *i)?,
InteropValue::Float(f) => encode_float(buf, *f, profile)?,
InteropValue::Str(s) => encode_str(buf, s)?,
InteropValue::Bytes(b) => encode_bin(buf, b)?,
InteropValue::Array(items) => {
encode_array_header(buf, items.len())?;
for item in items {
encode(buf, item, profile)?;
}
}
InteropValue::Map(map) => {
encode_map_header(buf, map.len())?;
for (key, val) in map {
encode_str(buf, key)?;
encode(buf, val, profile)?;
}
}
InteropValue::Set(items) => {
let mut encoded: Vec<Vec<u8>> = items
.iter()
.map(|item| {
let mut b = Vec::new();
encode(&mut b, item, profile)?;
Ok(b)
})
.collect::<Result<_, CachekitError>>()?;
encoded.sort();
encoded.dedup();
encode_array_header(buf, encoded.len())?;
for bytes in &encoded {
buf.extend_from_slice(bytes);
}
}
InteropValue::DateTime { unix_micros } => {
if matches!(profile, Profile::Value) {
return Err(CachekitError::Serialization(
"datetime interop VALUES use the wire-format sentinel map \
{\"__datetime__\": true, \"value\": \"<ISO-8601>\"} — build that map \
explicitly; the Unix-timestamp encoding is argument-hashing only"
.to_owned(),
));
}
#[allow(clippy::cast_precision_loss)] let ts = (*unix_micros as f64) / 1_000_000.0;
encode_float(buf, ts, profile)?;
}
InteropValue::Uuid(u) => {
encode_str(buf, &u.hyphenated().to_string())?;
}
}
Ok(())
}
fn encode_int(buf: &mut Vec<u8>, i: i128) -> Result<(), CachekitError> {
if !(INT_MIN..=INT_MAX).contains(&i) {
return Err(CachekitError::Serialization(format!(
"integer {i} is outside the interop range [-2^63, 2^64-1]"
)));
}
#[allow(clippy::cast_possible_truncation, clippy::cast_sign_loss)]
if i >= 0 {
encode_uint(buf, i as u64);
} else {
encode_negative_int(buf, i as i64);
}
Ok(())
}
#[allow(clippy::cast_possible_truncation)] fn encode_uint(buf: &mut Vec<u8>, n: u64) {
if n <= 0x7f {
buf.push(n as u8); } else if n <= 0xff {
buf.push(0xcc);
buf.push(n as u8);
} else if n <= 0xffff {
buf.push(0xcd);
buf.extend_from_slice(&(n as u16).to_be_bytes());
} else if n <= 0xffff_ffff {
buf.push(0xce);
buf.extend_from_slice(&(n as u32).to_be_bytes());
} else {
buf.push(0xcf);
buf.extend_from_slice(&n.to_be_bytes());
}
}
#[allow(clippy::cast_possible_truncation)] fn encode_negative_int(buf: &mut Vec<u8>, n: i64) {
debug_assert!(n < 0);
if n >= -32 {
buf.push((n as i8).to_be_bytes()[0]); } else if n >= i64::from(i8::MIN) {
buf.push(0xd0);
buf.push((n as i8).to_be_bytes()[0]);
} else if n >= i64::from(i16::MIN) {
buf.push(0xd1);
buf.extend_from_slice(&(n as i16).to_be_bytes());
} else if n >= i64::from(i32::MIN) {
buf.push(0xd2);
buf.extend_from_slice(&(n as i32).to_be_bytes());
} else {
buf.push(0xd3);
buf.extend_from_slice(&n.to_be_bytes());
}
}
fn encode_float(buf: &mut Vec<u8>, f: f64, profile: Profile) -> Result<(), CachekitError> {
if !f.is_finite() {
return Err(CachekitError::Serialization(format!(
"{f} is not allowed in interop values (NaN and infinities are rejected, \
never silently encoded)"
)));
}
if matches!(profile, Profile::Args)
&& f.trunc() == f
&& (F64_COLLAPSE_MIN..F64_COLLAPSE_MAX).contains(&f)
{
#[allow(clippy::cast_possible_truncation, clippy::cast_sign_loss)]
if f >= 0.0 {
encode_uint(buf, f as u64);
} else {
encode_negative_int(buf, f as i64);
}
} else {
buf.push(0xcb);
buf.extend_from_slice(&f.to_be_bytes());
}
Ok(())
}
fn encode_str(buf: &mut Vec<u8>, s: &str) -> Result<(), CachekitError> {
let len = checked_len(s.len())?;
#[allow(clippy::cast_possible_truncation)] if len <= 31 {
buf.push(0xa0 | (len as u8)); } else if len <= 0xff {
buf.push(0xd9);
buf.push(len as u8);
} else if len <= 0xffff {
buf.push(0xda);
buf.extend_from_slice(&(len as u16).to_be_bytes());
} else {
buf.push(0xdb);
buf.extend_from_slice(&len.to_be_bytes());
}
buf.extend_from_slice(s.as_bytes());
Ok(())
}
fn encode_bin(buf: &mut Vec<u8>, b: &[u8]) -> Result<(), CachekitError> {
let len = checked_len(b.len())?;
#[allow(clippy::cast_possible_truncation)] if len <= 0xff {
buf.push(0xc4);
buf.push(len as u8);
} else if len <= 0xffff {
buf.push(0xc5);
buf.extend_from_slice(&(len as u16).to_be_bytes());
} else {
buf.push(0xc6);
buf.extend_from_slice(&len.to_be_bytes());
}
buf.extend_from_slice(b);
Ok(())
}
fn encode_array_header(buf: &mut Vec<u8>, len: usize) -> Result<(), CachekitError> {
let len = checked_len(len)?;
#[allow(clippy::cast_possible_truncation)] if len <= 15 {
buf.push(0x90 | (len as u8)); } else if len <= 0xffff {
buf.push(0xdc);
buf.extend_from_slice(&(len as u16).to_be_bytes());
} else {
buf.push(0xdd);
buf.extend_from_slice(&len.to_be_bytes());
}
Ok(())
}
fn encode_map_header(buf: &mut Vec<u8>, len: usize) -> Result<(), CachekitError> {
let len = checked_len(len)?;
#[allow(clippy::cast_possible_truncation)] if len <= 15 {
buf.push(0x80 | (len as u8)); } else if len <= 0xffff {
buf.push(0xde);
buf.extend_from_slice(&(len as u16).to_be_bytes());
} else {
buf.push(0xdf);
buf.extend_from_slice(&len.to_be_bytes());
}
Ok(())
}
fn checked_len(len: usize) -> Result<u32, CachekitError> {
u32::try_from(len).map_err(|_| {
CachekitError::Serialization(format!("length {len} exceeds the MessagePack u32 maximum"))
})
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn deserialize_single_document() {
let value: i64 = deserialize(&[0x2a]).unwrap();
assert_eq!(value, 42);
}
#[test]
fn deserialize_rejects_trailing_bytes() {
let err = deserialize::<i64>(&[0x2a, 0x00]).unwrap_err();
assert!(
err.to_string().contains("trailing byte"),
"expected trailing-bytes error, got: {err}"
);
}
#[test]
fn deserialize_rejects_ck_frame_with_diagnostic() {
let ck_frame = b"CK\x03\x00\x00\x00\x02{}payload";
let err = deserialize::<i64>(ck_frame).unwrap_err();
let msg = err.to_string();
assert!(
msg.contains("Python-SDK-internal auto-mode entry"),
"expected the CK-frame diagnostic, got: {msg}"
);
}
#[test]
fn deserialize_matches_lenient_reader_on_clean_input() {
let bytes = rmp_serde::to_vec(&("hello", 7u8)).unwrap();
let strict: (String, u8) = deserialize(&bytes).unwrap();
let lenient: (String, u8) = rmp_serde::from_slice(&bytes).unwrap();
assert_eq!(strict, lenient);
}
#[test]
fn segment_rejects_trailing_newline() {
assert!(interop_key("users\n", "get_user", &[]).is_err());
}
#[test]
fn segment_rejects_empty_and_too_long() {
assert!(interop_key("", "op", &[]).is_err());
assert!(interop_key(&"a".repeat(65), "op", &[]).is_err());
assert!(interop_key(&"a".repeat(64), "op", &[]).is_ok());
}
#[test]
fn segment_rejects_leading_punctuation() {
assert!(interop_key(".users", "op", &[]).is_err());
assert!(interop_key("-users", "op", &[]).is_err());
assert!(interop_key("users.v2_x-y", "op", &[]).is_ok());
}
#[test]
fn systemtime_pre_epoch_floors_toward_negative_infinity() {
use std::time::{Duration, SystemTime, UNIX_EPOCH};
fn micros(t: SystemTime) -> Option<i64> {
match InteropValue::try_from(t) {
Ok(InteropValue::DateTime { unix_micros }) => Some(unix_micros),
_ => None,
}
}
assert_eq!(micros(UNIX_EPOCH - Duration::from_nanos(1500)), Some(-2));
assert_eq!(micros(UNIX_EPOCH - Duration::from_micros(3)), Some(-3));
}
#[test]
fn int_range_enforced_at_encode_time() {
assert!(canonical_args(&[InteropValue::Int(INT_MAX + 1)]).is_err());
assert!(canonical_args(&[InteropValue::Int(INT_MIN - 1)]).is_err());
assert!(canonical_args(&[InteropValue::Int(INT_MAX)]).is_ok());
assert!(canonical_args(&[InteropValue::Int(INT_MIN)]).is_ok());
}
#[test]
fn value_profile_rejects_datetime() {
let dt = InteropValue::datetime_from_unix_micros(1_704_067_200_000_000);
let err = serialize_value(&dt).unwrap_err();
assert!(err.to_string().contains("__datetime__"), "got: {err}");
assert!(canonical_args(&[dt]).is_ok());
}
#[test]
fn value_profile_preserves_floats() {
let bytes = serialize_value(&InteropValue::from(2.0f64)).unwrap();
assert_eq!(bytes, hex::decode("cb4000000000000000").unwrap());
let bytes = canonical_args(&[InteropValue::from(2.0f64)]).unwrap();
assert_eq!(bytes, hex::decode("9102").unwrap());
}
}