Function decode_aggregate_response

nodedb/src/event/alert/executor.rs:197–257

Decode the Data Plane's MessagePack response into (group_key, agg_value) pairs. The response is an `rmpv::Value::Array` of `Map` rows. Each row contains the group-by columns and the aggregate value (e.g., `"avg_temperature"`).

(
    payload: &[u8],
    group_by: &[String],
    agg_func: &str,
)

/// The response is an `rmpv::Value::Array` of `Map` rows. Each row contains
/// the group-by columns and the aggregate value (e.g., `"avg_temperature"`).
fn decode_aggregate_response(
    payload: &[u8],
    group_by: &[String],
    agg_func: &str,
) -> crate::Result<Vec<(String, f64)>> {
    if payload.is_empty() {
        return Ok(Vec::new());
    }

    let value: rmpv::Value =
        rmpv::decode::read_value(&mut &payload[..]).map_err(|e| crate::Error::Internal {
            detail: format!("decode aggregate response: {e}"),
        })?;

    let rows = match value {
        rmpv::Value::Array(rows) => rows,
        _ => return Ok(Vec::new()),
    };

    let mut results = Vec::with_capacity(rows.len());
    for row in &rows {
        let rmpv::Value::Map(fields) = row else {
            continue;
        };

        // Extract group key: concatenate group-by column values.
        let group_key = if group_by.is_empty() {
            "__all__".to_string()
        } else {
            group_by
                .iter()
                .filter_map(|col| {
                    fields.iter().find_map(|(k, v)| {
                        if k.as_str().is_some_and(|s| s == col) {
                            Some(v.as_str().unwrap_or("").to_string())
                        } else {
                            None
                        }
                    })
                })
                .collect::<Vec<_>>()
                .join("\0")
        };

        // Extract aggregate value by matching the agg key pattern: "{func}_{column}".
        let agg_value = fields.iter().find_map(|(k, v)| {
            let key_str = k.as_str()?;
            if key_str.starts_with(agg_func) {
                v.as_f64().or_else(|| v.as_i64().map(|i| i as f64))
            } else {
                None
            }
        });

        if let Some(val) = agg_value {
            results.push((group_key, val));
        }
    }