lodestone-types 0.0.28

//! Mapping-owned metric records and their canonical ordinary-source projection.

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

use serde::{Deserialize, Serialize};

pub const METRIC_RECORD_MODEL_FORMAT_VERSION: u16 = 1;
pub const METRIC_POINT_FORMAT_VERSION: u16 = 1;
pub const METRIC_SOURCE_FORMAT_VERSION: u16 = 1;

pub const MAX_METRIC_RESOURCE_ATTRIBUTES: usize = 64;
pub const MAX_METRIC_SCOPE_ATTRIBUTES: usize = 32;
pub const MAX_METRIC_SERIES_ATTRIBUTES: usize = 128;
pub const MAX_METRIC_GROUPING_ATTRIBUTES: usize = 16;
pub const MAX_METRIC_EXEMPLAR_ATTRIBUTES: usize = 16;
pub const MAX_METRIC_ATTRIBUTE_KEY_BYTES: usize = 256;
pub const MAX_METRIC_ATTRIBUTE_STRING_BYTES: usize = 4_096;
pub const MAX_METRIC_ATTRIBUTE_BYTES: usize = 64 * 1_024;
pub const MAX_METRIC_EXEMPLARS: usize = 8;
pub const MAX_METRIC_EXPLICIT_BOUNDS: usize = 4_096;
pub const MAX_METRIC_EXPONENTIAL_BUCKETS: usize = 4_096;
pub const MIN_METRIC_EXPONENTIAL_SCALE: i32 = -10;
pub const MAX_METRIC_EXPONENTIAL_SCALE: i32 = 20;
pub const MAX_METRIC_POINT_BYTES: usize = 1_048_576;

const RESOURCE_GROUPING_PREFIX: &str = "resource.";
const SCOPE_GROUPING_PREFIX: &str = "scope.";
const SERIES_GROUPING_PREFIX: &str = "series.";

#[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum IndexRecordModel {
    #[default]
    Document,
    MetricV1 {
        format_version: u16,
        grouping_attributes: Vec<String>,
    },
    RelationshipV1 {
        format_version: u16,
    },
    ProfileV1 {
        format_version: u16,
        privacy: crate::ProfilePrivacyPolicyV1,
    },
}

impl IndexRecordModel {
    pub fn validate(&self, mode: crate::IndexMode) -> Result<(), MetricRecordError> {
        match self {
            Self::Document => Ok(()),
            Self::MetricV1 {
                format_version,
                grouping_attributes,
            } => {
                if *format_version != METRIC_RECORD_MODEL_FORMAT_VERSION {
                    return Err(MetricRecordError::UnsupportedRecordModelVersion(*format_version));
                }
                if mode != crate::IndexMode::AppendLog {
                    return Err(MetricRecordError::MetricModelRequiresAppendLog);
                }
                validate_grouping_selectors(
                    grouping_attributes,
                    MAX_METRIC_GROUPING_ATTRIBUTES,
                    "grouping attributes",
                )
            }
            Self::RelationshipV1 { format_version } => {
                crate::validate_relationship_record_model_v1(*format_version, mode)
            }
            Self::ProfileV1 {
                format_version,
                privacy,
            } => crate::validate_profile_record_model_v1(*format_version, privacy, mode),
        }
    }

    #[must_use]
    pub fn metric_grouping_attributes(&self) -> Option<&[String]> {
        match self {
            Self::Document => None,
            Self::MetricV1 {
                grouping_attributes, ..
            } => Some(grouping_attributes),
            Self::RelationshipV1 { .. } => None,
            Self::ProfileV1 { .. } => None,
        }
    }

    #[must_use]
    pub fn semantic_digest_v1(&self) -> [u8; 32] {
        let mut hasher = blake3::Hasher::new();
        hasher.update(b"lodestone/index-record-model/v1\0");
        match self {
            Self::Document => {
                hasher.update(b"document\0");
            }
            Self::MetricV1 {
                format_version,
                grouping_attributes,
            } => {
                hasher.update(b"metric_v1\0");
                hasher.update(&format_version.to_be_bytes());
                hasher.update(&(grouping_attributes.len() as u64).to_be_bytes());
                for key in grouping_attributes {
                    hasher.update(&(key.len() as u64).to_be_bytes());
                    hasher.update(key.as_bytes());
                }
            }
            Self::RelationshipV1 { format_version } => {
                hasher.update(b"relationship_v1\0");
                hasher.update(&format_version.to_be_bytes());
            }
            Self::ProfileV1 {
                format_version,
                privacy,
            } => {
                hasher.update(b"profile_v1\0");
                hasher.update(&format_version.to_be_bytes());
                hasher.update(&privacy.semantic_digest_v1());
            }
        }
        *hasher.finalize().as_bytes()
    }
}

/// Resolves one mapping-admitted, namespace-qualified grouping selector.
///
/// The returned key remains qualified so equal attribute keys from distinct
/// metric namespaces cannot collapse into the same result group.
pub fn metric_grouping_attribute_v1(
    point: &MetricPointV1,
    selector: &str,
) -> Result<Option<MetricAttributeV1>, MetricRecordError> {
    let (attributes, key) = if let Some(key) = selector.strip_prefix(RESOURCE_GROUPING_PREFIX) {
        (&point.resource_attributes, key)
    } else if let Some(key) = selector.strip_prefix(SCOPE_GROUPING_PREFIX) {
        (&point.instrumentation_scope.attributes, key)
    } else if let Some(key) = selector.strip_prefix(SERIES_GROUPING_PREFIX) {
        (&point.series_attributes, key)
    } else {
        return Err(MetricRecordError::InvalidGroupingSelector);
    };
    validate_key(key)?;
    Ok(attributes
        .binary_search_by(|attribute| attribute.key.as_str().cmp(key))
        .ok()
        .and_then(|position| attributes.get(position))
        .map(|attribute| MetricAttributeV1 {
            key: selector.to_owned(),
            value: attribute.value.clone(),
        }))
}

#[derive(Clone, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(deny_unknown_fields)]
pub struct MetricAttributeV1 {
    pub key: String,
    pub value: MetricAttributeValueV1,
}

#[derive(Clone, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum MetricAttributeValueV1 {
    String { value: String },
    Boolean { value: bool },
    I64 { value: i64 },
    U64 { value: u64 },
    F64Bits { value: u64 },
}

impl MetricAttributeValueV1 {
    fn validate(&self) -> Result<(), MetricRecordError> {
        match self {
            Self::String { value } if value.len() > MAX_METRIC_ATTRIBUTE_STRING_BYTES => {
                Err(MetricRecordError::AttributeStringTooLong)
            }
            Self::F64Bits { value } if !f64::from_bits(*value).is_finite() => Err(MetricRecordError::NonFiniteNumber),
            Self::String { .. } | Self::Boolean { .. } | Self::I64 { .. } | Self::U64 { .. } | Self::F64Bits { .. } => {
                Ok(())
            }
        }
    }
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum MetricNumberV1 {
    I64 { value: i64 },
    F64Bits { value: u64 },
}

impl MetricNumberV1 {
    pub fn validate(self) -> Result<(), MetricRecordError> {
        match self {
            Self::F64Bits { value } if !f64::from_bits(value).is_finite() => Err(MetricRecordError::NonFiniteNumber),
            Self::I64 { .. } | Self::F64Bits { .. } => Ok(()),
        }
    }

    #[must_use]
    pub const fn kind(self) -> MetricNumberKindV1 {
        match self {
            Self::I64 { .. } => MetricNumberKindV1::I64,
            Self::F64Bits { .. } => MetricNumberKindV1::F64,
        }
    }

    #[must_use]
    pub const fn as_i64(self) -> Option<i64> {
        match self {
            Self::I64 { value } => Some(value),
            Self::F64Bits { .. } => None,
        }
    }

    #[must_use]
    pub fn as_f64(self) -> f64 {
        match self {
            Self::I64 { value } => value as f64,
            Self::F64Bits { value } => f64::from_bits(value),
        }
    }
}

#[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd)]
pub enum MetricNumberKindV1 {
    I64,
    F64,
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum MetricTemporalityV1 {
    Cumulative,
    Delta,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct MetricExponentialBucketsV1 {
    pub offset: i32,
    pub counts: Vec<u64>,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum MetricDataV1 {
    Gauge {
        value: MetricNumberV1,
    },
    Sum {
        value: MetricNumberV1,
        temporality: MetricTemporalityV1,
        monotonic: bool,
    },
    ExplicitHistogram {
        temporality: MetricTemporalityV1,
        count: u64,
        sum: Option<MetricNumberV1>,
        bounds_f64_bits: Vec<u64>,
        bucket_counts: Vec<u64>,
    },
    ExponentialHistogram {
        temporality: MetricTemporalityV1,
        count: u64,
        sum: Option<MetricNumberV1>,
        scale: i32,
        zero_count: u64,
        positive: MetricExponentialBucketsV1,
        negative: MetricExponentialBucketsV1,
    },
}

impl MetricDataV1 {
    #[must_use]
    pub const fn kind(&self) -> MetricPointKindV1 {
        match self {
            Self::Gauge { .. } => MetricPointKindV1::Gauge,
            Self::Sum { .. } => MetricPointKindV1::Sum,
            Self::ExplicitHistogram { .. } => MetricPointKindV1::ExplicitHistogram,
            Self::ExponentialHistogram { .. } => MetricPointKindV1::ExponentialHistogram,
        }
    }

    #[must_use]
    pub const fn temporality(&self) -> Option<MetricTemporalityV1> {
        match self {
            Self::Gauge { .. } => None,
            Self::Sum { temporality, .. }
            | Self::ExplicitHistogram { temporality, .. }
            | Self::ExponentialHistogram { temporality, .. } => Some(*temporality),
        }
    }

    #[must_use]
    pub const fn monotonic(&self) -> Option<bool> {
        match self {
            Self::Sum { monotonic, .. } => Some(*monotonic),
            Self::Gauge { .. } | Self::ExplicitHistogram { .. } | Self::ExponentialHistogram { .. } => None,
        }
    }

    fn validate(&self) -> Result<(), MetricRecordError> {
        match self {
            Self::Gauge { value } => value.validate(),
            Self::Sum {
                value, monotonic: true, ..
            } => {
                value.validate()?;
                if value.as_f64() < 0.0 {
                    return Err(MetricRecordError::NegativeMonotonicValue);
                }
                Ok(())
            }
            Self::Sum {
                value,
                monotonic: false,
                ..
            } => value.validate(),
            Self::ExplicitHistogram {
                count,
                sum,
                bounds_f64_bits,
                bucket_counts,
                ..
            } => {
                validate_optional_number(*sum)?;
                if bounds_f64_bits.len() > MAX_METRIC_EXPLICIT_BOUNDS
                    || bucket_counts.len() > MAX_METRIC_EXPLICIT_BOUNDS + 1
                {
                    return Err(MetricRecordError::HistogramBucketBudgetExceeded);
                }
                if bucket_counts.len() != bounds_f64_bits.len() + 1 {
                    return Err(MetricRecordError::InvalidExplicitHistogramLayout);
                }
                let mut previous = None;
                for bits in bounds_f64_bits {
                    let bound = f64::from_bits(*bits);
                    if !bound.is_finite() || previous.is_some_and(|prior: f64| bound <= prior) {
                        return Err(MetricRecordError::InvalidExplicitHistogramBounds);
                    }
                    previous = Some(bound);
                }
                let observed = checked_count(bucket_counts.iter().copied())?;
                if observed != *count {
                    return Err(MetricRecordError::HistogramCountMismatch);
                }
                Ok(())
            }
            Self::ExponentialHistogram {
                count,
                sum,
                scale,
                zero_count,
                positive,
                negative,
                ..
            } => {
                validate_optional_number(*sum)?;
                if positive.counts.len() > MAX_METRIC_EXPONENTIAL_BUCKETS
                    || negative.counts.len() > MAX_METRIC_EXPONENTIAL_BUCKETS
                {
                    return Err(MetricRecordError::HistogramBucketBudgetExceeded);
                }
                if !(MIN_METRIC_EXPONENTIAL_SCALE..=MAX_METRIC_EXPONENTIAL_SCALE).contains(scale) {
                    return Err(MetricRecordError::InvalidExponentialHistogramLayout);
                }
                let observed = checked_count(
                    positive
                        .counts
                        .iter()
                        .copied()
                        .chain(negative.counts.iter().copied())
                        .chain(std::iter::once(*zero_count)),
                )?;
                if observed != *count {
                    return Err(MetricRecordError::HistogramCountMismatch);
                }
                Ok(())
            }
        }
    }
}

fn validate_optional_number(value: Option<MetricNumberV1>) -> Result<(), MetricRecordError> {
    match value {
        Some(value) => value.validate(),
        None => Ok(()),
    }
}

fn checked_count(values: impl IntoIterator<Item = u64>) -> Result<u64, MetricRecordError> {
    values.into_iter().try_fold(0_u64, |total, value| {
        total.checked_add(value).ok_or(MetricRecordError::CountOverflow)
    })
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum MetricPointKindV1 {
    Gauge,
    Sum,
    ExplicitHistogram,
    ExponentialHistogram,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct MetricInstrumentationScopeV1 {
    pub name: String,
    pub version: Option<String>,
    pub schema_url: Option<String>,
    pub attributes: Vec<MetricAttributeV1>,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct MetricExemplarV1 {
    pub timestamp_unix_nanos: u64,
    pub value: MetricNumberV1,
    pub trace_id: Option<String>,
    pub span_id: Option<String>,
    pub filtered_attributes: Vec<MetricAttributeV1>,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct MetricPointV1 {
    pub format_version: u16,
    pub resource_attributes: Vec<MetricAttributeV1>,
    pub instrumentation_scope: MetricInstrumentationScopeV1,
    pub metric_name: String,
    pub description: Option<String>,
    pub unit: String,
    pub series_attributes: Vec<MetricAttributeV1>,
    pub timestamp_unix_nanos: u64,
    pub start_timestamp_unix_nanos: Option<u64>,
    pub data: MetricDataV1,
    pub exemplars: Vec<MetricExemplarV1>,
}

impl MetricPointV1 {
    pub fn validate(&self, model: &IndexRecordModel) -> Result<(), MetricRecordError> {
        if self.format_version != METRIC_POINT_FORMAT_VERSION {
            return Err(MetricRecordError::UnsupportedPointVersion(self.format_version));
        }
        if model.metric_grouping_attributes().is_none() {
            return Err(MetricRecordError::RecordModelMismatch);
        }
        model.validate(crate::IndexMode::AppendLog)?;
        validate_name(&self.metric_name, "metric name")?;
        validate_bounded_string(&self.unit)?;
        validate_optional_string(self.description.as_deref())?;
        validate_name(&self.instrumentation_scope.name, "instrumentation scope name")?;
        validate_optional_string(self.instrumentation_scope.version.as_deref())?;
        validate_optional_string(self.instrumentation_scope.schema_url.as_deref())?;
        validate_attributes(
            &self.resource_attributes,
            MAX_METRIC_RESOURCE_ATTRIBUTES,
            "resource attributes",
        )?;
        validate_attributes(
            &self.instrumentation_scope.attributes,
            MAX_METRIC_SCOPE_ATTRIBUTES,
            "scope attributes",
        )?;
        validate_attributes(
            &self.series_attributes,
            MAX_METRIC_SERIES_ATTRIBUTES,
            "series attributes",
        )?;
        if self
            .start_timestamp_unix_nanos
            .is_some_and(|start| start > self.timestamp_unix_nanos)
        {
            return Err(MetricRecordError::StartAfterObservation);
        }
        self.data.validate()?;
        if self.exemplars.len() > MAX_METRIC_EXEMPLARS {
            return Err(MetricRecordError::TooManyExemplars);
        }
        for exemplar in &self.exemplars {
            exemplar.value.validate()?;
            validate_hex_id(exemplar.trace_id.as_deref(), 32)?;
            validate_hex_id(exemplar.span_id.as_deref(), 16)?;
            validate_attributes(
                &exemplar.filtered_attributes,
                MAX_METRIC_EXEMPLAR_ATTRIBUTES,
                "exemplar attributes",
            )?;
        }
        let attribute_bytes = serde_json::to_vec(&(
            &self.resource_attributes,
            &self.instrumentation_scope.attributes,
            &self.series_attributes,
            self.exemplars
                .iter()
                .map(|exemplar| &exemplar.filtered_attributes)
                .collect::<Vec<_>>(),
        ))
        .map_err(|error| MetricRecordError::Codec(error.to_string()))?
        .len();
        if attribute_bytes > MAX_METRIC_ATTRIBUTE_BYTES {
            return Err(MetricRecordError::AttributeBudgetExceeded(attribute_bytes));
        }
        let encoded = serde_json::to_vec(self).map_err(|error| MetricRecordError::Codec(error.to_string()))?;
        if encoded.len() > MAX_METRIC_POINT_BYTES {
            return Err(MetricRecordError::PointBudgetExceeded(encoded.len()));
        }
        Ok(())
    }

    pub fn series_id_v1(&self, model: &IndexRecordModel) -> Result<[u8; 32], MetricRecordError> {
        self.validate(model)?;
        let identity = MetricSeriesIdentityV1 {
            format_version: METRIC_POINT_FORMAT_VERSION,
            resource_attributes: &self.resource_attributes,
            instrumentation_scope: &self.instrumentation_scope,
            metric_name: &self.metric_name,
            unit: &self.unit,
            point_kind: self.data.kind(),
            temporality: self.data.temporality(),
            monotonic: self.data.monotonic(),
            series_attributes: &self.series_attributes,
        };
        let encoded = serde_json::to_vec(&identity).map_err(|error| MetricRecordError::Codec(error.to_string()))?;
        let mut hasher = blake3::Hasher::new();
        hasher.update(b"lodestone/metric-series/v1\0");
        hasher.update(&(encoded.len() as u64).to_be_bytes());
        hasher.update(&encoded);
        Ok(*hasher.finalize().as_bytes())
    }
}

#[derive(Serialize)]
struct MetricSeriesIdentityV1<'a> {
    format_version: u16,
    resource_attributes: &'a [MetricAttributeV1],
    instrumentation_scope: &'a MetricInstrumentationScopeV1,
    metric_name: &'a str,
    unit: &'a str,
    point_kind: MetricPointKindV1,
    temporality: Option<MetricTemporalityV1>,
    monotonic: Option<bool>,
    series_attributes: &'a [MetricAttributeV1],
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct CanonicalMetricSourceV1 {
    pub format_version: u16,
    pub series_id: String,
    pub point: MetricPointV1,
}

impl CanonicalMetricSourceV1 {
    pub fn new(point: MetricPointV1, model: &IndexRecordModel) -> Result<Self, MetricRecordError> {
        let series_id = encode_lower_hex(point.series_id_v1(model)?);
        Ok(Self {
            format_version: METRIC_SOURCE_FORMAT_VERSION,
            series_id,
            point,
        })
    }

    pub fn validate(&self, model: &IndexRecordModel) -> Result<[u8; 32], MetricRecordError> {
        if self.format_version != METRIC_SOURCE_FORMAT_VERSION {
            return Err(MetricRecordError::UnsupportedSourceVersion(self.format_version));
        }
        let series_id = self.point.series_id_v1(model)?;
        if self.series_id != encode_lower_hex(series_id) {
            return Err(MetricRecordError::SeriesIdentityMismatch);
        }
        Ok(series_id)
    }
}

pub fn canonical_metric_source_v1(
    point: MetricPointV1,
    model: &IndexRecordModel,
) -> Result<(Vec<u8>, [u8; 32]), MetricRecordError> {
    let source = CanonicalMetricSourceV1::new(point, model)?;
    let series_id = source.validate(model)?;
    let encoded = serde_json::to_vec(&source).map_err(|error| MetricRecordError::Codec(error.to_string()))?;
    if encoded.len() > MAX_METRIC_POINT_BYTES {
        return Err(MetricRecordError::PointBudgetExceeded(encoded.len()));
    }
    Ok((encoded, series_id))
}

pub fn decode_canonical_metric_source_v1(
    bytes: &[u8],
    model: &IndexRecordModel,
) -> Result<(CanonicalMetricSourceV1, [u8; 32]), MetricRecordError> {
    if bytes.len() > MAX_METRIC_POINT_BYTES {
        return Err(MetricRecordError::PointBudgetExceeded(bytes.len()));
    }
    let source: CanonicalMetricSourceV1 =
        serde_json::from_slice(bytes).map_err(|error| MetricRecordError::Codec(error.to_string()))?;
    let series_id = source.validate(model)?;
    let canonical = serde_json::to_vec(&source).map_err(|error| MetricRecordError::Codec(error.to_string()))?;
    if canonical != bytes {
        return Err(MetricRecordError::NonCanonicalSource);
    }
    Ok((source, series_id))
}

#[must_use]
pub fn encode_lower_hex(bytes: [u8; 32]) -> String {
    const HEX: &[u8; 16] = b"0123456789abcdef";
    let mut output = String::with_capacity(64);
    for byte in bytes {
        output.push(char::from(HEX[usize::from(byte >> 4)]));
        output.push(char::from(HEX[usize::from(byte & 0x0f)]));
    }
    output
}

fn validate_attributes(
    values: &[MetricAttributeV1],
    maximum: usize,
    section: &'static str,
) -> Result<(), MetricRecordError> {
    if values.len() > maximum {
        return Err(MetricRecordError::TooManyAttributes(section));
    }
    let mut previous = None;
    for attribute in values {
        validate_key(&attribute.key)?;
        if previous.is_some_and(|value: &str| value >= attribute.key.as_str()) {
            return Err(MetricRecordError::AttributesNotSorted(section));
        }
        attribute.value.validate()?;
        previous = Some(attribute.key.as_str());
    }
    Ok(())
}

fn validate_grouping_selectors(
    values: &[String],
    maximum: usize,
    section: &'static str,
) -> Result<(), MetricRecordError> {
    if values.len() > maximum {
        return Err(MetricRecordError::TooManyAttributes(section));
    }
    let mut previous = None;
    for selector in values {
        if previous.is_some_and(|prior: &str| prior >= selector.as_str()) {
            return Err(MetricRecordError::AttributesNotSorted(section));
        }
        let key = selector
            .strip_prefix(RESOURCE_GROUPING_PREFIX)
            .or_else(|| selector.strip_prefix(SCOPE_GROUPING_PREFIX))
            .or_else(|| selector.strip_prefix(SERIES_GROUPING_PREFIX))
            .ok_or(MetricRecordError::InvalidGroupingSelector)?;
        validate_key(key)?;
        previous = Some(selector.as_str());
    }
    Ok(())
}

fn validate_key(value: &str) -> Result<(), MetricRecordError> {
    if value.is_empty()
        || value.len() > MAX_METRIC_ATTRIBUTE_KEY_BYTES
        || value.bytes().any(|byte| byte <= 0x1f || byte == 0x7f)
    {
        return Err(MetricRecordError::InvalidAttributeKey);
    }
    Ok(())
}

fn validate_name(value: &str, section: &'static str) -> Result<(), MetricRecordError> {
    if value.is_empty() {
        return Err(MetricRecordError::MissingName(section));
    }
    validate_bounded_string(value)
}

fn validate_optional_string(value: Option<&str>) -> Result<(), MetricRecordError> {
    match value {
        Some(value) => validate_bounded_string(value),
        None => Ok(()),
    }
}

fn validate_bounded_string(value: &str) -> Result<(), MetricRecordError> {
    if value.len() > MAX_METRIC_ATTRIBUTE_STRING_BYTES || value.bytes().any(|byte| byte == 0) {
        return Err(MetricRecordError::AttributeStringTooLong);
    }
    Ok(())
}

fn validate_hex_id(value: Option<&str>, length: usize) -> Result<(), MetricRecordError> {
    if value.is_some_and(|value| {
        value.len() != length
            || value
                .bytes()
                .any(|byte| !byte.is_ascii_digit() && !(b'a'..=b'f').contains(&byte))
    }) {
        return Err(MetricRecordError::InvalidCorrelationIdentity);
    }
    Ok(())
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub enum MetricRecordError {
    UnsupportedRecordModelVersion(u16),
    UnsupportedPointVersion(u16),
    UnsupportedSourceVersion(u16),
    MetricModelRequiresAppendLog,
    RelationshipModelRequiresMutableOrAppendLog,
    ProfileModelRequiresAppendLog,
    InvalidProfilePrivacyPolicy,
    RecordModelMismatch,
    InvalidGroupingSelector,
    TooManyAttributes(&'static str),
    AttributesNotSorted(&'static str),
    InvalidAttributeKey,
    AttributeStringTooLong,
    AttributeBudgetExceeded(usize),
    MissingName(&'static str),
    StartAfterObservation,
    NonFiniteNumber,
    NegativeMonotonicValue,
    TooManyExemplars,
    HistogramBucketBudgetExceeded,
    InvalidCorrelationIdentity,
    InvalidExplicitHistogramLayout,
    InvalidExplicitHistogramBounds,
    InvalidExponentialHistogramLayout,
    HistogramCountMismatch,
    CountOverflow,
    PointBudgetExceeded(usize),
    SeriesIdentityMismatch,
    NonCanonicalSource,
    Codec(String),
}

impl Display for MetricRecordError {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        write!(formatter, "metric record: {self:?}")
    }
}

impl Error for MetricRecordError {}

#[cfg(test)]
mod tests {
    use std::error::Error;

    use super::*;

    fn model() -> IndexRecordModel {
        IndexRecordModel::MetricV1 {
            format_version: METRIC_RECORD_MODEL_FORMAT_VERSION,
            grouping_attributes: vec!["series.host".to_owned()],
        }
    }

    fn point() -> MetricPointV1 {
        MetricPointV1 {
            format_version: METRIC_POINT_FORMAT_VERSION,
            resource_attributes: vec![MetricAttributeV1 {
                key: "service.name".to_owned(),
                value: MetricAttributeValueV1::String {
                    value: "checkout".to_owned(),
                },
            }],
            instrumentation_scope: MetricInstrumentationScopeV1 {
                name: "http".to_owned(),
                version: Some("1".to_owned()),
                schema_url: None,
                attributes: Vec::new(),
            },
            metric_name: "requests".to_owned(),
            description: Some("request count".to_owned()),
            unit: "{request}".to_owned(),
            series_attributes: vec![MetricAttributeV1 {
                key: "host".to_owned(),
                value: MetricAttributeValueV1::String { value: "a".to_owned() },
            }],
            timestamp_unix_nanos: 20,
            start_timestamp_unix_nanos: Some(10),
            data: MetricDataV1::Sum {
                value: MetricNumberV1::I64 { value: 4 },
                temporality: MetricTemporalityV1::Cumulative,
                monotonic: true,
            },
            exemplars: Vec::new(),
        }
    }

    #[test]
    fn metric_source_is_canonical_and_series_identity_ignores_description() -> Result<(), Box<dyn Error>> {
        let model = model();
        let (encoded, series_id) = canonical_metric_source_v1(point(), &model)?;
        let (decoded, decoded_id) = decode_canonical_metric_source_v1(&encoded, &model)?;
        assert_eq!(decoded_id, series_id);
        let mut changed = decoded.point;
        changed.description = Some("changed".to_owned());
        assert_eq!(changed.series_id_v1(&model)?, series_id);
        changed.unit = "ms".to_owned();
        assert_ne!(changed.series_id_v1(&model)?, series_id);
        Ok(())
    }

    #[test]
    fn metric_validation_rejects_unsorted_attributes_and_bad_histograms() {
        let model = model();
        let mut value = point();
        value.series_attributes = vec![
            MetricAttributeV1 {
                key: "z".to_owned(),
                value: MetricAttributeValueV1::Boolean { value: true },
            },
            MetricAttributeV1 {
                key: "a".to_owned(),
                value: MetricAttributeValueV1::Boolean { value: true },
            },
        ];
        assert_eq!(
            value.validate(&model),
            Err(MetricRecordError::AttributesNotSorted("series attributes"))
        );

        let mut value = point();
        value.data = MetricDataV1::ExplicitHistogram {
            temporality: MetricTemporalityV1::Delta,
            count: 2,
            sum: None,
            bounds_f64_bits: vec![1.0_f64.to_bits()],
            bucket_counts: vec![1],
        };
        assert_eq!(
            value.validate(&model),
            Err(MetricRecordError::InvalidExplicitHistogramLayout)
        );
    }

    #[test]
    fn record_model_is_mode_bound_immutable_and_digest_sensitive() -> Result<(), Box<dyn Error>> {
        let first = model();
        assert_eq!(first.validate(crate::IndexMode::AppendLog), Ok(()));
        assert_eq!(
            first.validate(crate::IndexMode::Mutable),
            Err(MetricRecordError::MetricModelRequiresAppendLog)
        );
        let second = IndexRecordModel::MetricV1 {
            format_version: METRIC_RECORD_MODEL_FORMAT_VERSION,
            grouping_attributes: vec!["series.zone".to_owned()],
        };
        assert_ne!(first.semantic_digest_v1(), second.semantic_digest_v1());
        assert_ne!(
            first.semantic_digest_v1(),
            IndexRecordModel::Document.semantic_digest_v1()
        );
        let invalid = IndexRecordModel::MetricV1 {
            format_version: METRIC_RECORD_MODEL_FORMAT_VERSION,
            grouping_attributes: vec!["host".to_owned()],
        };
        assert_eq!(
            invalid.validate(crate::IndexMode::AppendLog),
            Err(MetricRecordError::InvalidGroupingSelector)
        );
        let newer = IndexRecordModel::MetricV1 {
            format_version: METRIC_RECORD_MODEL_FORMAT_VERSION + 1,
            grouping_attributes: vec!["series.host".to_owned()],
        };
        assert_eq!(
            point().validate(&newer),
            Err(MetricRecordError::UnsupportedRecordModelVersion(
                METRIC_RECORD_MODEL_FORMAT_VERSION + 1
            ))
        );
        assert_eq!(
            metric_grouping_attribute_v1(&point(), "resource.service.name")?,
            Some(MetricAttributeV1 {
                key: "resource.service.name".to_owned(),
                value: MetricAttributeValueV1::String {
                    value: "checkout".to_owned(),
                },
            })
        );
        Ok(())
    }

    #[test]
    fn malformed_numbers_counts_correlation_and_noncanonical_sources_refuse() -> Result<(), Box<dyn Error>> {
        let model = model();
        let mut value = point();
        value.data = MetricDataV1::Gauge {
            value: MetricNumberV1::F64Bits {
                value: f64::NAN.to_bits(),
            },
        };
        assert_eq!(value.validate(&model), Err(MetricRecordError::NonFiniteNumber));

        let mut value = point();
        value.data = MetricDataV1::Sum {
            value: MetricNumberV1::I64 { value: -1 },
            temporality: MetricTemporalityV1::Delta,
            monotonic: true,
        };
        assert_eq!(value.validate(&model), Err(MetricRecordError::NegativeMonotonicValue));

        let mut value = point();
        value.data = MetricDataV1::ExponentialHistogram {
            temporality: MetricTemporalityV1::Delta,
            count: 2,
            sum: None,
            scale: 0,
            zero_count: 0,
            positive: MetricExponentialBucketsV1 {
                offset: 0,
                counts: vec![1],
            },
            negative: MetricExponentialBucketsV1 {
                offset: 0,
                counts: Vec::new(),
            },
        };
        assert_eq!(value.validate(&model), Err(MetricRecordError::HistogramCountMismatch));

        let (canonical, _) = canonical_metric_source_v1(point(), &model)?;
        let mut padded = canonical.clone();
        padded.push(b' ');
        assert_eq!(
            decode_canonical_metric_source_v1(&padded, &model),
            Err(MetricRecordError::NonCanonicalSource)
        );
        Ok(())
    }
}