relay_server/services/

processor.rs

use std::borrow::Cow;
use std::collections::{BTreeMap, BTreeSet, HashMap};
use std::error::Error;
use std::fmt::Debug;
use std::future::Future;
use std::io::Write;
use std::pin::Pin;
use std::sync::Arc;
use std::time::Duration;

use anyhow::Context;
use brotli::CompressorWriter as BrotliEncoder;
use bytes::Bytes;
use chrono::{DateTime, Utc};
use flate2::Compression;
use flate2::write::{GzEncoder, ZlibEncoder};
use futures::future::BoxFuture;
use relay_base_schema::project::{ProjectId, ProjectKey};
use relay_cogs::{AppFeature, Cogs, FeatureWeights, ResourceId, Token};
use relay_common::time::UnixTimestamp;
use relay_config::{Config, HttpEncoding, UpstreamDescriptor};
use relay_event_normalization::{ClockDriftProcessor, GeoIpLookup};
use relay_event_schema::processor::ProcessingAction;
use relay_event_schema::protocol::ClientReport;
use relay_filter::FilterStatKey;
use relay_log::sentry::SentryFutureExt;
use relay_metrics::{Bucket, BucketMetadata, BucketView, BucketsView, MetricNamespace};
use relay_quotas::{RateLimits, Scoping};
use relay_sampling::evaluation::SamplingDecision;
use relay_statsd::metric;
use relay_system::{Addr, FromMessage, NoResponse, Service};
use reqwest::header;
use zstd::stream::Encoder as ZstdEncoder;

use crate::envelope::{self, ContentType, Envelope, EnvelopeError, Item, ItemType};
use crate::extractors::{PartialDsn, RequestMeta, RequestTrust};
use crate::managed::ManagedEnvelope;
use crate::metrics::{MetricOutcomes, MetricsLimiter, MinimalTrackableBucket};
use crate::metrics_extraction::ExtractedMetrics;
use crate::processing::errors::SwitchProcessingError;
use crate::processing::relay::RelayProcessor;
use crate::processing::{Forward as _, Output, Outputs, QuotaRateLimiter};
use crate::service::ServiceError;
use crate::services::global_config::GlobalConfigHandle;
use crate::services::metrics::{Aggregator, FlushBuckets, MergeBuckets, ProjectBuckets};
use crate::services::outcome::{DiscardItemType, DiscardReason, Outcome, TrackOutcome};
use crate::services::projects::cache::ProjectCacheHandle;
use crate::services::projects::project::{ProjectInfo, ProjectState};
use crate::services::upstream::{
    SendRequest, Sign, SignatureType, UpstreamRelay, UpstreamRequest, UpstreamRequestError,
};
use crate::statsd::{RelayCounters, RelayDistributions, RelayTimers};
use crate::utils;
use crate::{http, processing};
use relay_threading::AsyncPool;
use symbolic_unreal::{Unreal4Error, Unreal4ErrorKind};
#[cfg(feature = "processing")]
use {
    crate::services::objectstore::Objectstore,
    crate::services::store::Store,
    itertools::Itertools,
    relay_dynamic_config::GlobalConfig,
    relay_quotas::{Quota, RateLimitingError, RedisRateLimiter},
    relay_redis::RedisClients,
    std::time::Instant,
};

mod metrics;

/// The minimum clock drift for correction to apply.
pub const MINIMUM_CLOCK_DRIFT: Duration = Duration::from_secs(55 * 60);

/// An error returned when handling [`ProcessEnvelope`].
#[derive(Debug, thiserror::Error)]
pub enum ProcessingError {
    #[error("invalid json in event")]
    InvalidJson(#[source] serde_json::Error),

    #[error("invalid message pack event payload")]
    InvalidMsgpack(#[from] rmp_serde::decode::Error),

    #[error("invalid unreal crash report")]
    InvalidUnrealReport(#[source] Unreal4Error),

    #[error("event payload too large")]
    PayloadTooLarge(DiscardItemType),

    #[error("invalid transaction event")]
    InvalidTransaction,

    #[error("the item is not allowed/supported in this envelope")]
    UnsupportedItem,

    #[error("envelope processor failed")]
    ProcessingFailed(#[from] ProcessingAction),

    #[error("duplicate {0} in event")]
    DuplicateItem(ItemType),

    #[error("failed to extract event payload")]
    NoEventPayload,

    #[error("invalid security report type: {0:?}")]
    InvalidSecurityType(Bytes),

    #[error("unsupported security report type")]
    UnsupportedSecurityType,

    #[error("invalid security report")]
    InvalidSecurityReport(#[source] serde_json::Error),

    #[error("event filtered with reason: {0:?}")]
    EventFiltered(FilterStatKey),

    #[error("could not serialize event payload")]
    SerializeFailed(#[source] serde_json::Error),

    #[cfg(feature = "processing")]
    #[error("failed to apply quotas")]
    QuotasFailed(#[from] RateLimitingError),

    #[error("nintendo switch dying message processing failed {0:?}")]
    InvalidNintendoDyingMessage(#[source] SwitchProcessingError),

    #[cfg(all(sentry, feature = "processing"))]
    #[error("playstation dump processing failed: {0}")]
    InvalidPlaystationDump(String),

    #[cfg(feature = "processing")]
    #[error("invalid attachment reference")]
    InvalidAttachmentRef,
}

impl ProcessingError {
    pub fn to_outcome(&self) -> Option<Outcome> {
        match self {
            Self::PayloadTooLarge(payload_type) => {
                Some(Outcome::Invalid(DiscardReason::ItemTooLarge(*payload_type)))
            }
            Self::InvalidJson(_) => Some(Outcome::Invalid(DiscardReason::InvalidJson)),
            Self::InvalidMsgpack(_) => Some(Outcome::Invalid(DiscardReason::InvalidMsgpack)),
            Self::InvalidSecurityType(_) => {
                Some(Outcome::Invalid(DiscardReason::SecurityReportType))
            }
            Self::UnsupportedItem => Some(Outcome::Invalid(DiscardReason::InvalidEnvelope)),
            Self::InvalidSecurityReport(_) => Some(Outcome::Invalid(DiscardReason::SecurityReport)),
            Self::UnsupportedSecurityType => Some(Outcome::Filtered(FilterStatKey::InvalidCsp)),
            Self::InvalidTransaction => Some(Outcome::Invalid(DiscardReason::InvalidTransaction)),
            Self::DuplicateItem(_) => Some(Outcome::Invalid(DiscardReason::DuplicateItem)),
            Self::NoEventPayload => Some(Outcome::Invalid(DiscardReason::NoEventPayload)),
            Self::InvalidNintendoDyingMessage(_) => Some(Outcome::Invalid(DiscardReason::Payload)),
            #[cfg(all(sentry, feature = "processing"))]
            Self::InvalidPlaystationDump(_) => Some(Outcome::Invalid(DiscardReason::Payload)),
            Self::InvalidUnrealReport(err) if err.kind() == Unreal4ErrorKind::BadCompression => {
                Some(Outcome::Invalid(DiscardReason::InvalidCompression))
            }
            Self::InvalidUnrealReport(_) => Some(Outcome::Invalid(DiscardReason::ProcessUnreal)),
            Self::SerializeFailed(_) | Self::ProcessingFailed(_) => {
                Some(Outcome::Invalid(DiscardReason::Internal))
            }
            #[cfg(feature = "processing")]
            Self::QuotasFailed(_) => Some(Outcome::Invalid(DiscardReason::Internal)),
            Self::EventFiltered(key) => Some(Outcome::Filtered(key.clone())),

            #[cfg(feature = "processing")]
            Self::InvalidAttachmentRef => {
                Some(Outcome::Invalid(DiscardReason::InvalidAttachmentRef))
            }
        }
    }
}

impl From<Unreal4Error> for ProcessingError {
    fn from(err: Unreal4Error) -> Self {
        match err.kind() {
            Unreal4ErrorKind::TooLarge => Self::PayloadTooLarge(ItemType::UnrealReport.into()),
            _ => ProcessingError::InvalidUnrealReport(err),
        }
    }
}

/// A container for extracted metrics during processing.
///
/// The container enforces that the extracted metrics are correctly tagged
/// with the dynamic sampling decision.
#[derive(Debug)]
pub struct ProcessingExtractedMetrics {
    metrics: ExtractedMetrics,
}

impl ProcessingExtractedMetrics {
    pub fn new() -> Self {
        Self {
            metrics: ExtractedMetrics::default(),
        }
    }

    pub fn into_inner(self) -> ExtractedMetrics {
        self.metrics
    }

    /// Extends the contained metrics with [`ExtractedMetrics`].
    pub fn extend(
        &mut self,
        extracted: ExtractedMetrics,
        sampling_decision: Option<SamplingDecision>,
    ) {
        self.extend_project_metrics(extracted.project_metrics, sampling_decision);
        self.extend_sampling_metrics(extracted.sampling_metrics, sampling_decision);
    }

    /// Extends the contained project metrics.
    pub fn extend_project_metrics<I>(
        &mut self,
        buckets: I,
        sampling_decision: Option<SamplingDecision>,
    ) where
        I: IntoIterator<Item = Bucket>,
    {
        self.metrics
            .project_metrics
            .extend(buckets.into_iter().map(|mut bucket| {
                bucket.metadata.extracted_from_indexed =
                    sampling_decision == Some(SamplingDecision::Keep);
                bucket
            }));
    }

    /// Extends the contained sampling metrics.
    pub fn extend_sampling_metrics<I>(
        &mut self,
        buckets: I,
        sampling_decision: Option<SamplingDecision>,
    ) where
        I: IntoIterator<Item = Bucket>,
    {
        self.metrics
            .sampling_metrics
            .extend(buckets.into_iter().map(|mut bucket| {
                bucket.metadata.extracted_from_indexed =
                    sampling_decision == Some(SamplingDecision::Keep);
                bucket
            }));
    }
}

fn send_metrics(
    metrics: ExtractedMetrics,
    project_key: ProjectKey,
    sampling_key: Option<ProjectKey>,
    aggregator: &Addr<Aggregator>,
) {
    let ExtractedMetrics {
        project_metrics,
        sampling_metrics,
    } = metrics;

    if !project_metrics.is_empty() {
        aggregator.send(MergeBuckets {
            project_key,
            buckets: project_metrics,
        });
    }

    if !sampling_metrics.is_empty() {
        // If no sampling project state is available, we associate the sampling
        // metrics with the current project.
        //
        // project_without_tracing         -> metrics goes to self
        // dependent_project_with_tracing  -> metrics goes to root
        // root_project_with_tracing       -> metrics goes to root == self
        let sampling_project_key = sampling_key.unwrap_or(project_key);
        aggregator.send(MergeBuckets {
            project_key: sampling_project_key,
            buckets: sampling_metrics,
        });
    }
}

/// Applies processing to all contents of the given envelope.
///
/// Depending on the contents of the envelope and Relay's mode, this includes:
///
///  - Basic normalization and validation for all item types.
///  - Clock drift correction if the required `sent_at` header is present.
///  - Expansion of certain item types (e.g. unreal).
///  - Store normalization for event payloads in processing mode.
///  - Rate limiters and inbound filters on events in processing mode.
#[derive(Debug)]
pub struct ProcessEnvelope {
    /// Envelope to process.
    pub envelope: ManagedEnvelope,
    /// The project info.
    pub project_info: Arc<ProjectInfo>,
    /// Currently active cached rate limits for this project.
    pub rate_limits: Arc<RateLimits>,
    /// Root sampling project info.
    pub sampling_project_info: Option<Arc<ProjectInfo>>,
}

/// Parses a list of metrics or metric buckets and pushes them to the project's aggregator.
///
/// This parses and validates the metrics:
///  - For [`Metrics`](ItemType::Statsd), each metric is parsed separately, and invalid metrics are
///    ignored independently.
///  - For [`MetricBuckets`](ItemType::MetricBuckets), the entire list of buckets is parsed and
///    dropped together on parsing failure.
///  - Other envelope items will be ignored with an error message.
///
/// Additionally, processing applies clock drift correction using the system clock of this Relay, if
/// the Envelope specifies the [`sent_at`](Envelope::sent_at) header.
#[derive(Debug)]
pub struct ProcessMetrics {
    /// A list of metric items.
    pub data: MetricData,
    /// The target project.
    pub project_key: ProjectKey,
    /// Whether to keep or reset the metric metadata.
    pub source: BucketSource,
    /// The wall clock time at which the request was received.
    pub received_at: DateTime<Utc>,
    /// The value of the Envelope's [`sent_at`](Envelope::sent_at) header for clock drift
    /// correction.
    pub sent_at: Option<DateTime<Utc>>,
}

/// Raw unparsed metric data.
#[derive(Debug)]
pub enum MetricData {
    /// Raw data, unparsed envelope items.
    Raw(Vec<Item>),
    /// Already parsed buckets but unprocessed.
    Parsed(Vec<Bucket>),
}

impl MetricData {
    /// Consumes the metric data and parses the contained buckets.
    ///
    /// If the contained data is already parsed the buckets are returned unchanged.
    /// Raw buckets are parsed and created with the passed `timestamp`.
    fn into_buckets(self, timestamp: UnixTimestamp) -> Vec<Bucket> {
        let items = match self {
            Self::Parsed(buckets) => return buckets,
            Self::Raw(items) => items,
        };

        let mut buckets = Vec::new();
        for item in items {
            let payload = item.payload();
            if item.ty() == &ItemType::Statsd {
                for bucket_result in Bucket::parse_all(&payload, timestamp) {
                    match bucket_result {
                        Ok(bucket) => buckets.push(bucket),
                        Err(error) => relay_log::debug!(
                            error = &error as &dyn Error,
                            "failed to parse metric bucket from statsd format",
                        ),
                    }
                }
            } else if item.ty() == &ItemType::MetricBuckets {
                match serde_json::from_slice::<Vec<Bucket>>(&payload) {
                    Ok(parsed_buckets) => {
                        // Re-use the allocation of `b` if possible.
                        if buckets.is_empty() {
                            buckets = parsed_buckets;
                        } else {
                            buckets.extend(parsed_buckets);
                        }
                    }
                    Err(error) => {
                        relay_log::debug!(
                            error = &error as &dyn Error,
                            "failed to parse metric bucket",
                        );
                        metric!(counter(RelayCounters::MetricBucketsParsingFailed) += 1);
                    }
                }
            } else {
                relay_log::error!(
                    "invalid item of type {} passed to ProcessMetrics",
                    item.ty()
                );
            }
        }
        buckets
    }
}

#[derive(Debug)]
pub struct ProcessBatchedMetrics {
    /// Metrics payload in JSON format.
    pub payload: Bytes,
    /// Whether to keep or reset the metric metadata.
    pub source: BucketSource,
    /// The wall clock time at which the request was received.
    pub received_at: DateTime<Utc>,
    /// The wall clock time at which the request was received.
    pub sent_at: Option<DateTime<Utc>>,
}

/// Source information where a metric bucket originates from.
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
pub enum BucketSource {
    /// The metric bucket originated from an internal Relay use case.
    ///
    /// The metric bucket originates either from within the same Relay
    /// or was accepted coming from another Relay which is registered as
    /// an internal Relay via Relay's configuration.
    Internal,
    /// The bucket source originated from an untrusted source.
    ///
    /// Managed Relays sending extracted metrics are considered external,
    /// it's a project use case but it comes from an untrusted source.
    External,
}

impl BucketSource {
    /// Infers the bucket source from [`RequestMeta::request_trust`].
    pub fn from_meta(meta: &RequestMeta) -> Self {
        match meta.request_trust() {
            RequestTrust::Trusted => Self::Internal,
            RequestTrust::Untrusted => Self::External,
        }
    }
}

/// Sends a client report to the upstream.
#[derive(Debug)]
pub struct SubmitClientReports {
    /// The client report to be sent.
    pub client_reports: Vec<ClientReport>,
    /// Scoping information for the client report.
    pub scoping: Scoping,
}

/// CPU-intensive processing tasks for envelopes.
#[derive(Debug)]
pub enum EnvelopeProcessor {
    ProcessEnvelope(Box<ProcessEnvelope>),
    ProcessProjectMetrics(Box<ProcessMetrics>),
    ProcessBatchedMetrics(Box<ProcessBatchedMetrics>),
    FlushBuckets(Box<FlushBuckets>),
    SubmitClientReports(Box<SubmitClientReports>),
}

impl EnvelopeProcessor {
    /// Returns the name of the message variant.
    pub fn variant(&self) -> &'static str {
        match self {
            EnvelopeProcessor::ProcessEnvelope(_) => "ProcessEnvelope",
            EnvelopeProcessor::ProcessProjectMetrics(_) => "ProcessProjectMetrics",
            EnvelopeProcessor::ProcessBatchedMetrics(_) => "ProcessBatchedMetrics",
            EnvelopeProcessor::FlushBuckets(_) => "FlushBuckets",
            EnvelopeProcessor::SubmitClientReports(_) => "SubmitClientReports",
        }
    }
}

impl relay_system::Interface for EnvelopeProcessor {}

impl FromMessage<ProcessEnvelope> for EnvelopeProcessor {
    type Response = relay_system::NoResponse;

    fn from_message(message: ProcessEnvelope, _sender: ()) -> Self {
        Self::ProcessEnvelope(Box::new(message))
    }
}

impl FromMessage<ProcessMetrics> for EnvelopeProcessor {
    type Response = NoResponse;

    fn from_message(message: ProcessMetrics, _: ()) -> Self {
        Self::ProcessProjectMetrics(Box::new(message))
    }
}

impl FromMessage<ProcessBatchedMetrics> for EnvelopeProcessor {
    type Response = NoResponse;

    fn from_message(message: ProcessBatchedMetrics, _: ()) -> Self {
        Self::ProcessBatchedMetrics(Box::new(message))
    }
}

impl FromMessage<FlushBuckets> for EnvelopeProcessor {
    type Response = NoResponse;

    fn from_message(message: FlushBuckets, _: ()) -> Self {
        Self::FlushBuckets(Box::new(message))
    }
}

impl FromMessage<SubmitClientReports> for EnvelopeProcessor {
    type Response = NoResponse;

    fn from_message(message: SubmitClientReports, _: ()) -> Self {
        Self::SubmitClientReports(Box::new(message))
    }
}

/// The asynchronous thread pool used for scheduling processing tasks in the processor.
pub type EnvelopeProcessorServicePool = AsyncPool<BoxFuture<'static, ()>>;

/// Service implementing the [`EnvelopeProcessor`] interface.
///
/// This service handles messages in a worker pool with configurable concurrency.
#[derive(Clone)]
pub struct EnvelopeProcessorService {
    inner: Arc<InnerProcessor>,
}

/// Contains the addresses of services that the processor publishes to.
pub struct Addrs {
    pub outcome_aggregator: Addr<TrackOutcome>,
    pub upstream_relay: Addr<UpstreamRelay>,
    #[cfg(feature = "processing")]
    pub objectstore: Option<Addr<Objectstore>>,
    #[cfg(feature = "processing")]
    pub store_forwarder: Option<Addr<Store>>,
    pub aggregator: Addr<Aggregator>,
}

impl Default for Addrs {
    fn default() -> Self {
        Addrs {
            outcome_aggregator: Addr::dummy(),
            upstream_relay: Addr::dummy(),
            #[cfg(feature = "processing")]
            objectstore: None,
            #[cfg(feature = "processing")]
            store_forwarder: None,
            aggregator: Addr::dummy(),
        }
    }
}

struct InnerProcessor {
    pool: EnvelopeProcessorServicePool,
    config: Arc<Config>,
    global_config: GlobalConfigHandle,
    project_cache: ProjectCacheHandle,
    cogs: Cogs,
    addrs: Addrs,
    #[cfg(feature = "processing")]
    rate_limiter: Option<Arc<RedisRateLimiter>>,
    metric_outcomes: MetricOutcomes,
    processor: RelayProcessor,
}

impl EnvelopeProcessorService {
    /// Creates a multi-threaded envelope processor.
    #[cfg_attr(feature = "processing", expect(clippy::too_many_arguments))]
    pub fn new(
        pool: EnvelopeProcessorServicePool,
        config: Arc<Config>,
        global_config: GlobalConfigHandle,
        project_cache: ProjectCacheHandle,
        cogs: Cogs,
        #[cfg(feature = "processing")] redis: Option<RedisClients>,
        addrs: Addrs,
        metric_outcomes: MetricOutcomes,
    ) -> Self {
        let geoip_lookup = config
            .geoip_path()
            .and_then(
                |p| match GeoIpLookup::open(p).context(ServiceError::GeoIp) {
                    Ok(geoip) => Some(geoip),
                    Err(err) => {
                        relay_log::error!("failed to open GeoIP db {p:?}: {err:?}");
                        None
                    }
                },
            )
            .unwrap_or_else(GeoIpLookup::empty);

        if let Some(build_epoch) = geoip_lookup.build_epoch() {
            relay_log::info!("Loaded GeoIP database (build: {build_epoch})");
        }

        #[cfg(feature = "processing")]
        let rate_limiter = redis.map(|redis| {
            RedisRateLimiter::new(redis.quotas)
                .max_limit(config.max_rate_limit())
                .cache(config.quota_cache_ratio(), config.quota_cache_max())
        });

        let quota_limiter = Arc::new(QuotaRateLimiter::new(
            #[cfg(feature = "processing")]
            project_cache.clone(),
            #[cfg(feature = "processing")]
            rate_limiter.clone(),
        ));
        #[cfg(feature = "processing")]
        let rate_limiter = rate_limiter.map(Arc::new);
        let inner = InnerProcessor {
            pool,
            global_config,
            project_cache,
            #[cfg(feature = "processing")]
            rate_limiter,
            processor: RelayProcessor::new(
                cogs.clone(),
                &quota_limiter,
                &geoip_lookup,
                addrs.outcome_aggregator.clone(),
            ),
            cogs,
            addrs,
            metric_outcomes,
            config,
        };

        Self {
            inner: Arc::new(inner),
        }
    }

    async fn process_envelope(
        &self,
        project_id: ProjectId,
        mut envelope: ManagedEnvelope,
        ctx: processing::Context<'_>,
    ) -> Vec<Output<Outputs>> {
        // Pre-process the envelope headers.
        if let Some(sampling_state) = ctx.sampling_project_info {
            // Both transactions and standalone span envelopes need a normalized DSC header
            // to make sampling rules based on the segment/transaction name work correctly.
            envelope
                .envelope_mut()
                .parametrize_dsc_transaction(&sampling_state.config.tx_name_rules);
        }

        // Set the event retention. Effectively, this value will only be available in processing
        // mode when the full project config is queried from the upstream.
        if let Some(retention) = ctx.project_info.config.event_retention {
            envelope.envelope_mut().set_retention(retention);
        }

        // Ensure the project ID is updated to the stored instance for this project cache. This can
        // differ in two cases:
        //  1. The envelope was sent to the legacy `/store/` endpoint without a project ID.
        //  2. The DSN was moved and the envelope sent to the old project ID.
        envelope
            .envelope_mut()
            .meta_mut()
            .set_project_id(project_id);

        self.inner.processor.run(envelope, ctx).await
    }

    /// Processes the envelope and returns the processed envelope back.
    ///
    /// Returns `Some` if the envelope passed inbound filtering and rate limiting. Invalid items are
    /// removed from the envelope. Otherwise, if the envelope is empty or the entire envelope needs
    /// to be dropped, this is `None`.
    async fn process<'a>(
        &self,
        mut envelope: ManagedEnvelope,
        ctx: processing::Context<'a>,
    ) -> Vec<Output<Outputs>> {
        // Prefer the project's project ID, and fall back to the stated project id from the
        // envelope. The project ID is available in all modes, other than in proxy mode, where
        // envelopes for unknown projects are forwarded blindly.
        //
        // Neither ID can be available in proxy mode on the /store/ endpoint. This is not supported,
        // since we cannot process an envelope without project ID, so drop it.
        let Some(project_id) = ctx
            .project_info
            .project_id
            .or_else(|| envelope.envelope().meta().project_id())
        else {
            relay_log::error!(
                tags.project_key = %envelope.envelope().meta().public_key(),
                "project info does not contain project id"
            );
            envelope.reject(Outcome::Invalid(DiscardReason::Internal));
            return Vec::new();
        };

        relay_log::configure_scope(|scope| {
            scope.set_tag("project_id", project_id);
        });

        self.process_envelope(project_id, envelope, ctx).await
    }

    async fn handle_process_envelope(&self, cogs: &mut Token, message: ProcessEnvelope) {
        let wait_time = message.envelope.age();
        metric!(timer(RelayTimers::EnvelopeWaitTime) = wait_time);

        // This COGS handling may need an overhaul in the future:
        // Cancel the passed in token, to start individual measurements per processor instead.
        cogs.cancel();

        let global_config = self.inner.global_config.current().unwrap_or_default();

        let ctx = processing::Context {
            config: &self.inner.config,
            global_config: &global_config,
            project_info: &message.project_info,
            sampling_project_info: message.sampling_project_info.as_deref(),
            rate_limits: &message.rate_limits,
        };

        let project_key = message.envelope.meta().public_key();
        // Only allow sending to the sampling key, if we successfully loaded a sampling project
        // info relating to it. This filters out unknown/invalid project keys as well as project
        // keys from different organizations.
        let sampling_key = ctx
            .sampling_project_info
            .and_then(|p| p.get_public_key_config())
            .map(|pkc| pkc.public_key);

        relay_log::configure_scope(|scope| {
            scope.set_tag("project_key", project_key);
            if let Some(sampling_key) = sampling_key {
                scope.set_tag("sampling_key", sampling_key);
            }
            let meta = message.envelope.envelope().meta();
            scope.set_tag("sdk_name", meta.client_name());
            if let Some(client) = meta.client() {
                scope.set_tag("sdk", client);
            }
            if let Some(user_agent) = meta.user_agent() {
                scope.set_extra("user_agent", user_agent.into());
            }
        });

        let outputs = metric!(timer(RelayTimers::EnvelopeProcessingTime), {
            self.process(message.envelope, ctx).await
        });

        let ctx = ctx.to_forward();
        for Output { main, metrics } in outputs {
            if let Some(metrics) = metrics {
                let agg = &self.inner.addrs.aggregator;
                metrics.accept(|metrics| {
                    send_metrics(metrics, project_key, sampling_key, agg);
                });
            }

            if let Some(output) = main {
                // Only counting processing time for COGS at the moment.
                self.submit_upstream(&mut Token::noop(), output, ctx);
            }
        }
    }

    fn handle_process_metrics(&self, cogs: &mut Token, message: ProcessMetrics) {
        let ProcessMetrics {
            data,
            project_key,
            received_at,
            sent_at,
            source,
        } = message;

        let received_timestamp =
            UnixTimestamp::from_datetime(received_at).unwrap_or(UnixTimestamp::now());

        let mut buckets = data.into_buckets(received_timestamp);
        if buckets.is_empty() {
            return;
        };
        cogs.update(relay_metrics::cogs::BySize(&buckets));

        let clock_drift_processor =
            ClockDriftProcessor::new(sent_at, received_at).at_least(MINIMUM_CLOCK_DRIFT);

        buckets.retain_mut(|bucket| {
            if let Err(error) = relay_metrics::normalize_bucket(bucket) {
                relay_log::debug!(error = &error as &dyn Error, "dropping bucket {bucket:?}");
                return false;
            }

            if !self::metrics::is_valid_namespace(bucket, source) {
                relay_log::debug!("dropping bucket in invalid namespace {bucket:?}");
                return false;
            }

            clock_drift_processor.process_timestamp(&mut bucket.timestamp);

            if !matches!(source, BucketSource::Internal) {
                bucket.metadata = BucketMetadata::new(received_timestamp);
            }

            true
        });

        let project = self.inner.project_cache.get(project_key);

        // Best effort check to filter and rate limit buckets, if there is no project state
        // available at the current time, we will check again after flushing.
        let buckets = match project.state() {
            ProjectState::Enabled(project_info) => {
                let rate_limits = project.rate_limits().current_limits();
                self.check_buckets(project_key, project_info, &rate_limits, buckets)
            }
            _ => buckets,
        };

        relay_log::trace!("merging metric buckets into the aggregator");
        self.inner
            .addrs
            .aggregator
            .send(MergeBuckets::new(project_key, buckets));
    }

    fn handle_process_batched_metrics(&self, cogs: &mut Token, message: ProcessBatchedMetrics) {
        let ProcessBatchedMetrics {
            payload,
            source,
            received_at,
            sent_at,
        } = message;

        #[derive(serde::Deserialize)]
        struct Wrapper {
            buckets: HashMap<ProjectKey, Vec<Bucket>>,
        }

        let buckets = match serde_json::from_slice(&payload) {
            Ok(Wrapper { buckets }) => buckets,
            Err(error) => {
                relay_log::debug!(
                    error = &error as &dyn Error,
                    "failed to parse batched metrics",
                );
                metric!(counter(RelayCounters::MetricBucketsParsingFailed) += 1);
                return;
            }
        };

        for (project_key, buckets) in buckets {
            self.handle_process_metrics(
                cogs,
                ProcessMetrics {
                    data: MetricData::Parsed(buckets),
                    project_key,
                    source,
                    received_at,
                    sent_at,
                },
            )
        }
    }

    /// Submits a processor [`Output`] to the appropriate upstream.
    ///
    /// If processing is enabled, the upstream is Kafka.
    fn submit_upstream(
        &self,
        cogs: &mut Token,
        output: Outputs,
        ctx: processing::ForwardContext<'_>,
    ) {
        let _submit = cogs.start_category("submit");

        #[cfg(feature = "processing")]
        if ctx.config.processing_enabled()
            && let Some(store_forwarder) = &self.inner.addrs.store_forwarder
        {
            use crate::processing::StoreHandle;

            let objectstore = self.inner.addrs.objectstore.as_ref();
            let handle = StoreHandle::new(store_forwarder, objectstore, ctx.global_config);

            output
                .forward_store(handle, ctx)
                .unwrap_or_else(|err| err.into_inner());

            return;
        }

        match output.serialize_envelope(ctx) {
            Ok(envelope) => {
                let envelope = ManagedEnvelope::from(envelope);
                self.submit_envelope_upstream(envelope, ctx.project_info.upstream.clone());
            }
            Err(_) => relay_log::error!("failed to serialize output to an envelope"),
        };
    }

    fn submit_envelope_upstream(
        &self,
        mut envelope: ManagedEnvelope,
        // Currently allowed to be optional as code is migrated to respect the upstream override
        // provided from the project config. Eventually must be available and is required.
        upstream: Option<UpstreamDescriptor>,
    ) {
        if envelope.envelope_mut().is_empty() {
            envelope.accept();
            return;
        }

        // No code path should hit this.
        //
        // Any item which is produced by processing is handled in `submit_upstream`,
        // metrics are sent to the store directly and outcomes must be produced to Kafka
        // instead of being sent onward as client report.
        if self.inner.config.processing_enabled() {
            relay_log::error!(
                "attempt to forward envelope to http upstream when processing is enabled"
            );
            return;
        }

        // Override the `sent_at` timestamp. Since the envelope went through basic
        // normalization, all timestamps have been corrected. We propagate the new
        // `sent_at` to allow the next Relay to double-check this timestamp and
        // potentially apply correction again. This is done as close to sending as
        // possible so that we avoid internal delays.
        envelope.envelope_mut().set_sent_at(Utc::now());

        relay_log::trace!("sending envelope to sentry endpoint");
        let http_encoding = self.inner.config.http_encoding();
        let result = envelope.envelope().to_vec().and_then(|v| {
            encode_payload(&v.into(), http_encoding).map_err(EnvelopeError::PayloadIoFailed)
        });

        match result {
            Ok(body) => {
                self.inner
                    .addrs
                    .upstream_relay
                    .send(SendRequest(SendEnvelope {
                        upstream,
                        envelope,
                        body,
                        http_encoding,
                        project_cache: self.inner.project_cache.clone(),
                    }));
            }
            Err(error) => {
                // Errors are only logged for what we consider an internal discard reason. These
                // indicate errors in the infrastructure or implementation bugs.
                relay_log::error!(
                    error = &error as &dyn Error,
                    tags.project_key = %envelope.scoping().project_key,
                    "failed to serialize envelope payload"
                );

                envelope.reject(Outcome::Invalid(DiscardReason::Internal));
            }
        }
    }

    fn handle_submit_client_reports(&self, message: SubmitClientReports) {
        let SubmitClientReports {
            client_reports,
            scoping,
        } = message;

        let upstream = self.inner.config.upstream();
        let dsn = PartialDsn::outbound(&scoping, upstream);

        let mut envelope = Envelope::from_request(None, RequestMeta::outbound(dsn));
        for client_report in client_reports {
            match client_report.serialize() {
                Ok(payload) => {
                    let mut item = Item::new(ItemType::ClientReport);
                    item.set_payload(ContentType::Json, payload);
                    envelope.add_item(item);
                }
                Err(error) => {
                    relay_log::error!(
                        error = &error as &dyn std::error::Error,
                        "failed to serialize client report"
                    );
                }
            }
        }

        let envelope = ManagedEnvelope::new(envelope, self.inner.addrs.outcome_aggregator.clone());
        self.submit_envelope_upstream(envelope, None);
    }

    fn check_buckets(
        &self,
        project_key: ProjectKey,
        project_info: &ProjectInfo,
        rate_limits: &RateLimits,
        buckets: Vec<Bucket>,
    ) -> Vec<Bucket> {
        let Some(scoping) = project_info.scoping(project_key) else {
            relay_log::error!(
                tags.project_key = project_key.as_str(),
                "there is no scoping: dropping {} buckets",
                buckets.len(),
            );
            return Vec::new();
        };

        let mut buckets = self::metrics::apply_project_info(
            buckets,
            &self.inner.metric_outcomes,
            project_info,
            scoping,
        );

        let mut namespaces: BTreeSet<MetricNamespace> = buckets
            .iter()
            .filter_map(|bucket| bucket.name.try_namespace())
            .collect();

        // Never rate limit outcomes.
        namespaces.remove(&MetricNamespace::Outcomes);

        for namespace in namespaces {
            let limits = rate_limits
                .check_with_quotas(project_info.get_quotas(), scoping.metric_bucket(namespace));

            if limits.is_limited() {
                let rejected;
                (buckets, rejected) = utils::split_off(buckets, |bucket| {
                    bucket.name.try_namespace() == Some(namespace)
                });

                let reason_code = limits.longest().and_then(|limit| limit.reason_code.clone());
                self.inner.metric_outcomes.track(
                    scoping,
                    &rejected,
                    Outcome::RateLimited(reason_code),
                );
            }
        }

        let quotas = project_info.config.quotas.clone();
        match MetricsLimiter::create(buckets, quotas, scoping) {
            Ok(mut bucket_limiter) => {
                bucket_limiter.enforce_limits(rate_limits, &self.inner.metric_outcomes);
                bucket_limiter.into_buckets()
            }
            Err(buckets) => buckets,
        }
    }

    #[cfg(feature = "processing")]
    async fn rate_limit_buckets(
        &self,
        scoping: Scoping,
        project_info: &ProjectInfo,
        mut buckets: Vec<Bucket>,
    ) -> Vec<Bucket> {
        let Some(rate_limiter) = &self.inner.rate_limiter else {
            return buckets;
        };

        let global_config = self.inner.global_config.current().unwrap_or_default();
        let mut namespaces = buckets
            .iter()
            .filter_map(|bucket| bucket.name.try_namespace())
            .counts();

        // Never rate limit outcomes.
        namespaces.remove(&MetricNamespace::Outcomes);

        let quotas = CombinedQuotas::new(&global_config, project_info.get_quotas());

        for (namespace, quantity) in namespaces {
            let item_scoping = scoping.metric_bucket(namespace);

            let limits = match rate_limiter
                .is_rate_limited(quotas, item_scoping, quantity, false)
                .await
            {
                Ok(limits) => limits,
                Err(err) => {
                    relay_log::error!(
                        error = &err as &dyn std::error::Error,
                        "failed to check redis rate limits"
                    );
                    break;
                }
            };

            if limits.is_limited() {
                let rejected;
                (buckets, rejected) = utils::split_off(buckets, |bucket| {
                    bucket.name.try_namespace() == Some(namespace)
                });

                let reason_code = limits.longest().and_then(|limit| limit.reason_code.clone());
                self.inner.metric_outcomes.track(
                    scoping,
                    &rejected,
                    Outcome::RateLimited(reason_code),
                );

                self.inner
                    .project_cache
                    .get(item_scoping.scoping.project_key)
                    .rate_limits()
                    .merge(limits);
            }
        }

        match MetricsLimiter::create(buckets, project_info.config.quotas.clone(), scoping) {
            Err(buckets) => buckets,
            Ok(bucket_limiter) => self.apply_other_rate_limits(bucket_limiter).await,
        }
    }

    /// Check and apply rate limits to metrics buckets for transactions and spans.
    #[cfg(feature = "processing")]
    async fn apply_other_rate_limits(&self, mut bucket_limiter: MetricsLimiter) -> Vec<Bucket> {
        relay_log::trace!("handle_rate_limit_buckets");

        let scoping = *bucket_limiter.scoping();

        if let Some(rate_limiter) = self.inner.rate_limiter.as_ref() {
            let global_config = self.inner.global_config.current().unwrap_or_default();
            let quotas = CombinedQuotas::new(&global_config, bucket_limiter.quotas());

            // We set over_accept_once such that the limit is actually reached, which allows subsequent
            // calls with quantity=0 to be rate limited.
            let over_accept_once = true;
            let mut rate_limits = RateLimits::new();

            let (category, count) = bucket_limiter.count();

            let timer = Instant::now();
            let mut is_limited = false;

            if let Some(count) = count {
                match rate_limiter
                    .is_rate_limited(quotas, scoping.item(category), count, over_accept_once)
                    .await
                {
                    Ok(limits) => {
                        is_limited = limits.is_limited();
                        rate_limits.merge(limits)
                    }
                    Err(e) => {
                        relay_log::error!(error = &e as &dyn Error, "rate limiting error")
                    }
                }
            }

            relay_statsd::metric!(
                timer(RelayTimers::RateLimitBucketsDuration) = timer.elapsed(),
                category = category.name(),
                limited = if is_limited { "true" } else { "false" },
                count = match count {
                    None => "none",
                    Some(0) => "0",
                    Some(1) => "1",
                    Some(1..=10) => "10",
                    Some(1..=25) => "25",
                    Some(1..=50) => "50",
                    Some(51..=100) => "100",
                    Some(101..=500) => "500",
                    _ => "> 500",
                },
            );

            if rate_limits.is_limited() {
                let was_enforced =
                    bucket_limiter.enforce_limits(&rate_limits, &self.inner.metric_outcomes);

                if was_enforced {
                    // Update the rate limits in the project cache.
                    self.inner
                        .project_cache
                        .get(scoping.project_key)
                        .rate_limits()
                        .merge(rate_limits);
                }
            }
        }

        bucket_limiter.into_buckets()
    }

    /// Processes metric buckets and sends them to Kafka.
    ///
    /// This function runs the following steps:
    ///  - rate limiting
    ///  - emit billing outcomes
    ///  - submit to `StoreForwarder`
    #[cfg(feature = "processing")]
    async fn encode_metrics_processing(
        &self,
        message: FlushBuckets,
        store_forwarder: &Addr<Store>,
    ) {
        use crate::constants::DEFAULT_EVENT_RETENTION;
        use crate::services::store::StoreMetrics;
        use relay_dynamic_config::Feature;

        for ProjectBuckets {
            buckets,
            scoping,
            project_info,
            ..
        } in message.buckets.into_values()
        {
            let mut buckets = self
                .rate_limit_buckets(scoping, &project_info, buckets)
                .await;

            if buckets.is_empty() {
                continue;
            }

            if project_info
                .config
                .features
                .has(Feature::GenerateBillingOutcome)
            {
                // Emit metric billing outcomes.
                self.inner
                    .metric_outcomes
                    .track_accepted_outcome(scoping, &mut buckets);
            }

            let retention = project_info
                .config
                .event_retention
                .unwrap_or(DEFAULT_EVENT_RETENTION);

            // The store forwarder takes care of bucket splitting internally, so we can submit the
            // entire list of buckets. There is no batching needed here.
            store_forwarder.send(StoreMetrics {
                buckets,
                scoping,
                retention,
            });
        }
    }

    /// Serializes metric buckets to JSON and sends them to the upstream.
    ///
    /// This function runs the following steps:
    ///  - partitioning
    ///  - batching by configured size limit
    ///  - serialize to JSON and pack in an envelope
    ///
    /// Rate limiting runs only in processing Relays as it requires access to the central Redis instance.
    /// Cached rate limits are applied in the project cache already.
    fn encode_metrics_envelope(&self, message: FlushBuckets) {
        let FlushBuckets {
            partition_key,
            buckets,
        } = message;

        let batch_size = self.inner.config.metrics_max_batch_size_bytes();
        let upstream = self.inner.config.upstream();

        for ProjectBuckets {
            buckets,
            scoping,
            project_info,
            ..
        } in buckets.values()
        {
            let dsn = PartialDsn::outbound(scoping, upstream);

            relay_statsd::metric!(
                distribution(RelayDistributions::PartitionKeys) = u64::from(partition_key)
            );

            let mut num_batches = 0;
            for batch in BucketsView::from(buckets).by_size(batch_size) {
                let mut envelope = Envelope::from_request(None, RequestMeta::outbound(dsn.clone()));

                let mut item = Item::new(ItemType::MetricBuckets);
                item.set_source_quantities(crate::metrics::extract_quantities(batch));
                item.set_payload(ContentType::Json, serde_json::to_vec(&buckets).unwrap());
                envelope.add_item(item);

                let mut envelope =
                    ManagedEnvelope::new(envelope, self.inner.addrs.outcome_aggregator.clone());
                envelope
                    .set_partition_key(Some(partition_key))
                    .scope(*scoping);

                relay_statsd::metric!(
                    distribution(RelayDistributions::BucketsPerBatch) = batch.len() as u64
                );

                self.submit_envelope_upstream(envelope, project_info.upstream.clone());
                num_batches += 1;
            }

            relay_statsd::metric!(
                distribution(RelayDistributions::BatchesPerPartition) = num_batches
            );
        }
    }

    /// Creates a [`SendMetricsRequest`] and sends it to the upstream relay.
    fn send_global_partition(
        &self,
        upstream: Option<UpstreamDescriptor>,
        partition_key: u32,
        partition: &mut Partition<'_>,
    ) {
        if partition.is_empty() {
            return;
        }

        let (unencoded, project_info) = partition.take();
        let http_encoding = self.inner.config.http_encoding();
        let encoded = match encode_payload(&unencoded, http_encoding) {
            Ok(payload) => payload,
            Err(error) => {
                let error = &error as &dyn std::error::Error;
                relay_log::error!(error, "failed to encode metrics payload");
                return;
            }
        };

        let request = SendMetricsRequest {
            upstream,
            partition_key: partition_key.to_string(),
            unencoded,
            encoded,
            project_info,
            http_encoding,
            metric_outcomes: self.inner.metric_outcomes.clone(),
        };

        self.inner.addrs.upstream_relay.send(SendRequest(request));
    }

    /// Serializes metric buckets to JSON and sends them to the upstream via the global endpoint.
    ///
    /// This function is similar to [`Self::encode_metrics_envelope`], but sends a global batched
    /// payload directly instead of per-project Envelopes.
    ///
    /// This function runs the following steps:
    ///  - partitioning
    ///  - batching by configured size limit
    ///  - serialize to JSON
    ///  - submit directly to the upstream
    fn encode_metrics_global(&self, message: FlushBuckets) {
        let FlushBuckets {
            partition_key,
            buckets,
        } = message;

        let batch_size = self.inner.config.metrics_max_batch_size_bytes();
        let mut partitions = BTreeMap::new();
        let mut partition_splits = 0;

        for ProjectBuckets {
            buckets,
            scoping,
            project_info,
            ..
        } in buckets.values()
        {
            let partition = match partitions.get_mut(&project_info.upstream) {
                Some(partition) => partition,
                None => partitions
                    .entry(project_info.upstream.clone())
                    .or_insert_with(|| Partition::new(batch_size)),
            };

            for bucket in buckets {
                let mut remaining = Some(BucketView::new(bucket));

                while let Some(bucket) = remaining.take() {
                    if let Some(next) = partition.insert(bucket, *scoping) {
                        // A part of the bucket could not be inserted. Take the partition and submit
                        // it immediately. Repeat until the final part was inserted. This should
                        // always result in a request, otherwise we would enter an endless loop.
                        self.send_global_partition(
                            project_info.upstream.clone(),
                            partition_key,
                            partition,
                        );
                        remaining = Some(next);
                        partition_splits += 1;
                    }
                }
            }
        }

        if partition_splits > 0 {
            metric!(distribution(RelayDistributions::PartitionSplits) = partition_splits);
        }

        for (upstream, mut partition) in partitions {
            self.send_global_partition(upstream, partition_key, &mut partition);
        }
    }

    async fn handle_flush_buckets(&self, mut message: FlushBuckets) {
        for (project_key, pb) in message.buckets.iter_mut() {
            let buckets = std::mem::take(&mut pb.buckets);
            pb.buckets =
                self.check_buckets(*project_key, &pb.project_info, &pb.rate_limits, buckets);
        }

        #[cfg(feature = "processing")]
        if self.inner.config.processing_enabled()
            && let Some(ref store_forwarder) = self.inner.addrs.store_forwarder
        {
            return self
                .encode_metrics_processing(message, store_forwarder)
                .await;
        }

        if self.inner.config.http_global_metrics() {
            self.encode_metrics_global(message)
        } else {
            self.encode_metrics_envelope(message)
        }
    }

    #[cfg(all(test, feature = "processing"))]
    fn redis_rate_limiter_enabled(&self) -> bool {
        self.inner.rate_limiter.is_some()
    }

    async fn handle_message(self, message: EnvelopeProcessor) {
        let ty = message.variant();
        let feature_weights = self.feature_weights(&message);

        metric!(timer(RelayTimers::ProcessMessageDuration), message = ty, {
            let mut cogs = self.inner.cogs.timed(ResourceId::Relay, feature_weights);

            match message {
                EnvelopeProcessor::ProcessEnvelope(m) => {
                    self.handle_process_envelope(&mut cogs, *m).await
                }
                EnvelopeProcessor::ProcessProjectMetrics(m) => {
                    self.handle_process_metrics(&mut cogs, *m)
                }
                EnvelopeProcessor::ProcessBatchedMetrics(m) => {
                    self.handle_process_batched_metrics(&mut cogs, *m)
                }
                EnvelopeProcessor::FlushBuckets(m) => self.handle_flush_buckets(*m).await,
                EnvelopeProcessor::SubmitClientReports(m) => self.handle_submit_client_reports(*m),
            }
        });
    }

    fn feature_weights(&self, message: &EnvelopeProcessor) -> FeatureWeights {
        match message {
            // Envelope is split later and tokens are attributed then.
            EnvelopeProcessor::ProcessEnvelope(_) => AppFeature::Unattributed.into(),
            EnvelopeProcessor::ProcessProjectMetrics(_) => AppFeature::Unattributed.into(),
            EnvelopeProcessor::ProcessBatchedMetrics(_) => AppFeature::Unattributed.into(),
            EnvelopeProcessor::FlushBuckets(v) => v
                .buckets
                .values()
                .map(|s| {
                    if self.inner.config.processing_enabled() {
                        // Processing does not encode the metrics but instead rate limit the metrics,
                        // which scales by count and not size.
                        relay_metrics::cogs::ByCount(&s.buckets).into()
                    } else {
                        relay_metrics::cogs::BySize(&s.buckets).into()
                    }
                })
                .fold(FeatureWeights::none(), FeatureWeights::merge),
            EnvelopeProcessor::SubmitClientReports(_) => AppFeature::ClientReports.into(),
        }
    }
}

impl Service for EnvelopeProcessorService {
    type Interface = EnvelopeProcessor;

    async fn run(self, mut rx: relay_system::Receiver<Self::Interface>) {
        while let Some(message) = rx.recv().await {
            let service = self.clone();
            // Create a new hub to prevent sentry scopes from bleeding to other tasks.
            let hub = relay_log::Hub::with(|h| relay_log::Hub::new_from_top(h));

            self.inner
                .pool
                .spawn_async(Box::pin(service.handle_message(message).bind_hub(hub)))
                .await;
        }
    }
}

pub fn encode_payload(body: &Bytes, http_encoding: HttpEncoding) -> Result<Bytes, std::io::Error> {
    let envelope_body: Vec<u8> = match http_encoding {
        HttpEncoding::Identity => return Ok(body.clone()),
        HttpEncoding::Deflate => {
            let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default());
            encoder.write_all(body.as_ref())?;
            encoder.finish()?
        }
        HttpEncoding::Gzip => {
            let mut encoder = GzEncoder::new(Vec::new(), Compression::default());
            encoder.write_all(body.as_ref())?;
            encoder.finish()?
        }
        HttpEncoding::Br => {
            // Use default buffer size (via 0), medium quality (5), and the default lgwin (22).
            let mut encoder = BrotliEncoder::new(Vec::new(), 0, 5, 22);
            encoder.write_all(body.as_ref())?;
            encoder.into_inner()
        }
        HttpEncoding::Zstd => {
            // Use the fastest compression level, our main objective here is to get the best
            // compression ratio for least amount of time spent.
            let mut encoder = ZstdEncoder::new(Vec::new(), 1)?;
            encoder.write_all(body.as_ref())?;
            encoder.finish()?
        }
    };

    Ok(envelope_body.into())
}

/// An upstream request that submits an envelope via HTTP.
#[derive(Debug)]
pub struct SendEnvelope {
    pub upstream: Option<UpstreamDescriptor>,
    pub envelope: ManagedEnvelope,
    pub body: Bytes,
    pub http_encoding: HttpEncoding,
    pub project_cache: ProjectCacheHandle,
}

impl UpstreamRequest for SendEnvelope {
    fn upstream(&self) -> Option<&UpstreamDescriptor> {
        self.upstream.as_ref()
    }

    fn method(&self) -> reqwest::Method {
        reqwest::Method::POST
    }

    fn path(&self) -> Cow<'_, str> {
        format!("/api/{}/envelope/", self.envelope.scoping().project_id).into()
    }

    fn route(&self) -> &'static str {
        "envelope"
    }

    fn build(&mut self, builder: &mut http::RequestBuilder) -> Result<(), http::HttpError> {
        let envelope_body = self.body.clone();
        metric!(
            distribution(RelayDistributions::UpstreamEnvelopeBodySize) = envelope_body.len() as u64
        );

        let meta = &self.envelope.meta();
        let shard = self.envelope.partition_key().map(|p| p.to_string());
        builder
            .content_encoding(self.http_encoding)
            .header_opt("Origin", meta.origin().map(|url| url.as_str()))
            .header_opt("User-Agent", meta.user_agent())
            .header("X-Sentry-Auth", meta.auth_header())
            .header("X-Forwarded-For", meta.forwarded_for())
            .header("Content-Type", envelope::CONTENT_TYPE)
            .header_opt("X-Sentry-Relay-Shard", shard)
            .body(envelope_body);

        Ok(())
    }

    fn sign(&mut self) -> Option<Sign> {
        Some(Sign::Optional(SignatureType::RequestSign))
    }

    fn respond(
        self: Box<Self>,
        result: Result<http::Response, UpstreamRequestError>,
    ) -> Pin<Box<dyn Future<Output = ()> + Send + Sync>> {
        Box::pin(async move {
            let result = match result {
                Ok(mut response) => response.consume().await.map_err(UpstreamRequestError::Http),
                Err(error) => Err(error),
            };

            match result {
                Ok(()) => self.envelope.accept(),
                Err(error) if error.is_received() => {
                    let scoping = self.envelope.scoping();
                    self.envelope.accept();

                    if let UpstreamRequestError::RateLimited(limits) = error {
                        self.project_cache
                            .get(scoping.project_key)
                            .rate_limits()
                            .merge(limits.scope(&scoping));
                    }
                }
                Err(error) => {
                    // Errors are only logged for what we consider an internal discard reason. These
                    // indicate errors in the infrastructure or implementation bugs.
                    let mut envelope = self.envelope;
                    envelope.reject(Outcome::Invalid(DiscardReason::Internal));
                    relay_log::error!(
                        error = &error as &dyn Error,
                        tags.project_key = %envelope.scoping().project_key,
                        "error sending envelope"
                    );
                }
            }
        })
    }
}

/// A container for metric buckets from multiple projects.
///
/// This container is used to send metrics to the upstream in global batches as part of the
/// [`FlushBuckets`] message if the `http.global_metrics` option is enabled. The container monitors
/// the size of all metrics and allows to split them into multiple batches. See
/// [`insert`](Self::insert) for more information.
#[derive(Debug)]
struct Partition<'a> {
    max_size: usize,
    remaining: usize,
    views: HashMap<ProjectKey, Vec<BucketView<'a>>>,
    project_info: HashMap<ProjectKey, Scoping>,
}

impl<'a> Partition<'a> {
    /// Creates a new partition with the given maximum size in bytes.
    pub fn new(size: usize) -> Self {
        Self {
            max_size: size,
            remaining: size,
            views: HashMap::new(),
            project_info: HashMap::new(),
        }
    }

    /// Inserts a bucket into the partition, splitting it if necessary.
    ///
    /// This function attempts to add the bucket to this partition. If the bucket does not fit
    /// entirely into the partition given its maximum size, the remaining part of the bucket is
    /// returned from this function call.
    ///
    /// If this function returns `Some(_)`, the partition is full and should be submitted to the
    /// upstream immediately. Use [`Self::take`] to retrieve the contents of the
    /// partition. Afterwards, the caller is responsible to call this function again with the
    /// remaining bucket until it is fully inserted.
    pub fn insert(&mut self, bucket: BucketView<'a>, scoping: Scoping) -> Option<BucketView<'a>> {
        let (current, next) = bucket.split(self.remaining, Some(self.max_size));

        if let Some(current) = current {
            self.remaining = self.remaining.saturating_sub(current.estimated_size());
            self.views
                .entry(scoping.project_key)
                .or_default()
                .push(current);

            self.project_info
                .entry(scoping.project_key)
                .or_insert(scoping);
        }

        next
    }

    /// Returns `true` if the partition does not hold any data.
    fn is_empty(&self) -> bool {
        self.views.is_empty()
    }

    /// Returns the serialized buckets for this partition.
    ///
    /// This empties the partition, so that it can be reused.
    fn take(&mut self) -> (Bytes, HashMap<ProjectKey, Scoping>) {
        #[derive(serde::Serialize)]
        struct Wrapper<'a> {
            buckets: &'a HashMap<ProjectKey, Vec<BucketView<'a>>>,
        }

        let buckets = &self.views;
        let payload = serde_json::to_vec(&Wrapper { buckets }).unwrap().into();

        let scopings = std::mem::take(&mut self.project_info);

        self.views.clear();
        self.remaining = self.max_size;

        (payload, scopings)
    }
}

/// An upstream request that submits metric buckets via HTTP.
///
/// This request is not awaited. It automatically tracks outcomes if the request is not received.
#[derive(Debug)]
struct SendMetricsRequest {
    /// Optional upstream override where the request will be sent to.
    upstream: Option<UpstreamDescriptor>,
    /// If the partition key is set, the request is marked with `X-Sentry-Relay-Shard`.
    partition_key: String,
    /// Serialized metric buckets without encoding applied, used for signing.
    unencoded: Bytes,
    /// Serialized metric buckets with the stated HTTP encoding applied.
    encoded: Bytes,
    /// Mapping of all contained project keys to their scoping and extraction mode.
    ///
    /// Used to track outcomes for transmission failures.
    project_info: HashMap<ProjectKey, Scoping>,
    /// Encoding (compression) of the payload.
    http_encoding: HttpEncoding,
    /// Metric outcomes instance to send outcomes on error.
    metric_outcomes: MetricOutcomes,
}

impl SendMetricsRequest {
    fn create_error_outcomes(self) {
        #[derive(serde::Deserialize)]
        struct Wrapper {
            buckets: HashMap<ProjectKey, Vec<MinimalTrackableBucket>>,
        }

        let buckets = match serde_json::from_slice(&self.unencoded) {
            Ok(Wrapper { buckets }) => buckets,
            Err(err) => {
                relay_log::error!(
                    error = &err as &dyn std::error::Error,
                    "failed to parse buckets from failed transmission"
                );
                return;
            }
        };

        for (key, buckets) in buckets {
            let Some(&scoping) = self.project_info.get(&key) else {
                relay_log::error!("missing scoping for project key");
                continue;
            };

            self.metric_outcomes.track(
                scoping,
                &buckets,
                Outcome::Invalid(DiscardReason::Internal),
            );
        }
    }
}

impl UpstreamRequest for SendMetricsRequest {
    fn upstream(&self) -> Option<&UpstreamDescriptor> {
        self.upstream.as_ref()
    }

    fn set_relay_id(&self) -> bool {
        true
    }

    fn sign(&mut self) -> Option<Sign> {
        Some(Sign::Required(SignatureType::Body(self.unencoded.clone())))
    }

    fn method(&self) -> reqwest::Method {
        reqwest::Method::POST
    }

    fn path(&self) -> Cow<'_, str> {
        "/api/0/relays/metrics/".into()
    }

    fn route(&self) -> &'static str {
        "global_metrics"
    }

    fn build(&mut self, builder: &mut http::RequestBuilder) -> Result<(), http::HttpError> {
        metric!(
            distribution(RelayDistributions::UpstreamMetricsBodySize) = self.encoded.len() as u64
        );

        builder
            .content_encoding(self.http_encoding)
            .header("X-Sentry-Relay-Shard", self.partition_key.as_bytes())
            .header(header::CONTENT_TYPE, b"application/json")
            .body(self.encoded.clone());

        Ok(())
    }

    fn respond(
        self: Box<Self>,
        result: Result<http::Response, UpstreamRequestError>,
    ) -> Pin<Box<dyn Future<Output = ()> + Send + Sync>> {
        Box::pin(async {
            match result {
                Ok(mut response) => {
                    response.consume().await.ok();
                }
                Err(error) => {
                    relay_log::error!(error = &error as &dyn Error, "Failed to send metrics batch");

                    // If the request did not arrive at the upstream, we are responsible for outcomes.
                    // Otherwise, the upstream is responsible to log outcomes.
                    if error.is_received() {
                        return;
                    }

                    self.create_error_outcomes()
                }
            }
        })
    }
}

/// Container for global and project level [`Quota`].
#[derive(Copy, Clone, Debug)]
#[cfg(feature = "processing")]
struct CombinedQuotas<'a> {
    global_quotas: &'a [Quota],
    project_quotas: &'a [Quota],
}

#[cfg(feature = "processing")]
impl<'a> CombinedQuotas<'a> {
    /// Returns a new [`CombinedQuotas`].
    pub fn new(global_config: &'a GlobalConfig, project_quotas: &'a [Quota]) -> Self {
        Self {
            global_quotas: &global_config.quotas,
            project_quotas,
        }
    }
}

#[cfg(feature = "processing")]
impl<'a> IntoIterator for CombinedQuotas<'a> {
    type Item = &'a Quota;
    type IntoIter = std::iter::Chain<std::slice::Iter<'a, Quota>, std::slice::Iter<'a, Quota>>;

    fn into_iter(self) -> Self::IntoIter {
        self.global_quotas.iter().chain(self.project_quotas.iter())
    }
}

#[cfg(test)]
mod tests {
    use insta::assert_debug_snapshot;
    use relay_common::glob2::LazyGlob;
    use relay_dynamic_config::ProjectConfig;
    use relay_event_normalization::{
        NormalizationConfig, RedactionRule, TransactionNameConfig, TransactionNameRule,
    };
    use relay_event_schema::protocol::{Event, EventId, TransactionSource};
    use relay_pii::DataScrubbingConfig;
    use relay_protocol::Annotated;
    #[cfg(feature = "processing")]
    use relay_quotas::DataCategory;
    use similar_asserts::assert_eq;

    use crate::testutils::{create_test_processor, create_test_processor_with_addrs};

    #[cfg(feature = "processing")]
    use {
        relay_metrics::BucketValue,
        relay_quotas::{QuotaScope, ReasonCode},
        relay_test::mock_service,
    };

    use super::*;

    async fn process_to_single_envelope<'a>(
        processor: &EnvelopeProcessorService,
        envelope: ManagedEnvelope,
        ctx: processing::Context<'a>,
    ) -> Box<Envelope> {
        let mut outputs = processor.process(envelope, ctx).await;
        assert_eq!(outputs.len(), 1);

        let Output { main, metrics } = outputs.pop().unwrap();

        if let Some(metrics) = metrics {
            metrics.accept(drop);
        }

        main.unwrap()
            .serialize_envelope(ctx.to_forward())
            .unwrap()
            .accept(|envelope| envelope)
    }

    #[cfg(feature = "processing")]
    fn mock_quota(id: &str) -> Quota {
        Quota {
            id: Some(id.into()),
            categories: [DataCategory::MetricBucket].into(),
            scope: QuotaScope::Organization,
            scope_id: None,
            limit: Some(0),
            window: None,
            reason_code: None,
            namespace: None,
        }
    }

    #[cfg(feature = "processing")]
    #[test]
    fn test_dynamic_quotas() {
        let global_config = relay_dynamic_config::GlobalConfig {
            quotas: vec![mock_quota("foo"), mock_quota("bar")],
            ..Default::default()
        };

        let project_quotas = vec![mock_quota("baz"), mock_quota("qux")];

        let dynamic_quotas = CombinedQuotas::new(&global_config, &project_quotas);

        let quota_ids = dynamic_quotas.into_iter().filter_map(|q| q.id.as_deref());
        assert!(quota_ids.eq(["foo", "bar", "baz", "qux"]));
    }

    /// Ensures that if we ratelimit one batch of buckets in [`FlushBuckets`] message, it won't
    /// also ratelimit the next batches in the same message automatically.
    #[cfg(feature = "processing")]
    #[tokio::test]
    async fn test_ratelimit_per_batch() {
        use relay_base_schema::organization::OrganizationId;
        use relay_protocol::FiniteF64;

        let rate_limited_org = Scoping {
            organization_id: OrganizationId::new(1),
            project_id: ProjectId::new(21),
            project_key: ProjectKey::parse("00000000000000000000000000000000").unwrap(),
            key_id: Some(17),
        };

        let not_rate_limited_org = Scoping {
            organization_id: OrganizationId::new(2),
            project_id: ProjectId::new(21),
            project_key: ProjectKey::parse("11111111111111111111111111111111").unwrap(),
            key_id: Some(17),
        };

        let message = {
            let project_info = {
                let quota = Quota {
                    id: Some("testing".into()),
                    categories: [DataCategory::MetricBucket].into(),
                    scope: relay_quotas::QuotaScope::Organization,
                    scope_id: Some(rate_limited_org.organization_id.to_string().into()),
                    limit: Some(0),
                    window: None,
                    reason_code: Some(ReasonCode::new("test")),
                    namespace: None,
                };

                let mut config = ProjectConfig::default();
                config.quotas.push(quota);

                Arc::new(ProjectInfo {
                    config,
                    ..Default::default()
                })
            };

            let project_metrics = |scoping| ProjectBuckets {
                buckets: vec![Bucket {
                    name: "d:spans/bar".into(),
                    value: BucketValue::Counter(FiniteF64::new(1.0).unwrap()),
                    timestamp: UnixTimestamp::now(),
                    tags: Default::default(),
                    width: 10,
                    metadata: BucketMetadata::default(),
                }],
                rate_limits: Default::default(),
                project_info: project_info.clone(),
                scoping,
            };

            let buckets = hashbrown::HashMap::from([
                (
                    rate_limited_org.project_key,
                    project_metrics(rate_limited_org),
                ),
                (
                    not_rate_limited_org.project_key,
                    project_metrics(not_rate_limited_org),
                ),
            ]);

            FlushBuckets {
                partition_key: 0,
                buckets,
            }
        };

        // ensure the order of the map while iterating is as expected.
        assert_eq!(message.buckets.keys().count(), 2);

        let config = {
            let config_json = serde_json::json!({
                "processing": {
                    "enabled": true,
                    "kafka_config": [],
                    "redis": {
                        "server": std::env::var("RELAY_REDIS_URL").unwrap_or_else(|_| "redis://127.0.0.1:6379".to_owned()),
                    }
                }
            });
            Config::from_json_value(config_json).unwrap()
        };

        let (store, handle) = {
            let f = |org_ids: &mut Vec<OrganizationId>, msg: Store| {
                let org_id = match msg {
                    Store::Metrics(x) => x.scoping.organization_id,
                    _ => panic!("received envelope when expecting only metrics"),
                };
                org_ids.push(org_id);
            };

            mock_service("store_forwarder", vec![], f)
        };

        let processor = create_test_processor(config).await;
        assert!(processor.redis_rate_limiter_enabled());

        processor.encode_metrics_processing(message, &store).await;

        drop(store);
        let orgs_not_ratelimited = handle.await.unwrap();

        assert_eq!(
            orgs_not_ratelimited,
            vec![not_rate_limited_org.organization_id]
        );
    }

    #[tokio::test]
    async fn test_browser_version_extraction_with_pii_like_data() {
        let processor = create_test_processor(Default::default()).await;
        let outcome_aggregator = Addr::dummy();
        let event_id = EventId::new();

        let dsn = "https://e12d836b15bb49d7bbf99e64295d995b:@sentry.io/42"
            .parse()
            .unwrap();

        let request_meta = RequestMeta::new(dsn);
        let mut envelope = Envelope::from_request(Some(event_id), request_meta);

        envelope.add_item({
                let mut item = Item::new(ItemType::Event);
                item.set_payload(
                    ContentType::Json,
                    r#"
                    {
                        "request": {
                            "headers": [
                                ["User-Agent", "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/103.0.0.0 Safari/537.36"]
                            ]
                        }
                    }
                "#,
                );
                item
            });

        let mut datascrubbing_settings = DataScrubbingConfig::default();
        // enable all the default scrubbing
        datascrubbing_settings.scrub_data = true;
        datascrubbing_settings.scrub_defaults = true;
        datascrubbing_settings.scrub_ip_addresses = true;

        // Make sure to mask any IP-like looking data
        let pii_config = serde_json::from_str(r#"{"applications": {"**": ["@ip:mask"]}}"#).unwrap();

        let config = ProjectConfig {
            datascrubbing_settings,
            pii_config: Some(pii_config),
            ..Default::default()
        };

        let project_info = ProjectInfo {
            config,
            ..Default::default()
        };

        let envelope = ManagedEnvelope::new(envelope, outcome_aggregator);

        let ctx = processing::Context {
            project_info: &project_info,
            ..processing::Context::for_test()
        };

        let new_envelope = process_to_single_envelope(&processor, envelope, ctx).await;

        let event_item = new_envelope.items().last().unwrap();
        let annotated_event: Annotated<Event> =
            Annotated::from_json_bytes(&event_item.payload()).unwrap();
        let event = annotated_event.into_value().unwrap();
        let headers = event
            .request
            .into_value()
            .unwrap()
            .headers
            .into_value()
            .unwrap();

        // IP-like data must be masked
        assert_eq!(
            Some(
                "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/********* Safari/537.36"
            ),
            headers.get_header("User-Agent")
        );
        // But we still get correct browser and version number
        let contexts = event.contexts.into_value().unwrap();
        let browser = contexts.0.get("browser").unwrap();
        assert_eq!(
            r#"{"browser":"Chrome 103.0.0","name":"Chrome","version":"103.0.0","type":"browser"}"#,
            browser.to_json().unwrap()
        );
    }

    #[tokio::test]
    #[cfg(feature = "processing")]
    async fn test_materialize_dsc() {
        use crate::services::projects::project::PublicKeyConfig;

        let dsn = "https://e12d836b15bb49d7bbf99e64295d995b:@sentry.io/42"
            .parse()
            .unwrap();
        let request_meta = RequestMeta::new(dsn);
        let mut envelope = Envelope::from_request(None, request_meta);

        let dsc = r#"{
            "trace_id": "00000000-0000-0000-0000-000000000001",
            "public_key": "e12d836b15bb49d7bbf99e64295d995b",
            "sample_rate": "0.2"
        }"#;
        envelope.set_dsc(serde_json::from_str(dsc).unwrap());

        let mut item = Item::new(ItemType::Event);
        item.set_payload(ContentType::Json, r#"{}"#);
        envelope.add_item(item);

        let outcome_aggregator = Addr::dummy();
        let managed_envelope = ManagedEnvelope::new(envelope, outcome_aggregator);

        let mut project_info = ProjectInfo::default();
        project_info.public_keys.push(PublicKeyConfig {
            public_key: ProjectKey::parse("e12d836b15bb49d7bbf99e64295d995b").unwrap(),
            numeric_id: Some(1),
        });

        let config = serde_json::json!({
            "processing": {
                "enabled": true,
                "kafka_config": [],
            }
        });

        let processor =
            create_test_processor(Config::from_json_value(config.clone()).unwrap()).await;
        let config = Config::from_json_value(config).unwrap();
        let ctx = processing::Context {
            config: &config,
            project_info: &project_info,
            sampling_project_info: Some(&project_info),
            ..processing::Context::for_test()
        };

        let envelope = process_to_single_envelope(&processor, managed_envelope, ctx).await;
        let event = envelope
            .get_item_by(|item| item.ty() == &ItemType::Event)
            .unwrap();

        let event = Annotated::<Event>::from_json_bytes(&event.payload()).unwrap();
        insta::assert_debug_snapshot!(event.value().unwrap()._dsc, @r###"
        Object(
            {
                "environment": ~,
                "public_key": String(
                    "e12d836b15bb49d7bbf99e64295d995b",
                ),
                "release": ~,
                "replay_id": ~,
                "sample_rate": String(
                    "0.2",
                ),
                "trace_id": String(
                    "00000000000000000000000000000001",
                ),
                "transaction": ~,
            },
        )
        "###);
    }

    fn capture_test_event(transaction_name: &str, source: TransactionSource) -> Vec<String> {
        let mut event = Annotated::<Event>::from_json(
            r#"
            {
                "type": "transaction",
                "transaction": "/foo/",
                "timestamp": 946684810.0,
                "start_timestamp": 946684800.0,
                "contexts": {
                    "trace": {
                        "trace_id": "4c79f60c11214eb38604f4ae0781bfb2",
                        "span_id": "fa90fdead5f74053",
                        "op": "http.server",
                        "type": "trace"
                    }
                },
                "transaction_info": {
                    "source": "url"
                }
            }
            "#,
        )
        .unwrap();
        let e = event.value_mut().as_mut().unwrap();
        e.transaction.set_value(Some(transaction_name.into()));

        e.transaction_info
            .value_mut()
            .as_mut()
            .unwrap()
            .source
            .set_value(Some(source));

        relay_statsd::with_capturing_test_client(|| {
            utils::log_transaction_name_metrics(&mut event, |event| {
                let config = NormalizationConfig {
                    transaction_name_config: TransactionNameConfig {
                        rules: &[TransactionNameRule {
                            pattern: LazyGlob::new("/foo/*/**".to_owned()),
                            expiry: DateTime::<Utc>::MAX_UTC,
                            redaction: RedactionRule::Replace {
                                substitution: "*".to_owned(),
                            },
                        }],
                    },
                    ..Default::default()
                };
                relay_event_normalization::normalize_event(event, &config)
            });
        })
    }

    #[test]
    fn test_log_transaction_metrics_none() {
        let captures = capture_test_event("/nothing", TransactionSource::Url);
        insta::assert_debug_snapshot!(captures, @r###"
        [
            "event.transaction_name_changes:1|c|#source_in:url,changes:none,source_out:sanitized,is_404:false",
        ]
        "###);
    }

    #[test]
    fn test_log_transaction_metrics_rule() {
        let captures = capture_test_event("/foo/john/denver", TransactionSource::Url);
        insta::assert_debug_snapshot!(captures, @r###"
        [
            "event.transaction_name_changes:1|c|#source_in:url,changes:rule,source_out:sanitized,is_404:false",
        ]
        "###);
    }

    #[test]
    fn test_log_transaction_metrics_pattern() {
        let captures = capture_test_event("/something/12345", TransactionSource::Url);
        insta::assert_debug_snapshot!(captures, @r###"
        [
            "event.transaction_name_changes:1|c|#source_in:url,changes:pattern,source_out:sanitized,is_404:false",
        ]
        "###);
    }

    #[test]
    fn test_log_transaction_metrics_both() {
        let captures = capture_test_event("/foo/john/12345", TransactionSource::Url);
        insta::assert_debug_snapshot!(captures, @r###"
        [
            "event.transaction_name_changes:1|c|#source_in:url,changes:both,source_out:sanitized,is_404:false",
        ]
        "###);
    }

    #[test]
    fn test_log_transaction_metrics_no_match() {
        let captures = capture_test_event("/foo/john/12345", TransactionSource::Route);
        insta::assert_debug_snapshot!(captures, @r###"
        [
            "event.transaction_name_changes:1|c|#source_in:route,changes:none,source_out:route,is_404:false",
        ]
        "###);
    }

    #[tokio::test]
    async fn test_process_metrics_bucket_metadata() {
        let mut token = Cogs::noop().timed(ResourceId::Relay, AppFeature::Unattributed);
        let project_key = ProjectKey::parse("a94ae32be2584e0bbd7a4cbb95971fee").unwrap();
        let received_at = Utc::now();
        let config = Config::default();

        let (aggregator, mut aggregator_rx) = Addr::custom();
        let processor = create_test_processor_with_addrs(
            config,
            Addrs {
                aggregator,
                ..Default::default()
            },
        )
        .await;

        let mut item = Item::new(ItemType::Statsd);
        item.set_payload(ContentType::Text, "spans/foo:3182887624:4267882815|s");
        for (source, expected_received_at) in [
            (
                BucketSource::External,
                Some(UnixTimestamp::from_datetime(received_at).unwrap()),
            ),
            (BucketSource::Internal, None),
        ] {
            let message = ProcessMetrics {
                data: MetricData::Raw(vec![item.clone()]),
                project_key,
                source,
                received_at,
                sent_at: Some(Utc::now()),
            };
            processor.handle_process_metrics(&mut token, message);

            let Aggregator::MergeBuckets(merge_buckets) = aggregator_rx.recv().await.unwrap();
            let buckets = merge_buckets.buckets;
            assert_eq!(buckets.len(), 1);
            assert_eq!(buckets[0].metadata.received_at, expected_received_at);
        }
    }

    #[tokio::test]
    async fn test_process_batched_metrics() {
        let mut token = Cogs::noop().timed(ResourceId::Relay, AppFeature::Unattributed);
        let received_at = Utc::now();
        let config = Config::default();

        let (aggregator, mut aggregator_rx) = Addr::custom();
        let processor = create_test_processor_with_addrs(
            config,
            Addrs {
                aggregator,
                ..Default::default()
            },
        )
        .await;

        let payload = r#"{
    "buckets": {
        "11111111111111111111111111111111": [
            {
                "timestamp": 1615889440,
                "width": 0,
                "name": "d:custom/endpoint.response_time@millisecond",
                "type": "d",
                "value": [
                  68.0
                ],
                "tags": {
                  "route": "user_index"
                }
            }
        ],
        "22222222222222222222222222222222": [
            {
                "timestamp": 1615889440,
                "width": 0,
                "name": "d:custom/endpoint.cache_rate@none",
                "type": "d",
                "value": [
                  36.0
                ]
            }
        ]
    }
}
"#;
        let message = ProcessBatchedMetrics {
            payload: Bytes::from(payload),
            source: BucketSource::Internal,
            received_at,
            sent_at: Some(Utc::now()),
        };
        processor.handle_process_batched_metrics(&mut token, message);

        let Aggregator::MergeBuckets(mb1) = aggregator_rx.recv().await.unwrap();
        let Aggregator::MergeBuckets(mb2) = aggregator_rx.recv().await.unwrap();

        let mut messages = vec![mb1, mb2];
        messages.sort_by_key(|mb| mb.project_key);

        let actual = messages
            .into_iter()
            .map(|mb| (mb.project_key, mb.buckets))
            .collect::<Vec<_>>();

        assert_debug_snapshot!(actual, @r###"
        [
            (
                ProjectKey("11111111111111111111111111111111"),
                [
                    Bucket {
                        timestamp: UnixTimestamp(1615889440),
                        width: 0,
                        name: MetricName(
                            "d:custom/endpoint.response_time@millisecond",
                        ),
                        value: Distribution(
                            [
                                68.0,
                            ],
                        ),
                        tags: {
                            "route": "user_index",
                        },
                        metadata: BucketMetadata {
                            merges: 1,
                            received_at: None,
                            extracted_from_indexed: false,
                        },
                    },
                ],
            ),
            (
                ProjectKey("22222222222222222222222222222222"),
                [
                    Bucket {
                        timestamp: UnixTimestamp(1615889440),
                        width: 0,
                        name: MetricName(
                            "d:custom/endpoint.cache_rate@none",
                        ),
                        value: Distribution(
                            [
                                36.0,
                            ],
                        ),
                        tags: {},
                        metadata: BucketMetadata {
                            merges: 1,
                            received_at: None,
                            extracted_from_indexed: false,
                        },
                    },
                ],
            ),
        ]
        "###);
    }
}