pub mod secure;
#[cfg(all(feature = "compression", not(target_arch = "wasm32")))]
pub mod zstd;
use crate::config::ConfigError;
use crate::domain::{DomainError, DomainResult};
use serde_json::{Value as JsonValue, json};
use std::collections::HashMap;
pub(crate) const DICT_SENTINEL: char = '\u{7F}';
#[derive(Debug, Clone)]
pub struct CompressionConfig {
pub min_array_length: usize,
pub min_string_length: usize,
pub min_frequency_count: u32,
pub uuid_compression_potential: f32,
pub min_net_savings: usize,
pub delta_threshold: f32,
pub min_delta_potential: f32,
pub run_length_threshold: f32,
pub min_compression_potential: f32,
pub min_numeric_sequence_size: usize,
}
impl Default for CompressionConfig {
fn default() -> Self {
Self {
min_array_length: 2,
min_string_length: 3,
min_frequency_count: 1,
uuid_compression_potential: 0.3,
min_net_savings: 10,
delta_threshold: 30.0,
min_delta_potential: 0.3,
run_length_threshold: 20.0,
min_compression_potential: 0.4,
min_numeric_sequence_size: 3,
}
}
}
impl CompressionConfig {
pub fn validate(&self) -> Result<(), ConfigError> {
for (value, message) in [
(
self.uuid_compression_potential,
"uuid_compression_potential must be in 0.0..=1.0",
),
(
self.min_delta_potential,
"min_delta_potential must be in 0.0..=1.0",
),
(
self.min_compression_potential,
"min_compression_potential must be in 0.0..=1.0",
),
] {
if !(0.0..=1.0).contains(&value) {
return Err(ConfigError::InconsistentBounds {
section: "compression",
message,
});
}
}
for (value, message) in [
(
self.delta_threshold,
"delta_threshold must be finite and non-negative",
),
(
self.run_length_threshold,
"run_length_threshold must be finite and non-negative",
),
] {
if !value.is_finite() || value < 0.0 {
return Err(ConfigError::InconsistentBounds {
section: "compression",
message,
});
}
}
Ok(())
}
}
#[derive(Debug, Clone, PartialEq)]
pub enum CompressionStrategy {
None,
Dictionary {
dictionary: HashMap<String, u16>,
},
Delta {
base_values: HashMap<String, f64>,
},
RunLength,
Hybrid {
string_dict: HashMap<String, u16>,
numeric_deltas: HashMap<String, f64>,
},
}
#[derive(Debug, Clone)]
pub struct SchemaAnalyzer {
patterns: HashMap<String, PatternInfo>,
numeric_fields: HashMap<String, NumericStats>,
string_repetitions: HashMap<String, u32>,
config: CompressionConfig,
}
#[derive(Debug, Clone)]
struct PatternInfo {
frequency: u32,
compression_potential: f32,
}
#[derive(Debug, Clone)]
struct NumericStats {
values: Vec<f64>,
delta_potential: f32,
base_value: f64,
}
impl SchemaAnalyzer {
pub fn new() -> Self {
Self {
patterns: HashMap::new(),
numeric_fields: HashMap::new(),
string_repetitions: HashMap::new(),
config: CompressionConfig::default(),
}
}
pub fn with_config(config: CompressionConfig) -> Self {
Self {
patterns: HashMap::new(),
numeric_fields: HashMap::new(),
string_repetitions: HashMap::new(),
config,
}
}
pub fn analyze(&mut self, data: &JsonValue) -> DomainResult<CompressionStrategy> {
self.patterns.clear();
self.numeric_fields.clear();
self.string_repetitions.clear();
self.analyze_recursive(data, "")?;
self.determine_strategy()
}
fn analyze_recursive(&mut self, value: &JsonValue, path: &str) -> DomainResult<()> {
match value {
JsonValue::Object(obj) => {
for (key, val) in obj {
let field_path = if path.is_empty() {
key.clone()
} else {
format!("{path}.{key}")
};
self.analyze_recursive(val, &field_path)?;
}
}
JsonValue::Array(arr) => {
if arr.len() > self.config.min_array_length {
self.analyze_array_patterns(arr, path)?;
}
for (idx, item) in arr.iter().enumerate() {
let item_path = format!("{path}[{idx}]");
self.analyze_recursive(item, &item_path)?;
}
}
JsonValue::String(s) => {
self.analyze_string_pattern(s, path);
}
JsonValue::Number(n) => {
if let Some(f) = n.as_f64() {
self.analyze_numeric_pattern(f, path);
}
}
_ => {}
}
Ok(())
}
fn analyze_array_patterns(&mut self, arr: &[JsonValue], path: &str) -> DomainResult<()> {
if let Some(JsonValue::Object(first)) = arr.first() {
let structure_key = format!("array_structure:{path}");
let field_names: Vec<&str> = first.keys().map(|k| k.as_str()).collect();
let pattern = field_names.join(",");
let matching_count = arr
.iter()
.filter_map(|v| v.as_object())
.filter(|obj| {
let obj_fields: Vec<&str> = obj.keys().map(|k| k.as_str()).collect();
obj_fields.join(",") == pattern
})
.count();
if matching_count > self.config.min_frequency_count as usize {
let info = PatternInfo {
frequency: matching_count as u32,
compression_potential: (matching_count as f32 - 1.0) / matching_count as f32,
};
self.patterns.insert(structure_key, info);
}
}
if arr.len() > 2 {
let mut value_counts = HashMap::new();
for value in arr {
let key = match value {
JsonValue::String(s) => format!("string:{s}"),
JsonValue::Number(n) => format!("number:{n}"),
JsonValue::Bool(b) => format!("bool:{b}"),
_ => continue,
};
*value_counts.entry(key).or_insert(0) += 1;
}
for (value_key, count) in value_counts {
if count > self.config.min_frequency_count {
let info = PatternInfo {
frequency: count,
compression_potential: (count as f32 - 1.0) / count as f32,
};
self.patterns
.insert(format!("array_value:{path}:{value_key}"), info);
}
}
}
Ok(())
}
fn analyze_string_pattern(&mut self, s: &str, _path: &str) {
*self.string_repetitions.entry(s.to_string()).or_insert(0) += 1;
if s.len() > 10 {
if s.starts_with("http://") || s.starts_with("https://") {
let prefix = if s.starts_with("https://") {
"https://"
} else {
"http://"
};
self.patterns
.entry(format!("url_prefix:{prefix}"))
.or_insert(PatternInfo {
frequency: 0,
compression_potential: 0.0,
})
.frequency += 1;
}
if s.len() == 36 && s.chars().filter(|&c| c == '-').count() == 4 {
self.patterns
.entry("uuid_pattern".to_string())
.or_insert(PatternInfo {
frequency: 0,
compression_potential: self.config.uuid_compression_potential,
})
.frequency += 1;
}
}
}
fn analyze_numeric_pattern(&mut self, value: f64, path: &str) {
self.numeric_fields
.entry(path.to_string())
.or_insert_with(|| NumericStats {
values: Vec::new(),
delta_potential: 0.0,
base_value: value,
})
.values
.push(value);
}
fn determine_strategy(&mut self) -> DomainResult<CompressionStrategy> {
let mut delta_score = 0.0;
let (string_dict, dict_net_savings) =
build_dictionary(&self.string_repetitions, &self.config);
let string_dict_selected =
!string_dict.is_empty() && dict_net_savings >= self.config.min_net_savings as i64;
let mut numeric_deltas = HashMap::new();
for (path, stats) in &mut self.numeric_fields {
if stats.values.len() > 2 {
stats
.values
.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));
let deltas: Vec<f64> = stats
.values
.windows(2)
.map(|window| window[1] - window[0])
.collect();
if !deltas.is_empty() {
let avg_delta = deltas.iter().sum::<f64>() / deltas.len() as f64;
let delta_variance =
deltas.iter().map(|d| (d - avg_delta).powi(2)).sum::<f64>()
/ deltas.len() as f64;
stats.delta_potential = 1.0 / (1.0 + delta_variance as f32);
if stats.delta_potential > self.config.min_delta_potential {
delta_score += stats.delta_potential * stats.values.len() as f32;
numeric_deltas.insert(path.clone(), stats.base_value);
}
}
}
}
match (
string_dict_selected,
delta_score >= self.config.delta_threshold,
) {
(true, true) => Ok(CompressionStrategy::Hybrid {
string_dict,
numeric_deltas,
}),
(true, false) => Ok(CompressionStrategy::Dictionary {
dictionary: string_dict,
}),
(false, true) => Ok(CompressionStrategy::Delta {
base_values: numeric_deltas,
}),
(false, false) => {
let run_length_score = self
.patterns
.values()
.filter(|p| p.compression_potential > self.config.min_compression_potential)
.map(|p| p.frequency as f32 * p.compression_potential)
.sum::<f32>();
if run_length_score >= self.config.run_length_threshold {
Ok(CompressionStrategy::RunLength)
} else {
Ok(CompressionStrategy::None)
}
}
}
}
}
fn decimal_digits(n: u16) -> usize {
n.to_string().len()
}
fn build_dictionary(
repetitions: &HashMap<String, u32>,
config: &CompressionConfig,
) -> (HashMap<String, u16>, i64) {
let mut candidates: Vec<(&String, u32)> = repetitions
.iter()
.filter_map(|(s, &count)| {
(count > config.min_frequency_count && s.len() > config.min_string_length)
.then_some((s, count))
})
.collect();
candidates.sort_by(|(s1, c1), (s2, c2)| {
let payoff1 = *c1 as usize * s1.len();
let payoff2 = *c2 as usize * s2.len();
payoff2.cmp(&payoff1).then_with(|| s1.cmp(s2))
});
let mut dictionary = HashMap::new();
let mut net: i64 = 0;
let mut index: u16 = 0;
for (s, count) in candidates {
if index == u16::MAX {
break;
}
let marker_len = 1 + decimal_digits(index);
let gain = count as i64 * (s.len() as i64 - marker_len as i64);
let cost = s.len() as i64 + 3;
if gain > cost {
net += gain - cost;
dictionary.insert(s.clone(), index);
index += 1;
}
}
if !dictionary.is_empty() {
net -= 10; }
(dictionary, net)
}
fn wire_size(data: &JsonValue, metadata: &HashMap<String, JsonValue>) -> DomainResult<usize> {
let mut size = serde_json::to_string(data)
.map_err(|e| DomainError::CompressionError(format!("JSON serialization failed: {e}")))?
.len();
if !metadata.is_empty() {
size += serde_json::to_string(metadata)
.map_err(|e| DomainError::CompressionError(format!("JSON serialization failed: {e}")))?
.len();
}
Ok(size)
}
fn dictionary_metadata(dictionary: &HashMap<String, u16>) -> JsonValue {
let mut ordered: Vec<Option<&str>> = vec![None; dictionary.len()];
for (s, &i) in dictionary {
if let Some(slot) = ordered.get_mut(i as usize) {
*slot = Some(s.as_str());
}
}
JsonValue::Array(
ordered
.into_iter()
.map(|s| JsonValue::String(s.unwrap_or_default().to_string()))
.collect(),
)
}
fn substitute_dictionary_strings(data: &JsonValue, dictionary: &HashMap<String, u16>) -> JsonValue {
match data {
JsonValue::Object(obj) => {
let mut out = serde_json::Map::with_capacity(obj.len());
for (key, value) in obj {
out.insert(
key.clone(),
substitute_dictionary_strings(value, dictionary),
);
}
JsonValue::Object(out)
}
JsonValue::Array(arr) => JsonValue::Array(
arr.iter()
.map(|v| substitute_dictionary_strings(v, dictionary))
.collect(),
),
JsonValue::String(s) => {
if let Some(&index) = dictionary.get(s) {
JsonValue::String(format!("{DICT_SENTINEL}{index}"))
} else if s.starts_with(DICT_SENTINEL) {
JsonValue::String(format!("{DICT_SENTINEL}{s}"))
} else {
data.clone()
}
}
_ => data.clone(),
}
}
#[derive(Debug, Clone)]
pub struct SchemaCompressor {
strategy: CompressionStrategy,
analyzer: SchemaAnalyzer,
config: CompressionConfig,
}
impl SchemaCompressor {
pub fn new() -> Self {
let config = CompressionConfig::default();
Self {
strategy: CompressionStrategy::None,
analyzer: SchemaAnalyzer::with_config(config.clone()),
config,
}
}
pub fn with_strategy(strategy: CompressionStrategy) -> Self {
let config = CompressionConfig::default();
Self {
strategy,
analyzer: SchemaAnalyzer::with_config(config.clone()),
config,
}
}
pub fn with_config(config: CompressionConfig) -> Self {
Self {
strategy: CompressionStrategy::None,
analyzer: SchemaAnalyzer::with_config(config.clone()),
config,
}
}
pub fn analyze_and_optimize(&mut self, data: &JsonValue) -> DomainResult<&CompressionStrategy> {
self.strategy = self.analyzer.analyze(data)?;
Ok(&self.strategy)
}
pub fn compress(&self, data: &JsonValue) -> DomainResult<CompressedData> {
match &self.strategy {
CompressionStrategy::None => {
let metadata = HashMap::new();
Ok(CompressedData {
strategy: self.strategy.clone(),
compressed_size: wire_size(data, &metadata)?,
data: data.clone(),
compression_metadata: metadata,
})
}
CompressionStrategy::Dictionary { dictionary } => {
self.compress_with_dictionary(data, dictionary)
}
CompressionStrategy::Delta { base_values } => {
self.compress_with_delta(data, base_values)
}
CompressionStrategy::RunLength => self.compress_with_run_length(data),
CompressionStrategy::Hybrid {
string_dict,
numeric_deltas,
} => self.compress_hybrid(data, string_dict, numeric_deltas),
}
}
fn compress_with_dictionary(
&self,
data: &JsonValue,
dictionary: &HashMap<String, u16>,
) -> DomainResult<CompressedData> {
let mut metadata = HashMap::new();
metadata.insert("dict".to_string(), dictionary_metadata(dictionary));
let compressed = substitute_dictionary_strings(data, dictionary);
let compressed_size = wire_size(&compressed, &metadata)?;
Ok(CompressedData {
strategy: self.strategy.clone(),
compressed_size,
data: compressed,
compression_metadata: metadata,
})
}
fn compress_with_delta(
&self,
data: &JsonValue,
base_values: &HashMap<String, f64>,
) -> DomainResult<CompressedData> {
let mut metadata = HashMap::new();
for (path, base) in base_values {
let number = serde_json::Number::from_f64(*base).ok_or_else(|| {
DomainError::CompressionError(format!(
"delta base value for path '{path}' is non-finite (NaN or Infinity); cannot compress"
))
})?;
metadata.insert(format!("base_{path}"), JsonValue::Number(number));
}
let compressed = self.apply_delta_compression(data, base_values)?;
let compressed_size = wire_size(&compressed, &metadata)?;
Ok(CompressedData {
strategy: self.strategy.clone(),
compressed_size,
data: compressed,
compression_metadata: metadata,
})
}
fn compress_with_run_length(&self, data: &JsonValue) -> DomainResult<CompressedData> {
let metadata = HashMap::new();
let compressed = self.apply_run_length_encoding(data)?;
let compressed_size = wire_size(&compressed, &metadata)?;
Ok(CompressedData {
strategy: self.strategy.clone(),
compressed_size,
data: compressed,
compression_metadata: metadata,
})
}
fn apply_run_length_encoding(&self, data: &JsonValue) -> DomainResult<JsonValue> {
match data {
JsonValue::Object(obj) => {
let mut compressed_obj = serde_json::Map::new();
for (key, value) in obj {
compressed_obj.insert(key.clone(), self.apply_run_length_encoding(value)?);
}
Ok(JsonValue::Object(compressed_obj))
}
JsonValue::Array(arr) if arr.len() > 2 => {
let mut compressed_runs = Vec::new();
let mut current_value = None;
let mut run_count = 0;
for item in arr {
if Some(item) == current_value.as_ref() {
run_count += 1;
} else {
if let Some(value) = current_value {
if run_count > self.config.min_frequency_count {
compressed_runs.push(json!({
"rle_value": value,
"rle_count": run_count
}));
} else {
compressed_runs.push(value);
}
}
current_value = Some(item.clone());
run_count = 1;
}
}
if let Some(value) = current_value {
if run_count > self.config.min_frequency_count {
compressed_runs.push(json!({
"rle_value": value,
"rle_count": run_count
}));
} else {
compressed_runs.push(value);
}
}
Ok(JsonValue::Array(compressed_runs))
}
JsonValue::Array(arr) => {
let compressed_arr: Result<Vec<_>, _> = arr
.iter()
.map(|item| self.apply_run_length_encoding(item))
.collect();
Ok(JsonValue::Array(compressed_arr?))
}
_ => Ok(data.clone()),
}
}
fn compress_hybrid(
&self,
data: &JsonValue,
string_dict: &HashMap<String, u16>,
numeric_deltas: &HashMap<String, f64>,
) -> DomainResult<CompressedData> {
let mut metadata = HashMap::new();
metadata.insert("dict".to_string(), dictionary_metadata(string_dict));
for (path, base) in numeric_deltas {
let number = serde_json::Number::from_f64(*base).ok_or_else(|| {
DomainError::CompressionError(format!(
"delta base value for path '{path}' is non-finite (NaN or Infinity); cannot compress"
))
})?;
metadata.insert(format!("base_{path}"), JsonValue::Number(number));
}
let dict_compressed = substitute_dictionary_strings(data, string_dict);
let final_compressed = self.apply_delta_compression(&dict_compressed, numeric_deltas)?;
let compressed_size = wire_size(&final_compressed, &metadata)?;
Ok(CompressedData {
strategy: self.strategy.clone(),
compressed_size,
data: final_compressed,
compression_metadata: metadata,
})
}
fn apply_delta_compression(
&self,
data: &JsonValue,
base_values: &HashMap<String, f64>,
) -> DomainResult<JsonValue> {
self.apply_delta_recursive(data, "", base_values)
}
fn apply_delta_recursive(
&self,
data: &JsonValue,
path: &str,
base_values: &HashMap<String, f64>,
) -> DomainResult<JsonValue> {
match data {
JsonValue::Object(obj) => {
let mut compressed_obj = serde_json::Map::new();
for (key, value) in obj {
let field_path = if path.is_empty() {
key.clone()
} else {
format!("{path}.{key}")
};
compressed_obj.insert(
key.clone(),
self.apply_delta_recursive(value, &field_path, base_values)?,
);
}
Ok(JsonValue::Object(compressed_obj))
}
JsonValue::Array(arr) if arr.len() > 2 => {
if self.is_numeric_sequence(arr) {
self.compress_numeric_array_with_delta(arr, path, base_values)
} else {
let compressed_arr: Result<Vec<_>, _> = arr
.iter()
.enumerate()
.map(|(idx, item)| {
let item_path = format!("{path}[{idx}]");
self.apply_delta_recursive(item, &item_path, base_values)
})
.collect();
Ok(JsonValue::Array(compressed_arr?))
}
}
JsonValue::Array(arr) => {
let compressed_arr: Result<Vec<_>, _> = arr
.iter()
.enumerate()
.map(|(idx, item)| {
let item_path = format!("{path}[{idx}]");
self.apply_delta_recursive(item, &item_path, base_values)
})
.collect();
Ok(JsonValue::Array(compressed_arr?))
}
_ => Ok(data.clone()),
}
}
fn is_numeric_sequence(&self, arr: &[JsonValue]) -> bool {
if arr.len() < self.config.min_numeric_sequence_size {
return false;
}
arr.iter().all(|v| v.is_number())
}
fn compress_numeric_array_with_delta(
&self,
arr: &[JsonValue],
path: &str,
base_values: &HashMap<String, f64>,
) -> DomainResult<JsonValue> {
let mut compressed_array = Vec::new();
let numbers: Vec<f64> = arr.iter().filter_map(|v| v.as_f64()).collect();
if numbers.is_empty() {
return Ok(JsonValue::Array(arr.to_vec()));
}
let base_value = base_values.get(path).copied().unwrap_or(numbers[0]);
compressed_array.push(json!({
"delta_base": base_value,
"delta_type": "numeric_sequence"
}));
let deltas: Vec<f64> = numbers.iter().map(|&num| num - base_value).collect();
let original_precision = numbers.iter().map(|n| format!("{n}").len()).sum::<usize>();
let delta_precision = deltas.iter().map(|d| format!("{d}").len()).sum::<usize>();
if delta_precision < original_precision {
compressed_array.extend(deltas.into_iter().map(JsonValue::from));
} else {
return Ok(JsonValue::Array(arr.to_vec()));
}
Ok(JsonValue::Array(compressed_array))
}
}
#[derive(Debug, Clone)]
pub struct CompressedData {
pub strategy: CompressionStrategy,
pub compressed_size: usize,
pub data: JsonValue,
pub compression_metadata: HashMap<String, JsonValue>,
}
impl CompressedData {
pub fn compression_ratio(&self, original_size: usize) -> f32 {
if original_size == 0 {
return 1.0;
}
self.compressed_size as f32 / original_size as f32
}
pub fn compression_savings(&self, original_size: usize) -> isize {
original_size as isize - self.compressed_size as isize
}
}
impl Default for SchemaAnalyzer {
fn default() -> Self {
Self::new()
}
}
impl Default for SchemaCompressor {
fn default() -> Self {
Self::new()
}
}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
#[test]
fn test_schema_analyzer_dictionary_potential() {
let mut analyzer = SchemaAnalyzer::new();
let data = json!({
"users": [
{"name": "John Doe", "role": "admin", "status": "active", "department": "engineering"},
{"name": "Jane Smith", "role": "admin", "status": "active", "department": "engineering"},
{"name": "Bob Wilson", "role": "admin", "status": "active", "department": "engineering"},
{"name": "Alice Brown", "role": "admin", "status": "active", "department": "engineering"},
{"name": "Charlie Davis", "role": "admin", "status": "active", "department": "engineering"},
{"name": "Diana Evans", "role": "admin", "status": "active", "department": "engineering"},
{"name": "Frank Miller", "role": "admin", "status": "active", "department": "engineering"},
{"name": "Grace Wilson", "role": "admin", "status": "active", "department": "engineering"}
]
});
let strategy = analyzer.analyze(&data).unwrap();
match strategy {
CompressionStrategy::Dictionary { .. } | CompressionStrategy::Hybrid { .. } => {
}
_ => panic!("Expected dictionary-based compression strategy"),
}
}
#[test]
fn test_schema_analyzer_realistic_ecommerce_payload() {
let mut analyzer = SchemaAnalyzer::new();
let data = json!({
"products": [
{"id": 1001, "name": "MacBook Pro", "category": "Electronics", "status": "available", "brand": "Apple", "price": 2399.99},
{"id": 1002, "name": "iPhone 15", "category": "Electronics", "status": "available", "brand": "Apple", "price": 999.99},
{"id": 1003, "name": "AirPods Pro", "category": "Electronics", "status": "available", "brand": "Apple", "price": 249.99}
],
"store": {"name": "Tech Store", "status": "operational", "location": "San Francisco"}
});
let strategy = analyzer.analyze(&data).unwrap();
match &strategy {
CompressionStrategy::Dictionary { .. } | CompressionStrategy::Hybrid { .. } => {}
other => panic!("Expected dictionary-based compression strategy, got {other:?}"),
}
let original_size = serde_json::to_string(&data).unwrap().len();
let compressed = SchemaCompressor::with_strategy(strategy)
.compress(&data)
.unwrap();
assert!(
compressed.compression_savings(original_size) > 0,
"expected genuine positive wire-byte savings, got {}",
compressed.compression_savings(original_size)
);
}
#[test]
fn test_schema_analyzer_realistic_api_response_payload() {
let mut analyzer = SchemaAnalyzer::new();
let data = json!({
"status": "success",
"data": {
"users": [
{"id": "user_001", "email": "alice@example.com", "status": "subscription_active", "role": "standard_user", "created_at": "2024-01-01T00:00:00Z", "last_login": "2024-01-15T10:30:00Z"},
{"id": "user_002", "email": "bob@example.com", "status": "subscription_active", "role": "standard_user", "created_at": "2024-01-02T00:00:00Z", "last_login": "2024-01-15T09:15:00Z"},
{"id": "user_003", "email": "charlie@example.com", "status": "subscription_active", "role": "standard_user", "created_at": "2024-01-03T00:00:00Z", "last_login": "2024-01-10T14:22:00Z"},
{"id": "user_004", "email": "dave@example.com", "status": "subscription_active", "role": "administrator", "created_at": "2024-01-04T00:00:00Z", "last_login": "2024-01-14T11:05:00Z"},
{"id": "user_005", "email": "erin@example.com", "status": "subscription_inactive", "role": "standard_user", "created_at": "2024-01-05T00:00:00Z", "last_login": "2024-01-09T08:40:00Z"}
]
},
"pagination": {"page": 1, "per_page": 25, "total_pages": 4, "total_items": 89},
"meta": {"request_id": "req_12345", "timestamp": "2024-01-15T10:30:15Z", "version": "v1.2.3"}
});
let strategy = analyzer.analyze(&data).unwrap();
match &strategy {
CompressionStrategy::Dictionary { .. } | CompressionStrategy::Hybrid { .. } => {}
other => panic!("Expected dictionary-based compression strategy, got {other:?}"),
}
let original_size = serde_json::to_string(&data).unwrap().len();
let compressed = SchemaCompressor::with_strategy(strategy)
.compress(&data)
.unwrap();
assert!(
compressed.compression_savings(original_size) > 0,
"expected genuine positive wire-byte savings, got {}",
compressed.compression_savings(original_size)
);
}
#[test]
fn test_schema_analyzer_no_repetition_stays_none() {
let mut analyzer = SchemaAnalyzer::new();
let data = json!({
"id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
"name": "Unique Product Name Alpha",
"description": "A completely unique description of this particular item with no repeats",
"vendor": "Acme Corporation International",
"location": "Building 12, Warehouse Section D",
"notes": "Handled with care during transit process"
});
let strategy = analyzer.analyze(&data).unwrap();
assert_eq!(strategy, CompressionStrategy::None);
}
#[test]
fn test_schema_analyzer_tiny_duplicate_stays_none_below_savings_floor() {
let mut analyzer = SchemaAnalyzer::new();
let data = json!({"a": "hello", "b": "hello", "c": "world"});
let strategy = analyzer.analyze(&data).unwrap();
assert_eq!(strategy, CompressionStrategy::None);
}
#[test]
fn test_schema_analyzer_long_repeated_string_selects_dictionary_and_shrinks() {
let mut analyzer = SchemaAnalyzer::new();
let data = json!({
"a": "premium_subscription",
"b": "premium_subscription",
"c": "premium_subscription",
"d": "unique"
});
let strategy = analyzer.analyze(&data).unwrap();
let dictionary = match &strategy {
CompressionStrategy::Dictionary { dictionary } => dictionary,
other => panic!("Expected Dictionary strategy, got {other:?}"),
};
let original_size = serde_json::to_string(&data).unwrap().len();
let compressed = SchemaCompressor::with_strategy(CompressionStrategy::Dictionary {
dictionary: dictionary.clone(),
})
.compress(&data)
.unwrap();
assert!(compressed.compression_savings(original_size) > 0);
}
#[test]
fn test_schema_compressor_basic() {
let compressor = SchemaCompressor::new();
let data = json!({
"message": "hello world",
"count": 42
});
let original_size = serde_json::to_string(&data).unwrap().len();
let compressed = compressor.compress(&data).unwrap();
assert!(compressed.compressed_size > 0);
assert!(compressed.compression_ratio(original_size) <= 1.0);
}
#[test]
fn test_dictionary_compression() {
let mut dictionary = HashMap::new();
dictionary.insert("active".to_string(), 0);
dictionary.insert("admin".to_string(), 1);
let compressor =
SchemaCompressor::with_strategy(CompressionStrategy::Dictionary { dictionary });
let data = json!({
"status": "active",
"role": "admin",
"description": "active admin user"
});
let result = compressor.compress(&data).unwrap();
assert_eq!(
result.compression_metadata.get("dict"),
Some(&json!(["active", "admin"]))
);
}
#[test]
fn test_dictionary_compression_never_produces_numbers_from_substitution() {
let mut dictionary = HashMap::new();
dictionary.insert("active".to_string(), 0);
let compressor =
SchemaCompressor::with_strategy(CompressionStrategy::Dictionary { dictionary });
let data = json!({
"status": "active",
"count": 0
});
let result = compressor.compress(&data).unwrap();
assert_eq!(result.data, json!({"status": "\u{7F}0", "count": 0}));
}
#[test]
fn test_dictionary_sentinel_escaping_encode_shape() {
let mut dictionary = HashMap::new();
dictionary.insert("greeting".to_string(), 0);
let data = json!({
"a": "\u{7F}foo",
"b": "\u{7F}\u{7F}bar",
"c": "\u{7F}0",
"d": "greeting"
});
let substituted = substitute_dictionary_strings(&data, &dictionary);
assert_eq!(
substituted,
json!({
"a": "\u{7F}\u{7F}foo",
"b": "\u{7F}\u{7F}\u{7F}bar",
"c": "\u{7F}\u{7F}0",
"d": "\u{7F}0"
})
);
}
#[test]
fn test_compressed_size_matches_wire_bytes_for_every_strategy() {
fn expected_wire_size(data: &JsonValue, metadata: &HashMap<String, JsonValue>) -> usize {
let mut size = serde_json::to_string(data).unwrap().len();
if !metadata.is_empty() {
size += serde_json::to_string(metadata).unwrap().len();
}
size
}
let data = json!({
"status": "active",
"count": 3,
"sequence": [1.0, 2.0, 3.0],
"repeated": [1, 1, 1, 2, 2]
});
let mut dictionary = HashMap::new();
dictionary.insert("active".to_string(), 0);
let mut base_values = HashMap::new();
base_values.insert("sequence".to_string(), 1.0);
for strategy in [
CompressionStrategy::None,
CompressionStrategy::Dictionary {
dictionary: dictionary.clone(),
},
CompressionStrategy::Delta {
base_values: base_values.clone(),
},
CompressionStrategy::RunLength,
CompressionStrategy::Hybrid {
string_dict: dictionary.clone(),
numeric_deltas: base_values.clone(),
},
] {
let compressor = SchemaCompressor::with_strategy(strategy);
let result = compressor.compress(&data).unwrap();
assert_eq!(
result.compressed_size,
expected_wire_size(&result.data, &result.compression_metadata),
"strategy {:?} mismatched wire size",
result.strategy
);
}
}
#[test]
fn test_build_dictionary_caps_index_at_u16_max_without_overflow() {
let mut repetitions = HashMap::new();
for i in 0..(u16::MAX as u32 + 2) {
repetitions.insert(format!("padding_string_{i:05}"), 2);
}
let (dictionary, _net) = build_dictionary(&repetitions, &CompressionConfig::default());
assert!(
dictionary.len() <= u16::MAX as usize,
"dictionary must never exceed the u16 index space, got {} entries",
dictionary.len()
);
let distinct_indices: std::collections::HashSet<u16> =
dictionary.values().copied().collect();
assert_eq!(
distinct_indices.len(),
dictionary.len(),
"every dictionary entry must have a unique index — a mismatch here means indices \
wrapped and collided"
);
}
#[test]
fn test_compression_strategy_selection() {
let mut analyzer = SchemaAnalyzer::new();
let simple_data = json!({
"unique_field_1": "unique_value_1",
"unique_field_2": "unique_value_2"
});
let strategy = analyzer.analyze(&simple_data).unwrap();
assert_eq!(strategy, CompressionStrategy::None);
}
#[test]
fn test_numeric_delta_analysis() {
let mut analyzer = SchemaAnalyzer::new();
let data = json!({
"measurements": [
{"time": 100, "value": 10.0},
{"time": 101, "value": 10.5},
{"time": 102, "value": 11.0},
{"time": 103, "value": 11.5}
]
});
let _strategy = analyzer.analyze(&data).unwrap();
assert!(!analyzer.numeric_fields.is_empty());
}
#[test]
fn test_run_length_encoding() {
let compressor = SchemaCompressor::with_strategy(CompressionStrategy::RunLength);
let data = json!({
"repeated_values": [1, 1, 1, 2, 2, 3, 3, 3, 3]
});
let result = compressor.compress(&data).unwrap();
assert!(result.compressed_size > 0);
let compressed_array = &result.data["repeated_values"];
assert!(compressed_array.is_array());
let array = compressed_array.as_array().unwrap();
let has_rle = array.iter().any(|v| v.get("rle_value").is_some());
assert!(has_rle);
}
#[test]
fn test_delta_compression() {
let mut base_values = HashMap::new();
base_values.insert("sequence".to_string(), 100.0);
let compressor =
SchemaCompressor::with_strategy(CompressionStrategy::Delta { base_values });
let data = json!({
"sequence": [100.0, 101.0, 102.0, 103.0, 104.0]
});
let result = compressor.compress(&data).unwrap();
assert!(result.compressed_size > 0);
let compressed_array = &result.data["sequence"];
assert!(compressed_array.is_array());
let array = compressed_array.as_array().unwrap();
let has_delta_base = array.iter().any(|v| v.get("delta_base").is_some());
assert!(has_delta_base);
}
#[test]
fn test_delta_compression_rejects_nan_base() {
let mut base_values = HashMap::new();
base_values.insert("sequence".to_string(), f64::NAN);
let compressor =
SchemaCompressor::with_strategy(CompressionStrategy::Delta { base_values });
let data = json!({ "sequence": [1.0, 2.0, 3.0] });
let err = compressor
.compress(&data)
.expect_err("expected error for NaN base");
match err {
DomainError::CompressionError(msg) => {
assert!(msg.contains("non-finite"), "unexpected message: {msg}");
assert!(msg.contains("sequence"), "expected path in message: {msg}");
}
other => panic!("expected CompressionError, got {other:?}"),
}
}
#[test]
fn test_delta_compression_rejects_infinity_base() {
let mut base_values = HashMap::new();
base_values.insert("sequence".to_string(), f64::INFINITY);
let compressor =
SchemaCompressor::with_strategy(CompressionStrategy::Delta { base_values });
let data = json!({ "sequence": [1.0, 2.0, 3.0] });
let err = compressor
.compress(&data)
.expect_err("expected error for Infinity base");
assert!(matches!(err, DomainError::CompressionError(_)));
}
#[test]
fn test_hybrid_compression_rejects_nan_base() {
let string_dict = HashMap::new();
let mut numeric_deltas = HashMap::new();
numeric_deltas.insert("sequence".to_string(), f64::NEG_INFINITY);
let compressor = SchemaCompressor::with_strategy(CompressionStrategy::Hybrid {
string_dict,
numeric_deltas,
});
let data = json!({ "sequence": [1.0, 2.0, 3.0] });
let err = compressor
.compress(&data)
.expect_err("expected error for non-finite base");
match err {
DomainError::CompressionError(msg) => {
assert!(msg.contains("non-finite"), "unexpected message: {msg}");
}
other => panic!("expected CompressionError, got {other:?}"),
}
}
}