lodestone-client 0.0.32

//! Byte-exact client adapter for the generic observability Gateway leaf.
//!
//! The adapter deliberately transports the caller's encoded message without
//! decoding OTLP or Elasticsearch bulk content. Collection credentials and
//! trust remain owned by the validated Lodestone client session.

use std::fmt::{self, Display, Formatter};
use std::time::Duration;

use http::{HeaderMap, HeaderValue, StatusCode, Version, header};
use reqwest::{Certificate, Url, redirect::Policy};

use crate::transport::{Client, authorization_value};
use crate::{ClientError, CredentialProviderError};

pub const MAX_OBSERVABILITY_INGEST_MESSAGE_BYTES: usize = 16 * 1024 * 1024;
const MAX_OBSERVABILITY_RESPONSE_BYTES: usize = 16 * 1024 * 1024;
const OPERATION_ID_HEADER: &str = "x-lodestone-operation-id";
const REQUEST_DIGEST_HEADER: &str = "x-lodestone-request-digest";
const MODE_HEADER: &str = "x-lodestone-ingest-mode";
const PROFILE_HEADER: &str = "x-lodestone-observability-profile";
const DISPOSITION_HEADER: &str = "x-lodestone-submission-disposition";
const ACCEPTED_HEADER: &str = "x-lodestone-accepted-records";
const REJECTED_HEADER: &str = "x-lodestone-rejected-records";
const RECEIPT_DIGEST_HEADER: &str = "x-lodestone-receipt-digest";
const INDEX_ID_HEADER: &str = "x-lodestone-index-id";
const PROTOCOL_HEADER: &str = "x-lodestone-observability-protocol";
const COLLECTION_ID_HEADER: &str = "x-lodestone-collection-id";
const INDEX_NAME_HEADER: &str = "x-lodestone-index-name";

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ObservabilityIngestEndpointV1 {
    origin: Url,
}

impl ObservabilityIngestEndpointV1 {
    pub fn parse(value: &str) -> Result<Self, ObservabilityIngestErrorV1> {
        let origin = Url::parse(value).map_err(|_| ObservabilityIngestErrorV1::InvalidRequest)?;
        if origin.scheme() != "https"
            || origin.cannot_be_a_base()
            || origin.username() != ""
            || origin.password().is_some()
            || origin.query().is_some()
            || origin.fragment().is_some()
            || origin.host_str().is_none()
            || !matches!(origin.path(), "" | "/")
        {
            return Err(ObservabilityIngestErrorV1::InvalidRequest);
        }
        Ok(Self { origin })
    }

    fn request_url(&self, path: &str) -> Result<Url, ObservabilityIngestErrorV1> {
        self.origin
            .join(path.trim_start_matches('/'))
            .map_err(|_| ObservabilityIngestErrorV1::InvalidRequest)
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ObservabilityIngestProtocolV1 {
    OtlpLogsHttpProtobuf,
    OtlpLogsHttpJson,
    OtlpTracesHttpProtobuf,
    OtlpTracesHttpJson,
    OtlpLogsGrpc,
    OtlpTracesGrpc,
    ElasticsearchBulkVector,
    ElasticsearchBulkFluentBit,
}

impl ObservabilityIngestProtocolV1 {
    #[must_use]
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::OtlpLogsHttpProtobuf => "otlp_logs_http_protobuf",
            Self::OtlpLogsHttpJson => "otlp_logs_http_json",
            Self::OtlpTracesHttpProtobuf => "otlp_traces_http_protobuf",
            Self::OtlpTracesHttpJson => "otlp_traces_http_json",
            Self::OtlpLogsGrpc => "otlp_logs_grpc",
            Self::OtlpTracesGrpc => "otlp_traces_grpc",
            Self::ElasticsearchBulkVector => "elasticsearch_bulk_vector",
            Self::ElasticsearchBulkFluentBit => "elasticsearch_bulk_fluent_bit",
        }
    }

    const fn path(self) -> &'static str {
        match self {
            Self::OtlpLogsHttpProtobuf | Self::OtlpLogsHttpJson => "/v1/logs",
            Self::OtlpTracesHttpProtobuf | Self::OtlpTracesHttpJson => "/v1/traces",
            Self::OtlpLogsGrpc => "/opentelemetry.proto.collector.logs.v1.LogsService/Export",
            Self::OtlpTracesGrpc => "/opentelemetry.proto.collector.trace.v1.TraceService/Export",
            Self::ElasticsearchBulkVector | Self::ElasticsearchBulkFluentBit => "/v1/elasticsearch/_bulk",
        }
    }

    const fn content_type(self) -> &'static str {
        match self {
            Self::OtlpLogsHttpProtobuf | Self::OtlpTracesHttpProtobuf => "application/x-protobuf",
            Self::OtlpLogsHttpJson | Self::OtlpTracesHttpJson => "application/json",
            Self::OtlpLogsGrpc | Self::OtlpTracesGrpc => "application/grpc",
            Self::ElasticsearchBulkVector | Self::ElasticsearchBulkFluentBit => "application/x-ndjson",
        }
    }

    const fn profile(self) -> Option<&'static str> {
        match self {
            Self::ElasticsearchBulkVector => Some("vector"),
            Self::ElasticsearchBulkFluentBit => Some("fluent_bit"),
            _ => None,
        }
    }

    const fn is_grpc(self) -> bool {
        matches!(self, Self::OtlpLogsGrpc | Self::OtlpTracesGrpc)
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ObservabilityContentEncodingV1 {
    Identity,
    Gzip,
}

impl ObservabilityContentEncodingV1 {
    const fn as_str(self) -> &'static str {
        match self {
            Self::Identity => "identity",
            Self::Gzip => "gzip",
        }
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ObservabilityIngestModeV1 {
    Submit,
    Resolve,
}

impl ObservabilityIngestModeV1 {
    const fn as_str(self) -> &'static str {
        match self {
            Self::Submit => "submit",
            Self::Resolve => "resolve",
        }
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ObservabilityIngestDispositionV1 {
    Committed,
    Replayed,
    SubmittedIndeterminate,
    Conflict,
    NotSubmitted,
    Retired,
}

impl ObservabilityIngestDispositionV1 {
    fn parse(value: &str) -> Option<Self> {
        match value {
            "committed" => Some(Self::Committed),
            "replayed" => Some(Self::Replayed),
            "submitted_indeterminate" => Some(Self::SubmittedIndeterminate),
            "conflict" => Some(Self::Conflict),
            "not_submitted" => Some(Self::NotSubmitted),
            "retired" => Some(Self::Retired),
            _ => None,
        }
    }
}

pub struct ObservabilityIngestRequestV1 {
    pub endpoint: ObservabilityIngestEndpointV1,
    pub operation_id: String,
    pub request_digest: [u8; 32],
    pub protocol: ObservabilityIngestProtocolV1,
    pub mode: ObservabilityIngestModeV1,
    pub expected_index_id: String,
    pub index_name: String,
    pub content_encoding: ObservabilityContentEncodingV1,
    pub exact_encoded_message: Vec<u8>,
    pub timeout: Duration,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ObservabilityIngestReceiptV1 {
    pub operation_id: String,
    pub request_digest: [u8; 32],
    pub protocol: ObservabilityIngestProtocolV1,
    pub disposition: ObservabilityIngestDispositionV1,
    pub accepted_records: u64,
    pub rejected_records: u64,
    pub source_bytes: u64,
    pub index_id: String,
    pub receipt_digest: [u8; 32],
}

pub struct ObservabilityIngestResponseV1 {
    pub receipt: ObservabilityIngestReceiptV1,
    pub status: StatusCode,
    pub content_type: String,
    pub transport_body: Vec<u8>,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ObservabilityIngestErrorV1 {
    InvalidRequest,
    CredentialUnavailable,
    DependencyUnavailable,
    SubmittedIndeterminate,
    Refused { status: u16, retry_safe: bool },
    InvalidResponse,
}

impl Display for ObservabilityIngestErrorV1 {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        formatter.write_str(match self {
            Self::InvalidRequest => "observability ingest request is invalid",
            Self::CredentialUnavailable => "observability ingest credential is unavailable",
            Self::DependencyUnavailable => "observability ingest dependency is unavailable",
            Self::SubmittedIndeterminate => "observability ingest submission is indeterminate",
            Self::Refused { .. } => "observability ingest was refused",
            Self::InvalidResponse => "observability ingest response is invalid",
        })
    }
}

impl std::error::Error for ObservabilityIngestErrorV1 {}

impl Client {
    pub async fn observability_ingest(
        &self,
        request: ObservabilityIngestRequestV1,
    ) -> Result<ObservabilityIngestResponseV1, ObservabilityIngestErrorV1> {
        validate_request(self, &request)?;
        let material = self.credential().await.map_err(map_client_error)?;
        let authorization = authorization_value(&material).map_err(map_client_error)?;
        let trust = self
            .transport_config()
            .trust
            .certificates_der()
            .iter()
            .map(|certificate| {
                Certificate::from_der(certificate).map_err(|_| ObservabilityIngestErrorV1::InvalidRequest)
            })
            .collect::<Result<Vec<_>, _>>()?;
        let mut builder = reqwest::Client::builder()
            .https_only(true)
            .no_proxy()
            .redirect(Policy::none())
            .connect_timeout(self.pool_config().connect_timeout)
            .timeout(request.timeout)
            .tls_certs_only(trust);
        if request.protocol.is_grpc() {
            builder = builder.http2_prior_knowledge();
        }
        let client = builder
            .build()
            .map_err(|_| ObservabilityIngestErrorV1::DependencyUnavailable)?;
        let body = if request.protocol.is_grpc() {
            grpc_frame(&request.exact_encoded_message, request.content_encoding)?
        } else {
            request.exact_encoded_message.clone()
        };
        let mut outbound = client
            .post(request.endpoint.request_url(request.protocol.path())?)
            .header(header::AUTHORIZATION, sensitive_header(authorization.as_slice())?)
            .header(COLLECTION_ID_HEADER, self.expected_collection_id())
            .header(OPERATION_ID_HEADER, &request.operation_id)
            .header(REQUEST_DIGEST_HEADER, lower_hex(&request.request_digest))
            .header(MODE_HEADER, request.mode.as_str())
            .header(INDEX_ID_HEADER, &request.expected_index_id)
            .header(header::CONTENT_TYPE, request.protocol.content_type())
            .body(body);
        if request.protocol.is_grpc() {
            outbound = outbound
                .version(Version::HTTP_2)
                .header("te", "trailers")
                .header(INDEX_NAME_HEADER, &request.index_name)
                .header("grpc-encoding", request.content_encoding.as_str());
        } else {
            outbound = outbound.header(header::CONTENT_ENCODING, request.content_encoding.as_str());
        }
        if let Some(profile) = request.protocol.profile() {
            outbound = outbound.header(PROFILE_HEADER, profile);
        }
        let response = outbound.send().await.map_err(map_reqwest_error)?;
        let status = response.status();
        let headers = response.headers().clone();
        let content_type = exact_header(&headers, header::CONTENT_TYPE.as_str())
            .unwrap_or("application/octet-stream")
            .to_owned();
        let body = response
            .bytes()
            .await
            .map_err(|_| ObservabilityIngestErrorV1::SubmittedIndeterminate)?;
        if body.len() > MAX_OBSERVABILITY_RESPONSE_BYTES {
            return Err(ObservabilityIngestErrorV1::InvalidResponse);
        }
        let receipt_headers_present = headers.contains_key(DISPOSITION_HEADER);
        if !receipt_headers_present {
            return Err(map_refusal(status));
        }
        let transport_body = if request.protocol.is_grpc() {
            unframe_grpc_response(&body)?
        } else {
            body.to_vec()
        };
        let receipt = parse_receipt(&headers, &request)?;
        Ok(ObservabilityIngestResponseV1 {
            receipt,
            status,
            content_type,
            transport_body,
        })
    }
}

fn validate_request(client: &Client, request: &ObservabilityIngestRequestV1) -> Result<(), ObservabilityIngestErrorV1> {
    let server_name = client.transport_config().identity.tls_server_name();
    if request.endpoint.origin.host_str() != Some(server_name)
        || !valid_uuid_v7(&request.operation_id)
        || request.request_digest == [0; 32]
        || blake3::hash(&request.exact_encoded_message).as_bytes() != &request.request_digest
        || !valid_lower_hex(&request.expected_index_id, 16)
        || request.index_name.is_empty()
        || request.index_name.len() > 255
        || request.index_name.bytes().any(|byte| byte.is_ascii_control())
        || request.exact_encoded_message.is_empty()
        || request.exact_encoded_message.len() > MAX_OBSERVABILITY_INGEST_MESSAGE_BYTES
        || request.timeout.is_zero()
        || request.timeout > Duration::from_secs(60)
    {
        return Err(ObservabilityIngestErrorV1::InvalidRequest);
    }
    Ok(())
}

fn parse_receipt(
    headers: &HeaderMap,
    request: &ObservabilityIngestRequestV1,
) -> Result<ObservabilityIngestReceiptV1, ObservabilityIngestErrorV1> {
    let disposition = ObservabilityIngestDispositionV1::parse(exact_header(headers, DISPOSITION_HEADER)?)
        .ok_or(ObservabilityIngestErrorV1::InvalidResponse)?;
    let accepted_records = canonical_u64(exact_header(headers, ACCEPTED_HEADER)?)?;
    let rejected_records = canonical_u64(exact_header(headers, REJECTED_HEADER)?)?;
    let receipt_digest = decode_digest(exact_header(headers, RECEIPT_DIGEST_HEADER)?)?;
    let index_id = exact_header(headers, INDEX_ID_HEADER)?;
    let protocol = exact_header(headers, PROTOCOL_HEADER)?;
    if index_id != request.expected_index_id
        || protocol != request.protocol.as_str()
        || receipt_digest == [0; 32]
        || matches!(
            disposition,
            ObservabilityIngestDispositionV1::Committed | ObservabilityIngestDispositionV1::Replayed
        ) && accepted_records == 0
    {
        return Err(ObservabilityIngestErrorV1::InvalidResponse);
    }
    Ok(ObservabilityIngestReceiptV1 {
        operation_id: request.operation_id.clone(),
        request_digest: request.request_digest,
        protocol: request.protocol,
        disposition,
        accepted_records,
        rejected_records,
        source_bytes: u64::try_from(request.exact_encoded_message.len())
            .map_err(|_| ObservabilityIngestErrorV1::InvalidResponse)?,
        index_id: index_id.to_owned(),
        receipt_digest,
    })
}

fn exact_header<'a>(headers: &'a HeaderMap, name: &'static str) -> Result<&'a str, ObservabilityIngestErrorV1> {
    let mut values = headers.get_all(name).iter();
    let value = values
        .next()
        .and_then(|value| value.to_str().ok())
        .filter(|value| !value.is_empty())
        .ok_or(ObservabilityIngestErrorV1::InvalidResponse)?;
    if values.next().is_some() {
        return Err(ObservabilityIngestErrorV1::InvalidResponse);
    }
    Ok(value)
}

fn canonical_u64(value: &str) -> Result<u64, ObservabilityIngestErrorV1> {
    if value.is_empty() || !value.bytes().all(|byte| byte.is_ascii_digit()) || value.starts_with('0') && value != "0" {
        return Err(ObservabilityIngestErrorV1::InvalidResponse);
    }
    value
        .parse::<u64>()
        .map_err(|_| ObservabilityIngestErrorV1::InvalidResponse)
}

fn decode_digest(value: &str) -> Result<[u8; 32], ObservabilityIngestErrorV1> {
    if !valid_lower_hex(value, 32) {
        return Err(ObservabilityIngestErrorV1::InvalidResponse);
    }
    let mut digest = [0_u8; 32];
    for (output, pair) in digest.iter_mut().zip(value.as_bytes().as_chunks::<2>().0) {
        let pair = std::str::from_utf8(pair).map_err(|_| ObservabilityIngestErrorV1::InvalidResponse)?;
        *output = u8::from_str_radix(pair, 16).map_err(|_| ObservabilityIngestErrorV1::InvalidResponse)?;
    }
    Ok(digest)
}

fn valid_lower_hex(value: &str, bytes: usize) -> bool {
    value.len() == bytes.saturating_mul(2)
        && value
            .bytes()
            .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
        && value.bytes().any(|byte| byte != b'0')
}

fn valid_uuid_v7(value: &str) -> bool {
    value.len() == 36
        && value.as_bytes().get(8) == Some(&b'-')
        && value.as_bytes().get(13) == Some(&b'-')
        && value.as_bytes().get(18) == Some(&b'-')
        && value.as_bytes().get(23) == Some(&b'-')
        && value.as_bytes().get(14) == Some(&b'7')
        && value
            .as_bytes()
            .get(19)
            .is_some_and(|value| matches!(value, b'8' | b'9' | b'a' | b'b'))
        && value.bytes().enumerate().all(|(index, byte)| {
            matches!(index, 8 | 13 | 18 | 23) || byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte)
        })
}

fn grpc_frame(message: &[u8], encoding: ObservabilityContentEncodingV1) -> Result<Vec<u8>, ObservabilityIngestErrorV1> {
    let length = u32::try_from(message.len()).map_err(|_| ObservabilityIngestErrorV1::InvalidRequest)?;
    let mut framed = Vec::with_capacity(message.len().saturating_add(5));
    framed.push(u8::from(encoding == ObservabilityContentEncodingV1::Gzip));
    framed.extend_from_slice(&length.to_be_bytes());
    framed.extend_from_slice(message);
    Ok(framed)
}

fn unframe_grpc_response(body: &[u8]) -> Result<Vec<u8>, ObservabilityIngestErrorV1> {
    if body.len() < 5 || body[0] != 0 {
        return Err(ObservabilityIngestErrorV1::InvalidResponse);
    }
    let declared = usize::try_from(u32::from_be_bytes([body[1], body[2], body[3], body[4]]))
        .map_err(|_| ObservabilityIngestErrorV1::InvalidResponse)?;
    if declared != body.len().saturating_sub(5) {
        return Err(ObservabilityIngestErrorV1::InvalidResponse);
    }
    Ok(body[5..].to_vec())
}

fn sensitive_header(value: &[u8]) -> Result<HeaderValue, ObservabilityIngestErrorV1> {
    let mut header = HeaderValue::from_bytes(value).map_err(|_| ObservabilityIngestErrorV1::CredentialUnavailable)?;
    header.set_sensitive(true);
    Ok(header)
}

fn lower_hex(bytes: &[u8]) -> String {
    let mut output = String::with_capacity(bytes.len().saturating_mul(2));
    for byte in bytes {
        use std::fmt::Write as _;
        let _ = write!(output, "{byte:02x}");
    }
    output
}

fn map_client_error(error: ClientError) -> ObservabilityIngestErrorV1 {
    match error {
        ClientError::Credential(
            CredentialProviderError::Unavailable
            | CredentialProviderError::NotFound
            | CredentialProviderError::InvalidMaterial
            | CredentialProviderError::RandomUnavailable,
        ) => ObservabilityIngestErrorV1::CredentialUnavailable,
        _ => ObservabilityIngestErrorV1::DependencyUnavailable,
    }
}

fn map_reqwest_error(error: reqwest::Error) -> ObservabilityIngestErrorV1 {
    if error.is_builder() || error.is_connect() {
        ObservabilityIngestErrorV1::DependencyUnavailable
    } else {
        ObservabilityIngestErrorV1::SubmittedIndeterminate
    }
}

fn map_refusal(status: StatusCode) -> ObservabilityIngestErrorV1 {
    ObservabilityIngestErrorV1::Refused {
        status: status.as_u16(),
        retry_safe: matches!(status, StatusCode::TOO_MANY_REQUESTS | StatusCode::SERVICE_UNAVAILABLE),
    }
}

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

    #[test]
    fn endpoint_is_an_exact_https_origin() {
        assert!(ObservabilityIngestEndpointV1::parse("https://core.example:8444").is_ok());
        assert!(ObservabilityIngestEndpointV1::parse("http://core.example:8444").is_err());
        assert!(ObservabilityIngestEndpointV1::parse("https://core.example:8444/v1/logs").is_err());
        assert!(ObservabilityIngestEndpointV1::parse("https://user@core.example:8444").is_err());
    }

    #[test]
    fn grpc_framing_preserves_the_exact_message() -> Result<(), Box<dyn std::error::Error>> {
        let message = b"exact opaque bytes";
        let framed = grpc_frame(message, ObservabilityContentEncodingV1::Identity)?;
        assert_eq!(unframe_grpc_response(&framed)?, message);
        let compressed = grpc_frame(message, ObservabilityContentEncodingV1::Gzip)?;
        assert_eq!(compressed[0], 1);
        Ok(())
    }

    #[test]
    fn receipt_headers_are_closed_and_request_bound() -> Result<(), Box<dyn std::error::Error>> {
        let message = b"opaque".to_vec();
        let request = ObservabilityIngestRequestV1 {
            endpoint: ObservabilityIngestEndpointV1::parse("https://core.example:8444")?,
            operation_id: "019a3bc2-0e66-7f38-a446-ef98386bfcd1".to_owned(),
            request_digest: *blake3::hash(&message).as_bytes(),
            protocol: ObservabilityIngestProtocolV1::OtlpLogsHttpProtobuf,
            mode: ObservabilityIngestModeV1::Submit,
            expected_index_id: "01".repeat(16),
            index_name: "logs".to_owned(),
            content_encoding: ObservabilityContentEncodingV1::Identity,
            exact_encoded_message: message,
            timeout: Duration::from_secs(1),
        };
        let mut headers = HeaderMap::new();
        for (name, value) in [
            (DISPOSITION_HEADER, "committed".to_owned()),
            (ACCEPTED_HEADER, "1".to_owned()),
            (REJECTED_HEADER, "0".to_owned()),
            (RECEIPT_DIGEST_HEADER, "02".repeat(32)),
            (INDEX_ID_HEADER, request.expected_index_id.clone()),
            (PROTOCOL_HEADER, request.protocol.as_str().to_owned()),
        ] {
            headers.insert(
                http::header::HeaderName::from_static(name),
                HeaderValue::from_str(&value)?,
            );
        }
        let receipt = parse_receipt(&headers, &request)?;
        assert_eq!(receipt.accepted_records, 1);
        headers.insert(INDEX_ID_HEADER, HeaderValue::from_static("01"));
        assert_eq!(
            parse_receipt(&headers, &request),
            Err(ObservabilityIngestErrorV1::InvalidResponse)
        );
        Ok(())
    }
}