stele/

metrics.rs

use std::fmt;

use datadog_protos::metrics::{v3::Payload as V3Payload, Dogsketch, MetricPayload, MetricType, SketchPayload};
use ddsketch::DDSketch;
use float_cmp::ApproxEqRatio as _;
use saluki_error::{generic_error, GenericError};
use serde::{Deserialize, Serialize};

/// JSON envelope for the legacy V1 series intake (`/api/v1/series`).
#[derive(Deserialize)]
struct V1SeriesEnvelope {
    series: Vec<V1Serie>,
}

/// A single `Serie` entry in a V1 series payload.
///
/// Mirrors the Datadog Agent's wire format (see `pkg/metrics/series.go`). `omitempty` fields default to empty.
// Other fields present in the JSON envelope (`source_type_name`, `unit`) are intentionally not deserialized;
// serde silently ignores unknown JSON fields, so omitting them here is sufficient.
#[derive(Deserialize)]
struct V1Serie {
    metric: String,
    points: Vec<(i64, f64)>,
    #[serde(default)]
    tags: Vec<String>,
    #[serde(default)]
    host: String,
    #[serde(default)]
    device: String,
    #[serde(rename = "type", default)]
    mtype: String,
    #[serde(default)]
    interval: i64,
}

/// A metric's unique identifier.
///
/// The host is normalized into the tag list as a `host:<value>` tag rather than carried as a separate field; the
/// Datadog backend treats host as a first-class dimension on the time series, equivalent to any other tag. This
/// keeps comparison logic uniform regardless of which wire format the metric originated from.
#[derive(Clone, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
pub struct MetricContext {
    name: String,
    tags: Vec<String>,
}

impl MetricContext {
    /// Returns the name of the context.
    pub fn name(&self) -> &str {
        &self.name
    }

    /// Returns the tags of the context.
    ///
    /// When the underlying payload carried a host, it appears here as a `host:<value>` tag.
    pub fn tags(&self) -> &[String] {
        &self.tags
    }

    /// Consumes this context, returning the name and tags.
    pub fn into_parts(self) -> (String, Vec<String>) {
        (self.name, self.tags)
    }
}

impl fmt::Display for MetricContext {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{}", self.name)?;

        if !self.tags.is_empty() {
            write!(f, " {{{}}}", self.tags.join(", "))?;
        }

        Ok(())
    }
}

/// A simplified metric representation.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct Metric {
    context: MetricContext,
    values: Vec<(u64, MetricValue)>,
}

impl Metric {
    /// Returns the context of the metric.
    pub fn context(&self) -> &MetricContext {
        &self.context
    }

    /// Returns the values associated with the metric.
    pub fn values(&self) -> &[(u64, MetricValue)] {
        &self.values
    }
}

/// A metric value.
///
/// # Equality
///
/// `MetricValue` implements `PartialEq` and `Eq`, the majority of which involves comparing floating-point (`f64`)
/// numbers. Comparing floating-point numbers for equality is inherently tricky ([this][bitbanging_io] is just one blog
/// post/article out of thousands on the subject). In the equality implementation for `MetricValue`, we use a
/// ratio-based approach.
///
/// This means that when comparing two floating-point numbers, we look at their _ratio_ to one another, with an upper
/// bound on the allowed difference. For example, if we compare 99 to 100, there's a difference of 1% (`1 - (99/100) =
/// 0.01 = 1%`), while the difference between 99.999 and 100 is only 0.001% (`1 - (99.999/100) = 0.00001 = 0.001%`). As
/// most comparisons are expected to be close, only differing by a few ULPs (units in the last place) due to slight
/// differences in how floating-point numbers are implemented between Go and Rust, this approach is sufficient to
/// compensate for the inherent imprecision while not falling victim to relying on ULPs or epsilon directly, whose
/// applicability depends on the number range being compared.
///
/// Specifically, we compare floating-point numbers using a ratio of `0.00000001` (0.0000001%), meaning the smaller of
/// the two values being compared must be within 99.999999% to 100% of the larger number, which is sufficiently precise
/// for our concerns.
///
/// [bitbanging_io]: https://bitbashing.io/comparing-floats.html
#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(tag = "mtype")]
pub enum MetricValue {
    /// A count.
    Count {
        /// The value of the count.
        value: f64,
    },

    /// A rate.
    ///
    /// Rates are per-second adjusted counts. For example, a count that increased by 100 over 10 seconds would be
    /// represented as a rate with an interval of 10 (seconds) and a value of 10 (`100 / 10 = 10`).
    Rate {
        /// The interval of the rate, in seconds.
        interval: u64,

        /// The per-second value of the rate.
        value: f64,
    },

    /// A gauge.
    Gauge {
        /// The value of the gauge.
        value: f64,
    },

    /// A sketch.
    Sketch {
        /// The sketch data.
        sketch: DDSketch,
    },
}

impl PartialEq for MetricValue {
    fn eq(&self, other: &Self) -> bool {
        // When comparing two values, the smaller value cannot deviate by more than 0.0000001% of the larger value.
        const RATIO_ERROR: f64 = 0.00000001;

        match (self, other) {
            (MetricValue::Count { value: value_a }, MetricValue::Count { value: value_b }) => {
                value_a.approx_eq_ratio(value_b, RATIO_ERROR)
            }
            (
                MetricValue::Rate {
                    interval: interval_a,
                    value: value_a,
                },
                MetricValue::Rate {
                    interval: interval_b,
                    value: value_b,
                },
            ) => interval_a == interval_b && value_a.approx_eq_ratio(value_b, RATIO_ERROR),
            (MetricValue::Gauge { value: value_a }, MetricValue::Gauge { value: value_b }) => {
                value_a.approx_eq_ratio(value_b, RATIO_ERROR)
            }
            (MetricValue::Sketch { sketch: sketch_a }, MetricValue::Sketch { sketch: sketch_b }) => {
                approx_eq_ratio_optional(sketch_a.min(), sketch_b.min(), RATIO_ERROR)
                    && approx_eq_ratio_optional(sketch_a.max(), sketch_b.max(), RATIO_ERROR)
                    && approx_eq_ratio_optional(sketch_a.avg(), sketch_b.avg(), RATIO_ERROR)
                    && approx_eq_ratio_optional(sketch_a.sum(), sketch_b.sum(), RATIO_ERROR)
                    && sketch_a.count() == sketch_b.count()
                    && sketch_a.bin_count() == sketch_b.bin_count()
            }
            _ => false,
        }
    }
}

impl Eq for MetricValue {}

impl Metric {
    /// Attempts to parse metrics from a series v1 payload.
    ///
    /// V1 keeps a separate `device` JSON field rather than a `device:<value>` tag like the V2 protobuf encoder. To
    /// keep the post-conversion `stele::Metric` representation comparable between V1 and V2 payloads, this re-injects
    /// `device:<value>` into the tag list when the JSON `device` field is non-empty.
    ///
    /// # Errors
    ///
    /// If the JSON can't be deserialized, contains invalid data (for example, an unknown `type`), or has out-of-range
    /// timestamps, an error is returned.
    pub fn try_from_series_v1(payload: &[u8]) -> Result<Vec<Self>, GenericError> {
        let envelope: V1SeriesEnvelope = serde_json::from_slice(payload)
            .map_err(|e| generic_error!("Failed to parse V1 series JSON payload: {}", e))?;

        let mut metrics = Vec::with_capacity(envelope.series.len());

        for serie in envelope.series {
            let mut tags = serie.tags;
            if !serie.host.is_empty() {
                tags.push(format!("host:{}", serie.host));
            }
            if !serie.device.is_empty() {
                tags.push(format!("device:{}", serie.device));
            }

            let mut values = Vec::with_capacity(serie.points.len());
            for (ts, value) in serie.points {
                let timestamp =
                    u64::try_from(ts).map_err(|_| generic_error!("Invalid timestamp in V1 series payload: {}", ts))?;

                let metric_value = match serie.mtype.as_str() {
                    "count" => MetricValue::Count { value },
                    "rate" => MetricValue::Rate {
                        interval: serie.interval as u64,
                        value,
                    },
                    "gauge" => MetricValue::Gauge { value },
                    other => {
                        return Err(generic_error!(
                            "Unknown metric type '{}' in V1 series payload (metric '{}')",
                            other,
                            serie.metric
                        ));
                    }
                };
                values.push((timestamp, metric_value));
            }

            metrics.push(Metric {
                context: MetricContext {
                    name: serie.metric,
                    tags,
                },
                values,
            });
        }

        Ok(metrics)
    }

    /// Attempts to parse metrics from a series v2 payload.
    ///
    /// # Errors
    ///
    /// If the metric payload contains invalid data, an error will be returned.
    pub fn try_from_series_v2(payload: MetricPayload) -> Result<Vec<Self>, GenericError> {
        let mut metrics = Vec::new();

        for series in payload.series {
            let name = series.metric().to_string();
            let mut tags: Vec<String> = series.tags().iter().map(|tag| tag.to_string()).collect();
            // V2 protobuf encodes the hostname as a resource with type="host". The Datadog Agent's wire format
            // contract requires at least one such resource per series, but we still tolerate its absence. The host
            // is appended to the tag list (as `host:<value>`) to match the backend's representation and to keep
            // comparison logic uniform with the V1 JSON path.
            if let Some(host) = series.resources.iter().find(|r| r.type_() == "host") {
                let host_name = host.name();
                if !host_name.is_empty() {
                    tags.push(format!("host:{}", host_name));
                }
            }
            let mut values = Vec::new();

            match series.type_() {
                MetricType::UNSPECIFIED => {
                    return Err(generic_error!("Received metric series with UNSPECIFIED type."));
                }
                MetricType::COUNT => {
                    for point in series.points {
                        let timestamp = u64::try_from(point.timestamp)
                            .map_err(|_| generic_error!("Invalid timestamp for point: {}", point.timestamp))?;
                        values.push((timestamp, MetricValue::Count { value: point.value }));
                    }
                }
                MetricType::RATE => {
                    for point in series.points {
                        let timestamp = u64::try_from(point.timestamp)
                            .map_err(|_| generic_error!("Invalid timestamp for point: {}", point.timestamp))?;
                        values.push((
                            timestamp,
                            MetricValue::Rate {
                                interval: series.interval as u64,
                                value: point.value,
                            },
                        ));
                    }
                }
                MetricType::GAUGE => {
                    for point in series.points {
                        let timestamp = u64::try_from(point.timestamp)
                            .map_err(|_| generic_error!("Invalid timestamp for point: {}", point.timestamp))?;
                        values.push((timestamp, MetricValue::Gauge { value: point.value }));
                    }
                }
            }

            metrics.push(Metric {
                context: MetricContext { name, tags },
                values,
            })
        }

        Ok(metrics)
    }

    /// Attempts to parse metrics from a sketch payload.
    ///
    /// # Errors
    ///
    /// If the sketch payload contains invalid data, an error will be returned.
    pub fn try_from_sketch(payload: SketchPayload) -> Result<Vec<Self>, GenericError> {
        let mut metrics = Vec::new();

        for sketch in payload.sketches {
            let name = sketch.metric().to_string();
            let mut tags: Vec<String> = sketch.tags().iter().map(|tag| tag.to_string()).collect();
            // V2 sketches carry the host on the sketch message itself rather than via a resources list. Fold it
            // into the tag list (as `host:<value>`) for comparison parity with the V1 JSON / V2 series paths.
            let host = sketch.host();
            if !host.is_empty() {
                tags.push(format!("host:{}", host));
            }
            let mut values = Vec::new();

            for dogsketch in sketch.dogsketches {
                let timestamp = u64::try_from(dogsketch.ts)
                    .map_err(|_| generic_error!("Invalid timestamp for sketch: {}", dogsketch.ts))?;
                let sketch = DDSketch::try_from(dogsketch)
                    .map_err(|e| generic_error!("Failed to convert DogSketch to DDSketch: {}", e))?;
                values.push((timestamp, MetricValue::Sketch { sketch }));
            }

            metrics.push(Metric {
                context: MetricContext { name, tags },
                values,
            })
        }

        Ok(metrics)
    }
}

// V3 metric type constants (from intake_v3.proto metricType enum).
const V3_METRIC_TYPE_COUNT: u64 = 1;
const V3_METRIC_TYPE_RATE: u64 = 2;
const V3_METRIC_TYPE_GAUGE: u64 = 3;
const V3_METRIC_TYPE_SKETCH: u64 = 4;

// V3 value type constants (from intake_v3.proto valueType enum).
const V3_VALUE_TYPE_ZERO: u64 = 0x00;
const V3_VALUE_TYPE_SINT64: u64 = 0x10;
const V3_VALUE_TYPE_FLOAT32: u64 = 0x20;
const V3_VALUE_TYPE_FLOAT64: u64 = 0x30;

/// Tracks cursors into the various value arrays of a v3 payload during decoding.
struct V3ValueCursors {
    timestamp: usize,
    sint64: usize,
    float32: usize,
    float64: usize,
    sketch_point: usize,
    sketch_bin_key: usize,
    sketch_bin_cnt: usize,
}

impl V3ValueCursors {
    fn new() -> Self {
        Self {
            timestamp: 0,
            sint64: 0,
            float32: 0,
            float64: 0,
            sketch_point: 0,
            sketch_bin_key: 0,
            sketch_bin_cnt: 0,
        }
    }
}

impl Metric {
    /// Attempts to parse metrics from a v3 payload.
    ///
    /// The v3 format uses columnar encoding with dictionary deduplication and delta encoding.
    ///
    /// # Errors
    ///
    /// If the payload contains invalid data, an error will be returned.
    pub fn try_from_v3(mut payload: V3Payload) -> Result<Vec<Self>, GenericError> {
        let data = payload
            .metricData
            .take()
            .ok_or_else(|| generic_error!("V3 payload missing metricData"))?;

        let num_metrics = data.types.len();
        if num_metrics == 0 {
            return Ok(Vec::new());
        }

        // Parse dictionaries.
        let names_dict = parse_dict_strings(&data.dictNameStr)?;
        let tags_dict = parse_dict_strings(&data.dictTagStr)?;
        let tagsets_dict = parse_tagsets(&data.dictTagsets, &tags_dict)?;
        let resources_dict = parse_resources(
            &data.dictResourceLen,
            &data.dictResourceType,
            &data.dictResourceName,
            &parse_dict_strings(&data.dictResourceStr)?,
        )?;

        // Delta-decode index arrays.
        let mut name_refs = data.nameRefs;
        let mut tagset_refs = data.tagsetRefs;
        let mut resources_refs = data.resourcesRefs;
        let mut timestamps = data.timestamps;
        delta_decode(&mut name_refs);
        delta_decode(&mut tagset_refs);
        delta_decode(&mut resources_refs);
        delta_decode(&mut timestamps);

        // Delta-decode sketch bin keys (per-sketch sequences are individually delta-encoded,
        // but we handle that during iteration).
        let mut sketch_bin_keys = data.sketchBinKeys;

        let mut cursors = V3ValueCursors::new();
        let mut metrics = Vec::with_capacity(num_metrics);

        for i in 0..num_metrics {
            let type_field = data
                .types
                .get(i)
                .copied()
                .ok_or_else(|| generic_error!("Ran out of metric types"))?;
            let metric_type = type_field & 0x0F;
            let value_type = type_field & 0xF0;
            let num_points = data
                .numPoints
                .get(i)
                .copied()
                .ok_or_else(|| generic_error!("Ran out of numPoints"))
                .and_then(|num_points| u64_to_usize(num_points, "numPoints"))?;

            // Resolve name (1-based index).
            let name_ref = name_refs
                .get(i)
                .copied()
                .ok_or_else(|| generic_error!("Ran out of nameRefs"))
                .and_then(|name_ref| i64_to_usize(name_ref, "name ref"))?;
            let name = if name_ref == 0 {
                String::new()
            } else {
                names_dict
                    .get(name_ref - 1)
                    .ok_or_else(|| generic_error!("Invalid name ref {} (dict size {})", name_ref, names_dict.len()))?
                    .clone()
            };

            // Resolve tags (1-based index).
            let tagset_ref = tagset_refs
                .get(i)
                .copied()
                .ok_or_else(|| generic_error!("Ran out of tagsetRefs"))
                .and_then(|tagset_ref| i64_to_usize(tagset_ref, "tagset ref"))?;
            let mut tags = if tagset_ref == 0 {
                Vec::new()
            } else {
                tagsets_dict
                    .get(tagset_ref - 1)
                    .ok_or_else(|| {
                        generic_error!("Invalid tagset ref {} (dict size {})", tagset_ref, tagsets_dict.len())
                    })?
                    .clone()
            };

            let resource_ref = resources_refs
                .get(i)
                .copied()
                .map(|resource_ref| i64_to_usize(resource_ref, "resource ref"))
                .transpose()?
                .unwrap_or(0);
            if resource_ref != 0 {
                let resources = resources_dict.get(resource_ref - 1).ok_or_else(|| {
                    generic_error!(
                        "Invalid resource ref {} (dict size {})",
                        resource_ref,
                        resources_dict.len()
                    )
                })?;
                if let Some((_, host_name)) = resources
                    .iter()
                    .find(|(resource_type, resource_name)| resource_type == "host" && !resource_name.is_empty())
                {
                    tags.push(format!("host:{}", host_name));
                }
            }

            let mut values = Vec::with_capacity(num_points);

            if metric_type == V3_METRIC_TYPE_SKETCH {
                for _ in 0..num_points {
                    // Read timestamp.
                    let ts = *timestamps
                        .get(cursors.timestamp)
                        .ok_or_else(|| generic_error!("Ran out of timestamps"))?;
                    let timestamp = u64::try_from(ts).map_err(|_| generic_error!("Invalid timestamp: {}", ts))?;
                    cursors.timestamp += 1;

                    // The Agent writes sketch summaries as sum, min, max, then count. Count is always in valsSint64,
                    // but integer summaries can share that column, so count must be read after the summary values.
                    let sum = read_value(
                        value_type,
                        &mut cursors,
                        &data.valsSint64,
                        &data.valsFloat32,
                        &data.valsFloat64,
                    )?;
                    let min = read_value(
                        value_type,
                        &mut cursors,
                        &data.valsSint64,
                        &data.valsFloat32,
                        &data.valsFloat64,
                    )?;
                    let max = read_value(
                        value_type,
                        &mut cursors,
                        &data.valsSint64,
                        &data.valsFloat32,
                        &data.valsFloat64,
                    )?;
                    let cnt = *data
                        .valsSint64
                        .get(cursors.sint64)
                        .ok_or_else(|| generic_error!("Ran out of sint64 values for sketch count"))?;
                    cursors.sint64 += 1;
                    let avg = if cnt != 0 { sum / cnt as f64 } else { 0.0 };

                    // Read bin data.
                    let num_bins = *data
                        .sketchNumBins
                        .get(cursors.sketch_point)
                        .ok_or_else(|| generic_error!("Ran out of sketchNumBins"))?
                        as usize;
                    cursors.sketch_point += 1;

                    let bin_key_start = cursors.sketch_bin_key;
                    let bin_key_end = bin_key_start + num_bins;
                    if bin_key_end > sketch_bin_keys.len() {
                        return Err(generic_error!("Ran out of sketch bin keys"));
                    }

                    // Delta-decode this sketch's bin keys.
                    delta_decode_i32(&mut sketch_bin_keys[bin_key_start..bin_key_end]);

                    let k: Vec<i32> = sketch_bin_keys[bin_key_start..bin_key_end].to_vec();
                    cursors.sketch_bin_key = bin_key_end;

                    let bin_cnt_start = cursors.sketch_bin_cnt;
                    let bin_cnt_end = bin_cnt_start + num_bins;
                    if bin_cnt_end > data.sketchBinCnts.len() {
                        return Err(generic_error!("Ran out of sketch bin counts"));
                    }
                    let n: Vec<u32> = data.sketchBinCnts[bin_cnt_start..bin_cnt_end].to_vec();
                    cursors.sketch_bin_cnt = bin_cnt_end;

                    // Build a Dogsketch proto and use the existing TryFrom conversion.
                    let mut dogsketch = Dogsketch::new();
                    dogsketch.ts = ts;
                    dogsketch.cnt = cnt;
                    dogsketch.min = min;
                    dogsketch.max = max;
                    dogsketch.avg = avg;
                    dogsketch.sum = sum;
                    dogsketch.set_k(k);
                    dogsketch.set_n(n);

                    let sketch = DDSketch::try_from(dogsketch)
                        .map_err(|e| generic_error!("Failed to convert v3 sketch to DDSketch: {}", e))?;
                    values.push((timestamp, MetricValue::Sketch { sketch }));
                }
            } else {
                for _ in 0..num_points {
                    // Read timestamp.
                    let ts = *timestamps
                        .get(cursors.timestamp)
                        .ok_or_else(|| generic_error!("Ran out of timestamps"))?;
                    let timestamp = u64::try_from(ts).map_err(|_| generic_error!("Invalid timestamp: {}", ts))?;
                    cursors.timestamp += 1;

                    // Read point value.
                    let value = read_value(
                        value_type,
                        &mut cursors,
                        &data.valsSint64,
                        &data.valsFloat32,
                        &data.valsFloat64,
                    )?;

                    let metric_value = match metric_type {
                        V3_METRIC_TYPE_COUNT => MetricValue::Count { value },
                        V3_METRIC_TYPE_RATE => MetricValue::Rate {
                            interval: data
                                .intervals
                                .get(i)
                                .copied()
                                .ok_or_else(|| generic_error!("Ran out of intervals"))?,
                            value,
                        },
                        V3_METRIC_TYPE_GAUGE => MetricValue::Gauge { value },
                        other => return Err(generic_error!("Unknown v3 metric type: {}", other)),
                    };

                    values.push((timestamp, metric_value));
                }
            }

            metrics.push(Metric {
                context: MetricContext { name, tags },
                values,
            });
        }

        Ok(metrics)
    }
}

/// Delta-decode in place: convert deltas to absolute values (prefix sum).
fn delta_decode(s: &mut [i64]) {
    for i in 1..s.len() {
        s[i] += s[i - 1];
    }
}

/// Delta-decode i32 values in place.
fn delta_decode_i32(s: &mut [i32]) {
    for i in 1..s.len() {
        s[i] += s[i - 1];
    }
}

/// Read a varint from a byte slice, returning `(value, bytes_consumed)`.
fn read_varint(data: &[u8]) -> Result<(u64, usize), GenericError> {
    let mut value: u64 = 0;
    let mut shift = 0;
    for (i, &byte) in data.iter().enumerate() {
        value |= ((byte & 0x7F) as u64) << shift;
        if byte & 0x80 == 0 {
            return Ok((value, i + 1));
        }
        shift += 7;
        if shift >= 64 {
            return Err(generic_error!("Varint too large"));
        }
    }
    Err(generic_error!("Unexpected end of data reading varint"))
}

/// Parse varint-length-prefixed strings from a byte buffer.
fn parse_dict_strings(data: &[u8]) -> Result<Vec<String>, GenericError> {
    let mut strings = Vec::new();
    let mut offset = 0;
    while offset < data.len() {
        let (len, varint_size) = read_varint(&data[offset..])?;
        offset += varint_size;
        let len = len as usize;
        if offset + len > data.len() {
            return Err(generic_error!("Dictionary string extends past end of buffer"));
        }
        let s = simdutf8::basic::from_utf8(&data[offset..offset + len])
            .map_err(|e| generic_error!("Invalid UTF-8 in dictionary string: {}", e))?;
        strings.push(s.to_string());
        offset += len;
    }
    Ok(strings)
}

/// Parse tagsets from the `dictTagsets` array using the tag dictionary.
///
/// Each tagset in `dict_tagsets` is encoded as: length, then that many sorted-then-delta-encoded
/// entries. Each decoded entry is one of:
///
/// - A positive value: a 1-based index into the tag-string dictionary.
/// - A negative value `-N`: a backreference to the previously interned tagset with 1-based ID
///   `N`, whose (already fully resolved) tags are included in this one. The Agent emits these when
///   it splits a metric's composite tags into two groups and encodes the second group
///   against the first (see `internTags` in the Agent's `iterable_series_v3.go`). Because the
///   referenced tagset is always interned before the tagset that references it, a single forward
///   pass can resolve the reference against the tagsets parsed so far.
fn parse_tagsets(dict_tagsets: &[i64], tags_dict: &[String]) -> Result<Vec<Vec<String>>, GenericError> {
    let mut tagsets: Vec<Vec<String>> = Vec::new();
    let mut offset = 0;
    while offset < dict_tagsets.len() {
        let count = i64_to_usize(dict_tagsets[offset], "tagset length")?;
        offset += 1;
        if offset + count > dict_tagsets.len() {
            return Err(generic_error!("Tagset extends past end of dictTagsets array"));
        }

        // Delta-decode the entries within this tagset.
        let mut entries: Vec<i64> = dict_tagsets[offset..offset + count].to_vec();
        delta_decode(&mut entries);

        let mut tags = Vec::with_capacity(count);
        for &entry in &entries {
            match entry.cmp(&0) {
                std::cmp::Ordering::Less => {
                    // Prefix reference to a previously interned tagset (1-based ID = -entry).
                    let prefix_id = i64_to_usize(-entry, "tagset prefix reference")?;
                    let prefix_idx = prefix_id
                        .checked_sub(1)
                        .ok_or_else(|| generic_error!("Invalid zero tagset prefix reference"))?;
                    let prefix_tags = tagsets.get(prefix_idx).ok_or_else(|| {
                        generic_error!(
                            "Invalid tagset prefix reference {} (only {} tagsets parsed so far)",
                            prefix_id,
                            tagsets.len()
                        )
                    })?;
                    tags.extend(prefix_tags.iter().cloned());
                }
                std::cmp::Ordering::Equal => continue,
                std::cmp::Ordering::Greater => {
                    let idx = i64_to_usize(entry, "tag index")?;
                    let tag = tags_dict
                        .get(idx - 1)
                        .ok_or_else(|| generic_error!("Invalid tag index {} (dict size {})", idx, tags_dict.len()))?;
                    tags.push(tag.clone());
                }
            }
        }
        tagsets.push(tags);
        offset += count;
    }
    Ok(tagsets)
}

fn u64_to_usize(value: u64, field: &str) -> Result<usize, GenericError> {
    usize::try_from(value).map_err(|_| generic_error!("Invalid {}: {}", field, value))
}

fn i64_to_usize(value: i64, field: &str) -> Result<usize, GenericError> {
    usize::try_from(value).map_err(|_| generic_error!("Invalid negative {}: {}", field, value))
}

/// Parse resource sets from V3 resource dictionaries.
///
/// Each resource set is encoded as one length entry plus that many locally delta-encoded type/name dictionary indexes.
fn parse_resources(
    dict_resource_len: &[i64], dict_resource_type: &[i64], dict_resource_name: &[i64], resource_strings: &[String],
) -> Result<Vec<Vec<(String, String)>>, GenericError> {
    let mut resources = Vec::with_capacity(dict_resource_len.len());
    let mut offset = 0;

    for &count in dict_resource_len {
        let count = usize::try_from(count).map_err(|_| generic_error!("Invalid negative resource count: {}", count))?;
        if offset + count > dict_resource_type.len() || offset + count > dict_resource_name.len() {
            return Err(generic_error!("Resource set extends past resource dictionary arrays"));
        }

        let mut type_indices = dict_resource_type[offset..offset + count].to_vec();
        let mut name_indices = dict_resource_name[offset..offset + count].to_vec();
        delta_decode(&mut type_indices);
        delta_decode(&mut name_indices);

        let mut resource_set = Vec::with_capacity(count);
        for (&type_idx, &name_idx) in type_indices.iter().zip(name_indices.iter()) {
            let resource_type = resource_strings
                .get(resource_index(type_idx)?)
                .ok_or_else(|| generic_error!("Invalid resource type index {}", type_idx))?
                .clone();
            let resource_name = resource_strings
                .get(resource_index(name_idx)?)
                .ok_or_else(|| generic_error!("Invalid resource name index {}", name_idx))?
                .clone();
            resource_set.push((resource_type, resource_name));
        }

        resources.push(resource_set);
        offset += count;
    }

    Ok(resources)
}

fn resource_index(idx: i64) -> Result<usize, GenericError> {
    let idx = usize::try_from(idx).map_err(|_| generic_error!("Invalid negative resource index: {}", idx))?;
    idx.checked_sub(1)
        .ok_or_else(|| generic_error!("Invalid zero resource index"))
}

/// Read the next f64 value from the appropriate value array based on `value_type`.
fn read_value(
    value_type: u64, cursors: &mut V3ValueCursors, vals_sint64: &[i64], vals_float32: &[f32], vals_float64: &[f64],
) -> Result<f64, GenericError> {
    match value_type {
        V3_VALUE_TYPE_ZERO => Ok(0.0),
        V3_VALUE_TYPE_SINT64 => {
            let v = *vals_sint64
                .get(cursors.sint64)
                .ok_or_else(|| generic_error!("Ran out of sint64 values"))?;
            cursors.sint64 += 1;
            Ok(v as f64)
        }
        V3_VALUE_TYPE_FLOAT32 => {
            let v = *vals_float32
                .get(cursors.float32)
                .ok_or_else(|| generic_error!("Ran out of float32 values"))?;
            cursors.float32 += 1;
            Ok(v as f64)
        }
        V3_VALUE_TYPE_FLOAT64 => {
            let v = *vals_float64
                .get(cursors.float64)
                .ok_or_else(|| generic_error!("Ran out of float64 values"))?;
            cursors.float64 += 1;
            Ok(v)
        }
        _ => Err(generic_error!("Unknown v3 value type: {:#x}", value_type)),
    }
}

fn approx_eq_ratio_optional(a: Option<f64>, b: Option<f64>, ratio: f64) -> bool {
    match (a, b) {
        (Some(a), Some(b)) => a.approx_eq_ratio(&b, ratio),
        (None, None) => true,
        _ => false,
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn try_from_series_v1_parses_count_gauge_rate() {
        let body = br#"{"series":[
            {"metric":"a.count","points":[[100,5.0]],"tags":["env:prod"],"host":"h","type":"count","interval":0},
            {"metric":"a.gauge","points":[[101,12.0]],"tags":[],"host":"h","type":"gauge","interval":0},
            {"metric":"a.rate","points":[[102,3.0]],"tags":[],"host":"h","type":"rate","interval":10}
        ]}"#;

        let metrics = Metric::try_from_series_v1(body).expect("parse should succeed");
        assert_eq!(metrics.len(), 3);

        assert_eq!(metrics[0].context.name, "a.count");
        assert!(metrics[0].context.tags.contains(&"env:prod".to_string()));
        assert!(metrics[0].context.tags.contains(&"host:h".to_string()));
        assert_eq!(metrics[0].values, vec![(100, MetricValue::Count { value: 5.0 })]);

        assert_eq!(metrics[1].context.name, "a.gauge");
        assert_eq!(metrics[1].context.tags, vec!["host:h".to_string()]);
        assert_eq!(metrics[1].values, vec![(101, MetricValue::Gauge { value: 12.0 })]);

        assert_eq!(metrics[2].context.name, "a.rate");
        assert_eq!(metrics[2].context.tags, vec!["host:h".to_string()]);
        assert_eq!(
            metrics[2].values,
            vec![(
                102,
                MetricValue::Rate {
                    interval: 10,
                    value: 3.0
                }
            )]
        );
    }

    #[test]
    fn try_from_series_v1_omits_host_tag_when_empty() {
        // A payload without `host` is uncommon in practice but the parser must remain permissive — it omits the
        // `host:` tag rather than emitting an empty one.
        let body = br#"{"series":[
            {"metric":"m","points":[[1,1.0]],"tags":[],"type":"count","interval":0}
        ]}"#;

        let metrics = Metric::try_from_series_v1(body).expect("parse should succeed");
        assert!(!metrics[0].context.tags.iter().any(|t| t.starts_with("host:")));
    }

    #[test]
    fn try_from_series_v1_reinjects_device_tag() {
        let body = br#"{"series":[
            {"metric":"m","points":[[1,1.0]],"tags":["env:prod"],"host":"h","device":"eth0","type":"count","interval":0}
        ]}"#;

        let metrics = Metric::try_from_series_v1(body).expect("parse should succeed");
        assert!(metrics[0].context.tags.contains(&"env:prod".to_string()));
        assert!(metrics[0].context.tags.contains(&"device:eth0".to_string()));
    }

    #[test]
    fn try_from_series_v1_rejects_unknown_type() {
        let body = br#"{"series":[
            {"metric":"m","points":[[1,1.0]],"tags":[],"host":"h","type":"weird","interval":0}
        ]}"#;

        assert!(Metric::try_from_series_v1(body).is_err());
    }

    #[test]
    fn try_from_series_v2_folds_host_into_tags() {
        use datadog_protos::metrics::metric_payload::{
            MetricPoint, MetricSeries, MetricType as ProtoMetricType, Resource,
        };
        use datadog_protos::metrics::MetricPayload;

        let mut payload = MetricPayload::new();

        let mut series = MetricSeries::new();
        series.set_metric("my.metric".into());
        series.set_type(ProtoMetricType::COUNT);
        series.tags.push("env:prod".into());

        let mut host_res = Resource::new();
        host_res.set_type("host".into());
        host_res.set_name("server-1".into());
        series.resources.push(host_res);

        // Non-host resources must not be folded into tags.
        let mut device_res = Resource::new();
        device_res.set_type("device".into());
        device_res.set_name("eth0".into());
        series.resources.push(device_res);

        let mut point = MetricPoint::new();
        point.value = 1.0;
        point.timestamp = 1;
        series.points.push(point);

        payload.series.push(series);

        let metrics = Metric::try_from_series_v2(payload).expect("parse should succeed");
        assert_eq!(metrics.len(), 1);
        assert!(metrics[0].context.tags.contains(&"host:server-1".to_string()));
        assert!(metrics[0].context.tags.contains(&"env:prod".to_string()));
        assert!(!metrics[0].context.tags.iter().any(|t| t.starts_with("device:")));
    }

    #[test]
    fn try_from_sketch_folds_host_into_tags() {
        use datadog_protos::metrics::sketch_payload::Sketch;
        use datadog_protos::metrics::SketchPayload;

        let mut payload = SketchPayload::new();
        let mut sketch = Sketch::new();
        sketch.set_metric("my.metric".into());
        sketch.set_host("server-1".into());
        sketch.tags.push("env:prod".into());
        payload.sketches.push(sketch);

        let metrics = Metric::try_from_sketch(payload).expect("parse should succeed");
        assert_eq!(metrics.len(), 1);
        assert!(metrics[0].context.tags.contains(&"host:server-1".to_string()));
        assert!(metrics[0].context.tags.contains(&"env:prod".to_string()));
    }

    #[test]
    fn try_from_v3_folds_host_resource_into_tags() {
        use datadog_protos::metrics::v3::{MetricData, Payload};

        let mut data = MetricData::new();
        data.dictNameStr = length_prefixed_strings(["my.metric"]);
        data.dictTagStr = length_prefixed_strings(["env:prod"]);
        data.dictTagsets = vec![1, 1];
        data.dictResourceStr = length_prefixed_strings(["host", "server-1", "device", "eth0"]);
        data.dictResourceLen = vec![2];
        data.dictResourceType = vec![1, 2];
        data.dictResourceName = vec![2, 2];
        data.types = vec![V3_METRIC_TYPE_COUNT | V3_VALUE_TYPE_ZERO];
        data.nameRefs = vec![1];
        data.tagsetRefs = vec![1];
        data.resourcesRefs = vec![1];
        data.intervals = vec![0];
        data.numPoints = vec![1];
        data.timestamps = vec![1];

        let mut payload = Payload::new();
        payload.metricData = Some(data).into();

        let metrics = Metric::try_from_v3(payload).expect("parse should succeed");
        assert_eq!(metrics.len(), 1);
        assert!(metrics[0].context.tags.contains(&"env:prod".to_string()));
        assert!(metrics[0].context.tags.contains(&"host:server-1".to_string()));
        assert!(!metrics[0].context.tags.iter().any(|tag| tag.starts_with("device:")));
    }

    #[test]
    fn try_from_v3_decodes_integer_sketch_summary_order() {
        use datadog_protos::metrics::v3::{MetricData, Payload};

        let mut data = MetricData::new();
        data.dictNameStr = length_prefixed_strings(["my.sketch"]);
        data.types = vec![V3_METRIC_TYPE_SKETCH | V3_VALUE_TYPE_SINT64];
        data.nameRefs = vec![1];
        data.tagsetRefs = vec![0];
        data.resourcesRefs = vec![0];
        data.intervals = vec![0];
        data.numPoints = vec![1];
        data.timestamps = vec![123];
        // Agent V3 sketch ordering is sum, min, max, count when integer summaries share valsSint64.
        data.valsSint64 = vec![10, 1, 4, 4];
        data.sketchNumBins = vec![1];
        data.sketchBinKeys = vec![0];
        data.sketchBinCnts = vec![4];

        let mut payload = Payload::new();
        payload.metricData = Some(data).into();

        let metrics = Metric::try_from_v3(payload).expect("parse should succeed");
        assert_eq!(metrics.len(), 1);

        let MetricValue::Sketch { sketch } = &metrics[0].values[0].1 else {
            panic!("expected sketch value");
        };
        assert_eq!(sketch.count(), 4);
        assert_eq!(sketch.sum(), Some(10.0));
        assert_eq!(sketch.min(), Some(1.0));
        assert_eq!(sketch.max(), Some(4.0));
        assert_eq!(sketch.avg(), Some(2.5));
    }

    fn length_prefixed_strings(strings: impl IntoIterator<Item = &'static str>) -> Vec<u8> {
        let mut bytes = Vec::new();
        for s in strings {
            bytes.push(s.len() as u8);
            bytes.extend_from_slice(s.as_bytes());
        }
        bytes
    }

    #[test]
    fn parse_tagsets_resolves_positive_indices() {
        // Two independent tagsets, each a length followed by sorted+delta-encoded 1-based tag
        // indices. Tagset 1: [a:1]. Tagset 2: [a:1, b:2] -> indices [1, 2] -> deltas [1, 1].
        let tags_dict = vec!["a:1".to_string(), "b:2".to_string(), "c:3".to_string()];
        let dict_tagsets = vec![1, 1, /* tagset 1 */ 2, 1, 1 /* tagset 2 */];
        let tagsets = parse_tagsets(&dict_tagsets, &tags_dict).expect("parse should succeed");
        assert_eq!(
            tagsets,
            vec![vec!["a:1".to_string()], vec!["a:1".to_string(), "b:2".to_string()]]
        );
    }

    #[test]
    fn parse_tagsets_resolves_negative_prefix_reference() {
        // The Agent prefix-compresses composite tags: tagset 2 references tagset 1 (id 1) via a
        // negative `-1` entry plus its own tag `b:2` (index 2). The buffer `[-1, 2]` is sorted
        // then delta-encoded to `[-1, 3]`.
        let tags_dict = vec!["a:1".to_string(), "b:2".to_string()];
        let dict_tagsets = vec![
            1, 1, // tagset 1 (id 1): [a:1]
            2, -1, 3, // tagset 2 (id 2): prefix ref to id 1, then b:2
        ];
        let tagsets = parse_tagsets(&dict_tagsets, &tags_dict).expect("parse should succeed");
        assert_eq!(tagsets.len(), 2);
        assert_eq!(tagsets[0], vec!["a:1".to_string()]);
        // Tagset 2 resolves the prefix (a:1) and adds its own tag (b:2).
        assert_eq!(tagsets[1], vec!["a:1".to_string(), "b:2".to_string()]);
    }

    #[test]
    fn parse_tagsets_rejects_forward_prefix_reference() {
        // A prefix reference to a tagset that hasn't been parsed yet is invalid.
        let tags_dict = vec!["a:1".to_string()];
        let dict_tagsets = vec![1, -5];
        assert!(parse_tagsets(&dict_tagsets, &tags_dict).is_err());
    }
}