liminal-server 0.24.10

//! Distinct framed keys and values for stream rows sharing one routed shard.

use std::collections::TryReserveError;

use liminal::durability::StoredEntry;

const KEY_PREFIX: &[u8] = b"liminal:routed-stream:v1:";
const HEAD_PREFIX: &[u8; 5] = b"LRSH\x01";
const ROW_PREFIX: &[u8; 5] = b"LRSR\x01";
const HEAD_BYTES: usize = 13;
const ROW_HEADER_BYTES: usize = 29;

#[derive(Debug, thiserror::Error)]
pub enum CodecError {
    #[error("routed stream identity is empty")]
    EmptyStream,
    #[error("routed stream identity exceeds its encoded width")]
    StreamTooLong,
    #[error("routed stream framing length overflow")]
    LengthOverflow,
    #[error("invalid routed stream key framing")]
    InvalidKey,
    #[error("invalid routed stream head framing")]
    InvalidHead,
    #[error("invalid routed stream row framing")]
    InvalidRow,
    #[error("routed stream row sequence mismatch: expected {expected}, actual {actual}")]
    SequenceMismatch { expected: u64, actual: u64 },
    #[error("routed stream framing allocation refused")]
    Allocation(#[source] TryReserveError),
}

#[derive(Debug, PartialEq, Eq)]
pub(super) enum StreamKey<'a> {
    Head { stream: &'a str },
    Row { stream: &'a str, sequence: u64 },
}

/// Encodes a head key without treating the route as a namespace.
///
/// # Errors
/// Refuses an empty or unrepresentable identity and allocation failure.
pub(super) fn head_key(stream: &str) -> Result<Vec<u8>, CodecError> {
    let mut key = key_prefix(stream, 1)?;
    key.push(0);
    Ok(key)
}

/// Encodes a row key using its full stream identity and API sequence.
///
/// # Errors
/// Refuses an empty or unrepresentable identity and allocation failure.
pub(super) fn row_key(stream: &str, sequence: u64) -> Result<Vec<u8>, CodecError> {
    let mut key = key_prefix(stream, 9)?;
    key.push(1);
    key.extend_from_slice(&sequence.to_be_bytes());
    Ok(key)
}

/// The window `[from, to)` holding the head and every row of one stream:
/// its key prefix followed by 0 (the head) or 1 (a row), bounded by 2.
///
/// # Errors
/// The stream identity's own refusal.
pub(super) fn stream_window(stream: &str) -> Result<(Vec<u8>, Vec<u8>), CodecError> {
    let mut from = key_prefix(stream, 1)?;
    let mut to = from.clone();
    from.push(0);
    to.push(2);
    Ok((from, to))
}

fn key_prefix(stream: &str, suffix: usize) -> Result<Vec<u8>, CodecError> {
    if stream.is_empty() {
        return Err(CodecError::EmptyStream);
    }
    let length = u32::try_from(stream.len()).map_err(|_| CodecError::StreamTooLong)?;
    let capacity = KEY_PREFIX
        .len()
        .checked_add(4)
        .and_then(|value| value.checked_add(stream.len()))
        .and_then(|value| value.checked_add(suffix))
        .ok_or(CodecError::LengthOverflow)?;
    let mut key = Vec::new();
    key.try_reserve_exact(capacity)
        .map_err(CodecError::Allocation)?;
    key.extend_from_slice(KEY_PREFIX);
    key.extend_from_slice(&length.to_be_bytes());
    key.extend_from_slice(stream.as_bytes());
    Ok(key)
}

/// The stream a routed row key names, or nothing for a head key or a key of
/// another shape (a checkpoint record, a receipt): what a barrier fixture
/// reads to cross the gate a native append of that stream crossed.
#[cfg(test)]
pub(super) fn row_stream(key: &[u8]) -> Option<&str> {
    match decode_key(key) {
        Ok(StreamKey::Row { stream, .. }) => Some(stream),
        Ok(StreamKey::Head { .. }) | Err(_) => None,
    }
}

/// Borrows the complete stream identity from a structurally valid key.
///
/// # Errors
/// Refuses unknown kinds, invalid identity bytes and incomplete or trailing data.
pub(super) fn decode_key(bytes: &[u8]) -> Result<StreamKey<'_>, CodecError> {
    if !bytes.starts_with(KEY_PREFIX) {
        return Err(CodecError::InvalidKey);
    }
    let length_end = KEY_PREFIX
        .len()
        .checked_add(4)
        .ok_or(CodecError::LengthOverflow)?;
    let length_bytes: [u8; 4] = bytes
        .get(KEY_PREFIX.len()..length_end)
        .ok_or(CodecError::InvalidKey)?
        .try_into()
        .map_err(|_| CodecError::InvalidKey)?;
    let length = usize::try_from(u32::from_be_bytes(length_bytes))
        .map_err(|_| CodecError::LengthOverflow)?;
    if length == 0 {
        return Err(CodecError::EmptyStream);
    }
    let stream_end = length_end
        .checked_add(length)
        .ok_or(CodecError::LengthOverflow)?;
    let stream = std::str::from_utf8(
        bytes
            .get(length_end..stream_end)
            .ok_or(CodecError::InvalidKey)?,
    )
    .map_err(|_| CodecError::InvalidKey)?;
    match bytes.get(stream_end..).ok_or(CodecError::InvalidKey)? {
        [0] => Ok(StreamKey::Head { stream }),
        [1, sequence @ ..] if sequence.len() == 8 => Ok(StreamKey::Row {
            stream,
            sequence: decode_integer(sequence, CodecError::InvalidKey)?,
        }),
        _ => Err(CodecError::InvalidKey),
    }
}

/// Encodes the stream counter without granting authority to an absent head.
#[must_use]
pub(super) fn encode_head(next_sequence: u64) -> [u8; HEAD_BYTES] {
    let mut bytes = [0; HEAD_BYTES];
    bytes[..5].copy_from_slice(HEAD_PREFIX);
    bytes[5..].copy_from_slice(&next_sequence.to_be_bytes());
    bytes
}

/// Decodes only the exact head envelope.
///
/// # Errors
/// Refuses a different version or incomplete or trailing bytes.
pub(super) fn decode_head(bytes: &[u8]) -> Result<u64, CodecError> {
    if bytes.len() != HEAD_BYTES || !bytes.starts_with(HEAD_PREFIX) {
        return Err(CodecError::InvalidHead);
    }
    decode_integer(
        bytes.get(5..).ok_or(CodecError::InvalidHead)?,
        CodecError::InvalidHead,
    )
}

/// Preserves the entry's API sequence, timestamp and payload.
///
/// # Errors
/// Refuses an unrepresentable length or allocation failure.
pub(super) fn encode_row(entry: &StoredEntry) -> Result<Vec<u8>, CodecError> {
    let length = u64::try_from(entry.payload.len()).map_err(|_| CodecError::LengthOverflow)?;
    let capacity = ROW_HEADER_BYTES
        .checked_add(entry.payload.len())
        .ok_or(CodecError::LengthOverflow)?;
    let mut bytes = Vec::new();
    bytes
        .try_reserve_exact(capacity)
        .map_err(CodecError::Allocation)?;
    bytes.extend_from_slice(ROW_PREFIX);
    bytes.extend_from_slice(&entry.sequence.to_be_bytes());
    bytes.extend_from_slice(&entry.timestamp.to_be_bytes());
    bytes.extend_from_slice(&length.to_be_bytes());
    bytes.extend_from_slice(&entry.payload);
    Ok(bytes)
}

/// Validates the row envelope against the sequence named by its key.
///
/// # Errors
/// Refuses inconsistent sequence, version or length and allocation failure.
pub(super) fn decode_row(bytes: &[u8], expected_sequence: u64) -> Result<StoredEntry, CodecError> {
    let (timestamp, bytes) = decode_row_payload(bytes, expected_sequence)?;
    let mut payload = Vec::new();
    payload
        .try_reserve_exact(bytes.len())
        .map_err(CodecError::Allocation)?;
    payload.extend_from_slice(bytes);
    Ok(StoredEntry {
        sequence: expected_sequence,
        timestamp,
        payload,
    })
}

/// Borrows the payload after validating the complete row frame.
///
/// # Errors
/// Refuses row identity, length or sequence drift without allocating.
pub(super) fn decode_row_payload(
    bytes: &[u8],
    expected_sequence: u64,
) -> Result<(u64, &[u8]), CodecError> {
    if bytes.len() < ROW_HEADER_BYTES || !bytes.starts_with(ROW_PREFIX) {
        return Err(CodecError::InvalidRow);
    }
    let sequence = decode_integer(&bytes[5..13], CodecError::InvalidRow)?;
    if sequence != expected_sequence {
        return Err(CodecError::SequenceMismatch {
            expected: expected_sequence,
            actual: sequence,
        });
    }
    let timestamp = decode_integer(&bytes[13..21], CodecError::InvalidRow)?;
    let length = usize::try_from(decode_integer(&bytes[21..29], CodecError::InvalidRow)?)
        .map_err(|_| CodecError::LengthOverflow)?;
    let end = ROW_HEADER_BYTES
        .checked_add(length)
        .ok_or(CodecError::LengthOverflow)?;
    if end != bytes.len() {
        return Err(CodecError::InvalidRow);
    }
    Ok((timestamp, &bytes[ROW_HEADER_BYTES..]))
}

fn decode_integer(bytes: &[u8], error: CodecError) -> Result<u64, CodecError> {
    let encoded = bytes.try_into().map_err(|_| error)?;
    Ok(u64::from_be_bytes(encoded))
}

#[cfg(test)]
mod tests;