use frame_core::component::ComponentId;
use haematite::{BranchKind, Hash};
use crate::error::StateError;
use crate::types::{
AbandonedWork, ComponentSchema, CrossComponentRef, EntityId, MergePolicy, MergeRecord,
MetaRecord, MetaState,
};
pub(crate) const CROSS_COMPONENT_REF_WIDTH: usize = 65;
pub(crate) const ENTITY_REGION: u8 = 0;
pub(crate) const MERGE_REGION: u8 = 1;
pub(crate) const META_REGION: u8 = 0;
pub(crate) const META_ARCHIVE_REGION: u8 = 1;
const RECORD_VERSION: u8 = 1;
pub(crate) fn entity_key(id: crate::EntityId) -> Vec<u8> {
let mut key = Vec::with_capacity(33);
key.push(ENTITY_REGION);
key.extend_from_slice(id.as_bytes());
key
}
pub(crate) fn entity_bounds() -> ([u8; 1], [u8; 1]) {
([ENTITY_REGION], [MERGE_REGION])
}
pub(crate) fn encode_cross_component_ref(
reference: CrossComponentRef,
) -> [u8; CROSS_COMPONENT_REF_WIDTH] {
let mut bytes = [0_u8; CROSS_COMPONENT_REF_WIDTH];
bytes[0] = RECORD_VERSION;
bytes[1..33].copy_from_slice(reference.target.as_bytes());
bytes[33..].copy_from_slice(reference.entity.as_bytes());
bytes
}
pub(crate) fn decode_cross_component_ref(bytes: &[u8]) -> Result<CrossComponentRef, StateError> {
if bytes.len() != CROSS_COMPONENT_REF_WIDTH {
return Err(corrupt(&format!(
"cross-component reference must be {CROSS_COMPONENT_REF_WIDTH} bytes, got {}",
bytes.len()
)));
}
if bytes[0] != RECORD_VERSION {
return Err(corrupt(&format!(
"cross-component reference has unsupported version {}",
bytes[0]
)));
}
let target = component_from_bytes(&bytes[1..33], "cross-component reference target")?;
let mut entity = [0_u8; 32];
entity.copy_from_slice(&bytes[33..]);
Ok(CrossComponentRef::new(target, EntityId::from_bytes(entity)))
}
pub(crate) fn merge_key(sequence: u64) -> Vec<u8> {
let mut key = Vec::with_capacity(9);
key.push(MERGE_REGION);
key.extend_from_slice(&sequence.to_be_bytes());
key
}
pub(crate) fn merge_bounds() -> ([u8; 1], [u8; 1]) {
([MERGE_REGION], [MERGE_REGION + 1])
}
pub(crate) fn meta_key(component: ComponentId) -> Vec<u8> {
let mut key = Vec::with_capacity(33);
key.push(META_REGION);
key.extend_from_slice(component.as_bytes());
key
}
pub(crate) fn meta_bounds() -> ([u8; 1], [u8; 1]) {
([META_REGION], [META_REGION + 1])
}
pub(crate) fn archive_meta_key(component: ComponentId, generation: u64) -> Vec<u8> {
let mut key = Vec::with_capacity(41);
key.push(META_ARCHIVE_REGION);
key.extend_from_slice(component.as_bytes());
key.extend_from_slice(&generation.to_be_bytes());
key
}
pub(crate) fn component_from_meta_key(key: &[u8]) -> Result<ComponentId, StateError> {
if key.len() != 33 || key[0] != META_REGION {
return Err(corrupt(
"meta key is not region byte plus 32-byte component id",
));
}
component_from_bytes(&key[1..], "meta key component id")
}
fn component_from_bytes(bytes: &[u8], field: &str) -> Result<ComponentId, StateError> {
if bytes.len() != 32 {
return Err(corrupt(&format!(
"{field} must be 32 bytes, got {}",
bytes.len()
)));
}
let mut text = String::with_capacity(64);
for byte in bytes {
use std::fmt::Write as _;
write!(text, "{byte:02x}").map_err(|error| corrupt(&format!("{field}: {error}")))?;
}
text.parse()
.map_err(|error: frame_core::component::ComponentIdParseError| {
corrupt(&format!("{field}: {error}"))
})
}
pub(crate) fn encode_meta(record: &MetaRecord) -> Result<Vec<u8>, StateError> {
let mut bytes = vec![RECORD_VERSION];
bytes.extend_from_slice(&record.incarnation.to_be_bytes());
encode_policy(&record.schema.merge_policy, &mut bytes)?;
match &record.state {
MetaState::Installing => bytes.push(0),
MetaState::Active => bytes.push(1),
MetaState::Archiving {
generation,
abandoned_work,
} => {
bytes.push(2);
encode_archive_state(*generation, abandoned_work, &mut bytes)?;
}
MetaState::Archived {
generation,
abandoned_work,
} => {
bytes.push(3);
encode_archive_state(*generation, abandoned_work, &mut bytes)?;
}
}
Ok(bytes)
}
pub(crate) fn decode_meta(component: ComponentId, bytes: &[u8]) -> Result<MetaRecord, StateError> {
let mut decoder = Decoder::new(bytes);
decoder.version()?;
let incarnation = decoder.u64()?;
let schema = ComponentSchema {
merge_policy: decode_policy(&mut decoder)?,
};
let state = match decoder.u8()? {
0 => MetaState::Installing,
1 => MetaState::Active,
2 => {
let (generation, abandoned_work) = decode_archive_state(&mut decoder)?;
MetaState::Archiving {
generation,
abandoned_work,
}
}
3 => {
let (generation, abandoned_work) = decode_archive_state(&mut decoder)?;
MetaState::Archived {
generation,
abandoned_work,
}
}
tag => return Err(corrupt(&format!("unknown meta state tag {tag}"))),
};
decoder.finish()?;
Ok(MetaRecord {
component,
incarnation,
schema,
state,
})
}
pub(crate) fn encode_merge(record: &MergeRecord) -> Result<Vec<u8>, StateError> {
let mut bytes = vec![RECORD_VERSION];
bytes.extend_from_slice(&record.sequence.to_be_bytes());
encode_policy(&record.policy, &mut bytes)?;
encode_string(&record.source_branch, &mut bytes)?;
encode_string(&record.destination_branch, &mut bytes)?;
encode_hash(record.source_root, &mut bytes);
encode_hash(record.destination_root_before, &mut bytes);
encode_hash(record.merged_content_root, &mut bytes);
Ok(bytes)
}
pub(crate) fn decode_merge(bytes: &[u8]) -> Result<MergeRecord, StateError> {
let mut decoder = Decoder::new(bytes);
decoder.version()?;
let record = MergeRecord {
sequence: decoder.u64()?,
policy: decode_policy(&mut decoder)?,
source_branch: decoder.string()?,
destination_branch: decoder.string()?,
source_root: decoder.hash()?,
destination_root_before: decoder.hash()?,
merged_content_root: decoder.hash()?,
};
decoder.finish()?;
Ok(record)
}
fn encode_archive_state(
generation: u64,
abandoned: &[AbandonedWork],
bytes: &mut Vec<u8>,
) -> Result<(), StateError> {
bytes.extend_from_slice(&generation.to_be_bytes());
encode_len(abandoned.len(), bytes)?;
for work in abandoned {
encode_string(&work.branch, bytes)?;
bytes.push(match work.kind {
BranchKind::Namespace => 0,
BranchKind::Work => 1,
});
encode_hash(work.last_root, bytes);
}
Ok(())
}
fn decode_archive_state(
decoder: &mut Decoder<'_>,
) -> Result<(u64, Vec<AbandonedWork>), StateError> {
let generation = decoder.u64()?;
let count = decoder.len()?;
let mut abandoned = Vec::with_capacity(count);
for _ in 0..count {
let branch = decoder.string()?;
let kind = match decoder.u8()? {
0 => BranchKind::Namespace,
1 => BranchKind::Work,
tag => return Err(corrupt(&format!("unknown branch kind tag {tag}"))),
};
abandoned.push(AbandonedWork {
branch,
kind,
last_root: decoder.hash()?,
});
}
Ok((generation, abandoned))
}
fn encode_policy(policy: &MergePolicy, bytes: &mut Vec<u8>) -> Result<(), StateError> {
match policy {
MergePolicy::Lww => bytes.push(0),
MergePolicy::VectorClock => bytes.push(1),
MergePolicy::Custom(name) => {
bytes.push(2);
encode_string(name, bytes)?;
}
}
Ok(())
}
fn decode_policy(decoder: &mut Decoder<'_>) -> Result<MergePolicy, StateError> {
match decoder.u8()? {
0 => Ok(MergePolicy::Lww),
1 => Ok(MergePolicy::VectorClock),
2 => Ok(MergePolicy::Custom(decoder.string()?)),
tag => Err(corrupt(&format!("unknown merge policy tag {tag}"))),
}
}
fn encode_hash(hash: Hash, bytes: &mut Vec<u8>) {
bytes.extend_from_slice(hash.as_bytes());
}
fn encode_string(value: &str, bytes: &mut Vec<u8>) -> Result<(), StateError> {
encode_len(value.len(), bytes)?;
bytes.extend_from_slice(value.as_bytes());
Ok(())
}
fn encode_len(length: usize, bytes: &mut Vec<u8>) -> Result<(), StateError> {
let length = u32::try_from(length).map_err(|_| corrupt("record field exceeds u32 length"))?;
bytes.extend_from_slice(&length.to_be_bytes());
Ok(())
}
fn corrupt(detail: &str) -> StateError {
StateError::CorruptRecord {
detail: detail.to_owned(),
}
}
struct Decoder<'a> {
remaining: &'a [u8],
}
impl<'a> Decoder<'a> {
const fn new(bytes: &'a [u8]) -> Self {
Self { remaining: bytes }
}
fn version(&mut self) -> Result<(), StateError> {
let version = self.u8()?;
if version == RECORD_VERSION {
Ok(())
} else {
Err(corrupt(&format!("unsupported record version {version}")))
}
}
fn u8(&mut self) -> Result<u8, StateError> {
let Some((&value, remaining)) = self.remaining.split_first() else {
return Err(corrupt("record ended before u8"));
};
self.remaining = remaining;
Ok(value)
}
fn u64(&mut self) -> Result<u64, StateError> {
let bytes = self.take(8)?;
let array = <[u8; 8]>::try_from(bytes).map_err(|_| corrupt("invalid u64 width"))?;
Ok(u64::from_be_bytes(array))
}
fn len(&mut self) -> Result<usize, StateError> {
let bytes = self.take(4)?;
let array = <[u8; 4]>::try_from(bytes).map_err(|_| corrupt("invalid u32 width"))?;
usize::try_from(u32::from_be_bytes(array))
.map_err(|_| corrupt("record length does not fit usize"))
}
fn string(&mut self) -> Result<String, StateError> {
let length = self.len()?;
let bytes = self.take(length)?;
String::from_utf8(bytes.to_vec()).map_err(|error| corrupt(&error.to_string()))
}
fn hash(&mut self) -> Result<Hash, StateError> {
let bytes = self.take(32)?;
let array = <[u8; 32]>::try_from(bytes).map_err(|_| corrupt("invalid hash width"))?;
Ok(Hash::from_bytes(array))
}
fn take(&mut self, length: usize) -> Result<&'a [u8], StateError> {
if self.remaining.len() < length {
return Err(corrupt("record ended before declared field length"));
}
let (value, remaining) = self.remaining.split_at(length);
self.remaining = remaining;
Ok(value)
}
fn finish(self) -> Result<(), StateError> {
if self.remaining.is_empty() {
Ok(())
} else {
Err(corrupt("record contains trailing bytes"))
}
}
}