lodestone-proto 0.0.26

//! Exact, bounded JSON canonicalization for the opt-in retained append-event
//! identity digest. Flag-off mutation digests remain byte-exact.

use std::collections::BTreeMap;
use std::error::Error;
use std::fmt::{self, Display, Formatter};

use lodestone_types::{
    CollectionIndexRef, DocumentId, IndexRecordModel, OperationFingerprint, decode_canonical_metric_source_v1,
    decode_canonical_relationship_source_v1,
};

pub const MAX_CANONICAL_EVENT_SOURCE_BYTES: usize = 1_048_576;
pub const MAX_CANONICAL_EVENT_NODES: usize = 4_096;
pub const MAX_CANONICAL_EVENT_DEPTH: usize = 64;

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum CanonicalEventJsonError {
    InvalidJson,
    DuplicateKey,
    NumberOutOfRange,
    DepthExceeded,
    BudgetExceeded,
    InvalidIdentityKey,
    InvalidPolicyGeneration,
    InvalidRecord,
}

impl Display for CanonicalEventJsonError {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        write!(formatter, "canonical append-event JSON: {self:?}")
    }
}

impl Error for CanonicalEventJsonError {}

/// Returns a typed canonical byte sequence, not a JSON rendering. Equal
/// sequences have equal JSON semantics under ADR-0241; callers domain-separate
/// the containing operation before hashing it.
pub fn canonical_event_source_v1(source: &[u8]) -> Result<Vec<u8>, CanonicalEventJsonError> {
    if source.is_empty() || source.len() > MAX_CANONICAL_EVENT_SOURCE_BYTES {
        return Err(CanonicalEventJsonError::BudgetExceeded);
    }
    let mut parser = Parser {
        source,
        offset: 0,
        nodes: 0,
    };
    let value = parser.value(0)?;
    parser.whitespace();
    if parser.offset != source.len() {
        return Err(CanonicalEventJsonError::InvalidJson);
    }
    Ok(value)
}

/// A private lookup commitment. It is never a bearer credential and does not
/// expose the event ID in an opaque resolver locator.
pub fn append_event_identity_v1(
    scope: CollectionIndexRef,
    event_id: &DocumentId,
    incarnation_identity_key: &[u8; 32],
) -> Result<[u8; 32], CanonicalEventJsonError> {
    if incarnation_identity_key == &[0; 32] {
        return Err(CanonicalEventJsonError::InvalidIdentityKey);
    }
    let mut hasher = blake3::Hasher::new_keyed(incarnation_identity_key);
    hasher.update(b"lodestone/retained-append-event-identity/v1");
    hasher.update(&scope.collection_id().to_be_bytes());
    hasher.update(&scope.index_id().to_be_bytes());
    hasher.update(&(event_id.as_str().len() as u64).to_be_bytes());
    hasher.update(event_id.as_str().as_bytes());
    Ok(*hasher.finalize().as_bytes())
}

/// The fixed Put/Any/event-ID-routing intent for retained append events.
/// A later policy generation must change this digest domain or generation.
pub fn append_event_put_fingerprint_v1(
    scope: CollectionIndexRef,
    event_id: &DocumentId,
    source: &[u8],
    identity_policy_generation: u64,
    incarnation_identity_key: &[u8; 32],
) -> Result<OperationFingerprint, CanonicalEventJsonError> {
    if identity_policy_generation == 0 {
        return Err(CanonicalEventJsonError::InvalidPolicyGeneration);
    }
    let canonical = canonical_event_source_v1(source)?;
    let event_identity = append_event_identity_v1(scope, event_id, incarnation_identity_key)?;
    let mut hasher = blake3::Hasher::new();
    hasher.update(b"lodestone/retained-append-put-any/event-id-routing/v1");
    hasher.update(&event_identity);
    hasher.update(&identity_policy_generation.to_be_bytes());
    hasher.update(&(canonical.len() as u64).to_be_bytes());
    hasher.update(&canonical);
    Ok(OperationFingerprint::new(*hasher.finalize().as_bytes()))
}

/// Selects the physical routing commitment admitted by the mapping-owned
/// record model. Event identity remains the idempotency and resolver identity.
pub fn retained_append_routing_identity_v2(
    scope: CollectionIndexRef,
    record_model: &IndexRecordModel,
    source: &[u8],
    event_identity: [u8; 32],
) -> Result<[u8; 32], CanonicalEventJsonError> {
    match record_model {
        IndexRecordModel::Document => Ok(event_identity),
        IndexRecordModel::MetricV1 { .. } => decode_canonical_metric_source_v1(source, record_model)
            .map(|(_, series_id)| series_id)
            .map_err(|_| CanonicalEventJsonError::InvalidRecord),
        IndexRecordModel::RelationshipV1 { .. } => {
            decode_canonical_relationship_source_v1(source, record_model, scope.collection_id())
                .map(|(_, relationship_id)| relationship_id)
                .map_err(|_| CanonicalEventJsonError::InvalidRecord)
        }
        IndexRecordModel::ProfileV1 { .. } => lodestone_types::decode_canonical_profile_source_v1(source, record_model)
            .map(|(_, profile_id)| profile_id)
            .map_err(|_| CanonicalEventJsonError::InvalidRecord),
    }
}

/// The current retained Put/Any intent. Routing is an independent commitment
/// so metric events can remain idempotent by event ID while co-locating a
/// series on one shard.
pub fn append_event_put_fingerprint_v2(
    scope: CollectionIndexRef,
    event_id: &DocumentId,
    routing_identity: [u8; 32],
    source: &[u8],
    identity_policy_generation: u64,
    incarnation_identity_key: &[u8; 32],
) -> Result<OperationFingerprint, CanonicalEventJsonError> {
    if identity_policy_generation == 0 || routing_identity == [0; 32] {
        return Err(CanonicalEventJsonError::InvalidPolicyGeneration);
    }
    let canonical = canonical_event_source_v1(source)?;
    let event_identity = append_event_identity_v1(scope, event_id, incarnation_identity_key)?;
    let mut hasher = blake3::Hasher::new();
    hasher.update(b"lodestone/retained-append-put-any/routing/v2");
    hasher.update(&event_identity);
    hasher.update(&routing_identity);
    hasher.update(&identity_policy_generation.to_be_bytes());
    hasher.update(&(canonical.len() as u64).to_be_bytes());
    hasher.update(&canonical);
    Ok(OperationFingerprint::new(*hasher.finalize().as_bytes()))
}

struct Parser<'a> {
    source: &'a [u8],
    offset: usize,
    nodes: usize,
}

impl Parser<'_> {
    fn whitespace(&mut self) {
        while self
            .source
            .get(self.offset)
            .is_some_and(|byte| matches!(byte, b' ' | b'\t' | b'\r' | b'\n'))
        {
            self.offset += 1;
        }
    }

    fn byte(&mut self, expected: u8) -> Result<(), CanonicalEventJsonError> {
        if self.source.get(self.offset) != Some(&expected) {
            return Err(CanonicalEventJsonError::InvalidJson);
        }
        self.offset += 1;
        Ok(())
    }

    fn value(&mut self, depth: usize) -> Result<Vec<u8>, CanonicalEventJsonError> {
        if depth > MAX_CANONICAL_EVENT_DEPTH {
            return Err(CanonicalEventJsonError::DepthExceeded);
        }
        self.nodes += 1;
        if self.nodes > MAX_CANONICAL_EVENT_NODES {
            return Err(CanonicalEventJsonError::BudgetExceeded);
        }
        self.whitespace();
        match self.source.get(self.offset).copied() {
            Some(b'"') => {
                let string = self.string()?;
                let mut output = vec![b'S'];
                append_bytes(&mut output, string.as_bytes())?;
                Ok(output)
            }
            Some(b'[') => self.array(depth + 1),
            Some(b'{') => self.object(depth + 1),
            Some(b't') => self.literal(b"true", b'T'),
            Some(b'f') => self.literal(b"false", b'F'),
            Some(b'n') => self.literal(b"null", b'Z'),
            Some(b'-' | b'0'..=b'9') => self.number(),
            _ => Err(CanonicalEventJsonError::InvalidJson),
        }
    }

    fn literal(&mut self, spelling: &[u8], tag: u8) -> Result<Vec<u8>, CanonicalEventJsonError> {
        if self.source.get(self.offset..self.offset.saturating_add(spelling.len())) != Some(spelling) {
            return Err(CanonicalEventJsonError::InvalidJson);
        }
        self.offset += spelling.len();
        Ok(vec![tag])
    }

    fn string(&mut self) -> Result<String, CanonicalEventJsonError> {
        let start = self.offset;
        self.byte(b'"')?;
        loop {
            let byte = *self
                .source
                .get(self.offset)
                .ok_or(CanonicalEventJsonError::InvalidJson)?;
            self.offset += 1;
            match byte {
                b'"' => break,
                b'\\' => {
                    self.offset = self
                        .offset
                        .checked_add(1)
                        .ok_or(CanonicalEventJsonError::BudgetExceeded)?;
                    if self.offset > self.source.len() {
                        return Err(CanonicalEventJsonError::InvalidJson);
                    }
                }
                0..=31 => return Err(CanonicalEventJsonError::InvalidJson),
                _ => {}
            }
        }
        serde_json::from_slice(&self.source[start..self.offset]).map_err(|_| CanonicalEventJsonError::InvalidJson)
    }

    fn array(&mut self, depth: usize) -> Result<Vec<u8>, CanonicalEventJsonError> {
        self.byte(b'[')?;
        self.whitespace();
        let mut entries = Vec::new();
        if self.source.get(self.offset) != Some(&b']') {
            loop {
                entries.push(self.value(depth)?);
                self.whitespace();
                match self.source.get(self.offset) {
                    Some(b',') => self.offset += 1,
                    Some(b']') => break,
                    _ => return Err(CanonicalEventJsonError::InvalidJson),
                }
            }
        }
        self.byte(b']')?;
        let mut output = vec![b'A'];
        append_count(&mut output, entries.len())?;
        for entry in entries {
            append_bytes(&mut output, &entry)?;
        }
        Ok(output)
    }

    fn object(&mut self, depth: usize) -> Result<Vec<u8>, CanonicalEventJsonError> {
        self.byte(b'{')?;
        self.whitespace();
        let mut entries = BTreeMap::new();
        if self.source.get(self.offset) != Some(&b'}') {
            loop {
                self.whitespace();
                let key = self.string()?;
                self.whitespace();
                self.byte(b':')?;
                let value = self.value(depth)?;
                if entries.insert(key, value).is_some() {
                    return Err(CanonicalEventJsonError::DuplicateKey);
                }
                self.whitespace();
                match self.source.get(self.offset) {
                    Some(b',') => self.offset += 1,
                    Some(b'}') => break,
                    _ => return Err(CanonicalEventJsonError::InvalidJson),
                }
            }
        }
        self.byte(b'}')?;
        let mut output = vec![b'O'];
        append_count(&mut output, entries.len())?;
        for (key, value) in entries {
            append_bytes(&mut output, key.as_bytes())?;
            append_bytes(&mut output, &value)?;
        }
        Ok(output)
    }

    fn number(&mut self) -> Result<Vec<u8>, CanonicalEventJsonError> {
        let negative = if self.source.get(self.offset) == Some(&b'-') {
            self.offset += 1;
            true
        } else {
            false
        };
        let integer_start = self.offset;
        match self.source.get(self.offset) {
            Some(b'0') => self.offset += 1,
            Some(b'1'..=b'9') => {
                self.offset += 1;
                while self.source.get(self.offset).is_some_and(u8::is_ascii_digit) {
                    self.offset += 1;
                }
            }
            _ => return Err(CanonicalEventJsonError::InvalidJson),
        }
        let integer_end = self.offset;
        let mut digits = self.source[integer_start..integer_end].to_vec();
        let mut fractional_digits = 0_usize;
        if self.source.get(self.offset) == Some(&b'.') {
            self.offset += 1;
            let fractional_start = self.offset;
            while self.source.get(self.offset).is_some_and(u8::is_ascii_digit) {
                self.offset += 1;
            }
            fractional_digits = self.offset - fractional_start;
            if fractional_digits == 0 {
                return Err(CanonicalEventJsonError::InvalidJson);
            }
            digits.extend_from_slice(&self.source[fractional_start..self.offset]);
        }
        let mut exponent = 0_i32;
        if matches!(self.source.get(self.offset), Some(b'e' | b'E')) {
            self.offset += 1;
            let exponent_negative = if matches!(self.source.get(self.offset), Some(b'+' | b'-')) {
                let value = self.source[self.offset] == b'-';
                self.offset += 1;
                value
            } else {
                false
            };
            let start = self.offset;
            while self.source.get(self.offset).is_some_and(u8::is_ascii_digit) {
                self.offset += 1;
            }
            if self.offset == start || self.offset - start > 4 {
                return Err(CanonicalEventJsonError::NumberOutOfRange);
            }
            for digit in &self.source[start..self.offset] {
                exponent = exponent * 10 + i32::from(digit - b'0');
            }
            if exponent_negative {
                exponent = -exponent;
            }
        }
        let nonzero = digits.iter().position(|digit| *digit != b'0');
        let mut output = vec![b'N', u8::from(negative)];
        if let Some(first) = nonzero {
            let last = digits
                .iter()
                .rposition(|digit| *digit != b'0')
                .ok_or(CanonicalEventJsonError::InvalidJson)?;
            let trailing = digits.len() - last - 1;
            let significant = &digits[first..=last];
            let adjusted = exponent
                - i32::try_from(fractional_digits).map_err(|_| CanonicalEventJsonError::NumberOutOfRange)?
                + i32::try_from(trailing).map_err(|_| CanonicalEventJsonError::NumberOutOfRange)?;
            if significant.len() > 38 || !(-308..=308).contains(&adjusted) {
                return Err(CanonicalEventJsonError::NumberOutOfRange);
            }
            append_bytes(&mut output, significant)?;
            output.extend_from_slice(&adjusted.to_be_bytes());
        } else {
            let adjusted =
                exponent - i32::try_from(fractional_digits).map_err(|_| CanonicalEventJsonError::NumberOutOfRange)?;
            if !(-308..=308).contains(&adjusted) {
                return Err(CanonicalEventJsonError::NumberOutOfRange);
            }
            append_bytes(&mut output, b"0")?;
            output.extend_from_slice(&0_i32.to_be_bytes());
        }
        Ok(output)
    }
}

fn append_count(output: &mut Vec<u8>, count: usize) -> Result<(), CanonicalEventJsonError> {
    let count = u32::try_from(count).map_err(|_| CanonicalEventJsonError::BudgetExceeded)?;
    output.extend_from_slice(&count.to_be_bytes());
    if output.len() > MAX_CANONICAL_EVENT_SOURCE_BYTES.saturating_mul(4) {
        return Err(CanonicalEventJsonError::BudgetExceeded);
    }
    Ok(())
}

fn append_bytes(output: &mut Vec<u8>, bytes: &[u8]) -> Result<(), CanonicalEventJsonError> {
    append_count(output, bytes.len())?;
    let maximum = MAX_CANONICAL_EVENT_SOURCE_BYTES.saturating_mul(4);
    if output
        .len()
        .checked_add(bytes.len())
        .is_none_or(|length| length > maximum)
    {
        return Err(CanonicalEventJsonError::BudgetExceeded);
    }
    output.extend_from_slice(bytes);
    Ok(())
}

#[cfg(test)]
mod tests {
    use lodestone_types::{
        CollectionId, CollectionIndexRef, DocumentId, EntityRefV1, IndexId, IndexRecordModel,
        METRIC_POINT_FORMAT_VERSION, METRIC_RECORD_MODEL_FORMAT_VERSION, MetricDataV1, MetricInstrumentationScopeV1,
        MetricNumberV1, MetricPointV1, RELATIONSHIP_RECORD_FORMAT_VERSION, RELATIONSHIP_RECORD_MODEL_FORMAT_VERSION,
        RelationshipDirectionalityV1, RelationshipRecordV1, canonical_metric_source_v1,
        canonical_relationship_source_v1,
    };

    use super::{
        CanonicalEventJsonError, append_event_identity_v1, append_event_put_fingerprint_v1,
        append_event_put_fingerprint_v2, canonical_event_source_v1, retained_append_routing_identity_v2,
    };

    #[test]
    fn ec7_5_json_equivalence_and_refusal() -> Result<(), CanonicalEventJsonError> {
        let left = canonical_event_source_v1(br#"{ "a": 1.00e+0, "b": ["\u00e9", -0.0] }"#)?;
        let right = canonical_event_source_v1("{\"b\":[\"é\",-0],\"a\":1}".as_bytes())?;
        assert_eq!(left, right);
        assert_ne!(canonical_event_source_v1(b"0")?, canonical_event_source_v1(b"-0")?);
        assert_ne!(
            canonical_event_source_v1("\"é\"".as_bytes())?,
            canonical_event_source_v1("\"é\"".as_bytes())?
        );
        for malformed in [
            br#"{"x":1,"\u0078":2}"#.as_slice(),
            br#"{"nested":{"x":1,"x":2}}"#,
            b"01",
            b"1e99999",
            b"1e309",
            b"0e9999",
            b"1.000000000000000000000000000000000000001",
            b"[1,]",
            b"\"\\ud800\"",
            b"\"\xff\"",
        ] {
            assert!(canonical_event_source_v1(malformed).is_err(), "{malformed:?}");
        }
        Ok(())
    }

    #[test]
    fn ec7_5_event_identity_is_incarnation_scoped_and_digest_is_semantic() -> Result<(), Box<dyn std::error::Error>> {
        let scope = CollectionIndexRef::new(CollectionId::new(1)?, IndexId::new(2)?);
        let next_incarnation = CollectionIndexRef::new(CollectionId::new(1)?, IndexId::new(3)?);
        let event = DocumentId::new("event-1")?;
        let key = [7; 32];
        let identity = append_event_identity_v1(scope, &event, &key)?;
        assert_ne!(identity, append_event_identity_v1(next_incarnation, &event, &key)?);
        assert_ne!(
            identity,
            append_event_identity_v1(scope, &DocumentId::new("event-2")?, &key)?
        );
        assert_ne!(identity, append_event_identity_v1(scope, &event, &[8; 32])?);
        assert_eq!(
            append_event_identity_v1(scope, &event, &[0; 32]),
            Err(CanonicalEventJsonError::InvalidIdentityKey)
        );
        assert_eq!(
            append_event_put_fingerprint_v1(scope, &event, br#"{"x": 1.0}"#, 1, &key)?,
            append_event_put_fingerprint_v1(scope, &event, br#"{"x":1e0}"#, 1, &key)?,
        );
        assert_ne!(
            append_event_put_fingerprint_v1(scope, &event, br#"{"x":1}"#, 1, &key)?,
            append_event_put_fingerprint_v1(scope, &event, br#"{"x":2}"#, 1, &key)?,
        );
        Ok(())
    }

    #[test]
    fn typed_routing_is_series_stable_while_event_identity_remains_unique() -> Result<(), Box<dyn std::error::Error>> {
        let scope = CollectionIndexRef::new(CollectionId::new(1)?, IndexId::new(2)?);
        let key = [7; 32];
        let first_event = DocumentId::new("event-1")?;
        let second_event = DocumentId::new("event-2")?;
        let first_identity = append_event_identity_v1(scope, &first_event, &key)?;
        let second_identity = append_event_identity_v1(scope, &second_event, &key)?;
        let model = IndexRecordModel::MetricV1 {
            format_version: METRIC_RECORD_MODEL_FORMAT_VERSION,
            grouping_attributes: Vec::new(),
        };
        let point = MetricPointV1 {
            format_version: METRIC_POINT_FORMAT_VERSION,
            resource_attributes: Vec::new(),
            instrumentation_scope: MetricInstrumentationScopeV1 {
                name: "test".to_owned(),
                version: None,
                schema_url: None,
                attributes: Vec::new(),
            },
            metric_name: "requests".to_owned(),
            description: None,
            unit: "{request}".to_owned(),
            series_attributes: Vec::new(),
            timestamp_unix_nanos: 1,
            start_timestamp_unix_nanos: None,
            data: MetricDataV1::Gauge {
                value: MetricNumberV1::I64 { value: 1 },
            },
            exemplars: Vec::new(),
        };
        let (source, series_id) = canonical_metric_source_v1(point, &model)?;
        assert_eq!(
            retained_append_routing_identity_v2(scope, &model, &source, first_identity)?,
            series_id
        );
        assert_eq!(
            retained_append_routing_identity_v2(scope, &model, &source, second_identity)?,
            series_id
        );
        assert_ne!(first_identity, second_identity);
        assert_ne!(
            append_event_put_fingerprint_v2(scope, &first_event, series_id, &source, 1, &key)?,
            append_event_put_fingerprint_v2(scope, &first_event, [9; 32], &source, 1, &key)?,
        );
        assert_eq!(
            retained_append_routing_identity_v2(scope, &IndexRecordModel::Document, br#"{"x":1}"#, first_identity)?,
            first_identity
        );
        assert_eq!(
            retained_append_routing_identity_v2(scope, &model, br#"{"x":1}"#, first_identity),
            Err(CanonicalEventJsonError::InvalidRecord)
        );
        Ok(())
    }

    #[test]
    fn typed_relationship_routing_is_stable_while_event_identity_remains_unique()
    -> Result<(), Box<dyn std::error::Error>> {
        let collection = CollectionId::new(1)?;
        let scope = CollectionIndexRef::new(collection, IndexId::new(2)?);
        let first_identity = append_event_identity_v1(scope, &DocumentId::new("event-1")?, &[7; 32])?;
        let second_identity = append_event_identity_v1(scope, &DocumentId::new("event-2")?, &[7; 32])?;
        let model = IndexRecordModel::RelationshipV1 {
            format_version: RELATIONSHIP_RECORD_MODEL_FORMAT_VERSION,
        };
        let relationship = RelationshipRecordV1 {
            format_version: RELATIONSHIP_RECORD_FORMAT_VERSION,
            relationship_id: "edge-1".to_owned(),
            source: EntityRefV1 {
                entity_kind: "service".to_owned(),
                entity_id: "api".to_owned(),
            },
            target: EntityRefV1 {
                entity_kind: "service".to_owned(),
                entity_id: "database".to_owned(),
            },
            relationship_type: "calls".to_owned(),
            directionality: RelationshipDirectionalityV1::Directed,
            valid_from_unix_nanos: None,
            valid_through_unix_nanos: None,
            provenance: vec!["runtime".to_owned()],
            evidence: Vec::new(),
            assertion_class: "observed".to_owned(),
        };
        let (source, routing) = canonical_relationship_source_v1(relationship, &model, collection)?;
        assert_eq!(
            retained_append_routing_identity_v2(scope, &model, &source, first_identity)?,
            routing
        );
        assert_eq!(
            retained_append_routing_identity_v2(scope, &model, &source, second_identity)?,
            routing
        );
        assert_ne!(first_identity, second_identity);
        Ok(())
    }
}