liminal-rs 0.16.3

use crate::durability::DurabilityError;

use super::{ClaimEvidence, DedupEntry};

const ENTRY_MAGIC: [u8; 4] = *b"LDE1";
const EVIDENCE_MAGIC: [u8; 4] = *b"LDE2";
const TOMBSTONE_MAGIC: [u8; 4] = *b"LDT1";
const JOIN_MAGIC: [u8; 4] = *b"LDX1";
const JOIN_CLAIM_MAGIC: [u8; 4] = *b"LDX2";
const ABSENT: u8 = 0;
const PRESENT: u8 = 1;

#[derive(Clone)]
pub(super) enum DedupRecord {
    Active(DedupEntry),
    Tombstone {
        idempotency_key: String,
        timestamp_millis: u64,
    },
}

impl DedupRecord {
    pub(super) const fn tombstone(idempotency_key: String, timestamp_millis: u64) -> Self {
        Self::Tombstone {
            idempotency_key,
            timestamp_millis,
        }
    }

    pub(super) fn idempotency_key(&self) -> &str {
        match self {
            Self::Active(entry) => entry.idempotency_key(),
            Self::Tombstone {
                idempotency_key, ..
            } => idempotency_key,
        }
    }

    pub(super) fn into_active(self) -> Option<DedupEntry> {
        match self {
            Self::Active(entry) => Some(entry),
            Self::Tombstone { .. } => None,
        }
    }

    pub(super) fn serialize(&self) -> Result<Vec<u8>, DurabilityError> {
        let mut bytes = Vec::new();
        match self {
            Self::Active(entry) => {
                if let Some(evidence) = entry.evidence() {
                    if entry.receipt().is_some() {
                        return Err(DurabilityError::EnvelopeError(
                            "evidenced in-flight claim cannot carry a completed receipt".to_owned(),
                        ));
                    }
                    bytes.extend_from_slice(&EVIDENCE_MAGIC);
                    write_string(&mut bytes, entry.idempotency_key())?;
                    write_u64(&mut bytes, entry.timestamp_millis());
                    write_string(&mut bytes, evidence.channel())?;
                    write_len(&mut bytes, evidence.positions().len())?;
                    for position in evidence.positions() {
                        write_u64(&mut bytes, *position);
                    }
                    return Ok(bytes);
                }
                bytes.extend_from_slice(&ENTRY_MAGIC);
                write_string(&mut bytes, entry.idempotency_key())?;
                write_optional_bytes(&mut bytes, entry.receipt())?;
                write_u64(&mut bytes, entry.timestamp_millis());
            }
            Self::Tombstone {
                idempotency_key,
                timestamp_millis,
            } => {
                bytes.extend_from_slice(&TOMBSTONE_MAGIC);
                write_string(&mut bytes, idempotency_key)?;
                write_u64(&mut bytes, *timestamp_millis);
            }
        }
        Ok(bytes)
    }

    pub(super) fn deserialize(bytes: &[u8]) -> Result<Self, DurabilityError> {
        let mut reader = DedupReader::new(bytes);
        let magic = reader.read_magic()?;
        let record = if magic == ENTRY_MAGIC {
            let idempotency_key = reader.read_string()?;
            let receipt = reader.read_optional_bytes()?.map(<[u8]>::to_vec);
            let timestamp_millis = reader.read_u64()?;
            Self::Active(DedupEntry::new(idempotency_key, receipt, timestamp_millis))
        } else if magic == EVIDENCE_MAGIC {
            let idempotency_key = reader.read_string()?;
            let timestamp_millis = reader.read_u64()?;
            let channel = reader.read_string()?;
            let count = reader.read_len()?;
            let mut positions = Vec::new();
            for _ in 0..count {
                positions.push(reader.read_u64()?);
            }
            let mut entry = DedupEntry::new(idempotency_key, None, timestamp_millis);
            entry.evidence = Some(ClaimEvidence::new(channel, positions));
            Self::Active(entry)
        } else if magic == TOMBSTONE_MAGIC {
            Self::tombstone(reader.read_string()?, reader.read_u64()?)
        } else {
            return Err(DurabilityError::EnvelopeError(
                "invalid dedup entry magic".to_owned(),
            ));
        };
        reader.finish()?;
        Ok(record)
    }
}

/// Encodes the in-flight index row naming `idempotency_key` and `claim`, the
/// sequence its claim takes in the key's own stream (`LDX2`).
pub(super) fn encode_join(idempotency_key: &str, claim: u64) -> Result<Vec<u8>, DurabilityError> {
    let mut bytes = JOIN_CLAIM_MAGIC.to_vec();
    write_string(&mut bytes, idempotency_key)?;
    write_u64(&mut bytes, claim);
    Ok(bytes)
}

/// Encodes an index row as a binary from before the claim sequence wrote it
/// (`LDX1`): the key alone.
#[cfg(test)]
pub(super) fn encode_join_without_claim(idempotency_key: &str) -> Result<Vec<u8>, DurabilityError> {
    let mut bytes = JOIN_MAGIC.to_vec();
    write_string(&mut bytes, idempotency_key)?;
    Ok(bytes)
}

/// Decodes an in-flight index row into the key it names and, for a row that
/// carries one (`LDX2`), the sequence of the key's claim. A row written by an
/// older binary (`LDX1`) answers `None`.
pub(super) fn decode_join(bytes: &[u8]) -> Result<(String, Option<u64>), DurabilityError> {
    let mut reader = DedupReader::new(bytes);
    let magic = reader.read_magic()?;
    let carries_claim = if magic == JOIN_CLAIM_MAGIC {
        true
    } else if magic == JOIN_MAGIC {
        false
    } else {
        return Err(DurabilityError::EnvelopeError(
            "invalid in-flight index row magic".to_owned(),
        ));
    };
    let idempotency_key = reader.read_string()?;
    let claim = if carries_claim {
        Some(reader.read_u64()?)
    } else {
        None
    };
    reader.finish()?;
    Ok((idempotency_key, claim))
}

fn write_optional_bytes(bytes: &mut Vec<u8>, value: Option<&[u8]>) -> Result<(), DurabilityError> {
    match value {
        Some(value) => {
            bytes.push(PRESENT);
            write_bytes(bytes, value)?;
        }
        None => bytes.push(ABSENT),
    }
    Ok(())
}

fn write_string(bytes: &mut Vec<u8>, value: &str) -> Result<(), DurabilityError> {
    write_bytes(bytes, value.as_bytes())
}

fn write_bytes(bytes: &mut Vec<u8>, value: &[u8]) -> Result<(), DurabilityError> {
    write_len(bytes, value.len())?;
    bytes.extend_from_slice(value);
    Ok(())
}

fn write_len(bytes: &mut Vec<u8>, len: usize) -> Result<(), DurabilityError> {
    let len = u64::try_from(len).map_err(|error| {
        DurabilityError::EnvelopeError(format!("field length cannot be encoded: {error}"))
    })?;
    write_u64(bytes, len);
    Ok(())
}

fn write_u64(bytes: &mut Vec<u8>, value: u64) {
    bytes.extend_from_slice(&value.to_be_bytes());
}

struct DedupReader<'a> {
    bytes: &'a [u8],
    cursor: usize,
}

impl<'a> DedupReader<'a> {
    const fn new(bytes: &'a [u8]) -> Self {
        Self { bytes, cursor: 0 }
    }

    fn read_magic(&mut self) -> Result<[u8; 4], DurabilityError> {
        let bytes = self.read_exact(4)?;
        let mut magic = [0_u8; 4];
        magic.copy_from_slice(bytes);
        Ok(magic)
    }

    fn read_optional_bytes(&mut self) -> Result<Option<&'a [u8]>, DurabilityError> {
        match self.read_u8()? {
            ABSENT => Ok(None),
            PRESENT => self.read_bytes().map(Some),
            value => Err(DurabilityError::EnvelopeError(format!(
                "invalid option marker {value}"
            ))),
        }
    }

    fn read_string(&mut self) -> Result<String, DurabilityError> {
        let bytes = self.read_bytes()?;
        String::from_utf8(bytes.to_vec()).map_err(|error| {
            DurabilityError::EnvelopeError(format!("invalid UTF-8 string field: {error}"))
        })
    }

    fn read_bytes(&mut self) -> Result<&'a [u8], DurabilityError> {
        let len = self.read_len()?;
        self.read_exact(len)
    }

    /// Reads a length, refusing one longer than the bytes left to read so a
    /// corrupt count cannot ask for more than the record holds.
    fn read_len(&mut self) -> Result<usize, DurabilityError> {
        let len = usize::try_from(self.read_u64()?).map_err(|error| {
            DurabilityError::EnvelopeError(format!("encoded length cannot fit memory: {error}"))
        })?;
        if len > self.bytes.len().saturating_sub(self.cursor) {
            return Err(DurabilityError::EnvelopeError(
                "truncated dedup entry bytes".to_owned(),
            ));
        }
        Ok(len)
    }

    fn read_u8(&mut self) -> Result<u8, DurabilityError> {
        Ok(self.read_exact(1)?[0])
    }

    fn read_u64(&mut self) -> Result<u64, DurabilityError> {
        let bytes = self.read_exact(8)?;
        let mut array = [0_u8; 8];
        array.copy_from_slice(bytes);
        Ok(u64::from_be_bytes(array))
    }

    fn read_exact(&mut self, len: usize) -> Result<&'a [u8], DurabilityError> {
        let end = self
            .cursor
            .checked_add(len)
            .ok_or_else(|| DurabilityError::EnvelopeError("dedup cursor overflow".to_owned()))?;
        if end > self.bytes.len() {
            return Err(DurabilityError::EnvelopeError(
                "truncated dedup entry bytes".to_owned(),
            ));
        }
        let slice = &self.bytes[self.cursor..end];
        self.cursor = end;
        Ok(slice)
    }

    fn finish(&self) -> Result<(), DurabilityError> {
        if self.cursor == self.bytes.len() {
            Ok(())
        } else {
            Err(DurabilityError::EnvelopeError(
                "trailing dedup entry bytes".to_owned(),
            ))
        }
    }
}