road-runner-common 0.21.0

//! The shipped [`AudienceResolver`]: cex-account's internal
//! `AccountInternalService.GetNotificationTarget` (feature `notification-grpc`).
//!
//! One lazily-connected channel per process with explicit connect/request timeouts;
//! every call carries the shared `x-internal-secret` as gRPC metadata. This exists so
//! that adopting notifications in a service is a wiring change, not another copy of the
//! same client.

use async_trait::async_trait;
use tonic::metadata::{Ascii, MetadataValue};
use tonic::transport::{Channel, Endpoint};
use tonic::{Code, Request};

use super::audience::{
    normalize_locale, Audience, AudienceError, AudienceResolver, RecipientRef, DEFAULT_LOCALE,
};

pub mod account_proto {
    tonic::include_proto!("exchange.cex.account");
}

use account_proto::account_internal_service_client::AccountInternalServiceClient;
use account_proto::get_notification_target_request::By;
use account_proto::GetNotificationTargetRequest;

/// Environment variable naming cex-account's internal gRPC endpoint. The same key in
/// every service that talks to it (cex-auth, cex-history, cex-ledger).
pub const ACCOUNT_GRPC_URL_ENV: &str = "ACCOUNT_INTERNAL_GRPC_URL";

/// Environment variable carrying the shared internal secret presented as
/// `x-internal-secret`.
pub const INTERNAL_SECRET_ENV: &str = "APIKEY_INTERNAL_SECRET";

const CONNECT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(3);
const REQUEST_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);

/// The resolver could not be built at all — a deployment error, not a runtime one.
#[derive(Debug, thiserror::Error)]
pub enum ResolverInitError {
    #[error("invalid cex-account gRPC url '{url}': {source}")]
    Url {
        url: String,
        #[source]
        source: tonic::transport::Error,
    },
    #[error("internal secret is not valid ASCII metadata")]
    Secret,
}

/// Resolves notification audiences against cex-account.
#[derive(Clone)]
pub struct AccountAudienceResolver {
    client: AccountInternalServiceClient<Channel>,
    secret: MetadataValue<Ascii>,
    fallback_locale: String,
}

impl AccountAudienceResolver {
    /// `url` is the value of e.g. `ACCOUNT_INTERNAL_GRPC_URL`, `secret` of
    /// `APIKEY_INTERNAL_SECRET`. The channel connects lazily, so cex-account does not
    /// need to be up at startup.
    pub fn new(url: &str, secret: &str) -> Result<Self, ResolverInitError> {
        let target = normalize_url(url);
        let channel = Endpoint::from_shared(target.clone())
            .map_err(|source| ResolverInitError::Url {
                url: target,
                source,
            })?
            .timeout(REQUEST_TIMEOUT)
            .connect_timeout(CONNECT_TIMEOUT)
            .connect_lazy();
        let secret: MetadataValue<Ascii> =
            secret.parse().map_err(|_| ResolverInitError::Secret)?;

        Ok(Self {
            client: AccountInternalServiceClient::new(channel),
            secret,
            fallback_locale: DEFAULT_LOCALE.to_string(),
        })
    }

    /// Override the language used when a subject's own language cannot be read.
    /// Defaults to [`DEFAULT_LOCALE`].
    pub fn with_fallback_locale(mut self, locale: impl Into<String>) -> Self {
        self.fallback_locale = locale.into();
        self
    }

    async fn lookup(&self, by: By) -> Result<Audience, tonic::Status> {
        let mut request = Request::new(GetNotificationTargetRequest { by: Some(by) });
        request
            .metadata_mut()
            .insert("x-internal-secret", self.secret.clone());

        let response = self
            .client
            .clone()
            .get_notification_target(request)
            .await?
            .into_inner();

        Ok(Audience::new(
            response.user_sub,
            normalize_locale(&response.locale)
                .unwrap_or_else(|| self.fallback_locale.clone()),
        ))
    }
}

#[async_trait]
impl AudienceResolver for AccountAudienceResolver {
    async fn resolve(&self, reference: &RecipientRef) -> Result<Audience, AudienceError> {
        match reference {
            // The subject is already the subscriber key, so only the language is at
            // stake. A message in the fallback language beats no message at all —
            // degrade rather than fail, the same posture cex-auth takes for OTP.
            RecipientRef::Subject(subject) => match self
                .lookup(By::UserId(subject.clone()))
                .await
            {
                Ok(audience) => Ok(audience),
                Err(status) => {
                    tracing::warn!(
                        code = ?status.code(),
                        "notification locale lookup failed; falling back to '{}'",
                        self.fallback_locale
                    );
                    Ok(Audience::new(subject.clone(), self.fallback_locale.clone()))
                }
            },
            // The lookup is the only source of the subject here, so there is nothing to
            // degrade to: without it there is no one to notify.
            RecipientRef::AccountId(account_id) => self
                .lookup(By::AccountId(*account_id))
                .await
                .map_err(|status| match status.code() {
                    Code::NotFound => AudienceError::NotFound(reference.log_label()),
                    _ => AudienceError::Unavailable {
                        reference: reference.log_label(),
                        message: status.message().to_string(),
                    },
                }),
            // Internal channels are routed from configuration by the notifier and never
            // reach a directory lookup. Reaching here would mean the notifier stopped
            // short-circuiting them, which is a bug rather than a runtime condition.
            RecipientRef::Admin(_) => Err(AudienceError::NotFound(reference.log_label())),
        }
    }
}

/// tonic needs a scheme; the configmap convention may already include one.
fn normalize_url(url: &str) -> String {
    let url = url.trim().trim_end_matches('/');
    if url.starts_with("http://") || url.starts_with("https://") {
        url.to_string()
    } else {
        format!("http://{url}")
    }
}

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

    #[test]
    fn a_scheme_is_added_only_when_missing() {
        assert_eq!(normalize_url("cex-account:50051"), "http://cex-account:50051");
        assert_eq!(
            normalize_url("https://cex-account:50051/"),
            "https://cex-account:50051"
        );
    }
}