sentinel_core/

event.rs

//! Core event types for the perf-sentinel pipeline.

use std::sync::Arc;

use serde::{Deserialize, Serialize};

/// The type of I/O operation a span represents.
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum EventType {
    Sql,
    HttpOut,
    /// Message published to a broker (Kafka, `RabbitMQ`, Pulsar, SQS, ...).
    /// Producer side only, see `ingest::otlp::classify_io_event`.
    Messaging,
}

/// Source context for the span (which endpoint/method triggered it).
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct EventSource {
    pub endpoint: String,
    pub method: String,
}

/// Maximum allowed length for a `trace_id` or `span_id`.
///
/// OpenTelemetry specifies 32 hex chars for trace IDs and 16 for span IDs.
/// We allow up to 128 chars to accommodate non-standard formats.
pub const MAX_ID_LENGTH: usize = 128;

/// Truncate an ID field (`trace_id`, `span_id`) to [`MAX_ID_LENGTH`].
///
/// Uses char-boundary-aware truncation to avoid panicking on multi-byte UTF-8.
/// Delegates to [`truncate_field`] after a one-time clone to keep the
/// char-boundary walk in a single place.
#[must_use]
pub fn sanitize_id(id: &str) -> String {
    let mut s = id.to_string();
    truncate_field(&mut s, MAX_ID_LENGTH);
    s
}

/// Maximum length for the `service` field (bytes).
pub const MAX_SERVICE_LENGTH: usize = 256;

/// Maximum length for the `operation` field (bytes).
pub const MAX_OPERATION_LENGTH: usize = 256;

/// Maximum length for the `target` field (bytes).
/// The SQL normalizer has its own 64 KB limit; this provides
/// defense-in-depth at the ingestion boundary.
pub const MAX_TARGET_LENGTH: usize = 65_536;

/// Maximum length for `source.endpoint` and `source.method` (bytes).
pub const MAX_SOURCE_LENGTH: usize = 512;

/// Maximum length for `code_function` and `code_namespace` (bytes).
pub const MAX_CODE_FUNCTION_LENGTH: usize = 512;

/// Maximum length for `code_filepath` (bytes).
pub const MAX_CODE_FILEPATH_LENGTH: usize = 1024;

/// Maximum length for `code_namespace` (bytes).
pub const MAX_CODE_NAMESPACE_LENGTH: usize = 512;

/// Maximum length for a single instrumentation scope name (bytes).
/// Real OpenTelemetry scope names are short (`io.opentelemetry.spring-data-3.0`
/// is 33 bytes), so 256 leaves comfortable headroom while bounding the
/// memory amplification of the per-finding Vec clone.
pub const MAX_SCOPE_NAME_LENGTH: usize = 256;

/// Maximum number of instrumentation scopes captured per span. Matches
/// the OTLP parent-walk depth bound (`CODE_ATTRS_MAX_DEPTH = 8`). The
/// JSON ingest path has no such structural bound, so the cap fires there.
pub const MAX_INSTRUMENTATION_SCOPES: usize = 8;

/// Truncate a string to `max_len` bytes on a char boundary.
///
/// Shared between span-event sanitization and the daemon ack store
/// (`crate::daemon::ack`). Keep behavior strictly bytes-and-char-boundary,
/// do not add domain-specific normalization here.
pub(crate) fn truncate_field(s: &mut String, max_len: usize) {
    if s.len() <= max_len {
        return;
    }
    let mut end = max_len;
    while end > 0 && !s.is_char_boundary(end) {
        end -= 1;
    }
    s.truncate(end);
}

/// Strip URL query string, fragment, and userinfo from an endpoint in place.
///
/// `source.endpoint` falls back to the raw request URL when `http.route` is
/// absent, so a `?token=...`, `#...`, or `user:pass@` can carry a secret into
/// stored findings, the query API, the NDJSON archive, and the ack signature.
/// Drop them at the ingestion boundary. Route templates and query-less URLs
/// (the common case) are left untouched.
fn strip_endpoint_secrets(endpoint: &mut String) {
    // Drop query/fragment first, so a '?', '#', '://' or '@' living inside them
    // cannot be mistaken for authority structure below.
    if let Some(cut) = endpoint.find(['?', '#']) {
        endpoint.truncate(cut);
    }
    // Authority begins after a scheme `://` (only when that `://` precedes the
    // first path '/'), after a leading `//`, or at 0 for a scheme-less
    // `user:pass@host`. Strip its userinfo up to the last '@' before the path.
    let authority_start = if endpoint.starts_with("//") {
        2
    } else if let Some(i) = endpoint
        .find("://")
        .filter(|&i| !endpoint[..i].contains('/'))
    {
        i + 3
    } else {
        0
    };
    let authority_end = endpoint[authority_start..]
        .find('/')
        .map_or(endpoint.len(), |i| authority_start + i);
    if let Some(at) = endpoint[authority_start..authority_end].rfind('@') {
        endpoint.replace_range(authority_start..=authority_start + at, "");
    }
}

/// Drop the field if it contains any ASCII control character, otherwise truncate.
///
/// Mirrors the silent-drop posture used for `cloud.region` invalid values.
/// Control characters in `code.*` would render badly in TUI/CLI output and
/// could enable log-injection if any future log site emitted them raw.
fn sanitize_optional_arc_str(field: &mut Option<Arc<str>>, max_len: usize) {
    if field
        .as_deref()
        .is_some_and(crate::config::has_control_char)
    {
        *field = None;
        return;
    }
    if let Some(s) = field.as_ref()
        && s.len() > max_len
    {
        let mut tmp = s.to_string();
        truncate_field(&mut tmp, max_len);
        *field = Some(Arc::from(tmp));
    }
}

/// Truncate an `Arc<str>` field in place via alloc-and-replace.
///
/// `Arc<str>` is immutable, so we materialize a `String`, truncate it
/// on a char boundary, and rebuild a fresh `Arc::<str>::from(String)`
/// that reuses the buffer (no double allocation).
fn truncate_arc_str(field: &mut Arc<str>, max_len: usize) {
    if field.len() <= max_len {
        return;
    }
    let mut tmp = field.to_string();
    truncate_field(&mut tmp, max_len);
    *field = Arc::from(tmp);
}

/// Drop entries with control characters, truncate the remainder to
/// `max_len` and cap the Vec at `max_count`.
///
/// Used for `instrumentation_scopes` (OpenTelemetry scope names from
/// arbitrary agents, including the JSON ingest path which has no
/// structural depth bound). Bounds both the per-element and per-event
/// memory amplification when those scope names propagate into the
/// per-finding clone.
fn sanitize_arc_str_vec(field: &mut Vec<Arc<str>>, max_len: usize, max_count: usize) {
    field.retain(|s| !crate::config::has_control_char(s));
    if field.len() > max_count {
        field.truncate(max_count);
    }
    for s in field.iter_mut() {
        truncate_arc_str(s, max_len);
    }
}

/// Sanitize all string fields on a [`SpanEvent`] to enforce length limits.
///
/// Maximum length for the `timestamp` field (bytes).
/// ISO 8601 with microseconds and timezone is at most ~30 chars.
const MAX_TIMESTAMP_LENGTH: usize = 64;

/// Called at every ingestion boundary (OTLP, JSON, Jaeger, Zipkin) to
/// prevent unbounded memory growth from oversized attribute values.
/// Also truncates IDs (`trace_id`, `span_id`, `parent_span_id`) that
/// are not already sanitized at the ingestion boundary for some formats
/// (Jaeger, Zipkin, native JSON).
pub fn sanitize_span_event(event: &mut SpanEvent) {
    truncate_field(&mut event.timestamp, MAX_TIMESTAMP_LENGTH);
    truncate_field(&mut event.trace_id, MAX_ID_LENGTH);
    truncate_field(&mut event.span_id, MAX_ID_LENGTH);
    if let Some(ref mut pid) = event.parent_span_id {
        truncate_field(pid, MAX_ID_LENGTH);
    }
    // is_valid_region_id at OTLP/JSON ingest already caps cloud_region at
    // 64 chars, but Jaeger/Zipkin paths leave the field None. Funnel
    // through the same control-char + truncate helper as code_* for
    // defense-in-depth on hand-crafted inputs.
    sanitize_optional_arc_str(&mut event.cloud_region, MAX_ID_LENGTH);
    // A grouping value reaches the terminal and the dashboard, and its key
    // is the label shown beside it. Both go through the same guard as the
    // service name, and either one rejected or blank drops the whole pair:
    // a half-pair renders as an unlabelled identity. Native JSON ingest
    // deserializes `grouping` verbatim, so both sides are untrusted here.
    event.grouping.retain_mut(|attr| {
        let mut key = Some(Arc::clone(&attr.key));
        let mut value = Some(Arc::clone(&attr.value));
        sanitize_optional_arc_str(&mut key, MAX_SERVICE_LENGTH);
        sanitize_optional_arc_str(&mut value, MAX_SERVICE_LENGTH);
        match (key, value) {
            (Some(k), Some(v)) if !k.is_empty() && !v.is_empty() => {
                attr.key = k;
                attr.value = v;
                true
            }
            _ => false,
        }
    });
    event
        .grouping
        .truncate(crate::config::MAX_GROUPING_ATTRIBUTES);
    sanitize_optional_arc_str(&mut event.link_trace_id, MAX_ID_LENGTH);
    truncate_arc_str(&mut event.service, MAX_SERVICE_LENGTH);
    truncate_field(&mut event.operation, MAX_OPERATION_LENGTH);
    // A messaging destination reaches the finding template verbatim, with
    // no tokenizer or URL parser in between. Scheme-gated on purpose, see
    // `docs/design/02-NORMALIZATION.md`.
    if event.event_type == EventType::Messaging && event.target.contains("://") {
        strip_endpoint_secrets(&mut event.target);
    }
    truncate_field(&mut event.target, MAX_TARGET_LENGTH);
    strip_endpoint_secrets(&mut event.source.endpoint);
    truncate_field(&mut event.source.endpoint, MAX_SOURCE_LENGTH);
    truncate_field(&mut event.source.method, MAX_SOURCE_LENGTH);
    sanitize_optional_arc_str(&mut event.code_function, MAX_CODE_FUNCTION_LENGTH);
    sanitize_optional_arc_str(&mut event.code_filepath, MAX_CODE_FILEPATH_LENGTH);
    sanitize_optional_arc_str(&mut event.code_namespace, MAX_CODE_NAMESPACE_LENGTH);
    sanitize_arc_str_vec(
        &mut event.instrumentation_scopes,
        MAX_SCOPE_NAME_LENGTH,
        MAX_INSTRUMENTATION_SCOPES,
    );
}

/// Source code location extracted from `OTel` `code.*` span attributes.
///
/// Not all instrumentation agents emit these attributes. When present,
/// they allow findings to point to the exact function and file where the
/// anti-pattern originates.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct CodeLocation {
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub function: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub filepath: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub lineno: Option<u32>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub namespace: Option<String>,
}

impl CodeLocation {
    /// Returns `true` when all fields are `None`.
    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.function.is_none()
            && self.filepath.is_none()
            && self.lineno.is_none()
            && self.namespace.is_none()
    }

    /// Render the location as `namespace.function (filepath:lineno)`,
    /// omitting absent parts. Returns an empty string when the location
    /// has nothing displayable, so callers can skip the line entirely
    /// rather than printing a bare `Source:` label.
    ///
    /// Single source of truth for the CLI text output, the SARIF
    /// `physicalLocation` message, and the TUI detail panel.
    #[must_use]
    pub fn display_string(&self) -> String {
        let mut src = String::new();
        if let Some(ref ns) = self.namespace {
            src.push_str(ns);
            src.push('.');
        }
        if let Some(ref func) = self.function {
            src.push_str(func);
        }
        let has_name = !src.is_empty();
        if let Some(ref fp) = self.filepath {
            if has_name {
                src.push_str(" (");
            }
            src.push_str(fp);
            if let Some(ln) = self.lineno {
                src.push(':');
                src.push_str(&ln.to_string());
            }
            if has_name {
                src.push(')');
            }
        }
        src
    }
}

/// One attribute captured for grouping, with the attribute name it came
/// from so an operator can tell `tenant.id=acme` from `k8s.namespace.name=acme`.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct GroupingAttribute {
    pub key: Arc<str>,
    pub value: Arc<str>,
}

/// A single span event representing an I/O operation (SQL query, HTTP call).
///
/// Strings repeated across events of the same workload are stored as
/// `Arc<str>` for cheap clones, per-event-unique strings stay as `String`.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SpanEvent {
    pub timestamp: String,
    pub trace_id: String,
    pub span_id: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub parent_span_id: Option<String>,
    /// Trace of the producer whose message triggered this span, from the `OTel`
    /// span link of the nearest CONSUMER ancestor. The two traces stay
    /// separate: this only makes the chain navigable across the broker.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub link_trace_id: Option<Arc<str>>,
    pub service: Arc<str>,
    /// Attributes captured for grouping, in `[detection] grouping_attributes`
    /// order, absent ones skipped. The first entry is the dimension identity
    /// keys use, the rest are kept because a report must show the telemetry
    /// it received, not only the value that won.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub grouping: Vec<GroupingAttribute>,
    /// Cloud region this span was emitted from, sourced from the `OTel`
    /// `cloud.region` resource attribute (or span attribute as fallback).
    ///
    /// Used by the carbon scoring stage to apply per-region carbon
    /// intensity coefficients in multi-region deployments. `None` when
    /// the attribute is absent or when ingesting from formats that don't
    /// carry it (`Jaeger`, `Zipkin`, raw `JSON` without explicit field).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub cloud_region: Option<Arc<str>>,
    #[serde(rename = "type")]
    pub event_type: EventType,
    /// SQL: `db.system` for OTLP (e.g. "postgresql"), verb for native JSON.
    /// HTTP: request method (e.g. "GET").
    pub operation: String,
    pub target: String,
    pub duration_us: u64,
    pub source: EventSource,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub status_code: Option<u16>,
    /// Payload size in bytes: the HTTP response body
    /// (`http.response.body.size`, legacy `http.response_content_length`) or
    /// the published message body (`messaging.message.body.size`). Drives the
    /// HTTP carbon size tiers and network transport estimation. `None` for SQL
    /// spans or when the attribute is absent.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub response_size_bytes: Option<u64>,
    /// `OTel` `code.function` attribute.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub code_function: Option<Arc<str>>,
    /// `OTel` `code.filepath` attribute.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub code_filepath: Option<Arc<str>>,
    /// `OTel` `code.lineno` attribute.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub code_lineno: Option<u32>,
    /// `OTel` `code.namespace` attribute (e.g. Java package).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub code_namespace: Option<Arc<str>>,
    /// OpenTelemetry instrumentation scope names captured at ingest time:
    /// the leaf span's scope at index 0, then each unique ancestor scope
    /// up to a bounded depth. Lets framework detection identify Spring
    /// Data, Hibernate, Quarkus, Helidon and friends from the
    /// `io.opentelemetry.<module>` strings emitted by the agent, without
    /// relying on user-code naming conventions.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub instrumentation_scopes: Vec<Arc<str>>,
}

impl SpanEvent {
    /// Value that separates deployments in reports and recurrence views.
    #[must_use]
    pub fn effective_grouping(&self) -> Option<&GroupingAttribute> {
        self.grouping.first()
    }

    /// Just the value, for the identity keys that do not display it.
    #[must_use]
    pub fn grouping_value(&self) -> Option<&str> {
        self.grouping.first().map(|g| g.value.as_ref())
    }

    /// Borrowed `(key, value)` for a `HashMap` partition key. Both halves,
    /// never the value alone: `tenant.id=prod` and `k8s.namespace.name=prod`
    /// are two deployments, not one.
    #[must_use]
    pub fn grouping_identity(&self) -> Option<(&str, &str)> {
        self.grouping
            .first()
            .map(|g| (g.key.as_ref(), g.value.as_ref()))
    }

    /// Build a [`CodeLocation`] from this span's `code_*` fields.
    ///
    /// Returns `None` when all four fields are absent.
    #[must_use]
    pub fn code_location(&self) -> Option<CodeLocation> {
        if self.code_function.is_none()
            && self.code_filepath.is_none()
            && self.code_lineno.is_none()
            && self.code_namespace.is_none()
        {
            return None;
        }
        Some(CodeLocation {
            function: self.code_function.as_deref().map(String::from),
            filepath: self.code_filepath.as_deref().map(String::from),
            lineno: self.code_lineno,
            namespace: self.code_namespace.as_deref().map(String::from),
        })
    }
}

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

    #[test]
    fn code_location_display_string_full() {
        let loc = CodeLocation {
            function: Some("OrderItemRepository.findByOrderId".to_string()),
            filepath: Some("order-service/src/main/java/OrderItemRepository.java".to_string()),
            lineno: Some(42),
            namespace: Some("com.example.order.repository".to_string()),
        };
        assert_eq!(
            loc.display_string(),
            "com.example.order.repository.OrderItemRepository.findByOrderId \
             (order-service/src/main/java/OrderItemRepository.java:42)"
        );
    }

    #[test]
    fn code_location_display_string_function_only() {
        let loc = CodeLocation {
            function: Some("fetchUser".to_string()),
            filepath: None,
            lineno: None,
            namespace: None,
        };
        assert_eq!(loc.display_string(), "fetchUser");
    }

    #[test]
    fn code_location_display_string_filepath_only() {
        let loc = CodeLocation {
            function: None,
            filepath: Some("src/main.rs".to_string()),
            lineno: Some(7),
            namespace: None,
        };
        // No function or namespace, so no parentheses wrap; filepath
        // still emits with its line number.
        assert_eq!(loc.display_string(), "src/main.rs:7");
    }

    #[test]
    fn code_location_display_string_empty_when_all_none() {
        let loc = CodeLocation {
            function: None,
            filepath: None,
            lineno: None,
            namespace: None,
        };
        assert_eq!(loc.display_string(), "");
        assert!(loc.is_empty());
    }

    fn sample_sql_json() -> &'static str {
        r#"{
            "timestamp": "2025-07-10T14:32:01.123Z",
            "trace_id": "abc123-def456",
            "span_id": "span-789",
            "service": "order-svc",
            "type": "sql",
            "operation": "SELECT",
            "target": "SELECT * FROM order_item WHERE order_id = 42",
            "duration_us": 1200,
            "source": {
                "endpoint": "POST /api/orders/42/submit",
                "method": "OrderService::create_order"
            }
        }"#
    }

    fn sample_http_json() -> &'static str {
        r#"{
            "timestamp": "2025-07-10T14:32:01.456Z",
            "trace_id": "abc123-def456",
            "span_id": "span-790",
            "service": "order-svc",
            "type": "http_out",
            "operation": "GET",
            "target": "http://user-svc:5000/api/users/user-123",
            "duration_us": 15000,
            "status_code": 200,
            "source": {
                "endpoint": "POST /api/orders/42/submit",
                "method": "OrderService::create_order"
            }
        }"#
    }

    #[test]
    fn deserialize_sql_event() {
        let event: SpanEvent = serde_json::from_str(sample_sql_json()).unwrap();
        assert_eq!(event.event_type, EventType::Sql);
        assert_eq!(event.trace_id, "abc123-def456");
        assert_eq!(&*event.service, "order-svc");
        assert_eq!(event.target, "SELECT * FROM order_item WHERE order_id = 42");
        assert_eq!(event.duration_us, 1200);
        assert!(event.status_code.is_none());
    }

    #[test]
    fn deserialize_http_event() {
        let event: SpanEvent = serde_json::from_str(sample_http_json()).unwrap();
        assert_eq!(event.event_type, EventType::HttpOut);
        assert_eq!(event.status_code, Some(200));
        assert_eq!(event.source.endpoint, "POST /api/orders/42/submit");
    }

    #[test]
    fn serde_roundtrip_sql() {
        let event: SpanEvent = serde_json::from_str(sample_sql_json()).unwrap();
        let json = serde_json::to_string(&event).unwrap();
        let back: SpanEvent = serde_json::from_str(&json).unwrap();
        assert_eq!(event, back);
    }

    #[test]
    fn serde_roundtrip_http() {
        let event: SpanEvent = serde_json::from_str(sample_http_json()).unwrap();
        let json = serde_json::to_string(&event).unwrap();
        let back: SpanEvent = serde_json::from_str(&json).unwrap();
        assert_eq!(event, back);
    }

    #[test]
    fn sql_event_omits_status_code_in_json() {
        let event: SpanEvent = serde_json::from_str(sample_sql_json()).unwrap();
        let json = serde_json::to_string(&event).unwrap();
        assert!(!json.contains("status_code"));
    }

    #[test]
    fn deserialize_event_without_cloud_region_defaults_to_none() {
        let event: SpanEvent = serde_json::from_str(sample_sql_json()).unwrap();
        assert!(event.cloud_region.is_none());
    }

    #[test]
    fn deserialize_event_without_grouping_defaults_to_empty() {
        let event: SpanEvent = serde_json::from_str(sample_sql_json()).unwrap();
        assert!(event.grouping.is_empty());
        assert_eq!(event.grouping_value(), None);
    }

    /// The identity value is the FIRST captured attribute, config order, and
    /// the others stay readable rather than being dropped.
    #[test]
    fn grouping_value_is_the_first_captured_attribute() {
        let mut event = make_event_with_field("service", "svc");
        event.grouping = vec![
            GroupingAttribute {
                key: Arc::from("k8s.namespace.name"),
                value: Arc::from("prod-eu"),
            },
            GroupingAttribute {
                key: Arc::from("service.namespace"),
                value: Arc::from("payments"),
            },
        ];
        assert_eq!(event.grouping_value(), Some("prod-eu"));
        assert_eq!(
            event.effective_grouping().map(|g| g.key.as_ref()),
            Some("k8s.namespace.name"),
            "the operator must be able to tell which attribute won"
        );
        assert_eq!(event.grouping[1].value.as_ref(), "payments");

        event.grouping.clear();
        assert_eq!(event.grouping_value(), None);
    }

    #[test]
    fn serde_roundtrip_with_grouping() {
        let mut value: serde_json::Value = serde_json::from_str(sample_sql_json()).unwrap();
        value["grouping"] = serde_json::json!([
            {"key": "tenant.id", "value": "acme"},
            {"key": "k8s.namespace.name", "value": "shared-cluster"},
        ]);

        let event: SpanEvent = serde_json::from_value(value).unwrap();
        assert_eq!(event.grouping_value(), Some("acme"));
        assert_eq!(event.grouping.len(), 2);

        let serialized = serde_json::to_string(&event).unwrap();
        let back: SpanEvent = serde_json::from_str(&serialized).unwrap();
        assert_eq!(back, event);
    }

    #[test]
    fn serde_roundtrip_with_cloud_region() {
        let json = r#"{
            "timestamp": "2025-07-10T14:32:01.123Z",
            "trace_id": "abc123-def456",
            "span_id": "span-789",
            "service": "order-svc",
            "cloud_region": "eu-west-3",
            "type": "sql",
            "operation": "SELECT",
            "target": "SELECT 1",
            "duration_us": 1200,
            "source": {
                "endpoint": "POST /api/orders/42/submit",
                "method": "OrderService::create_order"
            }
        }"#;
        let event: SpanEvent = serde_json::from_str(json).unwrap();
        assert_eq!(event.cloud_region.as_deref(), Some("eu-west-3"));
        let serialized = serde_json::to_string(&event).unwrap();
        assert!(serialized.contains("\"cloud_region\":\"eu-west-3\""));
        let back: SpanEvent = serde_json::from_str(&serialized).unwrap();
        assert_eq!(event, back);
    }

    #[test]
    fn cloud_region_omitted_when_none() {
        let event: SpanEvent = serde_json::from_str(sample_sql_json()).unwrap();
        let json = serde_json::to_string(&event).unwrap();
        assert!(!json.contains("cloud_region"));
    }

    #[test]
    fn sanitize_id_short_unchanged() {
        assert_eq!(sanitize_id("abc-123"), "abc-123");
    }

    #[test]
    fn sanitize_id_truncates_long() {
        let long = "a".repeat(200);
        let result = sanitize_id(&long);
        assert_eq!(result.len(), MAX_ID_LENGTH);
    }

    #[test]
    fn sanitize_id_exact_length_unchanged() {
        let exact = "b".repeat(MAX_ID_LENGTH);
        assert_eq!(sanitize_id(&exact), exact);
    }

    #[test]
    fn sanitize_id_multibyte_no_panic() {
        // 4-byte emoji repeated to exceed MAX_ID_LENGTH (200 bytes total)
        let id = "\u{1F600}".repeat(50);
        assert!(id.len() > MAX_ID_LENGTH);
        let result = sanitize_id(&id);
        assert!(result.len() <= MAX_ID_LENGTH);
        // Must be valid UTF-8 (would panic in .to_string() if not)
        assert!(result.is_char_boundary(result.len()));
    }

    #[test]
    fn sanitize_id_two_byte_chars_no_panic() {
        // 2-byte UTF-8 chars: é is 2 bytes
        let id = "é".repeat(100); // 200 bytes
        let result = sanitize_id(&id);
        assert!(result.len() <= MAX_ID_LENGTH);
        // Result should contain whole chars only (even byte count for 2-byte chars)
        assert_eq!(result.len() % 2, 0);
    }

    // ------------------------------------------------------------------
    // sanitize_span_event
    // ------------------------------------------------------------------

    fn make_event_with_field(field: &str, value: &str) -> SpanEvent {
        let mut event: SpanEvent = serde_json::from_str(sample_sql_json()).unwrap();
        match field {
            "service" => event.service = Arc::from(value),
            "operation" => event.operation = value.to_string(),
            "target" => event.target = value.to_string(),
            "endpoint" => event.source.endpoint = value.to_string(),
            "method" => event.source.method = value.to_string(),
            _ => panic!("unknown field: {field}"),
        }
        event
    }

    #[test]
    fn sanitize_bounds_link_trace_id() {
        let mut event = make_event_with_field("service", "svc");
        event.link_trace_id = Some(Arc::from("x".repeat(500)));
        sanitize_span_event(&mut event);
        assert_eq!(
            event.link_trace_id.as_deref().map(str::len),
            Some(MAX_ID_LENGTH)
        );

        event.link_trace_id = Some(Arc::from("trace\u{7}id"));
        sanitize_span_event(&mut event);
        assert_eq!(event.link_trace_id, None);
    }

    #[test]
    fn sanitize_strips_credentials_from_a_messaging_destination() {
        // A destination lands verbatim in the finding template, which
        // reaches the archive, the dashboard and the terminal.
        let mut event = make_event_with_field("target", "amqp://svc:hunter2@rabbit:5672/orders");
        event.event_type = EventType::Messaging;
        sanitize_span_event(&mut event);
        assert!(
            !event.target.contains("hunter2"),
            "credential survived: {}",
            event.target
        );
        assert!(event.target.contains("rabbit"), "host must survive");
    }

    #[test]
    fn sanitize_keeps_an_sqs_arn_intact() {
        // The account id is deliberately preserved: it is what keeps two
        // AWS accounts from merging into one template.
        let arn = "arn:aws:sqs:eu-west-3:123456789012:orders";
        let mut event = make_event_with_field("target", arn);
        event.event_type = EventType::Messaging;
        sanitize_span_event(&mut event);
        assert_eq!(event.target, arn);
    }

    #[test]
    fn sanitize_keeps_scheme_less_destinations_verbatim() {
        // These are names, not URLs. Running the authority parser over
        // them merges distinct destinations into one finding target.
        for name in [
            "ORDERS@QM1",   // IBM MQ queue@qmgr
            "logs.#",       // RabbitMQ routing key
            "my.topic?v=2", // a literal '?' in a queue name
            "orders",
        ] {
            let mut event = make_event_with_field("target", name);
            event.event_type = EventType::Messaging;
            sanitize_span_event(&mut event);
            assert_eq!(event.target, name, "rewrote a legitimate destination");
        }
    }

    #[test]
    fn sanitize_truncates_long_service() {
        let mut event = make_event_with_field("service", &"x".repeat(500));
        sanitize_span_event(&mut event);
        assert!(event.service.len() <= MAX_SERVICE_LENGTH);
    }

    /// A grouping value reaches the terminal and the dashboard, and its key
    /// comes from operator config: both are bounded, and a control character
    /// drops the whole pair rather than half of it.
    #[test]
    fn sanitize_bounds_and_rejects_grouping_values() {
        let mut event = make_event_with_field("service", "svc");
        event.grouping = vec![
            GroupingAttribute {
                key: Arc::from("k8s.namespace.name"),
                value: Arc::from("s".repeat(500)),
            },
            GroupingAttribute {
                key: Arc::from("tenant.id"),
                value: Arc::from("prod\u{7}hidden"),
            },
            GroupingAttribute {
                key: Arc::from("service.namespace"),
                value: Arc::from(""),
            },
        ];

        sanitize_span_event(&mut event);

        assert_eq!(event.grouping.len(), 1, "{:?}", event.grouping);
        assert_eq!(
            event.grouping_value().map(str::len),
            Some(MAX_SERVICE_LENGTH),
            "an over-long value is truncated, not dropped"
        );
    }

    #[test]
    fn sanitize_caps_oversize_grouping_vec() {
        let mut event = make_event_with_field("service", "svc");
        event.grouping = (0..32)
            .map(|i| GroupingAttribute {
                key: Arc::from(format!("tenant.dimension.{i}")),
                value: Arc::from(format!("value-{i}")),
            })
            .collect();

        sanitize_span_event(&mut event);

        assert_eq!(event.grouping.len(), crate::config::MAX_GROUPING_ATTRIBUTES);
        assert_eq!(event.grouping[0].key.as_ref(), "tenant.dimension.0");
    }

    #[test]
    fn sanitize_truncates_long_operation() {
        let mut event = make_event_with_field("operation", &"x".repeat(500));
        sanitize_span_event(&mut event);
        assert!(event.operation.len() <= MAX_OPERATION_LENGTH);
    }

    #[test]
    fn sanitize_truncates_long_target() {
        let mut event = make_event_with_field("target", &"x".repeat(100_000));
        sanitize_span_event(&mut event);
        assert!(event.target.len() <= MAX_TARGET_LENGTH);
    }

    #[test]
    fn sanitize_truncates_long_endpoint() {
        let mut event = make_event_with_field("endpoint", &"x".repeat(1000));
        sanitize_span_event(&mut event);
        assert!(event.source.endpoint.len() <= MAX_SOURCE_LENGTH);
    }

    #[test]
    fn sanitize_truncates_long_method() {
        let mut event = make_event_with_field("method", &"x".repeat(1000));
        sanitize_span_event(&mut event);
        assert!(event.source.method.len() <= MAX_SOURCE_LENGTH);
    }

    #[test]
    fn sanitize_strips_endpoint_query_and_userinfo() {
        for (raw, expected) in [
            (
                "https://svc/api/users?token=SECRET",
                "https://svc/api/users",
            ),
            ("http://user:pass@host/cb?code=abc#frag", "http://host/cb"),
            ("/api/reset?token=SECRET", "/api/reset"),
            // '@' or '://' in the query is not userinfo (path-less URLs).
            ("http://svc?filter=a@b&token=SECRET", "http://svc"),
            ("http://host?a=x@y", "http://host"),
            ("//user:tok@bmc/cb?redirect=https://app", "//bmc/cb"),
            ("user:pass@host/cb?next=https://x", "host/cb"),
            // Scheme-less and protocol-relative userinfo still stripped.
            ("user:pass@host/path", "host/path"),
            ("//svc-user:tok@order-svc/cb", "//order-svc/cb"),
            // '@' outside an authority (path) is not userinfo.
            ("/users/a@b.example/orders", "/users/a@b.example/orders"),
            // Common cases: route templates and query-less URLs are untouched.
            ("POST /api/orders/{id}", "POST /api/orders/{id}"),
            (
                "http://order-svc/api/orders/42",
                "http://order-svc/api/orders/42",
            ),
        ] {
            let mut event = make_event_with_field("endpoint", raw);
            sanitize_span_event(&mut event);
            assert_eq!(event.source.endpoint, expected, "input: {raw}");
        }
    }

    #[test]
    fn sanitize_short_fields_unchanged() {
        let mut event: SpanEvent = serde_json::from_str(sample_sql_json()).unwrap();
        let before = event.clone();
        sanitize_span_event(&mut event);
        assert_eq!(event, before);
    }

    #[test]
    fn sanitize_multibyte_char_boundary() {
        // Service with 4-byte emojis that would split mid-char at MAX_SERVICE_LENGTH
        let mut event = make_event_with_field("service", &"\u{1F600}".repeat(100));
        sanitize_span_event(&mut event);
        assert!(event.service.len() <= MAX_SERVICE_LENGTH);
        // Must be valid UTF-8 (String invariant guarantees this, but verify)
        assert!(event.service.is_char_boundary(event.service.len()));
    }

    // ------------------------------------------------------------------
    // CodeLocation and code_* fields
    // ------------------------------------------------------------------

    #[test]
    fn code_location_is_empty_when_all_none() {
        let loc = CodeLocation {
            function: None,
            filepath: None,
            lineno: None,
            namespace: None,
        };
        assert!(loc.is_empty());
    }

    #[test]
    fn code_location_not_empty_with_function() {
        let loc = CodeLocation {
            function: Some("processItems".to_string()),
            filepath: None,
            lineno: None,
            namespace: None,
        };
        assert!(!loc.is_empty());
    }

    #[test]
    fn span_event_code_location_none_when_all_absent() {
        let event: SpanEvent = serde_json::from_str(sample_sql_json()).unwrap();
        assert!(event.code_location().is_none());
    }

    #[test]
    fn span_event_code_location_some_when_present() {
        let mut event: SpanEvent = serde_json::from_str(sample_sql_json()).unwrap();
        event.code_function = Some(Arc::from("processItems"));
        event.code_filepath = Some(Arc::from("src/OrderService.java"));
        event.code_lineno = Some(42);
        event.code_namespace = Some(Arc::from("com.example"));
        let loc = event.code_location().unwrap();
        assert_eq!(loc.function.as_deref(), Some("processItems"));
        assert_eq!(loc.filepath.as_deref(), Some("src/OrderService.java"));
        assert_eq!(loc.lineno, Some(42));
        assert_eq!(loc.namespace.as_deref(), Some("com.example"));
    }

    #[test]
    fn serde_roundtrip_with_code_fields() {
        let json = r#"{
            "timestamp": "2025-07-10T14:32:01.123Z",
            "trace_id": "abc123",
            "span_id": "span-1",
            "service": "svc",
            "type": "sql",
            "operation": "SELECT",
            "target": "SELECT 1",
            "duration_us": 100,
            "source": { "endpoint": "GET /test", "method": "test" },
            "code_function": "processItems",
            "code_filepath": "src/OrderService.java",
            "code_lineno": 42,
            "code_namespace": "com.example"
        }"#;
        let event: SpanEvent = serde_json::from_str(json).unwrap();
        assert_eq!(event.code_function.as_deref(), Some("processItems"));
        assert_eq!(event.code_lineno, Some(42));
        let serialized = serde_json::to_string(&event).unwrap();
        let back: SpanEvent = serde_json::from_str(&serialized).unwrap();
        assert_eq!(event, back);
    }

    #[test]
    fn code_fields_omitted_when_none() {
        let event: SpanEvent = serde_json::from_str(sample_sql_json()).unwrap();
        let json = serde_json::to_string(&event).unwrap();
        assert!(!json.contains("code_function"));
        assert!(!json.contains("code_filepath"));
        assert!(!json.contains("code_lineno"));
        assert!(!json.contains("code_namespace"));
    }

    #[test]
    fn sanitize_truncates_long_code_function() {
        let mut event: SpanEvent = serde_json::from_str(sample_sql_json()).unwrap();
        event.code_function = Some(Arc::from("x".repeat(1000)));
        sanitize_span_event(&mut event);
        assert!(event.code_function.as_ref().unwrap().len() <= MAX_CODE_FUNCTION_LENGTH);
    }

    #[test]
    fn sanitize_truncates_long_code_filepath() {
        let mut event: SpanEvent = serde_json::from_str(sample_sql_json()).unwrap();
        event.code_filepath = Some(Arc::from("x".repeat(2000)));
        sanitize_span_event(&mut event);
        assert!(event.code_filepath.as_ref().unwrap().len() <= MAX_CODE_FILEPATH_LENGTH);
    }

    #[test]
    fn sanitize_drops_code_function_with_control_char() {
        let mut event: SpanEvent = serde_json::from_str(sample_sql_json()).unwrap();
        event.code_function = Some(Arc::from("findItems\x1b[31m"));
        sanitize_span_event(&mut event);
        assert!(event.code_function.is_none());
    }

    #[test]
    fn sanitize_drops_code_filepath_with_newline() {
        let mut event: SpanEvent = serde_json::from_str(sample_sql_json()).unwrap();
        event.code_filepath = Some(Arc::from("src/main.rs\nINJECT"));
        sanitize_span_event(&mut event);
        assert!(event.code_filepath.is_none());
    }

    #[test]
    fn sanitize_drops_code_namespace_with_del() {
        let mut event: SpanEvent = serde_json::from_str(sample_sql_json()).unwrap();
        event.code_namespace = Some(Arc::from("com.foo\x7fX"));
        sanitize_span_event(&mut event);
        assert!(event.code_namespace.is_none());
    }

    #[test]
    fn sanitize_keeps_clean_code_fields() {
        let mut event: SpanEvent = serde_json::from_str(sample_sql_json()).unwrap();
        event.code_function = Some(Arc::from("findItems"));
        event.code_filepath = Some(Arc::from("src/main/java/com/foo/Repo.java"));
        event.code_namespace = Some(Arc::from("com.foo.Repo"));
        sanitize_span_event(&mut event);
        assert_eq!(event.code_function.as_deref(), Some("findItems"));
        assert_eq!(
            event.code_filepath.as_deref(),
            Some("src/main/java/com/foo/Repo.java")
        );
        assert_eq!(event.code_namespace.as_deref(), Some("com.foo.Repo"));
    }

    // ── instrumentation_scopes sanitization ─────────────────────

    #[test]
    fn sanitize_truncates_long_instrumentation_scope() {
        let mut event: SpanEvent = serde_json::from_str(sample_sql_json()).unwrap();
        event.instrumentation_scopes = vec![Arc::from("x".repeat(1024))];
        sanitize_span_event(&mut event);
        assert_eq!(event.instrumentation_scopes.len(), 1);
        assert!(event.instrumentation_scopes[0].len() <= MAX_SCOPE_NAME_LENGTH);
    }

    #[test]
    fn sanitize_drops_instrumentation_scope_with_control_char() {
        let mut event: SpanEvent = serde_json::from_str(sample_sql_json()).unwrap();
        event.instrumentation_scopes = vec![
            Arc::from("io.opentelemetry.spring-data"),
            Arc::from("\x1b[31mio.opentelemetry.evil\x1b[0m"),
            Arc::from("io.opentelemetry.hibernate"),
        ];
        sanitize_span_event(&mut event);
        let scopes: Vec<&str> = event
            .instrumentation_scopes
            .iter()
            .map(AsRef::as_ref)
            .collect();
        assert_eq!(
            scopes,
            vec!["io.opentelemetry.spring-data", "io.opentelemetry.hibernate"]
        );
    }

    #[test]
    fn sanitize_caps_oversize_instrumentation_scopes_vec() {
        let mut event: SpanEvent = serde_json::from_str(sample_sql_json()).unwrap();
        event.instrumentation_scopes = (0..32)
            .map(|i| Arc::from(format!("io.opentelemetry.scope-{i}")))
            .collect();
        sanitize_span_event(&mut event);
        assert_eq!(
            event.instrumentation_scopes.len(),
            MAX_INSTRUMENTATION_SCOPES
        );
        assert_eq!(
            &*event.instrumentation_scopes[0],
            "io.opentelemetry.scope-0"
        );
    }
}