use std::collections::HashMap;
use std::time::Instant;
use super::query::{SnapshotConfig, SnapshotQuery};
use super::snapshot::{DeltaSnapshot, ProfileSnapshot, SnapshotIndex};
use super::types::{RetentionTier, SnapshotError, SnapshotResult};
#[derive(Debug)]
pub struct IncrementalSnapshotStore {
config: SnapshotConfig,
index: Vec<SnapshotIndex>,
pub(super) keyframes: HashMap<usize, ProfileSnapshot>,
pub(super) deltas: HashMap<usize, DeltaSnapshot>,
total_raw_size: usize,
total_compressed_size: usize,
next_index: usize,
}
impl IncrementalSnapshotStore {
pub fn new(config: SnapshotConfig) -> Self {
Self {
config,
index: Vec::new(),
keyframes: HashMap::new(),
deltas: HashMap::new(),
total_raw_size: 0,
total_compressed_size: 0,
next_index: 0,
}
}
pub fn append(&mut self, mut snapshot: ProfileSnapshot) -> SnapshotResult<usize> {
snapshot.index = self.next_index;
snapshot.compute_checksum();
let raw_size = snapshot.size_bytes();
self.total_raw_size += raw_size;
let is_keyframe = self
.next_index
.is_multiple_of(self.config.keyframe_interval)
|| self.keyframes.is_empty();
let (compressed_size, is_delta, base_index) = if is_keyframe {
let size = snapshot.size_bytes();
self.keyframes.insert(self.next_index, snapshot.clone());
(size, false, None)
} else {
let keyframe_idx =
(self.next_index / self.config.keyframe_interval) * self.config.keyframe_interval;
if let Some(base) = self.keyframes.get(&keyframe_idx) {
let delta = DeltaSnapshot::from_diff(base, &snapshot);
let size = delta.size_bytes();
self.deltas.insert(self.next_index, delta);
(size, true, Some(keyframe_idx))
} else {
let size = snapshot.size_bytes();
self.keyframes.insert(self.next_index, snapshot.clone());
(size, false, None)
}
};
self.total_compressed_size += compressed_size;
self.index.push(SnapshotIndex {
index: self.next_index,
timestamp_ns: snapshot.timestamp_ns,
fingerprint: snapshot.workload_fingerprint.clone(),
tier: RetentionTier::Raw,
offset: 0, size_bytes: compressed_size,
is_delta,
base_index,
});
let idx = self.next_index;
self.next_index += 1;
Ok(idx)
}
pub fn get(&self, index: usize) -> SnapshotResult<ProfileSnapshot> {
if index >= self.next_index {
return Err(SnapshotError::IndexOutOfBounds {
index,
max: self.next_index.saturating_sub(1),
});
}
if let Some(snapshot) = self.keyframes.get(&index) {
return Ok(snapshot.clone());
}
if let Some(delta) = self.deltas.get(&index) {
if let Some(base) = self.keyframes.get(&delta.base_index) {
let reconstructed = delta.apply_to(base);
if self.config.verify_checksums && !reconstructed.verify_checksum() {
return Err(SnapshotError::Corrupt {
reason: "Checksum verification failed".to_string(),
});
}
return Ok(reconstructed);
}
}
Err(SnapshotError::NotFound { index })
}
pub fn query(&self, query: &SnapshotQuery) -> SnapshotResult<Vec<ProfileSnapshot>> {
let mut results = Vec::new();
let mut memory_used = 0;
for idx_entry in &self.index {
if !query.matches(idx_entry) {
continue;
}
if memory_used + idx_entry.size_bytes > self.config.max_query_memory_bytes {
break;
}
let snapshot = self.get(idx_entry.index)?;
if let Some(ref metric_name) = query.metric_name {
if !snapshot.metrics.contains_key(metric_name) {
continue;
}
}
memory_used += idx_entry.size_bytes;
results.push(snapshot);
if let Some(limit) = query.limit {
if results.len() >= limit {
break;
}
}
}
Ok(results)
}
pub fn compression_ratio(&self) -> f64 {
if self.total_raw_size == 0 {
1.0
} else {
self.total_compressed_size as f64 / self.total_raw_size as f64
}
}
pub fn count(&self) -> usize {
self.next_index
}
pub fn total_raw_size(&self) -> usize {
self.total_raw_size
}
pub fn total_compressed_size(&self) -> usize {
self.total_compressed_size
}
pub fn cleanup(&mut self, _now: Instant, reference_time: u64) {
let _raw_cutoff = reference_time.saturating_sub(self.config.raw_max_age.as_nanos() as u64);
let compressed_cutoff =
reference_time.saturating_sub(self.config.compressed_max_age.as_nanos() as u64);
self.index.retain(|idx| {
if idx.timestamp_ns < compressed_cutoff {
self.keyframes.remove(&idx.index);
self.deltas.remove(&idx.index);
return false;
}
true
});
}
pub fn config(&self) -> &SnapshotConfig {
&self.config
}
pub fn index(&self) -> &[SnapshotIndex] {
&self.index
}
}