relay_config/

config.rs

use std::collections::btree_map::Entry;
use std::collections::{BTreeMap, HashMap};
use std::error::Error;
use std::io::Write;
use std::net::{IpAddr, SocketAddr};
use std::num::{NonZeroU8, NonZeroU16, NonZeroU32};
use std::path::{Path, PathBuf};
use std::str::FromStr;
use std::sync::{Arc, Mutex, PoisonError};
use std::time::Duration;
use std::{env, fmt, fs, io};

use anyhow::Context;
use arc_swap::ArcSwap;
use relay_auth::{PublicKey, RelayId, SecretKey, generate_key_pair, generate_relay_id};
use relay_common::Dsn;
use relay_kafka::{
    ConfigError as KafkaConfigError, KafkaConfigParam, KafkaTopic, KafkaTopicConfig,
    TopicAssignments,
};
use relay_metrics::MetricNamespace;
use serde::de::{DeserializeOwned, Unexpected, Visitor};
use serde::{Deserialize, Deserializer, Serialize, Serializer};
use uuid::Uuid;

use crate::aggregator::{AggregatorServiceConfig, ScopedAggregatorConfig};
use crate::byte_size::ByteSize;
use crate::upstream::UpstreamDescriptor;
use crate::{RedisConfig, RedisConfigs, RedisConfigsRef, build_redis_configs};

const DEFAULT_NETWORK_OUTAGE_GRACE_PERIOD: u64 = 10;

static CONFIG_YAML_HEADER: &str = r###"# Please see the relevant documentation.
# Performance tuning: https://docs.sentry.io/product/relay/operating-guidelines/
# All config options: https://docs.sentry.io/product/relay/options/
"###;

/// Indicates config related errors.
#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
#[non_exhaustive]
pub enum ConfigErrorKind {
    /// Failed to open the file.
    CouldNotOpenFile,
    /// Failed to save a file.
    CouldNotWriteFile,
    /// Parsing YAML failed.
    BadYaml,
    /// Parsing JSON failed.
    BadJson,
    /// Invalid config value
    InvalidValue,
    /// The user attempted to run Relay with processing enabled, but uses a binary that was
    /// compiled without the processing feature.
    ProcessingNotAvailable,
}

impl fmt::Display for ConfigErrorKind {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::CouldNotOpenFile => write!(f, "could not open config file"),
            Self::CouldNotWriteFile => write!(f, "could not write config file"),
            Self::BadYaml => write!(f, "could not parse yaml config file"),
            Self::BadJson => write!(f, "could not parse json config file"),
            Self::InvalidValue => write!(f, "invalid config value"),
            Self::ProcessingNotAvailable => write!(
                f,
                "was not compiled with processing, cannot enable processing"
            ),
        }
    }
}

/// Defines the source of a config error
#[derive(Debug, Default)]
enum ConfigErrorSource {
    /// An error occurring independently.
    #[default]
    None,
    /// An error originating from a configuration file.
    File(PathBuf),
    /// An error originating in a field override (an env var, or a CLI parameter).
    FieldOverride(String),
}

impl fmt::Display for ConfigErrorSource {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            ConfigErrorSource::None => Ok(()),
            ConfigErrorSource::File(file_name) => {
                write!(f, " (file {})", file_name.display())
            }
            ConfigErrorSource::FieldOverride(name) => write!(f, " (field {name})"),
        }
    }
}

/// Indicates config related errors.
#[derive(Debug)]
pub struct ConfigError {
    source: ConfigErrorSource,
    kind: ConfigErrorKind,
}

impl ConfigError {
    #[inline]
    fn new(kind: ConfigErrorKind) -> Self {
        Self {
            source: ConfigErrorSource::None,
            kind,
        }
    }

    #[inline]
    fn field(field: &'static str) -> Self {
        Self {
            source: ConfigErrorSource::FieldOverride(field.to_owned()),
            kind: ConfigErrorKind::InvalidValue,
        }
    }

    #[inline]
    fn file(kind: ConfigErrorKind, p: impl AsRef<Path>) -> Self {
        Self {
            source: ConfigErrorSource::File(p.as_ref().to_path_buf()),
            kind,
        }
    }

    /// Returns the error kind of the error.
    pub fn kind(&self) -> ConfigErrorKind {
        self.kind
    }
}

impl fmt::Display for ConfigError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{}{}", self.kind(), self.source)
    }
}

impl Error for ConfigError {}

enum ConfigFormat {
    Yaml,
    Json,
}

impl ConfigFormat {
    pub fn extension(&self) -> &'static str {
        match self {
            ConfigFormat::Yaml => "yml",
            ConfigFormat::Json => "json",
        }
    }
}

trait ConfigObject: DeserializeOwned + Serialize {
    /// The format in which to serialize this configuration.
    fn format() -> ConfigFormat;

    /// The basename of the config file.
    fn name() -> &'static str;

    /// The full filename of the config file, including the file extension.
    fn path(base: &Path) -> PathBuf {
        base.join(format!("{}.{}", Self::name(), Self::format().extension()))
    }

    /// Loads the config file from a file within the given directory location.
    fn load(base: &Path) -> anyhow::Result<Self> {
        let path = Self::path(base);

        let f = fs::File::open(&path)
            .with_context(|| ConfigError::file(ConfigErrorKind::CouldNotOpenFile, &path))?;
        let f = io::BufReader::new(f);

        let mut source = {
            let file = serde_vars::FileSource::default()
                .with_variable_prefix("${file:")
                .with_variable_suffix("}")
                .with_base_path(base);
            let env = serde_vars::EnvSource::default()
                .with_variable_prefix("${")
                .with_variable_suffix("}");
            (file, env)
        };
        match Self::format() {
            ConfigFormat::Yaml => {
                serde_vars::deserialize(serde_yaml::Deserializer::from_reader(f), &mut source)
                    .with_context(|| ConfigError::file(ConfigErrorKind::BadYaml, &path))
            }
            ConfigFormat::Json => {
                serde_vars::deserialize(&mut serde_json::Deserializer::from_reader(f), &mut source)
                    .with_context(|| ConfigError::file(ConfigErrorKind::BadJson, &path))
            }
        }
    }

    /// Writes the configuration to a file within the given directory location.
    fn save(&self, base: &Path) -> anyhow::Result<()> {
        let path = Self::path(base);
        let mut options = fs::OpenOptions::new();
        options.write(true).truncate(true).create(true);

        // Remove all non-user permissions for the newly created file
        #[cfg(unix)]
        {
            use std::os::unix::fs::OpenOptionsExt;
            options.mode(0o600);
        }

        let mut f = options
            .open(&path)
            .with_context(|| ConfigError::file(ConfigErrorKind::CouldNotWriteFile, &path))?;

        match Self::format() {
            ConfigFormat::Yaml => {
                f.write_all(CONFIG_YAML_HEADER.as_bytes())?;
                serde_yaml::to_writer(&mut f, self)
                    .with_context(|| ConfigError::file(ConfigErrorKind::CouldNotWriteFile, &path))?
            }
            ConfigFormat::Json => serde_json::to_writer_pretty(&mut f, self)
                .with_context(|| ConfigError::file(ConfigErrorKind::CouldNotWriteFile, &path))?,
        }

        f.write_all(b"\n").ok();

        Ok(())
    }
}

/// Structure used to hold information about configuration overrides via
/// CLI parameters or environment variables
#[derive(Debug, Default, Clone)]
pub struct OverridableConfig {
    /// The operation mode of this relay.
    pub mode: Option<String>,
    /// The instance type of this relay.
    pub instance: Option<String>,
    /// The log level of this relay.
    pub log_level: Option<String>,
    /// The log format of this relay.
    pub log_format: Option<String>,
    /// The upstream relay or sentry instance.
    pub upstream: Option<String>,
    /// Alternate upstream provided through a Sentry DSN. Key and project will be ignored.
    pub upstream_dsn: Option<String>,
    /// The host the relay should bind to (network interface).
    pub host: Option<String>,
    /// The port to bind for the unencrypted relay HTTP server.
    pub port: Option<String>,
    /// "true" if processing is enabled "false" otherwise
    pub processing: Option<String>,
    /// the kafka bootstrap.servers configuration string
    pub kafka_url: Option<String>,
    /// the redis server url
    pub redis_url: Option<String>,
    /// The globally unique ID of the relay.
    pub id: Option<String>,
    /// The secret key of the relay
    pub secret_key: Option<String>,
    /// The public key of the relay
    pub public_key: Option<String>,
    /// Outcome source
    pub outcome_source: Option<String>,
    /// shutdown timeout
    pub shutdown_timeout: Option<String>,
    /// Server name reported in the Sentry SDK.
    pub server_name: Option<String>,
}

/// The relay credentials
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
pub struct Credentials {
    /// The secret key of the relay
    pub secret_key: SecretKey,
    /// The public key of the relay
    pub public_key: PublicKey,
    /// The globally unique ID of the relay.
    pub id: RelayId,
}

impl Credentials {
    /// Generates new random credentials.
    pub fn generate() -> Self {
        relay_log::info!("generating new relay credentials");
        let (secret_key, public_key) = generate_key_pair();
        Self {
            secret_key,
            public_key,
            id: generate_relay_id(),
        }
    }

    /// Serializes this configuration to JSON.
    pub fn to_json_string(&self) -> anyhow::Result<String> {
        serde_json::to_string(self)
            .with_context(|| ConfigError::new(ConfigErrorKind::CouldNotWriteFile))
    }
}

impl ConfigObject for Credentials {
    fn format() -> ConfigFormat {
        ConfigFormat::Json
    }
    fn name() -> &'static str {
        "credentials"
    }
}

/// Information on a downstream Relay.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct RelayInfo {
    /// The public key that this Relay uses to authenticate and sign requests.
    pub public_key: PublicKey,

    /// Marks an internal relay that has privileged access to more project configuration.
    #[serde(default)]
    pub internal: bool,
}

impl RelayInfo {
    /// Creates a new RelayInfo
    pub fn new(public_key: PublicKey) -> Self {
        Self {
            public_key,
            internal: false,
        }
    }
}

/// The operation mode of a relay.
#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
pub enum RelayMode {
    /// This relay acts as a proxy for all requests and events.
    ///
    /// Events are normalized and rate limits from the upstream are enforced, but the relay will not
    /// fetch project configurations from the upstream or perform PII stripping. All events are
    /// accepted unless overridden on the file system.
    Proxy,

    /// Project configurations are managed by the upstream.
    ///
    /// Project configurations are always fetched from the upstream, unless they are statically
    /// overridden in the file system. This relay must be allowed in the upstream Sentry. This is
    /// only possible, if the upstream is Sentry directly, or another managed Relay.
    Managed,
}

impl<'de> Deserialize<'de> for RelayMode {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        let s = String::deserialize(deserializer)?;
        match s.as_str() {
            "proxy" => Ok(RelayMode::Proxy),
            "managed" => Ok(RelayMode::Managed),
            "static" => Err(serde::de::Error::custom(
                "Relay mode 'static' has been removed. Please use 'managed' or 'proxy' instead.",
            )),
            other => Err(serde::de::Error::unknown_variant(
                other,
                &["proxy", "managed"],
            )),
        }
    }
}

impl fmt::Display for RelayMode {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        match self {
            RelayMode::Proxy => write!(f, "proxy"),
            RelayMode::Managed => write!(f, "managed"),
        }
    }
}

/// The instance type of Relay.
#[derive(Clone, Copy, Debug, Eq, PartialEq, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
pub enum RelayInstance {
    /// This Relay is run as a default instance.
    Default,

    /// This Relay is run as a canary instance where experiments can be run.
    Canary,
}

impl RelayInstance {
    /// Returns `true` if the [`RelayInstance`] is of type [`RelayInstance::Canary`].
    pub fn is_canary(&self) -> bool {
        matches!(self, RelayInstance::Canary)
    }
}

impl fmt::Display for RelayInstance {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        match self {
            RelayInstance::Default => write!(f, "default"),
            RelayInstance::Canary => write!(f, "canary"),
        }
    }
}

impl FromStr for RelayInstance {
    type Err = fmt::Error;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        match s {
            "canary" => Ok(RelayInstance::Canary),
            _ => Ok(RelayInstance::Default),
        }
    }
}

/// Error returned when parsing an invalid [`RelayMode`].
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct ParseRelayModeError;

impl fmt::Display for ParseRelayModeError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "Relay mode must be one of: managed or proxy")
    }
}

impl Error for ParseRelayModeError {}

impl FromStr for RelayMode {
    type Err = ParseRelayModeError;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        match s {
            "proxy" => Ok(RelayMode::Proxy),
            "managed" => Ok(RelayMode::Managed),
            _ => Err(ParseRelayModeError),
        }
    }
}

/// Returns `true` if this value is equal to `Default::default()`.
fn is_default<T: Default + PartialEq>(t: &T) -> bool {
    *t == T::default()
}

/// Checks if we are running in docker.
fn is_docker() -> bool {
    if fs::metadata("/.dockerenv").is_ok() {
        return true;
    }

    fs::read_to_string("/proc/self/cgroup").is_ok_and(|s| s.contains("/docker"))
}

/// Default value for the "bind" configuration.
fn default_host() -> IpAddr {
    if is_docker() {
        // Docker images rely on this service being exposed
        "0.0.0.0".parse().unwrap()
    } else {
        "127.0.0.1".parse().unwrap()
    }
}

/// Controls responses from the readiness health check endpoint based on authentication.
///
/// Independent of the the readiness condition, shutdown always switches Relay into unready state.
#[derive(Clone, Copy, Debug, Eq, PartialEq, Deserialize, Serialize)]
#[serde(rename_all = "lowercase")]
#[derive(Default)]
pub enum ReadinessCondition {
    /// (default) Relay is ready when authenticated and connected to the upstream.
    ///
    /// Before authentication has succeeded and during network outages, Relay responds as not ready.
    /// Relay reauthenticates based on the `http.auth_interval` parameter. During reauthentication,
    /// Relay remains ready until authentication fails.
    ///
    /// Authentication is only required for Relays in managed mode. Other Relays will only check for
    /// network outages.
    #[default]
    Authenticated,
    /// Relay reports readiness regardless of the authentication and networking state.
    Always,
}

/// Relay specific configuration values.
#[derive(Serialize, Deserialize, Debug, Clone)]
#[serde(default)]
pub struct Relay {
    /// The operation mode of this Relay.
    pub mode: RelayMode,
    /// The instance type of this Relay.
    pub instance: RelayInstance,
    /// The upstream Relay or Sentry instance.
    pub upstream: UpstreamDescriptor,
    /// The upstream advertised to downstream Relay instances.
    ///
    /// This value will be advertised to downstream Relays as the upstream to use when forwarding
    /// data. It can be used for traffic routing and balancing, it must not redirect to a different
    /// Sentry instance.
    ///
    /// Downstream Relays will treat the advertised upstream as the same logical component as this instance
    /// and re-use already established authentication keys.
    pub advertised_upstream: Option<UpstreamDescriptor>,
    /// The host the relay should bind to (network interface).
    pub host: IpAddr,
    /// The port to bind for the unencrypted relay HTTP server.
    pub port: u16,
    /// The host the relay should bind to (network interface) for internally exposed APIs, like
    /// health checks.
    ///
    /// If not configured, internal routes are exposed on the main HTTP server.
    ///
    /// Note: configuring the internal http server on an address which overlaps with the main
    /// server (e.g. main on `0.0.0.0:3000` and internal on `127.0.0.1:3000`) is a misconfiguration
    /// resulting in approximately half of the requests sent to `127.0.0.1:3000` to fail, as the handling
    /// http server is chosen by the operating system 'at random'.
    ///
    /// As a best practice you should always choose different ports to avoid this issue.
    ///
    /// Defaults to [`Self::host`].
    pub internal_host: Option<IpAddr>,
    /// The port to bind for internally exposed APIs.
    ///
    /// Defaults to [`Self::port`].
    pub internal_port: Option<u16>,
    /// Always override project IDs from the URL and DSN with the identifier used at the upstream.
    ///
    /// Enable this setting for Relays used to redirect traffic to a migrated Sentry instance.
    /// Validation of project identifiers can be safely skipped in these cases.
    #[serde(skip_serializing_if = "is_default")]
    pub override_project_ids: bool,
    /// Interval in seconds for Relay to check if its configuration changed.
    ///
    /// If configured Relay will periodically check its configuration for changes
    /// and hot reload it.
    ///
    /// Hot reloading is only supported for a limited set of values.
    ///
    /// Defaults to `None` / off.
    pub config_reload_interval: Option<u64>,
}

impl Default for Relay {
    fn default() -> Self {
        Relay {
            mode: RelayMode::Managed,
            instance: RelayInstance::Default,
            upstream: "https://sentry.io/".parse().unwrap(),
            advertised_upstream: None,
            host: default_host(),
            port: 3000,
            internal_host: None,
            internal_port: None,
            override_project_ids: false,
            config_reload_interval: None,
        }
    }
}

/// Control the metrics.
#[derive(Serialize, Deserialize, Debug, Clone)]
#[serde(default)]
pub struct Metrics {
    /// Hostname and port of the statsd server.
    ///
    /// Defaults to `None`.
    pub statsd: Option<String>,
    /// Buffer size used for metrics sent to the statsd socket.
    ///
    /// Defaults to `None`.
    pub statsd_buffer_size: Option<usize>,
    /// Common prefix that should be added to all metrics.
    ///
    /// Defaults to `"sentry.relay"`.
    pub prefix: String,
    /// Default tags to apply to all metrics.
    pub default_tags: BTreeMap<String, String>,
    /// Tag name to report the hostname to for each metric. Defaults to not sending such a tag.
    pub hostname_tag: Option<String>,
    /// Interval for periodic metrics emitted from Relay.
    ///
    /// Setting it to `0` seconds disables the periodic metrics.
    /// Defaults to 5 seconds.
    pub periodic_secs: u64,
}

impl Default for Metrics {
    fn default() -> Self {
        Metrics {
            statsd: None,
            statsd_buffer_size: None,
            prefix: "sentry.relay".into(),
            default_tags: BTreeMap::new(),
            hostname_tag: None,
            periodic_secs: 5,
        }
    }
}

/// Controls various limits
#[derive(Serialize, Deserialize, Debug, Clone)]
#[serde(default)]
pub struct Limits {
    /// How many requests can be sent concurrently from Relay to the upstream before Relay starts
    /// buffering.
    pub max_concurrent_requests: usize,
    /// How many queries can be sent concurrently from Relay to the upstream before Relay starts
    /// buffering.
    ///
    /// The concurrency of queries is additionally constrained by `max_concurrent_requests`.
    pub max_concurrent_queries: usize,
    /// The maximum payload size for events.
    pub max_event_size: ByteSize,
    /// The maximum size for each attachment.
    pub max_attachment_size: ByteSize,
    /// The maximum amount of attachments in a single envelope.
    pub max_attachment_count: usize,
    /// The maximum combined size for all attachments in an envelope or request.
    pub max_attachments_size: ByteSize,
    /// The maximum size for a TUS upload request body.
    ///
    /// Applies to incoming request bytes at every hop and also to decoded bytes when writing to
    /// objectstore.
    pub max_upload_size: ByteSize,
    /// The maximum combined size for all client reports in an envelope or request.
    pub max_client_reports_size: ByteSize,
    /// The maximum number of client report items per envelope.
    pub max_client_reports_count: usize,
    /// The maximum payload size for a monitor check-in.
    pub max_check_in_size: ByteSize,
    /// The maximum payload size for an entire envelopes. Individual limits still apply.
    pub max_envelope_size: ByteSize,
    /// The maximum combined size for all sessions in an envelope in bytes.
    pub max_sessions_size: ByteSize,
    /// The maximum number of session items per envelope.
    pub max_session_count: usize,
    /// The maximum payload size for general API requests.
    pub max_api_payload_size: ByteSize,
    /// The maximum payload size for file uploads and chunks.
    pub max_api_file_upload_size: ByteSize,
    /// The maximum payload size for chunks
    pub max_api_chunk_upload_size: ByteSize,
    /// The maximum payload size for a profile
    pub max_profile_size: ByteSize,
    /// The maximum payload size for a trace metric.
    pub max_trace_metric_size: ByteSize,
    /// The maximum payload size for a log.
    pub max_log_size: ByteSize,
    /// The maximum number of operations that can occur in a log expansion.
    pub max_expanded_log_operations: usize,
    /// The maximum number of operations that can occur in a span expansion.
    pub max_expanded_span_operations: usize,
    /// The maximum number of operations that can occur in a trace metric expansion.
    pub max_expanded_trace_metric_operations: usize,
    /// The maximum payload size for a span.
    pub max_span_size: ByteSize,
    /// The maximum amount of standalone transaction spans per envelope.
    pub max_standalone_span_count: usize,
    /// The maximum payload size for an item container.
    pub max_container_size: ByteSize,
    /// The maximum payload size for metric buckets.
    pub max_metric_buckets_size: ByteSize,
    /// The maximum payload size for a compressed replay.
    pub max_replay_compressed_size: ByteSize,
    /// The maximum payload size for an uncompressed replay.
    #[serde(alias = "max_replay_size")]
    max_replay_uncompressed_size: ByteSize,
    /// The maximum size for a replay recording Kafka message.
    pub max_replay_message_size: ByteSize,
    /// The byte size limit up to which Relay will retain
    /// keys of invalid/removed attributes.
    ///
    /// This is only relevant for EAP items (spans, logs, …).
    /// In principle, we want to record all deletions of attributes,
    /// but we have to institute some limit to protect our infrastructure
    /// against excessive metadata sizes.
    ///
    /// Defaults to 10KiB.
    pub max_removed_attribute_key_size: ByteSize,
    /// The maximum number of threads to spawn for CPU and web work, each.
    ///
    /// The total number of threads spawned will roughly be `2 * max_thread_count`. Defaults to
    /// the number of logical CPU cores on the host.
    pub max_thread_count: usize,
    /// Controls the maximum concurrency of each worker thread.
    ///
    /// Increasing the concurrency, can lead to a better utilization of worker threads by
    /// increasing the amount of I/O done concurrently.
    //
    /// Currently has no effect on defaults to `1`.
    pub max_pool_concurrency: usize,
    /// The maximum number of seconds a query is allowed to take across retries. Individual requests
    /// have lower timeouts. Defaults to 30 seconds.
    pub query_timeout: u64,
    /// The maximum number of seconds to wait for pending envelopes after receiving a shutdown
    /// signal.
    pub shutdown_timeout: u64,
    /// Server keep-alive timeout in seconds.
    ///
    /// By default, keep-alive is set to 5 seconds.
    pub keepalive_timeout: u64,
    /// Server idle timeout in seconds.
    ///
    /// The idle timeout limits the amount of time a connection is kept open without activity.
    /// Setting this too short may abort connections before Relay is able to send a response.
    ///
    /// By default there is no idle timeout.
    pub idle_timeout: Option<u64>,
    /// Sets the maximum number of concurrent connections.
    ///
    /// Upon reaching the limit, the server will stop accepting connections.
    ///
    /// By default there is no limit.
    pub max_connections: Option<usize>,
    /// The TCP listen backlog.
    ///
    /// Configures the TCP listen backlog for the listening socket of Relay.
    /// See [`man listen(2)`](https://man7.org/linux/man-pages/man2/listen.2.html)
    /// for a more detailed description of the listen backlog.
    ///
    /// Defaults to `1024`, a value [google has been using for a long time](https://git.kernel.org/pub/scm/linux/kernel/git/torvalds/linux.git/commit/?id=19f92a030ca6d772ab44b22ee6a01378a8cb32d4).
    pub tcp_listen_backlog: u32,
}

impl Default for Limits {
    fn default() -> Self {
        Limits {
            max_concurrent_requests: 100,
            max_concurrent_queries: 5,
            max_event_size: ByteSize::mebibytes(1),
            max_attachment_size: ByteSize::mebibytes(200),
            max_attachment_count: 30,
            max_attachments_size: ByteSize::mebibytes(200),
            max_upload_size: ByteSize::mebibytes(10240),
            max_client_reports_size: ByteSize::kibibytes(100),
            max_client_reports_count: 100,
            max_check_in_size: ByteSize::kibibytes(100),
            max_envelope_size: ByteSize::mebibytes(200),
            max_sessions_size: ByteSize::mebibytes(10),
            max_session_count: 100,
            max_api_payload_size: ByteSize::mebibytes(20),
            max_api_file_upload_size: ByteSize::mebibytes(40),
            max_api_chunk_upload_size: ByteSize::mebibytes(100),
            max_profile_size: ByteSize::mebibytes(50),
            max_trace_metric_size: ByteSize::mebibytes(1),
            max_log_size: ByteSize::mebibytes(2),
            max_expanded_log_operations: 3_000_000,
            max_expanded_span_operations: 3_000_000,
            max_expanded_trace_metric_operations: 3_000_000,
            max_span_size: ByteSize::mebibytes(10),
            max_standalone_span_count: 25,
            max_container_size: ByteSize::mebibytes(12),
            max_metric_buckets_size: ByteSize::mebibytes(1),
            max_replay_compressed_size: ByteSize::mebibytes(10),
            max_replay_uncompressed_size: ByteSize::mebibytes(100),
            max_replay_message_size: ByteSize::mebibytes(15),
            max_thread_count: num_cpus::get(),
            max_pool_concurrency: 1,
            query_timeout: 30,
            shutdown_timeout: 10,
            keepalive_timeout: 5,
            idle_timeout: None,
            max_connections: None,
            tcp_listen_backlog: 1024,
            max_removed_attribute_key_size: ByteSize::kibibytes(10),
        }
    }
}

/// Controls traffic steering.
#[derive(Debug, Default, Deserialize, Serialize, Clone)]
#[serde(default)]
pub struct Routing {
    /// Accept and forward unknown Envelope items to the upstream.
    ///
    /// Forwarding unknown items should be enabled in most cases to allow proxying traffic for newer
    /// SDK versions. The upstream in Sentry makes the final decision on which items are valid. If
    /// this is disabled, just the unknown items are removed from Envelopes, and the rest is
    /// processed as usual.
    ///
    /// Defaults to `true` for all Relay modes other than processing mode. In processing mode, this
    /// is disabled by default since the item cannot be handled.
    pub accept_unknown_items: Option<bool>,
}

/// Http content encoding for both incoming and outgoing web requests.
#[derive(Clone, Copy, Debug, Default, Deserialize, Serialize, PartialEq, Eq)]
#[serde(rename_all = "lowercase")]
pub enum HttpEncoding {
    /// Identity function without no compression.
    ///
    /// This is the default encoding and does not require the presence of the `content-encoding`
    /// HTTP header.
    #[default]
    Identity,
    /// Compression using a [zlib](https://en.wikipedia.org/wiki/Zlib) structure with
    /// [deflate](https://en.wikipedia.org/wiki/DEFLATE) encoding.
    ///
    /// These structures are defined in [RFC 1950](https://datatracker.ietf.org/doc/html/rfc1950)
    /// and [RFC 1951](https://datatracker.ietf.org/doc/html/rfc1951).
    Deflate,
    /// A format using the [Lempel-Ziv coding](https://en.wikipedia.org/wiki/LZ77_and_LZ78#LZ77)
    /// (LZ77), with a 32-bit CRC.
    ///
    /// This is the original format of the UNIX gzip program. The HTTP/1.1 standard also recommends
    /// that the servers supporting this content-encoding should recognize `x-gzip` as an alias, for
    /// compatibility purposes.
    Gzip,
    /// A format using the [Brotli](https://en.wikipedia.org/wiki/Brotli) algorithm.
    Br,
    /// A format using the [Zstd](https://en.wikipedia.org/wiki/Zstd) compression algorithm.
    Zstd,
}

impl HttpEncoding {
    /// Parses a [`HttpEncoding`] from its `content-encoding` header value.
    ///
    /// Accepts aliases, whitespace, and mixed case; unknown values fall back to identity.
    /// Use [`Self::parse_strict`] when validating an incoming request header.
    pub fn parse(str: &str) -> Self {
        let str = str.trim();
        if str.eq_ignore_ascii_case("zstd") {
            Self::Zstd
        } else if str.eq_ignore_ascii_case("br") {
            Self::Br
        } else if str.eq_ignore_ascii_case("gzip") || str.eq_ignore_ascii_case("x-gzip") {
            Self::Gzip
        } else if str.eq_ignore_ascii_case("deflate") {
            Self::Deflate
        } else {
            Self::Identity
        }
    }

    /// Parses a canonical `Content-Encoding` value, returning `None` for unsupported values.
    pub fn parse_strict(value: &str) -> Option<Self> {
        match value {
            "identity" => Some(Self::Identity),
            "deflate" => Some(Self::Deflate),
            "gzip" => Some(Self::Gzip),
            "br" => Some(Self::Br),
            "zstd" => Some(Self::Zstd),
            _ => None,
        }
    }

    /// Returns the value for the `content-encoding` HTTP header.
    ///
    /// Returns `None` for [`Identity`](Self::Identity), and `Some` for other encodings.
    pub fn name(&self) -> Option<&'static str> {
        match self {
            Self::Identity => None,
            Self::Deflate => Some("deflate"),
            Self::Gzip => Some("gzip"),
            Self::Br => Some("br"),
            Self::Zstd => Some("zstd"),
        }
    }
}

/// Controls authentication with upstream.
#[derive(Serialize, Deserialize, Debug, Clone)]
#[serde(default)]
pub struct Http {
    /// Timeout for upstream requests in seconds.
    ///
    /// This timeout covers the time from sending the request until receiving response headers.
    /// Neither the connection process and handshakes, nor reading the response body is covered in
    /// this timeout.
    pub timeout: u32,
    /// Timeout for establishing connections with the upstream in seconds.
    ///
    /// This includes SSL handshakes. Relay reuses connections when the upstream supports connection
    /// keep-alive. Connections are retained for a maximum 75 seconds, or 15 seconds of inactivity.
    pub connection_timeout: u32,
    /// Maximum interval between failed request retries in seconds.
    pub max_retry_interval: u32,
    /// The custom HTTP Host header to send to the upstream.
    pub host_header: Option<String>,
    /// The interval in seconds at which Relay attempts to reauthenticate with the upstream server.
    ///
    /// Re-authentication happens even when Relay is idle. If authentication fails, Relay reverts
    /// back into startup mode and tries to establish a connection. During this time, incoming
    /// envelopes will be buffered.
    ///
    /// Defaults to `600` (10 minutes).
    pub auth_interval: Option<u64>,
    /// The maximum time of experiencing uninterrupted network failures until Relay considers that
    /// it has encountered a network outage in seconds.
    ///
    /// During a network outage relay will try to reconnect and will buffer all upstream messages
    /// until it manages to reconnect.
    pub outage_grace_period: u64,
    /// The time Relay waits before retrying an upstream request, in seconds.
    ///
    /// This time is only used before going into a network outage mode.
    pub retry_delay: u64,
    /// The interval in seconds for continued failed project fetches at which Relay will error.
    ///
    /// A successful fetch resets this interval. Relay does nothing during long
    /// times without emitting requests.
    pub project_failure_interval: u64,
    /// Content encoding to apply to upstream store requests.
    ///
    /// By default, Relay applies `zstd` content encoding to compress upstream requests. Compression
    /// can be disabled to reduce CPU consumption, but at the expense of increased network traffic.
    ///
    /// This setting applies to all store requests of SDK data, including events, transactions,
    /// envelopes and sessions. At the moment, this does not apply to Relay's internal queries.
    ///
    /// Available options are:
    ///
    ///  - `identity`: Disables compression.
    ///  - `deflate`: Compression using a zlib header with deflate encoding.
    ///  - `gzip` (default): Compression using gzip.
    ///  - `br`: Compression using the brotli algorithm.
    ///  - `zstd`: Compression using the zstd algorithm.
    pub encoding: HttpEncoding,
    /// Submit metrics globally through a shared endpoint.
    ///
    /// As opposed to regular envelopes which are sent to an endpoint inferred from the project's
    /// DSN, this submits metrics to the global endpoint with Relay authentication.
    ///
    /// This option does not have any effect on processing mode.
    pub global_metrics: bool,
    /// Controls whether the forward endpoint is enabled.
    ///
    /// The forward endpoint forwards unknown API requests to the upstream.
    ///
    /// Relay instances with processing enabled are expected to support the latest API and do never
    /// support forwarding requests to Sentry.
    pub forward: bool,
    /// Enables an async DNS resolver through the `hickory-dns` crate, which uses an LRU cache for
    /// the resolved entries. This helps to limit the amount of requests made to the upstream DNS
    /// server (important for K8s infrastructure).
    pub dns_cache: bool,
}

impl Default for Http {
    fn default() -> Self {
        Http {
            timeout: 5,
            connection_timeout: 3,
            max_retry_interval: 60, // 1 minute
            host_header: None,
            auth_interval: Some(600), // 10 minutes
            outage_grace_period: DEFAULT_NETWORK_OUTAGE_GRACE_PERIOD,
            retry_delay: 1,
            project_failure_interval: 90,
            encoding: HttpEncoding::Zstd,
            global_metrics: false,
            forward: true,
            dns_cache: true,
        }
    }
}

/// Strategy used to assign envelopes to buffer partitions.
#[derive(Clone, Copy, Debug, Eq, PartialEq, Default, Deserialize, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum EnvelopeSpoolPartitioning {
    /// Envelopes with the same project key pair land on the same partition.
    ///
    /// Keeps per-project state, disk files, and event ordering co-located on one partition.
    ProjectKeyPair,
    /// Envelopes are distributed across partitions in a round-robin fashion (default).
    ///
    /// This prevents "hot" partitions when a single project pair dominates traffic, but has
    /// trade-offs:
    /// - Per-project LIFO ordering is no longer preserved across partitions.
    /// - Per-partition memory footprint grows since every partition sees every project.
    #[default]
    RoundRobin,
}

/// Persistent buffering configuration for incoming envelopes.
#[derive(Debug, Serialize, Deserialize, Clone)]
#[serde(default)]
pub struct EnvelopeSpool {
    /// The path of the SQLite database file(s) which persist the data.
    ///
    /// Based on the number of partitions, more database files will be created within the same path.
    ///
    /// If not set, the envelopes will be buffered in memory.
    pub path: Option<PathBuf>,
    /// The maximum size of the buffer to keep, in bytes.
    ///
    /// When the on-disk buffer reaches this size, new envelopes will be dropped.
    ///
    /// Defaults to 500MB.
    pub max_disk_size: ByteSize,
    /// Size of the batch of compressed envelopes that are spooled to disk at once.
    ///
    /// Note that this is the size after which spooling will be triggered but it does not guarantee
    /// that exactly this size will be spooled, it can be greater or equal.
    ///
    /// Defaults to 10 KiB.
    pub batch_size_bytes: ByteSize,
    /// Time after which a batch is flushed, regardless of batch size.
    ///
    /// The age of the batch is only checked when a new envelope comes in, but in practice this
    /// has the desired effect: High-volume projects always form full batches, low-volume batches
    /// flush individual envelopes to keep memory usage low.
    pub flush_timeout_secs: Option<u64>,
    /// Maximum time between receiving the envelope and processing it.
    ///
    /// When envelopes spend too much time in the buffer (e.g. because their project cannot be loaded),
    /// they are dropped.
    ///
    /// Defaults to 24h.
    pub max_envelope_delay_secs: u64,
    /// The refresh frequency in ms of how frequently disk usage is updated by querying SQLite
    /// internal page stats.
    ///
    /// Defaults to 100ms.
    pub disk_usage_refresh_frequency_ms: u64,
    /// The relative memory usage above which the buffer service will stop dequeueing envelopes.
    ///
    /// Only applies when [`Self::path`] is set.
    ///
    /// This value should be lower than [`Health::max_memory_percent`] to prevent flip-flopping.
    ///
    /// Warning: This threshold can cause the buffer service to deadlock when the buffer consumes
    /// excessive memory (as influenced by [`Self::batch_size_bytes`]).
    ///
    /// This scenario arises when the buffer stops spooling due to reaching the
    /// [`Self::max_backpressure_memory_percent`] limit, but the batch threshold for spooling
    /// ([`Self::batch_size_bytes`]) is never reached. As a result, no data is spooled, memory usage
    /// continues to grow, and the system becomes deadlocked.
    ///
    /// ### Example
    /// Suppose the system has 1GB of available memory and is configured to spool only after
    /// accumulating 10GB worth of envelopes. If Relay consumes 900MB of memory, it will stop
    /// unspooling due to reaching the [`Self::max_backpressure_memory_percent`] threshold.
    ///
    /// However, because the buffer hasn't accumulated the 10GB needed to trigger spooling,
    /// no data will be offloaded. Memory usage keeps increasing until it hits the
    /// [`Health::max_memory_percent`] threshold, e.g., at 950MB. At this point:
    ///
    /// - No more envelopes are accepted.
    /// - The buffer remains stuck, as unspooling won’t resume until memory drops below 900MB which
    ///   will not happen.
    /// - A deadlock occurs, with the system unable to recover without manual intervention.
    ///
    /// Defaults to 90% (5% less than max memory).
    pub max_backpressure_memory_percent: f32,
    /// Number of partitions of the buffer.
    ///
    /// A partition is a separate instance of the buffer which has its own isolated queue, stacks
    /// and other resources.
    ///
    /// Defaults to 1.
    pub partitions: NonZeroU8,
    /// Strategy used to assign envelopes to buffer partitions.
    ///
    /// Defaults to partitioning by `ProjectKeyPair`, which keeps all envelopes of a given project
    /// pair on the same partition. See [`EnvelopeSpoolPartitioning`] for alternatives and
    /// trade-offs.
    pub partitioning: EnvelopeSpoolPartitioning,
    /// Maximum number of envelopes unspooled from disk per second.
    ///
    /// The limit applies per Relay instance, not per spool partition.
    ///
    /// Defaults to `None`, which disables the limit.
    pub max_unspool_envelopes_per_second: Option<NonZeroU32>,
    /// Whether the database defined in `path` is on an ephemeral storage disk.
    ///
    /// With `ephemeral: true`, Relay does not spool in-flight data to disk
    /// during graceful shutdown. Instead, it attempts to process all data before it terminates.
    ///
    /// Defaults to `false`.
    pub ephemeral: bool,
}

impl Default for EnvelopeSpool {
    fn default() -> Self {
        Self {
            path: None,
            max_disk_size: ByteSize::mebibytes(500),
            batch_size_bytes: ByteSize::kibibytes(10),
            max_envelope_delay_secs: 24 * 60 * 60,
            disk_usage_refresh_frequency_ms: 100,
            max_backpressure_memory_percent: 0.8,
            partitions: NonZeroU8::new(1).unwrap(),
            partitioning: EnvelopeSpoolPartitioning::default(),
            ephemeral: false,
            flush_timeout_secs: None,
            max_unspool_envelopes_per_second: None,
        }
    }
}

/// Persistent buffering configuration.
#[derive(Debug, Serialize, Deserialize, Default, Clone)]
#[serde(default)]
pub struct Spool {
    /// Configuration for envelope spooling.
    pub envelopes: EnvelopeSpool,
}

/// Controls internal caching behavior.
#[derive(Serialize, Deserialize, Debug, Clone)]
#[serde(default)]
pub struct Cache {
    /// The full project state will be requested by this Relay if set to `true`.
    ///
    /// Relay instances that receive the full project config have full access to quota config
    /// and perform dynamic sampling.
    pub project_request_full_config: bool,
    /// The cache timeout for project configurations in seconds.
    pub project_expiry: u32,
    /// Continue using project state this many seconds after cache expiry while a new state is
    /// being fetched. This is added on top of `project_expiry`.
    ///
    /// Default is 2 minutes.
    pub project_grace_period: u32,
    /// Refresh a project after the specified seconds.
    ///
    /// The time must be between expiry time and the grace period.
    ///
    /// By default there are no refreshes enabled.
    pub project_refresh_interval: Option<u32>,
    /// The cache timeout for downstream relay info (public keys) in seconds.
    pub relay_expiry: u32,
    /// The cache timeout for non-existing entries.
    pub miss_expiry: u32,
    /// The buffer timeout for batched project config queries before sending them upstream in ms.
    pub batch_interval: u32,
    /// The buffer timeout for batched queries of downstream relays in ms. Defaults to 100ms.
    pub downstream_relays_batch_interval: u32,
    /// The maximum number of project configs to fetch from Sentry at once. Defaults to 500.
    ///
    /// `cache.batch_interval` controls how quickly batches are sent, this controls the batch size.
    pub batch_size: usize,
    /// Interval for watching local cache override files in seconds.
    pub file_interval: u32,
    /// Interval for fetching new global configs from the upstream, in seconds.
    pub global_config_fetch_interval: u32,
}

impl Default for Cache {
    fn default() -> Self {
        Cache {
            project_request_full_config: false,
            project_expiry: 300,       // 5 minutes
            project_grace_period: 120, // 2 minutes
            project_refresh_interval: None,
            relay_expiry: 3600,                    // 1 hour
            miss_expiry: 60,                       // 1 minute
            batch_interval: 100,                   // 100ms
            downstream_relays_batch_interval: 100, // 100ms
            batch_size: 500,
            file_interval: 10,                // 10 seconds
            global_config_fetch_interval: 10, // 10 seconds
        }
    }
}

/// Controls Sentry-internal event processing.
#[derive(Serialize, Deserialize, Debug, Clone)]
#[serde(default)]
pub struct Processing {
    /// True if the Relay should do processing. Defaults to `false`.
    pub enabled: bool,
    /// GeoIp DB file source.
    pub geoip_path: Option<PathBuf>,
    /// Maximum future timestamp of ingested events.
    pub max_secs_in_future: u32,
    /// Maximum age of ingested sessions. Older sessions will be dropped.
    pub max_session_secs_in_past: u32,
    /// Kafka producer configurations.
    pub kafka_config: Vec<KafkaConfigParam>,
    /// Additional kafka producer configurations.
    ///
    /// The `kafka_config` is the default producer configuration used for all topics. A secondary
    /// kafka config can be referenced in `topics:` like this:
    ///
    /// ```yaml
    /// secondary_kafka_configs:
    ///   mycustomcluster:
    ///     - name: 'bootstrap.servers'
    ///       value: 'sentry_kafka_metrics:9093'
    ///
    /// topics:
    ///   transactions: ingest-transactions
    ///   metrics:
    ///     name: ingest-metrics
    ///     config: mycustomcluster
    /// ```
    ///
    /// Then metrics will be produced to an entirely different Kafka cluster.
    pub secondary_kafka_configs: BTreeMap<String, Vec<KafkaConfigParam>>,
    /// Kafka topic names.
    pub topics: TopicAssignments,
    /// Whether to validate the supplied topics by calling Kafka's metadata endpoints.
    pub kafka_validate_topics: bool,
    /// Upper bound, in seconds, on the random delay before a Kafka producer swaps its client
    /// after the Kafka config changed.
    ///
    /// Spreads swaps out so that a config change does not have every pod in the fleet reconnect
    /// at once. Keep it well below how often the Kafka config is expected to change. Tests set
    /// this to `0` to observe a swap without waiting.
    pub reloading_producer_jitter_secs: u64,
    /// Redis hosts to connect to for storing state for rate limits.
    pub redis: Option<RedisConfigs>,
    /// Maximum chunk size of attachments for Kafka.
    pub attachment_chunk_size: ByteSize,
    /// Prefix to use when looking up project configs in Redis. Defaults to "relayconfig".
    pub projectconfig_cache_prefix: String,
    /// Maximum rate limit to report to clients.
    pub max_rate_limit: Option<u32>,
    /// Configures the quota cache ratio between `0.0` and `1.0`.
    ///
    /// The quota cache, caches the specified ratio of remaining quota in memory to reduce the
    /// amount of synchronizations required with Redis.
    ///
    /// The ratio is applied to the (per second) rate of the quota, not the total limit.
    /// For example a quota with limit 100 with a 10 second window is treated equally to a quota of
    /// 10 with a 1 second window.
    ///
    /// By default quota caching is disabled.
    pub quota_cache_ratio: Option<f32>,
    /// Relative amount of the total quota limit to which quota caching is applied.
    ///
    /// If exceeded, the rate limiter will no longer cache the quota and sync with Redis on every call instead.
    /// Lowering this value reduces the probability of incorrectly over-accepting.
    ///
    /// Must be between `0.0` and `1.0`, by default there is no limit configured.
    pub quota_cache_max: Option<f32>,
    /// Configuration for the objectstore service.
    #[serde(alias = "upload")]
    pub objectstore: ObjectstoreServiceConfig,
}

impl Default for Processing {
    /// Constructs a disabled processing configuration.
    fn default() -> Self {
        Self {
            enabled: false,
            geoip_path: None,
            max_secs_in_future: 60,                  // 1 minute
            max_session_secs_in_past: 5 * 24 * 3600, // 5 days
            kafka_config: Vec::new(),
            secondary_kafka_configs: BTreeMap::new(),
            topics: TopicAssignments::default(),
            kafka_validate_topics: false,
            reloading_producer_jitter_secs: 60,
            redis: None,
            attachment_chunk_size: ByteSize::mebibytes(1),
            projectconfig_cache_prefix: "relayconfig".to_owned(),
            max_rate_limit: Some(300), // 5 minutes
            quota_cache_ratio: None,
            quota_cache_max: None,
            objectstore: ObjectstoreServiceConfig::default(),
        }
    }
}

/// Picks a cluster name for the default `kafka_config` that no secondary config uses.
///
/// Only relevant to [`ConfigSnapshot::kafka_config_blob`]: the blob names every cluster, while
/// Relay's own config leaves the default one anonymous. Appends underscores to `default` until
/// it is unused.
fn default_cluster_name(processing: &Processing) -> String {
    let mut name = String::from("default");
    while processing.secondary_kafka_configs.contains_key(&name) {
        name.push('_');
    }
    name
}

/// Configuration for normalization in this Relay.
#[derive(Debug, Default, Serialize, Deserialize, Clone)]
#[serde(default)]
pub struct Normalization {
    /// Level of normalization for Relay to apply to incoming data.
    pub level: NormalizationLevel,
}

/// Configuration for the level of normalization this Relay should do.
#[derive(Copy, Clone, Debug, Default, Serialize, Deserialize, Eq, PartialEq)]
#[serde(rename_all = "lowercase")]
pub enum NormalizationLevel {
    /// Runs normalization, excluding steps that break future compatibility.
    ///
    /// Processing Relays run [`NormalizationLevel::Full`] if this option is set.
    #[default]
    Default,
    /// Run full normalization.
    ///
    /// It includes steps that break future compatibility and should only run in
    /// the last layer of relays.
    Full,
}

/// Configuration options for objectstore's auth scheme.
#[derive(Serialize, Deserialize, Clone)]
pub struct ObjectstoreAuthConfig {
    /// Identifier for the private key used to sign objectstore's tokens. Must correspond to a
    /// public key configured in objectstore.
    pub key_id: String,

    /// EdDSA private key used to sign Objectstore's tokens, in PEM format.
    pub signing_key: String,
}

impl fmt::Debug for ObjectstoreAuthConfig {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("ObjectstoreAuthConfig")
            .field("key_id", &self.key_id)
            .field("signing_key", &"[redacted]")
            .finish()
    }
}

/// Configuration values for the objectstore service.
#[derive(Serialize, Deserialize, Debug, Clone)]
#[serde(default)]
pub struct ObjectstoreServiceConfig {
    /// The base URL for the objectstore service.
    ///
    /// This defaults to [`None`], which means that the service will be disabled,
    /// unless a proper configuration is provided.
    pub objectstore_url: Option<String>,

    /// Maximum concurrency of uploads.
    pub max_concurrent_requests: usize,

    /// Maximum size of the service input queue when `max_concurrent_requests` is saturated.
    ///
    /// The service will loadshed if this threshold is reached.
    pub max_backlog: usize,

    /// Maximum duration of an attachment upload in seconds. Uploads that take longer are discarded.
    ///
    /// NOTE: This timeout applies to attachments that are already in-memory. Streaming uploads
    /// might take longer and are restricted independently by [`Self::stream_timeout`].
    pub timeout: u64,

    /// Maximum duration of an upload stream.
    ///
    /// Streams get a larger default timeout because their duration depends on the client
    /// as well as the server.
    pub stream_timeout: u64,

    /// Time between upload attempts.
    pub retry_delay: f64,

    /// Maximum number of attempts made to upload.
    pub max_attempts: NonZeroU16,

    /// Configuration values for objectstore's auth scheme.
    pub auth: Option<ObjectstoreAuthConfig>,
}

impl Default for ObjectstoreServiceConfig {
    fn default() -> Self {
        Self {
            objectstore_url: None,
            max_concurrent_requests: 10,
            max_backlog: 20,
            timeout: 60,
            stream_timeout: 5 * 60, // synced with `Upload::timeout`
            retry_delay: 1.0,
            max_attempts: NonZeroU16::new(5).unwrap(),
            auth: None,
        }
    }
}

/// Determines how to emit outcomes.
/// For compatibility reasons, this can either be true, false or AsClientReports
#[derive(Copy, Clone, Debug, PartialEq, Eq)]

pub enum EmitOutcomes {
    /// Do not emit any outcomes.
    None,
    /// Emit outcomes as client reports.
    AsClientReports,
    /// Emit outcomes as outcomes.
    AsOutcomes,
}

impl EmitOutcomes {
    /// Returns true of outcomes are emitted via http, kafka, or client reports.
    pub fn any(&self) -> bool {
        !matches!(self, EmitOutcomes::None)
    }
}

impl Serialize for EmitOutcomes {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        // For compatibility, serialize None and AsOutcomes as booleans.
        match self {
            Self::None => serializer.serialize_bool(false),
            Self::AsClientReports => serializer.serialize_str("as_client_reports"),
            Self::AsOutcomes => serializer.serialize_bool(true),
        }
    }
}

struct EmitOutcomesVisitor;

impl Visitor<'_> for EmitOutcomesVisitor {
    type Value = EmitOutcomes;

    fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
        formatter.write_str("true, false, 'as_client_reports'")
    }

    fn visit_bool<E>(self, v: bool) -> Result<Self::Value, E>
    where
        E: serde::de::Error,
    {
        Ok(if v {
            EmitOutcomes::AsOutcomes
        } else {
            EmitOutcomes::None
        })
    }

    fn visit_str<E>(self, v: &str) -> Result<Self::Value, E>
    where
        E: serde::de::Error,
    {
        match v {
            "as_client_reports" => Ok(EmitOutcomes::AsClientReports),
            _ => Err(E::invalid_value(Unexpected::Str(v), &self)),
        }
    }
}

impl<'de> Deserialize<'de> for EmitOutcomes {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        deserializer.deserialize_any(EmitOutcomesVisitor)
    }
}

/// Outcome generation specific configuration values.
#[derive(Serialize, Deserialize, Debug, Clone)]
#[serde(default)]
pub struct Outcomes {
    /// Controls whether outcomes will be emitted when processing is disabled.
    /// Processing relays always emit outcomes (for backwards compatibility).
    /// Can take the following values: false, "as_client_reports", true
    pub emit_outcomes: EmitOutcomes,
    /// Defines the source string registered in the outcomes originating from
    /// this Relay (typically something like the region or the layer).
    pub source: Option<String>,
}

impl Default for Outcomes {
    fn default() -> Self {
        Outcomes {
            emit_outcomes: EmitOutcomes::AsClientReports,
            source: None,
        }
    }
}

/// Minimal version of a config for dumping out.
#[derive(Serialize, Deserialize, Debug, Default)]
pub struct MinimalConfig {
    /// The relay part of the config.
    pub relay: Relay,
}

impl MinimalConfig {
    /// Saves the config in the given config folder as config.yml
    pub fn save_in_folder<P: AsRef<Path>>(&self, p: P) -> anyhow::Result<()> {
        let path = p.as_ref();
        if fs::metadata(path).is_err() {
            fs::create_dir_all(path)
                .with_context(|| ConfigError::file(ConfigErrorKind::CouldNotOpenFile, path))?;
        }
        self.save(path)
    }
}

impl ConfigObject for MinimalConfig {
    fn format() -> ConfigFormat {
        ConfigFormat::Yaml
    }

    fn name() -> &'static str {
        "config"
    }
}

/// Alternative serialization of RelayInfo for config file using snake case.
mod config_relay_info {
    use serde::ser::SerializeMap;

    use super::*;

    // Uses snake_case as opposed to camelCase.
    #[derive(Debug, Serialize, Deserialize, Clone)]
    struct RelayInfoConfig {
        public_key: PublicKey,
        #[serde(default)]
        internal: bool,
    }

    impl From<RelayInfoConfig> for RelayInfo {
        fn from(v: RelayInfoConfig) -> Self {
            RelayInfo {
                public_key: v.public_key,
                internal: v.internal,
            }
        }
    }

    impl From<RelayInfo> for RelayInfoConfig {
        fn from(v: RelayInfo) -> Self {
            RelayInfoConfig {
                public_key: v.public_key,
                internal: v.internal,
            }
        }
    }

    pub(super) fn deserialize<'de, D>(des: D) -> Result<HashMap<RelayId, RelayInfo>, D::Error>
    where
        D: Deserializer<'de>,
    {
        let map = HashMap::<RelayId, RelayInfoConfig>::deserialize(des)?;
        Ok(map.into_iter().map(|(k, v)| (k, v.into())).collect())
    }

    pub(super) fn serialize<S>(elm: &HashMap<RelayId, RelayInfo>, ser: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        let mut map = ser.serialize_map(Some(elm.len()))?;

        for (k, v) in elm {
            map.serialize_entry(k, &RelayInfoConfig::from(v.clone()))?;
        }

        map.end()
    }
}

/// Authentication options.
#[derive(Serialize, Deserialize, Debug, Clone)]
#[serde(default)]
pub struct AuthConfig {
    /// Controls responses from the readiness health check endpoint based on authentication.
    #[serde(skip_serializing_if = "is_default")]
    pub ready: ReadinessCondition,

    /// Statically authenticated downstream relays.
    #[serde(with = "config_relay_info")]
    pub static_relays: HashMap<RelayId, RelayInfo>,

    /// How old a signature can be before it is considered invalid, in seconds.
    ///
    /// Defaults to 5 minutes.
    pub signature_max_age: u64,
}

impl Default for AuthConfig {
    fn default() -> Self {
        Self {
            ready: ReadinessCondition::default(),
            static_relays: HashMap::new(),
            signature_max_age: 300, // 5 minutes
        }
    }
}

/// GeoIp database configuration options.
#[derive(Serialize, Deserialize, Debug, Default, Clone)]
pub struct GeoIpConfig {
    /// The path to GeoIP database.
    pub path: Option<PathBuf>,
}

/// Settings to control Relay's health checks.
///
/// After breaching one of the configured thresholds, Relay will
/// return an `unhealthy` status from its health endpoint.
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq)]
#[serde(default)]
pub struct Health {
    /// Interval to refresh internal health checks.
    ///
    /// Shorter intervals will decrease the time it takes the health check endpoint to report
    /// issues, but can also increase sporadic unhealthy responses.
    ///
    /// Defaults to `3000`` (3 seconds).
    pub refresh_interval_ms: u64,
    /// Maximum memory watermark in bytes.
    ///
    /// By default, there is no absolute limit set and the watermark
    /// is only controlled by setting [`Self::max_memory_percent`].
    pub max_memory_bytes: Option<ByteSize>,
    /// Maximum memory watermark as a percentage of maximum system memory.
    ///
    /// Defaults to `0.95` (95%).
    pub max_memory_percent: f32,
    /// Health check probe timeout in milliseconds.
    ///
    /// Any probe exceeding the timeout will be considered failed.
    /// This limits the max execution time of Relay health checks.
    ///
    /// Defaults to 900 milliseconds.
    pub probe_timeout_ms: u64,
    /// The refresh frequency of memory stats which are used to poll memory
    /// usage of Relay.
    ///
    /// The implementation of memory stats guarantees that the refresh will happen at
    /// least every `x` ms since memory readings are lazy and are updated only if needed.
    pub memory_stat_refresh_frequency_ms: u64,
}

impl Default for Health {
    fn default() -> Self {
        Self {
            refresh_interval_ms: 3000,
            max_memory_bytes: None,
            max_memory_percent: 0.95,
            probe_timeout_ms: 900,
            memory_stat_refresh_frequency_ms: 100,
        }
    }
}

/// COGS configuration.
#[derive(Serialize, Deserialize, Debug, Clone)]
#[serde(default)]
pub struct Cogs {
    /// Maximium amount of COGS measurements allowed to backlog.
    ///
    /// Any additional COGS measurements recorded will be dropped.
    ///
    /// Defaults to `10_000`.
    pub max_queue_size: u64,
    /// Relay COGS resource id.
    ///
    /// All Relay related COGS measurements are emitted with this resource id.
    ///
    /// Defaults to `relay_service`.
    pub relay_resource_id: String,
}

impl Default for Cogs {
    fn default() -> Self {
        Self {
            max_queue_size: 10_000,
            relay_resource_id: "relay_service".to_owned(),
        }
    }
}

/// Configuration for the upload service.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(default)]
pub struct Upload {
    /// Maximum number of uploads that the service accepts.
    ///
    /// Additional uploads will be rejected.
    pub max_concurrent_requests: usize,
    /// Maximum time spent trying to upload, in seconds.
    pub timeout: u64,
    /// The maximum time between creating the upload and uploading the data / the attachment placeholder.
    ///
    /// In seconds.
    pub max_age: i64,

    /// Credentials used for signing & verifying upload locations.
    ///
    /// If omitted, relay's default [`Credentials`] are used.
    pub credentials: Option<UploadCredentials>,
}

impl Default for Upload {
    fn default() -> Self {
        Self {
            max_concurrent_requests: 100,
            timeout: 5 * 60,  // five minutes
            max_age: 60 * 60, // 1h
            credentials: None,
        }
    }
}

/// Credentials used for signing & verifying upload locations.
#[derive(Clone, Serialize, Deserialize)]
pub struct UploadCredentials {
    /// Key used to sign upload locations.
    #[cfg(feature = "processing")]
    pub signing_key: SecretKey,

    /// Key used to verify upload locations.
    pub verification_key: PublicKey,
}

impl fmt::Debug for UploadCredentials {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        let Self {
            #[cfg(feature = "processing")]
                signing_key: _,
            verification_key,
        } = self;
        let mut b = f.debug_struct("UploadCredentials");
        #[cfg(feature = "processing")]
        b.field("signing_key", &"[redacted]");
        b.field("verification_key", verification_key).finish()
    }
}

/// All configuration values that can be deserialized from `config.yml`.
#[derive(Serialize, Deserialize, Debug, Default, Clone)]
#[serde(default)]
#[allow(missing_docs)]
pub struct ConfigValues {
    pub relay: Relay,
    pub http: Http,
    pub cache: Cache,
    pub spool: Spool,
    pub limits: Limits,
    pub logging: relay_log::LogConfig,
    pub routing: Routing,
    pub metrics: Metrics,
    pub sentry: relay_log::SentryConfig,
    pub processing: Processing,
    pub outcomes: Outcomes,
    pub aggregator: AggregatorServiceConfig,
    pub secondary_aggregators: Vec<ScopedAggregatorConfig>,
    pub auth: AuthConfig,
    pub geoip: GeoIpConfig,
    pub normalization: Normalization,
    pub health: Health,
    pub cogs: Cogs,
    pub upload: Upload,
}

impl ConfigObject for ConfigValues {
    fn format() -> ConfigFormat {
        ConfigFormat::Yaml
    }

    fn name() -> &'static str {
        "config"
    }
}

#[derive(Default, Clone)]
struct ConfigInner {
    /// Relay's config values.
    values: ConfigValues,
    /// Configured Relay credentials.
    ///
    /// Credentials may be missing for proxy mode.
    credentials: Option<Credentials>,
}

impl ConfigInner {
    fn apply_overrides(&mut self, overrides: &OverridableConfig) -> anyhow::Result<()> {
        if let Some(log_level) = &overrides.log_level {
            self.values.logging.level = log_level.parse()?;
        }

        if let Some(log_format) = &overrides.log_format {
            self.values.logging.format = log_format.parse()?;
        }

        let relay = &mut self.values.relay;
        if let Some(mode) = &overrides.mode {
            relay.mode = mode
                .parse::<RelayMode>()
                .with_context(|| ConfigError::field("mode"))?;
        }
        if let Some(deployment) = &overrides.instance {
            relay.instance = deployment
                .parse::<RelayInstance>()
                .with_context(|| ConfigError::field("deployment"))?;
        }
        if let Some(upstream) = &overrides.upstream {
            relay.upstream = upstream
                .parse::<UpstreamDescriptor>()
                .with_context(|| ConfigError::field("upstream"))?;
        } else if let Some(upstream_dsn) = &overrides.upstream_dsn {
            relay.upstream = upstream_dsn
                .parse::<Dsn>()
                .map(|dsn| UpstreamDescriptor::from_dsn(&dsn))
                .with_context(|| ConfigError::field("upstream_dsn"))?;
        }
        if let Some(host) = &overrides.host {
            relay.host = host
                .parse::<IpAddr>()
                .with_context(|| ConfigError::field("host"))?;
        }
        if let Some(port) = &overrides.port {
            relay.port = port
                .as_str()
                .parse()
                .with_context(|| ConfigError::field("port"))?;
        }

        let processing = &mut self.values.processing;
        if let Some(enabled) = &overrides.processing {
            match enabled.to_lowercase().as_str() {
                "true" | "1" => processing.enabled = true,
                "false" | "0" | "" => processing.enabled = false,
                _ => return Err(ConfigError::field("processing").into()),
            }
        }
        if let Some(redis) = overrides.redis_url.clone() {
            processing.redis = Some(RedisConfigs::Unified(RedisConfig::single(redis)))
        }
        if let Some(kafka_url) = overrides.kafka_url.clone() {
            let existing = processing
                .kafka_config
                .iter_mut()
                .find(|e| e.name == "bootstrap.servers");

            if let Some(config_param) = existing {
                config_param.value = kafka_url;
            } else {
                self.values.processing.kafka_config.push(KafkaConfigParam {
                    name: "bootstrap.servers".to_owned(),
                    value: kafka_url,
                })
            }
        }

        if overrides.outcome_source.is_some() {
            self.values.outcomes.source = overrides.outcome_source.clone();
        }

        if let Some(shutdown_timeout) = &overrides.shutdown_timeout
            && let Ok(shutdown_timeout) = shutdown_timeout.parse::<u64>()
        {
            self.values.limits.shutdown_timeout = shutdown_timeout;
        }

        if let Some(server_name) = overrides.server_name.clone() {
            self.values.sentry.server_name = Some(server_name.into());
        }

        let id = if let Some(id) = &overrides.id {
            let id = Uuid::parse_str(id).with_context(|| ConfigError::field("id"))?;
            Some(id)
        } else {
            None
        };
        let public_key = if let Some(public_key) = &overrides.public_key {
            let public_key = public_key
                .parse::<PublicKey>()
                .with_context(|| ConfigError::field("public_key"))?;
            Some(public_key)
        } else {
            None
        };

        let secret_key = if let Some(secret_key) = &overrides.secret_key {
            let secret_key = secret_key
                .parse::<SecretKey>()
                .with_context(|| ConfigError::field("secret_key"))?;
            Some(secret_key)
        } else {
            None
        };

        if let Some(credentials) = &mut self.credentials {
            //we have existing credentials we may override some entries
            if let Some(id) = id {
                credentials.id = id;
            }
            if let Some(public_key) = public_key {
                credentials.public_key = public_key;
            }
            if let Some(secret_key) = secret_key {
                credentials.secret_key = secret_key
            }
        } else {
            //no existing credentials we may only create the full credentials
            match (id, public_key, secret_key) {
                (Some(id), Some(public_key), Some(secret_key)) => {
                    self.credentials = Some(Credentials {
                        secret_key,
                        public_key,
                        id,
                    })
                }
                (None, None, None) => {
                    // nothing provided, we'll just leave the credentials None, maybe we
                    // don't need them in the current command or we'll override them later
                }
                _ => {
                    return Err(ConfigError::field("incomplete credentials").into());
                }
            }
        }

        Ok(())
    }

    /// Merges reloadable parts of the config from `other` into `self`.
    ///
    /// Returns `true` if any parts of the config were updated.
    fn reload_with(&mut self, other: &Self) -> bool {
        let mut changed = false;

        if self.values.health != other.values.health {
            relay_log::debug!("updating health");
            self.values.health = other.values.health.clone();
            changed = true;
        }

        // Broker settings only. Producers hold their topic name and shard list from startup,
        // so `topics` changes cannot be applied here at all, including moving a topic to
        // another cluster via its `config` key. Those need a restart.
        //
        // Merging `topics` anyway would also break later pushes: `kafka_config_blob` reads it
        // live, so a renamed topic would be keyed by its new name while the producer still
        // selects the old one.
        //
        // Running Kafka clients pick this up via `ConfigReloadService`.
        if self.values.processing.kafka_config != other.values.processing.kafka_config
            || self.values.processing.secondary_kafka_configs
                != other.values.processing.secondary_kafka_configs
        {
            relay_log::debug!("updating kafka config");
            self.values.processing.kafka_config = other.values.processing.kafka_config.clone();
            self.values.processing.secondary_kafka_configs =
                other.values.processing.secondary_kafka_configs.clone();
            changed = true;
        }

        if self.values.spool.envelopes.max_unspool_envelopes_per_second
            != other
                .values
                .spool
                .envelopes
                .max_unspool_envelopes_per_second
        {
            relay_log::debug!("updating spool.envelopes.max_unspool_envelopes_per_second");
            self.values.spool.envelopes.max_unspool_envelopes_per_second = other
                .values
                .spool
                .envelopes
                .max_unspool_envelopes_per_second;
            changed = true;
        }

        changed
    }
}

/// Relay's Configuration.
pub struct Config {
    /// A mutex to serialize all write accesses to `inner`.
    ///
    /// Accessing the arc swap is done with compare and swap, but we still want serialized
    /// access and operations to guarantee consistency when dealing with e.g. the filesystem.
    ///
    /// The mutex must be acquired before changing `inner`. Methods taking `&mut` can omit
    /// this, as the `&mut` requirement already satisfies that there is no concurrent access
    /// possible.
    inner_access: Mutex<()>,
    /// The actual config.
    inner: ArcSwap<ConfigInner>,
    /// A list of overrides applied to the config, in order.
    ///
    /// When re-loading the configuration these overrides need to be applied again
    /// in the same order as they were applied originally.
    overrides: Vec<OverridableConfig>,
    /// Path from which the config is loaded.
    ///
    /// This is Relay's configuration directory.
    path: PathBuf,
}

impl fmt::Debug for Config {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        let inner = self.inner.load();

        f.debug_struct("Config")
            .field("path", &self.path)
            // Only print specific parts of `inner` to not leak the credentials.
            .field("values", &inner.values)
            .finish()
    }
}

impl Config {
    /// Loads a config from a given config folder.
    pub fn from_path<P: AsRef<Path>>(path: P) -> anyhow::Result<Config> {
        let path = env::current_dir()
            .map(|x| x.join(path.as_ref()))
            .unwrap_or_else(|_| path.as_ref().to_path_buf());

        let inner = ConfigInner {
            values: ConfigValues::load(&path)?,
            credentials: match Credentials::path(&path).exists() {
                true => Some(Credentials::load(&path)?),
                false => None,
            },
        };

        let config = Config {
            inner_access: Mutex::new(()),
            inner: ArcSwap::from_pointee(inner),
            overrides: Vec::new(),
            path: path.clone(),
        };

        if cfg!(not(feature = "processing")) && config.current().processing_enabled() {
            return Err(ConfigError::file(ConfigErrorKind::ProcessingNotAvailable, &path).into());
        }

        Ok(config)
    }

    /// Creates a config from a JSON value.
    ///
    /// This is mostly useful for tests.
    pub fn from_json_value(value: serde_json::Value) -> anyhow::Result<Config> {
        Ok(Config {
            inner_access: Mutex::new(()),
            inner: ArcSwap::from_pointee(ConfigInner {
                values: serde_json::from_value(value)
                    .with_context(|| ConfigError::new(ConfigErrorKind::BadJson))?,
                credentials: None,
            }),
            overrides: Vec::new(),
            path: PathBuf::new(),
        })
    }

    /// Override configuration with values coming from other sources (e.g. env variables or
    /// command line parameters).
    ///
    /// If applying the overrides fails, the config may be left in an inconsistent state.
    pub fn apply_override(&mut self, overrides: OverridableConfig) -> anyhow::Result<&mut Self> {
        // We could introduce a config builder which operates on mutable configs, which would eliminate
        // the need for this `try_rcu` dance here.
        crate::utils::try_rcu(&self.inner, |inner| {
            let mut new = ConfigInner::clone(inner);
            new.apply_overrides(&overrides)?;
            Ok::<_, anyhow::Error>(Arc::new(new))
        })?;

        // Overrides successfully applied.
        self.overrides.push(overrides);

        Ok(self)
    }

    /// Checks if the config is already initialized.
    pub fn config_exists<P: AsRef<Path>>(path: P) -> bool {
        fs::metadata(ConfigValues::path(path.as_ref())).is_ok()
    }

    /// Returns the filename of the config file.
    pub fn path(&self) -> &Path {
        &self.path
    }

    /// Dumps out a YAML string of the values.
    pub fn to_yaml_string(&self) -> anyhow::Result<String> {
        serde_yaml::to_string(&self.inner.load().values)
            .with_context(|| ConfigError::new(ConfigErrorKind::CouldNotWriteFile))
    }

    /// Set new credentials.
    ///
    /// This also writes the credentials back to the file, if this config was loaded from the file-system.
    pub fn replace_credentials(
        &mut self,
        credentials: Option<Credentials>,
    ) -> anyhow::Result<bool> {
        if self.inner.load().credentials == credentials {
            return Ok(false);
        }

        if !self.path.is_empty() {
            match &credentials {
                Some(creds) => {
                    creds.save(&self.path)?;
                }
                None => {
                    let path = Credentials::path(&self.path);
                    if fs::metadata(&path).is_ok() {
                        fs::remove_file(&path).with_context(|| {
                            ConfigError::file(ConfigErrorKind::CouldNotWriteFile, &path)
                        })?;
                    }
                }
            }
        }

        // Note: there is never anyone racing on the `ArcSwap` as long as `Self` borrowed mutably.
        //
        // We can improve this if we split out mutable operations into a separate struct and only
        // once `frozen()` we change to an `ArcSwap` internally.
        self.inner.rcu(|inner| {
            let mut inner = ConfigInner::clone(inner);
            inner.credentials = credentials.clone();
            Arc::new(inner)
        });

        Ok(true)
    }

    /// Reloads the configuration from disk.
    ///
    /// Returns `true` if the configuration changed.
    ///
    /// In order for a config to be reloadable it must've been loaded from a path. The original
    /// config will be re-read and reloadable parts of the config will be replaced with their updated
    /// values.
    ///
    /// If the reload fails for any reason, the current config is untouched.
    pub fn reload(&self) -> anyhow::Result<bool> {
        if self.path.is_empty() {
            return Ok(false);
        }

        let _access = self
            .inner_access
            .lock()
            .unwrap_or_else(PoisonError::into_inner);

        let mut new_config = Self::from_path(&self.path)?;
        for overrides in &self.overrides {
            new_config.apply_override(overrides.clone())?;
        }
        let new_config = new_config.current();

        let mut changed = false;

        // Since we do have the `_access` lock, this will always succeed on the first try.
        self.inner.rcu(|inner| {
            let mut new_inner = ConfigInner::clone(inner);
            changed = new_inner.reload_with(&new_config.inner);
            match changed {
                true => Arc::new(new_inner),
                false => Arc::clone(inner),
            }
        });

        Ok(changed)
    }

    /// Acquires a current [`snapshot`](ConfigSnapshot) of the config.
    ///
    /// A snapshot is the way to actually consume values from the config. A snapshot should ideally be acquired
    /// once per unit of work.
    pub fn current(&self) -> ConfigSnapshot {
        let inner = self.inner.load();
        ConfigSnapshot { inner }
    }
}

impl Default for Config {
    fn default() -> Self {
        Self {
            inner_access: Mutex::new(()),
            inner: ArcSwap::from_pointee(Default::default()),
            overrides: Vec::new(),
            path: PathBuf::new(),
        }
    }
}

/// A config snapshot is a point in time snapshot of the [`Config`].
///
/// The [`Config`] may change over time, to guarantee a consistent view of the config
/// a snapshot must be acquired first.
///
/// The snapshot should not be stored in a long lasting datastructure. As a rule of thumb it should
/// only exist on the stack.
pub struct ConfigSnapshot {
    inner: arc_swap::Guard<Arc<ConfigInner>>,
}

impl ConfigSnapshot {
    /// Returns `true` if the config is ready to use.
    pub fn has_credentials(&self) -> bool {
        self.inner.credentials.is_some()
    }

    /// Return the current credentials.
    pub fn credentials(&self) -> Option<&Credentials> {
        self.inner.credentials.as_ref()
    }

    /// Returns the secret key if set.
    pub fn secret_key(&self) -> Option<&SecretKey> {
        self.inner.credentials.as_ref().map(|x| &x.secret_key)
    }

    /// Returns the public key if set.
    pub fn public_key(&self) -> Option<&PublicKey> {
        self.inner.credentials.as_ref().map(|x| &x.public_key)
    }

    /// Returns the relay ID.
    pub fn relay_id(&self) -> Option<&RelayId> {
        self.inner.credentials.as_ref().map(|x| &x.id)
    }

    /// Returns the relay mode.
    pub fn relay_mode(&self) -> RelayMode {
        self.inner.values.relay.mode
    }

    /// Returns the instance type of relay.
    pub fn relay_instance(&self) -> RelayInstance {
        self.inner.values.relay.instance
    }

    /// Returns the upstream target as descriptor.
    pub fn upstream(&self) -> &UpstreamDescriptor {
        &self.inner.values.relay.upstream
    }

    /// Returns the advertised upstream for downstream instances as descriptor.
    pub fn advertised_upstream(&self) -> Option<&UpstreamDescriptor> {
        self.inner.values.relay.advertised_upstream.as_ref()
    }

    /// Returns the custom HTTP "Host" header.
    pub fn http_host_header(&self) -> Option<&str> {
        self.inner.values.http.host_header.as_deref()
    }

    /// Returns the listen address.
    pub fn listen_addr(&self) -> SocketAddr {
        (self.inner.values.relay.host, self.inner.values.relay.port).into()
    }

    /// Returns the listen address for internal APIs.
    ///
    /// Internal APIs are APIs which do not need to be publicly exposed,
    /// like health checks.
    ///
    /// Returns `None` when there is no explicit address configured for internal APIs,
    /// and they should instead be exposed on the main [`Self::listen_addr`].
    pub fn listen_addr_internal(&self) -> Option<SocketAddr> {
        match (
            self.inner.values.relay.internal_host,
            self.inner.values.relay.internal_port,
        ) {
            (Some(host), None) => Some((host, self.inner.values.relay.port).into()),
            (None, Some(port)) => Some((self.inner.values.relay.host, port).into()),
            (Some(host), Some(port)) => Some((host, port).into()),
            (None, None) => None,
        }
    }

    /// Returns `true` when project IDs should be overriden rather than validated.
    ///
    /// Defaults to `false`, which requires project ID validation.
    pub fn override_project_ids(&self) -> bool {
        self.inner.values.relay.override_project_ids
    }

    /// Returns the interval to check for configuration changes.
    ///
    /// `None` if the config should never be reloaded.
    pub fn config_reload_interval(&self) -> Option<Duration> {
        let interval = self.inner.values.relay.config_reload_interval?;
        Some(match interval {
            // Useful for tests to be able to configure this to a very small value,
            // but we also don't want it to be actually 0.
            0 => Duration::from_millis(50),
            secs => Duration::from_secs(secs),
        })
    }

    /// Returns `true` if Relay requires authentication for readiness.
    ///
    /// See [`ReadinessCondition`] for more information.
    pub fn requires_auth(&self) -> bool {
        match self.inner.values.auth.ready {
            ReadinessCondition::Authenticated => self.relay_mode() == RelayMode::Managed,
            ReadinessCondition::Always => false,
        }
    }

    /// Returns the interval at which Realy should try to re-authenticate with the upstream.
    ///
    /// Always disabled in processing mode.
    pub fn http_auth_interval(&self) -> Option<Duration> {
        if self.processing_enabled() {
            return None;
        }

        match self.inner.values.http.auth_interval {
            None | Some(0) => None,
            Some(secs) => Some(Duration::from_secs(secs)),
        }
    }

    /// The maximum time of experiencing uninterrupted network failures until Relay considers that
    /// it has encountered a network outage.
    pub fn http_outage_grace_period(&self) -> Duration {
        Duration::from_secs(self.inner.values.http.outage_grace_period)
    }

    /// Time Relay waits before retrying an upstream request.
    ///
    /// Before going into a network outage, Relay may fail to make upstream
    /// requests. This is the time Relay waits before retrying the same request.
    pub fn http_retry_delay(&self) -> Duration {
        Duration::from_secs(self.inner.values.http.retry_delay)
    }

    /// Time of continued project request failures before Relay emits an error.
    pub fn http_project_failure_interval(&self) -> Duration {
        Duration::from_secs(self.inner.values.http.project_failure_interval)
    }

    /// Content encoding of upstream requests.
    pub fn http_encoding(&self) -> HttpEncoding {
        self.inner.values.http.encoding
    }

    /// Returns whether metrics should be sent globally through a shared endpoint.
    pub fn http_global_metrics(&self) -> bool {
        self.inner.values.http.global_metrics
    }

    /// Returns `true` if Relay supports forwarding unknown API requests.
    ///
    /// Relay instances with processing enabled are expected to support the latest API and do never
    /// support forwarding requests to Sentry.
    pub fn http_forward(&self) -> bool {
        self.inner.values.http.forward && !self.processing_enabled()
    }

    /// Returns whether this Relay should emit outcomes.
    ///
    /// This is `true` either if `outcomes.emit_outcomes` is explicitly enabled, or if this Relay is
    /// in processing mode.
    pub fn emit_outcomes(&self) -> EmitOutcomes {
        if self.processing_enabled() {
            return EmitOutcomes::AsOutcomes;
        }
        self.inner.values.outcomes.emit_outcomes
    }

    /// The originating source of the outcome
    pub fn outcome_source(&self) -> Option<&str> {
        self.inner.values.outcomes.source.as_deref()
    }

    /// Returns logging configuration.
    pub fn logging(&self) -> &relay_log::LogConfig {
        &self.inner.values.logging
    }

    /// Returns logging configuration.
    pub fn sentry(&self) -> &relay_log::SentryConfig {
        &self.inner.values.sentry
    }

    /// Returns the addresses for statsd metrics.
    pub fn statsd_addr(&self) -> Option<&str> {
        self.inner.values.metrics.statsd.as_deref()
    }

    /// Returns the addresses for statsd metrics.
    pub fn statsd_buffer_size(&self) -> Option<usize> {
        self.inner.values.metrics.statsd_buffer_size
    }

    /// Return the prefix for statsd metrics.
    pub fn metrics_prefix(&self) -> &str {
        &self.inner.values.metrics.prefix
    }

    /// Returns the default tags for statsd metrics.
    pub fn metrics_default_tags(&self) -> &BTreeMap<String, String> {
        &self.inner.values.metrics.default_tags
    }

    /// Returns the name of the hostname tag that should be attached to each outgoing metric.
    pub fn metrics_hostname_tag(&self) -> Option<&str> {
        self.inner.values.metrics.hostname_tag.as_deref()
    }

    /// Returns the interval for periodic metrics emitted from Relay.
    ///
    /// `None` if periodic metrics are disabled.
    pub fn metrics_periodic_interval(&self) -> Option<Duration> {
        match self.inner.values.metrics.periodic_secs {
            0 => None,
            secs => Some(Duration::from_secs(secs)),
        }
    }

    /// Returns the default timeout for all upstream HTTP requests.
    pub fn http_timeout(&self) -> Duration {
        Duration::from_secs(self.inner.values.http.timeout.into())
    }

    /// Returns the connection timeout for all upstream HTTP requests.
    pub fn http_connection_timeout(&self) -> Duration {
        Duration::from_secs(self.inner.values.http.connection_timeout.into())
    }

    /// Returns the failed upstream request retry interval.
    pub fn http_max_retry_interval(&self) -> Duration {
        Duration::from_secs(self.inner.values.http.max_retry_interval.into())
    }

    /// Returns `true` if relay should use an in-process cache for DNS lookups.
    pub fn http_dns_cache(&self) -> bool {
        self.inner.values.http.dns_cache
    }

    /// Returns the expiry timeout for cached projects.
    pub fn project_cache_expiry(&self) -> Duration {
        Duration::from_secs(self.inner.values.cache.project_expiry.into())
    }

    /// Returns `true` if the full project state should be requested from upstream.
    pub fn request_full_project_config(&self) -> bool {
        self.inner.values.cache.project_request_full_config
    }

    /// Returns the expiry timeout for cached relay infos (public keys).
    pub fn relay_cache_expiry(&self) -> Duration {
        Duration::from_secs(self.inner.values.cache.relay_expiry.into())
    }

    /// Returns the expiry timeout for cached misses before trying to refetch.
    pub fn cache_miss_expiry(&self) -> Duration {
        Duration::from_secs(self.inner.values.cache.miss_expiry.into())
    }

    /// Returns the grace period for project caches.
    pub fn project_grace_period(&self) -> Duration {
        Duration::from_secs(self.inner.values.cache.project_grace_period.into())
    }

    /// Returns the refresh interval for a project.
    ///
    /// Validates the refresh time to be between the grace period and expiry.
    pub fn project_refresh_interval(&self) -> Option<Duration> {
        self.inner
            .values
            .cache
            .project_refresh_interval
            .map(Into::into)
            .map(Duration::from_secs)
    }

    /// Returns the duration in which batchable project config queries are
    /// collected before sending them in a single request.
    pub fn query_batch_interval(&self) -> Duration {
        Duration::from_millis(self.inner.values.cache.batch_interval.into())
    }

    /// Returns the duration in which downstream relays are requested from upstream.
    pub fn downstream_relays_batch_interval(&self) -> Duration {
        Duration::from_millis(
            self.inner
                .values
                .cache
                .downstream_relays_batch_interval
                .into(),
        )
    }

    /// Returns the interval in seconds in which local project configurations should be reloaded.
    pub fn local_cache_interval(&self) -> Duration {
        Duration::from_secs(self.inner.values.cache.file_interval.into())
    }

    /// Returns the interval in seconds in which fresh global configs should be
    /// fetched from  upstream.
    pub fn global_config_fetch_interval(&self) -> Duration {
        Duration::from_secs(self.inner.values.cache.global_config_fetch_interval.into())
    }

    /// Returns the path of the buffer file if the `cache.persistent_envelope_buffer.path` is configured.
    ///
    /// In case a partition with id > 0 is supplied, the filename of the envelopes path will be
    /// suffixed with `.{partition_id}`.
    pub fn spool_envelopes_path(&self, partition_id: u8) -> Option<PathBuf> {
        let mut path = self
            .inner
            .values
            .spool
            .envelopes
            .path
            .as_ref()
            .map(|path| path.to_owned())?;

        if partition_id == 0 {
            return Some(path);
        }

        let file_name = path.file_name().and_then(|f| f.to_str())?;
        let new_file_name = format!("{file_name}.{partition_id}");
        path.set_file_name(new_file_name);

        Some(path)
    }

    /// The maximum size of the buffer, in bytes.
    pub fn spool_envelopes_max_disk_size(&self) -> usize {
        self.inner.values.spool.envelopes.max_disk_size.as_bytes()
    }

    /// Number of encoded envelope bytes that need to be accumulated before
    /// flushing one batch to disk.
    pub fn spool_envelopes_batch_size_bytes(&self) -> usize {
        self.inner
            .values
            .spool
            .envelopes
            .batch_size_bytes
            .as_bytes()
    }

    /// Time after which a batch of envelopes is flushed to disk, regardless of its size.
    pub fn spool_envelopes_flush_timeout(&self) -> Option<Duration> {
        self.inner
            .values
            .spool
            .envelopes
            .flush_timeout_secs
            .map(Duration::from_secs)
    }

    /// Maximum number of envelopes unspooled from disk per second.
    pub fn spool_max_unspool_envelopes_per_second(&self) -> Option<NonZeroU32> {
        self.inner
            .values
            .spool
            .envelopes
            .max_unspool_envelopes_per_second
    }

    /// Returns the time after which we drop envelopes as a [`Duration`] object.
    pub fn spool_envelopes_max_age(&self) -> Duration {
        Duration::from_secs(self.inner.values.spool.envelopes.max_envelope_delay_secs)
    }

    /// Returns the refresh frequency for disk usage monitoring as a [`Duration`] object.
    pub fn spool_disk_usage_refresh_frequency_ms(&self) -> Duration {
        Duration::from_millis(
            self.inner
                .values
                .spool
                .envelopes
                .disk_usage_refresh_frequency_ms,
        )
    }

    /// Returns the relative memory usage up to which the disk buffer will unspool envelopes.
    pub fn spool_max_backpressure_memory_percent(&self) -> f32 {
        self.inner
            .values
            .spool
            .envelopes
            .max_backpressure_memory_percent
    }

    /// Returns the number of partitions for the buffer.
    pub fn spool_partitions(&self) -> NonZeroU8 {
        self.inner.values.spool.envelopes.partitions
    }

    /// Returns the strategy used to assign envelopes to buffer partitions.
    pub fn spool_partitioning(&self) -> EnvelopeSpoolPartitioning {
        self.inner.values.spool.envelopes.partitioning
    }

    /// Returns `true` if the data is stored on ephemeral disks.
    pub fn spool_ephemeral(&self) -> bool {
        self.inner.values.spool.envelopes.ephemeral
    }

    /// Returns the maximum size of an event payload in bytes.
    pub fn max_event_size(&self) -> usize {
        self.inner.values.limits.max_event_size.as_bytes()
    }

    /// Returns the maximum size of each attachment.
    pub fn max_attachment_size(&self) -> usize {
        self.inner.values.limits.max_attachment_size.as_bytes()
    }

    /// The maximum amount of attachments in a single envelope.
    pub fn max_attachment_count(&self) -> usize {
        self.inner.values.limits.max_attachment_count
    }

    /// Returns the maximum combined size of attachments or payloads containing attachments
    /// (minidump, unreal, standalone attachments) in bytes.
    pub fn max_attachments_size(&self) -> usize {
        self.inner.values.limits.max_attachments_size.as_bytes()
    }

    /// Returns the maximum size of a TUS upload request body.
    pub fn max_upload_size(&self) -> usize {
        self.inner.values.limits.max_upload_size.as_bytes()
    }

    /// Returns the maximum number of client reports per envelope.
    pub fn max_client_reports_count(&self) -> usize {
        self.inner.values.limits.max_client_reports_count
    }

    /// Returns the maximum combined size of client reports in bytes.
    pub fn max_client_reports_size(&self) -> usize {
        self.inner.values.limits.max_client_reports_size.as_bytes()
    }

    /// Returns the maximum payload size of a monitor check-in in bytes.
    pub fn max_check_in_size(&self) -> usize {
        self.inner.values.limits.max_check_in_size.as_bytes()
    }

    /// Returns the maximum payload size of a log in bytes.
    pub fn max_log_size(&self) -> usize {
        self.inner.values.limits.max_log_size.as_bytes()
    }

    /// Returns the maximum number of operations to allow for a log expansion.
    pub fn max_expanded_log_operations(&self) -> usize {
        self.inner.values.limits.max_expanded_log_operations
    }

    /// Returns the maximum number of operations to allow for a span expansion.
    pub fn max_expanded_span_operations(&self) -> usize {
        self.inner.values.limits.max_expanded_span_operations
    }

    /// Returns the maximum number of operations to allow for a span expansion.
    pub fn max_expanded_trace_metric_operations(&self) -> usize {
        self.inner
            .values
            .limits
            .max_expanded_trace_metric_operations
    }

    /// Returns the maximum payload size of a span in bytes.
    pub fn max_span_size(&self) -> usize {
        self.inner.values.limits.max_span_size.as_bytes()
    }

    /// Returns the maximum amount of standalone transaction spans per envelope.
    pub fn max_standalone_span_count(&self) -> usize {
        self.inner.values.limits.max_standalone_span_count
    }

    /// Returns the maximum payload size of an item container in bytes.
    pub fn max_container_size(&self) -> usize {
        self.inner.values.limits.max_container_size.as_bytes()
    }

    /// Returns the maximum size of an envelope payload in bytes.
    ///
    /// Individual item size limits still apply.
    pub fn max_envelope_size(&self) -> usize {
        self.inner.values.limits.max_envelope_size.as_bytes()
    }

    /// Returns the maximum number of sessions per envelope.
    pub fn max_session_count(&self) -> usize {
        self.inner.values.limits.max_session_count
    }

    /// Returns the maximum combined size for all sessions in an envelope in bytes.
    pub fn max_sessions_size(&self) -> usize {
        self.inner.values.limits.max_sessions_size.as_bytes()
    }

    /// Returns the maximum payload size of metric buckets in bytes.
    pub fn max_metric_buckets_size(&self) -> usize {
        self.inner.values.limits.max_metric_buckets_size.as_bytes()
    }

    /// Returns the maximum payload size for general API requests.
    pub fn max_api_payload_size(&self) -> usize {
        self.inner.values.limits.max_api_payload_size.as_bytes()
    }

    /// Returns the maximum payload size for file uploads and chunks.
    pub fn max_api_file_upload_size(&self) -> usize {
        self.inner.values.limits.max_api_file_upload_size.as_bytes()
    }

    /// Returns the maximum payload size for chunks
    pub fn max_api_chunk_upload_size(&self) -> usize {
        self.inner
            .values
            .limits
            .max_api_chunk_upload_size
            .as_bytes()
    }

    /// Returns the maximum payload size for a profile
    pub fn max_profile_size(&self) -> usize {
        self.inner.values.limits.max_profile_size.as_bytes()
    }

    /// Returns the maximum payload size for a trace metric.
    pub fn max_trace_metric_size(&self) -> usize {
        self.inner.values.limits.max_trace_metric_size.as_bytes()
    }

    /// Returns the maximum payload size for a compressed replay.
    pub fn max_replay_compressed_size(&self) -> usize {
        self.inner
            .values
            .limits
            .max_replay_compressed_size
            .as_bytes()
    }

    /// Returns the maximum payload size for an uncompressed replay.
    pub fn max_replay_uncompressed_size(&self) -> usize {
        self.inner
            .values
            .limits
            .max_replay_uncompressed_size
            .as_bytes()
    }

    /// Returns the maximum message size for an uncompressed replay.
    ///
    /// This is greater than max_replay_compressed_size because
    /// it can include additional metadata about the replay in
    /// addition to the recording.
    pub fn max_replay_message_size(&self) -> usize {
        self.inner.values.limits.max_replay_message_size.as_bytes()
    }

    /// Returns the maximum number of active requests
    pub fn max_concurrent_requests(&self) -> usize {
        self.inner.values.limits.max_concurrent_requests
    }

    /// Returns the maximum number of active queries
    pub fn max_concurrent_queries(&self) -> usize {
        self.inner.values.limits.max_concurrent_queries
    }

    /// Returns the maximum combined size of keys of invalid attributes.
    pub fn max_removed_attribute_key_size(&self) -> usize {
        self.inner
            .values
            .limits
            .max_removed_attribute_key_size
            .as_bytes()
    }

    /// The maximum number of seconds a query is allowed to take across retries.
    pub fn query_timeout(&self) -> Duration {
        Duration::from_secs(self.inner.values.limits.query_timeout)
    }

    /// The maximum number of seconds to wait for pending envelopes after receiving a shutdown
    /// signal.
    pub fn shutdown_timeout(&self) -> Duration {
        Duration::from_secs(self.inner.values.limits.shutdown_timeout)
    }

    /// Returns the server keep-alive timeout in seconds.
    ///
    /// By default keep alive is set to a 5 seconds.
    pub fn keepalive_timeout(&self) -> Duration {
        Duration::from_secs(self.inner.values.limits.keepalive_timeout)
    }

    /// Returns the server idle timeout in seconds.
    pub fn idle_timeout(&self) -> Option<Duration> {
        self.inner
            .values
            .limits
            .idle_timeout
            .map(Duration::from_secs)
    }

    /// Returns the maximum connections.
    pub fn max_connections(&self) -> Option<usize> {
        self.inner.values.limits.max_connections
    }

    /// TCP listen backlog to configure on Relay's listening socket.
    pub fn tcp_listen_backlog(&self) -> u32 {
        self.inner.values.limits.tcp_listen_backlog
    }

    /// Returns the number of cores to use for thread pools.
    pub fn cpu_concurrency(&self) -> usize {
        self.inner.values.limits.max_thread_count
    }

    /// Returns the number of tasks that can run concurrently in the worker pool.
    pub fn pool_concurrency(&self) -> usize {
        self.inner.values.limits.max_pool_concurrency
    }

    /// Returns the maximum size of a project config query.
    pub fn query_batch_size(&self) -> usize {
        self.inner.values.cache.batch_size
    }

    /// True if the Relay should do processing.
    pub fn processing_enabled(&self) -> bool {
        self.inner.values.processing.enabled
    }

    /// Level of normalization for Relay to apply to incoming data.
    pub fn normalization_level(&self) -> NormalizationLevel {
        self.inner.values.normalization.level
    }

    /// The path to the GeoIp database required for event processing.
    pub fn geoip_path(&self) -> Option<&Path> {
        self.inner.values.geoip.path.as_deref().or(self
            .inner
            .values
            .processing
            .geoip_path
            .as_deref())
    }

    /// Maximum future timestamp of ingested data.
    ///
    /// Events past this timestamp will be adjusted to `now()`. Sessions will be dropped.
    pub fn max_secs_in_future(&self) -> i64 {
        self.inner.values.processing.max_secs_in_future.into()
    }

    /// Maximum age of ingested sessions. Older sessions will be dropped.
    pub fn max_session_secs_in_past(&self) -> i64 {
        self.inner.values.processing.max_session_secs_in_past.into()
    }

    /// Configuration name and list of Kafka configuration parameters for a given topic.
    pub fn kafka_configs(
        &self,
        topic: KafkaTopic,
    ) -> Result<KafkaTopicConfig<'_>, KafkaConfigError> {
        self.inner
            .values
            .processing
            .topics
            .get(topic)
            .kafka_configs(
                &self.inner.values.processing.kafka_config,
                &self.inner.values.processing.secondary_kafka_configs,
            )
    }

    /// Whether to validate the topics against Kafka.
    pub fn kafka_validate_topics(&self) -> bool {
        self.inner.values.processing.kafka_validate_topics
    }

    /// Upper bound on the random delay before a Kafka producer swaps its client.
    pub fn reloading_producer_jitter(&self) -> Duration {
        Duration::from_secs(self.inner.values.processing.reloading_producer_jitter_secs)
    }

    /// Builds the Kafka config blob describing every configured topic.
    ///
    /// # This is a stand-in
    ///
    /// The blob is meant to be opaque to Relay: one is published per region covering every
    /// cluster and topic, and Relay passes the bytes to Arroyo without interpreting them. That
    /// delivery channel does not yet exist, so this builds a blob out of Relay's own
    /// `kafka_config`, `secondary_kafka_configs` and `topics` instead.
    pub fn kafka_config_blob(&self) -> Result<Vec<u8>, KafkaConfigError> {
        // Sorted throughout, so an unchanged config always serializes to the same bytes and the
        // caller can skip pushing by comparing them.
        let mut clusters = BTreeMap::new();
        let mut topics = BTreeMap::new();

        // A topic without a `config` uses the default `kafka_config`. The blob has no notion of
        // a default, so that becomes a cluster under a name derived to not collide with any
        // `secondary_kafka_configs` entry the user may have written.
        let default_cluster = default_cluster_name(&self.inner.values.processing);

        for topic in KafkaTopic::iter() {
            let topic_config = self.kafka_configs(*topic)?;

            for params in topic_config.topics() {
                let cluster_name = params.config_name.unwrap_or(&default_cluster);

                let config: BTreeMap<_, _> = params
                    .params
                    .iter()
                    .map(|param| (param.name.as_str(), param.value.as_str()))
                    .collect();

                clusters.insert(
                    cluster_name.to_owned(),
                    serde_json::json!({ "config": config }),
                );

                // Arroyo keys the blob by logical topic and maps it to one physical topic. Relay's
                // sharded topics are one logical topic to many physical ones, which Arroyo cannot
                // express yet, so each physical topic is its own entry and maps to itself.
                //
                // That also means one topic name cannot point at two clusters.
                let entry = serde_json::json!({ "physical_topics": [{ "cluster": cluster_name }] });
                match topics.entry(params.topic_name.clone()) {
                    Entry::Vacant(vacant) => {
                        vacant.insert(entry);
                    }
                    Entry::Occupied(occupied) if *occupied.get() != entry => {
                        return Err(KafkaConfigError::AmbiguousTopic(params.topic_name.clone()));
                    }
                    Entry::Occupied(_) => {}
                }
            }
        }

        // Serializing maps of strings cannot fail.
        Ok(serde_json::to_vec(&serde_json::json!({
            "clusters": clusters,
            "topics": topics,
        }))
        .expect("kafka config blob is serializable"))
    }

    /// All unused but configured topic assignments.
    pub fn unused_topic_assignments(&self) -> &relay_kafka::Unused {
        &self.inner.values.processing.topics.unused
    }

    /// Configuration of the objectstore service.
    pub fn objectstore(&self) -> &ObjectstoreServiceConfig {
        &self.inner.values.processing.objectstore
    }

    /// Configuration of the upload service.
    pub fn upload(&self) -> &Upload {
        &self.inner.values.upload
    }

    /// Returns the key used to sign upload locations.
    #[cfg(feature = "processing")]
    pub fn upload_signing_key(&self) -> Option<&SecretKey> {
        self.upload()
            .credentials
            .as_ref()
            .map(|c| &c.signing_key)
            .or(self.credentials().map(|c| &c.secret_key))
    }

    /// Returns the key used to verify upload locations.
    #[cfg(feature = "processing")]
    pub fn upload_verification_key(&self) -> Option<&PublicKey> {
        self.upload()
            .credentials
            .as_ref()
            .map(|c| &c.verification_key)
            .or(self.credentials().map(|c| &c.public_key))
    }

    /// Redis servers to connect to for project configs, rate limiting, and metrics metadata.
    pub fn redis(&self) -> Option<RedisConfigsRef<'_>> {
        let redis_configs = self.inner.values.processing.redis.as_ref()?;

        Some(build_redis_configs(
            redis_configs,
            self.cpu_concurrency() as u32,
            self.pool_concurrency() as u32,
        ))
    }

    /// Chunk size of attachments in bytes.
    pub fn attachment_chunk_size(&self) -> usize {
        self.inner
            .values
            .processing
            .attachment_chunk_size
            .as_bytes()
    }

    /// Maximum metrics batch size in bytes.
    pub fn metrics_max_batch_size_bytes(&self) -> usize {
        self.inner.values.aggregator.max_flush_bytes
    }

    /// Default prefix to use when looking up project configs in Redis. This is only done when
    /// Relay is in processing mode.
    pub fn projectconfig_cache_prefix(&self) -> &str {
        &self.inner.values.processing.projectconfig_cache_prefix
    }

    /// Maximum rate limit to report to clients in seconds.
    pub fn max_rate_limit(&self) -> Option<u64> {
        self.inner.values.processing.max_rate_limit.map(u32::into)
    }

    /// Amount of remaining quota which is cached in memory.
    pub fn quota_cache_ratio(&self) -> Option<f32> {
        self.inner.values.processing.quota_cache_ratio
    }

    /// Maximum limit (ratio) for the in memory quota cache.
    pub fn quota_cache_max(&self) -> Option<f32> {
        self.inner.values.processing.quota_cache_max
    }

    /// Interval to refresh internal health checks.
    pub fn health_refresh_interval(&self) -> Duration {
        Duration::from_millis(self.inner.values.health.refresh_interval_ms)
    }

    /// Maximum memory watermark in bytes.
    pub fn health_max_memory_watermark_bytes(&self) -> u64 {
        self.inner
            .values
            .health
            .max_memory_bytes
            .as_ref()
            .map_or(u64::MAX, |b| b.as_bytes() as u64)
    }

    /// Maximum memory watermark as a percentage of maximum system memory.
    pub fn health_max_memory_watermark_percent(&self) -> f32 {
        self.inner.values.health.max_memory_percent
    }

    /// Health check probe timeout.
    pub fn health_probe_timeout(&self) -> Duration {
        Duration::from_millis(self.inner.values.health.probe_timeout_ms)
    }

    /// Refresh frequency for polling new memory stats.
    pub fn memory_stat_refresh_frequency_ms(&self) -> u64 {
        self.inner.values.health.memory_stat_refresh_frequency_ms
    }

    /// Maximum amount of COGS measurements buffered in memory.
    pub fn cogs_max_queue_size(&self) -> u64 {
        self.inner.values.cogs.max_queue_size
    }

    /// Resource ID to use for Relay COGS measurements.
    pub fn cogs_relay_resource_id(&self) -> &str {
        &self.inner.values.cogs.relay_resource_id
    }

    /// Returns configuration for the default metrics aggregator.
    pub fn default_aggregator_config(&self) -> &AggregatorServiceConfig {
        &self.inner.values.aggregator
    }

    /// Returns configuration for non-default metrics aggregator.
    pub fn secondary_aggregator_configs(&self) -> &Vec<ScopedAggregatorConfig> {
        &self.inner.values.secondary_aggregators
    }

    /// Returns aggregator config for a given metrics namespace.
    pub fn aggregator_config_for(&self, namespace: MetricNamespace) -> &AggregatorServiceConfig {
        for entry in &self.inner.values.secondary_aggregators {
            if entry.condition.matches(Some(namespace)) {
                return &entry.config;
            }
        }
        &self.inner.values.aggregator
    }

    /// Return the statically configured Relays.
    pub fn static_relays(&self) -> &HashMap<RelayId, RelayInfo> {
        &self.inner.values.auth.static_relays
    }

    /// Returns the max age a signature is considered valid, in seconds.
    pub fn signature_max_age(&self) -> Duration {
        Duration::from_secs(self.inner.values.auth.signature_max_age)
    }

    /// Returns `true` if unknown items should be accepted and forwarded.
    pub fn accept_unknown_items(&self) -> bool {
        let forward = self.inner.values.routing.accept_unknown_items;
        forward.unwrap_or_else(|| !self.processing_enabled())
    }
}

impl fmt::Debug for ConfigSnapshot {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("ConfigSnapshot")
            .field("values", &self.inner.values)
            .finish()
    }
}

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

    #[test]
    fn test_http_encoding_parse_strict() {
        for (value, expected) in [
            ("identity", Some(HttpEncoding::Identity)),
            ("gzip", Some(HttpEncoding::Gzip)),
            ("deflate", Some(HttpEncoding::Deflate)),
            ("br", Some(HttpEncoding::Br)),
            ("zstd", Some(HttpEncoding::Zstd)),
            ("", None),
            ("unknown", None),
            ("x-gzip", None),
            ("GZIP", None),
            (" gzip ", None),
            ("zstd, gzip", None),
        ] {
            assert_eq!(HttpEncoding::parse_strict(value), expected, "{value:?}");
        }
    }

    #[cfg(feature = "processing")]
    #[test]
    fn test_upload_secret_key_from_file() {
        let path = env::temp_dir().join(Uuid::new_v4().to_string());
        fs::create_dir(&path).unwrap();
        fs::write(
            path.join("my_secret.txt"),
            "U3LSQM5NorvgnoYHW_aZpc_43nuuh3lhs3zjjcBwaks",
        )
        .unwrap();
        fs::write(
            ConfigValues::path(&path),
            r#"
    upload:
        credentials:
            signing_key: ${file:my_secret.txt}
            verification_key: "VNS8haF0VTnuMMDR2t-f7AgnmUcXmcdzV3SVksSk34s""#,
        )
        .unwrap();

        let config = Config::from_path(&path).unwrap().current();

        fs::remove_dir_all(path).unwrap();

        let signing_key = &config.upload().credentials.as_ref().unwrap().signing_key;
        assert_eq!(
            signing_key.to_string(),
            "U3LSQM5NorvgnoYHW_aZpc_43nuuh3lhs3zjjcBwaks"
        );
    }

    #[test]
    fn test_emit_outcomes() {
        for (serialized, deserialized) in &[
            ("true", EmitOutcomes::AsOutcomes),
            ("false", EmitOutcomes::None),
            ("\"as_client_reports\"", EmitOutcomes::AsClientReports),
        ] {
            let value: EmitOutcomes = serde_json::from_str(serialized).unwrap();
            assert_eq!(value, *deserialized);
            assert_eq!(serde_json::to_string(&value).unwrap(), *serialized);
        }
    }

    #[test]
    fn test_emit_outcomes_invalid() {
        assert!(serde_json::from_str::<EmitOutcomes>("asdf").is_err());
    }

    /// Builds a config from YAML under `processing`.
    #[cfg(feature = "processing")]
    fn processing_config(yaml: &str) -> Config {
        let path = env::temp_dir().join(Uuid::new_v4().to_string());
        fs::create_dir(&path).unwrap();
        fs::write(ConfigValues::path(&path), yaml).unwrap();
        Config::from_path(&path).unwrap()
    }

    /// A generated blob must resolve to the same broker params Relay would have built directly.
    #[cfg(feature = "processing")]
    #[test]
    fn test_kafka_config_blob_round_trips() {
        use sentry_arroyo::backends::kafka::config_blob::{ConfigBlob, ProducerSelector};

        // Covers sharding, a secondary cluster, and a secondary config named `default`.
        let config = processing_config(
            r#"
processing:
  enabled: true
  kafka_config:
    - {name: "bootstrap.servers", value: "primary:9092"}
    - {name: "compression.type", value: "lz4"}
  secondary_kafka_configs:
    default:
      - {name: "bootstrap.servers", value: "impostor:9092"}
  topics:
    ingest-events: ["events-1", "events-2"]
    ingest-metrics:
      name: "metrics-topic"
      config: "default"
"#,
        );
        let config = config.current();

        let blob = config.kafka_config_blob().unwrap();
        let parsed = ConfigBlob::parse(&blob).unwrap();

        let resolve = |topic: &str| {
            let resolved = parsed
                .producer_config(&ProducerSelector::new(topic, "relay"))
                .unwrap_or_else(|_| panic!("{topic} is missing from the blob"));

            // Each entry must produce to the topic it is keyed by.
            assert_eq!(resolved.physical_topic.as_str(), topic);

            resolved
                .config
                .config_map()
                .iter()
                // Arroyo adds these itself when building a producer config.
                .filter(|(key, _)| !matches!(key.as_str(), "statistics.interval.ms" | "client.id"))
                .map(|(key, value)| (key.clone(), value.clone()))
                .collect::<BTreeMap<_, _>>()
        };

        for (topic, physical) in [
            (KafkaTopic::Events, ["events-1", "events-2"].as_slice()),
            (KafkaTopic::MetricsSessions, ["metrics-topic"].as_slice()),
        ] {
            let params = config.kafka_configs(topic).unwrap();

            for (shard, physical) in params.topics().iter().zip(physical) {
                let expected: BTreeMap<_, _> = shard
                    .params
                    .iter()
                    .map(|param| (param.name.clone(), param.value.clone()))
                    .collect();

                assert_eq!(
                    resolve(physical),
                    expected,
                    "{physical} resolved differently"
                );
            }
        }

        // The secondary config named `default` must not have swallowed the anonymous one.
        assert_eq!(
            resolve("events-1").get("bootstrap.servers").unwrap(),
            "primary:9092"
        );
        assert_eq!(
            resolve("metrics-topic").get("bootstrap.servers").unwrap(),
            "impostor:9092"
        );
    }

    /// Moving a topic to a different cluster must change the broker it resolves to.
    #[cfg(feature = "processing")]
    #[test]
    fn test_topic_can_move_between_clusters() {
        use sentry_arroyo::backends::kafka::config_blob::{ConfigBlob, ProducerSelector};

        let resolve = |yaml: &str| {
            let config = processing_config(yaml);
            let blob = config.current().kafka_config_blob().unwrap();
            ConfigBlob::parse(&blob)
                .unwrap()
                .producer_config(&ProducerSelector::new("events-topic", "relay"))
                .unwrap()
                .config
                .get_config_value("bootstrap.servers")
                .unwrap()
                .clone()
        };

        let before = resolve(
            r#"
processing:
  enabled: true
  kafka_config:
    - {name: "bootstrap.servers", value: "primary:9092"}
  secondary_kafka_configs:
    other:
      - {name: "bootstrap.servers", value: "other:9092"}
  topics:
    ingest-events: "events-topic"
"#,
        );

        // Same physical topic, now pointed at the secondary cluster.
        let after = resolve(
            r#"
processing:
  enabled: true
  kafka_config:
    - {name: "bootstrap.servers", value: "primary:9092"}
  secondary_kafka_configs:
    other:
      - {name: "bootstrap.servers", value: "other:9092"}
  topics:
    ingest-events:
      name: "events-topic"
      config: "other"
"#,
        );

        assert_eq!(before, "primary:9092");
        assert_eq!(after, "other:9092");
    }

    /// The blob is keyed by topic name, so one name on two clusters cannot be expressed. That
    /// must fail loudly instead of sending one of the two to the wrong cluster.
    #[cfg(feature = "processing")]
    #[test]
    fn test_kafka_config_blob_rejects_topic_on_two_clusters() {
        let config = processing_config(
            r#"
processing:
  enabled: true
  kafka_config:
    - {name: "bootstrap.servers", value: "primary:9092"}
  secondary_kafka_configs:
    other:
      - {name: "bootstrap.servers", value: "other:9092"}
  topics:
    outcomes:
      name: "outcomes-topic"
      config: "other"
    outcomes-billing: "outcomes-topic"
"#,
        );

        assert!(matches!(
            config.current().kafka_config_blob(),
            Err(KafkaConfigError::AmbiguousTopic(topic)) if topic == "outcomes-topic"
        ));
    }

    /// Broker settings reload; topic assignments do not, since producers hold theirs from
    /// startup.
    #[test]
    fn test_reload_takes_kafka_config_but_not_topics() {
        let mut current = ConfigValues::default();
        current.processing.kafka_config = vec![KafkaConfigParam {
            name: "bootstrap.servers".to_owned(),
            value: "old:9092".to_owned(),
        }];
        current.processing.topics.events = "old-topic".to_owned().into();

        let mut other = ConfigValues::default();
        other.processing.kafka_config = vec![KafkaConfigParam {
            name: "bootstrap.servers".to_owned(),
            value: "new:9092".to_owned(),
        }];
        other.processing.topics.events = "new-topic".to_owned().into();

        let mut current = ConfigInner {
            values: current,
            ..Default::default()
        };
        let other = ConfigInner {
            values: other,
            ..Default::default()
        };

        assert!(current.reload_with(&other));
        assert_eq!(current.values.processing.kafka_config[0].value, "new:9092");

        let assignment = format!("{:?}", current.values.processing.topics.events);
        assert!(
            assignment.contains("old-topic"),
            "topic assignment was reloaded, but the producers cannot pick it up: {assignment}"
        );
    }
}