use std::collections::BTreeMap;
use super::super::super::restore_checkpoint::{CheckpointError, CheckpointPositions};
use super::super::super::restore_checkpoint_history_key::{CheckpointHistoryKey, HistoryEntryKey};
use super::super::super::state::checkpoint_body::{CheckpointBodyDecoder, invalid};
use super::super::{FateOccurrenceKey, FateOccurrenceRouter, FateOccurrenceState};
impl FateOccurrenceRouter {
pub(in crate::server::participant::production) fn decode_checkpoint(
bytes: &[u8],
positions: &CheckpointPositions,
) -> Result<Self, CheckpointError> {
let mut decoder = CheckpointBodyDecoder::new(bytes);
let count = decoder.count("fate occurrences", 8)?;
let mut occurrences = BTreeMap::new();
let mut previous = None;
for _ in 0..count {
let key_bytes = decoder.bytes("fate occurrence key")?;
let history_key =
CheckpointHistoryKey::decode_canonical(key_bytes, positions.conversation_id)
.map_err(|source| CheckpointError::LiveState {
source: Box::new(source),
})?;
if history_key.encode_canonical() != key_bytes {
return Err(invalid("canonical fate occurrence key"));
}
let HistoryEntryKey::FateOccurrence {
participant_id,
binding_epoch,
} = history_key.entry()
else {
return Err(invalid("fate occurrence key kind"));
};
if participant_id >= positions.next_participant {
return Err(invalid("fate occurrence participant"));
}
let key = FateOccurrenceKey {
conversation_id: positions.conversation_id,
participant_id,
binding_epoch,
};
if previous.is_some_and(|last| last >= key) {
return Err(invalid("fate occurrence order"));
}
previous = Some(key);
let value_bytes = decoder.bytes("fate occurrence value")?;
let value = FateOccurrenceState::decode_history_value(
value_bytes,
key,
positions.covered_log_sequence,
)
.map_err(|source| CheckpointError::LiveState {
source: Box::new(source),
})?;
if value
.encode_history_value(key, positions.covered_log_sequence)
.map_err(|source| CheckpointError::LiveState {
source: Box::new(source),
})?
!= value_bytes
{
return Err(invalid("canonical fate occurrence value"));
}
occurrences.insert(key, value);
}
decoder.finish()?;
Ok(Self { occurrences })
}
}