1pub mod secure;
7
8#[cfg(all(feature = "compression", not(target_arch = "wasm32")))]
9pub mod zstd;
10
11use crate::config::ConfigError;
12use crate::domain::{DomainError, DomainResult};
13use serde_json::{Value as JsonValue, json};
14use std::collections::HashMap;
15
16pub(crate) const DICT_SENTINEL: char = '\u{7F}';
24
25#[derive(Debug, Clone)]
27pub struct CompressionConfig {
28 pub min_array_length: usize,
30 pub min_string_length: usize,
32 pub min_frequency_count: u32,
34 pub uuid_compression_potential: f32,
36 pub min_net_savings: usize,
73 pub delta_threshold: f32,
75 pub min_delta_potential: f32,
77 pub run_length_threshold: f32,
79 pub min_compression_potential: f32,
81 pub min_numeric_sequence_size: usize,
83}
84
85impl Default for CompressionConfig {
86 fn default() -> Self {
87 Self {
88 min_array_length: 2,
89 min_string_length: 3,
90 min_frequency_count: 1,
91 uuid_compression_potential: 0.3,
92 min_net_savings: 10,
93 delta_threshold: 30.0,
94 min_delta_potential: 0.3,
95 run_length_threshold: 20.0,
96 min_compression_potential: 0.4,
97 min_numeric_sequence_size: 3,
98 }
99 }
100}
101
102impl CompressionConfig {
103 pub fn validate(&self) -> Result<(), ConfigError> {
121 for (value, message) in [
122 (
123 self.uuid_compression_potential,
124 "uuid_compression_potential must be in 0.0..=1.0",
125 ),
126 (
127 self.min_delta_potential,
128 "min_delta_potential must be in 0.0..=1.0",
129 ),
130 (
131 self.min_compression_potential,
132 "min_compression_potential must be in 0.0..=1.0",
133 ),
134 ] {
135 if !(0.0..=1.0).contains(&value) {
136 return Err(ConfigError::InconsistentBounds {
137 section: "compression",
138 message,
139 });
140 }
141 }
142
143 for (value, message) in [
144 (
145 self.delta_threshold,
146 "delta_threshold must be finite and non-negative",
147 ),
148 (
149 self.run_length_threshold,
150 "run_length_threshold must be finite and non-negative",
151 ),
152 ] {
153 if !value.is_finite() || value < 0.0 {
154 return Err(ConfigError::InconsistentBounds {
155 section: "compression",
156 message,
157 });
158 }
159 }
160
161 Ok(())
162 }
163}
164
165#[derive(Debug, Clone, PartialEq)]
167pub enum CompressionStrategy {
168 None,
170 Dictionary {
172 dictionary: HashMap<String, u16>,
174 },
175 Delta {
177 base_values: HashMap<String, f64>,
179 },
180 RunLength,
182 Hybrid {
184 string_dict: HashMap<String, u16>,
186 numeric_deltas: HashMap<String, f64>,
188 },
189}
190
191#[derive(Debug, Clone)]
193pub struct SchemaAnalyzer {
194 patterns: HashMap<String, PatternInfo>,
196 numeric_fields: HashMap<String, NumericStats>,
198 string_repetitions: HashMap<String, u32>,
200 config: CompressionConfig,
202}
203
204#[derive(Debug, Clone)]
205struct PatternInfo {
206 frequency: u32,
207 compression_potential: f32,
208}
209
210#[derive(Debug, Clone)]
211struct NumericStats {
212 values: Vec<f64>,
213 delta_potential: f32,
214 base_value: f64,
215}
216
217impl SchemaAnalyzer {
218 pub fn new() -> Self {
220 Self {
221 patterns: HashMap::new(),
222 numeric_fields: HashMap::new(),
223 string_repetitions: HashMap::new(),
224 config: CompressionConfig::default(),
225 }
226 }
227
228 pub fn with_config(config: CompressionConfig) -> Self {
230 Self {
231 patterns: HashMap::new(),
232 numeric_fields: HashMap::new(),
233 string_repetitions: HashMap::new(),
234 config,
235 }
236 }
237
238 pub fn analyze(&mut self, data: &JsonValue) -> DomainResult<CompressionStrategy> {
240 self.patterns.clear();
242 self.numeric_fields.clear();
243 self.string_repetitions.clear();
244
245 self.analyze_recursive(data, "")?;
247
248 self.determine_strategy()
250 }
251
252 fn analyze_recursive(&mut self, value: &JsonValue, path: &str) -> DomainResult<()> {
254 match value {
255 JsonValue::Object(obj) => {
256 for (key, val) in obj {
257 let field_path = if path.is_empty() {
258 key.clone()
259 } else {
260 format!("{path}.{key}")
261 };
262 self.analyze_recursive(val, &field_path)?;
263 }
264 }
265 JsonValue::Array(arr) => {
266 if arr.len() > self.config.min_array_length {
268 self.analyze_array_patterns(arr, path)?;
269 }
270 for (idx, item) in arr.iter().enumerate() {
271 let item_path = format!("{path}[{idx}]");
272 self.analyze_recursive(item, &item_path)?;
273 }
274 }
275 JsonValue::String(s) => {
276 self.analyze_string_pattern(s, path);
277 }
278 JsonValue::Number(n) => {
279 if let Some(f) = n.as_f64() {
280 self.analyze_numeric_pattern(f, path);
281 }
282 }
283 _ => {}
284 }
285 Ok(())
286 }
287
288 fn analyze_array_patterns(&mut self, arr: &[JsonValue], path: &str) -> DomainResult<()> {
290 if let Some(JsonValue::Object(first)) = arr.first() {
292 let structure_key = format!("array_structure:{path}");
293 let field_names: Vec<&str> = first.keys().map(|k| k.as_str()).collect();
294 let pattern = field_names.join(",");
295
296 let matching_count = arr
298 .iter()
299 .filter_map(|v| v.as_object())
300 .filter(|obj| {
301 let obj_fields: Vec<&str> = obj.keys().map(|k| k.as_str()).collect();
302 obj_fields.join(",") == pattern
303 })
304 .count();
305
306 if matching_count > self.config.min_frequency_count as usize {
307 let info = PatternInfo {
308 frequency: matching_count as u32,
309 compression_potential: (matching_count as f32 - 1.0) / matching_count as f32,
310 };
311 self.patterns.insert(structure_key, info);
312 }
313 }
314
315 if arr.len() > 2 {
317 let mut value_counts = HashMap::new();
318 for value in arr {
319 let key = match value {
320 JsonValue::String(s) => format!("string:{s}"),
321 JsonValue::Number(n) => format!("number:{n}"),
322 JsonValue::Bool(b) => format!("bool:{b}"),
323 _ => continue,
324 };
325 *value_counts.entry(key).or_insert(0) += 1;
326 }
327
328 for (value_key, count) in value_counts {
329 if count > self.config.min_frequency_count {
330 let info = PatternInfo {
331 frequency: count,
332 compression_potential: (count as f32 - 1.0) / count as f32,
333 };
334 self.patterns
335 .insert(format!("array_value:{path}:{value_key}"), info);
336 }
337 }
338 }
339
340 Ok(())
341 }
342
343 fn analyze_string_pattern(&mut self, s: &str, _path: &str) {
345 *self.string_repetitions.entry(s.to_string()).or_insert(0) += 1;
347
348 if s.len() > 10 {
350 if s.starts_with("http://") || s.starts_with("https://") {
352 let prefix = if s.starts_with("https://") {
353 "https://"
354 } else {
355 "http://"
356 };
357 self.patterns
358 .entry(format!("url_prefix:{prefix}"))
359 .or_insert(PatternInfo {
360 frequency: 0,
361 compression_potential: 0.0,
362 })
363 .frequency += 1;
364 }
365
366 if s.len() == 36 && s.chars().filter(|&c| c == '-').count() == 4 {
368 self.patterns
369 .entry("uuid_pattern".to_string())
370 .or_insert(PatternInfo {
371 frequency: 0,
372 compression_potential: self.config.uuid_compression_potential,
373 })
374 .frequency += 1;
375 }
376 }
377 }
378
379 fn analyze_numeric_pattern(&mut self, value: f64, path: &str) {
381 self.numeric_fields
382 .entry(path.to_string())
383 .or_insert_with(|| NumericStats {
384 values: Vec::new(),
385 delta_potential: 0.0,
386 base_value: value,
387 })
388 .values
389 .push(value);
390 }
391
392 fn determine_strategy(&mut self) -> DomainResult<CompressionStrategy> {
394 let mut delta_score = 0.0;
395
396 let (string_dict, dict_net_savings) =
399 build_dictionary(&self.string_repetitions, &self.config);
400 let string_dict_selected =
401 !string_dict.is_empty() && dict_net_savings >= self.config.min_net_savings as i64;
402
403 let mut numeric_deltas = HashMap::new();
405
406 for (path, stats) in &mut self.numeric_fields {
407 if stats.values.len() > 2 {
408 stats
410 .values
411 .sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));
412
413 let deltas: Vec<f64> = stats
414 .values
415 .windows(2)
416 .map(|window| window[1] - window[0])
417 .collect();
418
419 if !deltas.is_empty() {
420 let avg_delta = deltas.iter().sum::<f64>() / deltas.len() as f64;
421 let delta_variance =
422 deltas.iter().map(|d| (d - avg_delta).powi(2)).sum::<f64>()
423 / deltas.len() as f64;
424
425 stats.delta_potential = 1.0 / (1.0 + delta_variance as f32);
427
428 if stats.delta_potential > self.config.min_delta_potential {
429 delta_score += stats.delta_potential * stats.values.len() as f32;
430 numeric_deltas.insert(path.clone(), stats.base_value);
431 }
432 }
433 }
434 }
435
436 match (
438 string_dict_selected,
439 delta_score >= self.config.delta_threshold,
440 ) {
441 (true, true) => Ok(CompressionStrategy::Hybrid {
442 string_dict,
443 numeric_deltas,
444 }),
445 (true, false) => Ok(CompressionStrategy::Dictionary {
446 dictionary: string_dict,
447 }),
448 (false, true) => Ok(CompressionStrategy::Delta {
449 base_values: numeric_deltas,
450 }),
451 (false, false) => {
452 let run_length_score = self
454 .patterns
455 .values()
456 .filter(|p| p.compression_potential > self.config.min_compression_potential)
457 .map(|p| p.frequency as f32 * p.compression_potential)
458 .sum::<f32>();
459
460 if run_length_score >= self.config.run_length_threshold {
461 Ok(CompressionStrategy::RunLength)
462 } else {
463 Ok(CompressionStrategy::None)
464 }
465 }
466 }
467 }
468}
469
470fn decimal_digits(n: u16) -> usize {
472 n.to_string().len()
473}
474
475fn build_dictionary(
488 repetitions: &HashMap<String, u32>,
489 config: &CompressionConfig,
490) -> (HashMap<String, u16>, i64) {
491 let mut candidates: Vec<(&String, u32)> = repetitions
492 .iter()
493 .filter_map(|(s, &count)| {
494 (count > config.min_frequency_count && s.len() > config.min_string_length)
495 .then_some((s, count))
496 })
497 .collect();
498 candidates.sort_by(|(s1, c1), (s2, c2)| {
499 let payoff1 = *c1 as usize * s1.len();
500 let payoff2 = *c2 as usize * s2.len();
501 payoff2.cmp(&payoff1).then_with(|| s1.cmp(s2))
502 });
503
504 let mut dictionary = HashMap::new();
505 let mut net: i64 = 0;
506 let mut index: u16 = 0;
507 for (s, count) in candidates {
508 if index == u16::MAX {
513 break;
514 }
515 let marker_len = 1 + decimal_digits(index);
516 let gain = count as i64 * (s.len() as i64 - marker_len as i64);
517 let cost = s.len() as i64 + 3;
518 if gain > cost {
519 net += gain - cost;
520 dictionary.insert(s.clone(), index);
521 index += 1;
522 }
523 }
524 if !dictionary.is_empty() {
525 net -= 10; }
527 (dictionary, net)
528}
529
530fn wire_size(data: &JsonValue, metadata: &HashMap<String, JsonValue>) -> DomainResult<usize> {
539 let mut size = serde_json::to_string(data)
540 .map_err(|e| DomainError::CompressionError(format!("JSON serialization failed: {e}")))?
541 .len();
542 if !metadata.is_empty() {
543 size += serde_json::to_string(metadata)
544 .map_err(|e| DomainError::CompressionError(format!("JSON serialization failed: {e}")))?
545 .len();
546 }
547 Ok(size)
548}
549
550fn dictionary_metadata(dictionary: &HashMap<String, u16>) -> JsonValue {
556 let mut ordered: Vec<Option<&str>> = vec![None; dictionary.len()];
557 for (s, &i) in dictionary {
558 if let Some(slot) = ordered.get_mut(i as usize) {
559 *slot = Some(s.as_str());
560 }
561 }
562 JsonValue::Array(
563 ordered
564 .into_iter()
565 .map(|s| JsonValue::String(s.unwrap_or_default().to_string()))
566 .collect(),
567 )
568}
569
570fn substitute_dictionary_strings(data: &JsonValue, dictionary: &HashMap<String, u16>) -> JsonValue {
581 match data {
582 JsonValue::Object(obj) => {
583 let mut out = serde_json::Map::with_capacity(obj.len());
584 for (key, value) in obj {
585 out.insert(
586 key.clone(),
587 substitute_dictionary_strings(value, dictionary),
588 );
589 }
590 JsonValue::Object(out)
591 }
592 JsonValue::Array(arr) => JsonValue::Array(
593 arr.iter()
594 .map(|v| substitute_dictionary_strings(v, dictionary))
595 .collect(),
596 ),
597 JsonValue::String(s) => {
598 if let Some(&index) = dictionary.get(s) {
599 JsonValue::String(format!("{DICT_SENTINEL}{index}"))
600 } else if s.starts_with(DICT_SENTINEL) {
601 JsonValue::String(format!("{DICT_SENTINEL}{s}"))
602 } else {
603 data.clone()
604 }
605 }
606 _ => data.clone(),
607 }
608}
609
610#[derive(Debug, Clone)]
612pub struct SchemaCompressor {
613 strategy: CompressionStrategy,
614 analyzer: SchemaAnalyzer,
615 config: CompressionConfig,
616}
617
618impl SchemaCompressor {
619 pub fn new() -> Self {
621 let config = CompressionConfig::default();
622 Self {
623 strategy: CompressionStrategy::None,
624 analyzer: SchemaAnalyzer::with_config(config.clone()),
625 config,
626 }
627 }
628
629 pub fn with_strategy(strategy: CompressionStrategy) -> Self {
631 let config = CompressionConfig::default();
632 Self {
633 strategy,
634 analyzer: SchemaAnalyzer::with_config(config.clone()),
635 config,
636 }
637 }
638
639 pub fn with_config(config: CompressionConfig) -> Self {
641 Self {
642 strategy: CompressionStrategy::None,
643 analyzer: SchemaAnalyzer::with_config(config.clone()),
644 config,
645 }
646 }
647
648 pub fn analyze_and_optimize(&mut self, data: &JsonValue) -> DomainResult<&CompressionStrategy> {
650 self.strategy = self.analyzer.analyze(data)?;
651 Ok(&self.strategy)
652 }
653
654 pub fn compress(&self, data: &JsonValue) -> DomainResult<CompressedData> {
656 match &self.strategy {
657 CompressionStrategy::None => {
658 let metadata = HashMap::new();
659 Ok(CompressedData {
660 strategy: self.strategy.clone(),
661 compressed_size: wire_size(data, &metadata)?,
662 data: data.clone(),
663 compression_metadata: metadata,
664 })
665 }
666
667 CompressionStrategy::Dictionary { dictionary } => {
668 self.compress_with_dictionary(data, dictionary)
669 }
670
671 CompressionStrategy::Delta { base_values } => {
672 self.compress_with_delta(data, base_values)
673 }
674
675 CompressionStrategy::RunLength => self.compress_with_run_length(data),
676
677 CompressionStrategy::Hybrid {
678 string_dict,
679 numeric_deltas,
680 } => self.compress_hybrid(data, string_dict, numeric_deltas),
681 }
682 }
683
684 fn compress_with_dictionary(
686 &self,
687 data: &JsonValue,
688 dictionary: &HashMap<String, u16>,
689 ) -> DomainResult<CompressedData> {
690 let mut metadata = HashMap::new();
691 metadata.insert("dict".to_string(), dictionary_metadata(dictionary));
692
693 let compressed = substitute_dictionary_strings(data, dictionary);
694 let compressed_size = wire_size(&compressed, &metadata)?;
695
696 Ok(CompressedData {
697 strategy: self.strategy.clone(),
698 compressed_size,
699 data: compressed,
700 compression_metadata: metadata,
701 })
702 }
703
704 fn compress_with_delta(
706 &self,
707 data: &JsonValue,
708 base_values: &HashMap<String, f64>,
709 ) -> DomainResult<CompressedData> {
710 let mut metadata = HashMap::new();
711
712 for (path, base) in base_values {
714 let number = serde_json::Number::from_f64(*base).ok_or_else(|| {
715 DomainError::CompressionError(format!(
716 "delta base value for path '{path}' is non-finite (NaN or Infinity); cannot compress"
717 ))
718 })?;
719 metadata.insert(format!("base_{path}"), JsonValue::Number(number));
720 }
721
722 let compressed = self.apply_delta_compression(data, base_values)?;
724 let compressed_size = wire_size(&compressed, &metadata)?;
725
726 Ok(CompressedData {
727 strategy: self.strategy.clone(),
728 compressed_size,
729 data: compressed,
730 compression_metadata: metadata,
731 })
732 }
733
734 fn compress_with_run_length(&self, data: &JsonValue) -> DomainResult<CompressedData> {
736 let metadata = HashMap::new();
737 let compressed = self.apply_run_length_encoding(data)?;
738 let compressed_size = wire_size(&compressed, &metadata)?;
739
740 Ok(CompressedData {
741 strategy: self.strategy.clone(),
742 compressed_size,
743 data: compressed,
744 compression_metadata: metadata,
745 })
746 }
747
748 fn apply_run_length_encoding(&self, data: &JsonValue) -> DomainResult<JsonValue> {
750 match data {
751 JsonValue::Object(obj) => {
752 let mut compressed_obj = serde_json::Map::new();
753 for (key, value) in obj {
754 compressed_obj.insert(key.clone(), self.apply_run_length_encoding(value)?);
755 }
756 Ok(JsonValue::Object(compressed_obj))
757 }
758 JsonValue::Array(arr) if arr.len() > 2 => {
759 let mut compressed_runs = Vec::new();
761 let mut current_value = None;
762 let mut run_count = 0;
763
764 for item in arr {
765 if Some(item) == current_value.as_ref() {
766 run_count += 1;
767 } else {
768 if let Some(value) = current_value {
770 if run_count > self.config.min_frequency_count {
771 compressed_runs.push(json!({
773 "rle_value": value,
774 "rle_count": run_count
775 }));
776 } else {
777 compressed_runs.push(value);
779 }
780 }
781
782 current_value = Some(item.clone());
784 run_count = 1;
785 }
786 }
787
788 if let Some(value) = current_value {
790 if run_count > self.config.min_frequency_count {
791 compressed_runs.push(json!({
792 "rle_value": value,
793 "rle_count": run_count
794 }));
795 } else {
796 compressed_runs.push(value);
797 }
798 }
799
800 Ok(JsonValue::Array(compressed_runs))
801 }
802 JsonValue::Array(arr) => {
803 let compressed_arr: Result<Vec<_>, _> = arr
805 .iter()
806 .map(|item| self.apply_run_length_encoding(item))
807 .collect();
808 Ok(JsonValue::Array(compressed_arr?))
809 }
810 _ => Ok(data.clone()),
811 }
812 }
813
814 fn compress_hybrid(
816 &self,
817 data: &JsonValue,
818 string_dict: &HashMap<String, u16>,
819 numeric_deltas: &HashMap<String, f64>,
820 ) -> DomainResult<CompressedData> {
821 let mut metadata = HashMap::new();
822 metadata.insert("dict".to_string(), dictionary_metadata(string_dict));
823
824 for (path, base) in numeric_deltas {
826 let number = serde_json::Number::from_f64(*base).ok_or_else(|| {
827 DomainError::CompressionError(format!(
828 "delta base value for path '{path}' is non-finite (NaN or Infinity); cannot compress"
829 ))
830 })?;
831 metadata.insert(format!("base_{path}"), JsonValue::Number(number));
832 }
833
834 let dict_compressed = substitute_dictionary_strings(data, string_dict);
836 let final_compressed = self.apply_delta_compression(&dict_compressed, numeric_deltas)?;
837
838 let compressed_size = wire_size(&final_compressed, &metadata)?;
839
840 Ok(CompressedData {
841 strategy: self.strategy.clone(),
842 compressed_size,
843 data: final_compressed,
844 compression_metadata: metadata,
845 })
846 }
847
848 fn apply_delta_compression(
850 &self,
851 data: &JsonValue,
852 base_values: &HashMap<String, f64>,
853 ) -> DomainResult<JsonValue> {
854 self.apply_delta_recursive(data, "", base_values)
855 }
856
857 fn apply_delta_recursive(
859 &self,
860 data: &JsonValue,
861 path: &str,
862 base_values: &HashMap<String, f64>,
863 ) -> DomainResult<JsonValue> {
864 match data {
865 JsonValue::Object(obj) => {
866 let mut compressed_obj = serde_json::Map::new();
867 for (key, value) in obj {
868 let field_path = if path.is_empty() {
869 key.clone()
870 } else {
871 format!("{path}.{key}")
872 };
873 compressed_obj.insert(
874 key.clone(),
875 self.apply_delta_recursive(value, &field_path, base_values)?,
876 );
877 }
878 Ok(JsonValue::Object(compressed_obj))
879 }
880 JsonValue::Array(arr) if arr.len() > 2 => {
881 if self.is_numeric_sequence(arr) {
883 self.compress_numeric_array_with_delta(arr, path, base_values)
884 } else {
885 let compressed_arr: Result<Vec<_>, _> = arr
887 .iter()
888 .enumerate()
889 .map(|(idx, item)| {
890 let item_path = format!("{path}[{idx}]");
891 self.apply_delta_recursive(item, &item_path, base_values)
892 })
893 .collect();
894 Ok(JsonValue::Array(compressed_arr?))
895 }
896 }
897 JsonValue::Array(arr) => {
898 let compressed_arr: Result<Vec<_>, _> = arr
900 .iter()
901 .enumerate()
902 .map(|(idx, item)| {
903 let item_path = format!("{path}[{idx}]");
904 self.apply_delta_recursive(item, &item_path, base_values)
905 })
906 .collect();
907 Ok(JsonValue::Array(compressed_arr?))
908 }
909 _ => Ok(data.clone()),
910 }
911 }
912
913 fn is_numeric_sequence(&self, arr: &[JsonValue]) -> bool {
915 if arr.len() < self.config.min_numeric_sequence_size {
916 return false;
917 }
918
919 arr.iter().all(|v| v.is_number())
921 }
922
923 fn compress_numeric_array_with_delta(
925 &self,
926 arr: &[JsonValue],
927 path: &str,
928 base_values: &HashMap<String, f64>,
929 ) -> DomainResult<JsonValue> {
930 let mut compressed_array = Vec::new();
931
932 let numbers: Vec<f64> = arr.iter().filter_map(|v| v.as_f64()).collect();
934
935 if numbers.is_empty() {
936 return Ok(JsonValue::Array(arr.to_vec()));
937 }
938
939 let base_value = base_values.get(path).copied().unwrap_or(numbers[0]);
941
942 compressed_array.push(json!({
944 "delta_base": base_value,
945 "delta_type": "numeric_sequence"
946 }));
947
948 let deltas: Vec<f64> = numbers.iter().map(|&num| num - base_value).collect();
950
951 let original_precision = numbers.iter().map(|n| format!("{n}").len()).sum::<usize>();
953
954 let delta_precision = deltas.iter().map(|d| format!("{d}").len()).sum::<usize>();
955
956 if delta_precision < original_precision {
957 compressed_array.extend(deltas.into_iter().map(JsonValue::from));
959 } else {
960 return Ok(JsonValue::Array(arr.to_vec()));
962 }
963
964 Ok(JsonValue::Array(compressed_array))
965 }
966}
967
968#[derive(Debug, Clone)]
970pub struct CompressedData {
971 pub strategy: CompressionStrategy,
973 pub compressed_size: usize,
977 pub data: JsonValue,
979 pub compression_metadata: HashMap<String, JsonValue>,
981}
982
983impl CompressedData {
984 pub fn compression_ratio(&self, original_size: usize) -> f32 {
986 if original_size == 0 {
987 return 1.0;
988 }
989 self.compressed_size as f32 / original_size as f32
990 }
991
992 pub fn compression_savings(&self, original_size: usize) -> isize {
994 original_size as isize - self.compressed_size as isize
995 }
996}
997
998impl Default for SchemaAnalyzer {
999 fn default() -> Self {
1000 Self::new()
1001 }
1002}
1003
1004impl Default for SchemaCompressor {
1005 fn default() -> Self {
1006 Self::new()
1007 }
1008}
1009
1010#[cfg(test)]
1011mod tests {
1012 use super::*;
1013 use serde_json::json;
1014
1015 #[test]
1016 fn test_schema_analyzer_dictionary_potential() {
1017 let mut analyzer = SchemaAnalyzer::new();
1018
1019 let data = json!({
1020 "users": [
1021 {"name": "John Doe", "role": "admin", "status": "active", "department": "engineering"},
1022 {"name": "Jane Smith", "role": "admin", "status": "active", "department": "engineering"},
1023 {"name": "Bob Wilson", "role": "admin", "status": "active", "department": "engineering"},
1024 {"name": "Alice Brown", "role": "admin", "status": "active", "department": "engineering"},
1025 {"name": "Charlie Davis", "role": "admin", "status": "active", "department": "engineering"},
1026 {"name": "Diana Evans", "role": "admin", "status": "active", "department": "engineering"},
1027 {"name": "Frank Miller", "role": "admin", "status": "active", "department": "engineering"},
1028 {"name": "Grace Wilson", "role": "admin", "status": "active", "department": "engineering"}
1029 ]
1030 });
1031
1032 let strategy = analyzer.analyze(&data).unwrap();
1033
1034 match strategy {
1036 CompressionStrategy::Dictionary { .. } | CompressionStrategy::Hybrid { .. } => {
1037 }
1039 _ => panic!("Expected dictionary-based compression strategy"),
1040 }
1041 }
1042
1043 #[test]
1044 fn test_schema_analyzer_realistic_ecommerce_payload() {
1045 let mut analyzer = SchemaAnalyzer::new();
1049
1050 let data = json!({
1051 "products": [
1052 {"id": 1001, "name": "MacBook Pro", "category": "Electronics", "status": "available", "brand": "Apple", "price": 2399.99},
1053 {"id": 1002, "name": "iPhone 15", "category": "Electronics", "status": "available", "brand": "Apple", "price": 999.99},
1054 {"id": 1003, "name": "AirPods Pro", "category": "Electronics", "status": "available", "brand": "Apple", "price": 249.99}
1055 ],
1056 "store": {"name": "Tech Store", "status": "operational", "location": "San Francisco"}
1057 });
1058
1059 let strategy = analyzer.analyze(&data).unwrap();
1060
1061 match &strategy {
1062 CompressionStrategy::Dictionary { .. } | CompressionStrategy::Hybrid { .. } => {}
1063 other => panic!("Expected dictionary-based compression strategy, got {other:?}"),
1064 }
1065
1066 let original_size = serde_json::to_string(&data).unwrap().len();
1067 let compressed = SchemaCompressor::with_strategy(strategy)
1068 .compress(&data)
1069 .unwrap();
1070 assert!(
1071 compressed.compression_savings(original_size) > 0,
1072 "expected genuine positive wire-byte savings, got {}",
1073 compressed.compression_savings(original_size)
1074 );
1075 }
1076
1077 #[test]
1078 fn test_schema_analyzer_realistic_api_response_payload() {
1079 let mut analyzer = SchemaAnalyzer::new();
1085
1086 let data = json!({
1087 "status": "success",
1088 "data": {
1089 "users": [
1090 {"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"},
1091 {"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"},
1092 {"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"},
1093 {"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"},
1094 {"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"}
1095 ]
1096 },
1097 "pagination": {"page": 1, "per_page": 25, "total_pages": 4, "total_items": 89},
1098 "meta": {"request_id": "req_12345", "timestamp": "2024-01-15T10:30:15Z", "version": "v1.2.3"}
1099 });
1100
1101 let strategy = analyzer.analyze(&data).unwrap();
1102
1103 match &strategy {
1104 CompressionStrategy::Dictionary { .. } | CompressionStrategy::Hybrid { .. } => {}
1105 other => panic!("Expected dictionary-based compression strategy, got {other:?}"),
1106 }
1107
1108 let original_size = serde_json::to_string(&data).unwrap().len();
1109 let compressed = SchemaCompressor::with_strategy(strategy)
1110 .compress(&data)
1111 .unwrap();
1112 assert!(
1113 compressed.compression_savings(original_size) > 0,
1114 "expected genuine positive wire-byte savings, got {}",
1115 compressed.compression_savings(original_size)
1116 );
1117 }
1118
1119 #[test]
1120 fn test_schema_analyzer_no_repetition_stays_none() {
1121 let mut analyzer = SchemaAnalyzer::new();
1125
1126 let data = json!({
1127 "id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
1128 "name": "Unique Product Name Alpha",
1129 "description": "A completely unique description of this particular item with no repeats",
1130 "vendor": "Acme Corporation International",
1131 "location": "Building 12, Warehouse Section D",
1132 "notes": "Handled with care during transit process"
1133 });
1134
1135 let strategy = analyzer.analyze(&data).unwrap();
1136 assert_eq!(strategy, CompressionStrategy::None);
1137 }
1138
1139 #[test]
1140 fn test_schema_analyzer_tiny_duplicate_stays_none_below_savings_floor() {
1141 let mut analyzer = SchemaAnalyzer::new();
1146
1147 let data = json!({"a": "hello", "b": "hello", "c": "world"});
1148
1149 let strategy = analyzer.analyze(&data).unwrap();
1150 assert_eq!(strategy, CompressionStrategy::None);
1151 }
1152
1153 #[test]
1154 fn test_schema_analyzer_long_repeated_string_selects_dictionary_and_shrinks() {
1155 let mut analyzer = SchemaAnalyzer::new();
1159
1160 let data = json!({
1161 "a": "premium_subscription",
1162 "b": "premium_subscription",
1163 "c": "premium_subscription",
1164 "d": "unique"
1165 });
1166
1167 let strategy = analyzer.analyze(&data).unwrap();
1168 let dictionary = match &strategy {
1169 CompressionStrategy::Dictionary { dictionary } => dictionary,
1170 other => panic!("Expected Dictionary strategy, got {other:?}"),
1171 };
1172
1173 let original_size = serde_json::to_string(&data).unwrap().len();
1174 let compressed = SchemaCompressor::with_strategy(CompressionStrategy::Dictionary {
1175 dictionary: dictionary.clone(),
1176 })
1177 .compress(&data)
1178 .unwrap();
1179 assert!(compressed.compression_savings(original_size) > 0);
1180 }
1181
1182 #[test]
1183 fn test_schema_compressor_basic() {
1184 let compressor = SchemaCompressor::new();
1185
1186 let data = json!({
1187 "message": "hello world",
1188 "count": 42
1189 });
1190
1191 let original_size = serde_json::to_string(&data).unwrap().len();
1192 let compressed = compressor.compress(&data).unwrap();
1193
1194 assert!(compressed.compressed_size > 0);
1195 assert!(compressed.compression_ratio(original_size) <= 1.0);
1196 }
1197
1198 #[test]
1199 fn test_dictionary_compression() {
1200 let mut dictionary = HashMap::new();
1201 dictionary.insert("active".to_string(), 0);
1202 dictionary.insert("admin".to_string(), 1);
1203
1204 let compressor =
1205 SchemaCompressor::with_strategy(CompressionStrategy::Dictionary { dictionary });
1206
1207 let data = json!({
1208 "status": "active",
1209 "role": "admin",
1210 "description": "active admin user"
1211 });
1212
1213 let result = compressor.compress(&data).unwrap();
1214
1215 assert_eq!(
1217 result.compression_metadata.get("dict"),
1218 Some(&json!(["active", "admin"]))
1219 );
1220 }
1221
1222 #[test]
1223 fn test_dictionary_compression_never_produces_numbers_from_substitution() {
1224 let mut dictionary = HashMap::new();
1230 dictionary.insert("active".to_string(), 0);
1231
1232 let compressor =
1233 SchemaCompressor::with_strategy(CompressionStrategy::Dictionary { dictionary });
1234
1235 let data = json!({
1236 "status": "active",
1237 "count": 0
1238 });
1239
1240 let result = compressor.compress(&data).unwrap();
1241
1242 assert_eq!(result.data, json!({"status": "\u{7F}0", "count": 0}));
1243 }
1244
1245 #[test]
1246 fn test_dictionary_sentinel_escaping_encode_shape() {
1247 let mut dictionary = HashMap::new();
1254 dictionary.insert("greeting".to_string(), 0);
1255
1256 let data = json!({
1257 "a": "\u{7F}foo",
1258 "b": "\u{7F}\u{7F}bar",
1259 "c": "\u{7F}0",
1260 "d": "greeting"
1261 });
1262
1263 let substituted = substitute_dictionary_strings(&data, &dictionary);
1264 assert_eq!(
1265 substituted,
1266 json!({
1267 "a": "\u{7F}\u{7F}foo",
1268 "b": "\u{7F}\u{7F}\u{7F}bar",
1269 "c": "\u{7F}\u{7F}0",
1270 "d": "\u{7F}0"
1271 })
1272 );
1273 }
1274
1275 #[test]
1276 fn test_compressed_size_matches_wire_bytes_for_every_strategy() {
1277 fn expected_wire_size(data: &JsonValue, metadata: &HashMap<String, JsonValue>) -> usize {
1281 let mut size = serde_json::to_string(data).unwrap().len();
1282 if !metadata.is_empty() {
1283 size += serde_json::to_string(metadata).unwrap().len();
1284 }
1285 size
1286 }
1287
1288 let data = json!({
1289 "status": "active",
1290 "count": 3,
1291 "sequence": [1.0, 2.0, 3.0],
1292 "repeated": [1, 1, 1, 2, 2]
1293 });
1294
1295 let mut dictionary = HashMap::new();
1296 dictionary.insert("active".to_string(), 0);
1297 let mut base_values = HashMap::new();
1298 base_values.insert("sequence".to_string(), 1.0);
1299
1300 for strategy in [
1301 CompressionStrategy::None,
1302 CompressionStrategy::Dictionary {
1303 dictionary: dictionary.clone(),
1304 },
1305 CompressionStrategy::Delta {
1306 base_values: base_values.clone(),
1307 },
1308 CompressionStrategy::RunLength,
1309 CompressionStrategy::Hybrid {
1310 string_dict: dictionary.clone(),
1311 numeric_deltas: base_values.clone(),
1312 },
1313 ] {
1314 let compressor = SchemaCompressor::with_strategy(strategy);
1315 let result = compressor.compress(&data).unwrap();
1316 assert_eq!(
1317 result.compressed_size,
1318 expected_wire_size(&result.data, &result.compression_metadata),
1319 "strategy {:?} mismatched wire size",
1320 result.strategy
1321 );
1322 }
1323 }
1324
1325 #[test]
1326 fn test_build_dictionary_caps_index_at_u16_max_without_overflow() {
1327 let mut repetitions = HashMap::new();
1337 for i in 0..(u16::MAX as u32 + 2) {
1338 repetitions.insert(format!("padding_string_{i:05}"), 2);
1339 }
1340
1341 let (dictionary, _net) = build_dictionary(&repetitions, &CompressionConfig::default());
1342
1343 assert!(
1344 dictionary.len() <= u16::MAX as usize,
1345 "dictionary must never exceed the u16 index space, got {} entries",
1346 dictionary.len()
1347 );
1348
1349 let distinct_indices: std::collections::HashSet<u16> =
1350 dictionary.values().copied().collect();
1351 assert_eq!(
1352 distinct_indices.len(),
1353 dictionary.len(),
1354 "every dictionary entry must have a unique index — a mismatch here means indices \
1355 wrapped and collided"
1356 );
1357 }
1358
1359 #[test]
1360 fn test_compression_strategy_selection() {
1361 let mut analyzer = SchemaAnalyzer::new();
1362
1363 let simple_data = json!({
1365 "unique_field_1": "unique_value_1",
1366 "unique_field_2": "unique_value_2"
1367 });
1368
1369 let strategy = analyzer.analyze(&simple_data).unwrap();
1370 assert_eq!(strategy, CompressionStrategy::None);
1371 }
1372
1373 #[test]
1374 fn test_numeric_delta_analysis() {
1375 let mut analyzer = SchemaAnalyzer::new();
1376
1377 let data = json!({
1378 "measurements": [
1379 {"time": 100, "value": 10.0},
1380 {"time": 101, "value": 10.5},
1381 {"time": 102, "value": 11.0},
1382 {"time": 103, "value": 11.5}
1383 ]
1384 });
1385
1386 let _strategy = analyzer.analyze(&data).unwrap();
1387
1388 assert!(!analyzer.numeric_fields.is_empty());
1390 }
1391
1392 #[test]
1393 fn test_run_length_encoding() {
1394 let compressor = SchemaCompressor::with_strategy(CompressionStrategy::RunLength);
1395
1396 let data = json!({
1397 "repeated_values": [1, 1, 1, 2, 2, 3, 3, 3, 3]
1398 });
1399
1400 let result = compressor.compress(&data).unwrap();
1401
1402 assert!(result.compressed_size > 0);
1404
1405 let compressed_array = &result.data["repeated_values"];
1407 assert!(compressed_array.is_array());
1408
1409 let array = compressed_array.as_array().unwrap();
1411 let has_rle = array.iter().any(|v| v.get("rle_value").is_some());
1412 assert!(has_rle);
1413 }
1414
1415 #[test]
1416 fn test_delta_compression() {
1417 let mut base_values = HashMap::new();
1418 base_values.insert("sequence".to_string(), 100.0);
1419
1420 let compressor =
1421 SchemaCompressor::with_strategy(CompressionStrategy::Delta { base_values });
1422
1423 let data = json!({
1424 "sequence": [100.0, 101.0, 102.0, 103.0, 104.0]
1425 });
1426
1427 let result = compressor.compress(&data).unwrap();
1428
1429 assert!(result.compressed_size > 0);
1431
1432 let compressed_array = &result.data["sequence"];
1434 assert!(compressed_array.is_array());
1435
1436 let array = compressed_array.as_array().unwrap();
1438 let has_delta_base = array.iter().any(|v| v.get("delta_base").is_some());
1439 assert!(has_delta_base);
1440 }
1441
1442 #[test]
1443 fn test_delta_compression_rejects_nan_base() {
1444 let mut base_values = HashMap::new();
1445 base_values.insert("sequence".to_string(), f64::NAN);
1446
1447 let compressor =
1448 SchemaCompressor::with_strategy(CompressionStrategy::Delta { base_values });
1449
1450 let data = json!({ "sequence": [1.0, 2.0, 3.0] });
1451
1452 let err = compressor
1453 .compress(&data)
1454 .expect_err("expected error for NaN base");
1455 match err {
1456 DomainError::CompressionError(msg) => {
1457 assert!(msg.contains("non-finite"), "unexpected message: {msg}");
1458 assert!(msg.contains("sequence"), "expected path in message: {msg}");
1459 }
1460 other => panic!("expected CompressionError, got {other:?}"),
1461 }
1462 }
1463
1464 #[test]
1465 fn test_delta_compression_rejects_infinity_base() {
1466 let mut base_values = HashMap::new();
1467 base_values.insert("sequence".to_string(), f64::INFINITY);
1468
1469 let compressor =
1470 SchemaCompressor::with_strategy(CompressionStrategy::Delta { base_values });
1471
1472 let data = json!({ "sequence": [1.0, 2.0, 3.0] });
1473
1474 let err = compressor
1475 .compress(&data)
1476 .expect_err("expected error for Infinity base");
1477 assert!(matches!(err, DomainError::CompressionError(_)));
1478 }
1479
1480 #[test]
1481 fn test_hybrid_compression_rejects_nan_base() {
1482 let string_dict = HashMap::new();
1483 let mut numeric_deltas = HashMap::new();
1484 numeric_deltas.insert("sequence".to_string(), f64::NEG_INFINITY);
1485
1486 let compressor = SchemaCompressor::with_strategy(CompressionStrategy::Hybrid {
1487 string_dict,
1488 numeric_deltas,
1489 });
1490
1491 let data = json!({ "sequence": [1.0, 2.0, 3.0] });
1492
1493 let err = compressor
1494 .compress(&data)
1495 .expect_err("expected error for non-finite base");
1496 match err {
1497 DomainError::CompressionError(msg) => {
1498 assert!(msg.contains("non-finite"), "unexpected message: {msg}");
1499 }
1500 other => panic!("expected CompressionError, got {other:?}"),
1501 }
1502 }
1503}