haematite-cli 0.21.1

//! Bounded frames carry exact serialized bytes; only a verified whole receipt escapes.

use super::protocol::{WirePacket, WireRefusal};
use super::receipt_wire::WireReceipt;
use haematite::BatchReceipt;
use serde::{Deserialize, Serialize};
use serde_json::value::RawValue;
use sha2::{Digest, Sha256};
use std::collections::TryReserveError;
use std::fmt;

#[derive(Debug)]
pub enum ReceiptCodecError {
    IndexOverflow,
    Utf8 { cause: std::string::FromUtf8Error },
    Allocation { cause: TryReserveError },
    Encode { cause: serde_json::Error },
    Decode { cause: serde_json::Error },
    InvalidId,
    FrameTooSmall { offered: u32, minimum: usize },
    FrameLimit { size: usize, limit: u32 },
    Identity,
    Sequence { expected: u64, actual: u64 },
    ExtentOverflow,
    Extent { expected: u64, actual: u64 },
    Count { expected: u64, actual: u64 },
    Digest,
    Hex,
    Incomplete,
    ConnectionClosed,
    Finished,
}

impl ReceiptCodecError {
    pub(super) const fn allocation(cause: TryReserveError) -> Self {
        Self::Allocation { cause }
    }
}
impl fmt::Display for ReceiptCodecError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::IndexOverflow => f.write_str("receipt_index_overflow"),
            Self::Utf8 { cause } => write!(f, "receipt_utf8_invalid: {cause}"),
            Self::Allocation { cause } => write!(f, "receipt_allocation_failed: {cause}"),
            Self::Encode { cause } => write!(f, "receipt_encode_failed: {cause}"),
            Self::Decode { cause } => write!(f, "receipt_decode_failed: {cause}"),
            Self::InvalidId => f.write_str("receipt_invalid_request_id"),
            Self::FrameTooSmall { offered, minimum } => write!(
                f,
                "receipt_frame_too_small: offered {offered}, minimum {minimum}"
            ),
            Self::FrameLimit { size, limit } => {
                write!(f, "receipt_frame_limit: size {size}, limit {limit}")
            }
            Self::Identity => f.write_str("receipt_request_id_mismatch"),
            Self::Sequence { expected, actual } => write!(
                f,
                "receipt_fragment_sequence: expected {expected}, actual {actual}"
            ),
            Self::ExtentOverflow => f.write_str("receipt_extent_overflow"),
            Self::Extent { expected, actual } => write!(
                f,
                "receipt_extent_mismatch: expected {expected}, actual {actual}"
            ),
            Self::Count { expected, actual } => write!(
                f,
                "receipt_count_mismatch: expected {expected}, actual {actual}"
            ),
            Self::Digest => f.write_str("receipt_digest_mismatch"),
            Self::Hex => f.write_str("receipt_hex_invalid"),
            Self::Incomplete => f.write_str("receipt_incomplete"),
            Self::ConnectionClosed => f.write_str("receipt_connection_closed"),
            Self::Finished => f.write_str("receipt_already_finished"),
        }
    }
}
impl std::error::Error for ReceiptCodecError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::Allocation { cause } => Some(cause),
            Self::Utf8 { cause } => Some(cause),
            Self::Encode { cause } | Self::Decode { cause } => Some(cause),
            _ => None,
        }
    }
}

pub(super) fn reserve_bytes(
    bytes: &mut Vec<u8>,
    additional: usize,
) -> Result<(), ReceiptCodecError> {
    bytes
        .len()
        .checked_add(additional)
        .ok_or(ReceiptCodecError::ExtentOverflow)?;
    bytes
        .try_reserve_exact(additional)
        .map_err(ReceiptCodecError::allocation)
}

pub(super) fn encode_hex(bytes: &[u8]) -> Result<String, ReceiptCodecError> {
    const DIGITS: &[u8; 16] = b"0123456789abcdef";
    let length = bytes
        .len()
        .checked_mul(2)
        .ok_or(ReceiptCodecError::ExtentOverflow)?;
    let mut hex = String::new();
    hex.try_reserve_exact(length)
        .map_err(ReceiptCodecError::allocation)?;
    for byte in bytes {
        hex.push(char::from(DIGITS[usize::from(byte >> 4)]));
        hex.push(char::from(DIGITS[usize::from(byte & 15)]));
    }
    Ok(hex)
}

fn decode_hex(hex: &str) -> Result<Vec<u8>, ReceiptCodecError> {
    if !hex.len().is_multiple_of(2) {
        return Err(ReceiptCodecError::Hex);
    }
    let mut result = Vec::new();
    reserve_bytes(&mut result, hex.len() / 2)?;
    for pair in hex.as_bytes().as_chunks::<2>().0 {
        let high = char::from(pair[0])
            .to_digit(16)
            .ok_or(ReceiptCodecError::Hex)?;
        let low = char::from(pair[1])
            .to_digit(16)
            .ok_or(ReceiptCodecError::Hex)?;
        result.push(u8::try_from(high * 16 + low).map_err(|_| ReceiptCodecError::Hex)?);
    }
    Ok(result)
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct FragmentEnvelope {
    pub id: String,
    #[serde(rename = "receipt")]
    pub part: FragmentPart,
}
#[derive(Debug, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum FragmentPart {
    Chunk {
        sequence: u64,
        hex: String,
    },
    End {
        fragments: u64,
        outcomes: u64,
        bytes: u64,
        digest: String,
    },
}

fn encode(envelope: &FragmentEnvelope) -> Result<Vec<u8>, ReceiptCodecError> {
    serde_json::to_vec(envelope).map_err(|cause| ReceiptCodecError::Encode { cause })
}
const fn check_id(id: &str) -> Result<(), ReceiptCodecError> {
    if id.is_empty() || id.len() > 512 {
        Err(ReceiptCodecError::InvalidId)
    } else {
        Ok(())
    }
}

/// Derive the mandatory envelope floor from the actual serializer and maximum valid identity.
///
/// # Errors
///
/// Propagates envelope encoding failure.
pub fn minimum_cap() -> Result<usize, ReceiptCodecError> {
    let id = "\0".repeat(512);
    let chunk = encode(&FragmentEnvelope {
        id: id.clone(),
        part: FragmentPart::Chunk {
            sequence: u64::MAX,
            hex: "ff".into(),
        },
    })?;
    let end = encode(&FragmentEnvelope {
        id,
        part: FragmentPart::End {
            fragments: u64::MAX,
            outcomes: u64::MAX,
            bytes: u64::MAX,
            digest: "f".repeat(64),
        },
    })?;
    Ok(chunk.len().max(end.len()))
}
fn check_cap(cap: u32) -> Result<(), ReceiptCodecError> {
    let minimum = minimum_cap()?;
    if u64::from(cap) < u64::try_from(minimum).map_err(|_| ReceiptCodecError::ExtentOverflow)? {
        Err(ReceiptCodecError::FrameTooSmall {
            offered: cap,
            minimum,
        })
    } else {
        Ok(())
    }
}

#[derive(Debug)]
pub struct EncodedReceipt {
    bytes: Vec<u8>,
    outcomes: u64,
    digest: String,
}
impl EncodedReceipt {
    /// Project and serialize the complete receipt exactly once, hashing those same bytes.
    ///
    /// # Errors
    ///
    /// Propagates projection, serialization, index and allocation failures.
    pub fn project(
        receipt: &BatchReceipt,
        refusal: Option<WireRefusal>,
    ) -> Result<Self, ReceiptCodecError> {
        Self::seal(&WirePacket::Receipt {
            receipt: Box::new(WireReceipt::project(receipt)?),
            refusal,
        })
    }

    pub(crate) fn refusal(
        receipt: Option<&BatchReceipt>,
        refusal: WireRefusal,
    ) -> Result<Self, ReceiptCodecError> {
        let packet = match receipt {
            Some(receipt) => return Self::project(receipt, Some(refusal)),
            None => WirePacket::Refusal { refusal },
        };
        Self::seal(&packet)
    }

    fn seal(packet: &WirePacket) -> Result<Self, ReceiptCodecError> {
        let outcomes = u64::try_from(packet.receipt().map_or(0, |receipt| receipt.outcomes.len()))
            .map_err(|_| ReceiptCodecError::IndexOverflow)?;
        let bytes =
            serde_json::to_vec(packet).map_err(|cause| ReceiptCodecError::Encode { cause })?;
        let digest = encode_hex(&Sha256::digest(&bytes))?;
        Ok(Self {
            bytes,
            outcomes,
            digest,
        })
    }
    #[must_use]
    pub fn bytes(&self) -> &[u8] {
        &self.bytes
    }

    /// Produce ordered ordinary-cap envelopes without changing the serialized receipt.
    ///
    /// # Errors
    ///
    /// Refuses invalid identity, insufficient cap, overflow, allocation or encoding failure.
    pub fn frames(&self, id: &str, cap: u32) -> Result<Vec<Vec<u8>>, ReceiptCodecError> {
        check_id(id)?;
        check_cap(cap)?;
        let overhead = encode(&FragmentEnvelope {
            id: id.to_owned(),
            part: FragmentPart::Chunk {
                sequence: u64::MAX,
                hex: String::new(),
            },
        })?
        .len();
        let limit = usize::try_from(cap).map_err(|_| ReceiptCodecError::ExtentOverflow)?;
        let chunk_size = limit
            .checked_sub(overhead)
            .ok_or(ReceiptCodecError::ExtentOverflow)?
            / 2;
        if chunk_size == 0 {
            return Err(ReceiptCodecError::Incomplete);
        }
        let count = self.bytes.len().div_ceil(chunk_size);
        let mut frames = Vec::new();
        frames
            .try_reserve_exact(
                count
                    .checked_add(1)
                    .ok_or(ReceiptCodecError::ExtentOverflow)?,
            )
            .map_err(ReceiptCodecError::allocation)?;
        for (sequence, chunk) in self.bytes.chunks(chunk_size).enumerate() {
            let sequence =
                u64::try_from(sequence).map_err(|_| ReceiptCodecError::ExtentOverflow)?;
            let frame = encode(&FragmentEnvelope {
                id: id.to_owned(),
                part: FragmentPart::Chunk {
                    sequence,
                    hex: encode_hex(chunk)?,
                },
            })?;
            if frame.len() > limit {
                return Err(ReceiptCodecError::FrameLimit {
                    size: frame.len(),
                    limit: cap,
                });
            }
            frames.push(frame);
        }
        let final_frame = encode(&FragmentEnvelope {
            id: id.to_owned(),
            part: FragmentPart::End {
                fragments: u64::try_from(count).map_err(|_| ReceiptCodecError::ExtentOverflow)?,
                outcomes: self.outcomes,
                bytes: u64::try_from(self.bytes.len())
                    .map_err(|_| ReceiptCodecError::ExtentOverflow)?,
                digest: self.digest.clone(),
            },
        })?;
        if final_frame.len() > limit {
            return Err(ReceiptCodecError::FrameLimit {
                size: final_frame.len(),
                limit: cap,
            });
        }
        frames.push(final_frame);
        Ok(frames)
    }
}

/// A complete typed receipt owns the exact bytes whose digest and extent were verified.
///
/// `bytes` are those bytes as they arrived, whitespace included; `raw` is the
/// same JSON value as text, without the whitespace around it that a
/// [`RawValue`] does not keep.
#[derive(Debug)]
pub struct VerifiedReceipt {
    packet: WirePacket,
    bytes: Vec<u8>,
    raw: Box<RawValue>,
}
impl VerifiedReceipt {
    #[must_use]
    pub const fn packet(&self) -> &WirePacket {
        &self.packet
    }
    #[must_use]
    pub fn receipt(&self) -> Option<&WireReceipt> {
        self.packet.receipt()
    }
    #[must_use]
    pub const fn refusal(&self) -> Option<&WireRefusal> {
        self.packet.refusal()
    }
    #[must_use]
    pub fn raw(&self) -> &RawValue {
        &self.raw
    }
    #[must_use]
    pub fn bytes(&self) -> &[u8] {
        &self.bytes
    }
}

pub struct Assembler {
    id: String,
    cap: u32,
    next: u64,
    bytes: Vec<u8>,
    digest: Sha256,
    failed: bool,
    finished: bool,
}
impl Assembler {
    /// Own one complete response under its actual correlation identity and frame cap.
    ///
    /// # Errors
    ///
    /// Refuses invalid identity or an insufficient envelope cap.
    pub fn new(id: &str, cap: u32) -> Result<Self, ReceiptCodecError> {
        check_id(id)?;
        check_cap(cap)?;
        Ok(Self {
            id: id.to_owned(),
            cap,
            next: 0,
            bytes: Vec::new(),
            digest: Sha256::new(),
            failed: false,
            finished: false,
        })
    }
    /// Admit a frame, returning a receipt only after whole-byte and outcome verification.
    ///
    /// # Errors
    ///
    /// Any malformed, missing, repeated, unassociated or unverifiable part poisons assembly.
    pub fn accept(&mut self, frame: &[u8]) -> Result<Option<VerifiedReceipt>, ReceiptCodecError> {
        if self.finished {
            return Err(ReceiptCodecError::Finished);
        }
        if self.failed {
            return Err(ReceiptCodecError::Incomplete);
        }
        let result = self.accept_part(frame);
        if result.is_err() {
            self.failed = true;
            self.bytes.clear();
        }
        result
    }
    fn accept_part(&mut self, frame: &[u8]) -> Result<Option<VerifiedReceipt>, ReceiptCodecError> {
        let limit = usize::try_from(self.cap).map_err(|_| ReceiptCodecError::ExtentOverflow)?;
        if frame.len() > limit {
            return Err(ReceiptCodecError::FrameLimit {
                size: frame.len(),
                limit: self.cap,
            });
        }
        let envelope: FragmentEnvelope =
            serde_json::from_slice(frame).map_err(|cause| ReceiptCodecError::Decode { cause })?;
        if envelope.id != self.id {
            return Err(ReceiptCodecError::Identity);
        }
        match envelope.part {
            FragmentPart::Chunk { sequence, hex } => {
                if sequence != self.next {
                    return Err(ReceiptCodecError::Sequence {
                        expected: self.next,
                        actual: sequence,
                    });
                }
                let bytes = decode_hex(&hex)?;
                if bytes.is_empty() {
                    return Err(ReceiptCodecError::Incomplete);
                }
                let next = self
                    .next
                    .checked_add(1)
                    .ok_or(ReceiptCodecError::ExtentOverflow)?;
                reserve_bytes(&mut self.bytes, bytes.len())?;
                self.bytes.extend_from_slice(&bytes);
                self.digest.update(&bytes);
                self.next = next;
                Ok(None)
            }
            FragmentPart::End {
                fragments,
                outcomes,
                bytes,
                digest,
            } => {
                if self.next == 0 {
                    return Err(ReceiptCodecError::Incomplete);
                }
                if fragments != self.next {
                    return Err(ReceiptCodecError::Sequence {
                        expected: self.next,
                        actual: fragments,
                    });
                }
                let actual = u64::try_from(self.bytes.len())
                    .map_err(|_| ReceiptCodecError::ExtentOverflow)?;
                if bytes != actual {
                    return Err(ReceiptCodecError::Extent {
                        expected: bytes,
                        actual,
                    });
                }
                let expected_digest = decode_hex(&digest)?;
                let actual_digest = self.digest.clone().finalize();
                if expected_digest.as_slice() != &actual_digest[..] {
                    return Err(ReceiptCodecError::Digest);
                }
                let packet: WirePacket = serde_json::from_slice(&self.bytes)
                    .map_err(|cause| ReceiptCodecError::Decode { cause })?;
                let actual =
                    u64::try_from(packet.receipt().map_or(0, |receipt| receipt.outcomes.len()))
                        .map_err(|_| ReceiptCodecError::ExtentOverflow)?;
                if outcomes != actual {
                    return Err(ReceiptCodecError::Count {
                        expected: outcomes,
                        actual,
                    });
                }
                let exact = std::mem::take(&mut self.bytes);
                let mut text = Vec::new();
                reserve_bytes(&mut text, exact.len())?;
                text.extend_from_slice(&exact);
                let text =
                    String::from_utf8(text).map_err(|cause| ReceiptCodecError::Utf8 { cause })?;
                let raw = RawValue::from_string(text)
                    .map_err(|cause| ReceiptCodecError::Decode { cause })?;
                self.finished = true;
                Ok(Some(VerifiedReceipt {
                    packet,
                    bytes: exact,
                    raw,
                }))
            }
        }
    }
    /// Record an ended connection without ever exposing an unfinished receipt.
    ///
    /// # Errors
    ///
    /// Returns a named connection closure unless a whole receipt was already delivered.
    pub fn connection_closed(&mut self) -> Result<(), ReceiptCodecError> {
        if self.finished {
            return Ok(());
        }
        self.failed = true;
        self.bytes.clear();
        Err(ReceiptCodecError::ConnectionClosed)
    }
}